From 4644814a85da1284dda5eba0ef399df0d4bec440 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 3 Jul 2015 11:55:37 +0300 Subject: [PATCH] eth: add state retrieval, test it + block tests --- eth/backend.go | 2 +- eth/downloader/downloader.go | 1 + eth/handler.go | 45 +++++-- eth/handler_test.go | 235 ++++++++++++++++++++++++++++++++++- eth/helper_test.go | 23 ++-- eth/protocol_test.go | 6 +- ethdb/memory_database.go | 8 ++ 7 files changed, 295 insertions(+), 25 deletions(-) diff --git a/eth/backend.go b/eth/backend.go index 953a150ad3..ac6ebf15b5 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -372,7 +372,7 @@ func New(config *Config) (*Ethereum, error) { eth.blockProcessor = core.NewBlockProcessor(chainDb, eth.pow, eth.chainManager, eth.EventMux()) eth.chainManager.SetProcessor(eth.blockProcessor) - eth.protocolManager = NewProtocolManager(config.NetworkId, eth.eventMux, eth.txPool, eth.pow, eth.chainManager) + eth.protocolManager = NewProtocolManager(config.NetworkId, eth.eventMux, eth.txPool, eth.pow, eth.chainManager, chainDb) eth.miner = miner.New(eth, eth.EventMux(), eth.pow) eth.miner.SetGasPrice(config.GasPrice) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 27075fcf68..088e441e08 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -47,6 +47,7 @@ var ( MaxHashFetch = 512 // Amount of hashes to be fetched per retrieval request MaxBlockFetch = 128 // Amount of blocks to be fetched per retrieval request MaxHeaderFetch = 256 // Amount of block headers to be fetched per retrieval request + MaxStateFetch = 384 // Amount of node hashTTL = 5 * time.Second // Time it takes for a hash request to time out blockSoftTTL = 3 * time.Second // Request completion threshold for increasing or decreasing a peer's bandwidth diff --git a/eth/handler.go b/eth/handler.go index 7fbb47fb18..102901c086 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -36,7 +36,7 @@ import ( "github.com/ethereum/go-ethereum/rlp" ) -// This is the target maximum size of returned blocks or headers. +// This is the target maximum size of returned blocks, headers or node data. const softResponseLimit = 2 * 1024 * 1024 func errResp(code errCode, format string, v ...interface{}) error { @@ -59,6 +59,7 @@ func (ep extProt) GetBlock(hashes []common.Hash) error { return ep.getBlocks(has type ProtocolManager struct { txpool txPool chainman *core.ChainManager + statedb common.Database downloader *downloader.Downloader fetcher *fetcher.Fetcher peers *peerSet @@ -82,12 +83,13 @@ type ProtocolManager struct { // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable // with the ethereum network. -func NewProtocolManager(networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager) *ProtocolManager { +func NewProtocolManager(networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager, statedb common.Database) *ProtocolManager { // Create the protocol manager with the base fields manager := &ProtocolManager{ eventMux: mux, txpool: txpool, chainman: chainman, + statedb: statedb, peers: newPeerSet(), newPeerCh: make(chan *peer, 1), txsyncCh: make(chan *txsync), @@ -276,10 +278,8 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { case BlockHashesMsg: // A batch of hashes arrived to one of our previous requests - msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) - var hashes []common.Hash - if err := msgStream.Decode(&hashes); err != nil { + if err := msg.Decode(&hashes); err != nil { break } // Deliver them all to the downloader for queuing @@ -318,10 +318,8 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { case BlocksMsg: // Decode the arrived block message - msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) - var blocks []*types.Block - if err := msgStream.Decode(&blocks); err != nil { + if err := msg.Decode(&blocks); err != nil { glog.V(logger.Detail).Infoln("Decode error", err) blocks = nil } @@ -361,12 +359,37 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendBlockHeaders(headers) + case GetNodeDataMsg: + // Decode the retrieval message + msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) + if _, err := msgStream.List(); err != nil { + return err + } + // Gather state data until the fetch or network limits is reached + var ( + hash common.Hash + bytes int + data [][]byte + ) + for bytes < softResponseLimit && len(data) < downloader.MaxStateFetch { + //Retrieve the hash of the next state entry + if err := msgStream.Decode(&hash); err == rlp.EOL { + break + } else if err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + // Retrieve the requested state entry, stopping if enough was found + if entry, err := pm.statedb.Get(hash.Bytes()); err == nil { + data = append(data, entry) + bytes += len(entry) + } + } + return p.SendNodeData(data) + case NewBlockHashesMsg: // Retrieve and deseralize the remote new block hashes notification - msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) - var hashes []common.Hash - if err := msgStream.Decode(&hashes); err != nil { + if err := msg.Decode(&hashes); err != nil { break } // Mark the hashes as present at the remote node diff --git a/eth/handler_test.go b/eth/handler_test.go index f19717db5b..7cc7db16f4 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -1,17 +1,26 @@ package eth import ( + "fmt" + "math/big" + "math/rand" "testing" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/eth/downloader" + "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/params" ) // Tests that hashes can be retrieved from a remote chain by hashes in reverse // order. func TestGetBlockHashes(t *testing.T) { - pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil) + pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil) peer, _ := newTestPeer("peer", pm, true) defer peer.close() @@ -52,7 +61,7 @@ func TestGetBlockHashes(t *testing.T) { // Tests that hashes can be retrieved from a remote chain by numbers in forward // order. func TestGetBlockHashesFromNumber(t *testing.T) { - pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil) + pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil) peer, _ := newTestPeer("peer", pm, true) defer peer.close() @@ -87,3 +96,225 @@ func TestGetBlockHashesFromNumber(t *testing.T) { } } } + +// Tests that blocks can be retrieved from a remote chain based on their hashes. +func TestGetBlocks(t *testing.T) { + pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil) + peer, _ := newTestPeer("peer", pm, true) + defer peer.close() + + // Create a batch of tests for various scenarios + limit := downloader.MaxBlockFetch + tests := []struct { + random int // Number of blocks to fetch randomly from the chain + explicit []common.Hash // Explicitly requested blocks + available []bool // Availability of explicitly requested blocks + expected int // Total number of existing blocks to expect + }{ + {1, nil, nil, 1}, // A single random block should be retrievable + {10, nil, nil, 10}, // Multiple random blocks should be retrievable + {limit, nil, nil, limit}, // The maximum possible blocks should be retrievable + {limit + 1, nil, nil, limit}, // No more that the possible block count should be returned + {0, []common.Hash{pm.chainman.Genesis().Hash()}, []bool{true}, 1}, // The genesis block should be retrievable + {0, []common.Hash{pm.chainman.CurrentBlock().Hash()}, []bool{true}, 1}, // The chains head block should be retrievable + {0, []common.Hash{common.Hash{}}, []bool{false}, 0}, // A non existent block should not be returned + + // Existing and non-existing blocks interleaved should not cause problems + {0, []common.Hash{ + common.Hash{}, + pm.chainman.GetBlockByNumber(1).Hash(), + common.Hash{}, + pm.chainman.GetBlockByNumber(10).Hash(), + common.Hash{}, + pm.chainman.GetBlockByNumber(100).Hash(), + common.Hash{}, + }, []bool{false, true, false, true, false, true, false}, 3}, + } + // Run each of the tests and verify the results against the chain + for i, tt := range tests { + // Collect the hashes to request, and the response to expect + hashes, seen := []common.Hash{}, make(map[int64]bool) + blocks := []*types.Block{} + + for j := 0; j < tt.random; j++ { + for { + num := rand.Int63n(int64(pm.chainman.CurrentBlock().NumberU64())) + if !seen[num] { + seen[num] = true + + block := pm.chainman.GetBlockByNumber(uint64(num)) + hashes = append(hashes, block.Hash()) + if len(blocks) < tt.expected { + blocks = append(blocks, block) + } + break + } + } + } + for j, hash := range tt.explicit { + hashes = append(hashes, hash) + if tt.available[j] && len(blocks) < tt.expected { + blocks = append(blocks, pm.chainman.GetBlock(hash)) + } + } + // Send the hash request and verify the response + p2p.Send(peer.app, 0x05, hashes) + if err := p2p.ExpectMsg(peer.app, 0x06, blocks); err != nil { + t.Errorf("test %d: blocks mismatch: %v", i, err) + } + } +} + +// Tests that block headers can be retrieved from a remote chain based on their hashes. +func TestGetBlockHeaders(t *testing.T) { + pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil) + peer, _ := newTestPeer("peer", pm, true) + defer peer.close() + + // Create a batch of tests for various scenarios + limit := downloader.MaxHeaderFetch + tests := []struct { + random int // Number of blocks to fetch randomly from the chain + explicit []common.Hash // Explicitly requested blocks + available []bool // Availability of explicitly requested blocks + expected int // Total number of existing blocks to expect + }{ + {1, nil, nil, 1}, // A single random block should be retrievable + {10, nil, nil, 10}, // Multiple random blocks should be retrievable + {limit, nil, nil, limit}, // The maximum possible blocks should be retrievable + {limit + 1, nil, nil, limit}, // No more that the possible block count should be returned + {0, []common.Hash{pm.chainman.Genesis().Hash()}, []bool{true}, 1}, // The genesis block should be retrievable + {0, []common.Hash{pm.chainman.CurrentBlock().Hash()}, []bool{true}, 1}, // The chains head block should be retrievable + {0, []common.Hash{common.Hash{}}, []bool{false}, 0}, // A non existent block should not be returned + + // Existing and non-existing blocks interleaved should not cause problems + {0, []common.Hash{ + common.Hash{}, + pm.chainman.GetBlockByNumber(1).Hash(), + common.Hash{}, + pm.chainman.GetBlockByNumber(10).Hash(), + common.Hash{}, + pm.chainman.GetBlockByNumber(100).Hash(), + common.Hash{}, + }, []bool{false, true, false, true, false, true, false}, 3}, + } + // Run each of the tests and verify the results against the chain + for i, tt := range tests { + // Collect the hashes to request, and the response to expect + hashes, seen := []common.Hash{}, make(map[int64]bool) + headers := []*types.Header{} + + for j := 0; j < tt.random; j++ { + for { + num := rand.Int63n(int64(pm.chainman.CurrentBlock().NumberU64())) + if !seen[num] { + seen[num] = true + + block := pm.chainman.GetBlockByNumber(uint64(num)) + hashes = append(hashes, block.Hash()) + if len(headers) < tt.expected { + headers = append(headers, block.Header()) + } + break + } + } + } + for j, hash := range tt.explicit { + hashes = append(hashes, hash) + if tt.available[j] && len(headers) < tt.expected { + headers = append(headers, pm.chainman.GetBlock(hash).Header()) + } + } + // Send the hash request and verify the response + p2p.Send(peer.app, 0x09, hashes) + if err := p2p.ExpectMsg(peer.app, 0x0a, headers); err != nil { + t.Errorf("test %d: headers mismatch: %v", i, err) + } + } +} + +// Tests that the node state database can be retrieved based on hashes. +func TestGetNodeData(t *testing.T) { + // Define three accounts to simulate transactions with + acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a") + acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee") + acc1Addr := crypto.PubkeyToAddress(acc1Key.PublicKey) + acc2Addr := crypto.PubkeyToAddress(acc2Key.PublicKey) + + // Create a chain generator with some simple transactions (blatantly stolen from @fjl/chain_makerts_test) + generator := func(i int, block *core.BlockGen) { + switch i { + case 0: + // In block 1, the test bank sends account #1 some ether. + tx, _ := types.NewTransaction(block.TxNonce(testBankAddress), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil).SignECDSA(testBankKey) + block.AddTx(tx) + case 1: + // In block 2, the test bank sends some more ether to account #1. + // acc1Addr passes it on to account #2. + tx1, _ := types.NewTransaction(block.TxNonce(testBankAddress), acc1Addr, big.NewInt(1000), params.TxGas, nil, nil).SignECDSA(testBankKey) + tx2, _ := types.NewTransaction(block.TxNonce(acc1Addr), acc2Addr, big.NewInt(1000), params.TxGas, nil, nil).SignECDSA(acc1Key) + block.AddTx(tx1) + block.AddTx(tx2) + case 2: + // Block 3 is empty but was mined by account #2. + block.SetCoinbase(acc2Addr) + block.SetExtra([]byte("yeehaw")) + case 3: + // Block 4 includes blocks 2 and 3 as uncle headers (with modified extra data). + b2 := block.PrevBlock(1).Header() + b2.Extra = []byte("foo") + block.AddUncle(b2) + b3 := block.PrevBlock(2).Header() + b3.Extra = []byte("foo") + block.AddUncle(b3) + } + } + // Assemble the test environment + pm := newTestProtocolManager(4, generator, nil) + peer, _ := newTestPeer("peer", pm, true) + defer peer.close() + + // Fetch for now the entire state db + hashes := []common.Hash{} + for _, key := range pm.statedb.(*ethdb.MemDatabase).Keys() { + hashes = append(hashes, common.BytesToHash(key)) + } + p2p.Send(peer.app, 0x0b, hashes) + msg, err := peer.app.ReadMsg() + if err != nil { + t.Fatalf("failed to read node data response: %v", err) + } + if msg.Code != 0x0c { + t.Fatalf("response packet code mismatch: have %x, want %x", msg.Code, 0x0c) + } + var data [][]byte + if err := msg.Decode(&data); err != nil { + t.Fatalf("failed to decode response node data: %v", err) + } + // Verify that all hashes correspond to the requested data, and reconstruct a state tree + for i, want := range hashes { + if hash := crypto.Sha3Hash(data[i]); hash != want { + fmt.Errorf("data hash mismatch: have %x, want %x", hash, want) + } + } + statedb, _ := ethdb.NewMemDatabase() + for i := 0; i < len(data); i++ { + statedb.Put(hashes[i].Bytes(), data[i]) + } + accounts := []common.Address{testBankAddress, acc1Addr, acc2Addr} + for i := uint64(0); i <= pm.chainman.CurrentBlock().NumberU64(); i++ { + trie := state.New(pm.chainman.GetBlockByNumber(i).Root(), statedb) + + for j, acc := range accounts { + bw := pm.chainman.State().GetBalance(acc) + bh := trie.GetBalance(acc) + + if (bw != nil && bh == nil) || (bw == nil && bh != nil) { + t.Errorf("test %d, account %d: balance mismatch: have %v, want %v", i, j, bh, bw) + } + if bw != nil && bh != nil && bw.Cmp(bw) != 0 { + t.Errorf("test %d, account %d: balance mismatch: have %v, want %v", i, j, bh, bw) + } + } + } +} diff --git a/eth/helper_test.go b/eth/helper_test.go index efbf6f9706..533eaff6e4 100644 --- a/eth/helper_test.go +++ b/eth/helper_test.go @@ -19,22 +19,29 @@ import ( "github.com/ethereum/go-ethereum/p2p/discover" ) +var ( + testBankKey, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + testBankAddress = crypto.PubkeyToAddress(testBankKey.PublicKey) + testBankFunds = big.NewInt(1000000) +) + // 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(blocks int, newtx chan<- []*types.Transaction) *ProtocolManager { +func newTestProtocolManager(blocks int, generator func(int, *core.BlockGen), newtx chan<- []*types.Transaction) *ProtocolManager { var ( - emux = new(event.TypeMux) + evmux = new(event.TypeMux) pow = new(core.FakePow) db, _ = ethdb.NewMemDatabase() - genesis = core.WriteGenesisBlockForTesting(db, common.Address{}, big.NewInt(0)) - chainman, _ = core.NewChainManager(db, pow, emux) - blockproc = core.NewBlockProcessor(db, pow, chainman, emux) + genesis = core.WriteGenesisBlockForTesting(db, testBankAddress, testBankFunds) + chainman, _ = core.NewChainManager(db, pow, evmux) + blockproc = core.NewBlockProcessor(db, pow, chainman, evmux) ) chainman.SetProcessor(blockproc) - chainman.InsertChain(core.GenerateChain(genesis, db, blocks, nil)) - - pm := NewProtocolManager(NetworkId, emux, &testTxPool{added: newtx}, pow, chainman) + if _, err := chainman.InsertChain(core.GenerateChain(genesis, db, blocks, generator)); err != nil { + panic(err) + } + pm := NewProtocolManager(NetworkId, evmux, &testTxPool{added: newtx}, pow, chainman, db) pm.Start() return pm } diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 9a6964be2d..166488b79a 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -37,7 +37,7 @@ func init() { var testAccount = crypto.NewKey(rand.Reader) func TestStatusMsgErrors(t *testing.T) { - pm := newTestProtocolManager(0, nil) + pm := newTestProtocolManager(0, nil, nil) td, currentBlock, genesis := pm.chainman.Status() defer pm.Stop() @@ -87,7 +87,7 @@ func TestStatusMsgErrors(t *testing.T) { // This test checks that received transactions are added to the local pool. func TestRecvTransactions(t *testing.T) { txAdded := make(chan []*types.Transaction) - pm := newTestProtocolManager(0, txAdded) + pm := newTestProtocolManager(0, nil, txAdded) p, _ := newTestPeer("peer", pm, true) defer pm.Stop() defer p.close() @@ -110,7 +110,7 @@ func TestRecvTransactions(t *testing.T) { // This test checks that pending transactions are sent. func TestSendTransactions(t *testing.T) { - pm := newTestProtocolManager(0, nil) + pm := newTestProtocolManager(0, nil, nil) defer pm.Stop() // Fill the pool with big transactions. diff --git a/ethdb/memory_database.go b/ethdb/memory_database.go index 70b03dfade..d50f8f9d44 100644 --- a/ethdb/memory_database.go +++ b/ethdb/memory_database.go @@ -49,6 +49,14 @@ func (db *MemDatabase) Get(key []byte) ([]byte, error) { return db.db[string(key)], nil } +func (db *MemDatabase) Keys() [][]byte { + keys := [][]byte{} + for key, _ := range db.db { + keys = append(keys, []byte(key)) + } + return keys +} + /* func (db *MemDatabase) GetKeys() []*common.Key { data, _ := db.Get([]byte("KeyRing"))