1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237 |
- // 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"
- "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"
- )
- 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
- 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.StateDB, error)
- SubscribeChainHeadEvent(ch chan<- ChainHeadEvent) event.Subscription
- }
- // TxPoolConfig are the configuration parameters of the transaction pool.
- type TxPoolConfig struct {
- 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 // Minimum 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 {
- log.Warn("Sanitizing invalid txpool journal time", "provided", conf.Rejournal, "updated", time.Second)
- conf.Rejournal = time.Second
- }
- if conf.PriceLimit < 1 {
- log.Warn("Sanitizing invalid txpool price limit", "provided", conf.PriceLimit, "updated", DefaultTxPoolConfig.PriceLimit)
- conf.PriceLimit = DefaultTxPoolConfig.PriceLimit
- }
- if conf.PriceBump < 1 {
- log.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.StateDB // 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)
- 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 {
- log.Warn("Failed to load transaction journal", "err", err)
- }
- if err := pool.journal.rotate(pool.local()); err != nil {
- log.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 {
- log.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 {
- log.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 {
- log.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 {
- log.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 {
- log.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 {
- log.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 {
- log.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 {
- log.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
- log.Debug("Reinjecting stale transactions", "count", len(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()
- }
- log.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)
- }
- log.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, 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()
- pending := make(map[common.Address]types.Transactions)
- for addr, list := range pool.pending {
- pending[addr] = list.Flatten()
- }
- return pending, nil
- }
- // local retrieves all currently known local 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) 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 {
- log.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 {
- log.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) {
- log.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 {
- log.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)
- log.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 {
- pool.locals.add(from)
- }
- pool.journalTx(from, tx)
- log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
- return replace, 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 {
- log.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 {
- if !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()
- log.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()
- log.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) {
- log.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)
- log.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()
- 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, float32(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)
- }
- log.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)
- }
- log.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(addresssByHeartbeat, 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()
- log.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()
- log.Trace("Removed unpayable pending transaction", "hash", hash)
- pool.all.Remove(hash)
- pool.priced.Removed()
- pendingNofundsCounter.Inc(1)
- }
- for _, tx := range invalids {
- hash := tx.Hash()
- log.Trace("Demoting pending transaction", "hash", hash)
- pool.enqueueTx(hash, tx)
- }
- // If there's a gap in front, warn (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()
- log.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 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] }
- // 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
- }
- // 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{}{}
- }
- // 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)
- }
|