core/txpool: use slots to check queued, pending transactions limit

Since commit 8bd37a1d91 (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.
This commit is contained in:
Bui Quang Minh 2023-05-31 10:05:45 +07:00
parent e85fb69328
commit eaaa77715d
No known key found for this signature in database
GPG key ID: 9591AA13D0891B99
2 changed files with 67 additions and 53 deletions

View file

@ -1489,25 +1489,48 @@ func (pool *LegacyPool) promoteExecutables(accounts []common.Address) []*types.T
return promoted 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 // 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 // pending limit. The algorithm tries to reduce transaction counts by an approximately
// equal number for all for accounts with many pending transactions. // equal number for all for accounts with many pending transactions.
func (pool *LegacyPool) truncatePending() { func (pool *LegacyPool) truncatePending() {
pending := uint64(0) pending := uint64(0)
droppedTxs := 0
for _, list := range pool.pending { for _, list := range pool.pending {
pending += uint64(list.Len()) pending += uint64(list.TotalSlots())
} }
if pending <= pool.config.GlobalSlots { if pending <= pool.config.GlobalSlots {
return return
} }
pendingBeforeCap := pending
// Assemble a spam order to penalize large transactors first // Assemble a spam order to penalize large transactors first
spammers := prque.New[int64, common.Address](nil) spammers := prque.New[int64, common.Address](nil)
for addr, list := range pool.pending { for addr, list := range pool.pending {
// Only evict transactions from high rollers // Only evict transactions from high rollers
if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots { if !pool.locals.contains(addr) && uint64(list.TotalSlots()) > pool.config.AccountSlots {
spammers.Push(addr, int64(list.Len())) spammers.Push(addr, int64(list.TotalSlots()))
} }
} }
// Gradually drop transactions from offenders // Gradually drop transactions from offenders
@ -1520,29 +1543,23 @@ func (pool *LegacyPool) truncatePending() {
// Equalize balances until all the same or below threshold // Equalize balances until all the same or below threshold
if len(offenders) > 1 { if len(offenders) > 1 {
// Calculate the equalization threshold for all current offenders // 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 // 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++ { for i := 0; i < len(offenders)-1; i++ {
list := pool.pending[offenders[i]] if pool.pending[offenders[i]].TotalSlots() <= threshold {
lowerThanThreshold++
caps := list.Cap(list.Len() - 1) continue
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)
} }
pool.priced.Removed(len(caps))
pendingGauge.Dec(int64(len(caps))) pending -= pool.removeLastPendingTransaction(offenders[i])
if pool.locals.contains(offenders[i]) { droppedTxs++
localGauge.Dec(int64(len(caps))) }
}
pending-- 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 still above threshold, reduce to limit or min allowance
if pending > pool.config.GlobalSlots && len(offenders) > 0 { 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 { for _, addr := range offenders {
list := pool.pending[addr] if uint64(pool.pending[addr].TotalSlots()) <= pool.config.AccountSlots {
lowerThanThreshold++
caps := list.Cap(list.Len() - 1) continue
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)
} }
pool.priced.Removed(len(caps)) pending -= pool.removeLastPendingTransaction(addr)
pendingGauge.Dec(int64(len(caps))) droppedTxs++
if pool.locals.contains(addr) { }
localGauge.Dec(int64(len(caps)))
} if lowerThanThreshold == len(offenders) {
pending-- 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. // truncateQueue drops the oldest transactions in the queue if the pool is above the global queue limit.
func (pool *LegacyPool) truncateQueue() { func (pool *LegacyPool) truncateQueue() {
queued := uint64(0) queued := uint64(0)
for _, list := range pool.queue { for _, list := range pool.queue {
queued += uint64(list.Len()) queued += uint64(list.TotalSlots())
} }
if queued <= pool.config.GlobalQueue { if queued <= pool.config.GlobalQueue {
return return
@ -1596,26 +1606,26 @@ func (pool *LegacyPool) truncateQueue() {
sort.Sort(sort.Reverse(addresses)) sort.Sort(sort.Reverse(addresses))
// Drop transactions until the total is below the limit or only locals remain // 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] addr := addresses[len(addresses)-1]
list := pool.queue[addr.address] list := pool.queue[addr.address]
addresses = addresses[:len(addresses)-1] addresses = addresses[:len(addresses)-1]
// Drop all transactions if they are less than the overflow // 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() { for _, tx := range list.Flatten() {
pool.removeTx(tx.Hash(), true, true) pool.removeTx(tx.Hash(), true, true)
} }
drop -= size drop -= list.TotalSlots()
queuedRateLimitMeter.Mark(int64(size)) queuedRateLimitMeter.Mark(int64(list.Len()))
continue continue
} }
// Otherwise drop only last few transactions // Otherwise drop only last few transactions
txs := list.Flatten() txs := list.Flatten()
for i := len(txs) - 1; i >= 0 && drop > 0; i-- { for i := len(txs) - 1; i >= 0 && drop > 0; i-- {
pool.removeTx(txs[i].Hash(), true, true) pool.removeTx(txs[i].Hash(), true, true)
drop-- drop -= numSlots(txs[i])
queuedRateLimitMeter.Mark(1) queuedRateLimitMeter.Mark(1)
} }
} }

View file

@ -157,22 +157,26 @@ func (m *sortedMap) filter(filter func(*types.Transaction) bool) types.Transacti
return removed 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. // exceeding that limit.
func (m *sortedMap) Cap(threshold int) types.Transactions { func (m *sortedMap) Cap(threshold int) types.Transactions {
// Short circuit if the number of items is under the limit // Short circuit if the number of items is under the limit
if len(m.items) <= threshold { if m.totalslots <= threshold {
return nil return nil
} }
// Otherwise gather and drop the highest nonce'd transactions // Otherwise gather and drop the highest nonce'd transactions
var drops types.Transactions var (
drops types.Transactions
size int
)
sort.Sort(*m.index) 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]]) 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]) delete(m.items, (*m.index)[size-1])
} }
*m.index = (*m.index)[:threshold] *m.index = (*m.index)[:size]
heap.Init(m.index) heap.Init(m.index)
// If we had a cache, shift the back // If we had a cache, shift the back