From eaaa77715db15d44ef9423fa12b4751d33e894c3 Mon Sep 17 00:00:00 2001 From: Bui Quang Minh Date: Wed, 31 May 2023 10:05:45 +0700 Subject: [PATCH] core/txpool: use slots to check queued, pending transactions limit Since commit 8bd37a1d9191 (core: count tx size in slots, bump max size ot 4x32KB), txpool uses transaction slot for transaction accounting. However, when truncating queued and pending transactions, we still use number of transactions. This commit change all the places to consistently use transaction slot. --- core/txpool/legacypool/legacypool.go | 106 +++++++++++++++------------ core/txpool/legacypool/list.go | 14 ++-- 2 files changed, 67 insertions(+), 53 deletions(-) diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go index 99c436a193..129ffc1be8 100644 --- a/core/txpool/legacypool/legacypool.go +++ b/core/txpool/legacypool/legacypool.go @@ -1489,25 +1489,48 @@ func (pool *LegacyPool) promoteExecutables(accounts []common.Address) []*types.T return promoted } +// removeLastPendingTransaction removes the highest nonce transaction of the sender from +// the pending transactions, returns the number of transaction slot of that transaction +func (pool *LegacyPool) removeLastPendingTransaction(sender common.Address) uint64 { + list := pool.pending[sender] + + // Remove the transaction with highest nonce + lastTx := list.LastElement() + list.Remove(lastTx) + + // Drop the transaction from the global pools too + pool.all.Remove(lastTx.Hash()) + // Update the account nonce to the dropped transaction + pool.pendingNonces.setIfLower(sender, lastTx.Nonce()) + log.Trace("Removed fairness-exceeding pending transaction", "hash", lastTx.Hash()) + + pool.priced.Removed(1) + pendingGauge.Dec(1) + if pool.locals.contains(sender) { + localGauge.Dec(1) + } + return uint64(numSlots(lastTx)) +} + // truncatePending removes transactions from the pending queue if the pool is above the // pending limit. The algorithm tries to reduce transaction counts by an approximately // equal number for all for accounts with many pending transactions. func (pool *LegacyPool) truncatePending() { pending := uint64(0) + droppedTxs := 0 for _, list := range pool.pending { - pending += uint64(list.Len()) + pending += uint64(list.TotalSlots()) } if pending <= pool.config.GlobalSlots { return } - pendingBeforeCap := pending // Assemble a spam order to penalize large transactors first spammers := prque.New[int64, common.Address](nil) for addr, list := range pool.pending { // Only evict transactions from high rollers - if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots { - spammers.Push(addr, int64(list.Len())) + if !pool.locals.contains(addr) && uint64(list.TotalSlots()) > pool.config.AccountSlots { + spammers.Push(addr, int64(list.TotalSlots())) } } // Gradually drop transactions from offenders @@ -1520,29 +1543,23 @@ func (pool *LegacyPool) truncatePending() { // Equalize balances until all the same or below threshold if len(offenders) > 1 { // Calculate the equalization threshold for all current offenders - threshold := pool.pending[offender].Len() + threshold := pool.pending[offender].TotalSlots() // Iteratively reduce all offenders until below limit or threshold reached - for pending > pool.config.GlobalSlots && pool.pending[offenders[len(offenders)-2]].Len() > threshold { + for pending > pool.config.GlobalSlots { + lowerThanThreshold := 0 for i := 0; i < len(offenders)-1; i++ { - list := pool.pending[offenders[i]] - - caps := list.Cap(list.Len() - 1) - for _, tx := range caps { - // Drop the transaction from the global pools too - hash := tx.Hash() - pool.all.Remove(hash) - - // Update the account nonce to the dropped transaction - pool.pendingNonces.setIfLower(offenders[i], tx.Nonce()) - log.Trace("Removed fairness-exceeding pending transaction", "hash", hash) + if pool.pending[offenders[i]].TotalSlots() <= threshold { + lowerThanThreshold++ + continue } - pool.priced.Removed(len(caps)) - pendingGauge.Dec(int64(len(caps))) - if pool.locals.contains(offenders[i]) { - localGauge.Dec(int64(len(caps))) - } - pending-- + + pending -= pool.removeLastPendingTransaction(offenders[i]) + droppedTxs++ + } + + if lowerThanThreshold == len(offenders)-1 { + break } } } @@ -1550,37 +1567,30 @@ func (pool *LegacyPool) truncatePending() { // If still above threshold, reduce to limit or min allowance if pending > pool.config.GlobalSlots && len(offenders) > 0 { - for pending > pool.config.GlobalSlots && uint64(pool.pending[offenders[len(offenders)-1]].Len()) > pool.config.AccountSlots { + for pending > pool.config.GlobalSlots { + lowerThanThreshold := 0 for _, addr := range offenders { - list := pool.pending[addr] - - caps := list.Cap(list.Len() - 1) - for _, tx := range caps { - // Drop the transaction from the global pools too - hash := tx.Hash() - pool.all.Remove(hash) - - // Update the account nonce to the dropped transaction - pool.pendingNonces.setIfLower(addr, tx.Nonce()) - log.Trace("Removed fairness-exceeding pending transaction", "hash", hash) + if uint64(pool.pending[addr].TotalSlots()) <= pool.config.AccountSlots { + lowerThanThreshold++ + continue } - pool.priced.Removed(len(caps)) - pendingGauge.Dec(int64(len(caps))) - if pool.locals.contains(addr) { - localGauge.Dec(int64(len(caps))) - } - pending-- + pending -= pool.removeLastPendingTransaction(addr) + droppedTxs++ + } + + if lowerThanThreshold == len(offenders) { + break } } } - pendingRateLimitMeter.Mark(int64(pendingBeforeCap - pending)) + pendingRateLimitMeter.Mark(int64(droppedTxs)) } // truncateQueue drops the oldest transactions in the queue if the pool is above the global queue limit. func (pool *LegacyPool) truncateQueue() { queued := uint64(0) for _, list := range pool.queue { - queued += uint64(list.Len()) + queued += uint64(list.TotalSlots()) } if queued <= pool.config.GlobalQueue { return @@ -1596,26 +1606,26 @@ func (pool *LegacyPool) truncateQueue() { sort.Sort(sort.Reverse(addresses)) // Drop transactions until the total is below the limit or only locals remain - for drop := queued - pool.config.GlobalQueue; drop > 0 && len(addresses) > 0; { + for drop := int(queued - pool.config.GlobalQueue); drop > 0 && len(addresses) > 0; { addr := addresses[len(addresses)-1] list := pool.queue[addr.address] addresses = addresses[:len(addresses)-1] // Drop all transactions if they are less than the overflow - if size := uint64(list.Len()); size <= drop { + if list.TotalSlots() <= drop { for _, tx := range list.Flatten() { pool.removeTx(tx.Hash(), true, true) } - drop -= size - queuedRateLimitMeter.Mark(int64(size)) + drop -= list.TotalSlots() + queuedRateLimitMeter.Mark(int64(list.Len())) continue } // Otherwise drop only last few transactions txs := list.Flatten() for i := len(txs) - 1; i >= 0 && drop > 0; i-- { pool.removeTx(txs[i].Hash(), true, true) - drop-- + drop -= numSlots(txs[i]) queuedRateLimitMeter.Mark(1) } } diff --git a/core/txpool/legacypool/list.go b/core/txpool/legacypool/list.go index 39ec74624e..5c19d6fe8c 100644 --- a/core/txpool/legacypool/list.go +++ b/core/txpool/legacypool/list.go @@ -157,22 +157,26 @@ func (m *sortedMap) filter(filter func(*types.Transaction) bool) types.Transacti return removed } -// Cap places a hard limit on the number of items, returning all transactions +// Cap places a hard limit on the total slots of items, returning all transactions // exceeding that limit. func (m *sortedMap) Cap(threshold int) types.Transactions { // Short circuit if the number of items is under the limit - if len(m.items) <= threshold { + if m.totalslots <= threshold { return nil } // Otherwise gather and drop the highest nonce'd transactions - var drops types.Transactions + var ( + drops types.Transactions + size int + ) sort.Sort(*m.index) - for size := len(m.items); size > threshold; size-- { + for size = len(m.items); m.totalslots > threshold; size-- { drops = append(drops, m.items[(*m.index)[size-1]]) + m.totalslots -= numSlots(m.items[(*m.index)[size-1]]) delete(m.items, (*m.index)[size-1]) } - *m.index = (*m.index)[:threshold] + *m.index = (*m.index)[:size] heap.Init(m.index) // If we had a cache, shift the back