eth: clean up peer struct a bit, fix double txn bcast

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")
 }

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)
 		}
 	}

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

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.