les: fixes

This commit is contained in:
rjl493456442 2018-07-25 13:01:21 +08:00
parent da8abaf8cb
commit 8ff9a7613f
8 changed files with 94 additions and 93 deletions

View file

@ -1149,9 +1149,9 @@ func (pm *ProtocolManager) getAccount(statedb *state.StateDB, root, hash common.
func (pm *ProtocolManager) getHelperTrie(id uint, idx uint64) (common.Hash, string) { func (pm *ProtocolManager) getHelperTrie(id uint, idx uint64) (common.Hash, string) {
switch id { switch id {
case htCanonical: case htCanonical:
idxV2 := (idx+1)*(pm.indexerConfig.PairChtSize/pm.indexerConfig.ChtSize) - 1 idxV1 := (idx+1)*(pm.indexerConfig.PairChtSize/pm.indexerConfig.ChtSize) - 1
sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*pm.indexerConfig.PairChtSize-1) sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idxV1+1)*pm.indexerConfig.ChtSize-1)
return light.GetChtRoot(pm.chainDb, idxV2, sectionHead), light.ChtTablePrefix return light.GetChtRoot(pm.chainDb, idxV1, sectionHead), light.ChtTablePrefix
case htBloomBits: case htBloomBits:
sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*pm.indexerConfig.BloomTrieSize-1) sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*pm.indexerConfig.BloomTrieSize-1)
return light.GetBloomTrieRoot(pm.chainDb, idx, sectionHead), light.BloomTrieTablePrefix return light.GetBloomTrieRoot(pm.chainDb, idx, sectionHead), light.BloomTrieTablePrefix

View file

@ -51,12 +51,8 @@ func TestGetBlockHeadersLes1(t *testing.T) { testGetBlockHeaders(t, 1) }
func TestGetBlockHeadersLes2(t *testing.T) { testGetBlockHeaders(t, 2) } func TestGetBlockHeadersLes2(t *testing.T) { testGetBlockHeaders(t, 2) }
func testGetBlockHeaders(t *testing.T, protocol int) { func testGetBlockHeaders(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, 0) server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, nil)
defer func() { defer tearDown()
if tearDown != nil {
tearDown()
}
}()
bc := server.pm.blockchain.(*core.BlockChain) bc := server.pm.blockchain.(*core.BlockChain)
// Create a "random" unknown hash for testing // Create a "random" unknown hash for testing
@ -183,12 +179,8 @@ func TestGetBlockBodiesLes1(t *testing.T) { testGetBlockBodies(t, 1) }
func TestGetBlockBodiesLes2(t *testing.T) { testGetBlockBodies(t, 2) } func TestGetBlockBodiesLes2(t *testing.T) { testGetBlockBodies(t, 2) }
func testGetBlockBodies(t *testing.T, protocol int) { func testGetBlockBodies(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, protocol, 0) server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, protocol, nil)
defer func() { defer tearDown()
if tearDown != nil {
tearDown()
}
}()
bc := server.pm.blockchain.(*core.BlockChain) bc := server.pm.blockchain.(*core.BlockChain)
// Create a batch of tests for various scenarios // Create a batch of tests for various scenarios
@ -263,12 +255,8 @@ func TestGetCodeLes2(t *testing.T) { testGetCode(t, 2) }
func testGetCode(t *testing.T, protocol int) { func testGetCode(t *testing.T, protocol int) {
// Assemble the test environment // Assemble the test environment
server, tearDown := newServerEnv(t, 4, protocol, 0) server, tearDown := newServerEnv(t, 4, protocol, nil)
defer func() { defer tearDown()
if tearDown != nil {
tearDown()
}
}()
bc := server.pm.blockchain.(*core.BlockChain) bc := server.pm.blockchain.(*core.BlockChain)
var codereqs []*CodeReq var codereqs []*CodeReq
@ -299,12 +287,8 @@ func TestGetReceiptLes2(t *testing.T) { testGetReceipt(t, 2) }
func testGetReceipt(t *testing.T, protocol int) { func testGetReceipt(t *testing.T, protocol int) {
// Assemble the test environment // Assemble the test environment
server, tearDown := newServerEnv(t, 4, protocol, 0) server, tearDown := newServerEnv(t, 4, protocol, nil)
defer func() { defer tearDown()
if tearDown != nil {
tearDown()
}
}()
bc := server.pm.blockchain.(*core.BlockChain) bc := server.pm.blockchain.(*core.BlockChain)
// Collect the hashes to request, and the response to expect // Collect the hashes to request, and the response to expect
@ -329,12 +313,8 @@ func TestGetProofsLes2(t *testing.T) { testGetProofs(t, 2) }
func testGetProofs(t *testing.T, protocol int) { func testGetProofs(t *testing.T, protocol int) {
// Assemble the test environment // Assemble the test environment
server, tearDown := newServerEnv(t, 4, protocol, 0) server, tearDown := newServerEnv(t, 4, protocol, nil)
defer func() { defer tearDown()
if tearDown != nil {
tearDown()
}
}()
bc := server.pm.blockchain.(*core.BlockChain) bc := server.pm.blockchain.(*core.BlockChain)
var ( var (
@ -391,23 +371,30 @@ func testGetCHTProofs(t *testing.T, protocol int) {
config := light.TestServerIndexerConfig config := light.TestServerIndexerConfig
frequency := config.ChtSize frequency := config.ChtSize
if protocol == 2 { if protocol == 2 {
frequency = config.ChtSize * 8 frequency = config.PairChtSize
} }
server, tearDown := newServerEnv(t, int(frequency+config.ChtConfirm), protocol, 1) waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) {
defer func() { expectSections := frequency / config.ChtSize
if tearDown != nil { for {
tearDown() cs, _, _ := cIndexer.Sections()
bs, _, _ := bIndexer.Sections()
if cs >= expectSections && bs >= expectSections {
break
} }
}() time.Sleep(10 * time.Millisecond)
}
}
server, tearDown := newServerEnv(t, int(frequency+config.ChtConfirm), protocol, waitIndexers)
defer tearDown()
bc := server.pm.blockchain.(*core.BlockChain) bc := server.pm.blockchain.(*core.BlockChain)
// Assemble the proofs from the different protocols // Assemble the proofs from the different protocols
header := bc.GetHeaderByNumber(config.ChtSize - 1) header := bc.GetHeaderByNumber(frequency - 1)
rlp, _ := rlp.EncodeToBytes(header) rlp, _ := rlp.EncodeToBytes(header)
key := make([]byte, 8) key := make([]byte, 8)
binary.BigEndian.PutUint64(key, config.ChtSize-1) binary.BigEndian.PutUint64(key, frequency-1)
proofsV1 := []ChtResp{{ proofsV1 := []ChtResp{{
Header: header, Header: header,
@ -417,7 +404,7 @@ func testGetCHTProofs(t *testing.T, protocol int) {
} }
switch protocol { switch protocol {
case 1: case 1:
root := light.GetChtRoot(server.db, 0, bc.GetHeaderByNumber(config.ChtSize-1).Hash()) root := light.GetChtRoot(server.db, 0, bc.GetHeaderByNumber(frequency-1).Hash())
trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.ChtTablePrefix))) trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.ChtTablePrefix)))
var proof light.NodeList var proof light.NodeList
@ -425,18 +412,18 @@ func testGetCHTProofs(t *testing.T, protocol int) {
proofsV1[0].Proof = proof proofsV1[0].Proof = proof
case 2: case 2:
root := light.GetChtRoot(server.db, (config.ChtSize/config.PairChtSize)-1, bc.GetHeaderByNumber(config.ChtSize-1).Hash()) root := light.GetChtRoot(server.db, (frequency/config.ChtSize)-1, bc.GetHeaderByNumber(frequency-1).Hash())
trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.ChtTablePrefix))) trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.ChtTablePrefix)))
trie.Prove(key, 0, &proofsV2.Proofs) trie.Prove(key, 0, &proofsV2.Proofs)
} }
// Assemble the requests for the different protocols // Assemble the requests for the different protocols
requestsV1 := []ChtReq{{ requestsV1 := []ChtReq{{
ChtNum: 1, ChtNum: frequency / config.ChtSize,
BlockNum: config.ChtSize - 1, BlockNum: frequency - 1,
}} }}
requestsV2 := []HelperTrieReq{{ requestsV2 := []HelperTrieReq{{
Type: htCanonical, Type: htCanonical,
TrieIdx: 0, TrieIdx: frequency/config.PairChtSize - 1,
Key: key, Key: key,
AuxReq: auxHeader, AuxReq: auxHeader,
}} }}
@ -460,12 +447,20 @@ func testGetCHTProofs(t *testing.T, protocol int) {
// Tests that bloombits proofs can be correctly retrieved. // Tests that bloombits proofs can be correctly retrieved.
func TestGetBloombitsProofs(t *testing.T) { func TestGetBloombitsProofs(t *testing.T) {
config := light.TestServerIndexerConfig config := light.TestServerIndexerConfig
server, tearDown := newServerEnv(t, int(config.BloomSize+config.BloomConfirm), 2, 1)
defer func() { waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) {
if tearDown != nil { for {
tearDown() cs, _, _ := cIndexer.Sections()
bs, _, _ := bIndexer.Sections()
bts, _, _ := btIndexer.Sections()
if cs >= 8 && bs >= 8 && bts >= 1 {
break
} }
}() time.Sleep(10 * time.Millisecond)
}
}
server, tearDown := newServerEnv(t, int(config.BloomTrieSize+config.BloomTrieConfirm), 2, waitIndexers)
defer tearDown()
bc := server.pm.blockchain.(*core.BlockChain) bc := server.pm.blockchain.(*core.BlockChain)
// Request and verify each bit of the bloom bits proofs // Request and verify each bit of the bloom bits proofs
@ -474,7 +469,8 @@ func TestGetBloombitsProofs(t *testing.T) {
key := make([]byte, 10) key := make([]byte, 10)
binary.BigEndian.PutUint16(key[:2], uint16(bit)) binary.BigEndian.PutUint16(key[:2], uint16(bit))
binary.BigEndian.PutUint64(key[2:], config.BloomSize-1) // Only the first bloom section has data.
binary.BigEndian.PutUint64(key[2:], 0)
requests := []HelperTrieReq{{ requests := []HelperTrieReq{{
Type: htBloomBits, Type: htBloomBits,
@ -483,7 +479,7 @@ func TestGetBloombitsProofs(t *testing.T) {
}} }}
var proofs HelperTrieResps var proofs HelperTrieResps
root := light.GetBloomTrieRoot(server.db, 0, bc.GetHeaderByNumber(config.BloomSize-1).Hash()) root := light.GetBloomTrieRoot(server.db, 0, bc.GetHeaderByNumber(config.BloomTrieSize-1).Hash())
trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.BloomTrieTablePrefix))) trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.BloomTrieTablePrefix)))
trie.Prove(key, 0, &proofs.Proofs) trie.Prove(key, 0, &proofs.Proofs)

View file

@ -348,7 +348,7 @@ type TestEntity struct {
} }
// newServerEnv creates a server testing environment with a connected test peer for testing purpose. // 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()) { func newServerEnv(t *testing.T, blocks int, protocol int, waitIndexers func(*core.ChainIndexer, *core.ChainIndexer, *core.ChainIndexer)) (*TestEntity, func()) {
db := ethdb.NewMemDatabase() db := ethdb.NewMemDatabase()
cIndexer, bIndexer, btIndexer := testIndexers(db, light.TestServerIndexerConfig) cIndexer, bIndexer, btIndexer := testIndexers(db, light.TestServerIndexerConfig)
@ -359,16 +359,10 @@ func newServerEnv(t *testing.T, blocks int, protocol int, processSections uint64
bIndexer.Start(pm.blockchain.(*core.BlockChain)) bIndexer.Start(pm.blockchain.(*core.BlockChain))
// Wait until indexers generate enough index data. // Wait until indexers generate enough index data.
if processSections > 0 { if waitIndexers != nil {
for { waitIndexers(cIndexer, bIndexer, btIndexer)
cs, _, _ := cIndexer.Sections()
bs, _, _ := bIndexer.Sections()
if cs >= processSections && bs >= processSections {
break
}
time.Sleep(10 * time.Millisecond)
}
} }
return &TestEntity{ return &TestEntity{
db: db, db: db,
tPeer: peer, tPeer: peer,
@ -386,7 +380,7 @@ func newServerEnv(t *testing.T, blocks int, protocol int, processSections uint64
// newClientServerEnv creates a client/server arch environment with a connected les server and light client pair // newClientServerEnv creates a client/server arch environment with a connected les server and light client pair
// for testing purpose. // for testing purpose.
func newClientServerEnv(t *testing.T, blocks int, protocol int) (*TestEntity, *TestEntity, func()) { func newClientServerEnv(t *testing.T, blocks int, protocol int, waitIndexers func(*core.ChainIndexer, *core.ChainIndexer, *core.ChainIndexer), newPeer bool) (*TestEntity, *TestEntity, func()) {
db, ldb := ethdb.NewMemDatabase(), ethdb.NewMemDatabase() db, ldb := ethdb.NewMemDatabase(), ethdb.NewMemDatabase()
peers, lPeers := newPeerSet(), newPeerSet() peers, lPeers := newPeerSet(), newPeerSet()
@ -411,7 +405,17 @@ func newClientServerEnv(t *testing.T, blocks int, protocol int) (*TestEntity, *T
startIndexers(false, pm) startIndexers(false, pm)
startIndexers(true, lpm) startIndexers(true, lpm)
peer, err1, lPeer, err2 := newTestPeerPair("peer", protocol, pm, lpm) // Execute wait until function if it is specified.
if waitIndexers != nil {
waitIndexers(cIndexer, bIndexer, btIndexer)
}
var (
peer, lPeer *peer
err1, err2 <-chan error
)
if newPeer {
peer, err1, lPeer, err2 = newTestPeerPair("peer", protocol, pm, lpm)
select { select {
case <-time.After(time.Millisecond * 100): case <-time.After(time.Millisecond * 100):
case err := <-err1: case err := <-err1:
@ -419,6 +423,7 @@ func newClientServerEnv(t *testing.T, blocks int, protocol int) (*TestEntity, *T
case err := <-err2: case err := <-err2:
t.Fatalf("peer 2 handshake error: %v", err) t.Fatalf("peer 2 handshake error: %v", err)
} }
}
return &TestEntity{ return &TestEntity{
db: db, db: db,

View file

@ -387,10 +387,10 @@ func (r *ChtRequest) Request(reqID uint64, peer *peer, config *light.IndexerConf
} }
blockNum := binary.BigEndian.Uint64(req.Key) blockNum := binary.BigEndian.Uint64(req.Key)
// convert HelperTrie request to old CHT request // convert HelperTrie request to old CHT request
reqsV1 = ChtReq{ChtNum: (req.TrieIdx+1)*(config.ChtSize/config.PairChtSize) - 1, BlockNum: blockNum, FromLevel: req.FromLevel} reqsV1 = ChtReq{ChtNum: (req.TrieIdx + 1) * (config.ChtSize / config.PairChtSize), BlockNum: blockNum, FromLevel: req.FromLevel}
return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []interface{}{reqsV1}) return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []ChtReq{reqsV1})
case lpv2: case lpv2:
return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []interface{}{req}) return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []HelperTrieReq{req})
default: default:
panic(nil) panic(nil)
} }
@ -517,7 +517,7 @@ func (r *BloomRequest) Request(reqID uint64, peer *peer, config *light.IndexerCo
Key: common.CopyBytes(encNumber[:]), Key: common.CopyBytes(encNumber[:]),
} }
} }
return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []interface{}{reqs}) return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), reqs)
} }
// Valid processes an ODR request reply message from the LES network // Valid processes an ODR request reply message from the LES network

View file

@ -162,12 +162,8 @@ func odrContractCall(ctx context.Context, db ethdb.Database, config *params.Chai
// testOdr tests odr requests whose validation guaranteed by block headers. // testOdr tests odr requests whose validation guaranteed by block headers.
func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) { func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) {
// Assemble the test environment // Assemble the test environment
server, client, tearDown := newClientServerEnv(t, 4, protocol) server, client, tearDown := newClientServerEnv(t, 4, protocol, nil, true)
defer func() { defer tearDown()
if tearDown != nil {
tearDown()
}
}()
client.pm.synchronise(client.rPeer) client.pm.synchronise(client.rPeer)
test := func(expFail uint64) { test := func(expFail uint64) {

View file

@ -38,6 +38,7 @@ var (
errClosed = errors.New("peer set is closed") errClosed = errors.New("peer set is closed")
errAlreadyRegistered = errors.New("peer is already registered") errAlreadyRegistered = errors.New("peer is already registered")
errNotRegistered = errors.New("peer is not registered") errNotRegistered = errors.New("peer is not registered")
errInvalidHelpTrieReq = errors.New("invalid help trie request")
) )
const maxResponseErrors = 50 // number of invalid responses tolerated (makes the protocol less brittle but still avoids spam) const maxResponseErrors = 50 // number of invalid responses tolerated (makes the protocol less brittle but still avoids spam)
@ -283,12 +284,20 @@ func (p *peer) RequestProofs(reqID, cost uint64, reqs []ProofReq) error {
} }
// RequestHelperTrieProofs fetches a batch of HelperTrie merkle proofs from a remote node. // RequestHelperTrieProofs fetches a batch of HelperTrie merkle proofs from a remote node.
func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, reqs []interface{}) error { func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, data interface{}) error {
p.Log().Debug("Fetching batch of HelperTrie proofs", "count", len(reqs))
switch p.version { switch p.version {
case lpv1: case lpv1:
reqs, ok := data.([]ChtReq)
if !ok {
return errInvalidHelpTrieReq
}
p.Log().Debug("Fetching batch of HelperTrie proofs", "count", len(reqs))
return sendRequest(p.rw, GetHeaderProofsMsg, reqID, cost, reqs) return sendRequest(p.rw, GetHeaderProofsMsg, reqID, cost, reqs)
case lpv2: case lpv2:
reqs, ok := data.([]HelperTrieReq)
if !ok {
return errInvalidHelpTrieReq
}
return sendRequest(p.rw, GetHelperTrieProofsMsg, reqID, cost, reqs) return sendRequest(p.rw, GetHelperTrieProofsMsg, reqID, cost, reqs)
default: default:
panic(nil) panic(nil)

View file

@ -83,12 +83,8 @@ func tfCodeAccess(db ethdb.Database, bhash common.Hash, num uint64) light.OdrReq
func testAccess(t *testing.T, protocol int, fn accessTestFn) { func testAccess(t *testing.T, protocol int, fn accessTestFn) {
// Assemble the test environment // Assemble the test environment
server, client, tearDown := newClientServerEnv(t, 4, protocol) server, client, tearDown := newClientServerEnv(t, 4, protocol, nil, true)
defer func() { defer tearDown()
if tearDown != nil {
tearDown()
}
}()
client.pm.synchronise(client.rPeer) client.pm.synchronise(client.rPeer)
test := func(expFail uint64) { test := func(expFail uint64) {

View file

@ -335,6 +335,5 @@ func (b *BloomTrieIndexerBackend) Commit() error {
sectionHead := b.sectionHeads[b.bloomTrieRatio-1] 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", sectionHead, "root", root, "compression", float64(compSize)/float64(decompSize))
StoreBloomTrieRoot(b.diskdb, b.section, sectionHead, root) StoreBloomTrieRoot(b.diskdb, b.section, sectionHead, root)
return nil return nil
} }