light: CHT/BBT indexers working in light mode

core/chain_indexer.go

@@ -45,6 +45,17 @@ type ChainIndexerBackend interface {
 
 	// 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()
 }
 
 // ChainIndexerChain interface is used for connecting the indexer to a blockchain
@@ -138,6 +149,8 @@ func (c *ChainIndexer) Start(chain ChainIndexerChain) {
 func (c *ChainIndexer) Close() error {
 	var errs []error
 
+	c.backend.Closing()
+
 	// Tear down the primary update loop
 	errc := make(chan error)
 	c.quit <- errc

core/chain_indexer_test.go

@@ -235,3 +235,5 @@ func (b *testChainIndexBackend) Commit() error {
 	}
 	return nil
 }
+
+func (b *testChainIndexBackend) Closing() {}

eth/backend.go

@@ -130,7 +130,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 		gasPrice:       config.GasPrice,
 		etherbase:      config.Etherbase,
 		bloomRequests:  make(chan chan *bloombits.Retrieval),
-		bloomIndexer:   NewBloomIndexer(chainDb, params.BloomBitsBlocks),
+		bloomIndexer:   NewBloomIndexer(chainDb, params.BloomBitsBlocks, bloomConfirms),
 	}
 
 	log.Info("Initialising Ethereum protocol", "versions", ProtocolVersions, "network", config.NetworkId)

eth/bloombits.go

@@ -103,14 +103,14 @@ type BloomIndexer 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 uint64) *core.ChainIndexer {
+func NewBloomIndexer(db ethdb.Database, size, confReq uint64) *core.ChainIndexer {
 	backend := &BloomIndexer{
 		db:   db,
 		size: size,
 	}
 	table := ethdb.NewTable(db, string(rawdb.BloomBitsIndexPrefix))
 
-	return core.NewChainIndexer(db, table, backend, size, bloomConfirms, bloomThrottling, "bloombits")
+	return core.NewChainIndexer(db, table, backend, size, confReq, bloomThrottling, "bloombits")
 }
 
 // Reset implements core.ChainIndexerBackend, starting a new bloombits index
@@ -142,3 +142,6 @@ func (b *BloomIndexer) Commit() error {
 	}
 	return batch.Write()
 }
+
+// Cancel implements core.ChainIndexerBackend
+func (b *BloomIndexer) Closing() {}

les/backend.go

@@ -106,18 +106,24 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
 		shutdownChan:   make(chan bool),
 		networkId:      config.NetworkId,
 		bloomRequests:  make(chan chan *bloombits.Retrieval),
-		bloomIndexer:     eth.NewBloomIndexer(chainDb, light.BloomTrieFrequency),
+		bloomIndexer:   eth.NewBloomIndexer(chainDb, light.BloomTrieFrequency, light.HelperTrieConfirmations),
-		chtIndexer:       light.NewChtIndexer(chainDb, true),
-		bloomTrieIndexer: light.NewBloomTrieIndexer(chainDb, true),
 	}
 
 	leth.relay = NewLesTxRelay(peers, leth.reqDist)
 	leth.serverPool = newServerPool(chainDb, quitSync, &leth.wg)
 	leth.retriever = newRetrieveManager(peers, leth.reqDist, leth.serverPool)
-	leth.odr = NewLesOdr(chainDb, leth.chtIndexer, leth.bloomTrieIndexer, leth.bloomIndexer, leth.retriever)
+	leth.odr = NewLesOdr(chainDb, leth.retriever)
+	leth.chtIndexer = light.NewChtIndexer(chainDb, true, leth.odr)
+	leth.bloomTrieIndexer = light.NewBloomTrieIndexer(chainDb, true, leth.odr)
+	leth.odr.SetIndexers(leth.chtIndexer, leth.bloomTrieIndexer, leth.bloomIndexer)
+	// Note: NewLightChain adds the trusted checkpoint so it needs an ODR with
+	// indexers already set but not started yet
 	if leth.blockchain, err = light.NewLightChain(leth.odr, leth.chainConfig, leth.engine); err != nil {
 		return nil, err
 	}
+	// Note: AddChildIndexer starts the update process for the child
+	leth.bloomIndexer.AddChildIndexer(leth.bloomTrieIndexer)
+	leth.chtIndexer.Start(leth.blockchain)
 	leth.bloomIndexer.Start(leth.blockchain)
 	// Rewind the chain in case of an incompatible config upgrade.
 	if compat, ok := genesisErr.(*params.ConfigCompatError); ok {
@@ -242,9 +248,6 @@ func (s *LightEthereum) Stop() error {
 	if s.chtIndexer != nil {
 		s.chtIndexer.Close()
 	}
-	if s.bloomTrieIndexer != nil {
-		s.bloomTrieIndexer.Close()
-	}
 	s.blockchain.Stop()
 	s.protocolManager.Stop()
 	s.txPool.Stop()

les/distributor.go

@@ -20,14 +20,10 @@ package les
 
 import (
 	"container/list"
-	"errors"
 	"sync"
 	"time"
 )
 
-// ErrNoPeers is returned if no peers capable of serving a queued request are available
-var ErrNoPeers = errors.New("no suitable peers available")
-
 // requestDistributor implements a mechanism that distributes requests to
 // suitable peers, obeying flow control rules and prioritizing them in creation
 // order (even when a resend is necessary).

les/handler.go

@@ -1258,7 +1258,7 @@ func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, s
 	}
 	_, ok := <-pc.manager.reqDist.queue(rq)
 	if !ok {
-		return ErrNoPeers
+		return light.ErrNoPeers
 	}
 	return nil
 }
@@ -1282,7 +1282,7 @@ func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip
 	}
 	_, ok := <-pc.manager.reqDist.queue(rq)
 	if !ok {
-		return ErrNoPeers
+		return light.ErrNoPeers
 	}
 	return nil
 }

les/helper_test.go

@@ -156,12 +156,12 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
 	} else {
 		blockchain, _ := core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{})
 
-		chtIndexer := light.NewChtIndexer(db, false)
+		chtIndexer := light.NewChtIndexer(db, false, nil)
 		chtIndexer.Start(blockchain)
 
-		bbtIndexer := light.NewBloomTrieIndexer(db, false)
+		bbtIndexer := light.NewBloomTrieIndexer(db, false, nil)
 
-		bloomIndexer := eth.NewBloomIndexer(db, params.BloomBitsBlocks)
+		bloomIndexer := eth.NewBloomIndexer(db, params.BloomBitsBlocks, light.HelperTrieProcessConfirmations)
 		bloomIndexer.AddChildIndexer(bbtIndexer)
 		bloomIndexer.Start(blockchain)
 

les/odr.go

@@ -33,12 +33,9 @@ type LesOdr struct {
 	stop                                       chan struct{}
 }
 
-func NewLesOdr(db ethdb.Database, chtIndexer, bloomTrieIndexer, bloomIndexer *core.ChainIndexer, retriever *retrieveManager) *LesOdr {
+func NewLesOdr(db ethdb.Database, retriever *retrieveManager) *LesOdr {
 	return &LesOdr{
 		db:        db,
-		chtIndexer:       chtIndexer,
-		bloomTrieIndexer: bloomTrieIndexer,
-		bloomIndexer:     bloomIndexer,
 		retriever: retriever,
 		stop:      make(chan struct{}),
 	}
@@ -54,6 +51,13 @@ func (odr *LesOdr) Database() ethdb.Database {
 	return odr.db
 }
 
+// SetIndexers adds the necessary chain indexers to the ODR backend
+func (odr *LesOdr) SetIndexers(chtIndexer, bloomTrieIndexer, bloomIndexer *core.ChainIndexer) {
+	odr.chtIndexer = chtIndexer
+	odr.bloomTrieIndexer = bloomTrieIndexer
+	odr.bloomIndexer = bloomIndexer
+}
+
 // ChtIndexer returns the CHT chain indexer
 func (odr *LesOdr) ChtIndexer() *core.ChainIndexer {
 	return odr.chtIndexer

les/odr_test.go

@@ -167,7 +167,8 @@ func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) {
 	rm := newRetrieveManager(peers, dist, nil)
 	db := ethdb.NewMemDatabase()
 	ldb := ethdb.NewMemDatabase()
-	odr := NewLesOdr(ldb, light.NewChtIndexer(db, true), light.NewBloomTrieIndexer(db, true), eth.NewBloomIndexer(db, light.BloomTrieFrequency), rm)
+	odr := NewLesOdr(ldb, rm)
+	odr.SetIndexers(light.NewChtIndexer(db, true, nil), light.NewBloomTrieIndexer(db, true, nil), eth.NewBloomIndexer(db, light.BloomTrieFrequency, light.HelperTrieConfirmations))
 	pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
 	lpm := newTestProtocolManagerMust(t, true, 0, nil, peers, odr, ldb)
 	_, err1, lpeer, err2 := newTestPeerPair("peer", protocol, pm, lpm)

les/request_test.go

@@ -89,7 +89,8 @@ func testAccess(t *testing.T, protocol int, fn accessTestFn) {
 	rm := newRetrieveManager(peers, dist, nil)
 	db := ethdb.NewMemDatabase()
 	ldb := ethdb.NewMemDatabase()
-	odr := NewLesOdr(ldb, light.NewChtIndexer(db, true), light.NewBloomTrieIndexer(db, true), eth.NewBloomIndexer(db, light.BloomTrieFrequency), rm)
+	odr := NewLesOdr(ldb, rm)
+	odr.SetIndexers(light.NewChtIndexer(db, true, nil), light.NewBloomTrieIndexer(db, true, nil), eth.NewBloomIndexer(db, light.BloomTrieFrequency, light.HelperTrieConfirmations))
 
 	pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
 	lpm := newTestProtocolManagerMust(t, true, 0, nil, peers, odr, ldb)

les/retrieve.go

@@ -27,6 +27,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common/mclock"
+	"github.com/ethereum/go-ethereum/light"
 )
 
 var (
@@ -207,7 +208,7 @@ func (r *sentReq) stateRequesting() reqStateFn {
 					return r.stateNoMorePeers
 				}
 				// nothing to wait for, no more peers to ask, return with error
-				r.stop(ErrNoPeers)
+				r.stop(light.ErrNoPeers)
 				// no need to go to stopped state because waiting() already returned false
 				return nil
 			}

les/server.go

@@ -67,8 +67,8 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
 		protocolManager:  pm,
 		quitSync:         quitSync,
 		lesTopics:        lesTopics,
-		chtIndexer:       light.NewChtIndexer(eth.ChainDb(), false),
+		chtIndexer:       light.NewChtIndexer(eth.ChainDb(), false, nil),
-		bloomTrieIndexer: light.NewBloomTrieIndexer(eth.ChainDb(), false),
+		bloomTrieIndexer: light.NewBloomTrieIndexer(eth.ChainDb(), false, nil),
 	}
 	logger := log.New()
 

light/odr.go

@@ -20,6 +20,7 @@ package light
 
 import (
 	"context"
+	"errors"
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -33,6 +34,9 @@ import (
 // service is not required.
 var NoOdr = context.Background()
 
+// ErrNoPeers is returned if no peers capable of serving a queued request are available
+var ErrNoPeers = errors.New("no suitable peers available")
+
 // OdrBackend is an interface to a backend service that handles ODR retrievals type
 type OdrBackend interface {
 	Database() ethdb.Database

light/postprocess.go

@@ -17,8 +17,10 @@
 package light
 
 import (
+	"context"
 	"encoding/binary"
 	"errors"
+	"fmt"
 	"math/big"
 	"time"
 
@@ -119,15 +121,17 @@ func StoreChtRoot(db ethdb.Database, sectionIdx uint64, sectionHead, root common
 
 // ChtIndexerBackend implements core.ChainIndexerBackend
 type ChtIndexerBackend struct {
-	diskdb               ethdb.Database
+	diskdb, trieTable    ethdb.Database
+	odr                  OdrBackend
 	triedb               *trie.Database
 	section, sectionSize uint64
 	lastHash             common.Hash
 	trie                 *trie.Trie
+	quit                 chan struct{}
 }
 
 // NewBloomTrieIndexer creates a BloomTrie chain indexer
-func NewChtIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer {
+func NewChtIndexer(db ethdb.Database, clientMode bool, odr OdrBackend) *core.ChainIndexer {
 	var sectionSize, confirmReq uint64
 	if clientMode {
 		sectionSize = CHTFrequencyClient
@@ -137,14 +141,55 @@ func NewChtIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer {
 		confirmReq = HelperTrieProcessConfirmations
 	}
 	idb := ethdb.NewTable(db, "chtIndex-")
+	trieTable := ethdb.NewTable(db, ChtTablePrefix)
 	backend := &ChtIndexerBackend{
 		diskdb:      db,
-		triedb:      trie.NewDatabase(ethdb.NewTable(db, ChtTablePrefix)),
+		odr:         odr,
+		trieTable:   trieTable,
+		triedb:      trie.NewDatabase(trieTable),
 		sectionSize: sectionSize,
+		quit:        make(chan struct{}),
 	}
 	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)
+
+	batch := c.trieTable.NewBatch()
+	r := &ChtRequest{ChtRoot: root, ChtNum: section - 1, BlockNum: section*c.sectionSize - 1}
+	var err error
+	for {
+		err = c.odr.Retrieve(ctx, r)
+		if err == ErrNoPeers {
+			// if there are no peers to serve, retry later
+			select {
+			case <-c.quit:
+				return fmt.Errorf("Section processing cancelled")
+			case <-time.After(time.Second * 10):
+			}
+		} else {
+			break
+		}
+	}
+	if err == nil {
+		r.Proof.Store(batch)
+		err = batch.Write()
+	}
+	return err
+}
+
 // Reset implements core.ChainIndexerBackend
 func (c *ChtIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) error {
 	var root common.Hash
@@ -153,6 +198,14 @@ func (c *ChtIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) e
 	}
 	var err error
 	c.trie, err = trie.New(root, c.triedb)
+
+	if err != nil && c.odr != nil {
+		err = c.fetchMissingNodes(section, root)
+		if err == nil {
+			c.trie, err = trie.New(root, c.triedb)
+		}
+	}
+
 	c.section = section
 	return err
 }
@@ -181,16 +234,20 @@ func (c *ChtIndexerBackend) Commit() error {
 	c.triedb.Commit(root, false)
 
 	if ((c.section+1)*c.sectionSize)%CHTFrequencyClient == 0 {
-		log.Info("Storing CHT", "section", c.section*c.sectionSize/CHTFrequencyClient, "head", c.lastHash, "root", root)
+		log.Info("Storing CHT", "section", c.section*c.sectionSize/CHTFrequencyClient, "head", fmt.Sprintf("%064x", c.lastHash), "root", fmt.Sprintf("%064x", root))
 	}
 	StoreChtRoot(c.diskdb, c.section, c.lastHash, root)
 	return nil
 }
 
+// Cancel implements core.ChainIndexerBackend
+func (c *ChtIndexerBackend) Closing() {
+	close(c.quit)
+}
+
 const (
 	BloomTrieFrequency  = 32768
 	ethBloomBitsSection = 4096
-	ethBloomBitsConfirmations = 256
 )
 
 var (
@@ -215,32 +272,98 @@ func StoreBloomTrieRoot(db ethdb.Database, sectionIdx uint64, sectionHead, root
 
 // BloomTrieIndexerBackend implements core.ChainIndexerBackend
 type BloomTrieIndexerBackend struct {
-	diskdb                                     ethdb.Database
+	diskdb, trieTable                          ethdb.Database
+	odr                                        OdrBackend
 	triedb                                     *trie.Database
 	section, parentSectionSize, bloomTrieRatio uint64
 	trie                                       *trie.Trie
 	sectionHeads                               []common.Hash
+	quit                                       chan struct{}
 }
 
 // NewBloomTrieIndexer creates a BloomTrie chain indexer
-func NewBloomTrieIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer {
+func NewBloomTrieIndexer(db ethdb.Database, clientMode bool, odr OdrBackend) *core.ChainIndexer {
+	trieTable := ethdb.NewTable(db, BloomTrieTablePrefix)
 	backend := &BloomTrieIndexerBackend{
 		diskdb:    db,
-		triedb: trie.NewDatabase(ethdb.NewTable(db, BloomTrieTablePrefix)),
+		odr:       odr,
+		trieTable: trieTable,
+		triedb:    trie.NewDatabase(trieTable),
+		quit:      make(chan struct{}),
 	}
 	idb := ethdb.NewTable(db, "bltIndex-")
 
-	var confirmReq uint64
 	if clientMode {
 		backend.parentSectionSize = BloomTrieFrequency
-		confirmReq = HelperTrieConfirmations
 	} else {
 		backend.parentSectionSize = ethBloomBitsSection
-		confirmReq = HelperTrieProcessConfirmations
 	}
 	backend.bloomTrieRatio = BloomTrieFrequency / backend.parentSectionSize
 	backend.sectionHeads = make([]common.Hash, backend.bloomTrieRatio)
-	return core.NewChainIndexer(db, idb, backend, BloomTrieFrequency, confirmReq-ethBloomBitsConfirmations, time.Millisecond*100, "bloomtrie")
+	return core.NewChainIndexer(db, idb, backend, BloomTrieFrequency, 0, time.Millisecond*100, "bloomtrie")
+}
+
+// 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)
+
+	indexCh := make(chan uint, types.BloomBitLength)
+	type res struct {
+		nodes *NodeSet
+		err   error
+	}
+	resCh := make(chan res, types.BloomBitLength)
+	for i := 0; i < 20; i++ {
+		go func() {
+			for {
+				bitIndex, ok := <-indexCh
+				if !ok {
+					return
+				}
+				r := &BloomRequest{BloomTrieRoot: root, BloomTrieNum: section - 1, BitIdx: bitIndex, SectionIdxList: []uint64{section - 1}}
+				var err error
+				for {
+					err = b.odr.Retrieve(ctx, r)
+					if err == ErrNoPeers {
+						// if there are no peers to serve, retry later
+						select {
+						case <-b.quit:
+							resCh <- res{nil, fmt.Errorf("Section processing cancelled")}
+							return
+						case <-time.After(time.Second * 10):
+						}
+					} else {
+						break
+					}
+				}
+				resCh <- res{r.Proofs, err}
+			}
+		}()
+	}
+
+	for i := uint(0); i < types.BloomBitLength; i++ {
+		indexCh <- i
+	}
+	close(indexCh)
+	batch := b.trieTable.NewBatch()
+	for i := uint(0); i < types.BloomBitLength; i++ {
+		res := <-resCh
+		if res.err != nil {
+			return res.err
+		}
+		res.nodes.Store(batch)
+	}
+	return batch.Write()
 }
 
 // Reset implements core.ChainIndexerBackend
@@ -251,6 +374,12 @@ 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)
+		if err == nil {
+			b.trie, err = trie.New(root, b.triedb)
+		}
+	}
 	b.section = section
 	return err
 }
@@ -300,8 +429,13 @@ func (b *BloomTrieIndexerBackend) Commit() error {
 	b.triedb.Commit(root, false)
 
 	sectionHead := b.sectionHeads[b.bloomTrieRatio-1]
-	log.Info("Storing bloom trie", "section", b.section, "head", sectionHead, "root", root, "compression", float64(compSize)/float64(decompSize))
+	log.Info("Storing bloom trie", "section", b.section, "head", fmt.Sprintf("%064x", sectionHead), "root", fmt.Sprintf("%064x", root), "compression", float64(compSize)/float64(decompSize))
 	StoreBloomTrieRoot(b.diskdb, b.section, sectionHead, root)
 
 	return nil
 }
+
+// Cancel implements core.ChainIndexerBackend
+func (b *BloomTrieIndexerBackend) Closing() {
+	close(b.quit)
+}