core, light: simplified cancel mechanism

core/chain_indexer.go

@@ -17,8 +17,8 @@
 package core
 
 import (
+	"context"
 	"encoding/binary"
-	"errors"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -38,24 +38,16 @@ import (
 type ChainIndexerBackend interface {
 	// Reset initiates the processing of a new chain segment, potentially terminating
 	// any partially completed operations (in case of a reorg).
-	Reset(section uint64, prevHead common.Hash) error
+	Reset(ctx context.Context, section uint64, prevHead common.Hash) error
 
 	// Process crunches through the next header in the chain segment. The caller
 	// will ensure a sequential order of headers.
-	Process(header *types.Header) error
+	Process(ctx context.Context, header *types.Header) error
 
 	// Commit finalizes the section metadata and stores it into the database.
 	Commit() error
-
-	// 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.
@@ -83,6 +75,8 @@ type ChainIndexer struct {
 	active    uint32          // Flag whether the event loop was started
 	update    chan struct{}   // Notification channel that headers should be processed
 	quit      chan chan error // Quit channel to tear down running goroutines
+	ctx       context.Context
+	ctxCancel func()
 
 	sectionSize uint64 // Number of blocks in a single chain segment to process
 	confirmsReq uint64 // Number of confirmations before processing a completed segment
@@ -114,6 +108,8 @@ func NewChainIndexer(chainDb, indexDb ethdb.Database, backend ChainIndexerBacken
 	}
 	// Initialize database dependent fields and start the updater
 	c.loadValidSections()
+	c.ctx, c.ctxCancel = context.WithCancel(context.Background())
+
 	go c.updateLoop()
 
 	return c
@@ -147,7 +143,7 @@ func (c *ChainIndexer) Start(chain ChainIndexerChain) {
 func (c *ChainIndexer) Close() error {
 	var errs []error
 
-	c.backend.Close()
+	c.ctxCancel()
 
 	// Tear down the primary update loop
 	errc := make(chan error)
@@ -308,8 +304,11 @@ func (c *ChainIndexer) updateLoop() {
 				c.lock.Unlock()
 				newHead, err := c.processSection(section, oldHead)
 				if err != nil {
-					if err == ErrIndexerBackendClosed {
+					select {
-						continue
+					case <-c.ctx.Done():
+						<-c.quit <- nil
+						return
+					default:
 					}
 					c.log.Error("Section processing failed", "error", err)
 				}
@@ -358,7 +357,7 @@ func (c *ChainIndexer) processSection(section uint64, lastHead common.Hash) (com
 
 	// Reset and partial processing
 
-	if err := c.backend.Reset(section, lastHead); err != nil {
+	if err := c.backend.Reset(c.ctx, section, lastHead); err != nil {
 		c.setValidSections(0)
 		return common.Hash{}, err
 	}
@@ -374,7 +373,7 @@ 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")
 		}
-		if err := c.backend.Process(header); err != nil {
+		if err := c.backend.Process(c.ctx, header); err != nil {
 			return common.Hash{}, err
 		}
 		lastHead = header.Hash()

core/chain_indexer_test.go

@@ -17,6 +17,7 @@
 package core
 
 import (
+	"context"
 	"fmt"
 	"math/big"
 	"math/rand"
@@ -210,13 +211,13 @@ func (b *testChainIndexBackend) reorg(headNum uint64) uint64 {
 	return b.stored * b.indexer.sectionSize
 }
 
-func (b *testChainIndexBackend) Reset(section uint64, prevHead common.Hash) error {
+func (b *testChainIndexBackend) Reset(ctx context.Context, section uint64, prevHead common.Hash) error {
 	b.section = section
 	b.headerCnt = 0
 	return nil
 }
 
-func (b *testChainIndexBackend) Process(header *types.Header) error {
+func (b *testChainIndexBackend) Process(ctx context.Context, header *types.Header) error {
 	b.headerCnt++
 	if b.headerCnt > b.indexer.sectionSize {
 		b.t.Error("Processing too many headers")
@@ -236,5 +237,3 @@ func (b *testChainIndexBackend) Commit() error {
 	}
 	return nil
 }
-
-func (b *testChainIndexBackend) Close() {}

eth/bloombits.go

@@ -17,6 +17,7 @@
 package eth
 
 import (
+	"context"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -97,7 +98,6 @@ type BloomIndexer struct {
 	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
@@ -106,8 +106,6 @@ 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,14 +114,7 @@ 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 {
+func (b *BloomIndexer) Reset(ctx context.Context, 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
@@ -131,14 +122,7 @@ 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) error {
+func (b *BloomIndexer) Process(ctx context.Context, 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
@@ -147,13 +131,6 @@ func (b *BloomIndexer) Process(header *types.Header) error {
 // Commit implements core.ChainIndexerBackend, finalizing the bloom section and
 // writing it out into the database.
 func (b *BloomIndexer) Commit() error {
-	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))
@@ -164,9 +141,3 @@ func (b *BloomIndexer) Commit() error {
 	}
 	return batch.Write()
 }
-
-// Close implements core.ChainIndexerBackend
-func (b *BloomIndexer) Close() {
-	close(b.quit)
-	b.locked <- struct{}{}
-}

light/postprocess.go

@@ -127,7 +127,6 @@ type ChtIndexerBackend struct {
 	section, sectionSize uint64
 	lastHash             common.Hash
 	trie                 *trie.Trie
-	quit, locked         chan struct{}
 }
 
 // NewBloomTrieIndexer creates a BloomTrie chain indexer
@@ -148,26 +147,13 @@ func NewChtIndexer(db ethdb.Database, clientMode bool, odr OdrBackend) *core.Cha
 		trieTable:   trieTable,
 		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")
 }
 
 // fetchMissingNodes tries to retrieve the last entry of the latest trusted CHT from the
 // ODR backend in order to be able to add new entries and calculate subsequent root hashes
-func (c *ChtIndexerBackend) fetchMissingNodes(section uint64, root common.Hash) error {
+func (c *ChtIndexerBackend) fetchMissingNodes(ctx context.Context, section uint64, root common.Hash) error {
-	ctx, cancel := context.WithCancel(context.Background())
-	done := make(chan struct{})
-	go func() {
-		select {
-		case <-done:
-		case <-c.quit:
-		}
-		cancel()
-	}()
-	defer close(done)
-
 	batch := c.trieTable.NewBatch()
 	r := &ChtRequest{ChtRoot: root, ChtNum: section - 1, BlockNum: section*c.sectionSize - 1}
 	var err error
@@ -176,8 +162,8 @@ func (c *ChtIndexerBackend) fetchMissingNodes(section uint64, root common.Hash)
 		if err == ErrNoPeers {
 			// if there are no peers to serve, retry later
 			select {
-			case <-c.quit:
+			case <-ctx.Done():
-				return fmt.Errorf("Section processing cancelled")
+				return ctx.Err()
 			case <-time.After(time.Second * 10):
 			}
 		} else {
@@ -188,21 +174,11 @@ 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 {
+func (c *ChtIndexerBackend) Reset(ctx context.Context, 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 +187,7 @@ func (c *ChtIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) e
 	c.trie, err = trie.New(root, c.triedb)
 
 	if err != nil && c.odr != nil {
-		err = c.fetchMissingNodes(section, root)
+		err = c.fetchMissingNodes(ctx, section, root)
 		if err == nil {
 			c.trie, err = trie.New(root, c.triedb)
 		}
@@ -222,14 +198,7 @@ func (c *ChtIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) e
 }
 
 // Process implements core.ChainIndexerBackend
-func (c *ChtIndexerBackend) Process(header *types.Header) error {
+func (c *ChtIndexerBackend) Process(ctx context.Context, 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
 
@@ -246,13 +215,6 @@ func (c *ChtIndexerBackend) Process(header *types.Header) error {
 
 // 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
@@ -266,12 +228,6 @@ func (c *ChtIndexerBackend) Commit() error {
 	return nil
 }
 
-// Close implements core.ChainIndexerBackend
-func (c *ChtIndexerBackend) Close() {
-	close(c.quit)
-	c.locked <- struct{}{}
-}
-
 const (
 	BloomTrieFrequency  = 32768
 	ethBloomBitsSection = 4096
@@ -305,7 +261,6 @@ type BloomTrieIndexerBackend struct {
 	section, parentSectionSize, bloomTrieRatio uint64
 	trie                                       *trie.Trie
 	sectionHeads                               []common.Hash
-	quit, locked                               chan struct{}
 }
 
 // NewBloomTrieIndexer creates a BloomTrie chain indexer
@@ -316,8 +271,6 @@ func NewBloomTrieIndexer(db ethdb.Database, clientMode bool, odr OdrBackend) *co
 		odr:       odr,
 		trieTable: trieTable,
 		triedb:    trie.NewDatabase(trieTable),
-		quit:      make(chan struct{}),
-		locked:    make(chan struct{}, 1),
 	}
 	idb := ethdb.NewTable(db, "bltIndex-")
 
@@ -333,18 +286,7 @@ func NewBloomTrieIndexer(db ethdb.Database, clientMode bool, odr OdrBackend) *co
 
 // fetchMissingNodes tries to retrieve the last entries of the latest trusted bloom trie from the
 // ODR backend in order to be able to add new entries and calculate subsequent root hashes
-func (b *BloomTrieIndexerBackend) fetchMissingNodes(section uint64, root common.Hash) error {
+func (b *BloomTrieIndexerBackend) fetchMissingNodes(ctx context.Context, section uint64, root common.Hash) error {
-	ctx, cancel := context.WithCancel(context.Background())
-	done := make(chan struct{})
-	go func() {
-		select {
-		case <-done:
-		case <-b.quit:
-		}
-		cancel()
-	}()
-	defer close(done)
-
 	indexCh := make(chan uint, types.BloomBitLength)
 	type res struct {
 		nodes *NodeSet
@@ -365,8 +307,8 @@ func (b *BloomTrieIndexerBackend) fetchMissingNodes(section uint64, root common.
 					if err == ErrNoPeers {
 						// if there are no peers to serve, retry later
 						select {
-						case <-b.quit:
+						case <-ctx.Done():
-							resCh <- res{nil, fmt.Errorf("Section processing cancelled")}
+							resCh <- res{nil, ctx.Err()}
 							return
 						case <-time.After(time.Second * 10):
 						}
@@ -387,9 +329,6 @@ 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)
@@ -398,14 +337,7 @@ func (b *BloomTrieIndexerBackend) fetchMissingNodes(section uint64, root common.
 }
 
 // Reset implements core.ChainIndexerBackend
-func (b *BloomTrieIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) error {
+func (b *BloomTrieIndexerBackend) Reset(ctx context.Context, 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)
@@ -413,7 +345,7 @@ func (b *BloomTrieIndexerBackend) Reset(section uint64, lastSectionHead common.H
 	var err error
 	b.trie, err = trie.New(root, b.triedb)
 	if err != nil && b.odr != nil {
-		err = b.fetchMissingNodes(section, root)
+		err = b.fetchMissingNodes(ctx, section, root)
 		if err == nil {
 			b.trie, err = trie.New(root, b.triedb)
 		}
@@ -423,14 +355,7 @@ func (b *BloomTrieIndexerBackend) Reset(section uint64, lastSectionHead common.H
 }
 
 // Process implements core.ChainIndexerBackend
-func (b *BloomTrieIndexerBackend) Process(header *types.Header) error {
+func (b *BloomTrieIndexerBackend) Process(ctx context.Context, 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()
@@ -440,13 +365,6 @@ func (b *BloomTrieIndexerBackend) Process(header *types.Header) error {
 
 // 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++ {
@@ -487,9 +405,3 @@ func (b *BloomTrieIndexerBackend) Commit() error {
 
 	return nil
 }
-
-// Close implements core.ChainIndexerBackend
-func (b *BloomTrieIndexerBackend) Close() {
-	close(b.quit)
-	b.locked <- struct{}{}
-}