diff --git a/core/tx_pool.go b/core/tx_pool.go index 10a110e0bd..8e9663e144 100644 --- a/core/tx_pool.go +++ b/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() } diff --git a/core/types/transaction.go b/core/types/transaction.go index f0512ae7e7..86c843c6e4 100644 --- a/core/types/transaction.go +++ b/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 - heads TxByPrice // Next transaction for each unique account (price heap) + txs map[common.Address][]*Transaction // Per account nonce-sorted list of transactions + 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 { diff --git a/core/types/transaction_test.go b/core/types/transaction_test.go index 8b0b02c3ea..91582375d9 100644 --- a/core/types/transaction_test.go +++ b/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++ { diff --git a/eth/api_backend.go b/eth/api_backend.go index 42b84bf9ba..c83f0c1b41 100644 --- a/eth/api_backend.go +++ b/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 } diff --git a/eth/handler.go b/eth/handler.go index d72185dd3d..f2d6f3d1b4 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -80,10 +80,11 @@ type ProtocolManager struct { minedBlockSub event.Subscription // channels for fetcher, syncer, txsyncLoop - newPeerCh chan *peer - txsyncCh chan *txsync - quitSync chan struct{} - noMorePeers chan struct{} + newPeerCh chan *peer + txsyncInit chan *txsync + txsyncRemove chan *txsync + 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, - eventMux: mux, - txpool: txpool, - blockchain: blockchain, - chaindb: chaindb, - chainconfig: config, - peers: newPeerSet(), - newPeerCh: make(chan *peer), - noMorePeers: make(chan struct{}), - txsyncCh: make(chan *txsync), - quitSync: make(chan struct{}), + networkId: networkId, + eventMux: mux, + txpool: txpool, + blockchain: blockchain, + chaindb: chaindb, + chainconfig: config, + peers: newPeerSet(), + newPeerCh: make(chan *peer), + noMorePeers: make(chan struct{}), + txsyncInit: make(chan *txsync), + 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) diff --git a/eth/helper_test.go b/eth/helper_test.go index 732fe89ee7..a18db4de47 100644 --- a/eth/helper_test.go +++ b/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. diff --git a/eth/protocol.go b/eth/protocol.go index 3f65c204bc..31d3a4449c 100644 --- a/eth/protocol.go +++ b/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. diff --git a/eth/sync.go b/eth/sync.go index 6584bb1e2a..c351715fee 100644 --- a/eth/sync.go +++ b/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) diff --git a/miner/worker.go b/miner/worker.go index e5348cef42..f7e0546d3b 100644 --- a/miner/worker.go +++ b/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)