Merge 0f5c6e63ad into 289b30715d

core/tx_pool.go

@@ -211,10 +211,9 @@ func (pool *TxPool) Content() (map[common.Address]types.Transactions, map[common
 	return pending, queued
 }
 
-// Pending retrieves all currently processable transactions, groupped by origin
+// Pending retrieves all currently processable transactions The returned map is a copy and
-// account and sorted by nonce. The returned transaction set is a copy and can be
+// can be modified by calling code.
-// freely modified by calling code.
+func (pool *TxPool) PendingTransactions() map[common.Hash]*types.Transaction {
-func (pool *TxPool) Pending() map[common.Address]types.Transactions {
 	pool.mu.Lock()
 	defer pool.mu.Unlock()
 
@@ -224,7 +223,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,8 +448,8 @@ 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)
+	heads TxByPrice                         // Next transaction for each unique account (price heap)
 }
 
 // NewTransactionsByPriceAndNonce creates a transaction set that can retrieve
@@ -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() {
+	for _, tx := range b.eth.txPool.PendingTransactions() {
-		txs = append(txs, batch...)
+		txs = append(txs, tx)
 	}
 	return txs
 }

eth/handler.go

@@ -80,10 +80,11 @@ type ProtocolManager struct {
 	minedBlockSub event.Subscription
 
 	// channels for fetcher, syncer, txsyncLoop
-	newPeerCh   chan *peer
+	newPeerCh    chan *peer
-	txsyncCh    chan *txsync
+	txsyncInit   chan *txsync
-	quitSync    chan struct{}
+	txsyncRemove chan *txsync
-	noMorePeers chan struct{}
+	quitSync     chan struct{}
+	noMorePeers  chan struct{}
 
 	// wait group is used for graceful shutdowns during downloading
 	// and processing
@@ -97,17 +98,18 @@ type ProtocolManager struct {
 func NewProtocolManager(config *core.ChainConfig, fastSync bool, networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, blockchain *core.BlockChain, chaindb ethdb.Database) (*ProtocolManager, error) {
 	// Create the protocol manager with the base fields
 	manager := &ProtocolManager{
-		networkId:   networkId,
+		networkId:    networkId,
-		eventMux:    mux,
+		eventMux:     mux,
-		txpool:      txpool,
+		txpool:       txpool,
-		blockchain:  blockchain,
+		blockchain:   blockchain,
-		chaindb:     chaindb,
+		chaindb:      chaindb,
-		chainconfig: config,
+		chainconfig:  config,
-		peers:       newPeerSet(),
+		peers:        newPeerSet(),
-		newPeerCh:   make(chan *peer),
+		newPeerCh:    make(chan *peer),
-		noMorePeers: make(chan struct{}),
+		noMorePeers:  make(chan struct{}),
-		txsyncCh:    make(chan *txsync),
+		txsyncInit:   make(chan *txsync),
-		quitSync:    make(chan struct{}),
+		txsyncRemove: make(chan *txsync),
+		quitSync:     make(chan struct{}),
 	}
 	// Figure out whether to allow fast sync or not
 	if fastSync && blockchain.CurrentBlock().NumberU64() > 0 {
@@ -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()
+		pending[tx.Hash()] = tx
-		batches[from] = append(batches[from], tx)
 	}
-	for _, batch := range batches {
+	return pending
-		sort.Sort(types.TxByNonce(batch))
-	}
-	return batches
 }
 
 // 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.
+	// PendingTransactions should return all processable transactions.
-	// The slice should be modifiable by the caller.
+	// The map should be modifiable by the caller.
-	Pending() map[common.Address]types.Transactions
+	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
+	txs := pm.txpool.PendingTransactions()
-	for _, batch := range pm.txpool.Pending() {
-		txs = append(txs, batch...)
-	}
 	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    []*types.Transaction  // the pack that is being sent
-		pack    = new(txsync)         // the pack that is being sent
+		sending *peer                 // peer that pack is being sent to
-		done    = make(chan error, 1) // result of the send
+		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 = pack[:0]
-		pack.txs = pack.txs[:0]
+		for hash, tx := range s.txs {
-		for i := 0; i < len(s.txs) && size < txsyncPackSize; i++ {
+			if size > txsyncPackSize {
-			pack.txs = append(pack.txs, s.txs[i])
+				break
-			size += s.txs[i].Size()
+			}
+			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)
+		glog.V(logger.Detail).Infof("%v: sending %d transactions (%v)", s.p.Peer, len(pack), size)
-		sending = true
+		sending = s.p
-		go func() { done <- pack.p.SendTransactions(pack.txs) }()
+		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)
+				glog.V(logger.Debug).Infof("%v: tx send failed: %v", sending.Peer, err)
-				delete(pending, pack.p.ID())
+				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)