mirror of
https://gitlab.com/pulsechaincom/go-pulse.git
synced 2024-12-28 14:27:16 +00:00
94c71c171f
Logger updates
751 lines
23 KiB
Go
751 lines
23 KiB
Go
// Copyright 2014 The go-ethereum Authors
|
|
// This file is part of the go-ethereum library.
|
|
//
|
|
// The go-ethereum library is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU Lesser General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// The go-ethereum library is distributed in the hope that it will be useful,
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU Lesser General Public License for more details.
|
|
//
|
|
// You should have received a copy of the GNU Lesser General Public License
|
|
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
package core
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"math/big"
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/common"
|
|
"github.com/ethereum/go-ethereum/core/state"
|
|
"github.com/ethereum/go-ethereum/core/types"
|
|
"github.com/ethereum/go-ethereum/event"
|
|
"github.com/ethereum/go-ethereum/log"
|
|
"github.com/ethereum/go-ethereum/metrics"
|
|
"github.com/ethereum/go-ethereum/params"
|
|
"gopkg.in/karalabe/cookiejar.v2/collections/prque"
|
|
)
|
|
|
|
var (
|
|
// Transaction Pool Errors
|
|
ErrInvalidSender = errors.New("Invalid sender")
|
|
ErrNonce = errors.New("Nonce too low")
|
|
ErrCheap = errors.New("Gas price too low for acceptance")
|
|
ErrBalance = errors.New("Insufficient balance")
|
|
ErrInsufficientFunds = errors.New("Insufficient funds for gas * price + value")
|
|
ErrIntrinsicGas = errors.New("Intrinsic gas too low")
|
|
ErrGasLimit = errors.New("Exceeds block gas limit")
|
|
ErrNegativeValue = errors.New("Negative value")
|
|
)
|
|
|
|
var (
|
|
minPendingPerAccount = uint64(16) // Min number of guaranteed transaction slots per address
|
|
maxPendingTotal = uint64(4096) // Max limit of pending transactions from all accounts (soft)
|
|
maxQueuedPerAccount = uint64(64) // Max limit of queued transactions per address
|
|
maxQueuedInTotal = uint64(1024) // Max limit of queued transactions from all accounts
|
|
maxQueuedLifetime = 3 * time.Hour // Max amount of time transactions from idle accounts are queued
|
|
evictionInterval = time.Minute // Time interval to check for evictable transactions
|
|
)
|
|
|
|
var (
|
|
// Metrics for the pending pool
|
|
pendingDiscardCounter = metrics.NewCounter("txpool/pending/discard")
|
|
pendingReplaceCounter = metrics.NewCounter("txpool/pending/replace")
|
|
pendingRLCounter = metrics.NewCounter("txpool/pending/ratelimit") // Dropped due to rate limiting
|
|
pendingNofundsCounter = metrics.NewCounter("txpool/pending/nofunds") // Dropped due to out-of-funds
|
|
|
|
// Metrics for the queued pool
|
|
queuedDiscardCounter = metrics.NewCounter("txpool/queued/discard")
|
|
queuedReplaceCounter = metrics.NewCounter("txpool/queued/replace")
|
|
queuedRLCounter = metrics.NewCounter("txpool/queued/ratelimit") // Dropped due to rate limiting
|
|
queuedNofundsCounter = metrics.NewCounter("txpool/queued/nofunds") // Dropped due to out-of-funds
|
|
|
|
// General tx metrics
|
|
invalidTxCounter = metrics.NewCounter("txpool/invalid")
|
|
)
|
|
|
|
type stateFn func() (*state.StateDB, error)
|
|
|
|
// TxPool contains all currently known transactions. Transactions
|
|
// enter the pool when they are received from the network or submitted
|
|
// locally. They exit the pool when they are included in the blockchain.
|
|
//
|
|
// The pool separates processable transactions (which can be applied to the
|
|
// current state) and future transactions. Transactions move between those
|
|
// two states over time as they are received and processed.
|
|
type TxPool struct {
|
|
config *params.ChainConfig
|
|
currentState stateFn // The state function which will allow us to do some pre checks
|
|
pendingState *state.ManagedState
|
|
gasLimit func() *big.Int // The current gas limit function callback
|
|
minGasPrice *big.Int
|
|
eventMux *event.TypeMux
|
|
events *event.TypeMuxSubscription
|
|
localTx *txSet
|
|
signer types.Signer
|
|
mu sync.RWMutex
|
|
|
|
pending map[common.Address]*txList // All currently processable transactions
|
|
queue map[common.Address]*txList // Queued but non-processable transactions
|
|
all map[common.Hash]*types.Transaction // All transactions to allow lookups
|
|
beats map[common.Address]time.Time // Last heartbeat from each known account
|
|
|
|
wg sync.WaitGroup // for shutdown sync
|
|
quit chan struct{}
|
|
|
|
homestead bool
|
|
}
|
|
|
|
func NewTxPool(config *params.ChainConfig, eventMux *event.TypeMux, currentStateFn stateFn, gasLimitFn func() *big.Int) *TxPool {
|
|
pool := &TxPool{
|
|
config: config,
|
|
signer: types.NewEIP155Signer(config.ChainId),
|
|
pending: make(map[common.Address]*txList),
|
|
queue: make(map[common.Address]*txList),
|
|
all: make(map[common.Hash]*types.Transaction),
|
|
beats: make(map[common.Address]time.Time),
|
|
eventMux: eventMux,
|
|
currentState: currentStateFn,
|
|
gasLimit: gasLimitFn,
|
|
minGasPrice: new(big.Int),
|
|
pendingState: nil,
|
|
localTx: newTxSet(),
|
|
events: eventMux.Subscribe(ChainHeadEvent{}, GasPriceChanged{}, RemovedTransactionEvent{}),
|
|
quit: make(chan struct{}),
|
|
}
|
|
|
|
pool.resetState()
|
|
|
|
pool.wg.Add(2)
|
|
go pool.eventLoop()
|
|
go pool.expirationLoop()
|
|
|
|
return pool
|
|
}
|
|
|
|
func (pool *TxPool) eventLoop() {
|
|
defer pool.wg.Done()
|
|
|
|
// Track chain events. When a chain events occurs (new chain canon block)
|
|
// we need to know the new state. The new state will help us determine
|
|
// the nonces in the managed state
|
|
for ev := range pool.events.Chan() {
|
|
switch ev := ev.Data.(type) {
|
|
case ChainHeadEvent:
|
|
pool.mu.Lock()
|
|
if ev.Block != nil {
|
|
if pool.config.IsHomestead(ev.Block.Number()) {
|
|
pool.homestead = true
|
|
}
|
|
}
|
|
|
|
pool.resetState()
|
|
pool.mu.Unlock()
|
|
case GasPriceChanged:
|
|
pool.mu.Lock()
|
|
pool.minGasPrice = ev.Price
|
|
pool.mu.Unlock()
|
|
case RemovedTransactionEvent:
|
|
pool.AddBatch(ev.Txs)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (pool *TxPool) resetState() {
|
|
currentState, err := pool.currentState()
|
|
if err != nil {
|
|
log.Error("Failed reset txpool state", "err", err)
|
|
return
|
|
}
|
|
pool.pendingState = state.ManageState(currentState)
|
|
|
|
// validate the pool of pending transactions, this will remove
|
|
// any transactions that have been included in the block or
|
|
// have been invalidated because of another transaction (e.g.
|
|
// higher gas price)
|
|
pool.demoteUnexecutables(currentState)
|
|
|
|
// Update all accounts to the latest known pending nonce
|
|
for addr, list := range pool.pending {
|
|
txs := list.Flatten() // Heavy but will be cached and is needed by the miner anyway
|
|
pool.pendingState.SetNonce(addr, txs[len(txs)-1].Nonce()+1)
|
|
}
|
|
// Check the queue and move transactions over to the pending if possible
|
|
// or remove those that have become invalid
|
|
pool.promoteExecutables(currentState)
|
|
}
|
|
|
|
func (pool *TxPool) Stop() {
|
|
pool.events.Unsubscribe()
|
|
close(pool.quit)
|
|
pool.wg.Wait()
|
|
|
|
log.Info("Transaction pool stopped")
|
|
}
|
|
|
|
func (pool *TxPool) State() *state.ManagedState {
|
|
pool.mu.RLock()
|
|
defer pool.mu.RUnlock()
|
|
|
|
return pool.pendingState
|
|
}
|
|
|
|
// Stats retrieves the current pool stats, namely the number of pending and the
|
|
// number of queued (non-executable) transactions.
|
|
func (pool *TxPool) Stats() (pending int, queued int) {
|
|
pool.mu.RLock()
|
|
defer pool.mu.RUnlock()
|
|
|
|
for _, list := range pool.pending {
|
|
pending += list.Len()
|
|
}
|
|
for _, list := range pool.queue {
|
|
queued += list.Len()
|
|
}
|
|
return
|
|
}
|
|
|
|
// Content retrieves the data content of the transaction pool, returning all the
|
|
// pending as well as queued transactions, grouped by account and sorted by nonce.
|
|
func (pool *TxPool) Content() (map[common.Address]types.Transactions, map[common.Address]types.Transactions) {
|
|
pool.mu.RLock()
|
|
defer pool.mu.RUnlock()
|
|
|
|
pending := make(map[common.Address]types.Transactions)
|
|
for addr, list := range pool.pending {
|
|
pending[addr] = list.Flatten()
|
|
}
|
|
queued := make(map[common.Address]types.Transactions)
|
|
for addr, list := range pool.queue {
|
|
queued[addr] = list.Flatten()
|
|
}
|
|
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, error) {
|
|
pool.mu.Lock()
|
|
defer pool.mu.Unlock()
|
|
|
|
state, err := pool.currentState()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// check queue first
|
|
pool.promoteExecutables(state)
|
|
|
|
// invalidate any txs
|
|
pool.demoteUnexecutables(state)
|
|
|
|
pending := make(map[common.Address]types.Transactions)
|
|
for addr, list := range pool.pending {
|
|
pending[addr] = list.Flatten()
|
|
}
|
|
return pending, nil
|
|
}
|
|
|
|
// SetLocal marks a transaction as local, skipping gas price
|
|
// check against local miner minimum in the future
|
|
func (pool *TxPool) SetLocal(tx *types.Transaction) {
|
|
pool.mu.Lock()
|
|
defer pool.mu.Unlock()
|
|
pool.localTx.add(tx.Hash())
|
|
}
|
|
|
|
// validateTx checks whether a transaction is valid according
|
|
// to the consensus rules.
|
|
func (pool *TxPool) validateTx(tx *types.Transaction) error {
|
|
local := pool.localTx.contains(tx.Hash())
|
|
// Drop transactions under our own minimal accepted gas price
|
|
if !local && pool.minGasPrice.Cmp(tx.GasPrice()) > 0 {
|
|
return ErrCheap
|
|
}
|
|
|
|
currentState, err := pool.currentState()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
from, err := types.Sender(pool.signer, tx)
|
|
if err != nil {
|
|
return ErrInvalidSender
|
|
}
|
|
// Last but not least check for nonce errors
|
|
if currentState.GetNonce(from) > tx.Nonce() {
|
|
return ErrNonce
|
|
}
|
|
|
|
// Check the transaction doesn't exceed the current
|
|
// block limit gas.
|
|
if pool.gasLimit().Cmp(tx.Gas()) < 0 {
|
|
return ErrGasLimit
|
|
}
|
|
|
|
// Transactions can't be negative. This may never happen
|
|
// using RLP decoded transactions but may occur if you create
|
|
// a transaction using the RPC for example.
|
|
if tx.Value().Sign() < 0 {
|
|
return ErrNegativeValue
|
|
}
|
|
|
|
// Transactor should have enough funds to cover the costs
|
|
// cost == V + GP * GL
|
|
if currentState.GetBalance(from).Cmp(tx.Cost()) < 0 {
|
|
return ErrInsufficientFunds
|
|
}
|
|
|
|
intrGas := IntrinsicGas(tx.Data(), tx.To() == nil, pool.homestead)
|
|
if tx.Gas().Cmp(intrGas) < 0 {
|
|
return ErrIntrinsicGas
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// add validates a transaction and inserts it into the non-executable queue for
|
|
// later pending promotion and execution.
|
|
func (pool *TxPool) add(tx *types.Transaction) error {
|
|
// If the transaction is already known, discard it
|
|
hash := tx.Hash()
|
|
if pool.all[hash] != nil {
|
|
log.Trace("Discarding already known transaction", "hash", hash)
|
|
return fmt.Errorf("known transaction: %x", hash)
|
|
}
|
|
// Otherwise ensure basic validation passes and queue it up
|
|
if err := pool.validateTx(tx); err != nil {
|
|
log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
|
|
invalidTxCounter.Inc(1)
|
|
return err
|
|
}
|
|
pool.enqueueTx(hash, tx)
|
|
|
|
// Print a log message if low enough level is set
|
|
log.Debug("Pooled new transaction", "hash", hash, "from", log.Lazy{Fn: func() common.Address { from, _ := types.Sender(pool.signer, tx); return from }}, "to", tx.To())
|
|
return nil
|
|
}
|
|
|
|
// enqueueTx inserts a new transaction into the non-executable transaction queue.
|
|
//
|
|
// Note, this method assumes the pool lock is held!
|
|
func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) {
|
|
// Try to insert the transaction into the future queue
|
|
from, _ := types.Sender(pool.signer, tx) // already validated
|
|
if pool.queue[from] == nil {
|
|
pool.queue[from] = newTxList(false)
|
|
}
|
|
inserted, old := pool.queue[from].Add(tx)
|
|
if !inserted {
|
|
queuedDiscardCounter.Inc(1)
|
|
return // An older transaction was better, discard this
|
|
}
|
|
// Discard any previous transaction and mark this
|
|
if old != nil {
|
|
delete(pool.all, old.Hash())
|
|
queuedReplaceCounter.Inc(1)
|
|
}
|
|
pool.all[hash] = tx
|
|
}
|
|
|
|
// promoteTx adds a transaction to the pending (processable) list of transactions.
|
|
//
|
|
// Note, this method assumes the pool lock is held!
|
|
func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.Transaction) {
|
|
// Try to insert the transaction into the pending queue
|
|
if pool.pending[addr] == nil {
|
|
pool.pending[addr] = newTxList(true)
|
|
}
|
|
list := pool.pending[addr]
|
|
|
|
inserted, old := list.Add(tx)
|
|
if !inserted {
|
|
// An older transaction was better, discard this
|
|
delete(pool.all, hash)
|
|
pendingDiscardCounter.Inc(1)
|
|
return
|
|
}
|
|
// Otherwise discard any previous transaction and mark this
|
|
if old != nil {
|
|
delete(pool.all, old.Hash())
|
|
pendingReplaceCounter.Inc(1)
|
|
}
|
|
pool.all[hash] = tx // Failsafe to work around direct pending inserts (tests)
|
|
|
|
// Set the potentially new pending nonce and notify any subsystems of the new tx
|
|
pool.beats[addr] = time.Now()
|
|
pool.pendingState.SetNonce(addr, tx.Nonce()+1)
|
|
go pool.eventMux.Post(TxPreEvent{tx})
|
|
}
|
|
|
|
// Add queues a single transaction in the pool if it is valid.
|
|
func (pool *TxPool) Add(tx *types.Transaction) error {
|
|
pool.mu.Lock()
|
|
defer pool.mu.Unlock()
|
|
|
|
if err := pool.add(tx); err != nil {
|
|
return err
|
|
}
|
|
|
|
state, err := pool.currentState()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
pool.promoteExecutables(state)
|
|
|
|
return nil
|
|
}
|
|
|
|
// AddBatch attempts to queue a batch of transactions.
|
|
func (pool *TxPool) AddBatch(txs []*types.Transaction) error {
|
|
pool.mu.Lock()
|
|
defer pool.mu.Unlock()
|
|
|
|
// Add the batch of transaction, tracking the accepted ones
|
|
added := 0
|
|
for _, tx := range txs {
|
|
if err := pool.add(tx); err == nil {
|
|
added++
|
|
}
|
|
}
|
|
// Only reprocess the internal state if something was actually added
|
|
if added > 0 {
|
|
state, err := pool.currentState()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
pool.promoteExecutables(state)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Get returns a transaction if it is contained in the pool
|
|
// and nil otherwise.
|
|
func (pool *TxPool) Get(hash common.Hash) *types.Transaction {
|
|
pool.mu.RLock()
|
|
defer pool.mu.RUnlock()
|
|
|
|
return pool.all[hash]
|
|
}
|
|
|
|
// Remove removes the transaction with the given hash from the pool.
|
|
func (pool *TxPool) Remove(hash common.Hash) {
|
|
pool.mu.Lock()
|
|
defer pool.mu.Unlock()
|
|
|
|
pool.removeTx(hash)
|
|
}
|
|
|
|
// RemoveBatch removes all given transactions from the pool.
|
|
func (pool *TxPool) RemoveBatch(txs types.Transactions) {
|
|
pool.mu.Lock()
|
|
defer pool.mu.Unlock()
|
|
|
|
for _, tx := range txs {
|
|
pool.removeTx(tx.Hash())
|
|
}
|
|
}
|
|
|
|
// removeTx removes a single transaction from the queue, moving all subsequent
|
|
// transactions back to the future queue.
|
|
func (pool *TxPool) removeTx(hash common.Hash) {
|
|
// Fetch the transaction we wish to delete
|
|
tx, ok := pool.all[hash]
|
|
if !ok {
|
|
return
|
|
}
|
|
addr, _ := types.Sender(pool.signer, tx) // already validated during insertion
|
|
|
|
// Remove it from the list of known transactions
|
|
delete(pool.all, hash)
|
|
|
|
// Remove the transaction from the pending lists and reset the account nonce
|
|
if pending := pool.pending[addr]; pending != nil {
|
|
if removed, invalids := pending.Remove(tx); removed {
|
|
// If no more transactions are left, remove the list
|
|
if pending.Empty() {
|
|
delete(pool.pending, addr)
|
|
delete(pool.beats, addr)
|
|
} else {
|
|
// Otherwise postpone any invalidated transactions
|
|
for _, tx := range invalids {
|
|
pool.enqueueTx(tx.Hash(), tx)
|
|
}
|
|
}
|
|
// Update the account nonce if needed
|
|
if nonce := tx.Nonce(); pool.pendingState.GetNonce(addr) > nonce {
|
|
pool.pendingState.SetNonce(addr, tx.Nonce())
|
|
}
|
|
}
|
|
}
|
|
// Transaction is in the future queue
|
|
if future := pool.queue[addr]; future != nil {
|
|
future.Remove(tx)
|
|
if future.Empty() {
|
|
delete(pool.queue, addr)
|
|
}
|
|
}
|
|
}
|
|
|
|
// promoteExecutables moves transactions that have become processable from the
|
|
// future queue to the set of pending transactions. During this process, all
|
|
// invalidated transactions (low nonce, low balance) are deleted.
|
|
func (pool *TxPool) promoteExecutables(state *state.StateDB) {
|
|
// Iterate over all accounts and promote any executable transactions
|
|
queued := uint64(0)
|
|
for addr, list := range pool.queue {
|
|
// Drop all transactions that are deemed too old (low nonce)
|
|
for _, tx := range list.Forward(state.GetNonce(addr)) {
|
|
hash := tx.Hash()
|
|
log.Debug("Removed old queued transaction", "hash", hash)
|
|
delete(pool.all, hash)
|
|
}
|
|
// Drop all transactions that are too costly (low balance)
|
|
drops, _ := list.Filter(state.GetBalance(addr))
|
|
for _, tx := range drops {
|
|
hash := tx.Hash()
|
|
log.Debug("Removed unpayable queued transaction", "hash", hash)
|
|
delete(pool.all, hash)
|
|
queuedNofundsCounter.Inc(1)
|
|
}
|
|
// Gather all executable transactions and promote them
|
|
for _, tx := range list.Ready(pool.pendingState.GetNonce(addr)) {
|
|
hash := tx.Hash()
|
|
log.Debug("Promoting queued transaction", "hash", hash)
|
|
pool.promoteTx(addr, hash, tx)
|
|
}
|
|
// Drop all transactions over the allowed limit
|
|
for _, tx := range list.Cap(int(maxQueuedPerAccount)) {
|
|
hash := tx.Hash()
|
|
log.Debug("Removed cap-exceeding queued transaction", "hash", hash)
|
|
delete(pool.all, hash)
|
|
queuedRLCounter.Inc(1)
|
|
}
|
|
queued += uint64(list.Len())
|
|
|
|
// Delete the entire queue entry if it became empty.
|
|
if list.Empty() {
|
|
delete(pool.queue, addr)
|
|
}
|
|
}
|
|
// If the pending limit is overflown, start equalizing allowances
|
|
pending := uint64(0)
|
|
for _, list := range pool.pending {
|
|
pending += uint64(list.Len())
|
|
}
|
|
if pending > maxPendingTotal {
|
|
pendingBeforeCap := pending
|
|
// Assemble a spam order to penalize large transactors first
|
|
spammers := prque.New()
|
|
for addr, list := range pool.pending {
|
|
// Only evict transactions from high rollers
|
|
if uint64(list.Len()) > minPendingPerAccount {
|
|
// Skip local accounts as pools should maintain backlogs for themselves
|
|
for _, tx := range list.txs.items {
|
|
if !pool.localTx.contains(tx.Hash()) {
|
|
spammers.Push(addr, float32(list.Len()))
|
|
}
|
|
break // Checking on transaction for locality is enough
|
|
}
|
|
}
|
|
}
|
|
// Gradually drop transactions from offenders
|
|
offenders := []common.Address{}
|
|
for pending > maxPendingTotal && !spammers.Empty() {
|
|
// Retrieve the next offender if not local address
|
|
offender, _ := spammers.Pop()
|
|
offenders = append(offenders, offender.(common.Address))
|
|
|
|
// 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.(common.Address)].Len()
|
|
|
|
// Iteratively reduce all offenders until below limit or threshold reached
|
|
for pending > maxPendingTotal && pool.pending[offenders[len(offenders)-2]].Len() > threshold {
|
|
for i := 0; i < len(offenders)-1; i++ {
|
|
list := pool.pending[offenders[i]]
|
|
list.Cap(list.Len() - 1)
|
|
pending--
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// If still above threshold, reduce to limit or min allowance
|
|
if pending > maxPendingTotal && len(offenders) > 0 {
|
|
for pending > maxPendingTotal && uint64(pool.pending[offenders[len(offenders)-1]].Len()) > minPendingPerAccount {
|
|
for _, addr := range offenders {
|
|
list := pool.pending[addr]
|
|
list.Cap(list.Len() - 1)
|
|
pending--
|
|
}
|
|
}
|
|
}
|
|
pendingRLCounter.Inc(int64(pendingBeforeCap - pending))
|
|
}
|
|
// If we've queued more transactions than the hard limit, drop oldest ones
|
|
if queued > maxQueuedInTotal {
|
|
// Sort all accounts with queued transactions by heartbeat
|
|
addresses := make(addresssByHeartbeat, 0, len(pool.queue))
|
|
for addr := range pool.queue {
|
|
addresses = append(addresses, addressByHeartbeat{addr, pool.beats[addr]})
|
|
}
|
|
sort.Sort(addresses)
|
|
|
|
// Drop transactions until the total is below the limit
|
|
for drop := queued - maxQueuedInTotal; drop > 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 {
|
|
for _, tx := range list.Flatten() {
|
|
pool.removeTx(tx.Hash())
|
|
}
|
|
drop -= size
|
|
queuedRLCounter.Inc(int64(size))
|
|
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())
|
|
drop--
|
|
queuedRLCounter.Inc(1)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// demoteUnexecutables removes invalid and processed transactions from the pools
|
|
// executable/pending queue and any subsequent transactions that become unexecutable
|
|
// are moved back into the future queue.
|
|
func (pool *TxPool) demoteUnexecutables(state *state.StateDB) {
|
|
// Iterate over all accounts and demote any non-executable transactions
|
|
for addr, list := range pool.pending {
|
|
nonce := state.GetNonce(addr)
|
|
|
|
// Drop all transactions that are deemed too old (low nonce)
|
|
for _, tx := range list.Forward(nonce) {
|
|
hash := tx.Hash()
|
|
log.Debug("Removed old pending transaction", "hash", hash)
|
|
delete(pool.all, hash)
|
|
}
|
|
// Drop all transactions that are too costly (low balance), and queue any invalids back for later
|
|
drops, invalids := list.Filter(state.GetBalance(addr))
|
|
for _, tx := range drops {
|
|
hash := tx.Hash()
|
|
log.Debug("Removed unpayable pending transaction", "hash", hash)
|
|
delete(pool.all, hash)
|
|
pendingNofundsCounter.Inc(1)
|
|
}
|
|
for _, tx := range invalids {
|
|
hash := tx.Hash()
|
|
log.Debug("Demoting pending transaction", "hash", hash)
|
|
pool.enqueueTx(hash, tx)
|
|
}
|
|
// Delete the entire queue entry if it became empty.
|
|
if list.Empty() {
|
|
delete(pool.pending, addr)
|
|
delete(pool.beats, addr)
|
|
}
|
|
}
|
|
}
|
|
|
|
// expirationLoop is a loop that periodically iterates over all accounts with
|
|
// queued transactions and drop all that have been inactive for a prolonged amount
|
|
// of time.
|
|
func (pool *TxPool) expirationLoop() {
|
|
defer pool.wg.Done()
|
|
|
|
evict := time.NewTicker(evictionInterval)
|
|
defer evict.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-evict.C:
|
|
pool.mu.Lock()
|
|
for addr := range pool.queue {
|
|
if time.Since(pool.beats[addr]) > maxQueuedLifetime {
|
|
for _, tx := range pool.queue[addr].Flatten() {
|
|
pool.removeTx(tx.Hash())
|
|
}
|
|
}
|
|
}
|
|
pool.mu.Unlock()
|
|
|
|
case <-pool.quit:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// addressByHeartbeat is an account address tagged with its last activity timestamp.
|
|
type addressByHeartbeat struct {
|
|
address common.Address
|
|
heartbeat time.Time
|
|
}
|
|
|
|
type addresssByHeartbeat []addressByHeartbeat
|
|
|
|
func (a addresssByHeartbeat) Len() int { return len(a) }
|
|
func (a addresssByHeartbeat) Less(i, j int) bool { return a[i].heartbeat.Before(a[j].heartbeat) }
|
|
func (a addresssByHeartbeat) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
|
|
|
|
// txSet represents a set of transaction hashes in which entries
|
|
// are automatically dropped after txSetDuration time
|
|
type txSet struct {
|
|
txMap map[common.Hash]struct{}
|
|
txOrd map[uint64]txOrdType
|
|
addPtr, delPtr uint64
|
|
}
|
|
|
|
const txSetDuration = time.Hour * 2
|
|
|
|
// txOrdType represents an entry in the time-ordered list of transaction hashes
|
|
type txOrdType struct {
|
|
hash common.Hash
|
|
time time.Time
|
|
}
|
|
|
|
// newTxSet creates a new transaction set
|
|
func newTxSet() *txSet {
|
|
return &txSet{
|
|
txMap: make(map[common.Hash]struct{}),
|
|
txOrd: make(map[uint64]txOrdType),
|
|
}
|
|
}
|
|
|
|
// contains returns true if the set contains the given transaction hash
|
|
// (not thread safe, should be called from a locked environment)
|
|
func (self *txSet) contains(hash common.Hash) bool {
|
|
_, ok := self.txMap[hash]
|
|
return ok
|
|
}
|
|
|
|
// add adds a transaction hash to the set, then removes entries older than txSetDuration
|
|
// (not thread safe, should be called from a locked environment)
|
|
func (self *txSet) add(hash common.Hash) {
|
|
self.txMap[hash] = struct{}{}
|
|
now := time.Now()
|
|
self.txOrd[self.addPtr] = txOrdType{hash: hash, time: now}
|
|
self.addPtr++
|
|
delBefore := now.Add(-txSetDuration)
|
|
for self.delPtr < self.addPtr && self.txOrd[self.delPtr].time.Before(delBefore) {
|
|
delete(self.txMap, self.txOrd[self.delPtr].hash)
|
|
delete(self.txOrd, self.delPtr)
|
|
self.delPtr++
|
|
}
|
|
}
|