forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-23 21:26:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth, eth/downloader: support overlapping messages between proto versions

Browse source
This commit is contained in:
Péter Szilágyi 2015-07-06 13:21:44 +03:00
parent 4644814a85
commit 5c1cded016
7 changed files with 176 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
eth/downloader/downloader.go
View file

@ -39,7 +39,6 @@ import (
const ( const (
eth60 = 60 // Constant to check for old protocol support eth60 = 60 // Constant to check for old protocol support
eth61 = 61 // Constant to check for new protocol support eth61 = 61 // Constant to check for new protocol support
eth62 = 62 // Constant to check for experimental protocol support
) )
var ( 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 { if err = d.fetchBlocks60(); err != nil {
return err return err
} }
case eth61, eth62: case eth61:
// New eth/61, use forward, concurrent hash and block retrieval algorithm // New eth/61, use forward, concurrent hash and block retrieval algorithm
number, err := d.findAncestor(p) number, err := d.findAncestor(p)
if err != nil { if err != nil {

27
eth/handler.go
View file

@ -188,6 +188,9 @@ func (pm *ProtocolManager) handle(p *peer) error {
glog.V(logger.Debug).Infof("%v: handshake failed: %v", p, err) glog.V(logger.Debug).Infof("%v: handshake failed: %v", p, err)
return err return err
} }
if rw, ok := p.rw.(*meteredMsgReadWriter); ok {
rw.Init(p.version)
}
// Register the peer locally // Register the peer locally
glog.V(logger.Detail).Infof("%v: adding peer", p) glog.V(logger.Detail).Infof("%v: adding peer", p)
if err := pm.peers.Register(p); err != nil { if err := pm.peers.Register(p); err != nil {
@ -228,12 +231,12 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
defer msg.Discard() defer msg.Discard()
// Handle the message depending on its contents // Handle the message depending on its contents
switch msg.Code { switch {
case StatusMsg: case msg.Code == StatusMsg:
// Status messages should never arrive after the handshake // Status messages should never arrive after the handshake
return errResp(ErrExtraStatusMsg, "uncontrolled status message") 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 // Retrieve the number of hashes to return and from which origin hash
var request getBlockHashesData var request getBlockHashesData
if err := msg.Decode(&request); err != nil { if err := msg.Decode(&request); err != nil {
@ -249,7 +252,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
} }
return p.SendBlockHashes(hashes) 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 // Retrieve and decode the number of hashes to return and from which origin number
var request getBlockHashesFromNumberData var request getBlockHashesFromNumberData
if err := msg.Decode(&request); err != nil { if err := msg.Decode(&request); err != nil {
@ -276,7 +279,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
} }
return p.SendBlockHashes(hashes) 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 // A batch of hashes arrived to one of our previous requests
var hashes []common.Hash var hashes []common.Hash
if err := msg.Decode(&hashes); err != nil { if err := msg.Decode(&hashes); err != nil {
@ -288,7 +291,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
glog.V(logger.Debug).Infoln(err) glog.V(logger.Debug).Infoln(err)
} }
case GetBlocksMsg: case (p.version == eth60 || p.version == eth61) && msg.Code == GetBlocksMsg:
// Decode the retrieval message // Decode the retrieval message
msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
if _, err := msgStream.List(); err != nil { if _, err := msgStream.List(); err != nil {
@ -316,7 +319,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
} }
return p.SendBlocks(blocks) return p.SendBlocks(blocks)
case BlocksMsg: case (p.version == eth60 || p.version == eth61) && msg.Code == BlocksMsg:
// Decode the arrived block message // Decode the arrived block message
var blocks []*types.Block var blocks []*types.Block
if err := msg.Decode(&blocks); err != nil { if err := msg.Decode(&blocks); err != nil {
@ -332,7 +335,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
pm.downloader.DeliverBlocks(p.id, blocks) pm.downloader.DeliverBlocks(p.id, blocks)
} }
case GetBlockHeadersMsg: case p.version == eth62 && msg.Code == GetBlockHeadersMsg:
// Decode the retrieval message // Decode the retrieval message
msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
if _, err := msgStream.List(); err != nil { if _, err := msgStream.List(); err != nil {
@ -359,7 +362,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
} }
return p.SendBlockHeaders(headers) return p.SendBlockHeaders(headers)
case GetNodeDataMsg: case p.version == eth63 && msg.Code == GetNodeDataMsg:
// Decode the retrieval message // Decode the retrieval message
msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
if _, err := msgStream.List(); err != nil { if _, err := msgStream.List(); err != nil {
@ -386,7 +389,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
} }
return p.SendNodeData(data) return p.SendNodeData(data)
case NewBlockHashesMsg: case msg.Code == NewBlockHashesMsg:
// Retrieve and deseralize the remote new block hashes notification // Retrieve and deseralize the remote new block hashes notification
var hashes []common.Hash var hashes []common.Hash
if err := msg.Decode(&hashes); err != nil { 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) pm.fetcher.Notify(p.id, hash, time.Now(), p.RequestBlocks)
} }
case NewBlockMsg: case msg.Code == NewBlockMsg:
// Retrieve and decode the propagated block // Retrieve and decode the propagated block
var request newBlockData var request newBlockData
if err := msg.Decode(&request); err != nil { if err := msg.Decode(&request); err != nil {
@ -440,7 +443,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
go pm.synchronise(p) go pm.synchronise(p)
} }
case TxMsg: case msg.Code == TxMsg:
// Transactions arrived, parse all of them and deliver to the pool // Transactions arrived, parse all of them and deliver to the pool
var txs []*types.Transaction var txs []*types.Transaction
if err := msg.Decode(&txs); err != nil { if err := msg.Decode(&txs); err != nil {

37
eth/handler_test.go
View file

@ -19,9 +19,12 @@ import (
// Tests that hashes can be retrieved from a remote chain by hashes in reverse // Tests that hashes can be retrieved from a remote chain by hashes in reverse
// order. // 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) pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil)
peer, _ := newTestPeer("peer", pm, true) peer, _ := newTestPeer("peer", protocol, pm, true)
defer peer.close() defer peer.close()
// Create a batch of tests for various scenarios // 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 // Tests that hashes can be retrieved from a remote chain by numbers in forward
// order. // 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) pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil)
peer, _ := newTestPeer("peer", pm, true) peer, _ := newTestPeer("peer", protocol, pm, true)
defer peer.close() defer peer.close()
// Create a batch of tests for various scenarios // 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. // 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) pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil)
peer, _ := newTestPeer("peer", pm, true) peer, _ := newTestPeer("peer", protocol, pm, true)
defer peer.close() defer peer.close()
// Create a batch of tests for various scenarios // 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. // 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) pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil)
peer, _ := newTestPeer("peer", pm, true) peer, _ := newTestPeer("peer", 62, pm, true)
defer peer.close() defer peer.close()
// Create a batch of tests for various scenarios // 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 // Send the hash request and verify the response
p2p.Send(peer.app, 0x09, hashes) p2p.Send(peer.app, 0x03, hashes)
if err := p2p.ExpectMsg(peer.app, 0x0a, headers); err != nil { if err := p2p.ExpectMsg(peer.app, 0x04, headers); err != nil {
t.Errorf("test %d: headers mismatch: %v", i, err) t.Errorf("test %d: headers mismatch: %v", i, err)
} }
} }
} }
// Tests that the node state database can be retrieved based on hashes. // 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 // Define three accounts to simulate transactions with
acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a") acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a")
acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee") acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee")
@ -271,7 +280,7 @@ func TestGetNodeData(t *testing.T) {
} }
// Assemble the test environment // Assemble the test environment
pm := newTestProtocolManager(4, generator, nil) pm := newTestProtocolManager(4, generator, nil)
peer, _ := newTestPeer("peer", pm, true) peer, _ := newTestPeer("peer", 63, pm, true)
defer peer.close() defer peer.close()
// Fetch for now the entire state db // Fetch for now the entire state db
@ -279,12 +288,12 @@ func TestGetNodeData(t *testing.T) {
for _, key := range pm.statedb.(*ethdb.MemDatabase).Keys() { for _, key := range pm.statedb.(*ethdb.MemDatabase).Keys() {
hashes = append(hashes, common.BytesToHash(key)) hashes = append(hashes, common.BytesToHash(key))
} }
p2p.Send(peer.app, 0x0b, hashes) p2p.Send(peer.app, 0x0d, hashes)
msg, err := peer.app.ReadMsg() msg, err := peer.app.ReadMsg()
if err != nil { if err != nil {
t.Fatalf("failed to read node data response: %v", err) 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) t.Fatalf("response packet code mismatch: have %x, want %x", msg.Code, 0x0c)
} }
var data [][]byte var data [][]byte

6
eth/helper_test.go
View file

@ -93,7 +93,7 @@ type testPeer struct {
} }
// newTestPeer creates a new peer registered at the given protocol manager. // 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 // Create a message pipe to communicate through
app, net := p2p.MsgPipe() app, net := p2p.MsgPipe()
@ -101,7 +101,7 @@ func newTestPeer(name string, pm *ProtocolManager, shake bool) (*testPeer, <-cha
var id discover.NodeID var id discover.NodeID
rand.Read(id[:]) 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 // Start the peer on a new thread
errc := make(chan error, 1) 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. // remote side as we are simulating locally.
func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, genesis common.Hash) { func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, genesis common.Hash) {
msg := &statusData{ msg := &statusData{
ProtocolVersion: uint32(ProtocolVersions[0]), ProtocolVersion: uint32(p.version),
NetworkId: uint32(NetworkId), NetworkId: uint32(NetworkId),
TD: td, TD: td,
CurrentBlock: head, CurrentBlock: head,

131
eth/metrics.go
View file

@ -22,44 +22,53 @@ import (
) )
var ( var (
propTxnInPacketsMeter = metrics.NewMeter("eth/prop/txns/in/packets") propTxnInPacketsMeter = metrics.NewMeter("eth/prop/txns/in/packets")
propTxnInTrafficMeter = metrics.NewMeter("eth/prop/txns/in/traffic") propTxnInTrafficMeter = metrics.NewMeter("eth/prop/txns/in/traffic")
propTxnOutPacketsMeter = metrics.NewMeter("eth/prop/txns/out/packets") propTxnOutPacketsMeter = metrics.NewMeter("eth/prop/txns/out/packets")
propTxnOutTrafficMeter = metrics.NewMeter("eth/prop/txns/out/traffic") propTxnOutTrafficMeter = metrics.NewMeter("eth/prop/txns/out/traffic")
propHashInPacketsMeter = metrics.NewMeter("eth/prop/hashes/in/packets") propHashInPacketsMeter = metrics.NewMeter("eth/prop/hashes/in/packets")
propHashInTrafficMeter = metrics.NewMeter("eth/prop/hashes/in/traffic") propHashInTrafficMeter = metrics.NewMeter("eth/prop/hashes/in/traffic")
propHashOutPacketsMeter = metrics.NewMeter("eth/prop/hashes/out/packets") propHashOutPacketsMeter = metrics.NewMeter("eth/prop/hashes/out/packets")
propHashOutTrafficMeter = metrics.NewMeter("eth/prop/hashes/out/traffic") propHashOutTrafficMeter = metrics.NewMeter("eth/prop/hashes/out/traffic")
propBlockInPacketsMeter = metrics.NewMeter("eth/prop/blocks/in/packets") propBlockInPacketsMeter = metrics.NewMeter("eth/prop/blocks/in/packets")
propBlockInTrafficMeter = metrics.NewMeter("eth/prop/blocks/in/traffic") propBlockInTrafficMeter = metrics.NewMeter("eth/prop/blocks/in/traffic")
propBlockOutPacketsMeter = metrics.NewMeter("eth/prop/blocks/out/packets") propBlockOutPacketsMeter = metrics.NewMeter("eth/prop/blocks/out/packets")
propBlockOutTrafficMeter = metrics.NewMeter("eth/prop/blocks/out/traffic") propBlockOutTrafficMeter = metrics.NewMeter("eth/prop/blocks/out/traffic")
reqHashInPacketsMeter = metrics.NewMeter("eth/req/hashes/in/packets") reqHashInPacketsMeter = metrics.NewMeter("eth/req/hashes/in/packets")
reqHashInTrafficMeter = metrics.NewMeter("eth/req/hashes/in/traffic") reqHashInTrafficMeter = metrics.NewMeter("eth/req/hashes/in/traffic")
reqHashOutPacketsMeter = metrics.NewMeter("eth/req/hashes/out/packets") reqHashOutPacketsMeter = metrics.NewMeter("eth/req/hashes/out/packets")
reqHashOutTrafficMeter = metrics.NewMeter("eth/req/hashes/out/traffic") reqHashOutTrafficMeter = metrics.NewMeter("eth/req/hashes/out/traffic")
reqBlockInPacketsMeter = metrics.NewMeter("eth/req/blocks/in/packets") reqBlockInPacketsMeter = metrics.NewMeter("eth/req/blocks/in/packets")
reqBlockInTrafficMeter = metrics.NewMeter("eth/req/blocks/in/traffic") reqBlockInTrafficMeter = metrics.NewMeter("eth/req/blocks/in/traffic")
reqBlockOutPacketsMeter = metrics.NewMeter("eth/req/blocks/out/packets") reqBlockOutPacketsMeter = metrics.NewMeter("eth/req/blocks/out/packets")
reqBlockOutTrafficMeter = metrics.NewMeter("eth/req/blocks/out/traffic") reqBlockOutTrafficMeter = metrics.NewMeter("eth/req/blocks/out/traffic")
reqHeaderInPacketsMeter = metrics.NewMeter("eth/req/header/in/packets") reqHeaderInPacketsMeter = metrics.NewMeter("eth/req/header/in/packets")
reqHeaderInTrafficMeter = metrics.NewMeter("eth/req/header/in/traffic") reqHeaderInTrafficMeter = metrics.NewMeter("eth/req/header/in/traffic")
reqHeaderOutPacketsMeter = metrics.NewMeter("eth/req/header/out/packets") reqHeaderOutPacketsMeter = metrics.NewMeter("eth/req/header/out/packets")
reqHeaderOutTrafficMeter = metrics.NewMeter("eth/req/header/out/traffic") reqHeaderOutTrafficMeter = metrics.NewMeter("eth/req/header/out/traffic")
reqStateInPacketsMeter = metrics.NewMeter("eth/req/state/in/packets") reqBodyInPacketsMeter = metrics.NewMeter("eth/req/body/in/packets")
reqStateInTrafficMeter = metrics.NewMeter("eth/req/state/in/traffic") reqBodyInTrafficMeter = metrics.NewMeter("eth/req/body/in/traffic")
reqStateOutPacketsMeter = metrics.NewMeter("eth/req/state/out/packets") reqBodyOutPacketsMeter = metrics.NewMeter("eth/req/body/out/packets")
reqStateOutTrafficMeter = metrics.NewMeter("eth/req/state/out/traffic") reqBodyOutTrafficMeter = metrics.NewMeter("eth/req/body/out/traffic")
miscInPacketsMeter = metrics.NewMeter("eth/misc/in/packets") reqStateInPacketsMeter = metrics.NewMeter("eth/req/state/in/packets")
miscInTrafficMeter = metrics.NewMeter("eth/misc/in/traffic") reqStateInTrafficMeter = metrics.NewMeter("eth/req/state/in/traffic")
miscOutPacketsMeter = metrics.NewMeter("eth/misc/out/packets") reqStateOutPacketsMeter = metrics.NewMeter("eth/req/state/out/packets")
miscOutTrafficMeter = metrics.NewMeter("eth/misc/out/traffic") 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 // meteredMsgReadWriter is a wrapper around a p2p.MsgReadWriter, capable of
// accumulating the above defined metrics based on the data stream contents. // accumulating the above defined metrics based on the data stream contents.
type meteredMsgReadWriter struct { 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 // newMeteredMsgWriter wraps a p2p MsgReadWriter with metering support. If the
@ -68,7 +77,13 @@ func newMeteredMsgWriter(rw p2p.MsgReadWriter) p2p.MsgReadWriter {
if !metrics.Enabled { if !metrics.Enabled {
return rw 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) { func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) {
@ -79,20 +94,27 @@ func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) {
} }
// Account for the data traffic // Account for the data traffic
packets, traffic := miscInPacketsMeter, miscInTrafficMeter packets, traffic := miscInPacketsMeter, miscInTrafficMeter
switch msg.Code { switch {
case BlockHashesMsg: case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlockHashesMsg:
packets, traffic = reqHashInPacketsMeter, reqHashInTrafficMeter packets, traffic = reqHashInPacketsMeter, reqHashInTrafficMeter
case BlocksMsg: case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlocksMsg:
packets, traffic = reqBlockInPacketsMeter, reqBlockInTrafficMeter packets, traffic = reqBlockInPacketsMeter, reqBlockInTrafficMeter
case BlockHeadersMsg:
packets, traffic = reqHeaderInPacketsMeter, reqHeaderInTrafficMeter case rw.version == eth62 && msg.Code == BlockHeadersMsg:
case NodeDataMsg: 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 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 packets, traffic = propHashInPacketsMeter, propHashInTrafficMeter
case NewBlockMsg: case msg.Code == NewBlockMsg:
packets, traffic = propBlockInPacketsMeter, propBlockInTrafficMeter packets, traffic = propBlockInPacketsMeter, propBlockInTrafficMeter
case TxMsg: case msg.Code == TxMsg:
packets, traffic = propTxnInPacketsMeter, propTxnInTrafficMeter packets, traffic = propTxnInPacketsMeter, propTxnInTrafficMeter
} }
packets.Mark(1) packets.Mark(1)
@ -104,20 +126,27 @@ func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) {
func (rw *meteredMsgReadWriter) WriteMsg(msg p2p.Msg) error { func (rw *meteredMsgReadWriter) WriteMsg(msg p2p.Msg) error {
// Account for the data traffic // Account for the data traffic
packets, traffic := miscOutPacketsMeter, miscOutTrafficMeter packets, traffic := miscOutPacketsMeter, miscOutTrafficMeter
switch msg.Code { switch {
case BlockHashesMsg: case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlockHashesMsg:
packets, traffic = reqHashOutPacketsMeter, reqHashOutTrafficMeter packets, traffic = reqHashOutPacketsMeter, reqHashOutTrafficMeter
case BlocksMsg: case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlocksMsg:
packets, traffic = reqBlockOutPacketsMeter, reqBlockOutTrafficMeter packets, traffic = reqBlockOutPacketsMeter, reqBlockOutTrafficMeter
case BlockHeadersMsg:
case rw.version == eth62 && msg.Code == BlockHeadersMsg:
packets, traffic = reqHeaderOutPacketsMeter, reqHeaderOutTrafficMeter 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 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 packets, traffic = propHashOutPacketsMeter, propHashOutTrafficMeter
case NewBlockMsg: case msg.Code == NewBlockMsg:
packets, traffic = propBlockOutPacketsMeter, propBlockOutTrafficMeter packets, traffic = propBlockOutPacketsMeter, propBlockOutTrafficMeter
case TxMsg: case msg.Code == TxMsg:
packets, traffic = propTxnOutPacketsMeter, propTxnOutTrafficMeter packets, traffic = propTxnOutPacketsMeter, propTxnOutTrafficMeter
} }
packets.Mark(1) packets.Mark(1)

44
eth/protocol.go
View file

@ -23,11 +23,19 @@ import (
"github.com/ethereum/go-ethereum/core/types" "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). // 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. // Number of implemented message corresponding to different protocol versions.
var ProtocolLengths = []uint64{13, 9, 8} var ProtocolLengths = []uint64{9, 8}
const ( const (
NetworkId = 1 NetworkId = 1
@ -37,23 +45,29 @@ const (
// eth protocol message codes // eth protocol message codes
const ( const (
// Protocol messages belonging to eth/60 // Protocol messages belonging to eth/60
StatusMsg = iota StatusMsg = 0x00
NewBlockHashesMsg NewBlockHashesMsg = 0x01
TxMsg TxMsg = 0x02
GetBlockHashesMsg GetBlockHashesMsg = 0x03
BlockHashesMsg BlockHashesMsg = 0x04
GetBlocksMsg GetBlocksMsg = 0x05
BlocksMsg BlocksMsg = 0x06
NewBlockMsg NewBlockMsg = 0x07
// Protocol messages belonging to eth/61 // Protocol messages belonging to eth/61
GetBlockHashesFromNumberMsg GetBlockHashesFromNumberMsg = 0x08
// Protocol messages belonging to eth/62 // Protocol messages belonging to eth/62
GetBlockHeadersMsg GetBlockHeadersMsg = 0x03
BlockHeadersMsg BlockHeadersMsg = 0x04
GetNodeDataMsg GetBlockBodiesMsg = 0x05
NodeDataMsg BlockBodiesMsg = 0x06
// Protocol messages belonging to eth/63
GetNodeDataMsg = 0x0d
NodeDataMsg = 0x0e
GetReceiptsMsg = 0x0f
ReceiptsMsg = 0x10
) )
type errCode int type errCode int

34
eth/protocol_test.go
View file

@ -36,7 +36,13 @@ func init() {
var testAccount = crypto.NewKey(rand.Reader) 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) pm := newTestProtocolManager(0, nil, nil)
td, currentBlock, genesis := pm.chainman.Status() td, currentBlock, genesis := pm.chainman.Status()
defer pm.Stop() defer pm.Stop()
@ -52,20 +58,20 @@ func TestStatusMsgErrors(t *testing.T) {
}, },
{ {
code: StatusMsg, data: statusData{10, NetworkId, td, currentBlock, genesis}, 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)"), 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), wantError: errResp(ErrGenesisBlockMismatch, "0300000000000000000000000000000000000000000000000000000000000000 (!= %x)", genesis),
}, },
} }
for i, test := range tests { 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 send call might hang until reset because
// the protocol might not read the payload. // the protocol might not read the payload.
go p2p.Send(p.app, test.code, test.data) 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. // 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) txAdded := make(chan []*types.Transaction)
pm := newTestProtocolManager(0, nil, txAdded) pm := newTestProtocolManager(0, nil, txAdded)
p, _ := newTestPeer("peer", pm, true) p, _ := newTestPeer("peer", protocol, pm, true)
defer pm.Stop() defer pm.Stop()
defer p.close() defer p.close()
@ -109,7 +120,12 @@ func TestRecvTransactions(t *testing.T) {
} }
// This test checks that pending transactions are sent. // 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) pm := newTestProtocolManager(0, nil, nil)
defer pm.Stop() defer pm.Stop()
@ -156,7 +172,7 @@ func TestSendTransactions(t *testing.T) {
} }
} }
for i := 0; i < 3; i++ { 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) wg.Add(1)
go checktxs(p) go checktxs(p)
} }