forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-18 02:40:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core: expose various counter metrics for grafana (#19692)

Browse source
This commit is contained in:
Daniel Liu 2024-05-10 12:04:16 +08:00
parent 65baaaaf80
commit ddbf5d2782
4 changed files with 130 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
core/lending_pool.go
View file

@ -669,7 +669,7 @@ func (pool *LendingPool) add(tx *types.LendingTransaction, local bool) (bool, er
// If the transaction fails basic validation, discard it // If the transaction fails basic validation, discard it
if err := pool.validateTx(tx, local); err != nil { if err := pool.validateTx(tx, local); err != nil {
log.Debug("Discarding invalid lending transaction", "hash", hash, "userAddress", tx.UserAddress, "status", tx.Status, "err", err) log.Debug("Discarding invalid lending transaction", "hash", hash, "userAddress", tx.UserAddress, "status", tx.Status, "err", err)
invalidTxCounter.Inc(1) invalidTxMeter.Mark(1)
return false, err return false, err
} }
from, _ := types.LendingSender(pool.signer, tx) // already validated from, _ := types.LendingSender(pool.signer, tx) // already validated
@ -683,12 +683,12 @@ func (pool *LendingPool) add(tx *types.LendingTransaction, local bool) (bool, er
if list := pool.pending[from]; list != nil && list.Overlaps(tx) { if list := pool.pending[from]; list != nil && list.Overlaps(tx) {
inserted, old := list.Add(tx) inserted, old := list.Add(tx)
if !inserted { if !inserted {
pendingDiscardCounter.Inc(1) pendingDiscardMeter.Mark(1)
return false, ErrPendingNonceTooLow return false, ErrPendingNonceTooLow
} }
if old != nil { if old != nil {
delete(pool.all, old.Hash()) delete(pool.all, old.Hash())
pendingReplaceCounter.Inc(1) pendingReplaceMeter.Mark(1)
} }
pool.all[tx.Hash()] = tx pool.all[tx.Hash()] = tx
pool.journalTx(from, tx) pool.journalTx(from, tx)
@ -724,13 +724,13 @@ func (pool *LendingPool) enqueueTx(hash common.Hash, tx *types.LendingTransactio
inserted, old := pool.queue[from].Add(tx) inserted, old := pool.queue[from].Add(tx)
if !inserted { if !inserted {
// An older transaction was better, discard this // An older transaction was better, discard this
queuedDiscardCounter.Inc(1) pendingDiscardMeter.Mark(1)
return false, ErrPendingNonceTooLow return false, ErrPendingNonceTooLow
} }
// Discard any previous transaction and mark this // Discard any previous transaction and mark this
if old != nil { if old != nil {
delete(pool.all, old.Hash()) delete(pool.all, old.Hash())
queuedReplaceCounter.Inc(1) queuedReplaceMeter.Mark(1)
} }
pool.all[hash] = tx pool.all[hash] = tx
return old != nil, nil return old != nil, nil
@ -762,13 +762,13 @@ func (pool *LendingPool) promoteTx(addr common.Address, hash common.Hash, tx *ty
if !inserted { if !inserted {
// An older transaction was better, discard this // An older transaction was better, discard this
delete(pool.all, hash) delete(pool.all, hash)
pendingDiscardCounter.Inc(1) pendingDiscardMeter.Mark(1)
return return
} }
// Otherwise discard any previous transaction and mark this // Otherwise discard any previous transaction and mark this
if old != nil { if old != nil {
delete(pool.all, old.Hash()) delete(pool.all, old.Hash())
pendingReplaceCounter.Inc(1) pendingReplaceMeter.Mark(1)
} }
// Failsafe to work around direct pending inserts (tests) // Failsafe to work around direct pending inserts (tests)
if pool.all[hash] == nil { if pool.all[hash] == nil {
@ -979,7 +979,7 @@ func (pool *LendingPool) promoteExecutables(accounts []common.Address) {
hash := tx.Hash() hash := tx.Hash()
delete(pool.all, hash) delete(pool.all, hash)
queuedRateLimitCounter.Inc(1) queuedRateLimitMeter.Mark(1)
log.Trace("Removed cap-exceeding queued transaction", "hash", hash) log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
} }
} }
@ -1055,7 +1055,7 @@ func (pool *LendingPool) promoteExecutables(accounts []common.Address) {
} }
} }
} }
pendingRateLimitCounter.Inc(int64(pendingBeforeCap - pending)) pendingRateLimitMeter.Mark(int64(pendingBeforeCap - pending))
} }
// If we've queued more transactions than the hard limit, drop oldest ones // If we've queued more transactions than the hard limit, drop oldest ones
queued := uint64(0) queued := uint64(0)
@ -1085,7 +1085,7 @@ func (pool *LendingPool) promoteExecutables(accounts []common.Address) {
pool.removeTx(tx.Hash()) pool.removeTx(tx.Hash())
} }
drop -= size drop -= size
queuedRateLimitCounter.Inc(int64(size)) queuedRateLimitMeter.Mark(int64(size))
continue continue
} }
// Otherwise drop only last few transactions // Otherwise drop only last few transactions
@ -1093,7 +1093,7 @@ func (pool *LendingPool) promoteExecutables(accounts []common.Address) {
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()) pool.removeTx(txs[i].Hash())
drop-- drop--
queuedRateLimitCounter.Inc(1) queuedRateLimitMeter.Mark(1)
} }
} }
} }

22
core/order_pool.go
View file

@ -578,7 +578,7 @@ func (pool *OrderPool) add(tx *types.OrderTransaction, local bool) (bool, error)
// If the transaction fails basic validation, discard it // If the transaction fails basic validation, discard it
if err := pool.validateTx(tx, local); err != nil { if err := pool.validateTx(tx, local); err != nil {
log.Debug("Discarding invalid order transaction", "hash", hash, "userAddress", tx.UserAddress().Hex(), "status", tx.Status, "err", err) log.Debug("Discarding invalid order transaction", "hash", hash, "userAddress", tx.UserAddress().Hex(), "status", tx.Status, "err", err)
invalidTxCounter.Inc(1) invalidTxMeter.Mark(1)
return false, err return false, err
} }
from, _ := types.OrderSender(pool.signer, tx) // already validated from, _ := types.OrderSender(pool.signer, tx) // already validated
@ -592,12 +592,12 @@ func (pool *OrderPool) add(tx *types.OrderTransaction, local bool) (bool, error)
if list := pool.pending[from]; list != nil && list.Overlaps(tx) { if list := pool.pending[from]; list != nil && list.Overlaps(tx) {
inserted, old := list.Add(tx) inserted, old := list.Add(tx)
if !inserted { if !inserted {
pendingDiscardCounter.Inc(1) pendingDiscardMeter.Mark(1)
return false, ErrPendingNonceTooLow return false, ErrPendingNonceTooLow
} }
if old != nil { if old != nil {
delete(pool.all, old.Hash()) delete(pool.all, old.Hash())
pendingReplaceCounter.Inc(1) pendingReplaceMeter.Mark(1)
} }
pool.all[tx.Hash()] = tx pool.all[tx.Hash()] = tx
pool.journalTx(from, tx) pool.journalTx(from, tx)
@ -635,13 +635,13 @@ func (pool *OrderPool) enqueueTx(hash common.Hash, tx *types.OrderTransaction) (
inserted, old := pool.queue[from].Add(tx) inserted, old := pool.queue[from].Add(tx)
if !inserted { if !inserted {
// An older transaction was better, discard this // An older transaction was better, discard this
queuedDiscardCounter.Inc(1) queuedDiscardMeter.Mark(1)
return false, ErrPendingNonceTooLow return false, ErrPendingNonceTooLow
} }
// Discard any previous transaction and mark this // Discard any previous transaction and mark this
if old != nil { if old != nil {
delete(pool.all, old.Hash()) delete(pool.all, old.Hash())
queuedReplaceCounter.Inc(1) queuedReplaceMeter.Mark(1)
} }
pool.all[hash] = tx pool.all[hash] = tx
return old != nil, nil return old != nil, nil
@ -674,13 +674,13 @@ func (pool *OrderPool) promoteTx(addr common.Address, hash common.Hash, tx *type
if !inserted { if !inserted {
// An older transaction was better, discard this // An older transaction was better, discard this
delete(pool.all, hash) delete(pool.all, hash)
pendingDiscardCounter.Inc(1) pendingDiscardMeter.Mark(1)
return return
} }
// Otherwise discard any previous transaction and mark this // Otherwise discard any previous transaction and mark this
if old != nil { if old != nil {
delete(pool.all, old.Hash()) delete(pool.all, old.Hash())
pendingReplaceCounter.Inc(1) pendingReplaceMeter.Mark(1)
} }
// Failsafe to work around direct pending inserts (tests) // Failsafe to work around direct pending inserts (tests)
if pool.all[hash] == nil { if pool.all[hash] == nil {
@ -895,7 +895,7 @@ func (pool *OrderPool) promoteExecutables(accounts []common.Address) {
hash := tx.Hash() hash := tx.Hash()
delete(pool.all, hash) delete(pool.all, hash)
queuedRateLimitCounter.Inc(1) queuedRateLimitMeter.Mark(1)
log.Debug("Removed cap-exceeding queued transaction", "addr", tx.UserAddress().Hex(), "nonce", tx.Nonce(), "ohash", tx.OrderHash().Hex(), "status", tx.Status(), "orderid", tx.OrderID()) log.Debug("Removed cap-exceeding queued transaction", "addr", tx.UserAddress().Hex(), "nonce", tx.Nonce(), "ohash", tx.OrderHash().Hex(), "status", tx.Status(), "orderid", tx.OrderID())
} }
} }
@ -972,7 +972,7 @@ func (pool *OrderPool) promoteExecutables(accounts []common.Address) {
} }
} }
} }
pendingRateLimitCounter.Inc(int64(pendingBeforeCap - pending)) pendingRateLimitMeter.Mark(int64(pendingBeforeCap - pending))
} }
// If we've queued more transactions than the hard limit, drop oldest ones // If we've queued more transactions than the hard limit, drop oldest ones
queued := uint64(0) queued := uint64(0)
@ -1002,7 +1002,7 @@ func (pool *OrderPool) promoteExecutables(accounts []common.Address) {
pool.removeTx(tx.Hash()) pool.removeTx(tx.Hash())
} }
drop -= size drop -= size
queuedRateLimitCounter.Inc(int64(size)) queuedRateLimitMeter.Mark(int64(size))
continue continue
} }
// Otherwise drop only last few transactions // Otherwise drop only last few transactions
@ -1010,7 +1010,7 @@ func (pool *OrderPool) promoteExecutables(accounts []common.Address) {
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()) pool.removeTx(txs[i].Hash())
drop-- drop--
queuedRateLimitCounter.Inc(1) queuedRateLimitMeter.Mark(1)
} }
} }
} }

4
core/tx_list.go
View file

@ -429,9 +429,9 @@ func (l *txPricedList) Put(tx *types.Transaction) {
// Removed notifies the prices transaction list that an old transaction dropped // Removed notifies the prices transaction list that an old transaction dropped
// from the pool. The list will just keep a counter of stale objects and update // from the pool. The list will just keep a counter of stale objects and update
// the heap if a large enough ratio of transactions go stale. // the heap if a large enough ratio of transactions go stale.
func (l *txPricedList) Removed() { func (l *txPricedList) Removed(count int) {
// Bump the stale counter, but exit if still too low (< 25%) // Bump the stale counter, but exit if still too low (< 25%)
l.stales++ l.stales += count
if l.stales <= len(*l.items)/4 { if l.stales <= len(*l.items)/4 {
return return
} }

156
core/tx_pool.go
View file

@ -94,20 +94,25 @@ var (
var ( var (
// Metrics for the pending pool // Metrics for the pending pool
pendingDiscardCounter = metrics.NewRegisteredCounter("txpool/pending/discard", nil) pendingDiscardMeter = metrics.NewRegisteredMeter("txpool/pending/discard", nil)
pendingReplaceCounter = metrics.NewRegisteredCounter("txpool/pending/replace", nil) pendingReplaceMeter = metrics.NewRegisteredMeter("txpool/pending/replace", nil)
pendingRateLimitCounter = metrics.NewRegisteredCounter("txpool/pending/ratelimit", nil) // Dropped due to rate limiting pendingRateLimitMeter = metrics.NewRegisteredMeter("txpool/pending/ratelimit", nil) // Dropped due to rate limiting
pendingNofundsCounter = metrics.NewRegisteredCounter("txpool/pending/nofunds", nil) // Dropped due to out-of-funds pendingNofundsMeter = metrics.NewRegisteredMeter("txpool/pending/nofunds", nil) // Dropped due to out-of-funds
// Metrics for the queued pool // Metrics for the queued pool
queuedDiscardCounter = metrics.NewRegisteredCounter("txpool/queued/discard", nil) queuedDiscardMeter = metrics.NewRegisteredMeter("txpool/queued/discard", nil)
queuedReplaceCounter = metrics.NewRegisteredCounter("txpool/queued/replace", nil) queuedReplaceMeter = metrics.NewRegisteredMeter("txpool/queued/replace", nil)
queuedRateLimitCounter = metrics.NewRegisteredCounter("txpool/queued/ratelimit", nil) // Dropped due to rate limiting queuedRateLimitMeter = metrics.NewRegisteredMeter("txpool/queued/ratelimit", nil) // Dropped due to rate limiting
queuedNofundsCounter = metrics.NewRegisteredCounter("txpool/queued/nofunds", nil) // Dropped due to out-of-funds queuedNofundsMeter = metrics.NewRegisteredMeter("txpool/queued/nofunds", nil) // Dropped due to out-of-funds
// General tx metrics // General tx metrics
invalidTxCounter = metrics.NewRegisteredCounter("txpool/invalid", nil) validMeter = metrics.NewRegisteredMeter("txpool/valid", nil)
underpricedTxCounter = metrics.NewRegisteredCounter("txpool/underpriced", nil) invalidTxMeter = metrics.NewRegisteredMeter("txpool/invalid", nil)
underpricedTxMeter = metrics.NewRegisteredMeter("txpool/underpriced", nil)
pendingCounter = metrics.NewRegisteredCounter("txpool/pending", nil)
queuedCounter = metrics.NewRegisteredCounter("txpool/queued", nil)
localCounter = metrics.NewRegisteredCounter("txpool/local", nil)
) )
// TxStatus is the current status of a transaction as seen by the pool. // TxStatus is the current status of a transaction as seen by the pool.
@ -760,7 +765,7 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
// If the transaction fails basic validation, discard it // If the transaction fails basic validation, discard it
if err := pool.validateTx(tx, local); err != nil { if err := pool.validateTx(tx, local); err != nil {
log.Trace("Discarding invalid transaction", "hash", hash, "err", err) log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
invalidTxCounter.Inc(1) invalidTxMeter.Mark(1)
return false, err return false, err
} }
from, _ := types.Sender(pool.signer, tx) // already validated from, _ := types.Sender(pool.signer, tx) // already validated
@ -773,14 +778,14 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
// If the new transaction is underpriced, don't accept it // If the new transaction is underpriced, don't accept it
if !local && pool.priced.Underpriced(tx, pool.locals) { if !local && pool.priced.Underpriced(tx, pool.locals) {
log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice()) log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
underpricedTxCounter.Inc(1) underpricedTxMeter.Mark(1)
return false, ErrUnderpriced return false, ErrUnderpriced
} }
// New transaction is better than our worse ones, make room for it // New transaction is better than our worse ones, make room for it
drop := pool.priced.Discard(pool.all.Count()-int(pool.config.GlobalSlots+pool.config.GlobalQueue-1), pool.locals) drop := pool.priced.Discard(pool.all.Count()-int(pool.config.GlobalSlots+pool.config.GlobalQueue-1), pool.locals)
for _, tx := range drop { for _, tx := range drop {
log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice()) log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice())
underpricedTxCounter.Inc(1) underpricedTxMeter.Mark(1)
pool.removeTx(tx.Hash(), false) pool.removeTx(tx.Hash(), false)
} }
} }
@ -789,14 +794,14 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
// Nonce already pending, check if required price bump is met // Nonce already pending, check if required price bump is met
inserted, old := list.Add(tx, pool.config.PriceBump) inserted, old := list.Add(tx, pool.config.PriceBump)
if !inserted { if !inserted {
pendingDiscardCounter.Inc(1) pendingDiscardMeter.Mark(1)
return false, ErrReplaceUnderpriced return false, ErrReplaceUnderpriced
} }
// New transaction is better, replace old one // New transaction is better, replace old one
if old != nil { if old != nil {
pool.all.Remove(old.Hash()) pool.all.Remove(old.Hash())
pool.priced.Removed() pool.priced.Removed(1)
pendingReplaceCounter.Inc(1) pendingReplaceMeter.Mark(1)
} }
pool.all.Add(tx) pool.all.Add(tx)
pool.priced.Put(tx) pool.priced.Put(tx)
@ -821,6 +826,9 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
pool.locals.add(from) pool.locals.add(from)
} }
} }
if local || pool.locals.contains(from) {
localCounter.Inc(1)
}
pool.journalTx(from, tx) pool.journalTx(from, tx)
log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To()) log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
@ -839,14 +847,17 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, er
inserted, old := pool.queue[from].Add(tx, pool.config.PriceBump) inserted, old := pool.queue[from].Add(tx, pool.config.PriceBump)
if !inserted { if !inserted {
// An older transaction was better, discard this // An older transaction was better, discard this
queuedDiscardCounter.Inc(1) queuedDiscardMeter.Mark(1)
return false, ErrReplaceUnderpriced return false, ErrReplaceUnderpriced
} }
// Discard any previous transaction and mark this // Discard any previous transaction and mark this
if old != nil { if old != nil {
pool.all.Remove(old.Hash()) pool.all.Remove(old.Hash())
pool.priced.Removed() pool.priced.Removed(1)
queuedReplaceCounter.Inc(1) queuedReplaceMeter.Mark(1)
} else {
// Nothing was replaced, bump the queued counter
queuedCounter.Inc(1)
} }
if pool.all.Get(hash) == nil { if pool.all.Get(hash) == nil {
pool.all.Add(tx) pool.all.Add(tx)
@ -882,17 +893,20 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
if !inserted { if !inserted {
// An older transaction was better, discard this // An older transaction was better, discard this
pool.all.Remove(hash) pool.all.Remove(hash)
pool.priced.Removed() pool.priced.Removed(1)
pendingDiscardCounter.Inc(1) pendingDiscardMeter.Mark(1)
return false return false
} }
// Otherwise discard any previous transaction and mark this // Otherwise discard any previous transaction and mark this
if old != nil { if old != nil {
pool.all.Remove(old.Hash()) pool.all.Remove(old.Hash())
pool.priced.Removed() pool.priced.Removed(1)
pendingReplaceCounter.Inc(1) pendingReplaceMeter.Mark(1)
} else {
// Nothing was replaced, bump the pending counter
pendingCounter.Inc(1)
} }
// Failsafe to work around direct pending inserts (tests) // Failsafe to work around direct pending inserts (tests)
if pool.all.Get(hash) == nil { if pool.all.Get(hash) == nil {
@ -919,8 +933,11 @@ func (pool *TxPool) promoteSpecialTx(addr common.Address, tx *types.Transaction)
// Otherwise discard any previous transaction and mark this // Otherwise discard any previous transaction and mark this
if old != nil { if old != nil {
pool.all.Remove(old.Hash()) pool.all.Remove(old.Hash())
pool.priced.Removed() pool.priced.Removed(1)
pendingReplaceCounter.Inc(1) pendingReplaceMeter.Mark(1)
} else {
// Nothing was replaced, bump the pending counter
pendingCounter.Inc(1)
} }
list.txs.Put(tx) list.txs.Put(tx)
if cost := tx.Cost(); list.costcap.Cmp(cost) < 0 { if cost := tx.Cost(); list.costcap.Cmp(cost) < 0 {
@ -981,6 +998,8 @@ func (pool *TxPool) addTx(tx *types.Transaction, local bool) error {
if err != nil { if err != nil {
return err return err
} }
validMeter.Mark(1)
// If we added a new transaction, run promotion checks and return // If we added a new transaction, run promotion checks and return
if !replace { if !replace {
from, _ := types.Sender(pool.signer, tx) // already validated from, _ := types.Sender(pool.signer, tx) // already validated
@ -1065,7 +1084,10 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
// Remove it from the list of known transactions // Remove it from the list of known transactions
pool.all.Remove(hash) pool.all.Remove(hash)
if outofbound { if outofbound {
pool.priced.Removed() pool.priced.Removed(1)
}
if pool.locals.contains(addr) {
localCounter.Dec(1)
} }
// Remove the transaction from the pending lists and reset the account nonce // Remove the transaction from the pending lists and reset the account nonce
if pending := pool.pending[addr]; pending != nil { if pending := pool.pending[addr]; pending != nil {
@ -1083,12 +1105,17 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
if nonce := tx.Nonce(); pool.pendingState.GetNonce(addr) > nonce { if nonce := tx.Nonce(); pool.pendingState.GetNonce(addr) > nonce {
pool.pendingState.SetNonce(addr, nonce) pool.pendingState.SetNonce(addr, nonce)
} }
// Reduce the pending counter
pendingCounter.Dec(int64(1 + len(invalids)))
return return
} }
} }
// Transaction is in the future queue // Transaction is in the future queue
if future := pool.queue[addr]; future != nil { if future := pool.queue[addr]; future != nil {
future.Remove(tx) if removed, _ := future.Remove(tx); removed {
// Reduce the queued counter
queuedCounter.Dec(1)
}
if future.Empty() { if future.Empty() {
delete(pool.queue, addr) delete(pool.queue, addr)
} }
@ -1121,11 +1148,11 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
continue // Just in case someone calls with a non existing account continue // Just in case someone calls with a non existing account
} }
// Drop all transactions that are deemed too old (low nonce) // Drop all transactions that are deemed too old (low nonce)
for _, tx := range list.Forward(pool.currentState.GetNonce(addr)) { forwards := list.Forward(pool.currentState.GetNonce(addr))
for _, tx := range forwards {
hash := tx.Hash() hash := tx.Hash()
log.Trace("Removed old queued transaction", "hash", hash)
pool.all.Remove(hash) pool.all.Remove(hash)
pool.priced.Removed() log.Trace("Removed old queued transaction", "hash", hash)
} }
// Drop all transactions that are too costly (low balance or out of gas) // Drop all transactions that are too costly (low balance or out of gas)
var number *big.Int = nil var number *big.Int = nil
@ -1135,28 +1162,38 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas, pool.trc21FeeCapacity, number) drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas, pool.trc21FeeCapacity, number)
for _, tx := range drops { for _, tx := range drops {
hash := tx.Hash() hash := tx.Hash()
log.Trace("Removed unpayable queued transaction", "hash", hash)
pool.all.Remove(hash) pool.all.Remove(hash)
pool.priced.Removed() log.Trace("Removed unpayable queued transaction", "hash", hash)
queuedNofundsCounter.Inc(1)
} }
queuedNofundsMeter.Mark(int64(len(drops)))
// Gather all executable transactions and promote them // Gather all executable transactions and promote them
for _, tx := range list.Ready(pool.pendingState.GetNonce(addr)) { readies := list.Ready(pool.pendingState.GetNonce(addr))
for _, tx := range readies {
hash := tx.Hash() hash := tx.Hash()
if pool.promoteTx(addr, hash, tx) { if pool.promoteTx(addr, hash, tx) {
log.Trace("Promoting queued transaction", "hash", hash) log.Trace("Promoting queued transaction", "hash", hash)
promoted = append(promoted, tx) promoted = append(promoted, tx)
} }
} }
queuedCounter.Dec(int64(len(readies)))
// Drop all transactions over the allowed limit // Drop all transactions over the allowed limit
var caps types.Transactions
if !pool.locals.contains(addr) { if !pool.locals.contains(addr) {
for _, tx := range list.Cap(int(pool.config.AccountQueue)) { caps = list.Cap(int(pool.config.AccountQueue))
for _, tx := range caps {
hash := tx.Hash() hash := tx.Hash()
pool.all.Remove(hash) pool.all.Remove(hash)
pool.priced.Removed()
queuedRateLimitCounter.Inc(1)
log.Trace("Removed cap-exceeding queued transaction", "hash", hash) log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
} }
queuedRateLimitMeter.Mark(int64(len(caps)))
}
// Mark all the items dropped as removed
pool.priced.Removed(len(forwards) + len(drops) + len(caps))
queuedCounter.Dec(int64(len(forwards) + len(drops) + len(caps)))
if pool.locals.contains(addr) {
localCounter.Dec(int64(len(forwards) + len(drops) + len(caps)))
} }
// Delete the entire queue entry if it became empty. // Delete the entire queue entry if it became empty.
if list.Empty() { if list.Empty() {
@ -1198,11 +1235,12 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
for pending > pool.config.GlobalSlots && pool.pending[offenders[len(offenders)-2]].Len() > threshold { for pending > pool.config.GlobalSlots && pool.pending[offenders[len(offenders)-2]].Len() > threshold {
for i := 0; i < len(offenders)-1; i++ { for i := 0; i < len(offenders)-1; i++ {
list := pool.pending[offenders[i]] list := pool.pending[offenders[i]]
for _, tx := range list.Cap(list.Len() - 1) {
caps := list.Cap(list.Len() - 1)
for _, tx := range caps {
// Drop the transaction from the global pools too // Drop the transaction from the global pools too
hash := tx.Hash() hash := tx.Hash()
pool.all.Remove(hash) pool.all.Remove(hash)
pool.priced.Removed()
// Update the account nonce to the dropped transaction // Update the account nonce to the dropped transaction
if nonce := tx.Nonce(); pool.pendingState.GetNonce(offenders[i]) > nonce { if nonce := tx.Nonce(); pool.pendingState.GetNonce(offenders[i]) > nonce {
@ -1210,6 +1248,11 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
} }
log.Trace("Removed fairness-exceeding pending transaction", "hash", hash) log.Trace("Removed fairness-exceeding pending transaction", "hash", hash)
} }
pool.priced.Removed(len(caps))
pendingCounter.Dec(int64(len(caps)))
if pool.locals.contains(offenders[i]) {
localCounter.Dec(int64(len(caps)))
}
pending-- pending--
} }
} }
@ -1220,11 +1263,12 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
for pending > pool.config.GlobalSlots && uint64(pool.pending[offenders[len(offenders)-1]].Len()) > pool.config.AccountSlots { for pending > pool.config.GlobalSlots && uint64(pool.pending[offenders[len(offenders)-1]].Len()) > pool.config.AccountSlots {
for _, addr := range offenders { for _, addr := range offenders {
list := pool.pending[addr] list := pool.pending[addr]
for _, tx := range list.Cap(list.Len() - 1) {
caps := list.Cap(list.Len() - 1)
for _, tx := range caps {
// Drop the transaction from the global pools too // Drop the transaction from the global pools too
hash := tx.Hash() hash := tx.Hash()
pool.all.Remove(hash) pool.all.Remove(hash)
pool.priced.Removed()
// Update the account nonce to the dropped transaction // Update the account nonce to the dropped transaction
if nonce := tx.Nonce(); pool.pendingState.GetNonce(addr) > nonce { if nonce := tx.Nonce(); pool.pendingState.GetNonce(addr) > nonce {
@ -1232,11 +1276,16 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
} }
log.Trace("Removed fairness-exceeding pending transaction", "hash", hash) log.Trace("Removed fairness-exceeding pending transaction", "hash", hash)
} }
pool.priced.Removed(len(caps))
pendingCounter.Dec(int64(len(caps)))
if pool.locals.contains(addr) {
localCounter.Dec(int64(len(caps)))
}
pending-- pending--
} }
} }
} }
pendingRateLimitCounter.Inc(int64(pendingBeforeCap - pending)) pendingRateLimitMeter.Mark(int64(pendingBeforeCap - pending))
} }
// If we've queued more transactions than the hard limit, drop oldest ones // If we've queued more transactions than the hard limit, drop oldest ones
queued := uint64(0) queued := uint64(0)
@ -1266,7 +1315,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
pool.removeTx(tx.Hash(), true) pool.removeTx(tx.Hash(), true)
} }
drop -= size drop -= size
queuedRateLimitCounter.Inc(int64(size)) queuedRateLimitMeter.Mark(int64(size))
continue continue
} }
// Otherwise drop only last few transactions // Otherwise drop only last few transactions
@ -1274,7 +1323,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
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) pool.removeTx(txs[i].Hash(), true)
drop-- drop--
queuedRateLimitCounter.Inc(1) queuedRateLimitMeter.Mark(1)
} }
} }
} }
@ -1289,11 +1338,11 @@ func (pool *TxPool) demoteUnexecutables() {
nonce := pool.currentState.GetNonce(addr) nonce := pool.currentState.GetNonce(addr)
// Drop all transactions that are deemed too old (low nonce) // Drop all transactions that are deemed too old (low nonce)
for _, tx := range list.Forward(nonce) { olds := list.Forward(nonce)
for _, tx := range olds {
hash := tx.Hash() hash := tx.Hash()
log.Trace("Removed old pending transaction", "hash", hash)
pool.all.Remove(hash) pool.all.Remove(hash)
pool.priced.Removed() log.Trace("Removed old pending transaction", "hash", hash)
} }
// Drop all transactions that are too costly (low balance or out of gas), and queue any invalids back for later // Drop all transactions that are too costly (low balance or out of gas), and queue any invalids back for later
var number *big.Int = nil var number *big.Int = nil
@ -1305,21 +1354,28 @@ func (pool *TxPool) demoteUnexecutables() {
hash := tx.Hash() hash := tx.Hash()
log.Trace("Removed unpayable pending transaction", "hash", hash) log.Trace("Removed unpayable pending transaction", "hash", hash)
pool.all.Remove(hash) pool.all.Remove(hash)
pool.priced.Removed()
pendingNofundsCounter.Inc(1)
} }
pool.priced.Removed(len(olds) + len(drops))
pendingNofundsMeter.Mark(int64(len(drops)))
for _, tx := range invalids { for _, tx := range invalids {
hash := tx.Hash() hash := tx.Hash()
log.Trace("Demoting pending transaction", "hash", hash) log.Trace("Demoting pending transaction", "hash", hash)
pool.enqueueTx(hash, tx) pool.enqueueTx(hash, tx)
} }
pendingCounter.Dec(int64(len(olds) + len(drops) + len(invalids)))
if pool.locals.contains(addr) {
localCounter.Dec(int64(len(olds) + len(drops) + len(invalids)))
}
// If there's a gap in front, alert (should never happen) and postpone all transactions // If there's a gap in front, alert (should never happen) and postpone all transactions
if list.Len() > 0 && list.txs.Get(nonce) == nil { if list.Len() > 0 && list.txs.Get(nonce) == nil {
for _, tx := range list.Cap(0) { gapped := list.Cap(0)
for _, tx := range gapped {
hash := tx.Hash() hash := tx.Hash()
log.Warn("Demoting invalidated transaction", "hash", hash) log.Warn("Demoting invalidated transaction", "hash", hash)
pool.enqueueTx(hash, tx) pool.enqueueTx(hash, tx)
} }
pendingCounter.Inc(int64(len(gapped)))
} }
// Delete the entire queue entry if it became empty. // Delete the entire queue entry if it became empty.
if list.Empty() { if list.Empty() {