From 0f5c6e63ad1a630e7d39002f2719c1d4596db485 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Mon, 10 Oct 2016 09:35:49 +0200 Subject: [PATCH] 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 | 31 +++++++++++++--- core/types/transaction.go | 6 +-- core/types/transaction_test.go | 2 +- eth/api_backend.go | 4 +- eth/handler.go | 33 +++++++++-------- eth/helper_test.go | 13 ++----- eth/protocol.go | 6 +-- eth/sync.go | 68 ++++++++++++++++++++++------------ miner/worker.go | 4 +- 9 files changed, 103 insertions(+), 64 deletions(-) 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)