diff --git a/core/chain_indexer.go b/core/chain_indexer.go index 678444d7eb..a66dbadd00 100644 --- a/core/chain_indexer.go +++ b/core/chain_indexer.go @@ -18,6 +18,7 @@ package core import ( "encoding/binary" + "errors" "fmt" "sync" "sync/atomic" @@ -41,23 +42,20 @@ type ChainIndexerBackend interface { // Process crunches through the next header in the chain segment. The caller // will ensure a sequential order of headers. - Process(header *types.Header) + Process(header *types.Header) error // Commit finalizes the section metadata and stores it into the database. Commit() error - // Closing signals the backend about the indexer shutting down. After Closing - // is called the backend should not block waiting for any external events and - // may return with an error if it cannot finish the current operation. If there - // are no blocking operations and result is guaranteed in a limited time then - // Closing can be ignored. - // - // Note: Closing may be called during any phase of section processing or also - // when no processing is happening at all. It applies to all current and - // subsequent blocking operations. - Closing() + // Close shuts the backend down. After Close returns, all subsequent calls to + // the backend should return with ErrIndexerBackendClosed. Any blocking operations + // should be cancelled and return with ErrIndexerBackendClosed. Shorter operations + // may be finished and return normally while Close is blocking. + Close() } +var ErrIndexerBackendClosed = errors.New("Indexer already closed") + // ChainIndexerChain interface is used for connecting the indexer to a blockchain type ChainIndexerChain interface { // CurrentHeader retrieves the latest locally known header. @@ -149,7 +147,7 @@ func (c *ChainIndexer) Start(chain ChainIndexerChain) { func (c *ChainIndexer) Close() error { var errs []error - c.backend.Closing() + c.backend.Close() // Tear down the primary update loop errc := make(chan error) @@ -310,6 +308,9 @@ func (c *ChainIndexer) updateLoop() { c.lock.Unlock() newHead, err := c.processSection(section, oldHead) if err != nil { + if err == ErrIndexerBackendClosed { + continue + } c.log.Error("Section processing failed", "error", err) } c.lock.Lock() @@ -373,11 +374,12 @@ func (c *ChainIndexer) processSection(section uint64, lastHead common.Hash) (com } else if header.ParentHash != lastHead { return common.Hash{}, fmt.Errorf("chain reorged during section processing") } - c.backend.Process(header) + if err := c.backend.Process(header); err != nil { + return common.Hash{}, err + } lastHead = header.Hash() } if err := c.backend.Commit(); err != nil { - c.log.Error("Section commit failed", "error", err) return common.Hash{}, err } return lastHead, nil diff --git a/core/chain_indexer_test.go b/core/chain_indexer_test.go index 8f0872ce04..d595ee7da9 100644 --- a/core/chain_indexer_test.go +++ b/core/chain_indexer_test.go @@ -216,7 +216,7 @@ func (b *testChainIndexBackend) Reset(section uint64, prevHead common.Hash) erro return nil } -func (b *testChainIndexBackend) Process(header *types.Header) { +func (b *testChainIndexBackend) Process(header *types.Header) error { b.headerCnt++ if b.headerCnt > b.indexer.sectionSize { b.t.Error("Processing too many headers") @@ -227,6 +227,7 @@ func (b *testChainIndexBackend) Process(header *types.Header) { b.t.Fatal("Unexpected call to Process") case b.processCh <- header.Number.Uint64(): } + return nil } func (b *testChainIndexBackend) Commit() error { @@ -236,4 +237,4 @@ func (b *testChainIndexBackend) Commit() error { return nil } -func (b *testChainIndexBackend) Closing() {} +func (b *testChainIndexBackend) Close() {} diff --git a/eth/bloombits.go b/eth/bloombits.go index 1a1bd9af04..f55f73bd32 100644 --- a/eth/bloombits.go +++ b/eth/bloombits.go @@ -92,21 +92,22 @@ const ( // BloomIndexer implements a core.ChainIndexer, building up a rotated bloom bits index // for the Ethereum header bloom filters, permitting blazing fast filtering. type BloomIndexer struct { - size uint64 // section size to generate bloombits for - - db ethdb.Database // database instance to write index data and metadata into - gen *bloombits.Generator // generator to rotate the bloom bits crating the bloom index - - section uint64 // Section is the section number being processed currently - head common.Hash // Head is the hash of the last header processed + size uint64 // section size to generate bloombits for + db ethdb.Database // database instance to write index data and metadata into + gen *bloombits.Generator // generator to rotate the bloom bits crating the bloom index + section uint64 // Section is the section number being processed currently + head common.Hash // Head is the hash of the last header processed + quit, locked chan struct{} } // NewBloomIndexer returns a chain indexer that generates bloom bits data for the // canonical chain for fast logs filtering. func NewBloomIndexer(db ethdb.Database, size, confReq uint64) *core.ChainIndexer { backend := &BloomIndexer{ - db: db, - size: size, + db: db, + size: size, + quit: make(chan struct{}), + locked: make(chan struct{}, 1), } table := ethdb.NewTable(db, string(rawdb.BloomBitsIndexPrefix)) @@ -116,6 +117,13 @@ func NewBloomIndexer(db ethdb.Database, size, confReq uint64) *core.ChainIndexer // Reset implements core.ChainIndexerBackend, starting a new bloombits index // section. func (b *BloomIndexer) Reset(section uint64, lastSectionHead common.Hash) error { + select { + case b.locked <- struct{}{}: + defer func() { <-b.locked }() + case <-b.quit: + return core.ErrIndexerBackendClosed + } + gen, err := bloombits.NewGenerator(uint(b.size)) b.gen, b.section, b.head = gen, section, common.Hash{} return err @@ -123,16 +131,30 @@ func (b *BloomIndexer) Reset(section uint64, lastSectionHead common.Hash) error // Process implements core.ChainIndexerBackend, adding a new header's bloom into // the index. -func (b *BloomIndexer) Process(header *types.Header) { +func (b *BloomIndexer) Process(header *types.Header) error { + select { + case b.locked <- struct{}{}: + defer func() { <-b.locked }() + case <-b.quit: + return core.ErrIndexerBackendClosed + } + b.gen.AddBloom(uint(header.Number.Uint64()-b.section*b.size), header.Bloom) b.head = header.Hash() + return nil } // Commit implements core.ChainIndexerBackend, finalizing the bloom section and // writing it out into the database. func (b *BloomIndexer) Commit() error { - batch := b.db.NewBatch() + select { + case b.locked <- struct{}{}: + defer func() { <-b.locked }() + case <-b.quit: + return core.ErrIndexerBackendClosed + } + batch := b.db.NewBatch() for i := 0; i < types.BloomBitLength; i++ { bits, err := b.gen.Bitset(uint(i)) if err != nil { @@ -143,5 +165,8 @@ func (b *BloomIndexer) Commit() error { return batch.Write() } -// Cancel implements core.ChainIndexerBackend -func (b *BloomIndexer) Closing() {} +// Close implements core.ChainIndexerBackend +func (b *BloomIndexer) Close() { + close(b.quit) + b.locked <- struct{}{} +} diff --git a/light/postprocess.go b/light/postprocess.go index 007eded2f1..c23d410879 100644 --- a/light/postprocess.go +++ b/light/postprocess.go @@ -127,7 +127,7 @@ type ChtIndexerBackend struct { section, sectionSize uint64 lastHash common.Hash trie *trie.Trie - quit chan struct{} + quit, locked chan struct{} } // NewBloomTrieIndexer creates a BloomTrie chain indexer @@ -149,6 +149,7 @@ func NewChtIndexer(db ethdb.Database, clientMode bool, odr OdrBackend) *core.Cha triedb: trie.NewDatabase(trieTable), sectionSize: sectionSize, quit: make(chan struct{}), + locked: make(chan struct{}, 1), } return core.NewChainIndexer(db, idb, backend, sectionSize, confirmReq, time.Millisecond*100, "cht") } @@ -187,11 +188,21 @@ func (c *ChtIndexerBackend) fetchMissingNodes(section uint64, root common.Hash) r.Proof.Store(batch) err = batch.Write() } + if err == ctx.Err() { + return core.ErrIndexerBackendClosed + } return err } // Reset implements core.ChainIndexerBackend func (c *ChtIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) error { + select { + case c.locked <- struct{}{}: + defer func() { <-c.locked }() + case <-c.quit: + return core.ErrIndexerBackendClosed + } + var root common.Hash if section > 0 { root = GetChtRoot(c.diskdb, section-1, lastSectionHead) @@ -211,7 +222,14 @@ func (c *ChtIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) e } // Process implements core.ChainIndexerBackend -func (c *ChtIndexerBackend) Process(header *types.Header) { +func (c *ChtIndexerBackend) Process(header *types.Header) error { + select { + case c.locked <- struct{}{}: + defer func() { <-c.locked }() + case <-c.quit: + return core.ErrIndexerBackendClosed + } + hash, num := header.Hash(), header.Number.Uint64() c.lastHash = hash @@ -223,10 +241,18 @@ func (c *ChtIndexerBackend) Process(header *types.Header) { binary.BigEndian.PutUint64(encNumber[:], num) data, _ := rlp.EncodeToBytes(ChtNode{hash, td}) c.trie.Update(encNumber[:], data) + return nil } // Commit implements core.ChainIndexerBackend func (c *ChtIndexerBackend) Commit() error { + select { + case c.locked <- struct{}{}: + defer func() { <-c.locked }() + case <-c.quit: + return core.ErrIndexerBackendClosed + } + root, err := c.trie.Commit(nil) if err != nil { return err @@ -240,9 +266,10 @@ func (c *ChtIndexerBackend) Commit() error { return nil } -// Cancel implements core.ChainIndexerBackend -func (c *ChtIndexerBackend) Closing() { +// Close implements core.ChainIndexerBackend +func (c *ChtIndexerBackend) Close() { close(c.quit) + c.locked <- struct{}{} } const ( @@ -278,7 +305,7 @@ type BloomTrieIndexerBackend struct { section, parentSectionSize, bloomTrieRatio uint64 trie *trie.Trie sectionHeads []common.Hash - quit chan struct{} + quit, locked chan struct{} } // NewBloomTrieIndexer creates a BloomTrie chain indexer @@ -290,6 +317,7 @@ func NewBloomTrieIndexer(db ethdb.Database, clientMode bool, odr OdrBackend) *co trieTable: trieTable, triedb: trie.NewDatabase(trieTable), quit: make(chan struct{}), + locked: make(chan struct{}, 1), } idb := ethdb.NewTable(db, "bltIndex-") @@ -359,6 +387,9 @@ func (b *BloomTrieIndexerBackend) fetchMissingNodes(section uint64, root common. for i := uint(0); i < types.BloomBitLength; i++ { res := <-resCh if res.err != nil { + if res.err == ctx.Err() { + return core.ErrIndexerBackendClosed + } return res.err } res.nodes.Store(batch) @@ -368,6 +399,13 @@ func (b *BloomTrieIndexerBackend) fetchMissingNodes(section uint64, root common. // Reset implements core.ChainIndexerBackend func (b *BloomTrieIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) error { + select { + case b.locked <- struct{}{}: + defer func() { <-b.locked }() + case <-b.quit: + return core.ErrIndexerBackendClosed + } + var root common.Hash if section > 0 { root = GetBloomTrieRoot(b.diskdb, section-1, lastSectionHead) @@ -385,15 +423,30 @@ func (b *BloomTrieIndexerBackend) Reset(section uint64, lastSectionHead common.H } // Process implements core.ChainIndexerBackend -func (b *BloomTrieIndexerBackend) Process(header *types.Header) { +func (b *BloomTrieIndexerBackend) Process(header *types.Header) error { + select { + case b.locked <- struct{}{}: + defer func() { <-b.locked }() + case <-b.quit: + return core.ErrIndexerBackendClosed + } + num := header.Number.Uint64() - b.section*BloomTrieFrequency if (num+1)%b.parentSectionSize == 0 { b.sectionHeads[num/b.parentSectionSize] = header.Hash() } + return nil } // Commit implements core.ChainIndexerBackend func (b *BloomTrieIndexerBackend) Commit() error { + select { + case b.locked <- struct{}{}: + defer func() { <-b.locked }() + case <-b.quit: + return core.ErrIndexerBackendClosed + } + var compSize, decompSize uint64 for i := uint(0); i < types.BloomBitLength; i++ { @@ -435,7 +488,8 @@ func (b *BloomTrieIndexerBackend) Commit() error { return nil } -// Cancel implements core.ChainIndexerBackend -func (b *BloomTrieIndexerBackend) Closing() { +// Close implements core.ChainIndexerBackend +func (b *BloomTrieIndexerBackend) Close() { close(b.quit) + b.locked <- struct{}{} }