diff --git a/core/blockchain.go b/core/blockchain.go index 59be355895..2cf57012ce 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -69,6 +69,7 @@ var ( blockPrefetchInterruptMeter = metrics.NewRegisteredMeter("chain/prefetch/interrupts", nil) errInsertionInterrupted = errors.New("insertion is interrupted") + errClosed = errors.New("blockchain is closed") ) const ( @@ -157,11 +158,9 @@ type BlockChain struct { blockCache *lru.Cache // Cache for the most recent entire blocks futureBlocks *lru.Cache // future blocks are blocks added for later processing - quit chan struct{} // blockchain quit channel - running int32 // running must be called atomically - // procInterrupt must be atomically called - procInterrupt int32 // interrupt signaler for block processing - wg sync.WaitGroup // chain processing wait group for shutting down + quit chan struct{} // blockchain quit channel + closed int32 // Indicator whether the blockchain is still running. + wg sync.WaitGroup // chain processing wait group for shutting down engine consensus.Engine validator Validator // Block and state validator interface @@ -214,7 +213,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par bc.processor = NewStateProcessor(chainConfig, bc, engine) var err error - bc.hc, err = NewHeaderChain(db, chainConfig, engine, bc.getProcInterrupt) + bc.hc, err = NewHeaderChain(db, chainConfig, engine, bc.isClosed) if err != nil { return nil, err } @@ -285,8 +284,8 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par return bc, nil } -func (bc *BlockChain) getProcInterrupt() bool { - return atomic.LoadInt32(&bc.procInterrupt) == 1 +func (bc *BlockChain) isClosed() bool { + return atomic.LoadInt32(&bc.closed) == 1 } // GetVMConfig returns the block chain VM config. @@ -381,6 +380,13 @@ func (bc *BlockChain) SetHead(head uint64) error { bc.chainmu.Lock() defer bc.chainmu.Unlock() + bc.wg.Add(1) + defer bc.wg.Done() + + // Short circuit if the blockchain is already closed. + if bc.isClosed() { + return errClosed + } updateFn := func(db ethdb.KeyValueWriter, header *types.Header) { // Rewind the block chain, ensuring we don't end up with a stateless head block if currentBlock := bc.CurrentBlock(); currentBlock != nil && header.Number.Uint64() < currentBlock.NumberU64() { @@ -794,16 +800,14 @@ func (bc *BlockChain) TrieNode(hash common.Hash) ([]byte, error) { } // Stop stops the blockchain service. If any imports are currently in progress -// it will abort them using the procInterrupt. +// it will abort them using the closed. func (bc *BlockChain) Stop() { - if !atomic.CompareAndSwapInt32(&bc.running, 0, 1) { + if !atomic.CompareAndSwapInt32(&bc.closed, 0, 1) { return } // Unsubscribe all subscriptions registered from blockchain bc.scope.Close() close(bc.quit) - atomic.StoreInt32(&bc.procInterrupt, 1) - bc.wg.Wait() // Ensure the state of a recent block is also stored to disk before exiting. @@ -866,6 +870,13 @@ func (bc *BlockChain) Rollback(chain []common.Hash) { bc.chainmu.Lock() defer bc.chainmu.Unlock() + bc.wg.Add(1) + defer bc.wg.Done() + + // Short circuit if the blockchain is already closed + if bc.isClosed() { + return + } for i := len(chain) - 1; i >= 0; i-- { hash := chain[i] @@ -941,6 +952,10 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ bc.wg.Add(1) defer bc.wg.Done() + // Short circuit is the blockchain is already closed + if bc.isClosed() { + return 0, errClosed + } var ( ancientBlocks, liveBlocks types.Blocks ancientReceipts, liveReceipts []types.Receipts @@ -1007,7 +1022,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ var deleted []*numberHash for i, block := range blockChain { // Short circuit insertion if shutting down or processing failed - if atomic.LoadInt32(&bc.procInterrupt) == 1 { + if bc.isClosed() { return 0, errInsertionInterrupted } // Short circuit insertion if it is required(used in testing only) @@ -1140,7 +1155,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ batch := bc.db.NewBatch() for i, block := range blockChain { // Short circuit insertion if shutting down or processing failed - if atomic.LoadInt32(&bc.procInterrupt) == 1 { + if bc.isClosed() { return 0, errInsertionInterrupted } // Short circuit if the owner header is unknown @@ -1212,9 +1227,6 @@ var lastWrite uint64 // but does not write any state. This is used to construct competing side forks // up to the point where they exceed the canonical total difficulty. func (bc *BlockChain) writeBlockWithoutState(block *types.Block, td *big.Int) (err error) { - bc.wg.Add(1) - defer bc.wg.Done() - if err := bc.hc.WriteTd(block.Hash(), block.NumberU64(), td); err != nil { return err } @@ -1226,9 +1238,6 @@ func (bc *BlockChain) writeBlockWithoutState(block *types.Block, td *big.Int) (e // writeKnownBlock updates the head block flag with a known block // and introduces chain reorg if necessary. func (bc *BlockChain) writeKnownBlock(block *types.Block) error { - bc.wg.Add(1) - defer bc.wg.Done() - current := bc.CurrentBlock() if block.ParentHash() != current.Hash() { if err := bc.reorg(current, block); err != nil { @@ -1248,15 +1257,19 @@ func (bc *BlockChain) WriteBlockWithState(block *types.Block, receipts []*types. bc.chainmu.Lock() defer bc.chainmu.Unlock() + bc.wg.Add(1) + defer bc.wg.Done() + + // Short circuit if the blockchain is already closed + if bc.isClosed() { + return NonStatTy, errClosed + } return bc.writeBlockWithState(block, receipts, state) } // writeBlockWithState writes the block and all associated state to the database, // but is expects the chain mutex to be held. func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.Receipt, state *state.StateDB) (status WriteStatus, err error) { - bc.wg.Add(1) - defer bc.wg.Done() - // Calculate the total difficulty of the block ptd := bc.GetTd(block.ParentHash(), block.NumberU64()-1) if ptd == nil { @@ -1404,15 +1417,8 @@ func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) { if len(chain) == 0 { return 0, nil } - - bc.blockProcFeed.Send(true) - defer bc.blockProcFeed.Send(false) - - // Remove already known canon-blocks - var ( - block, prev *types.Block - ) // Do a sanity check that the provided chain is actually ordered and linked + var block, prev *types.Block for i := 1; i < len(chain); i++ { block = chain[i] prev = chain[i-1] @@ -1426,11 +1432,21 @@ func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) { } } // Pre-checks passed, start the full block imports - bc.wg.Add(1) bc.chainmu.Lock() + bc.wg.Add(1) + + // Short circuit if the blockchain is already closed. + if bc.isClosed() { + bc.wg.Done() + bc.chainmu.Unlock() + return 0, errClosed + } + bc.blockProcFeed.Send(true) + defer bc.blockProcFeed.Send(false) + n, events, logs, err := bc.insertChain(chain, true) - bc.chainmu.Unlock() bc.wg.Done() + bc.chainmu.Unlock() bc.PostChainEvents(events, logs) return n, err @@ -1445,10 +1461,6 @@ func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) { // is imported, but then new canon-head is added before the actual sidechain // completes, then the historic state could be pruned again func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, []interface{}, []*types.Log, error) { - // If the chain is terminating, don't even bother starting up - if atomic.LoadInt32(&bc.procInterrupt) == 1 { - return 0, nil, nil, nil - } // Start a parallel signature recovery (signer will fluke on fork transition, minimal perf loss) senderCacher.recoverFromBlocks(types.MakeSigner(bc.chainConfig, chain[0].Number()), chain) @@ -1548,7 +1560,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, [] // No validation errors for the first block (or chain prefix skipped) for ; block != nil && err == nil || err == ErrKnownBlock; block, err = it.next() { // If the chain is terminating, stop processing blocks - if atomic.LoadInt32(&bc.procInterrupt) == 1 { + if bc.isClosed() { log.Debug("Premature abort during blocks processing") break } @@ -1825,7 +1837,7 @@ func (bc *BlockChain) insertSideChain(block *types.Block, it *insertIterator) (i blocks, memory = blocks[:0], 0 // If the chain is terminating, stop processing blocks - if atomic.LoadInt32(&bc.procInterrupt) == 1 { + if bc.isClosed() { log.Debug("Premature abort during blocks processing") return 0, nil, nil, nil } @@ -2061,6 +2073,8 @@ Error: %v // InsertHeaderChain attempts to insert the given header chain in to the local // chain, possibly creating a reorg. If an error is returned, it will return the // index number of the failing header as well an error describing what went wrong. +// If the blockchain is closed, all mutation operations including this function +// will be rejected. // // The verify parameter can be used to fine tune whether nonce verification // should be done or not. The reason behind the optional check is because some @@ -2079,6 +2093,10 @@ func (bc *BlockChain) InsertHeaderChain(chain []*types.Header, checkFreq int) (i bc.wg.Add(1) defer bc.wg.Done() + // Short circuit if blockchain is already closed. + if bc.isClosed() { + return 0, errClosed + } whFunc := func(header *types.Header) error { _, err := bc.hc.WriteHeader(header) return err @@ -2134,9 +2152,6 @@ func (bc *BlockChain) GetBlockHashesFromHash(hash common.Hash, max uint64) []com // // Note: ancestor == 0 returns the same block, 1 returns its parent and so on. func (bc *BlockChain) GetAncestor(hash common.Hash, number, ancestor uint64, maxNonCanonical *uint64) (common.Hash, uint64) { - bc.chainmu.RLock() - defer bc.chainmu.RUnlock() - return bc.hc.GetAncestor(hash, number, ancestor, maxNonCanonical) } diff --git a/core/headerchain.go b/core/headerchain.go index 034858f651..74b8a1fb77 100644 --- a/core/headerchain.go +++ b/core/headerchain.go @@ -60,7 +60,7 @@ type HeaderChain struct { tdCache *lru.Cache // Cache for the most recent block total difficulties numberCache *lru.Cache // Cache for the most recent block numbers - procInterrupt func() bool + isClosed func() bool // Callback whether the upper level chain is closed. rand *mrand.Rand engine consensus.Engine @@ -70,7 +70,7 @@ type HeaderChain struct { // getValidator should return the parent's validator // procInterrupt points to the parent's interrupt semaphore // wg points to the parent's shutdown wait group -func NewHeaderChain(chainDb ethdb.Database, config *params.ChainConfig, engine consensus.Engine, procInterrupt func() bool) (*HeaderChain, error) { +func NewHeaderChain(chainDb ethdb.Database, config *params.ChainConfig, engine consensus.Engine, isClosed func() bool) (*HeaderChain, error) { headerCache, _ := lru.New(headerCacheLimit) tdCache, _ := lru.New(tdCacheLimit) numberCache, _ := lru.New(numberCacheLimit) @@ -82,14 +82,14 @@ func NewHeaderChain(chainDb ethdb.Database, config *params.ChainConfig, engine c } hc := &HeaderChain{ - config: config, - chainDb: chainDb, - headerCache: headerCache, - tdCache: tdCache, - numberCache: numberCache, - procInterrupt: procInterrupt, - rand: mrand.New(mrand.NewSource(seed.Int64())), - engine: engine, + config: config, + chainDb: chainDb, + headerCache: headerCache, + tdCache: tdCache, + numberCache: numberCache, + isClosed: isClosed, + rand: mrand.New(mrand.NewSource(seed.Int64())), + engine: engine, } hc.genesisHeader = hc.GetHeaderByNumber(0) @@ -239,7 +239,7 @@ func (hc *HeaderChain) ValidateHeaderChain(chain []*types.Header, checkFreq int) // Iterate over the headers and ensure they all check out for i, header := range chain { // If the chain is terminating, stop processing blocks - if hc.procInterrupt() { + if hc.isClosed() { log.Debug("Premature abort during headers verification") return 0, errors.New("aborted") } @@ -270,7 +270,7 @@ func (hc *HeaderChain) InsertHeaderChain(chain []*types.Header, writeHeader WhCa // All headers passed verification, import them into the database for i, header := range chain { // Short circuit insertion if shutting down - if hc.procInterrupt() { + if hc.isClosed() { log.Debug("Premature abort during headers import") return i, errors.New("aborted") } diff --git a/light/lightchain.go b/light/lightchain.go index 7f64d1c28b..d4bebf02d1 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -43,6 +43,8 @@ import ( var ( bodyCacheLimit = 256 blockCacheLimit = 256 + + errClosed = errors.New("lightchain is closed") ) // LightChain represents a canonical chain that by default only handles block @@ -69,8 +71,7 @@ type LightChain struct { wg sync.WaitGroup // Atomic boolean switches: - running int32 // whether LightChain is running or stopped - procInterrupt int32 // interrupts chain insert + closed int32 // whether LightChain is already closed disableCheckFreq int32 // disables header verification } @@ -93,7 +94,7 @@ func NewLightChain(odr OdrBackend, config *params.ChainConfig, engine consensus. engine: engine, } var err error - bc.hc, err = core.NewHeaderChain(odr.Database(), config, bc.engine, bc.getProcInterrupt) + bc.hc, err = core.NewHeaderChain(odr.Database(), config, bc.engine, bc.isClosed) if err != nil { return nil, err } @@ -134,8 +135,8 @@ func (lc *LightChain) AddTrustedCheckpoint(cp *params.TrustedCheckpoint) { log.Info("Added trusted checkpoint", "block", (cp.SectionIndex+1)*lc.indexerConfig.ChtSize-1, "hash", cp.SectionHead) } -func (lc *LightChain) getProcInterrupt() bool { - return atomic.LoadInt32(&lc.procInterrupt) == 1 +func (lc *LightChain) isClosed() bool { + return atomic.LoadInt32(&lc.closed) == 1 } // Odr returns the ODR backend of the chain @@ -302,11 +303,10 @@ func (lc *LightChain) GetBlockByNumber(ctx context.Context, number uint64) (*typ // Stop stops the blockchain service. If any imports are currently in progress // it will abort them using the procInterrupt. func (lc *LightChain) Stop() { - if !atomic.CompareAndSwapInt32(&lc.running, 0, 1) { + if !atomic.CompareAndSwapInt32(&lc.closed, 0, 1) { return } close(lc.quit) - atomic.StoreInt32(&lc.procInterrupt, 1) lc.wg.Wait() log.Info("Blockchain manager stopped") @@ -370,6 +370,10 @@ func (lc *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int) (i lc.wg.Add(1) defer lc.wg.Done() + // Short circuit if lightchain is already closed + if lc.isClosed() { + return 0, errClosed + } var events []interface{} whFunc := func(header *types.Header) error { status, err := lc.hc.WriteHeader(header) @@ -438,9 +442,6 @@ func (lc *LightChain) GetBlockHashesFromHash(hash common.Hash, max uint64) []com // // Note: ancestor == 0 returns the same block, 1 returns its parent and so on. func (lc *LightChain) GetAncestor(hash common.Hash, number, ancestor uint64, maxNonCanonical *uint64) (common.Hash, uint64) { - lc.chainmu.RLock() - defer lc.chainmu.RUnlock() - return lc.hc.GetAncestor(hash, number, ancestor, maxNonCanonical) }