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++
 | 
						|
	}
 | 
						|
}
 |