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/handler.go | 29 +++++++++------- eth/peer.go | 95 +++++++++++++++++++++++++++------------------------- eth/protocol_test.go | 6 ++-- eth/sync.go | 10 +----- 4 files changed, 69 insertions(+), 71 deletions(-) (limited to 'eth') diff --git a/eth/handler.go b/eth/handler.go index d25118337..d0456446d 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -158,9 +158,7 @@ func (pm *ProtocolManager) Stop() { } func (pm *ProtocolManager) newPeer(pv, nv int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { - td, current, genesis := pm.chainman.Status() - - return newPeer(pv, nv, genesis, current, td, p, rw) + return newPeer(pv, nv, p, rw) } // handle is the callback invoked to manage the life cycle of an eth peer. When @@ -169,7 +167,8 @@ func (pm *ProtocolManager) handle(p *peer) error { glog.V(logger.Debug).Infof("%v: peer connected", p) // Execute the Ethereum handshake - if err := p.handleStatus(); err != nil { + td, head, genesis := pm.chainman.Status() + if err := p.Handshake(td, head, genesis); err != nil { glog.V(logger.Debug).Infof("%v: handshake failed: %v", p, err) return err } @@ -182,7 +181,7 @@ func (pm *ProtocolManager) handle(p *peer) error { defer pm.removePeer(p.id) // Register the peer in the downloader. If the downloader considers it banned, we disconnect - if err := pm.downloader.RegisterPeer(p.id, p.Head(), p.requestHashes, p.requestBlocks); err != nil { + if err := pm.downloader.RegisterPeer(p.id, p.Head(), p.RequestHashes, p.RequestBlocks); err != nil { return err } // Propagate existing transactions. new transactions appearing @@ -225,9 +224,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } propTxnInPacketsMeter.Mark(1) for i, tx := range txs { + // Validate and mark the remote transaction if tx == nil { return errResp(ErrDecode, "transaction %d is nil", i) } + p.MarkTransaction(tx.Hash()) + + // Log it's arrival for later analysis propTxnInTrafficMeter.Mark(tx.Size().Int64()) jsonlogger.LogJson(&logger.EthTxReceived{ TxHash: tx.Hash().Hex(), @@ -255,7 +258,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } // returns either requested hashes or nothing (i.e. not found) - return p.sendBlockHashes(hashes) + return p.SendBlockHashes(hashes) case BlockHashesMsg: // A batch of hashes arrived to one of our previous requests @@ -314,7 +317,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { glog.Infof("%v: no blocks found for requested hashes %s", p, list) } - return p.sendBlocks(blocks) + return p.SendBlocks(blocks) case BlocksMsg: // Decode the arrived block message @@ -349,7 +352,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { // Mark the hashes as present at the remote node for _, hash := range hashes { - p.blockHashes.Add(hash) + p.MarkBlock(hash) p.SetHead(hash) } // Schedule all the unknown hashes for retrieval @@ -360,7 +363,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } for _, hash := range unknown { - pm.fetcher.Notify(p.id, hash, time.Now(), p.requestBlocks) + pm.fetcher.Notify(p.id, hash, time.Now(), p.RequestBlocks) } case NewBlockMsg: @@ -387,7 +390,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { RemoteId: p.ID().String(), }) // Mark the peer as owning the block and schedule it for import - p.blockHashes.Add(request.Block.Hash()) + p.MarkBlock(request.Block.Hash()) p.SetHead(request.Block.Hash()) pm.fetcher.Enqueue(p.id, request.Block) @@ -412,14 +415,14 @@ func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) { if propagate { transfer := peers[:int(math.Sqrt(float64(len(peers))))] for _, peer := range transfer { - peer.sendNewBlock(block) + peer.SendNewBlock(block) } glog.V(logger.Detail).Infof("propagated block %x to %d peers in %v", hash[:4], len(transfer), time.Since(block.ReceivedAt)) } // Otherwise if the block is indeed in out own chain, announce it if pm.chainman.HasBlock(hash) { for _, peer := range peers { - peer.sendNewBlockHashes([]common.Hash{hash}) + peer.SendNewBlockHashes([]common.Hash{hash}) } glog.V(logger.Detail).Infof("announced block %x to %d peers in %v", hash[:4], len(peers), time.Since(block.ReceivedAt)) } @@ -432,7 +435,7 @@ func (pm *ProtocolManager) BroadcastTx(hash common.Hash, tx *types.Transaction) peers := pm.peers.PeersWithoutTx(hash) //FIXME include this again: peers = peers[:int(math.Sqrt(float64(len(peers))))] for _, peer := range peers { - peer.sendTransactions(types.Transactions{tx}) + peer.SendTransactions(types.Transactions{tx}) } glog.V(logger.Detail).Infoln("broadcast tx to", len(peers), "peers") } 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) } } diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 60fa35443..ffd4ca19f 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -43,11 +43,11 @@ func TestStatusMsgErrors(t *testing.T) { wantError: errResp(ErrProtocolVersionMismatch, "10 (!= 0)"), }, { - code: StatusMsg, data: statusMsgData{ProtocolVersion, 999, td, currentBlock, genesis}, + code: StatusMsg, data: statusMsgData{uint32(ProtocolVersions[0]), 999, td, currentBlock, genesis}, wantError: errResp(ErrNetworkIdMismatch, "999 (!= 0)"), }, { - code: StatusMsg, data: statusMsgData{ProtocolVersion, NetworkId, td, currentBlock, common.Hash{3}}, + code: StatusMsg, data: statusMsgData{uint32(ProtocolVersions[0]), NetworkId, td, currentBlock, common.Hash{3}}, wantError: errResp(ErrGenesisBlockMismatch, "0300000000000000000000000000000000000000000000000000000000000000 (!= %x)", genesis), }, } @@ -167,7 +167,7 @@ func newProtocolManagerForTesting(txAdded chan<- []*types.Transaction) *Protocol db, _ = ethdb.NewMemDatabase() chain, _ = core.NewChainManager(core.GenesisBlock(0, db), db, db, core.FakePow{}, em) txpool = &fakeTxPool{added: txAdded} - pm = NewProtocolManager(ProtocolVersion, 0, em, txpool, core.FakePow{}, chain) + pm = NewProtocolManager(0, em, txpool, core.FakePow{}, chain) ) pm.Start() return pm diff --git a/eth/sync.go b/eth/sync.go index 82abb725f..47fd7363e 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -20,14 +20,6 @@ const ( txsyncPackSize = 100 * 1024 ) -// blockAnnounce is the hash notification of the availability of a new block in -// the network. -type blockAnnounce struct { - hash common.Hash - peer *peer - time time.Time -} - type txsync struct { p *peer txs []*types.Transaction @@ -75,7 +67,7 @@ func (pm *ProtocolManager) txsyncLoop() { // Send the pack in the background. glog.V(logger.Detail).Infof("%v: sending %d transactions (%v)", s.p.Peer, len(pack.txs), size) sending = true - go func() { done <- pack.p.sendTransactions(pack.txs) }() + go func() { done <- pack.p.SendTransactions(pack.txs) }() } // pick chooses the next pending sync. -- cgit v1.2.3