chain_manager.go 22 KB

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