From e131a83e302b05162c4c92982f3527971c847ab4 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Wed, 18 Jul 2018 14:12:42 +0800 Subject: [PATCH] light, les: indexer config --- les/backend.go | 4 +-- les/handler.go | 68 +++++++++++++++++++++------------------- les/helper_test.go | 12 +++++-- les/server.go | 2 +- light/lightchain.go | 26 ++++++++------- light/lightchain_test.go | 6 ++-- light/odr_test.go | 2 +- light/postprocess.go | 30 ++++++++++++++++++ light/txpool_test.go | 2 +- 9 files changed, 96 insertions(+), 56 deletions(-) diff --git a/les/backend.go b/les/backend.go index 7fa7a597c5..645cb03668 100644 --- a/les/backend.go +++ b/les/backend.go @@ -116,7 +116,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { 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) - if leth.blockchain, err = light.NewLightChain(leth.odr, leth.chainConfig, leth.engine); err != nil { + if leth.blockchain, err = light.NewLightChain(leth.odr, leth.chainConfig, light.DefaultClientIndexerConfig, leth.engine); err != nil { return nil, err } leth.bloomIndexer.Start(leth.blockchain) @@ -128,7 +128,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { } leth.txPool = light.NewTxPool(leth.chainConfig, leth.blockchain, leth.relay) - if leth.protocolManager, err = NewProtocolManager(leth.chainConfig, true, ClientProtocolVersions, config.NetworkId, leth.eventMux, leth.engine, leth.peers, leth.blockchain, nil, chainDb, leth.odr, leth.relay, leth.serverPool, quitSync, &leth.wg); err != nil { + if leth.protocolManager, err = NewProtocolManager(leth.chainConfig, light.DefaultClientIndexerConfig, true, ClientProtocolVersions, config.NetworkId, leth.eventMux, leth.engine, leth.peers, leth.blockchain, nil, chainDb, leth.odr, leth.relay, leth.serverPool, quitSync, &leth.wg); err != nil { return nil, err } leth.ApiBackend = &LesApiBackend{leth, nil} diff --git a/les/handler.go b/les/handler.go index 1e4670f354..f0a2f06347 100644 --- a/les/handler.go +++ b/les/handler.go @@ -94,19 +94,20 @@ type txPool interface { } type ProtocolManager struct { - lightSync bool - txpool txPool - txrelay *LesTxRelay - networkId uint64 - chainConfig *params.ChainConfig - blockchain BlockChain - chainDb ethdb.Database - odr *LesOdr - server *LesServer - serverPool *serverPool - lesTopic discv5.Topic - reqDist *requestDistributor - retriever *retrieveManager + lightSync bool + txpool txPool + txrelay *LesTxRelay + networkId uint64 + chainConfig *params.ChainConfig + indexerConfig *light.IndexerConfig + blockchain BlockChain + chainDb ethdb.Database + odr *LesOdr + server *LesServer + serverPool *serverPool + lesTopic discv5.Topic + reqDist *requestDistributor + retriever *retrieveManager downloader *downloader.Downloader fetcher *lightFetcher @@ -129,24 +130,25 @@ type ProtocolManager struct { // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable // with the ethereum network. -func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, protocolVersions []uint, networkId uint64, mux *event.TypeMux, engine consensus.Engine, peers *peerSet, blockchain BlockChain, txpool txPool, chainDb ethdb.Database, odr *LesOdr, txrelay *LesTxRelay, serverPool *serverPool, quitSync chan struct{}, wg *sync.WaitGroup) (*ProtocolManager, error) { +func NewProtocolManager(chainConfig *params.ChainConfig, indexerConfig *light.IndexerConfig, lightSync bool, protocolVersions []uint, networkId uint64, mux *event.TypeMux, engine consensus.Engine, peers *peerSet, blockchain BlockChain, txpool txPool, chainDb ethdb.Database, odr *LesOdr, txrelay *LesTxRelay, serverPool *serverPool, quitSync chan struct{}, wg *sync.WaitGroup) (*ProtocolManager, error) { // Create the protocol manager with the base fields manager := &ProtocolManager{ - lightSync: lightSync, - eventMux: mux, - blockchain: blockchain, - chainConfig: chainConfig, - chainDb: chainDb, - odr: odr, - networkId: networkId, - txpool: txpool, - txrelay: txrelay, - serverPool: serverPool, - peers: peers, - newPeerCh: make(chan *peer), - quitSync: quitSync, - wg: wg, - noMorePeers: make(chan struct{}), + lightSync: lightSync, + eventMux: mux, + blockchain: blockchain, + chainConfig: chainConfig, + indexerConfig: indexerConfig, + chainDb: chainDb, + odr: odr, + networkId: networkId, + txpool: txpool, + txrelay: txrelay, + serverPool: serverPool, + peers: peers, + newPeerCh: make(chan *peer), + quitSync: quitSync, + wg: wg, + noMorePeers: make(chan struct{}), } if odr != nil { manager.retriever = odr.retriever @@ -892,7 +894,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix)) for _, req := range req.Reqs { if header := pm.blockchain.GetHeaderByNumber(req.BlockNum); header != nil { - sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*params.CHTFrequencyServer-1) + sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*pm.indexerConfig.ChtSize-1) if root := light.GetChtRoot(pm.chainDb, req.ChtNum-1, sectionHead); root != (common.Hash{}) { trie, err := trie.New(root, trieDb) if err != nil { @@ -1147,11 +1149,11 @@ func (pm *ProtocolManager) getAccount(statedb *state.StateDB, root, hash common. func (pm *ProtocolManager) getHelperTrie(id uint, idx uint64) (common.Hash, string) { switch id { case htCanonical: - idxV2 := (idx+1)*(params.CHTFrequencyClient/params.CHTFrequencyServer) - 1 - sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*params.CHTFrequencyClient-1) + idxV2 := (idx+1)*(pm.indexerConfig.ChtClientSize/pm.indexerConfig.ChtSize) - 1 + sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*pm.indexerConfig.ChtClientSize-1) return light.GetChtRoot(pm.chainDb, idxV2, sectionHead), light.ChtTablePrefix case htBloomBits: - sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*params.BloomTrieFrequency-1) + sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*pm.indexerConfig.BloomTrieSize-1) return light.GetBloomTrieRoot(pm.chainDb, idx, sectionHead), light.BloomTrieTablePrefix } return common.Hash{}, "" diff --git a/les/helper_test.go b/les/helper_test.go index 04ff0d786b..1dd0ec60d0 100644 --- a/les/helper_test.go +++ b/les/helper_test.go @@ -152,7 +152,7 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor } if lightSync { - chain, _ = light.NewLightChain(odr, gspec.Config, engine) + chain, _ = light.NewLightChain(odr, gspec.Config, light.DefaultClientIndexerConfig, engine) } else { blockchain, _ := core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}) @@ -173,13 +173,19 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor chain = blockchain } - var protocolVersions []uint + var ( + protocolVersions []uint + indexConfig *light.IndexerConfig + ) + if lightSync { protocolVersions = ClientProtocolVersions + indexConfig = light.DefaultClientIndexerConfig } else { protocolVersions = ServerProtocolVersions + indexConfig = light.DefaultServerIndexerConfig } - pm, err := NewProtocolManager(gspec.Config, lightSync, protocolVersions, NetworkId, evmux, engine, peers, chain, nil, db, odr, nil, nil, make(chan struct{}), new(sync.WaitGroup)) + pm, err := NewProtocolManager(gspec.Config, indexConfig, lightSync, protocolVersions, NetworkId, evmux, engine, peers, chain, nil, db, odr, nil, nil, make(chan struct{}), new(sync.WaitGroup)) if err != nil { return nil, err } diff --git a/les/server.go b/les/server.go index 7e3ebc9d24..ba5c6a6b15 100644 --- a/les/server.go +++ b/les/server.go @@ -53,7 +53,7 @@ type LesServer struct { func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { quitSync := make(chan struct{}) - pm, err := NewProtocolManager(eth.BlockChain().Config(), false, ServerProtocolVersions, config.NetworkId, eth.EventMux(), eth.Engine(), newPeerSet(), eth.BlockChain(), eth.TxPool(), eth.ChainDb(), nil, nil, nil, quitSync, new(sync.WaitGroup)) + pm, err := NewProtocolManager(eth.BlockChain().Config(), light.DefaultServerIndexerConfig, false, ServerProtocolVersions, config.NetworkId, eth.EventMux(), eth.Engine(), newPeerSet(), eth.BlockChain(), eth.TxPool(), eth.ChainDb(), nil, nil, nil, quitSync, new(sync.WaitGroup)) if err != nil { return nil, err } diff --git a/light/lightchain.go b/light/lightchain.go index 5a4ee62cb6..ac6ee8b937 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -48,6 +48,7 @@ var ( // interface. It only does header validation during chain insertion. type LightChain struct { hc *core.HeaderChain + indexerConfig *IndexerConfig chainDb ethdb.Database odr OdrBackend chainFeed event.Feed @@ -75,19 +76,20 @@ type LightChain struct { // NewLightChain returns a fully initialised light chain using information // available in the database. It initialises the default Ethereum header // validator. -func NewLightChain(odr OdrBackend, config *params.ChainConfig, engine consensus.Engine) (*LightChain, error) { +func NewLightChain(odr OdrBackend, config *params.ChainConfig, indexerConfig *IndexerConfig, engine consensus.Engine) (*LightChain, error) { bodyCache, _ := lru.New(bodyCacheLimit) bodyRLPCache, _ := lru.New(bodyCacheLimit) blockCache, _ := lru.New(blockCacheLimit) bc := &LightChain{ - chainDb: odr.Database(), - odr: odr, - quit: make(chan struct{}), - bodyCache: bodyCache, - bodyRLPCache: bodyRLPCache, - blockCache: blockCache, - engine: engine, + chainDb: odr.Database(), + indexerConfig: indexerConfig, + odr: odr, + quit: make(chan struct{}), + bodyCache: bodyCache, + bodyRLPCache: bodyRLPCache, + blockCache: blockCache, + engine: engine, } var err error bc.hc, err = core.NewHeaderChain(odr.Database(), config, bc.engine, bc.getProcInterrupt) @@ -457,7 +459,7 @@ func (self *LightChain) GetHeaderByNumberOdr(ctx context.Context, number uint64) if header := self.hc.GetHeaderByNumber(number); header != nil { return header, nil } - return GetHeaderByNumber(ctx, params.CHTFrequencyClient, params.HelperTrieConfirmations, self.odr, number) + return GetHeaderByNumber(ctx, self.indexerConfig.ChtSize, self.indexerConfig.ChtConfirm, self.odr, number) } // Config retrieves the header chain's chain configuration. @@ -469,9 +471,9 @@ func (self *LightChain) SyncCht(ctx context.Context) bool { } headNum := self.CurrentHeader().Number.Uint64() chtCount, _, _ := self.odr.ChtIndexer().Sections() - if headNum+1 < chtCount*params.CHTFrequencyClient { - num := chtCount*params.CHTFrequencyClient - 1 - header, err := GetHeaderByNumber(ctx, params.CHTFrequencyClient, params.HelperTrieConfirmations, self.odr, num) + if headNum+1 < chtCount*self.indexerConfig.ChtSize { + num := chtCount*self.indexerConfig.ChtSize - 1 + header, err := GetHeaderByNumber(ctx, self.indexerConfig.ChtSize, self.indexerConfig.ChtConfirm, self.odr, num) if header != nil && err == nil { self.mu.Lock() if self.hc.CurrentHeader().Number.Uint64() < header.Number.Uint64() { diff --git a/light/lightchain_test.go b/light/lightchain_test.go index c0aa51da29..c32636e347 100644 --- a/light/lightchain_test.go +++ b/light/lightchain_test.go @@ -55,7 +55,7 @@ func newCanonical(n int) (ethdb.Database, *LightChain, error) { db := ethdb.NewMemDatabase() gspec := core.Genesis{Config: params.TestChainConfig} genesis := gspec.MustCommit(db) - blockchain, _ := NewLightChain(&dummyOdr{db: db}, gspec.Config, ethash.NewFaker()) + blockchain, _ := NewLightChain(&dummyOdr{db: db}, gspec.Config, DefaultClientIndexerConfig, ethash.NewFaker()) // Create and inject the requested chain if n == 0 { @@ -75,7 +75,7 @@ func newTestLightChain() *LightChain { Config: params.TestChainConfig, } gspec.MustCommit(db) - lc, err := NewLightChain(&dummyOdr{db: db}, gspec.Config, ethash.NewFullFaker()) + lc, err := NewLightChain(&dummyOdr{db: db}, gspec.Config, DefaultClientIndexerConfig, ethash.NewFullFaker()) if err != nil { panic(err) } @@ -339,7 +339,7 @@ func TestReorgBadHeaderHashes(t *testing.T) { defer func() { delete(core.BadHashes, headers[3].Hash()) }() // Create a new LightChain and check that it rolled back the state. - ncm, err := NewLightChain(&dummyOdr{db: bc.chainDb}, params.TestChainConfig, ethash.NewFaker()) + ncm, err := NewLightChain(&dummyOdr{db: bc.chainDb}, params.TestChainConfig, DefaultClientIndexerConfig, ethash.NewFaker()) if err != nil { t.Fatalf("failed to create new chain manager: %v", err) } diff --git a/light/odr_test.go b/light/odr_test.go index 3e7ac10118..c244945b4f 100644 --- a/light/odr_test.go +++ b/light/odr_test.go @@ -259,7 +259,7 @@ func testChainOdr(t *testing.T, protocol int, fn odrTestFn) { } odr := &testOdr{sdb: sdb, ldb: ldb} - lightchain, err := NewLightChain(odr, params.TestChainConfig, ethash.NewFullFaker()) + lightchain, err := NewLightChain(odr, params.TestChainConfig, DefaultClientIndexerConfig, ethash.NewFullFaker()) if err != nil { t.Fatal(err) } diff --git a/light/postprocess.go b/light/postprocess.go index 06546b6201..4e1793564c 100644 --- a/light/postprocess.go +++ b/light/postprocess.go @@ -34,6 +34,36 @@ import ( "github.com/ethereum/go-ethereum/trie" ) +// IndexerConfig specifies a set of configs for chain indexers. +type IndexerConfig struct { + ChtSize uint64 + ChtClientSize uint64 + ChtConfirm uint64 + BloomSize uint64 + BloomConfirm uint64 + BloomTrieSize uint64 + BloomTrieConfirm uint64 +} + +var DefaultServerIndexerConfig = &IndexerConfig{ + ChtSize: params.CHTFrequencyServer, + ChtClientSize: params.CHTFrequencyClient, + ChtConfirm: params.HelperTrieProcessConfirmations, + BloomSize: params.BloomBitsBlocks, + BloomConfirm: params.BloomConfirms, + BloomTrieSize: params.BloomTrieFrequency, + BloomTrieConfirm: params.HelperTrieProcessConfirmations, +} + +var DefaultClientIndexerConfig = &IndexerConfig{ + ChtSize: params.CHTFrequencyClient, + ChtConfirm: params.HelperTrieConfirmations, + BloomSize: params.BloomBitsBlocksClient, + BloomConfirm: params.HelperTrieConfirmations, + BloomTrieSize: params.BloomTrieFrequency, + BloomTrieConfirm: params.HelperTrieConfirmations, +} + // trustedCheckpoint represents a set of post-processed trie roots (CHT and BloomTrie) associated with // the appropriate section index and head hash. It is used to start light syncing from this checkpoint // and avoid downloading the entire header chain while still being able to securely access old headers/logs. diff --git a/light/txpool_test.go b/light/txpool_test.go index ccbd83a943..7c19624597 100644 --- a/light/txpool_test.go +++ b/light/txpool_test.go @@ -100,7 +100,7 @@ func TestTxPool(t *testing.T) { discard: make(chan int, 1), mined: make(chan int, 1), } - lightchain, _ := NewLightChain(odr, params.TestChainConfig, ethash.NewFullFaker()) + lightchain, _ := NewLightChain(odr, params.TestChainConfig, DefaultClientIndexerConfig, ethash.NewFullFaker()) txPermanent = 50 pool := NewTxPool(params.TestChainConfig, lightchain, relay) ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)