From 2c8ed76e01161d9fe4e69064404cd888b4e327f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 26 Jun 2015 20:42:27 +0300 Subject: eth: start cleaning up old protocol implementation, add metrics --- eth/peer.go | 64 ++++++++++++++++++++++++++++++++++++++++--------------------- 1 file changed, 42 insertions(+), 22 deletions(-) (limited to 'eth/peer.go') diff --git a/eth/peer.go b/eth/peer.go index c7045282b..b0002ce81 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -38,7 +38,8 @@ type peer struct { rw p2p.MsgReadWriter - protv, netid int + version int // Protocol version negotiated + network int // Network ID being on id string @@ -53,7 +54,7 @@ type peer struct { blockHashes *set.Set } -func newPeer(protv, netid int, genesis, head common.Hash, td *big.Int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { +func newPeer(version, network int, genesis, head common.Hash, td *big.Int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { id := p.ID() return &peer{ @@ -62,8 +63,8 @@ func newPeer(protv, netid int, genesis, head common.Hash, td *big.Int, p *p2p.Pe genesis: genesis, ourHash: head, ourTd: td, - protv: protv, - netid: netid, + version: version, + network: network, id: fmt.Sprintf("%x", id[:8]), txHashes: set.New(), blockHashes: set.New(), @@ -104,46 +105,58 @@ func (p *peer) SetTd(td *big.Int) { } // sendTransactions sends transactions to the peer and includes the hashes -// in it's tx hash set for future reference. The tx hash will allow the -// manager to check whether the peer has already received this particular -// transaction +// 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.txHashes.Add(tx.Hash()) } - return p2p.Send(p.rw, TxMsg, txs) } +// 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.blockHashes.Add(hash) } return p2p.Send(p.rw, NewBlockHashesMsg, hashes) } +// sendNewBlock propagates an entire block to a remote peer. func (p *peer) sendNewBlock(block *types.Block) error { - p.blockHashes.Add(block.Hash()) + propBlockOutPacketsMeter.Mark(1) + propBlockOutTrafficMeter.Mark(block.Size().Int64()) + p.blockHashes.Add(block.Hash()) return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, block.Td}) } -func (p *peer) sendTransaction(tx *types.Transaction) error { - p.txHashes.Add(tx.Hash()) - - return p2p.Send(p.rw, TxMsg, []*types.Transaction{tx}) -} - +// 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("[%s] fetching hashes (%d) %x...\n", p.id, downloader.MaxHashFetch, from[:4]) + glog.V(logger.Debug).Infof("Peer [%s] fetching hashes (%d) %x...\n", p.id, downloader.MaxHashFetch, from[:4]) return p2p.Send(p.rw, GetBlockHashesMsg, getBlockHashesMsgData{from, uint64(downloader.MaxHashFetch)}) } @@ -156,8 +169,8 @@ func (p *peer) handleStatus() error { errc := make(chan error, 1) go func() { errc <- p2p.Send(p.rw, StatusMsg, &statusMsgData{ - ProtocolVersion: uint32(p.protv), - NetworkId: uint32(p.netid), + ProtocolVersion: uint32(p.version), + NetworkId: uint32(p.network), TD: p.ourTd, CurrentBlock: p.ourHash, GenesisBlock: p.genesis, @@ -185,12 +198,12 @@ func (p *peer) handleStatus() error { return errResp(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, p.genesis) } - if int(status.NetworkId) != p.netid { - return errResp(ErrNetworkIdMismatch, "%d (!= %d)", status.NetworkId, p.netid) + if int(status.NetworkId) != p.network { + return errResp(ErrNetworkIdMismatch, "%d (!= %d)", status.NetworkId, p.network) } - if int(status.ProtocolVersion) != p.protv { - return errResp(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, p.protv) + if int(status.ProtocolVersion) != p.version { + return errResp(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, p.version) } // Set the total difficulty of the peer p.td = status.TD @@ -200,6 +213,13 @@ func (p *peer) handleStatus() error { return <-errc } +// String implements fmt.Stringer. +func (p *peer) String() string { + return fmt.Sprintf("Peer %s [%s]", p.id, + fmt.Sprintf("eth/%2d", p.version), + ) +} + // peerSet represents the collection of active peers currently participating in // the Ethereum sub-protocol. type peerSet struct { -- cgit v1.2.3 From 6fc85f1ec221f976399af071a75ad7264b0ee905 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 29 Jun 2015 12:44:00 +0300 Subject: eth: clean up peer struct a bit, fix double txn bcast --- eth/peer.go | 95 +++++++++++++++++++++++++++++++------------------------------ 1 file changed, 49 insertions(+), 46 deletions(-) (limited to 'eth/peer.go') diff --git a/eth/peer.go b/eth/peer.go index b0002ce81..9160ac718 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -47,27 +47,21 @@ type peer struct { td *big.Int lock sync.RWMutex - genesis, ourHash common.Hash - ourTd *big.Int - - txHashes *set.Set - blockHashes *set.Set + knownTxs *set.Set // Set of transaction hashes known to be known by this peer + knownBlocks *set.Set // Set of block hashes known to be known by this peer } -func newPeer(version, network int, genesis, head common.Hash, td *big.Int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { +func newPeer(version, network int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { id := p.ID() return &peer{ Peer: p, rw: rw, - genesis: genesis, - ourHash: head, - ourTd: td, version: version, network: network, id: fmt.Sprintf("%x", id[:8]), - txHashes: set.New(), - blockHashes: set.New(), + knownTxs: set.New(), + knownBlocks: set.New(), } } @@ -104,27 +98,39 @@ func (p *peer) SetTd(td *big.Int) { p.td.Set(td) } -// sendTransactions sends transactions to the peer and includes the hashes +// MarkBlock marks a block as known for the peer, ensuring that the block will +// never be propagated to this particular peer. +func (p *peer) MarkBlock(hash common.Hash) { + p.knownBlocks.Add(hash) +} + +// MarkTransaction marks a transaction as known for the peer, ensuring that it +// will never be propagated to this particular peer. +func (p *peer) MarkTransaction(hash common.Hash) { + p.knownTxs.Add(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 { +func (p *peer) SendTransactions(txs types.Transactions) error { propTxnOutPacketsMeter.Mark(1) for _, tx := range txs { propTxnOutTrafficMeter.Mark(tx.Size().Int64()) - p.txHashes.Add(tx.Hash()) + p.knownTxs.Add(tx.Hash()) } return p2p.Send(p.rw, TxMsg, txs) } -// sendBlockHashes sends a batch of known hashes to the remote peer. -func (p *peer) sendBlockHashes(hashes []common.Hash) 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 { +// 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()) @@ -132,52 +138,55 @@ func (p *peer) sendBlocks(blocks []*types.Block) error { return p2p.Send(p.rw, BlocksMsg, blocks) } -// sendNewBlockHashes announces the availability of a number of blocks through +// SendNewBlockHashes announces the availability of a number of blocks through // a hash notification. -func (p *peer) sendNewBlockHashes(hashes []common.Hash) error { +func (p *peer) SendNewBlockHashes(hashes []common.Hash) error { propHashOutPacketsMeter.Mark(1) propHashOutTrafficMeter.Mark(int64(32 * len(hashes))) for _, hash := range hashes { - p.blockHashes.Add(hash) + p.knownBlocks.Add(hash) } return p2p.Send(p.rw, NewBlockHashesMsg, hashes) } -// sendNewBlock propagates an entire block to a remote peer. -func (p *peer) sendNewBlock(block *types.Block) error { +// SendNewBlock propagates an entire block to a remote peer. +func (p *peer) SendNewBlock(block *types.Block) error { propBlockOutPacketsMeter.Mark(1) propBlockOutTrafficMeter.Mark(block.Size().Int64()) - p.blockHashes.Add(block.Hash()) + p.knownBlocks.Add(block.Hash()) return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, block.Td}) } -// requestHashes fetches a batch of hashes from a peer, starting at from, going +// 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 { +func (p *peer) RequestHashes(from common.Hash) error { glog.V(logger.Debug).Infof("Peer [%s] fetching hashes (%d) %x...\n", p.id, downloader.MaxHashFetch, from[:4]) return p2p.Send(p.rw, GetBlockHashesMsg, getBlockHashesMsgData{from, uint64(downloader.MaxHashFetch)}) } -func (p *peer) requestBlocks(hashes []common.Hash) error { +// 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)) return p2p.Send(p.rw, GetBlocksMsg, hashes) } -func (p *peer) handleStatus() error { +// 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 { + // Send out own handshake in a new thread errc := make(chan error, 1) go func() { errc <- p2p.Send(p.rw, StatusMsg, &statusMsgData{ ProtocolVersion: uint32(p.version), NetworkId: uint32(p.network), - TD: p.ourTd, - CurrentBlock: p.ourHash, - GenesisBlock: p.genesis, + TD: td, + CurrentBlock: head, + GenesisBlock: genesis, }) }() - - // read and handle remote status + // In the mean time retrieve the remote status message msg, err := p.rw.ReadMsg() if err != nil { return err @@ -188,28 +197,22 @@ func (p *peer) handleStatus() error { if msg.Size > ProtocolMaxMsgSize { return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) } - + // Decode the handshake and make sure everything matches var status statusMsgData if err := msg.Decode(&status); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - - if status.GenesisBlock != p.genesis { - return errResp(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, p.genesis) + if status.GenesisBlock != genesis { + return errResp(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesis) } - if int(status.NetworkId) != p.network { return errResp(ErrNetworkIdMismatch, "%d (!= %d)", status.NetworkId, p.network) } - if int(status.ProtocolVersion) != p.version { return errResp(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, p.version) } - // Set the total difficulty of the peer - p.td = status.TD - // set the best hash of the peer - p.head = status.CurrentBlock - + // Configure the remote peer, and sanity check out handshake too + p.td, p.head = status.TD, status.CurrentBlock return <-errc } @@ -284,7 +287,7 @@ func (ps *peerSet) PeersWithoutBlock(hash common.Hash) []*peer { list := make([]*peer, 0, len(ps.peers)) for _, p := range ps.peers { - if !p.blockHashes.Has(hash) { + if !p.knownBlocks.Has(hash) { list = append(list, p) } } @@ -299,7 +302,7 @@ func (ps *peerSet) PeersWithoutTx(hash common.Hash) []*peer { list := make([]*peer, 0, len(ps.peers)) for _, p := range ps.peers { - if !p.txHashes.Has(hash) { + if !p.knownTxs.Has(hash) { list = append(list, p) } } -- cgit v1.2.3 From 5db8f447d597e55718263ba5ed1770290e3e0893 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 29 Jun 2015 17:06:07 +0300 Subject: eth: fix #1319, put an upper limit on the known txns and blocks --- eth/peer.go | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) (limited to 'eth/peer.go') diff --git a/eth/peer.go b/eth/peer.go index 9160ac718..0120cd033 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -20,6 +20,11 @@ var ( errNotRegistered = errors.New("peer is not registered") ) +const ( + maxKnownTxs = 32768 // Maximum transactions hashes to keep in the known list (prevent DOS) + maxKnownBlocks = 1024 // Maximum block hashes to keep in the known list (prevent DOS) +) + type statusMsgData struct { ProtocolVersion uint32 NetworkId uint32 @@ -101,12 +106,26 @@ func (p *peer) SetTd(td *big.Int) { // MarkBlock marks a block as known for the peer, ensuring that the block will // never be propagated to this particular peer. func (p *peer) MarkBlock(hash common.Hash) { + // If we reached the memory allowance, drop a previously known block hash + if p.knownBlocks.Size() >= maxKnownBlocks { + p.knownBlocks.Each(func(item interface{}) bool { + p.knownBlocks.Remove(item) + return p.knownBlocks.Size() >= maxKnownBlocks + }) + } p.knownBlocks.Add(hash) } // MarkTransaction marks a transaction as known for the peer, ensuring that it // will never be propagated to this particular peer. func (p *peer) MarkTransaction(hash common.Hash) { + // If we reached the memory allowance, drop a previously known transaction hash + if p.knownTxs.Size() >= maxKnownTxs { + p.knownTxs.Each(func(item interface{}) bool { + p.knownTxs.Remove(item) + return p.knownTxs.Size() >= maxKnownTxs + }) + } p.knownTxs.Add(hash) } -- cgit v1.2.3 From aac2b6ae4c5cf6f78547159c47f9192babe3e6dd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 29 Jun 2015 17:32:14 +0300 Subject: eth: add the blocks from numbers protocol message --- eth/peer.go | 28 +++++++++++----------------- 1 file changed, 11 insertions(+), 17 deletions(-) (limited to 'eth/peer.go') diff --git a/eth/peer.go b/eth/peer.go index 0120cd033..a5d56249d 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -25,19 +25,6 @@ const ( maxKnownBlocks = 1024 // Maximum block hashes to keep in the known list (prevent DOS) ) -type statusMsgData struct { - ProtocolVersion uint32 - NetworkId uint32 - TD *big.Int - CurrentBlock common.Hash - GenesisBlock common.Hash -} - -type getBlockHashesMsgData struct { - Hash common.Hash - Amount uint64 -} - type peer struct { *p2p.Peer @@ -181,8 +168,15 @@ func (p *peer) SendNewBlock(block *types.Block) error { // 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) %x...\n", p.id, downloader.MaxHashFetch, from[:4]) - return p2p.Send(p.rw, GetBlockHashesMsg, getBlockHashesMsgData{from, uint64(downloader.MaxHashFetch)}) + glog.V(logger.Debug).Infof("Peer [%s] fetching hashes (%d) from %x...\n", p.id, 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. +func (p *peer) RequestHashesFromNumber(from uint64) error { + glog.V(logger.Debug).Infof("Peer [%s] fetching hashes (%d) from #%d...\n", p.id, downloader.MaxHashFetch, from) + return p2p.Send(p.rw, GetBlockHashesFromNumberMsg, getBlockHashesFromNumberData{from, uint64(downloader.MaxHashFetch)}) } // RequestBlocks fetches a batch of blocks corresponding to the specified hashes. @@ -197,7 +191,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, genesis common.Hash) err // Send out own handshake in a new thread errc := make(chan error, 1) go func() { - errc <- p2p.Send(p.rw, StatusMsg, &statusMsgData{ + errc <- p2p.Send(p.rw, StatusMsg, &statusData{ ProtocolVersion: uint32(p.version), NetworkId: uint32(p.network), TD: td, @@ -217,7 +211,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, genesis common.Hash) err return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) } // Decode the handshake and make sure everything matches - var status statusMsgData + var status statusData if err := msg.Decode(&status); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } -- cgit v1.2.3 From f43c07cb3ca0d96fd005aa7ce2ddd40876a06d11 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 30 Jun 2015 19:05:06 +0300 Subject: eth, eth/downloader: transition to eth 61 --- eth/peer.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'eth/peer.go') diff --git a/eth/peer.go b/eth/peer.go index a5d56249d..c8b8457b9 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -174,9 +174,9 @@ func (p *peer) RequestHashes(from common.Hash) error { // 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) error { - glog.V(logger.Debug).Infof("Peer [%s] fetching hashes (%d) from #%d...\n", p.id, downloader.MaxHashFetch, from) - return p2p.Send(p.rw, GetBlockHashesFromNumberMsg, getBlockHashesFromNumberData{from, uint64(downloader.MaxHashFetch)}) +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) + return p2p.Send(p.rw, GetBlockHashesFromNumberMsg, getBlockHashesFromNumberData{from, uint64(count)}) } // RequestBlocks fetches a batch of blocks corresponding to the specified hashes. -- cgit v1.2.3 From 1ae80aaf64c5acfde19fee5ee3bc4579db7ea76e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 1 Jul 2015 11:12:05 +0300 Subject: eth: fix #1371, double lock during block/txn known set limitation --- eth/peer.go | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) (limited to 'eth/peer.go') diff --git a/eth/peer.go b/eth/peer.go index c8b8457b9..088417aab 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -94,11 +94,8 @@ func (p *peer) SetTd(td *big.Int) { // never be propagated to this particular peer. func (p *peer) MarkBlock(hash common.Hash) { // If we reached the memory allowance, drop a previously known block hash - if p.knownBlocks.Size() >= maxKnownBlocks { - p.knownBlocks.Each(func(item interface{}) bool { - p.knownBlocks.Remove(item) - return p.knownBlocks.Size() >= maxKnownBlocks - }) + for p.knownBlocks.Size() >= maxKnownBlocks { + p.knownBlocks.Pop() } p.knownBlocks.Add(hash) } @@ -107,11 +104,8 @@ func (p *peer) MarkBlock(hash common.Hash) { // will never be propagated to this particular peer. func (p *peer) MarkTransaction(hash common.Hash) { // If we reached the memory allowance, drop a previously known transaction hash - if p.knownTxs.Size() >= maxKnownTxs { - p.knownTxs.Each(func(item interface{}) bool { - p.knownTxs.Remove(item) - return p.knownTxs.Size() >= maxKnownTxs - }) + for p.knownTxs.Size() >= maxKnownTxs { + p.knownTxs.Pop() } p.knownTxs.Add(hash) } -- cgit v1.2.3 From ea54283b304a1d308141d21e3ef75b7de0f4519d Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Tue, 7 Jul 2015 02:54:22 +0200 Subject: all: update license information --- eth/peer.go | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) (limited to 'eth/peer.go') diff --git a/eth/peer.go b/eth/peer.go index 088417aab..ccd5d3c6f 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -1,3 +1,19 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of go-ethereum. +// +// go-ethereum is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// go-ethereum is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with go-ethereum. If not, see . + package eth import ( -- cgit v1.2.3