From 7a49759c3eb8bbd292f3612a4b58016f0d7de6bf Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Wed, 15 Aug 2018 17:47:20 +0200 Subject: [PATCH] core, light: simplified cancel mechanism --- core/chain_indexer.go | 37 ++++++------ core/chain_indexer_test.go | 7 +-- eth/bloombits.go | 49 ++++------------ light/postprocess.go | 112 ++++--------------------------------- 4 files changed, 43 insertions(+), 162 deletions(-) diff --git a/core/chain_indexer.go b/core/chain_indexer.go index a66dbadd00..11a7c96fa0 100644 --- a/core/chain_indexer.go +++ b/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. @@ -80,9 +72,11 @@ type ChainIndexer struct { backend ChainIndexerBackend // Background processor generating the index data content children []*ChainIndexer // Child indexers to cascade chain updates to - 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 + 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 { - continue + select { + 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() diff --git a/core/chain_indexer_test.go b/core/chain_indexer_test.go index d595ee7da9..a029dec626 100644 --- a/core/chain_indexer_test.go +++ b/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() {} diff --git a/eth/bloombits.go b/eth/bloombits.go index f55f73bd32..eb18565e2c 100644 --- a/eth/bloombits.go +++ b/eth/bloombits.go @@ -17,6 +17,7 @@ package eth import ( + "context" "time" "github.com/ethereum/go-ethereum/common" @@ -92,22 +93,19 @@ 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 - quit, locked chan 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 } // 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, - quit: make(chan struct{}), - locked: make(chan struct{}, 1), + db: db, + size: size, } 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 { - select { - case b.locked <- struct{}{}: - defer func() { <-b.locked }() - case <-b.quit: - return core.ErrIndexerBackendClosed - } - +func (b *BloomIndexer) Reset(ctx context.Context, section uint64, lastSectionHead common.Hash) error { 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 { - select { - case b.locked <- struct{}{}: - defer func() { <-b.locked }() - case <-b.quit: - return core.ErrIndexerBackendClosed - } - +func (b *BloomIndexer) Process(ctx context.Context, header *types.Header) error { 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{}{} -} diff --git a/light/postprocess.go b/light/postprocess.go index c23d410879..41f7dba5c4 100644 --- a/light/postprocess.go +++ b/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 { - ctx, cancel := context.WithCancel(context.Background()) - done := make(chan struct{}) - go func() { - select { - case <-done: - case <-c.quit: - } - cancel() - }() - defer close(done) - +func (c *ChtIndexerBackend) fetchMissingNodes(ctx context.Context, section uint64, root common.Hash) error { 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: - return fmt.Errorf("Section processing cancelled") + case <-ctx.Done(): + 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 { - select { - case c.locked <- struct{}{}: - defer func() { <-c.locked }() - case <-c.quit: - return core.ErrIndexerBackendClosed - } - +func (c *ChtIndexerBackend) Reset(ctx context.Context, section uint64, lastSectionHead common.Hash) error { 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 { - select { - case c.locked <- struct{}{}: - defer func() { <-c.locked }() - case <-c.quit: - return core.ErrIndexerBackendClosed - } - +func (c *ChtIndexerBackend) Process(ctx context.Context, header *types.Header) error { 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 { - ctx, cancel := context.WithCancel(context.Background()) - done := make(chan struct{}) - go func() { - select { - case <-done: - case <-b.quit: - } - cancel() - }() - defer close(done) - +func (b *BloomTrieIndexerBackend) fetchMissingNodes(ctx context.Context, section uint64, root common.Hash) error { 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: - resCh <- res{nil, fmt.Errorf("Section processing cancelled")} + case <-ctx.Done(): + 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 { - select { - case b.locked <- struct{}{}: - defer func() { <-b.locked }() - case <-b.quit: - return core.ErrIndexerBackendClosed - } - +func (b *BloomTrieIndexerBackend) Reset(ctx context.Context, section uint64, lastSectionHead common.Hash) error { 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 { - select { - case b.locked <- struct{}{}: - defer func() { <-b.locked }() - case <-b.quit: - return core.ErrIndexerBackendClosed - } - +func (b *BloomTrieIndexerBackend) Process(ctx context.Context, header *types.Header) error { 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{}{} -}