From c51e153b5c5327f971e7b410e49e7babfc3f0b8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 2 Jul 2015 14:13:46 +0300 Subject: [PATCH 1/8] eth, metrics, p2p: prepare metrics and net packets to eth/62 --- eth/downloader/downloader.go | 3 +- eth/handler.go | 15 +------ eth/metrics.go | 84 ++++++++++++++++++++++++++++++++++++ eth/peer.go | 50 ++++++++++++--------- eth/protocol.go | 25 ++++++++--- metrics/metrics.go | 12 +++--- p2p/metrics.go | 8 +++- 7 files changed, 149 insertions(+), 48 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index e3e22a7848..23d2e045e9 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -39,6 +39,7 @@ import ( const ( eth60 = 60 // Constant to check for old protocol support eth61 = 61 // Constant to check for new protocol support + eth62 = 62 // Constant to check for experimental protocol support ) var ( @@ -329,7 +330,7 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e if err = d.fetchBlocks60(); err != nil { return err } - case eth61: + case eth61, eth62: // New eth/61, use forward, concurrent hash and block retrieval algorithm number, err := d.findAncestor(p) if err != nil { diff --git a/eth/handler.go b/eth/handler.go index 5d233dd968..6c1895bbd0 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -176,7 +176,7 @@ func (pm *ProtocolManager) Stop() { } func (pm *ProtocolManager) newPeer(pv, nv int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { - return newPeer(pv, nv, p, rw) + return newPeer(pv, nv, p, newMeteredMsgWriter(rw)) } // handle is the callback invoked to manage the life cycle of an eth peer. When @@ -281,14 +281,11 @@ 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)) - reqHashInPacketsMeter.Mark(1) var hashes []common.Hash if err := msgStream.Decode(&hashes); err != nil { break } - reqHashInTrafficMeter.Mark(int64(32 * len(hashes))) - // Deliver them all to the downloader for queuing err := pm.downloader.DeliverHashes(p.id, hashes) if err != nil { @@ -340,7 +337,6 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { case BlocksMsg: // Decode the arrived block message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) - reqBlockInPacketsMeter.Mark(1) var blocks []*types.Block if err := msgStream.Decode(&blocks); err != nil { @@ -349,7 +345,6 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } // Update the receive timestamp of each block for _, block := range blocks { - reqBlockInTrafficMeter.Mark(block.Size().Int64()) block.ReceivedAt = msg.ReceivedAt } // Filter out any explicitly requested blocks, deliver the rest to the downloader @@ -365,9 +360,6 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msgStream.Decode(&hashes); err != nil { break } - propHashInPacketsMeter.Mark(1) - propHashInTrafficMeter.Mark(int64(32 * len(hashes))) - // Mark the hashes as present at the remote node for _, hash := range hashes { p.MarkBlock(hash) @@ -390,9 +382,6 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&request); err != nil { return errResp(ErrDecode, "%v: %v", msg, err) } - propBlockInPacketsMeter.Mark(1) - propBlockInTrafficMeter.Mark(request.Block.Size().Int64()) - if err := request.Block.ValidateFields(); err != nil { return errResp(ErrDecode, "block validation %v: %v", msg, err) } @@ -427,7 +416,6 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&txs); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - propTxnInPacketsMeter.Mark(1) for i, tx := range txs { // Validate and mark the remote transaction if tx == nil { @@ -436,7 +424,6 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { p.MarkTransaction(tx.Hash()) // Log it's arrival for later analysis - propTxnInTrafficMeter.Mark(tx.Size().Int64()) jsonlogger.LogJson(&logger.EthTxReceived{ TxHash: tx.Hash().Hex(), RemoteId: p.ID().String(), diff --git a/eth/metrics.go b/eth/metrics.go index 625b90b671..13745dc43d 100644 --- a/eth/metrics.go +++ b/eth/metrics.go @@ -18,6 +18,7 @@ package eth import ( "github.com/ethereum/go-ethereum/metrics" + "github.com/ethereum/go-ethereum/p2p" ) var ( @@ -41,4 +42,87 @@ var ( reqBlockInTrafficMeter = metrics.NewMeter("eth/req/blocks/in/traffic") reqBlockOutPacketsMeter = metrics.NewMeter("eth/req/blocks/out/packets") reqBlockOutTrafficMeter = metrics.NewMeter("eth/req/blocks/out/traffic") + reqHeaderInPacketsMeter = metrics.NewMeter("eth/req/header/in/packets") + reqHeaderInTrafficMeter = metrics.NewMeter("eth/req/header/in/traffic") + reqHeaderOutPacketsMeter = metrics.NewMeter("eth/req/header/out/packets") + reqHeaderOutTrafficMeter = metrics.NewMeter("eth/req/header/out/traffic") + reqStateInPacketsMeter = metrics.NewMeter("eth/req/state/in/packets") + reqStateInTrafficMeter = metrics.NewMeter("eth/req/state/in/traffic") + reqStateOutPacketsMeter = metrics.NewMeter("eth/req/state/out/packets") + reqStateOutTrafficMeter = metrics.NewMeter("eth/req/state/out/traffic") + miscInPacketsMeter = metrics.NewMeter("eth/misc/in/packets") + miscInTrafficMeter = metrics.NewMeter("eth/misc/in/traffic") + miscOutPacketsMeter = metrics.NewMeter("eth/misc/out/packets") + miscOutTrafficMeter = metrics.NewMeter("eth/misc/out/traffic") ) + +// meteredMsgReadWriter is a wrapper around a p2p.MsgReadWriter, capable of +// accumulating the above defined metrics based on the data stream contents. +type meteredMsgReadWriter struct { + p2p.MsgReadWriter +} + +// newMeteredMsgWriter wraps a p2p MsgReadWriter with metering support. If the +// metrics system is disabled, this fucntion returns the original object. +func newMeteredMsgWriter(rw p2p.MsgReadWriter) p2p.MsgReadWriter { + if !metrics.Enabled { + return rw + } + return &meteredMsgReadWriter{rw} +} + +func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) { + // Read the message and short circuit in case of an error + msg, err := rw.MsgReadWriter.ReadMsg() + if err != nil { + return msg, err + } + // Account for the data traffic + packets, traffic := miscInPacketsMeter, miscInTrafficMeter + switch msg.Code { + case BlockHashesMsg: + packets, traffic = reqHashInPacketsMeter, reqHashInTrafficMeter + case BlocksMsg: + packets, traffic = reqBlockInPacketsMeter, reqBlockInTrafficMeter + case BlockHeadersMsg: + packets, traffic = reqHeaderInPacketsMeter, reqHeaderInTrafficMeter + case NodeDataMsg: + packets, traffic = reqStateInPacketsMeter, reqStateInTrafficMeter + case NewBlockHashesMsg: + packets, traffic = propHashInPacketsMeter, propHashInTrafficMeter + case NewBlockMsg: + packets, traffic = propBlockInPacketsMeter, propBlockInTrafficMeter + case TxMsg: + packets, traffic = propTxnInPacketsMeter, propTxnInTrafficMeter + } + packets.Mark(1) + traffic.Mark(int64(msg.Size)) + + return msg, err +} + +func (rw *meteredMsgReadWriter) WriteMsg(msg p2p.Msg) error { + // Account for the data traffic + packets, traffic := miscOutPacketsMeter, miscOutTrafficMeter + switch msg.Code { + case BlockHashesMsg: + packets, traffic = reqHashOutPacketsMeter, reqHashOutTrafficMeter + case BlocksMsg: + packets, traffic = reqBlockOutPacketsMeter, reqBlockOutTrafficMeter + case BlockHeadersMsg: + packets, traffic = reqHeaderOutPacketsMeter, reqHeaderOutTrafficMeter + case NodeDataMsg: + packets, traffic = reqStateOutPacketsMeter, reqStateOutTrafficMeter + case NewBlockHashesMsg: + packets, traffic = propHashOutPacketsMeter, propHashOutTrafficMeter + case NewBlockMsg: + packets, traffic = propBlockOutPacketsMeter, propBlockOutTrafficMeter + case TxMsg: + packets, traffic = propTxnOutPacketsMeter, propTxnOutTrafficMeter + } + packets.Mark(1) + traffic.Mark(int64(msg.Size)) + + // Send the packet to the p2p layer + return rw.MsgReadWriter.WriteMsg(msg) +} diff --git a/eth/peer.go b/eth/peer.go index ade1f37eaa..c17cdfca78 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -129,9 +129,7 @@ func (p *peer) MarkTransaction(hash common.Hash) { // SendTransactions sends transactions to the peer and includes the hashes // in its transaction hash set for future reference. func (p *peer) SendTransactions(txs types.Transactions) error { - propTxnOutPacketsMeter.Mark(1) for _, tx := range txs { - propTxnOutTrafficMeter.Mark(tx.Size().Int64()) p.knownTxs.Add(tx.Hash()) } return p2p.Send(p.rw, TxMsg, txs) @@ -139,27 +137,17 @@ func (p *peer) SendTransactions(txs types.Transactions) error { // SendBlockHashes sends a batch of known hashes to the remote peer. func (p *peer) SendBlockHashes(hashes []common.Hash) error { - reqHashOutPacketsMeter.Mark(1) - reqHashOutTrafficMeter.Mark(int64(32 * len(hashes))) - return p2p.Send(p.rw, BlockHashesMsg, hashes) } // SendBlocks sends a batch of blocks to the remote peer. func (p *peer) SendBlocks(blocks []*types.Block) error { - reqBlockOutPacketsMeter.Mark(1) - for _, block := range blocks { - reqBlockOutTrafficMeter.Mark(block.Size().Int64()) - } return p2p.Send(p.rw, BlocksMsg, blocks) } // SendNewBlockHashes announces the availability of a number of blocks through // a hash notification. func (p *peer) SendNewBlockHashes(hashes []common.Hash) error { - propHashOutPacketsMeter.Mark(1) - propHashOutTrafficMeter.Mark(int64(32 * len(hashes))) - for _, hash := range hashes { p.knownBlocks.Add(hash) } @@ -168,33 +156,55 @@ func (p *peer) SendNewBlockHashes(hashes []common.Hash) error { // SendNewBlock propagates an entire block to a remote peer. func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error { - propBlockOutPacketsMeter.Mark(1) - propBlockOutTrafficMeter.Mark(block.Size().Int64()) - p.knownBlocks.Add(block.Hash()) return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td}) } +// SendBlockHeaders sends a batch of block headers to the remote peer. +func (p *peer) SendBlockHeaders(headers []*types.Header) error { + return p2p.Send(p.rw, BlockHeadersMsg, headers) +} + +// SendNodeData sends a batch of arbitrary internal data, corresponding to the +// hashes requested. +func (p *peer) SendNodeData(data [][]byte) error { + return p2p.Send(p.rw, NodeDataMsg, data) +} + // RequestHashes fetches a batch of hashes from a peer, starting at from, going // towards the genesis block. func (p *peer) RequestHashes(from common.Hash) error { - glog.V(logger.Debug).Infof("Peer [%s] fetching hashes (%d) from %x...\n", p.id, downloader.MaxHashFetch, from[:4]) + glog.V(logger.Debug).Infof("%v fetching hashes (%d) from %x...\n", p, downloader.MaxHashFetch, from[:4]) return p2p.Send(p.rw, GetBlockHashesMsg, getBlockHashesData{from, uint64(downloader.MaxHashFetch)}) } -// RequestHashesFromNumber fetches a batch of hashes from a peer, starting at the -// requested block number, going upwards towards the genesis block. +// RequestHashesFromNumber fetches a batch of hashes from a peer, starting at +// the requested block number, going upwards towards the genesis block. func (p *peer) RequestHashesFromNumber(from uint64, count int) error { - glog.V(logger.Debug).Infof("Peer [%s] fetching hashes (%d) from #%d...\n", p.id, count, from) + glog.V(logger.Debug).Infof("%v fetching hashes (%d) from #%d...\n", p, count, from) return p2p.Send(p.rw, GetBlockHashesFromNumberMsg, getBlockHashesFromNumberData{from, uint64(count)}) } // RequestBlocks fetches a batch of blocks corresponding to the specified hashes. func (p *peer) RequestBlocks(hashes []common.Hash) error { - glog.V(logger.Debug).Infof("[%s] fetching %v blocks\n", p.id, len(hashes)) + glog.V(logger.Debug).Infof("%v fetching %v blocks\n", p, len(hashes)) return p2p.Send(p.rw, GetBlocksMsg, hashes) } +// RequestHeaders fetches a batch of blocks' headers corresponding to the +// specified hashes. +func (p *peer) RequestHeaders(hashes []common.Hash) error { + glog.V(logger.Debug).Infof("%v fetching %v headers\n", p, len(hashes)) + return p2p.Send(p.rw, GetBlockHeadersMsg, hashes) +} + +// RequestNodeData fetches a batch of arbitrary data from a node's known state +// data, corresponding to the specified hashes. +func (p *peer) RequestNodeData(hashes []common.Hash) error { + glog.V(logger.Debug).Infof("%v fetching %v state data\n", p, len(hashes)) + return p2p.Send(p.rw, GetNodeDataMsg, hashes) +} + // Handshake executes the eth protocol handshake, negotiating version number, // network IDs, difficulties, head and genesis blocks. func (p *peer) Handshake(td *big.Int, head common.Hash, genesis common.Hash) error { diff --git a/eth/protocol.go b/eth/protocol.go index b8c9b50d0e..fcc5f21e2e 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -24,10 +24,10 @@ import ( ) // Supported versions of the eth protocol (first is primary). -var ProtocolVersions = []uint{61, 60} +var ProtocolVersions = []uint{62, 61, 60} // Number of implemented message corresponding to different protocol versions. -var ProtocolLengths = []uint64{9, 8} +var ProtocolLengths = []uint64{13, 9, 8} const ( NetworkId = 1 @@ -36,6 +36,7 @@ const ( // eth protocol message codes const ( + // Protocol messages belonging to eth/60 StatusMsg = iota NewBlockHashesMsg TxMsg @@ -44,7 +45,15 @@ const ( GetBlocksMsg BlocksMsg NewBlockMsg + + // Protocol messages belonging to eth/61 GetBlockHashesFromNumberMsg + + // Protocol messages belonging to eth/62 + GetBlockHeadersMsg + BlockHeadersMsg + GetNodeDataMsg + NodeDataMsg ) type errCode int @@ -102,15 +111,14 @@ type statusData struct { GenesisBlock common.Hash } -// getBlockHashesData is the network packet for the hash based block retrieval -// message. +// getBlockHashesData is the network packet for the hash based hash retrieval. type getBlockHashesData struct { Hash common.Hash Amount uint64 } -// getBlockHashesFromNumberData is the network packet for the number based block -// retrieval message. +// getBlockHashesFromNumberData is the network packet for the number based hash +// retrieval. type getBlockHashesFromNumberData struct { Number uint64 Amount uint64 @@ -121,3 +129,8 @@ type newBlockData struct { Block *types.Block TD *big.Int } + +// nodeDataData is the network response packet for a node data retrieval. +type nodeDataData []struct { + Value []byte +} diff --git a/metrics/metrics.go b/metrics/metrics.go index 6d1a065edb..fcf8b5c32b 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -31,8 +31,8 @@ import ( // MetricsEnabledFlag is the CLI flag name to use to enable metrics collections. var MetricsEnabledFlag = "metrics" -// enabled is the flag specifying if metrics are enable or not. -var enabled = false +// Enabled is the flag specifying if metrics are enable or not. +var Enabled = false // Init enables or disables the metrics system. Since we need this to run before // any other code gets to create meters and timers, we'll actually do an ugly hack @@ -41,7 +41,7 @@ func init() { for _, arg := range os.Args { if strings.TrimLeft(arg, "-") == MetricsEnabledFlag { glog.V(logger.Info).Infof("Enabling metrics collection") - enabled = true + Enabled = true } } } @@ -49,7 +49,7 @@ func init() { // NewMeter create a new metrics Meter, either a real one of a NOP stub depending // on the metrics flag. func NewMeter(name string) metrics.Meter { - if !enabled { + if !Enabled { return new(metrics.NilMeter) } return metrics.GetOrRegisterMeter(name, metrics.DefaultRegistry) @@ -58,7 +58,7 @@ func NewMeter(name string) metrics.Meter { // NewTimer create a new metrics Timer, either a real one of a NOP stub depending // on the metrics flag. func NewTimer(name string) metrics.Timer { - if !enabled { + if !Enabled { return new(metrics.NilTimer) } return metrics.GetOrRegisterTimer(name, metrics.DefaultRegistry) @@ -68,7 +68,7 @@ func NewTimer(name string) metrics.Timer { // process. func CollectProcessMetrics(refresh time.Duration) { // Short circuit if the metrics system is disabled - if !enabled { + if !Enabled { return } // Create the various data collectors diff --git a/p2p/metrics.go b/p2p/metrics.go index f98cac2742..98b61901d5 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -38,8 +38,14 @@ type meteredConn struct { } // newMeteredConn creates a new metered connection, also bumping the ingress or -// egress connection meter. +// egress connection meter. If the metrics system is disabled, this function +// returns the original object. func newMeteredConn(conn net.Conn, ingress bool) net.Conn { + // Short circuit if metrics are disabled + if !metrics.Enabled { + return conn + } + // Otherwise bump the connection counters and wrap the connection if ingress { ingressConnectMeter.Mark(1) } else { From b31fb6d91e53a2af7523e5aef25ea31dac4e8250 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 2/8] 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 6c1895bbd0..b90981e30a 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 -} From e04fdd9d22d4722b246b079276f8f98050029735 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 3/8] 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 2b21a7c960..ad2a2c1f94 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -373,7 +373,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 b90981e30a..53df9ff427 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")) From b21cd21a7e4e37ac208d2daa7aed5070197db4b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 6 Jul 2015 13:21:44 +0300 Subject: [PATCH 4/8] eth, eth/downloader: support overlapping messages between proto versions --- eth/downloader/downloader.go | 3 +- eth/handler.go | 27 ++++---- eth/handler_test.go | 37 ++++++---- eth/helper_test.go | 6 +- eth/metrics.go | 131 +++++++++++++++++++++-------------- eth/protocol.go | 44 ++++++++---- eth/protocol_test.go | 34 ++++++--- 7 files changed, 176 insertions(+), 106 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 088e441e08..bd45af655c 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -39,7 +39,6 @@ import ( const ( eth60 = 60 // Constant to check for old protocol support eth61 = 61 // Constant to check for new protocol support - eth62 = 62 // Constant to check for experimental protocol support ) var ( @@ -332,7 +331,7 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e if err = d.fetchBlocks60(); err != nil { return err } - case eth61, eth62: + case eth61: // New eth/61, use forward, concurrent hash and block retrieval algorithm number, err := d.findAncestor(p) if err != nil { diff --git a/eth/handler.go b/eth/handler.go index 53df9ff427..b64ee869aa 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -188,6 +188,9 @@ func (pm *ProtocolManager) handle(p *peer) error { glog.V(logger.Debug).Infof("%v: handshake failed: %v", p, err) return err } + if rw, ok := p.rw.(*meteredMsgReadWriter); ok { + rw.Init(p.version) + } // Register the peer locally glog.V(logger.Detail).Infof("%v: adding peer", p) if err := pm.peers.Register(p); err != nil { @@ -228,12 +231,12 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { defer msg.Discard() // Handle the message depending on its contents - switch msg.Code { - case StatusMsg: + switch { + case msg.Code == StatusMsg: // Status messages should never arrive after the handshake return errResp(ErrExtraStatusMsg, "uncontrolled status message") - case GetBlockHashesMsg: + case (p.version == eth60 || p.version == eth61) && msg.Code == GetBlockHashesMsg: // Retrieve the number of hashes to return and from which origin hash var request getBlockHashesData if err := msg.Decode(&request); err != nil { @@ -249,7 +252,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendBlockHashes(hashes) - case GetBlockHashesFromNumberMsg: + case (p.version == eth60 || p.version == eth61) && msg.Code == GetBlockHashesFromNumberMsg: // Retrieve and decode the number of hashes to return and from which origin number var request getBlockHashesFromNumberData if err := msg.Decode(&request); err != nil { @@ -276,7 +279,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendBlockHashes(hashes) - case BlockHashesMsg: + case (p.version == eth60 || p.version == eth61) && msg.Code == BlockHashesMsg: // A batch of hashes arrived to one of our previous requests var hashes []common.Hash if err := msg.Decode(&hashes); err != nil { @@ -288,7 +291,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { glog.V(logger.Debug).Infoln(err) } - case GetBlocksMsg: + case (p.version == eth60 || p.version == eth61) && msg.Code == GetBlocksMsg: // Decode the retrieval message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) if _, err := msgStream.List(); err != nil { @@ -316,7 +319,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendBlocks(blocks) - case BlocksMsg: + case (p.version == eth60 || p.version == eth61) && msg.Code == BlocksMsg: // Decode the arrived block message var blocks []*types.Block if err := msg.Decode(&blocks); err != nil { @@ -332,7 +335,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { pm.downloader.DeliverBlocks(p.id, blocks) } - case GetBlockHeadersMsg: + case p.version == eth62 && msg.Code == GetBlockHeadersMsg: // Decode the retrieval message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) if _, err := msgStream.List(); err != nil { @@ -359,7 +362,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendBlockHeaders(headers) - case GetNodeDataMsg: + case p.version == eth63 && msg.Code == GetNodeDataMsg: // Decode the retrieval message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) if _, err := msgStream.List(); err != nil { @@ -386,7 +389,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendNodeData(data) - case NewBlockHashesMsg: + case msg.Code == NewBlockHashesMsg: // Retrieve and deseralize the remote new block hashes notification var hashes []common.Hash if err := msg.Decode(&hashes); err != nil { @@ -408,7 +411,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { pm.fetcher.Notify(p.id, hash, time.Now(), p.RequestBlocks) } - case NewBlockMsg: + case msg.Code == NewBlockMsg: // Retrieve and decode the propagated block var request newBlockData if err := msg.Decode(&request); err != nil { @@ -442,7 +445,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } - case TxMsg: + case msg.Code == TxMsg: // Transactions arrived, parse all of them and deliver to the pool var txs []*types.Transaction if err := msg.Decode(&txs); err != nil { diff --git a/eth/handler_test.go b/eth/handler_test.go index 7cc7db16f4..84c977b0e3 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -19,9 +19,12 @@ import ( // Tests that hashes can be retrieved from a remote chain by hashes in reverse // order. -func TestGetBlockHashes(t *testing.T) { +func TestGetBlockHashes60(t *testing.T) { testGetBlockHashes(t, 60) } +func TestGetBlockHashes61(t *testing.T) { testGetBlockHashes(t, 61) } + +func testGetBlockHashes(t *testing.T, protocol int) { pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil) - peer, _ := newTestPeer("peer", pm, true) + peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() // Create a batch of tests for various scenarios @@ -60,9 +63,12 @@ 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) { +func TestGetBlockHashesFromNumber60(t *testing.T) { testGetBlockHashesFromNumber(t, 60) } +func TestGetBlockHashesFromNumber61(t *testing.T) { testGetBlockHashesFromNumber(t, 61) } + +func testGetBlockHashesFromNumber(t *testing.T, protocol int) { pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil) - peer, _ := newTestPeer("peer", pm, true) + peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() // Create a batch of tests for various scenarios @@ -98,9 +104,12 @@ func TestGetBlockHashesFromNumber(t *testing.T) { } // Tests that blocks can be retrieved from a remote chain based on their hashes. -func TestGetBlocks(t *testing.T) { +func TestGetBlocks60(t *testing.T) { testGetBlocks(t, 60) } +func TestGetBlocks61(t *testing.T) { testGetBlocks(t, 61) } + +func testGetBlocks(t *testing.T, protocol int) { pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil) - peer, _ := newTestPeer("peer", pm, true) + peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() // Create a batch of tests for various scenarios @@ -166,9 +175,9 @@ func TestGetBlocks(t *testing.T) { } // Tests that block headers can be retrieved from a remote chain based on their hashes. -func TestGetBlockHeaders(t *testing.T) { +func TestGetBlockHeaders62(t *testing.T) { pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil) - peer, _ := newTestPeer("peer", pm, true) + peer, _ := newTestPeer("peer", 62, pm, true) defer peer.close() // Create a batch of tests for various scenarios @@ -226,15 +235,15 @@ func TestGetBlockHeaders(t *testing.T) { } } // Send the hash request and verify the response - p2p.Send(peer.app, 0x09, hashes) - if err := p2p.ExpectMsg(peer.app, 0x0a, headers); err != nil { + p2p.Send(peer.app, 0x03, hashes) + if err := p2p.ExpectMsg(peer.app, 0x04, 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) { +func TestGetNodeData63(t *testing.T) { // Define three accounts to simulate transactions with acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a") acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee") @@ -271,7 +280,7 @@ func TestGetNodeData(t *testing.T) { } // Assemble the test environment pm := newTestProtocolManager(4, generator, nil) - peer, _ := newTestPeer("peer", pm, true) + peer, _ := newTestPeer("peer", 63, pm, true) defer peer.close() // Fetch for now the entire state db @@ -279,12 +288,12 @@ func TestGetNodeData(t *testing.T) { for _, key := range pm.statedb.(*ethdb.MemDatabase).Keys() { hashes = append(hashes, common.BytesToHash(key)) } - p2p.Send(peer.app, 0x0b, hashes) + p2p.Send(peer.app, 0x0d, hashes) msg, err := peer.app.ReadMsg() if err != nil { t.Fatalf("failed to read node data response: %v", err) } - if msg.Code != 0x0c { + if msg.Code != 0x0e { t.Fatalf("response packet code mismatch: have %x, want %x", msg.Code, 0x0c) } var data [][]byte diff --git a/eth/helper_test.go b/eth/helper_test.go index 533eaff6e4..3a799e6f69 100644 --- a/eth/helper_test.go +++ b/eth/helper_test.go @@ -93,7 +93,7 @@ type testPeer struct { } // newTestPeer creates a new peer registered at the given protocol manager. -func newTestPeer(name string, pm *ProtocolManager, shake bool) (*testPeer, <-chan error) { +func newTestPeer(name string, version int, pm *ProtocolManager, shake bool) (*testPeer, <-chan error) { // Create a message pipe to communicate through app, net := p2p.MsgPipe() @@ -101,7 +101,7 @@ func newTestPeer(name string, pm *ProtocolManager, shake bool) (*testPeer, <-cha var id discover.NodeID rand.Read(id[:]) - peer := pm.newPeer(int(ProtocolVersions[0]), NetworkId, p2p.NewPeer(id, name, nil), net) + peer := pm.newPeer(version, NetworkId, p2p.NewPeer(id, name, nil), net) // Start the peer on a new thread errc := make(chan error, 1) @@ -126,7 +126,7 @@ func newTestPeer(name string, pm *ProtocolManager, shake bool) (*testPeer, <-cha // 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]), + ProtocolVersion: uint32(p.version), NetworkId: uint32(NetworkId), TD: td, CurrentBlock: head, diff --git a/eth/metrics.go b/eth/metrics.go index 13745dc43d..778747210d 100644 --- a/eth/metrics.go +++ b/eth/metrics.go @@ -22,44 +22,53 @@ import ( ) var ( - propTxnInPacketsMeter = metrics.NewMeter("eth/prop/txns/in/packets") - propTxnInTrafficMeter = metrics.NewMeter("eth/prop/txns/in/traffic") - propTxnOutPacketsMeter = metrics.NewMeter("eth/prop/txns/out/packets") - propTxnOutTrafficMeter = metrics.NewMeter("eth/prop/txns/out/traffic") - propHashInPacketsMeter = metrics.NewMeter("eth/prop/hashes/in/packets") - propHashInTrafficMeter = metrics.NewMeter("eth/prop/hashes/in/traffic") - propHashOutPacketsMeter = metrics.NewMeter("eth/prop/hashes/out/packets") - propHashOutTrafficMeter = metrics.NewMeter("eth/prop/hashes/out/traffic") - propBlockInPacketsMeter = metrics.NewMeter("eth/prop/blocks/in/packets") - propBlockInTrafficMeter = metrics.NewMeter("eth/prop/blocks/in/traffic") - propBlockOutPacketsMeter = metrics.NewMeter("eth/prop/blocks/out/packets") - propBlockOutTrafficMeter = metrics.NewMeter("eth/prop/blocks/out/traffic") - reqHashInPacketsMeter = metrics.NewMeter("eth/req/hashes/in/packets") - reqHashInTrafficMeter = metrics.NewMeter("eth/req/hashes/in/traffic") - reqHashOutPacketsMeter = metrics.NewMeter("eth/req/hashes/out/packets") - reqHashOutTrafficMeter = metrics.NewMeter("eth/req/hashes/out/traffic") - reqBlockInPacketsMeter = metrics.NewMeter("eth/req/blocks/in/packets") - reqBlockInTrafficMeter = metrics.NewMeter("eth/req/blocks/in/traffic") - reqBlockOutPacketsMeter = metrics.NewMeter("eth/req/blocks/out/packets") - reqBlockOutTrafficMeter = metrics.NewMeter("eth/req/blocks/out/traffic") - reqHeaderInPacketsMeter = metrics.NewMeter("eth/req/header/in/packets") - reqHeaderInTrafficMeter = metrics.NewMeter("eth/req/header/in/traffic") - reqHeaderOutPacketsMeter = metrics.NewMeter("eth/req/header/out/packets") - reqHeaderOutTrafficMeter = metrics.NewMeter("eth/req/header/out/traffic") - reqStateInPacketsMeter = metrics.NewMeter("eth/req/state/in/packets") - reqStateInTrafficMeter = metrics.NewMeter("eth/req/state/in/traffic") - reqStateOutPacketsMeter = metrics.NewMeter("eth/req/state/out/packets") - reqStateOutTrafficMeter = metrics.NewMeter("eth/req/state/out/traffic") - miscInPacketsMeter = metrics.NewMeter("eth/misc/in/packets") - miscInTrafficMeter = metrics.NewMeter("eth/misc/in/traffic") - miscOutPacketsMeter = metrics.NewMeter("eth/misc/out/packets") - miscOutTrafficMeter = metrics.NewMeter("eth/misc/out/traffic") + propTxnInPacketsMeter = metrics.NewMeter("eth/prop/txns/in/packets") + propTxnInTrafficMeter = metrics.NewMeter("eth/prop/txns/in/traffic") + propTxnOutPacketsMeter = metrics.NewMeter("eth/prop/txns/out/packets") + propTxnOutTrafficMeter = metrics.NewMeter("eth/prop/txns/out/traffic") + propHashInPacketsMeter = metrics.NewMeter("eth/prop/hashes/in/packets") + propHashInTrafficMeter = metrics.NewMeter("eth/prop/hashes/in/traffic") + propHashOutPacketsMeter = metrics.NewMeter("eth/prop/hashes/out/packets") + propHashOutTrafficMeter = metrics.NewMeter("eth/prop/hashes/out/traffic") + propBlockInPacketsMeter = metrics.NewMeter("eth/prop/blocks/in/packets") + propBlockInTrafficMeter = metrics.NewMeter("eth/prop/blocks/in/traffic") + propBlockOutPacketsMeter = metrics.NewMeter("eth/prop/blocks/out/packets") + propBlockOutTrafficMeter = metrics.NewMeter("eth/prop/blocks/out/traffic") + reqHashInPacketsMeter = metrics.NewMeter("eth/req/hashes/in/packets") + reqHashInTrafficMeter = metrics.NewMeter("eth/req/hashes/in/traffic") + reqHashOutPacketsMeter = metrics.NewMeter("eth/req/hashes/out/packets") + reqHashOutTrafficMeter = metrics.NewMeter("eth/req/hashes/out/traffic") + reqBlockInPacketsMeter = metrics.NewMeter("eth/req/blocks/in/packets") + reqBlockInTrafficMeter = metrics.NewMeter("eth/req/blocks/in/traffic") + reqBlockOutPacketsMeter = metrics.NewMeter("eth/req/blocks/out/packets") + reqBlockOutTrafficMeter = metrics.NewMeter("eth/req/blocks/out/traffic") + reqHeaderInPacketsMeter = metrics.NewMeter("eth/req/header/in/packets") + reqHeaderInTrafficMeter = metrics.NewMeter("eth/req/header/in/traffic") + reqHeaderOutPacketsMeter = metrics.NewMeter("eth/req/header/out/packets") + reqHeaderOutTrafficMeter = metrics.NewMeter("eth/req/header/out/traffic") + reqBodyInPacketsMeter = metrics.NewMeter("eth/req/body/in/packets") + reqBodyInTrafficMeter = metrics.NewMeter("eth/req/body/in/traffic") + reqBodyOutPacketsMeter = metrics.NewMeter("eth/req/body/out/packets") + reqBodyOutTrafficMeter = metrics.NewMeter("eth/req/body/out/traffic") + reqStateInPacketsMeter = metrics.NewMeter("eth/req/state/in/packets") + reqStateInTrafficMeter = metrics.NewMeter("eth/req/state/in/traffic") + reqStateOutPacketsMeter = metrics.NewMeter("eth/req/state/out/packets") + reqStateOutTrafficMeter = metrics.NewMeter("eth/req/state/out/traffic") + reqReceiptInPacketsMeter = metrics.NewMeter("eth/req/receipt/in/packets") + reqReceiptInTrafficMeter = metrics.NewMeter("eth/req/receipt/in/traffic") + reqReceiptOutPacketsMeter = metrics.NewMeter("eth/req/receipt/out/packets") + reqReceiptOutTrafficMeter = metrics.NewMeter("eth/req/receipt/out/traffic") + miscInPacketsMeter = metrics.NewMeter("eth/misc/in/packets") + miscInTrafficMeter = metrics.NewMeter("eth/misc/in/traffic") + miscOutPacketsMeter = metrics.NewMeter("eth/misc/out/packets") + miscOutTrafficMeter = metrics.NewMeter("eth/misc/out/traffic") ) // meteredMsgReadWriter is a wrapper around a p2p.MsgReadWriter, capable of // accumulating the above defined metrics based on the data stream contents. type meteredMsgReadWriter struct { - p2p.MsgReadWriter + p2p.MsgReadWriter // Wrapped message stream to meter + version int // Protocol version to select correct meters } // newMeteredMsgWriter wraps a p2p MsgReadWriter with metering support. If the @@ -68,7 +77,13 @@ func newMeteredMsgWriter(rw p2p.MsgReadWriter) p2p.MsgReadWriter { if !metrics.Enabled { return rw } - return &meteredMsgReadWriter{rw} + return &meteredMsgReadWriter{MsgReadWriter: rw} +} + +// Init sets the protocol version used by the stream to know which meters to +// increment in case of overlapping message ids between protocol versions. +func (rw *meteredMsgReadWriter) Init(version int) { + rw.version = version } func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) { @@ -79,20 +94,27 @@ func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) { } // Account for the data traffic packets, traffic := miscInPacketsMeter, miscInTrafficMeter - switch msg.Code { - case BlockHashesMsg: + switch { + case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlockHashesMsg: packets, traffic = reqHashInPacketsMeter, reqHashInTrafficMeter - case BlocksMsg: + case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlocksMsg: packets, traffic = reqBlockInPacketsMeter, reqBlockInTrafficMeter - case BlockHeadersMsg: - packets, traffic = reqHeaderInPacketsMeter, reqHeaderInTrafficMeter - case NodeDataMsg: + + case rw.version == eth62 && msg.Code == BlockHeadersMsg: + packets, traffic = reqBlockInPacketsMeter, reqBlockInTrafficMeter + case rw.version == eth62 && msg.Code == BlockBodiesMsg: + packets, traffic = reqBodyInPacketsMeter, reqBodyInTrafficMeter + + case rw.version == eth63 && msg.Code == NodeDataMsg: packets, traffic = reqStateInPacketsMeter, reqStateInTrafficMeter - case NewBlockHashesMsg: + case rw.version == eth63 && msg.Code == ReceiptsMsg: + packets, traffic = reqReceiptInPacketsMeter, reqReceiptInTrafficMeter + + case msg.Code == NewBlockHashesMsg: packets, traffic = propHashInPacketsMeter, propHashInTrafficMeter - case NewBlockMsg: + case msg.Code == NewBlockMsg: packets, traffic = propBlockInPacketsMeter, propBlockInTrafficMeter - case TxMsg: + case msg.Code == TxMsg: packets, traffic = propTxnInPacketsMeter, propTxnInTrafficMeter } packets.Mark(1) @@ -104,20 +126,27 @@ func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) { func (rw *meteredMsgReadWriter) WriteMsg(msg p2p.Msg) error { // Account for the data traffic packets, traffic := miscOutPacketsMeter, miscOutTrafficMeter - switch msg.Code { - case BlockHashesMsg: + switch { + case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlockHashesMsg: packets, traffic = reqHashOutPacketsMeter, reqHashOutTrafficMeter - case BlocksMsg: + case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlocksMsg: packets, traffic = reqBlockOutPacketsMeter, reqBlockOutTrafficMeter - case BlockHeadersMsg: + + case rw.version == eth62 && msg.Code == BlockHeadersMsg: packets, traffic = reqHeaderOutPacketsMeter, reqHeaderOutTrafficMeter - case NodeDataMsg: + case rw.version == eth62 && msg.Code == BlockBodiesMsg: + packets, traffic = reqBodyOutPacketsMeter, reqBodyOutTrafficMeter + + case rw.version == eth63 && msg.Code == NodeDataMsg: packets, traffic = reqStateOutPacketsMeter, reqStateOutTrafficMeter - case NewBlockHashesMsg: + case rw.version == eth63 && msg.Code == ReceiptsMsg: + packets, traffic = reqReceiptOutPacketsMeter, reqReceiptOutTrafficMeter + + case msg.Code == NewBlockHashesMsg: packets, traffic = propHashOutPacketsMeter, propHashOutTrafficMeter - case NewBlockMsg: + case msg.Code == NewBlockMsg: packets, traffic = propBlockOutPacketsMeter, propBlockOutTrafficMeter - case TxMsg: + case msg.Code == TxMsg: packets, traffic = propTxnOutPacketsMeter, propTxnOutTrafficMeter } packets.Mark(1) diff --git a/eth/protocol.go b/eth/protocol.go index fcc5f21e2e..fc8446589b 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -23,11 +23,19 @@ import ( "github.com/ethereum/go-ethereum/core/types" ) +// Constants to match up protocol versions and messages +const ( + eth60 = 60 + eth61 = 61 + eth62 = 62 + eth63 = 63 +) + // Supported versions of the eth protocol (first is primary). -var ProtocolVersions = []uint{62, 61, 60} +var ProtocolVersions = []uint{61, 60} // Number of implemented message corresponding to different protocol versions. -var ProtocolLengths = []uint64{13, 9, 8} +var ProtocolLengths = []uint64{9, 8} const ( NetworkId = 1 @@ -37,23 +45,29 @@ const ( // eth protocol message codes const ( // Protocol messages belonging to eth/60 - StatusMsg = iota - NewBlockHashesMsg - TxMsg - GetBlockHashesMsg - BlockHashesMsg - GetBlocksMsg - BlocksMsg - NewBlockMsg + StatusMsg = 0x00 + NewBlockHashesMsg = 0x01 + TxMsg = 0x02 + GetBlockHashesMsg = 0x03 + BlockHashesMsg = 0x04 + GetBlocksMsg = 0x05 + BlocksMsg = 0x06 + NewBlockMsg = 0x07 // Protocol messages belonging to eth/61 - GetBlockHashesFromNumberMsg + GetBlockHashesFromNumberMsg = 0x08 // Protocol messages belonging to eth/62 - GetBlockHeadersMsg - BlockHeadersMsg - GetNodeDataMsg - NodeDataMsg + GetBlockHeadersMsg = 0x03 + BlockHeadersMsg = 0x04 + GetBlockBodiesMsg = 0x05 + BlockBodiesMsg = 0x06 + + // Protocol messages belonging to eth/63 + GetNodeDataMsg = 0x0d + NodeDataMsg = 0x0e + GetReceiptsMsg = 0x0f + ReceiptsMsg = 0x10 ) type errCode int diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 166488b79a..4b89bc1f75 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -36,7 +36,13 @@ func init() { var testAccount = crypto.NewKey(rand.Reader) -func TestStatusMsgErrors(t *testing.T) { +// Tests that handshake failures are detected and reported correctly. +func TestStatusMsgErrors60(t *testing.T) { testStatusMsgErrors(t, 60) } +func TestStatusMsgErrors61(t *testing.T) { testStatusMsgErrors(t, 61) } +func TestStatusMsgErrors62(t *testing.T) { testStatusMsgErrors(t, 62) } +func TestStatusMsgErrors63(t *testing.T) { testStatusMsgErrors(t, 63) } + +func testStatusMsgErrors(t *testing.T, protocol int) { pm := newTestProtocolManager(0, nil, nil) td, currentBlock, genesis := pm.chainman.Status() defer pm.Stop() @@ -52,20 +58,20 @@ func TestStatusMsgErrors(t *testing.T) { }, { code: StatusMsg, data: statusData{10, NetworkId, td, currentBlock, genesis}, - wantError: errResp(ErrProtocolVersionMismatch, "10 (!= %d)", ProtocolVersions[0]), + wantError: errResp(ErrProtocolVersionMismatch, "10 (!= %d)", protocol), }, { - code: StatusMsg, data: statusData{uint32(ProtocolVersions[0]), 999, td, currentBlock, genesis}, + code: StatusMsg, data: statusData{uint32(protocol), 999, td, currentBlock, genesis}, wantError: errResp(ErrNetworkIdMismatch, "999 (!= 1)"), }, { - code: StatusMsg, data: statusData{uint32(ProtocolVersions[0]), NetworkId, td, currentBlock, common.Hash{3}}, + code: StatusMsg, data: statusData{uint32(protocol), NetworkId, td, currentBlock, common.Hash{3}}, wantError: errResp(ErrGenesisBlockMismatch, "0300000000000000000000000000000000000000000000000000000000000000 (!= %x)", genesis), }, } for i, test := range tests { - p, errc := newTestPeer("peer", pm, false) + p, errc := newTestPeer("peer", protocol, pm, false) // The send call might hang until reset because // the protocol might not read the payload. go p2p.Send(p.app, test.code, test.data) @@ -85,10 +91,15 @@ func TestStatusMsgErrors(t *testing.T) { } // This test checks that received transactions are added to the local pool. -func TestRecvTransactions(t *testing.T) { +func TestRecvTransactions60(t *testing.T) { testRecvTransactions(t, 60) } +func TestRecvTransactions61(t *testing.T) { testRecvTransactions(t, 61) } +func TestRecvTransactions62(t *testing.T) { testRecvTransactions(t, 62) } +func TestRecvTransactions63(t *testing.T) { testRecvTransactions(t, 63) } + +func testRecvTransactions(t *testing.T, protocol int) { txAdded := make(chan []*types.Transaction) pm := newTestProtocolManager(0, nil, txAdded) - p, _ := newTestPeer("peer", pm, true) + p, _ := newTestPeer("peer", protocol, pm, true) defer pm.Stop() defer p.close() @@ -109,7 +120,12 @@ func TestRecvTransactions(t *testing.T) { } // This test checks that pending transactions are sent. -func TestSendTransactions(t *testing.T) { +func TestSendTransactions60(t *testing.T) { testSendTransactions(t, 60) } +func TestSendTransactions61(t *testing.T) { testSendTransactions(t, 61) } +func TestSendTransactions62(t *testing.T) { testSendTransactions(t, 62) } +func TestSendTransactions63(t *testing.T) { testSendTransactions(t, 63) } + +func testSendTransactions(t *testing.T, protocol int) { pm := newTestProtocolManager(0, nil, nil) defer pm.Stop() @@ -156,7 +172,7 @@ func TestSendTransactions(t *testing.T) { } } for i := 0; i < 3; i++ { - p, _ := newTestPeer(fmt.Sprintf("peer #%d", i), pm, true) + p, _ := newTestPeer(fmt.Sprintf("peer #%d", i), protocol, pm, true) wg.Add(1) go checktxs(p) } From 7e89046d7d6ffefe85028baab9e390340772562c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 6 Jul 2015 14:00:27 +0300 Subject: [PATCH 5/8] eth, eth/downloader: support receipt retrieval (eth/63) --- eth/downloader/downloader.go | 11 ++++--- eth/handler.go | 42 ++++++++++++++++++++---- eth/handler_test.go | 63 ++++++++++++++++++++++++++++++++++-- eth/peer.go | 12 +++++++ 4 files changed, 114 insertions(+), 14 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index bd45af655c..6a6bce6445 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -42,11 +42,12 @@ 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 - MaxHeaderFetch = 256 // Amount of block headers to be fetched per retrieval request - MaxStateFetch = 384 // Amount of node + 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 + MaxStateFetch = 384 // Amount of node state values to allow fetching per request + MaxReceiptsFetch = 384 // Amount of transaction receipts to allow fetching per 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 b64ee869aa..f17ebe1803 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -57,9 +57,10 @@ 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 { - txpool txPool - chainman *core.ChainManager - statedb common.Database + txpool txPool + chainman *core.ChainManager + chaindb common.Database + downloader *downloader.Downloader fetcher *fetcher.Fetcher peers *peerSet @@ -83,13 +84,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, statedb common.Database) *ProtocolManager { +func NewProtocolManager(networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager, chaindb common.Database) *ProtocolManager { // Create the protocol manager with the base fields manager := &ProtocolManager{ eventMux: mux, txpool: txpool, chainman: chainman, - statedb: statedb, + chaindb: chaindb, peers: newPeerSet(), newPeerCh: make(chan *peer, 1), txsyncCh: make(chan *txsync), @@ -375,20 +376,47 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { data [][]byte ) for bytes < softResponseLimit && len(data) < downloader.MaxStateFetch { - //Retrieve the hash of the next state entry + // 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 { + if entry, err := pm.chaindb.Get(hash.Bytes()); err == nil { data = append(data, entry) bytes += len(entry) } } return p.SendNodeData(data) + case p.version == eth63 && msg.Code == GetReceiptsMsg: + // 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 + receipts []*types.Receipt + ) + for bytes < softResponseLimit && len(receipts) < downloader.MaxReceiptsFetch { + // Retrieve the hash of the next transaction receipt + if err := msgStream.Decode(&hash); err == rlp.EOL { + break + } else if err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + // Retrieve the requested receipt, stopping if enough was found + if receipt := core.GetReceipt(pm.chaindb, hash); receipt != nil { + receipts = append(receipts, receipt) + bytes += len(receipt.RlpEncode()) + } + } + return p.SendReceipts(receipts) + case msg.Code == NewBlockHashesMsg: // Retrieve and deseralize the remote new block hashes notification var hashes []common.Hash diff --git a/eth/handler_test.go b/eth/handler_test.go index 84c977b0e3..1acbdfc7a0 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -283,9 +283,9 @@ func TestGetNodeData63(t *testing.T) { peer, _ := newTestPeer("peer", 63, pm, true) defer peer.close() - // Fetch for now the entire state db + // Fetch for now the entire chain db hashes := []common.Hash{} - for _, key := range pm.statedb.(*ethdb.MemDatabase).Keys() { + for _, key := range pm.chaindb.(*ethdb.MemDatabase).Keys() { hashes = append(hashes, common.BytesToHash(key)) } p2p.Send(peer.app, 0x0d, hashes) @@ -327,3 +327,62 @@ func TestGetNodeData63(t *testing.T) { } } } + +// Tests that the transaction receipts can be retrieved based on hashes. +func TestGetReceipts63(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", 63, pm, true) + defer peer.close() + + // Collect the hashes to request, and the response to expect + hashes := []common.Hash{} + for i := uint64(0); i <= pm.chainman.CurrentBlock().NumberU64(); i++ { + for _, tx := range pm.chainman.GetBlockByNumber(i).Transactions() { + hashes = append(hashes, tx.Hash()) + } + } + receipts := make([]*types.Receipt, len(hashes)) + for i, hash := range hashes { + receipts[i] = core.GetReceipt(pm.chaindb, hash) + } + // Send the hash request and verify the response + p2p.Send(peer.app, 0x0f, hashes) + if err := p2p.ExpectMsg(peer.app, 0x10, receipts); err != nil { + t.Errorf("receipts mismatch: %v", err) + } +} diff --git a/eth/peer.go b/eth/peer.go index c17cdfca78..65159a7d10 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -171,6 +171,12 @@ func (p *peer) SendNodeData(data [][]byte) error { return p2p.Send(p.rw, NodeDataMsg, data) } +// SendReceipts sends a batch of transaction receipts, corresponding to the ones +// requested. +func (p *peer) SendReceipts(receipts []*types.Receipt) error { + return p2p.Send(p.rw, ReceiptsMsg, receipts) +} + // RequestHashes fetches a batch of hashes from a peer, starting at from, going // towards the genesis block. func (p *peer) RequestHashes(from common.Hash) error { @@ -205,6 +211,12 @@ func (p *peer) RequestNodeData(hashes []common.Hash) error { return p2p.Send(p.rw, GetNodeDataMsg, hashes) } +// RequestReceipts fetches a batch of transaction receipts from a remote node. +func (p *peer) RequestReceipts(hashes []common.Hash) error { + glog.V(logger.Debug).Infof("%v fetching %v receipts\n", p, len(hashes)) + return p2p.Send(p.rw, GetReceiptsMsg, hashes) +} + // Handshake executes the eth protocol handshake, negotiating version number, // network IDs, difficulties, head and genesis blocks. func (p *peer) Handshake(td *big.Int, head common.Hash, genesis common.Hash) error { From d4e11c312d34b1658d56993ff64bfc41fa9422a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 11 Aug 2015 18:34:33 +0300 Subject: [PATCH 6/8] eth, eth/fetcher: add missing eth62 messages, test bodies --- eth/fetcher/fetcher.go | 8 ++-- eth/fetcher/fetcher_test.go | 26 +++++----- eth/handler.go | 94 +++++++++++++++++++++++++++--------- eth/handler_test.go | 95 ++++++++++++++++++++++++++++++++++--- eth/peer.go | 5 ++ eth/protocol.go | 28 ++++++++++- eth/protocol_test.go | 3 ++ 7 files changed, 212 insertions(+), 47 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 55b6c5c1c5..07eb165dc5 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -69,8 +69,9 @@ type peerDropFn func(id string) // announce is the hash notification of the availability of a new block in the // network. type announce struct { - hash common.Hash // Hash of the block being announced - time time.Time // Timestamp of the announcement + hash common.Hash // Hash of the block being announced + number uint64 // Number of the block being announced (0 = unknown | old protocol) + time time.Time // Timestamp of the announcement origin string // Identifier of the peer originating the notification fetch blockRequesterFn // Fetcher function to retrieve @@ -152,9 +153,10 @@ func (f *Fetcher) Stop() { // Notify announces the fetcher of the potential availability of a new block in // the network. -func (f *Fetcher) Notify(peer string, hash common.Hash, time time.Time, fetcher blockRequesterFn) error { +func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time.Time, fetcher blockRequesterFn) error { block := &announce{ hash: hash, + number: number, time: time, origin: peer, fetch: fetcher, diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index ecbb3f868c..b0d9ce843a 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -194,7 +194,7 @@ func TestSequentialAnnouncements(t *testing.T) { tester.fetcher.importedHook = func(block *types.Block) { imported <- block } for i := len(hashes) - 2; i >= 0; i-- { - tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher) + tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher) verifyImportEvent(t, imported) } verifyImportDone(t, imported) @@ -221,9 +221,9 @@ func TestConcurrentAnnouncements(t *testing.T) { tester.fetcher.importedHook = func(block *types.Block) { imported <- block } for i := len(hashes) - 2; i >= 0; i-- { - tester.fetcher.Notify("first", hashes[i], time.Now().Add(-arriveTimeout), wrapper) - tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout+time.Millisecond), wrapper) - tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout-time.Millisecond), wrapper) + tester.fetcher.Notify("first", hashes[i], 0, time.Now().Add(-arriveTimeout), wrapper) + tester.fetcher.Notify("second", hashes[i], 0, time.Now().Add(-arriveTimeout+time.Millisecond), wrapper) + tester.fetcher.Notify("second", hashes[i], 0, time.Now().Add(-arriveTimeout-time.Millisecond), wrapper) verifyImportEvent(t, imported) } @@ -252,7 +252,7 @@ func TestOverlappingAnnouncements(t *testing.T) { tester.fetcher.importedHook = func(block *types.Block) { imported <- block } for i := len(hashes) - 2; i >= 0; i-- { - tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher) + tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher) select { case <-fetching: case <-time.After(time.Second): @@ -286,7 +286,7 @@ func TestPendingDeduplication(t *testing.T) { } // Announce the same block many times until it's fetched (wait for any pending ops) for tester.getBlock(hashes[0]) == nil { - tester.fetcher.Notify("repeater", hashes[0], time.Now().Add(-arriveTimeout), wrapper) + tester.fetcher.Notify("repeater", hashes[0], 0, time.Now().Add(-arriveTimeout), wrapper) time.Sleep(time.Millisecond) } time.Sleep(delay) @@ -317,12 +317,12 @@ func TestRandomArrivalImport(t *testing.T) { for i := len(hashes) - 1; i >= 0; i-- { if i != skip { - tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher) + tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher) time.Sleep(time.Millisecond) } } // Finally announce the skipped entry and check full import - tester.fetcher.Notify("valid", hashes[skip], time.Now().Add(-arriveTimeout), fetcher) + tester.fetcher.Notify("valid", hashes[skip], 0, time.Now().Add(-arriveTimeout), fetcher) verifyImportCount(t, imported, len(hashes)-1) } @@ -343,7 +343,7 @@ func TestQueueGapFill(t *testing.T) { for i := len(hashes) - 1; i >= 0; i-- { if i != skip { - tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher) + tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher) time.Sleep(time.Millisecond) } } @@ -374,7 +374,7 @@ func TestImportDeduplication(t *testing.T) { tester.fetcher.importedHook = func(block *types.Block) { imported <- block } // Announce the duplicating block, wait for retrieval, and also propagate directly - tester.fetcher.Notify("valid", hashes[0], time.Now().Add(-arriveTimeout), fetcher) + tester.fetcher.Notify("valid", hashes[0], 0, time.Now().Add(-arriveTimeout), fetcher) <-fetching tester.fetcher.Enqueue("valid", blocks[hashes[0]]) @@ -437,9 +437,9 @@ func TestHashMemoryExhaustionAttack(t *testing.T) { // Feed the tester a huge hashset from the attacker, and a limited from the valid peer for i := 0; i < len(attack); i++ { if i < maxQueueDist { - tester.fetcher.Notify("valid", hashes[len(hashes)-2-i], time.Now(), valid) + tester.fetcher.Notify("valid", hashes[len(hashes)-2-i], 0, time.Now(), valid) } - tester.fetcher.Notify("attacker", attack[i], time.Now(), attacker) + tester.fetcher.Notify("attacker", attack[i], 0, time.Now(), attacker) } if len(tester.fetcher.announced) != hashLimit+maxQueueDist { t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), hashLimit+maxQueueDist) @@ -449,7 +449,7 @@ func TestHashMemoryExhaustionAttack(t *testing.T) { // Feed the remaining valid hashes to ensure DOS protection state remains clean for i := len(hashes) - maxQueueDist - 2; i >= 0; i-- { - tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), valid) + tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), valid) verifyImportEvent(t, imported) } verifyImportDone(t, imported) diff --git a/eth/handler.go b/eth/handler.go index f17ebe1803..689d6d5ae2 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -237,7 +237,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { // Status messages should never arrive after the handshake return errResp(ErrExtraStatusMsg, "uncontrolled status message") - case (p.version == eth60 || p.version == eth61) && msg.Code == GetBlockHashesMsg: + case p.version < eth62 && msg.Code == GetBlockHashesMsg: // Retrieve the number of hashes to return and from which origin hash var request getBlockHashesData if err := msg.Decode(&request); err != nil { @@ -253,7 +253,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendBlockHashes(hashes) - case (p.version == eth60 || p.version == eth61) && msg.Code == GetBlockHashesFromNumberMsg: + case p.version < eth62 && msg.Code == GetBlockHashesFromNumberMsg: // Retrieve and decode the number of hashes to return and from which origin number var request getBlockHashesFromNumberData if err := msg.Decode(&request); err != nil { @@ -280,7 +280,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendBlockHashes(hashes) - case (p.version == eth60 || p.version == eth61) && msg.Code == BlockHashesMsg: + case p.version < eth62 && msg.Code == BlockHashesMsg: // A batch of hashes arrived to one of our previous requests var hashes []common.Hash if err := msg.Decode(&hashes); err != nil { @@ -292,7 +292,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { glog.V(logger.Debug).Infoln(err) } - case (p.version == eth60 || p.version == eth61) && msg.Code == GetBlocksMsg: + case p.version < eth62 && msg.Code == GetBlocksMsg: // Decode the retrieval message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) if _, err := msgStream.List(); err != nil { @@ -320,7 +320,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendBlocks(blocks) - case (p.version == eth60 || p.version == eth61) && msg.Code == BlocksMsg: + case p.version < eth62 && msg.Code == BlocksMsg: // Decode the arrived block message var blocks []*types.Block if err := msg.Decode(&blocks); err != nil { @@ -336,7 +336,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { pm.downloader.DeliverBlocks(p.id, blocks) } - case p.version == eth62 && msg.Code == GetBlockHeadersMsg: + case p.version >= eth62 && msg.Code == GetBlockHeadersMsg: // Decode the retrieval message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) if _, err := msgStream.List(); err != nil { @@ -363,7 +363,34 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendBlockHeaders(headers) - case p.version == eth63 && msg.Code == GetNodeDataMsg: + case p.version >= eth62 && msg.Code == GetBlockBodiesMsg: + // 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 + bodies []*blockBody + ) + for bytes < softResponseLimit && len(bodies) < downloader.MaxBlockFetch { + //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 { + bodies = append(bodies, &blockBody{Transactions: block.Transactions(), Uncles: block.Uncles()}) + bytes += block.Size() + } + } + return p.SendBlockBodies(bodies) + + case p.version >= eth63 && msg.Code == GetNodeDataMsg: // Decode the retrieval message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) if _, err := msgStream.List(); err != nil { @@ -390,7 +417,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } return p.SendNodeData(data) - case p.version == eth63 && msg.Code == GetReceiptsMsg: + case p.version >= eth63 && msg.Code == GetReceiptsMsg: // Decode the retrieval message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) if _, err := msgStream.List(); err != nil { @@ -419,24 +446,45 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { case msg.Code == NewBlockHashesMsg: // Retrieve and deseralize the remote new block hashes notification - var hashes []common.Hash - if err := msg.Decode(&hashes); err != nil { - break + type announce struct { + Hash common.Hash + Number uint64 } - // Mark the hashes as present at the remote node - for _, hash := range hashes { - p.MarkBlock(hash) - p.SetHead(hash) - } - // Schedule all the unknown hashes for retrieval - unknown := make([]common.Hash, 0, len(hashes)) - for _, hash := range hashes { - if !pm.chainman.HasBlock(hash) { - unknown = append(unknown, hash) + var announces = []announce{} + + if p.version < eth62 { + // We're running the old protocol, make block number unknown (0) + var hashes []common.Hash + if err := msg.Decode(&hashes); err != nil { + return errResp(ErrDecode, "%v: %v", msg, err) + } + for _, hash := range hashes { + announces = append(announces, announce{hash, 0}) + } + } else { + // Otherwise extract both block hash and number + var request newBlockHashesData + if err := msg.Decode(&request); err != nil { + return errResp(ErrDecode, "%v: %v", msg, err) + } + for _, block := range request { + announces = append(announces, announce{block.Hash, block.Number}) } } - for _, hash := range unknown { - pm.fetcher.Notify(p.id, hash, time.Now(), p.RequestBlocks) + // Mark the hashes as present at the remote node + for _, block := range announces { + p.MarkBlock(block.Hash) + p.SetHead(block.Hash) + } + // Schedule all the unknown hashes for retrieval + unknown := make([]announce, 0, len(announces)) + for _, block := range announces { + if !pm.chainman.HasBlock(block.Hash) { + unknown = append(unknown, block) + } + } + for _, block := range unknown { + pm.fetcher.Notify(p.id, block.Hash, block.Number, time.Now(), p.RequestBlocks) } case msg.Code == NewBlockMsg: diff --git a/eth/handler_test.go b/eth/handler_test.go index 1acbdfc7a0..bde44a22c7 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -175,9 +175,13 @@ func testGetBlocks(t *testing.T, protocol int) { } // Tests that block headers can be retrieved from a remote chain based on their hashes. -func TestGetBlockHeaders62(t *testing.T) { +func TestGetBlockHeaders62(t *testing.T) { testGetBlockHeaders(t, 62) } +func TestGetBlockHeaders63(t *testing.T) { testGetBlockHeaders(t, 63) } +func TestGetBlockHeaders64(t *testing.T) { testGetBlockHeaders(t, 64) } + +func testGetBlockHeaders(t *testing.T, protocol int) { pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil) - peer, _ := newTestPeer("peer", 62, pm, true) + peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() // Create a batch of tests for various scenarios @@ -242,8 +246,84 @@ func TestGetBlockHeaders62(t *testing.T) { } } +// Tests that block contents can be retrieved from a remote chain based on their hashes. +func TestGetBlockBodies62(t *testing.T) { testGetBlockBodies(t, 62) } +func TestGetBlockBodies63(t *testing.T) { testGetBlockBodies(t, 63) } +func TestGetBlockBodies64(t *testing.T) { testGetBlockBodies(t, 64) } + +func testGetBlockBodies(t *testing.T, protocol int) { + pm := newTestProtocolManager(downloader.MaxBlockFetch+15, nil, nil) + peer, _ := newTestPeer("peer", protocol, 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) + bodies := []*blockBody{} + + 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(bodies) < tt.expected { + bodies = append(bodies, &blockBody{Transactions: block.Transactions(), Uncles: block.Uncles()}) + } + break + } + } + } + for j, hash := range tt.explicit { + hashes = append(hashes, hash) + if tt.available[j] && len(bodies) < tt.expected { + block := pm.chainman.GetBlock(hash) + bodies = append(bodies, &blockBody{Transactions: block.Transactions(), Uncles: block.Uncles()}) + } + } + // Send the hash request and verify the response + p2p.Send(peer.app, 0x05, hashes) + if err := p2p.ExpectMsg(peer.app, 0x06, bodies); err != nil { + t.Errorf("test %d: bodies mismatch: %v", i, err) + } + } +} + // Tests that the node state database can be retrieved based on hashes. -func TestGetNodeData63(t *testing.T) { +func TestGetNodeData63(t *testing.T) { testGetNodeData(t, 63) } +func TestGetNodeData64(t *testing.T) { testGetNodeData(t, 64) } + +func testGetNodeData(t *testing.T, protocol int) { // Define three accounts to simulate transactions with acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a") acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee") @@ -280,7 +360,7 @@ func TestGetNodeData63(t *testing.T) { } // Assemble the test environment pm := newTestProtocolManager(4, generator, nil) - peer, _ := newTestPeer("peer", 63, pm, true) + peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() // Fetch for now the entire chain db @@ -329,7 +409,10 @@ func TestGetNodeData63(t *testing.T) { } // Tests that the transaction receipts can be retrieved based on hashes. -func TestGetReceipts63(t *testing.T) { +func TestGetReceipt63(t *testing.T) { testGetReceipt(t, 63) } +func TestGetReceipt64(t *testing.T) { testGetReceipt(t, 64) } + +func testGetReceipt(t *testing.T, protocol int) { // Define three accounts to simulate transactions with acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a") acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee") @@ -366,7 +449,7 @@ func TestGetReceipts63(t *testing.T) { } // Assemble the test environment pm := newTestProtocolManager(4, generator, nil) - peer, _ := newTestPeer("peer", 63, pm, true) + peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() // Collect the hashes to request, and the response to expect diff --git a/eth/peer.go b/eth/peer.go index 65159a7d10..78de8a9d34 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -165,6 +165,11 @@ func (p *peer) SendBlockHeaders(headers []*types.Header) error { return p2p.Send(p.rw, BlockHeadersMsg, headers) } +// SendBlockBodies sends a batch of block contents to the remote peer. +func (p *peer) SendBlockBodies(bodies []*blockBody) error { + return p2p.Send(p.rw, BlockBodiesMsg, blockBodiesData(bodies)) +} + // SendNodeData sends a batch of arbitrary internal data, corresponding to the // hashes requested. func (p *peer) SendNodeData(data [][]byte) error { diff --git a/eth/protocol.go b/eth/protocol.go index fc8446589b..8baabef27f 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -29,6 +29,7 @@ const ( eth61 = 61 eth62 = 62 eth63 = 63 + eth64 = 64 ) // Supported versions of the eth protocol (first is primary). @@ -54,10 +55,13 @@ const ( BlocksMsg = 0x06 NewBlockMsg = 0x07 - // Protocol messages belonging to eth/61 + // Protocol messages belonging to eth/61 (extension of eth/60) GetBlockHashesFromNumberMsg = 0x08 - // Protocol messages belonging to eth/62 + // Protocol messages belonging to eth/62 (new protocol from scratch) + // StatusMsg = 0x00 (uncomment after eth/61 deprecation) + // NewBlockHashesMsg = 0x01 (uncomment after eth/61 deprecation) + // TxMsg = 0x02 (uncomment after eth/61 deprecation) GetBlockHeadersMsg = 0x03 BlockHeadersMsg = 0x04 GetBlockBodiesMsg = 0x05 @@ -68,6 +72,11 @@ const ( NodeDataMsg = 0x0e GetReceiptsMsg = 0x0f ReceiptsMsg = 0x10 + + // Protocol messages belonging to eth/64 + GetAcctProofMsg = 0x11 + GetStorageDataProof = 0x12 + Proof = 0x13 ) type errCode int @@ -125,6 +134,12 @@ type statusData struct { GenesisBlock common.Hash } +// newBlockHashesData is the network packet for the block announcements. +type newBlockHashesData []struct { + Hash common.Hash // Hash of one particular block being announced + Number uint64 // Number of one particular block being announced +} + // getBlockHashesData is the network packet for the hash based hash retrieval. type getBlockHashesData struct { Hash common.Hash @@ -144,6 +159,15 @@ type newBlockData struct { TD *big.Int } +// blockBody represents the data content of a single block. +type blockBody struct { + Transactions []*types.Transaction // Transactions contained within a block + Uncles []*types.Header // Uncles contained within a block +} + +// blockBodiesData is the network packet for block content distribution. +type blockBodiesData []*blockBody + // nodeDataData is the network response packet for a node data retrieval. type nodeDataData []struct { Value []byte diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 4b89bc1f75..32d08ca587 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -41,6 +41,7 @@ func TestStatusMsgErrors60(t *testing.T) { testStatusMsgErrors(t, 60) } func TestStatusMsgErrors61(t *testing.T) { testStatusMsgErrors(t, 61) } func TestStatusMsgErrors62(t *testing.T) { testStatusMsgErrors(t, 62) } func TestStatusMsgErrors63(t *testing.T) { testStatusMsgErrors(t, 63) } +func TestStatusMsgErrors64(t *testing.T) { testStatusMsgErrors(t, 64) } func testStatusMsgErrors(t *testing.T, protocol int) { pm := newTestProtocolManager(0, nil, nil) @@ -95,6 +96,7 @@ func TestRecvTransactions60(t *testing.T) { testRecvTransactions(t, 60) } func TestRecvTransactions61(t *testing.T) { testRecvTransactions(t, 61) } func TestRecvTransactions62(t *testing.T) { testRecvTransactions(t, 62) } func TestRecvTransactions63(t *testing.T) { testRecvTransactions(t, 63) } +func TestRecvTransactions64(t *testing.T) { testRecvTransactions(t, 64) } func testRecvTransactions(t *testing.T, protocol int) { txAdded := make(chan []*types.Transaction) @@ -124,6 +126,7 @@ func TestSendTransactions60(t *testing.T) { testSendTransactions(t, 60) } func TestSendTransactions61(t *testing.T) { testSendTransactions(t, 61) } func TestSendTransactions62(t *testing.T) { testSendTransactions(t, 62) } func TestSendTransactions63(t *testing.T) { testSendTransactions(t, 63) } +func TestSendTransactions64(t *testing.T) { testSendTransactions(t, 64) } func testSendTransactions(t *testing.T, protocol int) { pm := newTestProtocolManager(0, nil, nil) From 0df317dbeb19c08d835e93299ac3ceb29b0c577e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 12 Aug 2015 13:46:36 +0300 Subject: [PATCH 7/8] eth: upgrade the block header query packet to new spec --- eth/handler.go | 68 +++++++++++++++----- eth/handler_test.go | 150 +++++++++++++++++++++++++++++-------------- eth/protocol.go | 47 ++++++++++++++ eth/protocol_test.go | 45 +++++++++++++ 4 files changed, 247 insertions(+), 63 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index 689d6d5ae2..25ff0eef03 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -336,29 +336,67 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { pm.downloader.DeliverBlocks(p.id, blocks) } + // Block header query, collect the requested headers and reply case p.version >= eth62 && msg.Code == GetBlockHeadersMsg: - // Decode the retrieval message - msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) - if _, err := msgStream.List(); err != nil { - return err + // Decode the complex header query + var query getBlockHeadersData + if err := msg.Decode(&query); err != nil { + return errResp(ErrDecode, "%v: %v", msg, err) } // Gather blocks until the fetch or network limits is reached var ( - hash common.Hash bytes common.StorageSize headers []*types.Header + unknown bool ) - 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) + for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit && len(headers) < downloader.MaxHeaderFetch { + // Retrieve the next block satisfying the query + var origin *types.Block + if query.Origin.Hash != (common.Hash{}) { + origin = pm.chainman.GetBlock(query.Origin.Hash) + } else { + origin = pm.chainman.GetBlockByNumber(query.Origin.Number) } - // 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() + if origin == nil { + break + } + headers = append(headers, origin.Header()) + bytes += origin.Size() + + // Advance to the next block of the query + switch { + case query.Origin.Hash != (common.Hash{}) && query.Reverse: + // Hash based traversal towards the genesis block + for i := 0; i < int(query.Skip)+1; i++ { + if block := pm.chainman.GetBlock(query.Origin.Hash); block != nil { + query.Origin.Hash = block.ParentHash() + } else { + unknown = true + break + } + } + case query.Origin.Hash != (common.Hash{}) && !query.Reverse: + // Hash based traversal towards the leaf block + if block := pm.chainman.GetBlockByNumber(origin.NumberU64() + query.Skip + 1); block != nil { + if pm.chainman.GetBlockHashesFromHash(block.Hash(), query.Skip+1)[query.Skip] == query.Origin.Hash { + query.Origin.Hash = block.Hash() + } else { + unknown = true + } + } else { + unknown = true + } + case query.Reverse: + // Number based traversal towards the genesis block + if query.Origin.Number >= query.Skip+1 { + query.Origin.Number -= (query.Skip + 1) + } else { + unknown = true + } + + case !query.Reverse: + // Number based traversal towards the leaf block + query.Origin.Number += (query.Skip + 1) } } return p.SendBlockHeaders(headers) diff --git a/eth/handler_test.go b/eth/handler_test.go index bde44a22c7..63c94faa12 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -174,7 +174,7 @@ func testGetBlocks(t *testing.T, protocol int) { } } -// Tests that block headers can be retrieved from a remote chain based on their hashes. +// Tests that block headers can be retrieved from a remote chain based on user queries. func TestGetBlockHeaders62(t *testing.T) { testGetBlockHeaders(t, 62) } func TestGetBlockHeaders63(t *testing.T) { testGetBlockHeaders(t, 63) } func TestGetBlockHeaders64(t *testing.T) { testGetBlockHeaders(t, 64) } @@ -184,62 +184,116 @@ func testGetBlockHeaders(t *testing.T, protocol int) { peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() + // Create a "random" unknown hash for testing + var unknown common.Hash + for i, _ := range unknown { + unknown[i] = byte(i) + } // Create a batch of tests for various scenarios - limit := downloader.MaxHeaderFetch + limit := uint64(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 + query *getBlockHeadersData // The query to execute for header retrieval + expect []common.Hash // The hashes of the block whose headers are expected }{ - {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}, + // A single random block should be retrievable by hash and number too + { + &getBlockHeadersData{Origin: hashOrNumber{Hash: pm.chainman.GetBlockByNumber(limit / 2).Hash()}, Amount: 1}, + []common.Hash{pm.chainman.GetBlockByNumber(limit / 2).Hash()}, + }, { + &getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 1}, + []common.Hash{pm.chainman.GetBlockByNumber(limit / 2).Hash()}, + }, + // Multiple headers should be retrievable in both directions + { + &getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 3}, + []common.Hash{ + pm.chainman.GetBlockByNumber(limit / 2).Hash(), + pm.chainman.GetBlockByNumber(limit/2 + 1).Hash(), + pm.chainman.GetBlockByNumber(limit/2 + 2).Hash(), + }, + }, { + &getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 3, Reverse: true}, + []common.Hash{ + pm.chainman.GetBlockByNumber(limit / 2).Hash(), + pm.chainman.GetBlockByNumber(limit/2 - 1).Hash(), + pm.chainman.GetBlockByNumber(limit/2 - 2).Hash(), + }, + }, + // Multiple headers with skip lists should be retrievable + { + &getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3}, + []common.Hash{ + pm.chainman.GetBlockByNumber(limit / 2).Hash(), + pm.chainman.GetBlockByNumber(limit/2 + 4).Hash(), + pm.chainman.GetBlockByNumber(limit/2 + 8).Hash(), + }, + }, { + &getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3, Reverse: true}, + []common.Hash{ + pm.chainman.GetBlockByNumber(limit / 2).Hash(), + pm.chainman.GetBlockByNumber(limit/2 - 4).Hash(), + pm.chainman.GetBlockByNumber(limit/2 - 8).Hash(), + }, + }, + // The chain endpoints should be retrievable + { + &getBlockHeadersData{Origin: hashOrNumber{Number: 0}, Amount: 1}, + []common.Hash{pm.chainman.GetBlockByNumber(0).Hash()}, + }, { + &getBlockHeadersData{Origin: hashOrNumber{Number: pm.chainman.CurrentBlock().NumberU64()}, Amount: 1}, + []common.Hash{pm.chainman.CurrentBlock().Hash()}, + }, + // Ensure protocol limits are honored + { + &getBlockHeadersData{Origin: hashOrNumber{Number: pm.chainman.CurrentBlock().NumberU64() - 1}, Amount: limit + 10, Reverse: true}, + pm.chainman.GetBlockHashesFromHash(pm.chainman.CurrentBlock().Hash(), limit), + }, + // Check that requesting more than available is handled gracefully + { + &getBlockHeadersData{Origin: hashOrNumber{Number: pm.chainman.CurrentBlock().NumberU64() - 4}, Skip: 3, Amount: 3}, + []common.Hash{ + pm.chainman.GetBlockByNumber(pm.chainman.CurrentBlock().NumberU64() - 4).Hash(), + pm.chainman.GetBlockByNumber(pm.chainman.CurrentBlock().NumberU64()).Hash(), + }, + }, { + &getBlockHeadersData{Origin: hashOrNumber{Number: 4}, Skip: 3, Amount: 3, Reverse: true}, + []common.Hash{ + pm.chainman.GetBlockByNumber(4).Hash(), + pm.chainman.GetBlockByNumber(0).Hash(), + }, + }, + // Check that requesting more than available is handled gracefully, even if mid skip + { + &getBlockHeadersData{Origin: hashOrNumber{Number: pm.chainman.CurrentBlock().NumberU64() - 4}, Skip: 2, Amount: 3}, + []common.Hash{ + pm.chainman.GetBlockByNumber(pm.chainman.CurrentBlock().NumberU64() - 4).Hash(), + pm.chainman.GetBlockByNumber(pm.chainman.CurrentBlock().NumberU64() - 1).Hash(), + }, + }, { + &getBlockHeadersData{Origin: hashOrNumber{Number: 4}, Skip: 2, Amount: 3, Reverse: true}, + []common.Hash{ + pm.chainman.GetBlockByNumber(4).Hash(), + pm.chainman.GetBlockByNumber(1).Hash(), + }, + }, + // Check that non existing headers aren't returned + { + &getBlockHeadersData{Origin: hashOrNumber{Hash: unknown}, Amount: 1}, + []common.Hash{}, + }, { + &getBlockHeadersData{Origin: hashOrNumber{Number: pm.chainman.CurrentBlock().NumberU64() + 1}, Amount: 1}, + []common.Hash{}, + }, } // 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) + // Collect the headers to expect in the response 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()) - } + for _, hash := range tt.expect { + headers = append(headers, pm.chainman.GetBlock(hash).Header()) } // Send the hash request and verify the response - p2p.Send(peer.app, 0x03, hashes) + p2p.Send(peer.app, 0x03, tt.query) if err := p2p.ExpectMsg(peer.app, 0x04, headers); err != nil { t.Errorf("test %d: headers mismatch: %v", i, err) } diff --git a/eth/protocol.go b/eth/protocol.go index 8baabef27f..8bed8d22ec 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -17,10 +17,13 @@ package eth import ( + "fmt" + "io" "math/big" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/rlp" ) // Constants to match up protocol versions and messages @@ -153,6 +156,50 @@ type getBlockHashesFromNumberData struct { Amount uint64 } +// getBlockHeadersData represents a block header query. +type getBlockHeadersData struct { + Origin hashOrNumber // Block from which to retrieve headers + Amount uint64 // Maximum number of headers to retrieve + Skip uint64 // Blocks to skip between consecutive headers + Reverse bool // Query direction (false = rising towards latest, true = falling towards genesis) +} + +// hashOrNumber is a combined field for specifying an origin block. +type hashOrNumber struct { + Hash common.Hash // Block hash from which to retrieve headers (excludes Number) + Number uint64 // Block hash from which to retrieve headers (excludes Hash) +} + +// EncodeRLP is a specialized encoder for hashOrNumber to encode only one of the +// two contained union fields. +func (hn *hashOrNumber) EncodeRLP(w io.Writer) error { + if hn.Hash == (common.Hash{}) { + return rlp.Encode(w, hn.Number) + } + if hn.Number != 0 { + return fmt.Errorf("both origin hash (%x) and number (%d) provided", hn.Hash, hn.Number) + } + return rlp.Encode(w, hn.Hash) +} + +// DecodeRLP is a specialized decoder for hashOrNumber to decode the contents +// into either a block hash or a block number. +func (hn *hashOrNumber) DecodeRLP(s *rlp.Stream) error { + _, size, _ := s.Kind() + origin, err := s.Raw() + if err == nil { + switch { + case size == 32: + err = rlp.DecodeBytes(origin, &hn.Hash) + case size <= 8: + err = rlp.DecodeBytes(origin, &hn.Number) + default: + err = fmt.Errorf("invalid input size %d for origin", size) + } + } + return err +} + // newBlockData is the network packet for the block propagation message. type newBlockData struct { Block *types.Block diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 32d08ca587..263088099d 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -27,6 +27,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rlp" ) func init() { @@ -181,3 +182,47 @@ func testSendTransactions(t *testing.T, protocol int) { } wg.Wait() } + +// Tests that the custom union field encoder and decoder works correctly. +func TestGetBlockHeadersDataEncodeDecode(t *testing.T) { + // Create a "random" hash for testing + var hash common.Hash + for i, _ := range hash { + hash[i] = byte(i) + } + // Assemble some table driven tests + tests := []struct { + packet *getBlockHeadersData + fail bool + }{ + // Providing the origin as either a hash or a number should both work + {fail: false, packet: &getBlockHeadersData{Origin: hashOrNumber{Number: 314}}}, + {fail: false, packet: &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}}}, + + // Providing arbitrary query field should also work + {fail: false, packet: &getBlockHeadersData{Origin: hashOrNumber{Number: 314}, Amount: 314, Skip: 1, Reverse: true}}, + {fail: false, packet: &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}, Amount: 314, Skip: 1, Reverse: true}}, + + // Providing both the origin hash and origin number must fail + {fail: true, packet: &getBlockHeadersData{Origin: hashOrNumber{Hash: hash, Number: 314}}}, + } + // Iterate over each of the tests and try to encode and then decode + for i, tt := range tests { + bytes, err := rlp.EncodeToBytes(tt.packet) + if err != nil && !tt.fail { + t.Fatalf("test %d: failed to encode packet: %v", i, err) + } else if err == nil && tt.fail { + t.Fatalf("test %d: encode should have failed", i) + } + if !tt.fail { + packet := new(getBlockHeadersData) + if err := rlp.DecodeBytes(bytes, packet); err != nil { + t.Fatalf("test %d: failed to decode packet: %v", i, err) + } + if packet.Origin.Hash != tt.packet.Origin.Hash || packet.Origin.Number != tt.packet.Origin.Number || packet.Amount != tt.packet.Amount || + packet.Skip != tt.packet.Skip || packet.Reverse != tt.packet.Reverse { + t.Fatalf("test %d: encode decode mismatch: have %+v, want %+v", i, packet, tt.packet) + } + } + } +} From 16f84cf1a639e65e7022c16e08b38a7b069551d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 13 Aug 2015 13:36:08 +0300 Subject: [PATCH 8/8] cmd, eth: flag for enabling eth protocol versions --- cmd/geth/main.go | 2 ++ cmd/utils/flags.go | 17 +++++++++++++++++ eth/protocol.go | 5 +++-- 3 files changed, 22 insertions(+), 2 deletions(-) diff --git a/cmd/geth/main.go b/cmd/geth/main.go index ff556c984e..dc7e19c61a 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -283,6 +283,7 @@ JavaScript API. See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Conso utils.DataDirFlag, utils.BlockchainVersionFlag, utils.OlympicFlag, + utils.EthVersionFlag, utils.CacheFlag, utils.JSpathFlag, utils.ListenPortFlag, @@ -333,6 +334,7 @@ JavaScript API. See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Conso app.Before = func(ctx *cli.Context) error { utils.SetupLogger(ctx) utils.SetupVM(ctx) + utils.SetupEth(ctx) if ctx.GlobalBool(utils.PProfEanbledFlag.Name) { utils.StartPProf(ctx) } diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index af2929d102..5ebc4ea61e 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -138,6 +138,11 @@ var ( Name: "olympic", Usage: "Use olympic style protocol", } + EthVersionFlag = cli.IntFlag{ + Name: "eth", + Value: 61, + Usage: "Highest eth protocol to advertise (temporary, dev option)", + } // miner settings MinerThreadsFlag = cli.IntFlag{ @@ -459,6 +464,18 @@ func SetupVM(ctx *cli.Context) { vm.SetJITCacheSize(ctx.GlobalInt(VMJitCacheFlag.Name)) } +// SetupEth configures the eth packages global settings +func SetupEth(ctx *cli.Context) { + version := ctx.GlobalInt(EthVersionFlag.Name) + for len(eth.ProtocolVersions) > 0 && eth.ProtocolVersions[0] > uint(version) { + eth.ProtocolVersions = eth.ProtocolVersions[1:] + eth.ProtocolLengths = eth.ProtocolLengths[1:] + } + if len(eth.ProtocolVersions) == 0 { + Fatalf("No valid eth protocols remaining") + } +} + // MakeChain creates a chain manager from set command line flags. func MakeChain(ctx *cli.Context) (chain *core.ChainManager, chainDb common.Database) { datadir := ctx.GlobalString(DataDirFlag.Name) diff --git a/eth/protocol.go b/eth/protocol.go index 8bed8d22ec..c16223ccf5 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -36,10 +36,10 @@ const ( ) // Supported versions of the eth protocol (first is primary). -var ProtocolVersions = []uint{61, 60} +var ProtocolVersions = []uint{eth64, eth63, eth62, eth61, eth60} // Number of implemented message corresponding to different protocol versions. -var ProtocolLengths = []uint64{9, 8} +var ProtocolLengths = []uint64{15, 12, 8, 9, 8} const ( NetworkId = 1 @@ -69,6 +69,7 @@ const ( BlockHeadersMsg = 0x04 GetBlockBodiesMsg = 0x05 BlockBodiesMsg = 0x06 + // NewBlockMsg = 0x07 (uncomment after eth/61 deprecation) // Protocol messages belonging to eth/63 GetNodeDataMsg = 0x0d