core: indexer backend interface changed

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

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() {}

eth/bloombits.go

@@ -93,12 +93,11 @@ const (
 // 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
+	quit, locked chan struct{}
 }
 
 // NewBloomIndexer returns a chain indexer that generates bloom bits data for the
@@ -107,6 +106,8 @@ func NewBloomIndexer(db ethdb.Database, size, confReq uint64) *core.ChainIndexer
 	backend := &BloomIndexer{
 		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{}{}
+}

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{}{}
 }