// 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 . package core import ( "context" "errors" "fmt" "math" "math/big" "sort" "sync" "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/prque" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/params" "github.com/harmony-one/harmony/core/state" "github.com/harmony-one/harmony/core/types" "github.com/harmony-one/harmony/internal/utils" ) const ( // chainHeadChanSize is the size of channel listening to ChainHeadEvent. chainHeadChanSize = 10 ) var ( // ErrInvalidSender is returned if the transaction contains an invalid signature. ErrInvalidSender = errors.New("invalid sender") // ErrNonceTooLow is returned if the nonce of a transaction is lower than the // one present in the local chain. ErrNonceTooLow = errors.New("nonce too low") // ErrUnderpriced is returned if a transaction's gas price is below the minimum // configured for the transaction pool. ErrUnderpriced = errors.New("transaction underpriced") // ErrReplaceUnderpriced is returned if a transaction is attempted to be replaced // with a different one without the required price bump. ErrReplaceUnderpriced = errors.New("replacement transaction underpriced") // ErrInsufficientFunds is returned if the total cost of executing a transaction // is higher than the balance of the user's account. ErrInsufficientFunds = errors.New("insufficient funds for gas * price + value") // ErrIntrinsicGas is returned if the transaction is specified to use less gas // than required to start the invocation. ErrIntrinsicGas = errors.New("intrinsic gas too low") // ErrGasLimit is returned if a transaction's requested gas limit exceeds the // maximum allowance of the current block. ErrGasLimit = errors.New("exceeds block gas limit") // ErrNegativeValue is a sanity error to ensure noone is able to specify a // transaction with a negative value. ErrNegativeValue = errors.New("negative value") // ErrOversizedData is returned if the input data of a transaction is greater // than some meaningful limit a user might use. This is not a consensus error // making the transaction invalid, rather a DOS protection. ErrOversizedData = errors.New("oversized data") ) var ( evictionInterval = time.Minute // Time interval to check for evictable transactions statsReportInterval = 8 * time.Second // Time interval to report transaction pool stats ) var ( // Metrics for the pending pool pendingDiscardCounter = metrics.NewRegisteredCounter("txpool/pending/discard", nil) pendingReplaceCounter = metrics.NewRegisteredCounter("txpool/pending/replace", nil) pendingRateLimitCounter = metrics.NewRegisteredCounter("txpool/pending/ratelimit", nil) // Dropped due to rate limiting pendingNofundsCounter = metrics.NewRegisteredCounter("txpool/pending/nofunds", nil) // Dropped due to out-of-funds // Metrics for the queued pool queuedDiscardCounter = metrics.NewRegisteredCounter("txpool/queued/discard", nil) queuedReplaceCounter = metrics.NewRegisteredCounter("txpool/queued/replace", nil) queuedRateLimitCounter = metrics.NewRegisteredCounter("txpool/queued/ratelimit", nil) // Dropped due to rate limiting queuedNofundsCounter = metrics.NewRegisteredCounter("txpool/queued/nofunds", nil) // Dropped due to out-of-funds // General tx metrics invalidTxCounter = metrics.NewRegisteredCounter("txpool/invalid", nil) underpricedTxCounter = metrics.NewRegisteredCounter("txpool/underpriced", nil) ) // TxStatus is the current status of a transaction as seen by the pool. type TxStatus uint // Constants for TxStatus. const ( TxStatusUnknown TxStatus = iota TxStatusQueued TxStatusPending TxStatusIncluded ) // blockChain provides the state of blockchain and current gas limit to do // some pre checks in tx pool and event subscribers. type blockChain interface { CurrentBlock() *types.Block GetBlock(hash common.Hash, number uint64) *types.Block StateAt(root common.Hash) (*state.DB, error) SubscribeChainHeadEvent(ch chan<- ChainHeadEvent) event.Subscription } // TxPoolConfig are the configuration parameters of the transaction pool. type TxPoolConfig struct { Locals []common.Address // Addresses that should be treated by default as local NoLocals bool // Whether local transaction handling should be disabled Journal string // Journal of local transactions to survive node restarts Rejournal time.Duration // Time interval to regenerate the local transaction journal PriceLimit uint64 // Minimum gas price to enforce for acceptance into the pool PriceBump uint64 // Minimum price bump percentage to replace an already existing transaction (nonce) AccountSlots uint64 // Number of executable transaction slots guaranteed per account GlobalSlots uint64 // Maximum number of executable transaction slots for all accounts AccountQueue uint64 // Maximum number of non-executable transaction slots permitted per account GlobalQueue uint64 // Maximum number of non-executable transaction slots for all accounts Lifetime time.Duration // Maximum amount of time non-executable transaction are queued } // DefaultTxPoolConfig contains the default configurations for the transaction // pool. var DefaultTxPoolConfig = TxPoolConfig{ Journal: "transactions.rlp", Rejournal: time.Hour, PriceLimit: 1, PriceBump: 10, AccountSlots: 16, GlobalSlots: 4096, AccountQueue: 64, GlobalQueue: 1024, Lifetime: 3 * time.Hour, } // sanitize checks the provided user configurations and changes anything that's // unreasonable or unworkable. func (config *TxPoolConfig) sanitize() TxPoolConfig { conf := *config if conf.Rejournal < time.Second { utils.GetLogger().Warn("Sanitizing invalid txpool journal time", "provided", conf.Rejournal, "updated", time.Second) conf.Rejournal = time.Second } if conf.PriceLimit < 1 { utils.GetLogger().Warn("Sanitizing invalid txpool price limit", "provided", conf.PriceLimit, "updated", DefaultTxPoolConfig.PriceLimit) conf.PriceLimit = DefaultTxPoolConfig.PriceLimit } if conf.PriceBump < 1 { utils.GetLogger().Warn("Sanitizing invalid txpool price bump", "provided", conf.PriceBump, "updated", DefaultTxPoolConfig.PriceBump) conf.PriceBump = DefaultTxPoolConfig.PriceBump } return conf } // 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 TxPoolConfig chainconfig *params.ChainConfig chain blockChain gasPrice *big.Int txFeed event.Feed scope event.SubscriptionScope chainHeadCh chan ChainHeadEvent chainHeadSub event.Subscription signer types.Signer mu sync.RWMutex currentState *state.DB // Current state in the blockchain head pendingState *state.ManagedState // Pending state tracking virtual nonces currentMaxGas uint64 // Current gas limit for transaction caps locals *accountSet // Set of local transaction to exempt from eviction rules journal *txJournal // Journal of local transaction to back up to disk pending map[common.Address]*txList // All currently processable transactions queue map[common.Address]*txList // Queued but non-processable transactions beats map[common.Address]time.Time // Last heartbeat from each known account all *txLookup // All transactions to allow lookups priced *txPricedList // All transactions sorted by price wg sync.WaitGroup // for shutdown sync homestead bool } // NewTxPool creates a new transaction pool to gather, sort and filter inbound // transactions from the network. func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain blockChain) *TxPool { // Sanitize the input to ensure no vulnerable gas prices are set config = (&config).sanitize() // Create the transaction pool with its initial settings pool := &TxPool{ config: config, chainconfig: chainconfig, chain: chain, signer: types.NewEIP155Signer(chainconfig.ChainID), pending: make(map[common.Address]*txList), queue: make(map[common.Address]*txList), beats: make(map[common.Address]time.Time), all: newTxLookup(), chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize), gasPrice: new(big.Int).SetUint64(config.PriceLimit), } pool.locals = newAccountSet(pool.signer) for _, addr := range config.Locals { utils.GetLogger().Info("Setting new local account", "address", addr) pool.locals.add(addr) } pool.priced = newTxPricedList(pool.all) pool.reset(nil, chain.CurrentBlock().Header()) // If local transactions and journaling is enabled, load from disk if !config.NoLocals && config.Journal != "" { pool.journal = newTxJournal(config.Journal) if err := pool.journal.load(pool.AddLocals); err != nil { utils.GetLogger().Warn("Failed to load transaction journal", "err", err) } if err := pool.journal.rotate(pool.local()); err != nil { utils.GetLogger().Warn("Failed to rotate transaction journal", "err", err) } } // Subscribe events from blockchain pool.chainHeadSub = pool.chain.SubscribeChainHeadEvent(pool.chainHeadCh) // Start the event loop and return pool.wg.Add(1) go pool.loop() return pool } // loop is the transaction pool's main event loop, waiting for and reacting to // outside blockchain events as well as for various reporting and transaction // eviction events. func (pool *TxPool) loop() { defer pool.wg.Done() // Start the stats reporting and transaction eviction tickers var prevPending, prevQueued, prevStales int report := time.NewTicker(statsReportInterval) defer report.Stop() evict := time.NewTicker(evictionInterval) defer evict.Stop() journal := time.NewTicker(pool.config.Rejournal) defer journal.Stop() // Track the previous head headers for transaction reorgs //head := pool.chain.CurrentBlock() // Keep waiting for and reacting to the various events for { select { // Handle ChainHeadEvent case ev := <-pool.chainHeadCh: if ev.Block != nil { pool.mu.Lock() //if pool.chainconfig.IsHomestead(ev.Block.Number()) { // pool.homestead = true //} //pool.reset(head.Header(), ev.Block.Header()) //head = ev.Block pool.mu.Unlock() } // Be unsubscribed due to system stopped case <-pool.chainHeadSub.Err(): return // Handle stats reporting ticks case <-report.C: pool.mu.RLock() pending, queued := pool.stats() stales := pool.priced.stales pool.mu.RUnlock() if pending != prevPending || queued != prevQueued || stales != prevStales { utils.GetLogger().Debug("Transaction pool status report", "executable", pending, "queued", queued, "stales", stales) prevPending, prevQueued, prevStales = pending, queued, stales } // Handle inactive account transaction eviction case <-evict.C: pool.mu.Lock() for addr := range pool.queue { // Skip local transactions from the eviction mechanism if pool.locals.contains(addr) { continue } // Any non-locals old enough should be removed if time.Since(pool.beats[addr]) > pool.config.Lifetime { for _, tx := range pool.queue[addr].Flatten() { pool.removeTx(tx.Hash(), true) } } } pool.mu.Unlock() // Handle local transaction journal rotation case <-journal.C: if pool.journal != nil { pool.mu.Lock() if err := pool.journal.rotate(pool.local()); err != nil { utils.GetLogger().Warn("Failed to rotate local tx journal", "err", err) } pool.mu.Unlock() } } } } // lockedReset is a wrapper around reset to allow calling it in a thread safe // manner. This method is only ever used in the tester! func (pool *TxPool) lockedReset(oldHead, newHead *types.Header) { pool.mu.Lock() defer pool.mu.Unlock() pool.reset(oldHead, newHead) } // reset retrieves the current state of the blockchain and ensures the content // of the transaction pool is valid with regard to the chain state. func (pool *TxPool) reset(oldHead, newHead *types.Header) { // If we're reorging an old state, reinject all dropped transactions var reinject types.Transactions if oldHead != nil && oldHead.Hash() != newHead.ParentHash { // If the reorg is too deep, avoid doing it (will happen during fast sync) oldNum := oldHead.Number.Uint64() newNum := newHead.Number.Uint64() if depth := uint64(math.Abs(float64(oldNum) - float64(newNum))); depth > 64 { utils.GetLogger().Debug("Skipping deep transaction reorg", "depth", depth) } else { // Reorg seems shallow enough to pull in all transactions into memory var discarded, included types.Transactions var ( rem = pool.chain.GetBlock(oldHead.Hash(), oldHead.Number.Uint64()) add = pool.chain.GetBlock(newHead.Hash(), newHead.Number.Uint64()) ) for rem.NumberU64() > add.NumberU64() { discarded = append(discarded, rem.Transactions()...) if rem = pool.chain.GetBlock(rem.ParentHash(), rem.NumberU64()-1); rem == nil { utils.GetLogger().Error("Unrooted old chain seen by tx pool", "block", oldHead.Number, "hash", oldHead.Hash()) return } } for add.NumberU64() > rem.NumberU64() { included = append(included, add.Transactions()...) if add = pool.chain.GetBlock(add.ParentHash(), add.NumberU64()-1); add == nil { utils.GetLogger().Error("Unrooted new chain seen by tx pool", "block", newHead.Number, "hash", newHead.Hash()) return } } for rem.Hash() != add.Hash() { discarded = append(discarded, rem.Transactions()...) if rem = pool.chain.GetBlock(rem.ParentHash(), rem.NumberU64()-1); rem == nil { utils.GetLogger().Error("Unrooted old chain seen by tx pool", "block", oldHead.Number, "hash", oldHead.Hash()) return } included = append(included, add.Transactions()...) if add = pool.chain.GetBlock(add.ParentHash(), add.NumberU64()-1); add == nil { utils.GetLogger().Error("Unrooted new chain seen by tx pool", "block", newHead.Number, "hash", newHead.Hash()) return } } reinject = types.TxDifference(discarded, included) } } // Initialize the internal state to the current head if newHead == nil { newHead = pool.chain.CurrentBlock().Header() // Special case during testing } statedb, err := pool.chain.StateAt(newHead.Root) if err != nil { utils.GetLogger().Error("Failed to reset txpool state", "err", err) return } pool.currentState = statedb pool.pendingState = state.ManageState(statedb) pool.currentMaxGas = newHead.GasLimit // Inject any transactions discarded due to reorgs utils.GetLogger().Debug("Reinjecting stale transactions", "count", len(reinject)) //senderCacher.recover(pool.signer, reinject) pool.addTxsLocked(reinject, false) // 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() // 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(nil) } // Stop terminates the transaction pool. func (pool *TxPool) Stop() { // Unsubscribe all subscriptions registered from txpool pool.scope.Close() // Unsubscribe subscriptions registered from blockchain pool.chainHeadSub.Unsubscribe() pool.wg.Wait() if pool.journal != nil { pool.journal.close() } utils.GetLogger().Info("Transaction pool stopped") } // SubscribeNewTxsEvent registers a subscription of NewTxsEvent and // starts sending event to the given channel. func (pool *TxPool) SubscribeNewTxsEvent(ch chan<- NewTxsEvent) event.Subscription { return pool.scope.Track(pool.txFeed.Subscribe(ch)) } // GasPrice returns the current gas price enforced by the transaction pool. func (pool *TxPool) GasPrice() *big.Int { pool.mu.RLock() defer pool.mu.RUnlock() return new(big.Int).Set(pool.gasPrice) } // SetGasPrice updates the minimum price required by the transaction pool for a // new transaction, and drops all transactions below this threshold. func (pool *TxPool) SetGasPrice(price *big.Int) { pool.mu.Lock() defer pool.mu.Unlock() pool.gasPrice = price for _, tx := range pool.priced.Cap(price, pool.locals) { pool.removeTx(tx.Hash(), false) } utils.GetLogger().Info("Transaction pool price threshold updated", "price", price) } // State returns the virtual managed state of the transaction pool. 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() (int, int) { pool.mu.RLock() defer pool.mu.RUnlock() return pool.stats() } // stats retrieves the current pool stats, namely the number of pending and the // number of queued (non-executable) transactions. func (pool *TxPool) stats() (int, int) { pending := 0 for _, list := range pool.pending { pending += list.Len() } queued := 0 for _, list := range pool.queue { queued += list.Len() } return pending, queued } // 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.Lock() defer pool.mu.Unlock() 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, grouped 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() pending := make(map[common.Address]types.Transactions) for addr, list := range pool.pending { pending[addr] = list.Flatten() } return pending, nil } // Locals retrieves the accounts currently considered local by the pool. func (pool *TxPool) Locals() []common.Address { pool.mu.Lock() defer pool.mu.Unlock() return pool.locals.flatten() } // local retrieves all currently known local transactions, grouped 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) local() map[common.Address]types.Transactions { txs := make(map[common.Address]types.Transactions) for addr := range pool.locals.accounts { if pending := pool.pending[addr]; pending != nil { txs[addr] = append(txs[addr], pending.Flatten()...) } if queued := pool.queue[addr]; queued != nil { txs[addr] = append(txs[addr], queued.Flatten()...) } } return txs } // validateTx checks whether a transaction is valid according to the consensus // rules and adheres to some heuristic limits of the local node (price and size). func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error { // Heuristic limit, reject transactions over 32KB to prevent DOS attacks if tx.Size() > 32*1024 { return ErrOversizedData } // Transactions can't be negative. This may never happen using RLP decoded // transactions but may occur if you create a transaction using the RPC. if tx.Value().Sign() < 0 { return ErrNegativeValue } // Ensure the transaction doesn't exceed the current block limit gas. if pool.currentMaxGas < tx.Gas() { return ErrGasLimit } // Make sure the transaction is signed properly from, err := types.Sender(pool.signer, tx) if err != nil { return ErrInvalidSender } // Drop non-local transactions under our own minimal accepted gas price local = local || pool.locals.contains(from) // account may be local even if the transaction arrived from the network if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 { return ErrUnderpriced } // Ensure the transaction adheres to nonce ordering if pool.currentState.GetNonce(from) > tx.Nonce() { return ErrNonceTooLow } // Transactor should have enough funds to cover the costs // cost == V + GP * GL if pool.currentState.GetBalance(from).Cmp(tx.Cost()) < 0 { return ErrInsufficientFunds } intrGas, err := IntrinsicGas(tx.Data(), tx.To() == nil, pool.homestead) if err != nil { return err } if tx.Gas() < intrGas { return ErrIntrinsicGas } return nil } // add validates a transaction and inserts it into the non-executable queue for // later pending promotion and execution. If the transaction is a replacement for // an already pending or queued one, it overwrites the previous and returns this // so outer code doesn't uselessly call promote. // // If a newly added transaction is marked as local, its sending account will be // whitelisted, preventing any associated transaction from being dropped out of // the pool due to pricing constraints. func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) { // If the transaction is already known, discard it hash := tx.Hash() if pool.all.Get(hash) != nil { utils.GetLogger().Trace("Discarding already known transaction", "hash", hash) return false, fmt.Errorf("known transaction: %x", hash) } // If the transaction fails basic validation, discard it if err := pool.validateTx(tx, local); err != nil { utils.GetLogger().Trace("Discarding invalid transaction", "hash", hash, "err", err) invalidTxCounter.Inc(1) return false, err } // If the transaction pool is full, discard underpriced transactions if uint64(pool.all.Count()) >= pool.config.GlobalSlots+pool.config.GlobalQueue { // If the new transaction is underpriced, don't accept it if !local && pool.priced.Underpriced(tx, pool.locals) { utils.GetLogger().Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice()) underpricedTxCounter.Inc(1) return false, ErrUnderpriced } // 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) for _, tx := range drop { utils.GetLogger().Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice()) underpricedTxCounter.Inc(1) pool.removeTx(tx.Hash(), false) } } // If the transaction is replacing an already pending one, do directly from, _ := types.Sender(pool.signer, tx) // already validated if list := pool.pending[from]; list != nil && list.Overlaps(tx) { // Nonce already pending, check if required price bump is met inserted, old := list.Add(tx, pool.config.PriceBump) if !inserted { pendingDiscardCounter.Inc(1) return false, ErrReplaceUnderpriced } // New transaction is better, replace old one if old != nil { pool.all.Remove(old.Hash()) pool.priced.Removed() pendingReplaceCounter.Inc(1) } pool.all.Add(tx) pool.priced.Put(tx) pool.journalTx(from, tx) utils.GetLogger().Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To()) // We've directly injected a replacement transaction, notify subsystems // go pool.txFeed.Send(NewTxsEvent{types.Transactions{tx}}) return old != nil, nil } // New transaction isn't replacing a pending one, push into queue replace, err := pool.enqueueTx(hash, tx) if err != nil { return false, err } // Mark local addresses and journal local transactions if local { if !pool.locals.contains(from) { utils.GetLogger().Info("Setting new local account", "address", from) pool.locals.add(from) } } pool.journalTx(from, tx) utils.GetLogger().Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To()) return replace, nil } // Add adds a transaction to the pool if valid and passes it to the tx relay // backend func (pool *TxPool) Add(ctx context.Context, tx *types.Transaction) error { // TODO(ricl): placeholder // TODO(minhdoan): follow with richard why we need this. As of now TxPool is not used now. 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) (bool, error) { // 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, pool.config.PriceBump) if !inserted { // An older transaction was better, discard this queuedDiscardCounter.Inc(1) return false, ErrReplaceUnderpriced } // Discard any previous transaction and mark this if old != nil { pool.all.Remove(old.Hash()) pool.priced.Removed() queuedReplaceCounter.Inc(1) } if pool.all.Get(hash) == nil { pool.all.Add(tx) pool.priced.Put(tx) } return old != nil, nil } // journalTx adds the specified transaction to the local disk journal if it is // deemed to have been sent from a local account. func (pool *TxPool) journalTx(from common.Address, tx *types.Transaction) { // Only journal if it's enabled and the transaction is local if pool.journal == nil || !pool.locals.contains(from) { return } if err := pool.journal.insert(tx); err != nil { utils.GetLogger().Warn("Failed to journal local transaction", "err", err) } } // promoteTx adds a transaction to the pending (processable) list of transactions // and returns whether it was inserted or an older was better. // // Note, this method assumes the pool lock is held! func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.Transaction) bool { // 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, pool.config.PriceBump) if !inserted { // An older transaction was better, discard this pool.all.Remove(hash) pool.priced.Removed() pendingDiscardCounter.Inc(1) return false } // Otherwise discard any previous transaction and mark this if old != nil { pool.all.Remove(old.Hash()) pool.priced.Removed() pendingReplaceCounter.Inc(1) } // Failsafe to work around direct pending inserts (tests) if pool.all.Get(hash) == nil { pool.all.Add(tx) pool.priced.Put(tx) } // 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) return true } // AddLocal enqueues a single transaction into the pool if it is valid, marking // the sender as a local one in the mean time, ensuring it goes around the local // pricing constraints. func (pool *TxPool) AddLocal(tx *types.Transaction) error { return pool.addTx(tx, !pool.config.NoLocals) } // AddRemote enqueues a single transaction into the pool if it is valid. If the // sender is not among the locally tracked ones, full pricing constraints will // apply. func (pool *TxPool) AddRemote(tx *types.Transaction) error { return pool.addTx(tx, false) } // AddLocals enqueues a batch of transactions into the pool if they are valid, // marking the senders as a local ones in the mean time, ensuring they go around // the local pricing constraints. func (pool *TxPool) AddLocals(txs []*types.Transaction) []error { return pool.addTxs(txs, !pool.config.NoLocals) } // AddRemotes enqueues a batch of transactions into the pool if they are valid. // If the senders are not among the locally tracked ones, full pricing constraints // will apply. func (pool *TxPool) AddRemotes(txs []*types.Transaction) []error { return pool.addTxs(txs, false) } // addTx enqueues a single transaction into the pool if it is valid. func (pool *TxPool) addTx(tx *types.Transaction, local bool) error { pool.mu.Lock() defer pool.mu.Unlock() // Try to inject the transaction and update any state replace, err := pool.add(tx, local) if err != nil { return err } // If we added a new transaction, run promotion checks and return if !replace { from, _ := types.Sender(pool.signer, tx) // already validated pool.promoteExecutables([]common.Address{from}) } return nil } // addTxs attempts to queue a batch of transactions if they are valid. func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) []error { pool.mu.Lock() defer pool.mu.Unlock() return pool.addTxsLocked(txs, local) } // addTxsLocked attempts to queue a batch of transactions if they are valid, // whilst assuming the transaction pool lock is already held. func (pool *TxPool) addTxsLocked(txs []*types.Transaction, local bool) []error { // Add the batch of transaction, tracking the accepted ones dirty := make(map[common.Address]struct{}) errs := make([]error, len(txs)) for i, tx := range txs { var replace bool if replace, errs[i] = pool.add(tx, local); errs[i] == nil && !replace { from, _ := types.Sender(pool.signer, tx) // already validated dirty[from] = struct{}{} } } // Only reprocess the internal state if something was actually added if len(dirty) > 0 { addrs := make([]common.Address, 0, len(dirty)) for addr := range dirty { addrs = append(addrs, addr) } pool.promoteExecutables(addrs) } return errs } // Status returns the status (unknown/pending/queued) of a batch of transactions // identified by their hashes. func (pool *TxPool) Status(hashes []common.Hash) []TxStatus { pool.mu.RLock() defer pool.mu.RUnlock() status := make([]TxStatus, len(hashes)) for i, hash := range hashes { if tx := pool.all.Get(hash); tx != nil { from, _ := types.Sender(pool.signer, tx) // already validated if pool.pending[from] != nil && pool.pending[from].txs.items[tx.Nonce()] != nil { status[i] = TxStatusPending } else { status[i] = TxStatusQueued } } } return status } // Get returns a transaction if it is contained in the pool // and nil otherwise. func (pool *TxPool) Get(hash common.Hash) *types.Transaction { return pool.all.Get(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, outofbound bool) { // Fetch the transaction we wish to delete tx := pool.all.Get(hash) if tx == nil { return } addr, _ := types.Sender(pool.signer, tx) // already validated during insertion // Remove it from the list of known transactions pool.all.Remove(hash) if outofbound { pool.priced.Removed() } // 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 pending transactions are left, remove the list if pending.Empty() { delete(pool.pending, addr) delete(pool.beats, addr) } // 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, nonce) } return } } // 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(accounts []common.Address) { // Track the promoted transactions to broadcast them at once var promoted []*types.Transaction // Gather all the accounts potentially needing updates if accounts == nil { accounts = make([]common.Address, 0, len(pool.queue)) for addr := range pool.queue { accounts = append(accounts, addr) } } // Iterate over all accounts and promote any executable transactions for _, addr := range accounts { list := pool.queue[addr] if list == nil { continue // Just in case someone calls with a non existing account } // Drop all transactions that are deemed too old (low nonce) for _, tx := range list.Forward(pool.currentState.GetNonce(addr)) { hash := tx.Hash() utils.GetLogger().Trace("Removed old queued transaction", "hash", hash) pool.all.Remove(hash) pool.priced.Removed() } // Drop all transactions that are too costly (low balance or out of gas) drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas) for _, tx := range drops { hash := tx.Hash() utils.GetLogger().Trace("Removed unpayable queued transaction", "hash", hash) pool.all.Remove(hash) pool.priced.Removed() queuedNofundsCounter.Inc(1) } // Gather all executable transactions and promote them for _, tx := range list.Ready(pool.pendingState.GetNonce(addr)) { hash := tx.Hash() if pool.promoteTx(addr, hash, tx) { utils.GetLogger().Trace("Promoting queued transaction", "hash", hash) promoted = append(promoted, tx) } } // Drop all transactions over the allowed limit if !pool.locals.contains(addr) { for _, tx := range list.Cap(int(pool.config.AccountQueue)) { hash := tx.Hash() pool.all.Remove(hash) pool.priced.Removed() queuedRateLimitCounter.Inc(1) utils.GetLogger().Trace("Removed cap-exceeding queued transaction", "hash", hash) } } // Delete the entire queue entry if it became empty. if list.Empty() { delete(pool.queue, addr) } } // Notify subsystem for new promoted transactions. //if len(promoted) > 0 { // go pool.txFeed.Send(NewTxsEvent{promoted}) //} // If the pending limit is overflown, start equalizing allowances pending := uint64(0) for _, list := range pool.pending { pending += uint64(list.Len()) } if pending > pool.config.GlobalSlots { pendingBeforeCap := pending // Assemble a spam order to penalize large transactors first spammers := prque.New(nil) for addr, list := range pool.pending { // Only evict transactions from high rollers if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots { spammers.Push(addr, int64(list.Len())) } } // Gradually drop transactions from offenders offenders := []common.Address{} for pending > pool.config.GlobalSlots && !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 > pool.config.GlobalSlots && pool.pending[offenders[len(offenders)-2]].Len() > threshold { for i := 0; i < len(offenders)-1; i++ { list := pool.pending[offenders[i]] for _, tx := range list.Cap(list.Len() - 1) { // Drop the transaction from the global pools too hash := tx.Hash() pool.all.Remove(hash) pool.priced.Removed() // Update the account nonce to the dropped transaction if nonce := tx.Nonce(); pool.pendingState.GetNonce(offenders[i]) > nonce { pool.pendingState.SetNonce(offenders[i], nonce) } utils.GetLogger().Trace("Removed fairness-exceeding pending transaction", "hash", hash) } pending-- } } } } // If still above threshold, reduce to limit or min allowance if pending > pool.config.GlobalSlots && len(offenders) > 0 { for pending > pool.config.GlobalSlots && uint64(pool.pending[offenders[len(offenders)-1]].Len()) > pool.config.AccountSlots { for _, addr := range offenders { list := pool.pending[addr] for _, tx := range list.Cap(list.Len() - 1) { // Drop the transaction from the global pools too hash := tx.Hash() pool.all.Remove(hash) pool.priced.Removed() // Update the account nonce to the dropped transaction if nonce := tx.Nonce(); pool.pendingState.GetNonce(addr) > nonce { pool.pendingState.SetNonce(addr, nonce) } utils.GetLogger().Trace("Removed fairness-exceeding pending transaction", "hash", hash) } pending-- } } } pendingRateLimitCounter.Inc(int64(pendingBeforeCap - pending)) } // If we've queued more transactions than the hard limit, drop oldest ones queued := uint64(0) for _, list := range pool.queue { queued += uint64(list.Len()) } if queued > pool.config.GlobalQueue { // Sort all accounts with queued transactions by heartbeat addresses := make(addressesByHeartbeat, 0, len(pool.queue)) for addr := range pool.queue { if !pool.locals.contains(addr) { // don't drop locals addresses = append(addresses, addressByHeartbeat{addr, pool.beats[addr]}) } } sort.Sort(addresses) // Drop transactions until the total is below the limit or only locals remain for drop := queued - pool.config.GlobalQueue; drop > 0 && len(addresses) > 0; { 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(), true) } drop -= size queuedRateLimitCounter.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(), true) drop-- queuedRateLimitCounter.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() { // Iterate over all accounts and demote any non-executable transactions for addr, list := range pool.pending { nonce := pool.currentState.GetNonce(addr) // Drop all transactions that are deemed too old (low nonce) for _, tx := range list.Forward(nonce) { hash := tx.Hash() utils.GetLogger().Trace("Removed old pending transaction", "hash", hash) pool.all.Remove(hash) pool.priced.Removed() } // Drop all transactions that are too costly (low balance or out of gas), and queue any invalids back for later drops, invalids := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas) for _, tx := range drops { hash := tx.Hash() utils.GetLogger().Trace("Removed unpayable pending transaction", "hash", hash) pool.all.Remove(hash) pool.priced.Removed() pendingNofundsCounter.Inc(1) } for _, tx := range invalids { hash := tx.Hash() utils.GetLogger().Trace("Demoting pending transaction", "hash", hash) pool.enqueueTx(hash, tx) } // If there's a gap in front, alert (should never happen) and postpone all transactions if list.Len() > 0 && list.txs.Get(nonce) == nil { for _, tx := range list.Cap(0) { hash := tx.Hash() utils.GetLogger().Error("Demoting invalidated 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) } } } // addressByHeartbeat is an account address tagged with its last activity timestamp. type addressByHeartbeat struct { address common.Address heartbeat time.Time } type addressesByHeartbeat []addressByHeartbeat func (a addressesByHeartbeat) Len() int { return len(a) } func (a addressesByHeartbeat) Less(i, j int) bool { return a[i].heartbeat.Before(a[j].heartbeat) } func (a addressesByHeartbeat) Swap(i, j int) { a[i], a[j] = a[j], a[i] } // accountSet is simply a set of addresses to check for existence, and a signer // capable of deriving addresses from transactions. type accountSet struct { accounts map[common.Address]struct{} signer types.Signer cache *[]common.Address } // newAccountSet creates a new address set with an associated signer for sender // derivations. func newAccountSet(signer types.Signer) *accountSet { return &accountSet{ accounts: make(map[common.Address]struct{}), signer: signer, } } // contains checks if a given address is contained within the set. func (as *accountSet) contains(addr common.Address) bool { _, exist := as.accounts[addr] return exist } // containsTx checks if the sender of a given tx is within the set. If the sender // cannot be derived, this method returns false. func (as *accountSet) containsTx(tx *types.Transaction) bool { if addr, err := types.Sender(as.signer, tx); err == nil { return as.contains(addr) } return false } // add inserts a new address into the set to track. func (as *accountSet) add(addr common.Address) { as.accounts[addr] = struct{}{} as.cache = nil } // flatten returns the list of addresses within this set, also caching it for later // reuse. The returned slice should not be changed! func (as *accountSet) flatten() []common.Address { if as.cache == nil { accounts := make([]common.Address, 0, len(as.accounts)) for account := range as.accounts { accounts = append(accounts, account) } as.cache = &accounts } return *as.cache } // txLookup is used internally by TxPool to track transactions while allowing lookup without // mutex contention. // // Note, although this type is properly protected against concurrent access, it // is **not** a type that should ever be mutated or even exposed outside of the // transaction pool, since its internal state is tightly coupled with the pools // internal mechanisms. The sole purpose of the type is to permit out-of-bound // peeking into the pool in TxPool.Get without having to acquire the widely scoped // TxPool.mu mutex. type txLookup struct { all map[common.Hash]*types.Transaction lock sync.RWMutex } // newTxLookup returns a new txLookup structure. func newTxLookup() *txLookup { return &txLookup{ all: make(map[common.Hash]*types.Transaction), } } // Range calls f on each key and value present in the map. func (t *txLookup) Range(f func(hash common.Hash, tx *types.Transaction) bool) { t.lock.RLock() defer t.lock.RUnlock() for key, value := range t.all { if !f(key, value) { break } } } // Get returns a transaction if it exists in the lookup, or nil if not found. func (t *txLookup) Get(hash common.Hash) *types.Transaction { t.lock.RLock() defer t.lock.RUnlock() return t.all[hash] } // Count returns the current number of items in the lookup. func (t *txLookup) Count() int { t.lock.RLock() defer t.lock.RUnlock() return len(t.all) } // Add adds a transaction to the lookup. func (t *txLookup) Add(tx *types.Transaction) { t.lock.Lock() defer t.lock.Unlock() t.all[tx.Hash()] = tx } // Remove removes a transaction from the lookup. func (t *txLookup) Remove(hash common.Hash) { t.lock.Lock() defer t.lock.Unlock() delete(t.all, hash) }