From da8abaf8cb987ad919a122cd8ee8e0a7c7688806 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Sun, 15 Jul 2018 16:51:41 +0800 Subject: [PATCH] les: refactor test a bit --- core/chain_indexer.go | 1 - les/backend.go | 2 +- les/handler_test.go | 179 ++++++++++++++++++++------------------- les/helper_test.go | 153 ++++++++++++++++++++++++++++----- les/odr_test.go | 50 ++++------- les/request_test.go | 49 ++++------- light/lightchain.go | 4 +- light/lightchain_test.go | 13 ++- light/odr_test.go | 8 +- light/postprocess.go | 63 +++++++++----- light/trie_test.go | 2 +- light/txpool_test.go | 4 +- 12 files changed, 323 insertions(+), 205 deletions(-) diff --git a/core/chain_indexer.go b/core/chain_indexer.go index 0b927116d0..52e914659a 100644 --- a/core/chain_indexer.go +++ b/core/chain_indexer.go @@ -309,7 +309,6 @@ func (c *ChainIndexer) updateLoop() { updating = false c.log.Info("Finished upgrading chain index") } - c.cascadedHead = c.storedSections*c.sectionSize - 1 for _, child := range c.children { c.log.Trace("Cascading chain index update", "head", c.cascadedHead) diff --git a/les/backend.go b/les/backend.go index 9bd8c9282a..f745544abb 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, light.DefaultClientIndexerConfig, leth.chtIndexer, leth.bloomTrieIndexer, leth.bloomIndexer, leth.retriever) - if leth.blockchain, err = light.NewLightChain(leth.odr, leth.chainConfig, light.DefaultClientIndexerConfig, leth.engine); err != nil { + if leth.blockchain, err = light.NewLightChain(leth.odr, leth.chainConfig, leth.engine); err != nil { return nil, err } leth.bloomIndexer.Start(leth.blockchain) diff --git a/les/handler_test.go b/les/handler_test.go index 0e3035c57d..81297d1a8f 100644 --- a/les/handler_test.go +++ b/les/handler_test.go @@ -51,10 +51,13 @@ func TestGetBlockHeadersLes1(t *testing.T) { testGetBlockHeaders(t, 1) } func TestGetBlockHeadersLes2(t *testing.T) { testGetBlockHeaders(t, 2) } func testGetBlockHeaders(t *testing.T, protocol int) { - pm := newTestProtocolManagerMust(t, false, downloader.MaxHashFetch+15, nil, nil, nil, ethdb.NewMemDatabase()) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() + server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, 0) + defer func() { + if tearDown != nil { + tearDown() + } + }() + bc := server.pm.blockchain.(*core.BlockChain) // Create a "random" unknown hash for testing var unknown common.Hash @@ -167,9 +170,9 @@ func testGetBlockHeaders(t *testing.T, protocol int) { } // Send the hash request and verify the response reqID++ - cost := peer.GetRequestCost(GetBlockHeadersMsg, int(tt.query.Amount)) - sendRequest(peer.app, GetBlockHeadersMsg, reqID, cost, tt.query) - if err := expectResponse(peer.app, BlockHeadersMsg, reqID, testBufLimit, headers); err != nil { + cost := server.tPeer.GetRequestCost(GetBlockHeadersMsg, int(tt.query.Amount)) + sendRequest(server.tPeer.app, GetBlockHeadersMsg, reqID, cost, tt.query) + if err := expectResponse(server.tPeer.app, BlockHeadersMsg, reqID, testBufLimit, headers); err != nil { t.Errorf("test %d: headers mismatch: %v", i, err) } } @@ -180,10 +183,13 @@ func TestGetBlockBodiesLes1(t *testing.T) { testGetBlockBodies(t, 1) } func TestGetBlockBodiesLes2(t *testing.T) { testGetBlockBodies(t, 2) } func testGetBlockBodies(t *testing.T, protocol int) { - pm := newTestProtocolManagerMust(t, false, downloader.MaxBlockFetch+15, nil, nil, nil, ethdb.NewMemDatabase()) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() + server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, protocol, 0) + defer func() { + if tearDown != nil { + tearDown() + } + }() + bc := server.pm.blockchain.(*core.BlockChain) // Create a batch of tests for various scenarios limit := MaxBodyFetch @@ -243,9 +249,9 @@ func testGetBlockBodies(t *testing.T, protocol int) { } reqID++ // Send the hash request and verify the response - cost := peer.GetRequestCost(GetBlockBodiesMsg, len(hashes)) - sendRequest(peer.app, GetBlockBodiesMsg, reqID, cost, hashes) - if err := expectResponse(peer.app, BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil { + cost := server.tPeer.GetRequestCost(GetBlockBodiesMsg, len(hashes)) + sendRequest(server.tPeer.app, GetBlockBodiesMsg, reqID, cost, hashes) + if err := expectResponse(server.tPeer.app, BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil { t.Errorf("test %d: bodies mismatch: %v", i, err) } } @@ -257,10 +263,13 @@ func TestGetCodeLes2(t *testing.T) { testGetCode(t, 2) } func testGetCode(t *testing.T, protocol int) { // Assemble the test environment - pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, ethdb.NewMemDatabase()) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() + server, tearDown := newServerEnv(t, 4, protocol, 0) + defer func() { + if tearDown != nil { + tearDown() + } + }() + bc := server.pm.blockchain.(*core.BlockChain) var codereqs []*CodeReq var codes [][]byte @@ -277,9 +286,9 @@ func testGetCode(t *testing.T, protocol int) { } } - cost := peer.GetRequestCost(GetCodeMsg, len(codereqs)) - sendRequest(peer.app, GetCodeMsg, 42, cost, codereqs) - if err := expectResponse(peer.app, CodeMsg, 42, testBufLimit, codes); err != nil { + cost := server.tPeer.GetRequestCost(GetCodeMsg, len(codereqs)) + sendRequest(server.tPeer.app, GetCodeMsg, 42, cost, codereqs) + if err := expectResponse(server.tPeer.app, CodeMsg, 42, testBufLimit, codes); err != nil { t.Errorf("codes mismatch: %v", err) } } @@ -290,11 +299,13 @@ func TestGetReceiptLes2(t *testing.T) { testGetReceipt(t, 2) } func testGetReceipt(t *testing.T, protocol int) { // Assemble the test environment - db := ethdb.NewMemDatabase() - pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() + server, tearDown := newServerEnv(t, 4, protocol, 0) + defer func() { + if tearDown != nil { + tearDown() + } + }() + bc := server.pm.blockchain.(*core.BlockChain) // Collect the hashes to request, and the response to expect hashes, receipts := []common.Hash{}, []types.Receipts{} @@ -302,12 +313,12 @@ func testGetReceipt(t *testing.T, protocol int) { block := bc.GetBlockByNumber(i) hashes = append(hashes, block.Hash()) - receipts = append(receipts, rawdb.ReadReceipts(db, block.Hash(), block.NumberU64())) + receipts = append(receipts, rawdb.ReadReceipts(server.db, block.Hash(), block.NumberU64())) } // Send the hash request and verify the response - cost := peer.GetRequestCost(GetReceiptsMsg, len(hashes)) - sendRequest(peer.app, GetReceiptsMsg, 42, cost, hashes) - if err := expectResponse(peer.app, ReceiptsMsg, 42, testBufLimit, receipts); err != nil { + cost := server.tPeer.GetRequestCost(GetReceiptsMsg, len(hashes)) + sendRequest(server.tPeer.app, GetReceiptsMsg, 42, cost, hashes) + if err := expectResponse(server.tPeer.app, ReceiptsMsg, 42, testBufLimit, receipts); err != nil { t.Errorf("receipts mismatch: %v", err) } } @@ -318,11 +329,13 @@ func TestGetProofsLes2(t *testing.T) { testGetProofs(t, 2) } func testGetProofs(t *testing.T, protocol int) { // Assemble the test environment - db := ethdb.NewMemDatabase() - pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() + server, tearDown := newServerEnv(t, 4, protocol, 0) + defer func() { + if tearDown != nil { + tearDown() + } + }() + bc := server.pm.blockchain.(*core.BlockChain) var ( proofreqs []ProofReq @@ -334,7 +347,7 @@ func testGetProofs(t *testing.T, protocol int) { for i := uint64(0); i <= bc.CurrentBlock().NumberU64(); i++ { header := bc.GetHeaderByNumber(i) root := header.Root - trie, _ := trie.New(root, trie.NewDatabase(db)) + trie, _ := trie.New(root, trie.NewDatabase(server.db)) for _, acc := range accounts { req := ProofReq{ @@ -356,15 +369,15 @@ func testGetProofs(t *testing.T, protocol int) { // Send the proof request and verify the response switch protocol { case 1: - cost := peer.GetRequestCost(GetProofsV1Msg, len(proofreqs)) - sendRequest(peer.app, GetProofsV1Msg, 42, cost, proofreqs) - if err := expectResponse(peer.app, ProofsV1Msg, 42, testBufLimit, proofsV1); err != nil { + cost := server.tPeer.GetRequestCost(GetProofsV1Msg, len(proofreqs)) + sendRequest(server.tPeer.app, GetProofsV1Msg, 42, cost, proofreqs) + if err := expectResponse(server.tPeer.app, ProofsV1Msg, 42, testBufLimit, proofsV1); err != nil { t.Errorf("proofs mismatch: %v", err) } case 2: - cost := peer.GetRequestCost(GetProofsV2Msg, len(proofreqs)) - sendRequest(peer.app, GetProofsV2Msg, 42, cost, proofreqs) - if err := expectResponse(peer.app, ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil { + cost := server.tPeer.GetRequestCost(GetProofsV2Msg, len(proofreqs)) + sendRequest(server.tPeer.app, GetProofsV2Msg, 42, cost, proofreqs) + if err := expectResponse(server.tPeer.app, ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil { t.Errorf("proofs mismatch: %v", err) } } @@ -375,28 +388,26 @@ func TestGetCHTProofsLes1(t *testing.T) { testGetCHTProofs(t, 1) } func TestGetCHTProofsLes2(t *testing.T) { testGetCHTProofs(t, 2) } func testGetCHTProofs(t *testing.T, protocol int) { - // Figure out the client's CHT frequency - frequency := uint64(params.CHTFrequencyClient) - if protocol == 1 { - frequency = uint64(params.CHTFrequencyServer) + config := light.TestServerIndexerConfig + frequency := config.ChtSize + if protocol == 2 { + frequency = config.ChtSize * 8 } - // Assemble the test environment - db := ethdb.NewMemDatabase() - pm := newTestProtocolManagerMust(t, false, int(frequency)+params.HelperTrieProcessConfirmations, testChainGen, nil, nil, db) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() - // Wait a while for the CHT indexer to process the new headers - time.Sleep(100 * time.Millisecond * time.Duration(frequency/params.CHTFrequencyServer)) // Chain indexer throttling - time.Sleep(250 * time.Millisecond) // CI tester slack + server, tearDown := newServerEnv(t, int(frequency+config.ChtConfirm), protocol, 1) + defer func() { + if tearDown != nil { + tearDown() + } + }() + bc := server.pm.blockchain.(*core.BlockChain) // Assemble the proofs from the different protocols - header := bc.GetHeaderByNumber(frequency) + header := bc.GetHeaderByNumber(config.ChtSize - 1) rlp, _ := rlp.EncodeToBytes(header) key := make([]byte, 8) - binary.BigEndian.PutUint64(key, frequency) + binary.BigEndian.PutUint64(key, config.ChtSize-1) proofsV1 := []ChtResp{{ Header: header, @@ -406,22 +417,22 @@ func testGetCHTProofs(t *testing.T, protocol int) { } switch protocol { case 1: - root := light.GetChtRoot(db, 0, bc.GetHeaderByNumber(frequency-1).Hash()) - trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(db, light.ChtTablePrefix))) + root := light.GetChtRoot(server.db, 0, bc.GetHeaderByNumber(config.ChtSize-1).Hash()) + trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.ChtTablePrefix))) var proof light.NodeList trie.Prove(key, 0, &proof) proofsV1[0].Proof = proof case 2: - root := light.GetChtRoot(db, (params.CHTFrequencyClient/params.CHTFrequencyServer)-1, bc.GetHeaderByNumber(frequency-1).Hash()) - trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(db, light.ChtTablePrefix))) + root := light.GetChtRoot(server.db, (config.ChtSize/config.PairChtSize)-1, bc.GetHeaderByNumber(config.ChtSize-1).Hash()) + trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.ChtTablePrefix))) trie.Prove(key, 0, &proofsV2.Proofs) } // Assemble the requests for the different protocols requestsV1 := []ChtReq{{ ChtNum: 1, - BlockNum: frequency, + BlockNum: config.ChtSize - 1, }} requestsV2 := []HelperTrieReq{{ Type: htCanonical, @@ -432,15 +443,15 @@ func testGetCHTProofs(t *testing.T, protocol int) { // Send the proof request and verify the response switch protocol { case 1: - cost := peer.GetRequestCost(GetHeaderProofsMsg, len(requestsV1)) - sendRequest(peer.app, GetHeaderProofsMsg, 42, cost, requestsV1) - if err := expectResponse(peer.app, HeaderProofsMsg, 42, testBufLimit, proofsV1); err != nil { + cost := server.tPeer.GetRequestCost(GetHeaderProofsMsg, len(requestsV1)) + sendRequest(server.tPeer.app, GetHeaderProofsMsg, 42, cost, requestsV1) + if err := expectResponse(server.tPeer.app, HeaderProofsMsg, 42, testBufLimit, proofsV1); err != nil { t.Errorf("proofs mismatch: %v", err) } case 2: - cost := peer.GetRequestCost(GetHelperTrieProofsMsg, len(requestsV2)) - sendRequest(peer.app, GetHelperTrieProofsMsg, 42, cost, requestsV2) - if err := expectResponse(peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil { + cost := server.tPeer.GetRequestCost(GetHelperTrieProofsMsg, len(requestsV2)) + sendRequest(server.tPeer.app, GetHelperTrieProofsMsg, 42, cost, requestsV2) + if err := expectResponse(server.tPeer.app, HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil { t.Errorf("proofs mismatch: %v", err) } } @@ -448,24 +459,22 @@ func testGetCHTProofs(t *testing.T, protocol int) { // Tests that bloombits proofs can be correctly retrieved. func TestGetBloombitsProofs(t *testing.T) { - // Assemble the test environment - db := ethdb.NewMemDatabase() - pm := newTestProtocolManagerMust(t, false, int(params.BloomBitsBlocksClient)+256, testChainGen, nil, nil, db) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", 2, pm, true) - defer peer.close() - - // Wait a while for the bloombits indexer to process the new headers - time.Sleep(100 * time.Millisecond * time.Duration(params.BloomBitsBlocksClient/4096)) // Chain indexer throttling - time.Sleep(250 * time.Millisecond) // CI tester slack + config := light.TestServerIndexerConfig + server, tearDown := newServerEnv(t, int(config.BloomSize+config.BloomConfirm), 2, 1) + defer func() { + if tearDown != nil { + tearDown() + } + }() + bc := server.pm.blockchain.(*core.BlockChain) // Request and verify each bit of the bloom bits proofs for bit := 0; bit < 2048; bit++ { - // Assemble therequest and proofs for the bloombits + // Assemble the request and proofs for the bloombits key := make([]byte, 10) binary.BigEndian.PutUint16(key[:2], uint16(bit)) - binary.BigEndian.PutUint64(key[2:], params.BloomBitsBlocksClient) + binary.BigEndian.PutUint64(key[2:], config.BloomSize-1) requests := []HelperTrieReq{{ Type: htBloomBits, @@ -474,14 +483,14 @@ func TestGetBloombitsProofs(t *testing.T) { }} var proofs HelperTrieResps - root := light.GetBloomTrieRoot(db, 0, bc.GetHeaderByNumber(params.BloomBitsBlocksClient-1).Hash()) - trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(db, light.BloomTrieTablePrefix))) + root := light.GetBloomTrieRoot(server.db, 0, bc.GetHeaderByNumber(config.BloomSize-1).Hash()) + trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.BloomTrieTablePrefix))) trie.Prove(key, 0, &proofs.Proofs) // Send the proof request and verify the response - cost := peer.GetRequestCost(GetHelperTrieProofsMsg, len(requests)) - sendRequest(peer.app, GetHelperTrieProofsMsg, 42, cost, requests) - if err := expectResponse(peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil { + cost := server.tPeer.GetRequestCost(GetHelperTrieProofsMsg, len(requests)) + sendRequest(server.tPeer.app, GetHelperTrieProofsMsg, 42, cost, requests) + if err := expectResponse(server.tPeer.app, HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil { t.Errorf("bit %d: proofs mismatch: %v", bit, err) } } diff --git a/les/helper_test.go b/les/helper_test.go index 1dd0ec60d0..d17efa62ed 100644 --- a/les/helper_test.go +++ b/les/helper_test.go @@ -24,6 +24,7 @@ import ( "math/big" "sync" "testing" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/consensus/ethash" @@ -123,6 +124,16 @@ func testChainGen(i int, block *core.BlockGen) { } } +// testIndexers creates a set of indexers with specified params for testing purpose. +func testIndexers(db ethdb.Database, indexerConfig *light.IndexerConfig) (*core.ChainIndexer, *core.ChainIndexer, *core.ChainIndexer) { + chtIndexer := light.NewChtIndexer(db, indexerConfig.ChtSize, indexerConfig.ChtConfirm) + bloomIndexer := eth.NewBloomIndexer(db, indexerConfig.BloomSize, indexerConfig.BloomConfirm) + bloomTrieIndexer := light.NewBloomTrieIndexer(db, indexerConfig.BloomSize, indexerConfig.BloomConfirm, + indexerConfig.BloomTrieSize, indexerConfig.BloomTrieConfirm) + bloomIndexer.AddChildIndexer(bloomTrieIndexer) + return chtIndexer, bloomIndexer, bloomTrieIndexer +} + func testRCL() RequestCostList { cl := make(RequestCostList, len(reqList)) for i, code := range reqList { @@ -134,9 +145,9 @@ func testRCL() RequestCostList { } // newTestProtocolManager creates a new protocol manager for testing purposes, -// with the given number of blocks already known, and potential notification -// channels for different events. -func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) (*ProtocolManager, error) { +// with the given number of blocks already known, potential notification +// channels for different events and relative chain indexers array. +func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *core.BlockGen), odr *LesOdr, peers *peerSet, db ethdb.Database) (*ProtocolManager, error) { var ( evmux = new(event.TypeMux) engine = ethash.NewFaker() @@ -152,20 +163,9 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor } if lightSync { - chain, _ = light.NewLightChain(odr, gspec.Config, light.DefaultClientIndexerConfig, engine) + chain, _ = light.NewLightChain(odr, gspec.Config, engine) } else { blockchain, _ := core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}) - - chtIndexer := light.NewChtIndexer(db, params.CHTFrequencyServer, params.HelperTrieProcessConfirmations) - chtIndexer.Start(blockchain) - - bbtIndexer := light.NewBloomTrieIndexer(db, params.BloomBitsBlocks, params.BloomConfirms, - params.BloomTrieFrequency, params.HelperTrieProcessConfirmations) - - bloomIndexer := eth.NewBloomIndexer(db, params.BloomBitsBlocks, params.BloomConfirms) - bloomIndexer.AddChildIndexer(bbtIndexer) - bloomIndexer.Start(blockchain) - gchain, _ := core.GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, blocks, generator) if _, err := blockchain.InsertChain(gchain); err != nil { panic(err) @@ -180,10 +180,10 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor if lightSync { protocolVersions = ClientProtocolVersions - indexConfig = light.DefaultClientIndexerConfig + indexConfig = light.TestClientIndexerConfig } else { protocolVersions = ServerProtocolVersions - indexConfig = light.DefaultServerIndexerConfig + indexConfig = light.TestServerIndexerConfig } 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 { @@ -206,11 +206,11 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor } // newTestProtocolManagerMust creates a new protocol manager for testing purposes, -// with the given number of blocks already known, and potential notification -// channels for different events. In case of an error, the constructor force- +// with the given number of blocks already known, potential notification +// channels for different events and relative chain indexers array. In case of an error, the constructor force- // fails the test. -func newTestProtocolManagerMust(t *testing.T, lightSync bool, blocks int, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) *ProtocolManager { - pm, err := newTestProtocolManager(lightSync, blocks, generator, peers, odr, db) +func newTestProtocolManagerMust(t *testing.T, lightSync bool, blocks int, generator func(int, *core.BlockGen), odr *LesOdr, peers *peerSet, db ethdb.Database) *ProtocolManager { + pm, err := newTestProtocolManager(lightSync, blocks, generator, odr, peers, db) if err != nil { t.Fatalf("Failed to create protocol manager: %v", err) } @@ -333,3 +333,114 @@ func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNu func (p *testPeer) close() { p.app.Close() } + +// TestEntity represents a network entity for testing with necessary auxiliary fields. +type TestEntity struct { + db ethdb.Database + rPeer *peer + tPeer *testPeer + peers *peerSet + pm *ProtocolManager + // Indexers + chtIndexer *core.ChainIndexer + bloomIndexer *core.ChainIndexer + bloomTrieIndexer *core.ChainIndexer +} + +// newServerEnv creates a server testing environment with a connected test peer for testing purpose. +func newServerEnv(t *testing.T, blocks int, protocol int, processSections uint64) (*TestEntity, func()) { + db := ethdb.NewMemDatabase() + cIndexer, bIndexer, btIndexer := testIndexers(db, light.TestServerIndexerConfig) + + pm := newTestProtocolManagerMust(t, false, blocks, testChainGen, nil, nil, db) + peer, _ := newTestPeer(t, "peer", protocol, pm, true) + + cIndexer.Start(pm.blockchain.(*core.BlockChain)) + bIndexer.Start(pm.blockchain.(*core.BlockChain)) + + // Wait until indexers generate enough index data. + if processSections > 0 { + for { + cs, _, _ := cIndexer.Sections() + bs, _, _ := bIndexer.Sections() + if cs >= processSections && bs >= processSections { + break + } + time.Sleep(10 * time.Millisecond) + } + } + return &TestEntity{ + db: db, + tPeer: peer, + pm: pm, + chtIndexer: cIndexer, + bloomIndexer: bIndexer, + bloomTrieIndexer: btIndexer, + }, func() { + peer.close() + // Note bloom trie indexer will be closed by it parent recursively. + cIndexer.Close() + bIndexer.Close() + } +} + +// newClientServerEnv creates a client/server arch environment with a connected les server and light client pair +// for testing purpose. +func newClientServerEnv(t *testing.T, blocks int, protocol int) (*TestEntity, *TestEntity, func()) { + db, ldb := ethdb.NewMemDatabase(), ethdb.NewMemDatabase() + peers, lPeers := newPeerSet(), newPeerSet() + + cIndexer, bIndexer, btIndexer := testIndexers(db, light.TestServerIndexerConfig) + lcIndexer, lbIndexer, lbtIndexer := testIndexers(ldb, light.TestClientIndexerConfig) + + startIndexers := func(clientMode bool, pm *ProtocolManager) { + if clientMode { + lcIndexer.Start(pm.blockchain.(*light.LightChain)) + lbIndexer.Start(pm.blockchain.(*light.LightChain)) + } else { + cIndexer.Start(pm.blockchain.(*core.BlockChain)) + bIndexer.Start(pm.blockchain.(*core.BlockChain)) + } + } + + pm := newTestProtocolManagerMust(t, false, blocks, testChainGen, nil, peers, db) + dist := newRequestDistributor(lPeers, make(chan struct{})) + rm := newRetrieveManager(lPeers, dist, nil) + lpm := newTestProtocolManagerMust(t, true, 0, nil, NewLesOdr(ldb, light.TestClientIndexerConfig, lcIndexer, lbtIndexer, lbIndexer, rm), lPeers, ldb) + + startIndexers(false, pm) + startIndexers(true, lpm) + + peer, err1, lPeer, err2 := newTestPeerPair("peer", protocol, pm, lpm) + select { + case <-time.After(time.Millisecond * 100): + case err := <-err1: + t.Fatalf("peer 1 handshake error: %v", err) + case err := <-err2: + t.Fatalf("peer 2 handshake error: %v", err) + } + + return &TestEntity{ + db: db, + pm: pm, + rPeer: peer, + peers: peers, + chtIndexer: cIndexer, + bloomIndexer: bIndexer, + bloomTrieIndexer: btIndexer, + }, &TestEntity{ + db: ldb, + pm: lpm, + rPeer: lPeer, + peers: lPeers, + chtIndexer: lcIndexer, + bloomIndexer: lbIndexer, + bloomTrieIndexer: lbtIndexer, + }, func() { + // Note bloom trie indexers will be closed by their parents recursively. + cIndexer.Close() + bIndexer.Close() + lcIndexer.Close() + lbIndexer.Close() + } +} diff --git a/les/odr_test.go b/les/odr_test.go index 8ba663c92f..f041927139 100644 --- a/les/odr_test.go +++ b/les/odr_test.go @@ -30,7 +30,6 @@ import ( "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/vm" - "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/params" @@ -160,37 +159,25 @@ func odrContractCall(ctx context.Context, db ethdb.Database, config *params.Chai return res } +// testOdr tests odr requests whose validation guaranteed by block headers. func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) { // Assemble the test environment - peers := newPeerSet() - dist := newRequestDistributor(peers, make(chan struct{})) - rm := newRetrieveManager(peers, dist, nil) - db := ethdb.NewMemDatabase() - ldb := ethdb.NewMemDatabase() - odr := NewLesOdr(ldb, light.DefaultClientIndexerConfig, light.NewChtIndexer(db, params.CHTFrequencyClient, params.HelperTrieConfirmations), - light.NewBloomTrieIndexer(db, params.BloomBitsBlocksClient, params.HelperTrieConfirmations, params.BloomTrieFrequency, params.HelperTrieConfirmations), - eth.NewBloomIndexer(db, params.BloomBitsBlocksClient, params.HelperTrieConfirmations), rm) - 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) - select { - case <-time.After(time.Millisecond * 100): - case err := <-err1: - t.Fatalf("peer 1 handshake error: %v", err) - case err := <-err2: - t.Fatalf("peer 1 handshake error: %v", err) - } - - lpm.synchronise(lpeer) + server, client, tearDown := newClientServerEnv(t, 4, protocol) + defer func() { + if tearDown != nil { + tearDown() + } + }() + client.pm.synchronise(client.rPeer) test := func(expFail uint64) { - for i := uint64(0); i <= pm.blockchain.CurrentHeader().Number.Uint64(); i++ { - bhash := rawdb.ReadCanonicalHash(db, i) - b1 := fn(light.NoOdr, db, pm.chainConfig, pm.blockchain.(*core.BlockChain), nil, bhash) + for i := uint64(0); i <= server.pm.blockchain.CurrentHeader().Number.Uint64(); i++ { + bhash := rawdb.ReadCanonicalHash(server.db, i) + b1 := fn(light.NoOdr, server.db, server.pm.chainConfig, server.pm.blockchain.(*core.BlockChain), nil, bhash) ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) defer cancel() - b2 := fn(ctx, ldb, lpm.chainConfig, nil, lpm.blockchain.(*light.LightChain), bhash) + b2 := fn(ctx, client.db, client.pm.chainConfig, nil, client.pm.blockchain.(*light.LightChain), bhash) eq := bytes.Equal(b1, b2) exp := i < expFail @@ -202,21 +189,20 @@ func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) { } } } - // temporarily remove peer to test odr fails // expect retrievals to fail (except genesis block) without a les peer - peers.Unregister(lpeer.id) + client.peers.Unregister(client.rPeer.id) time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed test(expFail) // expect all retrievals to pass - peers.Register(lpeer) + client.peers.Register(client.rPeer) time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed - lpeer.lock.Lock() - lpeer.hasBlock = func(common.Hash, uint64) bool { return true } - lpeer.lock.Unlock() + client.peers.lock.Lock() + client.rPeer.hasBlock = func(common.Hash, uint64) bool { return true } + client.peers.lock.Unlock() test(5) // still expect all retrievals to pass, now data should be cached locally - peers.Unregister(lpeer.id) + client.peers.Unregister(client.rPeer.id) time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed test(5) } diff --git a/les/request_test.go b/les/request_test.go index 34848c6081..e465187a87 100644 --- a/les/request_test.go +++ b/les/request_test.go @@ -24,10 +24,8 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/light" - "github.com/ethereum/go-ethereum/params" ) var testBankSecureTrieKey = secAddr(testBankAddress) @@ -85,36 +83,21 @@ func tfCodeAccess(db ethdb.Database, bhash common.Hash, num uint64) light.OdrReq func testAccess(t *testing.T, protocol int, fn accessTestFn) { // Assemble the test environment - peers := newPeerSet() - dist := newRequestDistributor(peers, make(chan struct{})) - rm := newRetrieveManager(peers, dist, nil) - db := ethdb.NewMemDatabase() - ldb := ethdb.NewMemDatabase() - odr := NewLesOdr(ldb, light.DefaultClientIndexerConfig, light.NewChtIndexer(db, params.CHTFrequencyClient, params.HelperTrieConfirmations), - light.NewBloomTrieIndexer(db, params.BloomBitsBlocksClient, params.HelperTrieConfirmations, params.BloomTrieFrequency, params.HelperTrieConfirmations), - eth.NewBloomIndexer(db, params.BloomBitsBlocksClient, params.HelperTrieConfirmations), rm) - - 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) - select { - case <-time.After(time.Millisecond * 100): - case err := <-err1: - t.Fatalf("peer 1 handshake error: %v", err) - case err := <-err2: - t.Fatalf("peer 1 handshake error: %v", err) - } - - lpm.synchronise(lpeer) + server, client, tearDown := newClientServerEnv(t, 4, protocol) + defer func() { + if tearDown != nil { + tearDown() + } + }() + client.pm.synchronise(client.rPeer) test := func(expFail uint64) { - for i := uint64(0); i <= pm.blockchain.CurrentHeader().Number.Uint64(); i++ { - bhash := rawdb.ReadCanonicalHash(db, i) - if req := fn(ldb, bhash, i); req != nil { + for i := uint64(0); i <= server.pm.blockchain.CurrentHeader().Number.Uint64(); i++ { + bhash := rawdb.ReadCanonicalHash(server.db, i) + if req := fn(client.db, bhash, i); req != nil { ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) defer cancel() - - err := odr.Retrieve(ctx, req) + err := client.pm.odr.Retrieve(ctx, req) got := err == nil exp := i < expFail if exp && !got { @@ -128,16 +111,16 @@ func testAccess(t *testing.T, protocol int, fn accessTestFn) { } // temporarily remove peer to test odr fails - peers.Unregister(lpeer.id) + client.peers.Unregister(client.rPeer.id) time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed // expect retrievals to fail (except genesis block) without a les peer test(0) - peers.Register(lpeer) + client.peers.Register(client.rPeer) time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed - lpeer.lock.Lock() - lpeer.hasBlock = func(common.Hash, uint64) bool { return true } - lpeer.lock.Unlock() + client.rPeer.lock.Lock() + client.rPeer.hasBlock = func(common.Hash, uint64) bool { return true } + client.rPeer.lock.Unlock() // expect all retrievals to pass test(5) } diff --git a/light/lightchain.go b/light/lightchain.go index bb01e7d310..5cd161fd69 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -76,14 +76,14 @@ 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, indexerConfig *IndexerConfig, engine consensus.Engine) (*LightChain, error) { +func NewLightChain(odr OdrBackend, config *params.ChainConfig, engine consensus.Engine) (*LightChain, error) { bodyCache, _ := lru.New(bodyCacheLimit) bodyRLPCache, _ := lru.New(bodyCacheLimit) blockCache, _ := lru.New(blockCacheLimit) bc := &LightChain{ chainDb: odr.Database(), - indexerConfig: indexerConfig, + indexerConfig: odr.IndexerConfig(), odr: odr, quit: make(chan struct{}), bodyCache: bodyCache, diff --git a/light/lightchain_test.go b/light/lightchain_test.go index c32636e347..87392850b5 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, DefaultClientIndexerConfig, ethash.NewFaker()) + blockchain, _ := NewLightChain(&dummyOdr{db: db, indexerConfig: TestClientIndexerConfig}, gspec.Config, 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, DefaultClientIndexerConfig, ethash.NewFullFaker()) + lc, err := NewLightChain(&dummyOdr{db: db}, gspec.Config, ethash.NewFullFaker()) if err != nil { panic(err) } @@ -265,7 +265,8 @@ func makeHeaderChainWithDiff(genesis *types.Block, d []int, seed byte) []*types. type dummyOdr struct { OdrBackend - db ethdb.Database + db ethdb.Database + indexerConfig *IndexerConfig } func (odr *dummyOdr) Database() ethdb.Database { @@ -276,6 +277,10 @@ func (odr *dummyOdr) Retrieve(ctx context.Context, req OdrRequest) error { return nil } +func (odr *dummyOdr) IndexerConfig() *IndexerConfig { + return odr.indexerConfig +} + // Tests that reorganizing a long difficult chain after a short easy one // overwrites the canonical numbers and links in the database. func TestReorgLongHeaders(t *testing.T) { @@ -339,7 +344,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, DefaultClientIndexerConfig, ethash.NewFaker()) + ncm, err := NewLightChain(&dummyOdr{db: bc.chainDb}, params.TestChainConfig, 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 a118a5ad3a..9b92fd585b 100644 --- a/light/odr_test.go +++ b/light/odr_test.go @@ -93,6 +93,10 @@ func (odr *testOdr) Retrieve(ctx context.Context, req OdrRequest) error { return nil } +func (odr *testOdr) IndexerConfig() *IndexerConfig { + return odr.indexerConfig +} + type odrTestFn func(ctx context.Context, db ethdb.Database, bc *core.BlockChain, lc *LightChain, bhash common.Hash) ([]byte, error) func TestOdrGetBlockLes1(t *testing.T) { testChainOdr(t, 1, odrGetBlock) } @@ -259,8 +263,8 @@ func testChainOdr(t *testing.T, protocol int, fn odrTestFn) { t.Fatal(err) } - odr := &testOdr{sdb: sdb, ldb: ldb, indexerConfig: DefaultClientIndexerConfig} - lightchain, err := NewLightChain(odr, params.TestChainConfig, DefaultClientIndexerConfig, ethash.NewFullFaker()) + odr := &testOdr{sdb: sdb, ldb: ldb, indexerConfig: TestClientIndexerConfig} + lightchain, err := NewLightChain(odr, params.TestChainConfig, ethash.NewFullFaker()) if err != nil { t.Fatal(err) } diff --git a/light/postprocess.go b/light/postprocess.go index 07fdb53b1f..7c71911057 100644 --- a/light/postprocess.go +++ b/light/postprocess.go @@ -59,27 +59,48 @@ type IndexerConfig struct { BloomTrieConfirm uint64 } -// DefaultServerIndexerConfig wraps a set of configs as a default indexer config for server side. -var DefaultServerIndexerConfig = &IndexerConfig{ - ChtSize: params.CHTFrequencyServer, - PairChtSize: params.CHTFrequencyClient, - ChtConfirm: params.HelperTrieProcessConfirmations, - BloomSize: params.BloomBitsBlocks, - BloomConfirm: params.BloomConfirms, - BloomTrieSize: params.BloomTrieFrequency, - BloomTrieConfirm: params.HelperTrieProcessConfirmations, -} - -// DefaultClientIndexerConfig wraps a set of configs as a default indexer config for client side. -var DefaultClientIndexerConfig = &IndexerConfig{ - ChtSize: params.CHTFrequencyClient, - PairChtSize: params.CHTFrequencyServer, - ChtConfirm: params.HelperTrieConfirmations, - BloomSize: params.BloomBitsBlocksClient, - BloomConfirm: params.HelperTrieConfirmations, - BloomTrieSize: params.BloomTrieFrequency, - BloomTrieConfirm: params.HelperTrieConfirmations, -} +var ( + // DefaultServerIndexerConfig wraps a set of configs as a default indexer config for server side. + DefaultServerIndexerConfig = &IndexerConfig{ + ChtSize: params.CHTFrequencyServer, + PairChtSize: params.CHTFrequencyClient, + ChtConfirm: params.HelperTrieProcessConfirmations, + BloomSize: params.BloomBitsBlocks, + BloomConfirm: params.BloomConfirms, + BloomTrieSize: params.BloomTrieFrequency, + BloomTrieConfirm: params.HelperTrieProcessConfirmations, + } + // DefaultClientIndexerConfig wraps a set of configs as a default indexer config for client side. + DefaultClientIndexerConfig = &IndexerConfig{ + ChtSize: params.CHTFrequencyClient, + PairChtSize: params.CHTFrequencyServer, + ChtConfirm: params.HelperTrieConfirmations, + BloomSize: params.BloomBitsBlocksClient, + BloomConfirm: params.HelperTrieConfirmations, + BloomTrieSize: params.BloomTrieFrequency, + BloomTrieConfirm: params.HelperTrieConfirmations, + } + // TestServerIndexerConfig wraps a set of configs as a test indexer config for server side. + TestServerIndexerConfig = &IndexerConfig{ + ChtSize: 256, + PairChtSize: 2048, + ChtConfirm: 16, + BloomSize: 256, + BloomConfirm: 16, + BloomTrieSize: 2048, + BloomTrieConfirm: 16, + } + // TestClientIndexerConfig wraps a set of configs as a test indexer config for client side. + TestClientIndexerConfig = &IndexerConfig{ + ChtSize: 2048, + PairChtSize: 256, + ChtConfirm: 128, + BloomSize: 2048, + BloomConfirm: 128, + BloomTrieSize: 2048, + BloomTrieConfirm: 128, + } +) // 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 diff --git a/light/trie_test.go b/light/trie_test.go index f366a904ce..6bddfefe28 100644 --- a/light/trie_test.go +++ b/light/trie_test.go @@ -47,7 +47,7 @@ func TestNodeIterator(t *testing.T) { } ctx := context.Background() - odr := &testOdr{sdb: fulldb, ldb: lightdb, indexerConfig: DefaultClientIndexerConfig} + odr := &testOdr{sdb: fulldb, ldb: lightdb, indexerConfig: TestClientIndexerConfig} head := blockchain.CurrentHeader() lightTrie, _ := NewStateDatabase(ctx, head, odr).OpenTrie(head.Root) fullTrie, _ := state.NewDatabase(fulldb).OpenTrie(head.Root) diff --git a/light/txpool_test.go b/light/txpool_test.go index bcaae73f20..204347a6e4 100644 --- a/light/txpool_test.go +++ b/light/txpool_test.go @@ -94,13 +94,13 @@ func TestTxPool(t *testing.T) { panic(err) } - odr := &testOdr{sdb: sdb, ldb: ldb, indexerConfig: DefaultClientIndexerConfig} + odr := &testOdr{sdb: sdb, ldb: ldb, indexerConfig: TestClientIndexerConfig} relay := &testTxRelay{ send: make(chan int, 1), discard: make(chan int, 1), mined: make(chan int, 1), } - lightchain, _ := NewLightChain(odr, params.TestChainConfig, DefaultClientIndexerConfig, ethash.NewFullFaker()) + lightchain, _ := NewLightChain(odr, params.TestChainConfig, ethash.NewFullFaker()) txPermanent = 50 pool := NewTxPool(params.TestChainConfig, lightchain, relay) ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)