chain_manager.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834
  1. // Copyright 2014 The go-ethereum Authors
  2. // This file is part of the go-ethereum library.
  3. //
  4. // The go-ethereum library is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Lesser General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // The go-ethereum library is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Lesser General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Lesser General Public License
  15. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16. // Package core implements the Ethereum consensus protocol.
  17. package core
  18. import (
  19. "errors"
  20. "fmt"
  21. "io"
  22. "math/big"
  23. "runtime"
  24. "sync"
  25. "sync/atomic"
  26. "time"
  27. "github.com/ethereum/go-ethereum/common"
  28. "github.com/ethereum/go-ethereum/core/state"
  29. "github.com/ethereum/go-ethereum/core/types"
  30. "github.com/ethereum/go-ethereum/event"
  31. "github.com/ethereum/go-ethereum/logger"
  32. "github.com/ethereum/go-ethereum/logger/glog"
  33. "github.com/ethereum/go-ethereum/metrics"
  34. "github.com/ethereum/go-ethereum/pow"
  35. "github.com/hashicorp/golang-lru"
  36. )
  37. var (
  38. chainlogger = logger.NewLogger("CHAIN")
  39. jsonlogger = logger.NewJsonLogger()
  40. blockInsertTimer = metrics.NewTimer("chain/inserts")
  41. ErrNoGenesis = errors.New("Genesis not found in chain")
  42. )
  43. const (
  44. blockCacheLimit = 256
  45. maxFutureBlocks = 256
  46. maxTimeFutureBlocks = 30
  47. checkpointLimit = 200
  48. )
  49. type ChainManager struct {
  50. //eth EthManager
  51. blockDb common.Database
  52. stateDb common.Database
  53. extraDb common.Database
  54. processor types.BlockProcessor
  55. eventMux *event.TypeMux
  56. genesisBlock *types.Block
  57. // Last known total difficulty
  58. mu sync.RWMutex
  59. chainmu sync.RWMutex
  60. tsmu sync.RWMutex
  61. checkpoint int // checkpoint counts towards the new checkpoint
  62. td *big.Int
  63. currentBlock *types.Block
  64. lastBlockHash common.Hash
  65. currentGasLimit *big.Int
  66. transState *state.StateDB
  67. txState *state.ManagedState
  68. cache *lru.Cache // cache is the LRU caching
  69. futureBlocks *lru.Cache // future blocks are blocks added for later processing
  70. quit chan struct{}
  71. // procInterrupt must be atomically called
  72. procInterrupt int32 // interrupt signaler for block processing
  73. wg sync.WaitGroup
  74. pow pow.PoW
  75. }
  76. func NewChainManager(blockDb, stateDb, extraDb common.Database, pow pow.PoW, mux *event.TypeMux) (*ChainManager, error) {
  77. cache, _ := lru.New(blockCacheLimit)
  78. bc := &ChainManager{
  79. blockDb: blockDb,
  80. stateDb: stateDb,
  81. extraDb: extraDb,
  82. eventMux: mux,
  83. quit: make(chan struct{}),
  84. cache: cache,
  85. pow: pow,
  86. }
  87. bc.genesisBlock = bc.GetBlockByNumber(0)
  88. if bc.genesisBlock == nil {
  89. // XXX Uncomment me before Frontier
  90. //return nil, ErrNoGenesis
  91. genesis, err := WriteTestNetGenesisBlock(bc.stateDb, bc.blockDb, 42)
  92. if err != nil {
  93. glog.Fatalln("genisis err", err)
  94. }
  95. bc.genesisBlock = genesis
  96. }
  97. if err := bc.setLastState(); err != nil {
  98. return nil, err
  99. }
  100. // Check the current state of the block hashes and make sure that we do not have any of the bad blocks in our chain
  101. for hash, _ := range BadHashes {
  102. if block := bc.GetBlock(hash); block != nil {
  103. glog.V(logger.Error).Infof("Found bad hash. Reorganising chain to state %x\n", block.ParentHash().Bytes()[:4])
  104. block = bc.GetBlock(block.ParentHash())
  105. if block == nil {
  106. glog.Fatal("Unable to complete. Parent block not found. Corrupted DB?")
  107. }
  108. bc.SetHead(block)
  109. glog.V(logger.Error).Infoln("Chain reorg was successfull. Resuming normal operation")
  110. }
  111. }
  112. bc.transState = bc.State().Copy()
  113. // Take ownership of this particular state
  114. bc.txState = state.ManageState(bc.State().Copy())
  115. bc.futureBlocks, _ = lru.New(maxFutureBlocks)
  116. bc.makeCache()
  117. go bc.update()
  118. return bc, nil
  119. }
  120. func (bc *ChainManager) SetHead(head *types.Block) {
  121. bc.mu.Lock()
  122. defer bc.mu.Unlock()
  123. for block := bc.currentBlock; block != nil && block.Hash() != head.Hash(); block = bc.GetBlock(block.ParentHash()) {
  124. bc.removeBlock(block)
  125. }
  126. bc.cache, _ = lru.New(blockCacheLimit)
  127. bc.currentBlock = head
  128. bc.makeCache()
  129. statedb := state.New(head.Root(), bc.stateDb)
  130. bc.txState = state.ManageState(statedb)
  131. bc.transState = statedb.Copy()
  132. bc.setTotalDifficulty(head.Td)
  133. bc.insert(head)
  134. bc.setLastState()
  135. }
  136. func (self *ChainManager) Td() *big.Int {
  137. self.mu.RLock()
  138. defer self.mu.RUnlock()
  139. return new(big.Int).Set(self.td)
  140. }
  141. func (self *ChainManager) GasLimit() *big.Int {
  142. self.mu.RLock()
  143. defer self.mu.RUnlock()
  144. return self.currentBlock.GasLimit()
  145. }
  146. func (self *ChainManager) LastBlockHash() common.Hash {
  147. self.mu.RLock()
  148. defer self.mu.RUnlock()
  149. return self.lastBlockHash
  150. }
  151. func (self *ChainManager) CurrentBlock() *types.Block {
  152. self.mu.RLock()
  153. defer self.mu.RUnlock()
  154. return self.currentBlock
  155. }
  156. func (self *ChainManager) Status() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash) {
  157. self.mu.RLock()
  158. defer self.mu.RUnlock()
  159. return new(big.Int).Set(self.td), self.currentBlock.Hash(), self.genesisBlock.Hash()
  160. }
  161. func (self *ChainManager) SetProcessor(proc types.BlockProcessor) {
  162. self.processor = proc
  163. }
  164. func (self *ChainManager) State() *state.StateDB {
  165. return state.New(self.CurrentBlock().Root(), self.stateDb)
  166. }
  167. func (self *ChainManager) TransState() *state.StateDB {
  168. self.tsmu.RLock()
  169. defer self.tsmu.RUnlock()
  170. return self.transState
  171. }
  172. func (self *ChainManager) setTransState(statedb *state.StateDB) {
  173. self.transState = statedb
  174. }
  175. func (bc *ChainManager) recover() bool {
  176. data, _ := bc.blockDb.Get([]byte("checkpoint"))
  177. if len(data) != 0 {
  178. block := bc.GetBlock(common.BytesToHash(data))
  179. if block != nil {
  180. err := bc.blockDb.Put([]byte("LastBlock"), block.Hash().Bytes())
  181. if err != nil {
  182. glog.Fatalln("db write err:", err)
  183. }
  184. bc.currentBlock = block
  185. bc.lastBlockHash = block.Hash()
  186. return true
  187. }
  188. }
  189. return false
  190. }
  191. func (bc *ChainManager) setLastState() error {
  192. data, _ := bc.blockDb.Get([]byte("LastBlock"))
  193. if len(data) != 0 {
  194. block := bc.GetBlock(common.BytesToHash(data))
  195. if block != nil {
  196. bc.currentBlock = block
  197. bc.lastBlockHash = block.Hash()
  198. } else {
  199. glog.Infof("LastBlock (%x) not found. Recovering...\n", data)
  200. if bc.recover() {
  201. glog.Infof("Recover successful")
  202. } else {
  203. glog.Fatalf("Recover failed. Please report")
  204. }
  205. }
  206. } else {
  207. bc.Reset()
  208. }
  209. bc.td = bc.currentBlock.Td
  210. bc.currentGasLimit = CalcGasLimit(bc.currentBlock)
  211. if glog.V(logger.Info) {
  212. glog.Infof("Last block (#%v) %x TD=%v\n", bc.currentBlock.Number(), bc.currentBlock.Hash(), bc.td)
  213. }
  214. return nil
  215. }
  216. func (bc *ChainManager) makeCache() {
  217. bc.cache, _ = lru.New(blockCacheLimit)
  218. // load in last `blockCacheLimit` - 1 blocks. Last block is the current.
  219. bc.cache.Add(bc.genesisBlock.Hash(), bc.genesisBlock)
  220. for _, block := range bc.GetBlocksFromHash(bc.currentBlock.Hash(), blockCacheLimit) {
  221. bc.cache.Add(block.Hash(), block)
  222. }
  223. }
  224. func (bc *ChainManager) Reset() {
  225. bc.mu.Lock()
  226. defer bc.mu.Unlock()
  227. for block := bc.currentBlock; block != nil; block = bc.GetBlock(block.ParentHash()) {
  228. bc.removeBlock(block)
  229. }
  230. bc.cache, _ = lru.New(blockCacheLimit)
  231. // Prepare the genesis block
  232. err := WriteBlock(bc.blockDb, bc.genesisBlock)
  233. if err != nil {
  234. glog.Fatalln("db err:", err)
  235. }
  236. bc.insert(bc.genesisBlock)
  237. bc.currentBlock = bc.genesisBlock
  238. bc.makeCache()
  239. bc.setTotalDifficulty(common.Big("0"))
  240. }
  241. func (bc *ChainManager) removeBlock(block *types.Block) {
  242. bc.blockDb.Delete(append(blockHashPre, block.Hash().Bytes()...))
  243. }
  244. func (bc *ChainManager) ResetWithGenesisBlock(gb *types.Block) {
  245. bc.mu.Lock()
  246. defer bc.mu.Unlock()
  247. for block := bc.currentBlock; block != nil; block = bc.GetBlock(block.ParentHash()) {
  248. bc.removeBlock(block)
  249. }
  250. // Prepare the genesis block
  251. gb.Td = gb.Difficulty()
  252. bc.genesisBlock = gb
  253. err := WriteBlock(bc.blockDb, bc.genesisBlock)
  254. if err != nil {
  255. glog.Fatalln("db err:", err)
  256. }
  257. bc.insert(bc.genesisBlock)
  258. bc.currentBlock = bc.genesisBlock
  259. bc.makeCache()
  260. bc.td = gb.Difficulty()
  261. }
  262. // Export writes the active chain to the given writer.
  263. func (self *ChainManager) Export(w io.Writer) error {
  264. if err := self.ExportN(w, uint64(0), self.currentBlock.NumberU64()); err != nil {
  265. return err
  266. }
  267. return nil
  268. }
  269. // ExportN writes a subset of the active chain to the given writer.
  270. func (self *ChainManager) ExportN(w io.Writer, first uint64, last uint64) error {
  271. self.mu.RLock()
  272. defer self.mu.RUnlock()
  273. if first > last {
  274. return fmt.Errorf("export failed: first (%d) is greater than last (%d)", first, last)
  275. }
  276. glog.V(logger.Info).Infof("exporting %d blocks...\n", last-first+1)
  277. for nr := first; nr <= last; nr++ {
  278. block := self.GetBlockByNumber(nr)
  279. if block == nil {
  280. return fmt.Errorf("export failed on #%d: not found", nr)
  281. }
  282. if err := block.EncodeRLP(w); err != nil {
  283. return err
  284. }
  285. }
  286. return nil
  287. }
  288. // insert injects a block into the current chain block chain. Note, this function
  289. // assumes that the `mu` mutex is held!
  290. func (bc *ChainManager) insert(block *types.Block) {
  291. err := WriteHead(bc.blockDb, block)
  292. if err != nil {
  293. glog.Fatal("db write fail:", err)
  294. }
  295. bc.checkpoint++
  296. if bc.checkpoint > checkpointLimit {
  297. err = bc.blockDb.Put([]byte("checkpoint"), block.Hash().Bytes())
  298. if err != nil {
  299. glog.Fatal("db write fail:", err)
  300. }
  301. bc.checkpoint = 0
  302. }
  303. bc.currentBlock = block
  304. bc.lastBlockHash = block.Hash()
  305. }
  306. // Accessors
  307. func (bc *ChainManager) Genesis() *types.Block {
  308. return bc.genesisBlock
  309. }
  310. // Block fetching methods
  311. func (bc *ChainManager) HasBlock(hash common.Hash) bool {
  312. if bc.cache.Contains(hash) {
  313. return true
  314. }
  315. data, _ := bc.blockDb.Get(append(blockHashPre, hash[:]...))
  316. return len(data) != 0
  317. }
  318. func (self *ChainManager) GetBlockHashesFromHash(hash common.Hash, max uint64) (chain []common.Hash) {
  319. block := self.GetBlock(hash)
  320. if block == nil {
  321. return
  322. }
  323. // XXX Could be optimised by using a different database which only holds hashes (i.e., linked list)
  324. for i := uint64(0); i < max; i++ {
  325. block = self.GetBlock(block.ParentHash())
  326. if block == nil {
  327. break
  328. }
  329. chain = append(chain, block.Hash())
  330. if block.Number().Cmp(common.Big0) <= 0 {
  331. break
  332. }
  333. }
  334. return
  335. }
  336. func (self *ChainManager) GetBlock(hash common.Hash) *types.Block {
  337. if block, ok := self.cache.Get(hash); ok {
  338. return block.(*types.Block)
  339. }
  340. block := GetBlockByHash(self.blockDb, hash)
  341. if block == nil {
  342. return nil
  343. }
  344. // Add the block to the cache
  345. self.cache.Add(hash, (*types.Block)(block))
  346. return (*types.Block)(block)
  347. }
  348. func (self *ChainManager) GetBlockByNumber(num uint64) *types.Block {
  349. self.mu.RLock()
  350. defer self.mu.RUnlock()
  351. return self.getBlockByNumber(num)
  352. }
  353. // GetBlocksFromHash returns the block corresponding to hash and up to n-1 ancestors.
  354. func (self *ChainManager) GetBlocksFromHash(hash common.Hash, n int) (blocks []*types.Block) {
  355. for i := 0; i < n; i++ {
  356. block := self.GetBlock(hash)
  357. if block == nil {
  358. break
  359. }
  360. blocks = append(blocks, block)
  361. hash = block.ParentHash()
  362. }
  363. return
  364. }
  365. // non blocking version
  366. func (self *ChainManager) getBlockByNumber(num uint64) *types.Block {
  367. return GetBlockByNumber(self.blockDb, num)
  368. }
  369. func (self *ChainManager) GetUnclesInChain(block *types.Block, length int) (uncles []*types.Header) {
  370. for i := 0; block != nil && i < length; i++ {
  371. uncles = append(uncles, block.Uncles()...)
  372. block = self.GetBlock(block.ParentHash())
  373. }
  374. return
  375. }
  376. // setTotalDifficulty updates the TD of the chain manager. Note, this function
  377. // assumes that the `mu` mutex is held!
  378. func (bc *ChainManager) setTotalDifficulty(td *big.Int) {
  379. bc.td = new(big.Int).Set(td)
  380. }
  381. func (bc *ChainManager) Stop() {
  382. close(bc.quit)
  383. atomic.StoreInt32(&bc.procInterrupt, 1)
  384. bc.wg.Wait()
  385. glog.V(logger.Info).Infoln("Chain manager stopped")
  386. }
  387. type queueEvent struct {
  388. queue []interface{}
  389. canonicalCount int
  390. sideCount int
  391. splitCount int
  392. }
  393. func (self *ChainManager) procFutureBlocks() {
  394. blocks := make([]*types.Block, self.futureBlocks.Len())
  395. for i, hash := range self.futureBlocks.Keys() {
  396. block, _ := self.futureBlocks.Get(hash)
  397. blocks[i] = block.(*types.Block)
  398. }
  399. if len(blocks) > 0 {
  400. types.BlockBy(types.Number).Sort(blocks)
  401. self.InsertChain(blocks)
  402. }
  403. }
  404. type writeStatus byte
  405. const (
  406. NonStatTy writeStatus = iota
  407. CanonStatTy
  408. SplitStatTy
  409. SideStatTy
  410. )
  411. // WriteBlock writes the block to the chain (or pending queue)
  412. func (self *ChainManager) WriteBlock(block *types.Block, queued bool) (status writeStatus, err error) {
  413. self.wg.Add(1)
  414. defer self.wg.Done()
  415. cblock := self.currentBlock
  416. // Compare the TD of the last known block in the canonical chain to make sure it's greater.
  417. // At this point it's possible that a different chain (fork) becomes the new canonical chain.
  418. if block.Td.Cmp(self.Td()) > 0 {
  419. // chain fork
  420. if block.ParentHash() != cblock.Hash() {
  421. // during split we merge two different chains and create the new canonical chain
  422. err := self.merge(cblock, block)
  423. if err != nil {
  424. return NonStatTy, err
  425. }
  426. status = SplitStatTy
  427. }
  428. self.mu.Lock()
  429. self.setTotalDifficulty(block.Td)
  430. self.insert(block)
  431. self.mu.Unlock()
  432. self.setTransState(state.New(block.Root(), self.stateDb))
  433. self.txState.SetState(state.New(block.Root(), self.stateDb))
  434. status = CanonStatTy
  435. } else {
  436. status = SideStatTy
  437. }
  438. err = WriteBlock(self.blockDb, block)
  439. if err != nil {
  440. glog.Fatalln("db err:", err)
  441. }
  442. // Delete from future blocks
  443. self.futureBlocks.Remove(block.Hash())
  444. return
  445. }
  446. // InsertChain will attempt to insert the given chain in to the canonical chain or, otherwise, create a fork. It an error is returned
  447. // it will return the index number of the failing block as well an error describing what went wrong (for possible errors see core/errors.go).
  448. func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
  449. self.wg.Add(1)
  450. defer self.wg.Done()
  451. self.chainmu.Lock()
  452. defer self.chainmu.Unlock()
  453. // A queued approach to delivering events. This is generally
  454. // faster than direct delivery and requires much less mutex
  455. // acquiring.
  456. var (
  457. queue = make([]interface{}, len(chain))
  458. queueEvent = queueEvent{queue: queue}
  459. stats struct{ queued, processed, ignored int }
  460. tstart = time.Now()
  461. nonceDone = make(chan nonceResult, len(chain))
  462. nonceQuit = make(chan struct{})
  463. nonceChecked = make([]bool, len(chain))
  464. )
  465. // Start the parallel nonce verifier.
  466. go verifyNonces(self.pow, chain, nonceQuit, nonceDone)
  467. defer close(nonceQuit)
  468. txcount := 0
  469. for i, block := range chain {
  470. if atomic.LoadInt32(&self.procInterrupt) == 1 {
  471. glog.V(logger.Debug).Infoln("Premature abort during chain processing")
  472. break
  473. }
  474. bstart := time.Now()
  475. // Wait for block i's nonce to be verified before processing
  476. // its state transition.
  477. for !nonceChecked[i] {
  478. r := <-nonceDone
  479. nonceChecked[r.i] = true
  480. if !r.valid {
  481. block := chain[r.i]
  482. return r.i, &BlockNonceErr{Hash: block.Hash(), Number: block.Number(), Nonce: block.Nonce()}
  483. }
  484. }
  485. if BadHashes[block.Hash()] {
  486. err := fmt.Errorf("Found known bad hash in chain %x", block.Hash())
  487. blockErr(block, err)
  488. return i, err
  489. }
  490. // Setting block.Td regardless of error (known for example) prevents errors down the line
  491. // in the protocol handler
  492. block.Td = new(big.Int).Set(CalcTD(block, self.GetBlock(block.ParentHash())))
  493. // Call in to the block processor and check for errors. It's likely that if one block fails
  494. // all others will fail too (unless a known block is returned).
  495. logs, receipts, err := self.processor.Process(block)
  496. if err != nil {
  497. if IsKnownBlockErr(err) {
  498. stats.ignored++
  499. continue
  500. }
  501. if err == BlockFutureErr {
  502. // Allow up to MaxFuture second in the future blocks. If this limit
  503. // is exceeded the chain is discarded and processed at a later time
  504. // if given.
  505. if max := uint64(time.Now().Unix()) + maxTimeFutureBlocks; block.Time() > max {
  506. return i, fmt.Errorf("%v: BlockFutureErr, %v > %v", BlockFutureErr, block.Time(), max)
  507. }
  508. self.futureBlocks.Add(block.Hash(), block)
  509. stats.queued++
  510. continue
  511. }
  512. if IsParentErr(err) && self.futureBlocks.Contains(block.ParentHash()) {
  513. self.futureBlocks.Add(block.Hash(), block)
  514. stats.queued++
  515. continue
  516. }
  517. blockErr(block, err)
  518. go ReportBlock(block, err)
  519. return i, err
  520. }
  521. txcount += len(block.Transactions())
  522. // write the block to the chain and get the status
  523. status, err := self.WriteBlock(block, true)
  524. if err != nil {
  525. return i, err
  526. }
  527. switch status {
  528. case CanonStatTy:
  529. if glog.V(logger.Debug) {
  530. glog.Infof("[%v] inserted block #%d (%d TXs %d UNCs) (%x...). Took %v\n", time.Now().UnixNano(), block.Number(), len(block.Transactions()), len(block.Uncles()), block.Hash().Bytes()[0:4], time.Since(bstart))
  531. }
  532. queue[i] = ChainEvent{block, block.Hash(), logs}
  533. queueEvent.canonicalCount++
  534. // This puts transactions in a extra db for rpc
  535. PutTransactions(self.extraDb, block, block.Transactions())
  536. // store the receipts
  537. PutReceipts(self.extraDb, receipts)
  538. case SideStatTy:
  539. if glog.V(logger.Detail) {
  540. glog.Infof("inserted forked block #%d (TD=%v) (%d TXs %d UNCs) (%x...). Took %v\n", block.Number(), block.Difficulty(), len(block.Transactions()), len(block.Uncles()), block.Hash().Bytes()[0:4], time.Since(bstart))
  541. }
  542. queue[i] = ChainSideEvent{block, logs}
  543. queueEvent.sideCount++
  544. case SplitStatTy:
  545. queue[i] = ChainSplitEvent{block, logs}
  546. queueEvent.splitCount++
  547. }
  548. PutBlockReceipts(self.extraDb, block, receipts)
  549. stats.processed++
  550. }
  551. if (stats.queued > 0 || stats.processed > 0 || stats.ignored > 0) && bool(glog.V(logger.Info)) {
  552. tend := time.Since(tstart)
  553. start, end := chain[0], chain[len(chain)-1]
  554. glog.Infof("imported %d block(s) (%d queued %d ignored) including %d txs in %v. #%v [%x / %x]\n", stats.processed, stats.queued, stats.ignored, txcount, tend, end.Number(), start.Hash().Bytes()[:4], end.Hash().Bytes()[:4])
  555. }
  556. go self.eventMux.Post(queueEvent)
  557. return 0, nil
  558. }
  559. // diff takes two blocks, an old chain and a new chain and will reconstruct the blocks and inserts them
  560. // to be part of the new canonical chain.
  561. func (self *ChainManager) diff(oldBlock, newBlock *types.Block) (types.Blocks, error) {
  562. var (
  563. newChain types.Blocks
  564. commonBlock *types.Block
  565. oldStart = oldBlock
  566. newStart = newBlock
  567. )
  568. // first reduce whoever is higher bound
  569. if oldBlock.NumberU64() > newBlock.NumberU64() {
  570. // reduce old chain
  571. for oldBlock = oldBlock; oldBlock != nil && oldBlock.NumberU64() != newBlock.NumberU64(); oldBlock = self.GetBlock(oldBlock.ParentHash()) {
  572. }
  573. } else {
  574. // reduce new chain and append new chain blocks for inserting later on
  575. for newBlock = newBlock; newBlock != nil && newBlock.NumberU64() != oldBlock.NumberU64(); newBlock = self.GetBlock(newBlock.ParentHash()) {
  576. newChain = append(newChain, newBlock)
  577. }
  578. }
  579. if oldBlock == nil {
  580. return nil, fmt.Errorf("Invalid old chain")
  581. }
  582. if newBlock == nil {
  583. return nil, fmt.Errorf("Invalid new chain")
  584. }
  585. numSplit := newBlock.Number()
  586. for {
  587. if oldBlock.Hash() == newBlock.Hash() {
  588. commonBlock = oldBlock
  589. break
  590. }
  591. newChain = append(newChain, newBlock)
  592. oldBlock, newBlock = self.GetBlock(oldBlock.ParentHash()), self.GetBlock(newBlock.ParentHash())
  593. if oldBlock == nil {
  594. return nil, fmt.Errorf("Invalid old chain")
  595. }
  596. if newBlock == nil {
  597. return nil, fmt.Errorf("Invalid new chain")
  598. }
  599. }
  600. if glog.V(logger.Debug) {
  601. commonHash := commonBlock.Hash()
  602. glog.Infof("Chain split detected @ %x. Reorganising chain from #%v %x to %x", commonHash[:4], numSplit, oldStart.Hash().Bytes()[:4], newStart.Hash().Bytes()[:4])
  603. }
  604. return newChain, nil
  605. }
  606. // merge merges two different chain to the new canonical chain
  607. func (self *ChainManager) merge(oldBlock, newBlock *types.Block) error {
  608. newChain, err := self.diff(oldBlock, newBlock)
  609. if err != nil {
  610. return fmt.Errorf("chain reorg failed: %v", err)
  611. }
  612. // insert blocks. Order does not matter. Last block will be written in ImportChain itself which creates the new head properly
  613. self.mu.Lock()
  614. for _, block := range newChain {
  615. // insert the block in the canonical way, re-writing history
  616. self.insert(block)
  617. // write canonical receipts and transactions
  618. PutTransactions(self.extraDb, block, block.Transactions())
  619. PutReceipts(self.extraDb, GetBlockReceipts(self.extraDb, block.Hash()))
  620. }
  621. self.mu.Unlock()
  622. return nil
  623. }
  624. func (self *ChainManager) update() {
  625. events := self.eventMux.Subscribe(queueEvent{})
  626. futureTimer := time.Tick(5 * time.Second)
  627. out:
  628. for {
  629. select {
  630. case ev := <-events.Chan():
  631. switch ev := ev.(type) {
  632. case queueEvent:
  633. for _, event := range ev.queue {
  634. switch event := event.(type) {
  635. case ChainEvent:
  636. // We need some control over the mining operation. Acquiring locks and waiting for the miner to create new block takes too long
  637. // and in most cases isn't even necessary.
  638. if self.lastBlockHash == event.Hash {
  639. self.currentGasLimit = CalcGasLimit(event.Block)
  640. self.eventMux.Post(ChainHeadEvent{event.Block})
  641. }
  642. }
  643. self.eventMux.Post(event)
  644. }
  645. }
  646. case <-futureTimer:
  647. self.procFutureBlocks()
  648. case <-self.quit:
  649. break out
  650. }
  651. }
  652. }
  653. func blockErr(block *types.Block, err error) {
  654. h := block.Header()
  655. glog.V(logger.Error).Infof("Bad block #%v (%x)\n", h.Number, h.Hash().Bytes())
  656. glog.V(logger.Error).Infoln(err)
  657. glog.V(logger.Debug).Infoln(verifyNonces)
  658. }
  659. type nonceResult struct {
  660. i int
  661. valid bool
  662. }
  663. // block verifies nonces of the given blocks in parallel and returns
  664. // an error if one of the blocks nonce verifications failed.
  665. func verifyNonces(pow pow.PoW, blocks []*types.Block, quit <-chan struct{}, done chan<- nonceResult) {
  666. // Spawn a few workers. They listen for blocks on the in channel
  667. // and send results on done. The workers will exit in the
  668. // background when in is closed.
  669. var (
  670. in = make(chan int)
  671. nworkers = runtime.GOMAXPROCS(0)
  672. )
  673. defer close(in)
  674. if len(blocks) < nworkers {
  675. nworkers = len(blocks)
  676. }
  677. for i := 0; i < nworkers; i++ {
  678. go func() {
  679. for i := range in {
  680. done <- nonceResult{i: i, valid: pow.Verify(blocks[i])}
  681. }
  682. }()
  683. }
  684. // Feed block indices to the workers.
  685. for i := range blocks {
  686. select {
  687. case in <- i:
  688. continue
  689. case <-quit:
  690. return
  691. }
  692. }
  693. }