From 6151b2316bb4228cdb4de60b8041706ea3cc7ae8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 2 Jul 2015 19:55:18 +0300 Subject: [PATCH] eth, eth/downloader: handle header requests, table driven proto tests --- eth/downloader/downloader.go | 7 +- eth/handler.go | 63 +++++++++------- eth/handler_test.go | 89 ++++++++++++++++++++++ eth/helper_test.go | 140 +++++++++++++++++++++++++++++++++++ eth/protocol_test.go | 119 ++++------------------------- 5 files changed, 282 insertions(+), 136 deletions(-) create mode 100644 eth/handler_test.go create mode 100644 eth/helper_test.go diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 23d2e045e9..27075fcf68 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -43,9 +43,10 @@ const ( ) var ( - MinHashFetch = 512 // Minimum amount of hashes to not consider a peer stalling - MaxHashFetch = 512 // Amount of hashes to be fetched per retrieval request - MaxBlockFetch = 128 // Amount of blocks to be fetched per retrieval request + MinHashFetch = 512 // Minimum amount of hashes to not consider a peer stalling + 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 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 55f37656db..7fbb47fb18 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -36,10 +36,8 @@ import ( "github.com/ethereum/go-ethereum/rlp" ) -// This is the target maximum size of returned blocks for the -// getBlocks message. The reply message may exceed it -// if a single block is larger than the limit. -const maxBlockRespSize = 2 * 1024 * 1024 +// This is the target maximum size of returned blocks or headers. +const softResponseLimit = 2 * 1024 * 1024 func errResp(code errCode, format string, v ...interface{}) error { return fmt.Errorf("%v - %v", code, fmt.Sprintf(format, v...)) @@ -59,12 +57,11 @@ func (ep extProt) GetHashes(hash common.Hash) error { return ep.getHashes(has func (ep extProt) GetBlock(hashes []common.Hash) error { return ep.getBlocks(hashes) } type ProtocolManager struct { - protVer, netId int - txpool txPool - chainman *core.ChainManager - downloader *downloader.Downloader - fetcher *fetcher.Fetcher - peers *peerSet + txpool txPool + chainman *core.ChainManager + downloader *downloader.Downloader + fetcher *fetcher.Fetcher + peers *peerSet SubProtocols []p2p.Protocol @@ -95,7 +92,6 @@ func NewProtocolManager(networkId int, mux *event.TypeMux, txpool txPool, pow po newPeerCh: make(chan *peer, 1), txsyncCh: make(chan *txsync), quitSync: make(chan struct{}), - netId: networkId, } // Initiate a sub-protocol for every implemented version we can handle manager.SubProtocols = make([]p2p.Protocol, len(ProtocolVersions)) @@ -302,36 +298,22 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { var ( hash common.Hash bytes common.StorageSize - hashes []common.Hash blocks []*types.Block ) - for { + for len(blocks) < downloader.MaxBlockFetch && bytes < softResponseLimit { + //Retrieve the hash of the next block err := msgStream.Decode(&hash) if err == rlp.EOL { break } else if err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - hashes = append(hashes, hash) - // Retrieve the requested block, stopping if enough was found if block := pm.chainman.GetBlock(hash); block != nil { blocks = append(blocks, block) bytes += block.Size() - if len(blocks) >= downloader.MaxBlockFetch || bytes > maxBlockRespSize { - break - } } } - if glog.V(logger.Detail) && len(blocks) == 0 && len(hashes) > 0 { - list := "[" - for _, hash := range hashes { - list += fmt.Sprintf("%x, ", hash[:4]) - } - list = list[:len(list)-2] + "]" - - glog.Infof("%v: no blocks found for requested hashes %s", p, list) - } return p.SendBlocks(blocks) case BlocksMsg: @@ -352,6 +334,33 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { pm.downloader.DeliverBlocks(p.id, blocks) } + case GetBlockHeadersMsg: + // Decode the retrieval message + msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) + if _, err := msgStream.List(); err != nil { + return err + } + // Gather blocks until the fetch or network limits is reached + var ( + hash common.Hash + bytes common.StorageSize + headers []*types.Header + ) + for bytes < softResponseLimit && len(headers) < downloader.MaxHeaderFetch { + //Retrieve the hash of the next block + if err := msgStream.Decode(&hash); err == rlp.EOL { + break + } else if err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + // Retrieve the requested block, stopping if enough was found + if block := pm.chainman.GetBlock(hash); block != nil { + headers = append(headers, block.Header()) + bytes += block.Size() + } + } + return p.SendBlockHeaders(headers) + case NewBlockHashesMsg: // Retrieve and deseralize the remote new block hashes notification msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) diff --git a/eth/handler_test.go b/eth/handler_test.go new file mode 100644 index 0000000000..f19717db5b --- /dev/null +++ b/eth/handler_test.go @@ -0,0 +1,89 @@ +package eth + +import ( + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/eth/downloader" + "github.com/ethereum/go-ethereum/p2p" +) + +// 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) + peer, _ := newTestPeer("peer", pm, true) + defer peer.close() + + // Create a batch of tests for various scenarios + limit := downloader.MaxHashFetch + tests := []struct { + origin common.Hash + number int + result int + }{ + {common.Hash{}, 1, 0}, // Make sure non existent hashes don't return results + {pm.chainman.Genesis().Hash(), 1, 0}, // There are no hashes to retrieve up from the genesis + {pm.chainman.GetBlockByNumber(5).Hash(), 5, 5}, // All the hashes including the genesis requested + {pm.chainman.GetBlockByNumber(5).Hash(), 10, 5}, // More hashes than available till the genesis requested + {pm.chainman.GetBlockByNumber(100).Hash(), 10, 10}, // All hashes available from the middle of the chain + {pm.chainman.CurrentBlock().Hash(), 10, 10}, // All hashes available from the head of the chain + {pm.chainman.CurrentBlock().Hash(), limit, limit}, // Request the maximum allowed hash count + {pm.chainman.CurrentBlock().Hash(), limit + 1, limit}, // Request more than the maximum allowed hash count + } + // Run each of the tests and verify the results against the chain + for i, tt := range tests { + // Assemble the hash response we would like to receive + resp := make([]common.Hash, tt.result) + if len(resp) > 0 { + from := pm.chainman.GetBlock(tt.origin).NumberU64() - 1 + for j := 0; j < len(resp); j++ { + resp[j] = pm.chainman.GetBlockByNumber(uint64(int(from) - j)).Hash() + } + } + // Send the hash request and verify the response + p2p.Send(peer.app, 0x03, getBlockHashesData{tt.origin, uint64(tt.number)}) + if err := p2p.ExpectMsg(peer.app, 0x04, resp); err != nil { + t.Errorf("test %d: block hashes mismatch: %v", i, err) + } + } +} + +// 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) + peer, _ := newTestPeer("peer", pm, true) + defer peer.close() + + // Create a batch of tests for various scenarios + limit := downloader.MaxHashFetch + tests := []struct { + origin uint64 + number int + result int + }{ + {pm.chainman.CurrentBlock().NumberU64() + 1, 1, 0}, // Out of bounds requests should return empty + {pm.chainman.CurrentBlock().NumberU64(), 1, 1}, // Make sure the head hash can be retrieved + {pm.chainman.CurrentBlock().NumberU64() - 4, 5, 5}, // All hashes, including the head hash requested + {pm.chainman.CurrentBlock().NumberU64() - 4, 10, 5}, // More hashes requested than available till the head + {pm.chainman.CurrentBlock().NumberU64() - 100, 10, 10}, // All hashes available from the middle of the chain + {0, 10, 10}, // All hashes available from the root of the chain + {0, limit, limit}, // Request the maximum allowed hash count + {0, limit + 1, limit}, // Request more than the maximum allowed hash count + {0, 1, 1}, // Make sure the genesis hash can be retrieved + } + // Run each of the tests and verify the results against the chain + for i, tt := range tests { + // Assemble the hash response we would like to receive + resp := make([]common.Hash, tt.result) + for j := 0; j < len(resp); j++ { + resp[j] = pm.chainman.GetBlockByNumber(tt.origin + uint64(j)).Hash() + } + // Send the hash request and verify the response + p2p.Send(peer.app, 0x08, getBlockHashesFromNumberData{tt.origin, uint64(tt.number)}) + if err := p2p.ExpectMsg(peer.app, 0x04, resp); err != nil { + t.Errorf("test %d: block hashes mismatch: %v", i, err) + } + } +} diff --git a/eth/helper_test.go b/eth/helper_test.go new file mode 100644 index 0000000000..efbf6f9706 --- /dev/null +++ b/eth/helper_test.go @@ -0,0 +1,140 @@ +// This file contains some shares testing functionality, common to multiple +// different files and modules being tested. + +package eth + +import ( + "crypto/rand" + "math/big" + "sync" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" +) + +// 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 { + var ( + emux = 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) + ) + chainman.SetProcessor(blockproc) + chainman.InsertChain(core.GenerateChain(genesis, db, blocks, nil)) + + pm := NewProtocolManager(NetworkId, emux, &testTxPool{added: newtx}, pow, chainman) + pm.Start() + return pm +} + +// testTxPool is a fake, helper transaction pool for testing purposes +type testTxPool struct { + pool []*types.Transaction // Collection of all transactions + added chan<- []*types.Transaction // Notification channel for new transactions + + lock sync.RWMutex // Protects the transaction pool +} + +// AddTransactions appends a batch of transactions to the pool, and notifies any +// listeners if the addition channel is non nil +func (p *testTxPool) AddTransactions(txs []*types.Transaction) { + p.lock.Lock() + defer p.lock.Unlock() + + p.pool = append(p.pool, txs...) + if p.added != nil { + p.added <- txs + } +} + +// GetTransactions returns all the transactions known to the pool +func (p *testTxPool) GetTransactions() types.Transactions { + p.lock.RLock() + defer p.lock.RUnlock() + + txs := make([]*types.Transaction, len(p.pool)) + copy(txs, p.pool) + + return txs +} + +// newTestTransaction create a new dummy transaction. +func newTestTransaction(from *crypto.Key, nonce uint64, datasize int) *types.Transaction { + tx := types.NewTransaction(nonce, common.Address{}, big.NewInt(0), big.NewInt(100000), big.NewInt(0), make([]byte, datasize)) + tx, _ = tx.SignECDSA(from.PrivateKey) + + return tx +} + +// testPeer is a simulated peer to allow testing direct network calls. +type testPeer struct { + net p2p.MsgReadWriter // Network layer reader/writer to simulate remote messaging + app *p2p.MsgPipeRW // Application layer reader/writer to simulate the local side + *peer +} + +// newTestPeer creates a new peer registered at the given protocol manager. +func newTestPeer(name string, pm *ProtocolManager, shake bool) (*testPeer, <-chan error) { + // Create a message pipe to communicate through + app, net := p2p.MsgPipe() + + // Generate a random id and create the peer + var id discover.NodeID + rand.Read(id[:]) + + peer := pm.newPeer(int(ProtocolVersions[0]), NetworkId, p2p.NewPeer(id, name, nil), net) + + // Start the peer on a new thread + errc := make(chan error, 1) + go func() { + pm.newPeerCh <- peer + errc <- pm.handle(peer) + }() + tp := &testPeer{ + app: app, + net: net, + peer: peer, + } + // Execute any implicitly requested handshakes and return + if shake { + td, head, genesis := pm.chainman.Status() + tp.handshake(nil, td, head, genesis) + } + return tp, errc +} + +// handshake simulates a trivial handshake that expects the same state from the +// remote side as we are simulating locally. +func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, genesis common.Hash) { + msg := &statusData{ + ProtocolVersion: uint32(ProtocolVersions[0]), + NetworkId: uint32(NetworkId), + TD: td, + CurrentBlock: head, + GenesisBlock: genesis, + } + if err := p2p.ExpectMsg(p.app, StatusMsg, msg); err != nil { + t.Fatalf("status recv: %v", err) + } + if err := p2p.Send(p.app, StatusMsg, msg); err != nil { + t.Fatalf("status send: %v", err) + } +} + +// close terminates the local side of the peer, notifying the remote protocol +// manager of termination. +func (p *testPeer) close() { + p.app.Close() +} diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 08c9b6a063..9a6964be2d 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -18,19 +18,15 @@ package eth import ( "crypto/rand" - "math/big" + "fmt" "sync" "testing" "time" "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/ethdb" - "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/p2p" - "github.com/ethereum/go-ethereum/p2p/discover" ) func init() { @@ -41,7 +37,7 @@ func init() { var testAccount = crypto.NewKey(rand.Reader) func TestStatusMsgErrors(t *testing.T) { - pm := newProtocolManagerForTesting(nil) + pm := newTestProtocolManager(0, nil) td, currentBlock, genesis := pm.chainman.Status() defer pm.Stop() @@ -56,7 +52,7 @@ func TestStatusMsgErrors(t *testing.T) { }, { code: StatusMsg, data: statusData{10, NetworkId, td, currentBlock, genesis}, - wantError: errResp(ErrProtocolVersionMismatch, "10 (!= 0)"), + wantError: errResp(ErrProtocolVersionMismatch, "10 (!= %d)", ProtocolVersions[0]), }, { code: StatusMsg, data: statusData{uint32(ProtocolVersions[0]), 999, td, currentBlock, genesis}, @@ -69,10 +65,10 @@ func TestStatusMsgErrors(t *testing.T) { } for i, test := range tests { - p, errc := newTestPeer(pm) + p, errc := newTestPeer("peer", pm, false) // The send call might hang until reset because // the protocol might not read the payload. - go p2p.Send(p, test.code, test.data) + go p2p.Send(p.app, test.code, test.data) select { case err := <-errc: @@ -91,14 +87,13 @@ 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 := newProtocolManagerForTesting(txAdded) - p, _ := newTestPeer(pm) + pm := newTestProtocolManager(0, txAdded) + p, _ := newTestPeer("peer", pm, true) defer pm.Stop() defer p.close() - p.handshake(t) - tx := newtx(testAccount, 0, 0) - if err := p2p.Send(p, TxMsg, []interface{}{tx}); err != nil { + tx := newTestTransaction(testAccount, 0, 0) + if err := p2p.Send(p.app, TxMsg, []interface{}{tx}); err != nil { t.Fatalf("send error: %v", err) } select { @@ -115,14 +110,14 @@ func TestRecvTransactions(t *testing.T) { // This test checks that pending transactions are sent. func TestSendTransactions(t *testing.T) { - pm := newProtocolManagerForTesting(nil) + pm := newTestProtocolManager(0, nil) defer pm.Stop() // Fill the pool with big transactions. const txsize = txsyncPackSize / 10 alltxs := make([]*types.Transaction, 100) for nonce := range alltxs { - alltxs[nonce] = newtx(testAccount, uint64(nonce), txsize) + alltxs[nonce] = newTestTransaction(testAccount, uint64(nonce), txsize) } pm.txpool.AddTransactions(alltxs) @@ -137,7 +132,7 @@ func TestSendTransactions(t *testing.T) { } for n := 0; n < len(alltxs) && !t.Failed(); { var txs []*types.Transaction - msg, err := p.ReadMsg() + msg, err := p.app.ReadMsg() if err != nil { t.Errorf("%v: read error: %v", p.Peer, err) } else if msg.Code != TxMsg { @@ -161,97 +156,9 @@ func TestSendTransactions(t *testing.T) { } } for i := 0; i < 3; i++ { - p, _ := newTestPeer(pm) - p.handshake(t) + p, _ := newTestPeer(fmt.Sprintf("peer #%d", i), pm, true) wg.Add(1) go checktxs(p) } wg.Wait() } - -// testPeer wraps all peer-related data for tests. -type testPeer struct { - p2p.MsgReadWriter // writing to the test peer feeds the protocol - pipe *p2p.MsgPipeRW // the protocol read/writes on this end - pm *ProtocolManager - *peer -} - -func newProtocolManagerForTesting(txAdded chan<- []*types.Transaction) *ProtocolManager { - db, _ := ethdb.NewMemDatabase() - core.WriteTestNetGenesisBlock(db, 0) - var ( - em = new(event.TypeMux) - chain, _ = core.NewChainManager(db, core.FakePow{}, em) - txpool = &fakeTxPool{added: txAdded} - pm = NewProtocolManager(NetworkId, em, txpool, core.FakePow{}, chain) - ) - pm.Start() - return pm -} - -func newTestPeer(pm *ProtocolManager) (*testPeer, <-chan error) { - var id discover.NodeID - rand.Read(id[:]) - rw1, rw2 := p2p.MsgPipe() - peer := pm.newPeer(pm.protVer, pm.netId, p2p.NewPeer(id, "test peer", nil), rw2) - errc := make(chan error, 1) - go func() { - pm.newPeerCh <- peer - errc <- pm.handle(peer) - }() - return &testPeer{rw1, rw2, pm, peer}, errc -} - -func (p *testPeer) handshake(t *testing.T) { - td, currentBlock, genesis := p.pm.chainman.Status() - msg := &statusData{ - ProtocolVersion: uint32(p.pm.protVer), - NetworkId: uint32(p.pm.netId), - TD: td, - CurrentBlock: currentBlock, - GenesisBlock: genesis, - } - if err := p2p.ExpectMsg(p, StatusMsg, msg); err != nil { - t.Fatalf("status recv: %v", err) - } - if err := p2p.Send(p, StatusMsg, msg); err != nil { - t.Fatalf("status send: %v", err) - } -} - -func (p *testPeer) close() { - p.pipe.Close() -} - -type fakeTxPool struct { - // all transactions are collected. - mu sync.Mutex - all []*types.Transaction - // if added is non-nil, it receives added transactions. - added chan<- []*types.Transaction -} - -func (pool *fakeTxPool) AddTransactions(txs []*types.Transaction) { - pool.mu.Lock() - defer pool.mu.Unlock() - pool.all = append(pool.all, txs...) - if pool.added != nil { - pool.added <- txs - } -} - -func (pool *fakeTxPool) GetTransactions() types.Transactions { - pool.mu.Lock() - defer pool.mu.Unlock() - txs := make([]*types.Transaction, len(pool.all)) - copy(txs, pool.all) - return types.Transactions(txs) -} - -func newtx(from *crypto.Key, nonce uint64, datasize int) *types.Transaction { - data := make([]byte, datasize) - tx := types.NewTransaction(nonce, common.Address{}, big.NewInt(0), big.NewInt(100000), big.NewInt(0), data) - tx, _ = tx.SignECDSA(from.PrivateKey) - return tx -}