eth: optimise initial transaction sync

When a connection starts up, all current transactions are sent in both
directions. We can mimise the amount of transactions sent by tracking
which txs where sent by the remote side and excluding those from the
ongoing sync. Transactions to send are picked randomly, ensuring that
most transactions are only sent once.

core/tx_pool.go

@@ -208,10 +208,9 @@ func (pool *TxPool) Content() (map[common.Address]types.Transactions, map[common
 	return pending, queued
 }
 
-// Pending retrieves all currently processable transactions, groupped by origin
-// account and sorted by nonce. The returned transaction set is a copy and can be
-// freely modified by calling code.
-func (pool *TxPool) Pending() map[common.Address]types.Transactions {
+// Pending retrieves all currently processable transactions The returned map is a copy and
+// can be modified by calling code.
+func (pool *TxPool) PendingTransactions() map[common.Hash]*types.Transaction {
 	pool.mu.Lock()
 	defer pool.mu.Unlock()
 
@@ -221,7 +220,29 @@ func (pool *TxPool) Pending() map[common.Address]types.Transactions {
 	// invalidate any txs
 	pool.demoteUnexecutables()
 
-	pending := make(map[common.Address]types.Transactions)
+	pending := make(map[common.Hash]*types.Transaction)
+	for _, list := range pool.pending {
+		for _, tx := range list.txs.items {
+			pending[tx.Hash()] = tx
+		}
+	}
+	return pending
+}
+
+// PendingTransactionsByAccount retrieves all currently processable transactions, grouped
+// by origin account and sorted by nonce. The returned transaction set is a copy and can
+// be modified by the caller.
+func (pool *TxPool) PendingTransactionsByAccount() map[common.Address][]*types.Transaction {
+	pool.mu.Lock()
+	defer pool.mu.Unlock()
+
+	// check queue first
+	pool.promoteExecutables()
+
+	// invalidate any txs
+	pool.demoteUnexecutables()
+
+	pending := make(map[common.Address][]*types.Transaction)
 	for addr, list := range pool.pending {
 		pending[addr] = list.Flatten()
 	}

core/types/transaction.go

@@ -448,7 +448,7 @@ func (s *TxByPrice) Pop() interface{} {
 // transactions in a profit-maximising sorted order, while supporting removing
 // entire batches of transactions for non-executable accounts.
 type TransactionsByPriceAndNonce struct {
-	txs   map[common.Address]Transactions // Per account nonce-sorted list of transactions
+	txs   map[common.Address][]*Transaction // Per account nonce-sorted list of transactions
 	heads TxByPrice                         // Next transaction for each unique account (price heap)
 }
 
@@ -457,7 +457,7 @@ type TransactionsByPriceAndNonce struct {
 //
 // Note, the input map is reowned so the caller should not interact any more with
 // if after providng it to the constructor.
-func NewTransactionsByPriceAndNonce(txs map[common.Address]Transactions) *TransactionsByPriceAndNonce {
+func NewTransactionsByPriceAndNonce(txs map[common.Address][]*Transaction) *TransactionsByPriceAndNonce {
 	// Initialize a price based heap with the head transactions
 	heads := make(TxByPrice, 0, len(txs))
 	for acc, accTxs := range txs {

core/types/transaction_test.go

@@ -128,7 +128,7 @@ func TestTransactionPriceNonceSort(t *testing.T) {
 		keys[i], _ = crypto.GenerateKey()
 	}
 	// Generate a batch of transactions with overlapping values, but shifted nonces
-	groups := map[common.Address]Transactions{}
+	groups := make(map[common.Address][]*Transaction)
 	for start, key := range keys {
 		addr := crypto.PubkeyToAddress(key.PublicKey)
 		for i := 0; i < 25; i++ {

eth/api_backend.go

@@ -126,8 +126,8 @@ func (b *EthApiBackend) GetPoolTransactions() types.Transactions {
 	defer b.eth.txMu.Unlock()
 
 	var txs types.Transactions
-	for _, batch := range b.eth.txPool.Pending() {
-		txs = append(txs, batch...)
+	for _, tx := range b.eth.txPool.PendingTransactions() {
+		txs = append(txs, tx)
 	}
 	return txs
 }

eth/handler.go

@@ -81,7 +81,8 @@ type ProtocolManager struct {
 
 	// channels for fetcher, syncer, txsyncLoop
 	newPeerCh    chan *peer
-	txsyncCh    chan *txsync
+	txsyncInit   chan *txsync
+	txsyncRemove chan *txsync
 	quitSync     chan struct{}
 	noMorePeers  chan struct{}
 
@@ -106,7 +107,8 @@ func NewProtocolManager(config *core.ChainConfig, fastSync bool, networkId int,
 		peers:        newPeerSet(),
 		newPeerCh:    make(chan *peer),
 		noMorePeers:  make(chan struct{}),
-		txsyncCh:    make(chan *txsync),
+		txsyncInit:   make(chan *txsync),
+		txsyncRemove: make(chan *txsync),
 		quitSync:     make(chan struct{}),
 	}
 	// Figure out whether to allow fast sync or not
@@ -689,6 +691,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			p.MarkTransaction(tx.Hash())
 		}
 		pm.txpool.AddBatch(txs)
+		pm.removeSyncTransactions(p, txs)
 
 	default:
 		return errResp(ErrInvalidMsgCode, "%v", msg.Code)

eth/helper_test.go

@@ -23,7 +23,6 @@ import (
 	"crypto/ecdsa"
 	"crypto/rand"
 	"math/big"
-	"sort"
 	"sync"
 	"testing"
 
@@ -103,19 +102,15 @@ func (p *testTxPool) AddBatch(txs []*types.Transaction) {
 }
 
 // Pending returns all the transactions known to the pool
-func (p *testTxPool) Pending() map[common.Address]types.Transactions {
+func (p *testTxPool) PendingTransactions() map[common.Hash]*types.Transaction {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
-	batches := make(map[common.Address]types.Transactions)
+	pending := make(map[common.Hash]*types.Transaction)
 	for _, tx := range p.pool {
-		from, _ := tx.From()
-		batches[from] = append(batches[from], tx)
+		pending[tx.Hash()] = tx
 	}
-	for _, batch := range batches {
-		sort.Sort(types.TxByNonce(batch))
-	}
-	return batches
+	return pending
 }
 
 // newTestTransaction create a new dummy transaction.

eth/protocol.go

@@ -100,9 +100,9 @@ type txPool interface {
 	// AddBatch should add the given transactions to the pool.
 	AddBatch([]*types.Transaction)
 
-	// Pending should return pending transactions.
-	// The slice should be modifiable by the caller.
-	Pending() map[common.Address]types.Transactions
+	// PendingTransactions should return all processable transactions.
+	// The map should be modifiable by the caller.
+	PendingTransactions() map[common.Hash]*types.Transaction
 }
 
 // statusData is the network packet for the status message.

eth/sync.go

@@ -40,20 +40,28 @@ const (
 
 type txsync struct {
 	p   *peer
-	txs []*types.Transaction
+	txs map[common.Hash]*types.Transaction
 }
 
 // syncTransactions starts sending all currently pending transactions to the given peer.
 func (pm *ProtocolManager) syncTransactions(p *peer) {
-	var txs types.Transactions
-	for _, batch := range pm.txpool.Pending() {
-		txs = append(txs, batch...)
-	}
+	txs := pm.txpool.PendingTransactions()
 	if len(txs) == 0 {
 		return
 	}
 	select {
-	case pm.txsyncCh <- &txsync{p, txs}:
+	case pm.txsyncInit <- &txsync{p, txs}:
+	case <-pm.quitSync:
+	}
+}
+
+func (pm *ProtocolManager) removeSyncTransactions(p *peer, txs []*types.Transaction) {
+	set := make(map[common.Hash]*types.Transaction, len(txs))
+	for _, tx := range txs {
+		set[tx.Hash()] = tx
+	}
+	select {
+	case pm.txsyncRemove <- &txsync{p, set}:
 	case <-pm.quitSync:
 	}
 }
@@ -65,30 +73,31 @@ func (pm *ProtocolManager) syncTransactions(p *peer) {
 func (pm *ProtocolManager) txsyncLoop() {
 	var (
 		pending = make(map[discover.NodeID]*txsync)
-		sending = false               // whether a send is active
-		pack    = new(txsync)         // the pack that is being sent
-		done    = make(chan error, 1) // result of the send
+		pack    []*types.Transaction  // the pack that is being sent
+		sending *peer                 // peer that pack is being sent to
+		done    = make(chan error, 1) // send error
 	)
 
 	// send starts a sending a pack of transactions from the sync.
 	send := func(s *txsync) {
 		// Fill pack with transactions up to the target size.
 		size := common.StorageSize(0)
-		pack.p = s.p
-		pack.txs = pack.txs[:0]
-		for i := 0; i < len(s.txs) && size < txsyncPackSize; i++ {
-			pack.txs = append(pack.txs, s.txs[i])
-			size += s.txs[i].Size()
+		pack = pack[:0]
+		for hash, tx := range s.txs {
+			if size > txsyncPackSize {
+				break
+			}
+			pack = append(pack, tx)
+			size += tx.Size()
+			delete(s.txs, hash)
 		}
-		// Remove the transactions that will be sent.
-		s.txs = s.txs[:copy(s.txs, s.txs[len(pack.txs):])]
 		if len(s.txs) == 0 {
 			delete(pending, s.p.ID())
 		}
 		// 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) }()
+		glog.V(logger.Detail).Infof("%v: sending %d transactions (%v)", s.p.Peer, len(pack), size)
+		sending = s.p
+		go func() { done <- s.p.SendTransactions(pack) }()
 	}
 
 	// pick chooses the next pending sync.
@@ -107,18 +116,29 @@ func (pm *ProtocolManager) txsyncLoop() {
 
 	for {
 		select {
-		case s := <-pm.txsyncCh:
+		case s := <-pm.txsyncInit:
 			pending[s.p.ID()] = s
-			if !sending {
+			if sending == nil {
 				send(s)
 			}
+		case s := <-pm.txsyncRemove:
+			// The peer has sent us a pack of transactions, remove
+			// them from the set txs that we have yet to send to them.
+			if set, ok := pending[s.p.ID()]; ok {
+				for _, tx := range s.txs {
+					delete(set.txs, tx.Hash())
+				}
+				if len(set.txs) == 0 {
+					delete(pending, s.p.ID())
+				}
+			}
 		case err := <-done:
-			sending = false
 			// Stop tracking peers that cause send failures.
 			if err != nil {
-				glog.V(logger.Debug).Infof("%v: tx send failed: %v", pack.p.Peer, err)
-				delete(pending, pack.p.ID())
+				glog.V(logger.Debug).Infof("%v: tx send failed: %v", sending.Peer, err)
+				delete(pending, sending.ID())
 			}
+			sending = nil
 			// Schedule the next send.
 			if s := pick(); s != nil {
 				send(s)

miner/worker.go

@@ -236,7 +236,7 @@ func (self *worker) update() {
 				self.currentMu.Lock()
 
 				acc, _ := ev.Tx.From()
-				txs := map[common.Address]types.Transactions{acc: types.Transactions{ev.Tx}}
+				txs := map[common.Address][]*types.Transaction{acc: types.Transactions{ev.Tx}}
 				txset := types.NewTransactionsByPriceAndNonce(txs)
 
 				self.current.commitTransactions(self.mux, txset, self.gasPrice, self.chain)
@@ -495,7 +495,7 @@ func (self *worker) commitNewWork() {
 	if self.config.DAOForkSupport && self.config.DAOForkBlock != nil && self.config.DAOForkBlock.Cmp(header.Number) == 0 {
 		core.ApplyDAOHardFork(work.state)
 	}
-	txs := types.NewTransactionsByPriceAndNonce(self.eth.TxPool().Pending())
+	txs := types.NewTransactionsByPriceAndNonce(self.eth.TxPool().PendingTransactionsByAccount())
 	work.commitTransactions(self.mux, txs, self.gasPrice, self.chain)
 
 	self.eth.TxPool().RemoveBatch(work.lowGasTxs)