123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785 |
- // 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"
- "math"
- "math/big"
- "sort"
- "sync"
- "time"
- "github.com/ethereum/go-ethereum/common"
- "github.com/ethereum/go-ethereum/common/prque"
- "github.com/ethereum/go-ethereum/core/mps"
- "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"
- pcore "github.com/ethereum/go-ethereum/permission/core"
- "github.com/ethereum/go-ethereum/private"
- )
- const (
- // chainHeadChanSize is the size of channel listening to ChainHeadEvent.
- chainHeadChanSize = 10
- // txSlotSize is used to calculate how many data slots a single transaction
- // takes up based on its size. The slots are used as DoS protection, ensuring
- // that validating a new transaction remains a constant operation (in reality
- // O(maxslots), where max slots are 4 currently).
- txSlotSize = 32 * 1024
- // txMaxSize is the maximum size a single transaction can have. This field has
- // non-trivial consequences: larger transactions are significantly harder and
- // more expensive to propagate; larger transactions also take more resources
- // to validate whether they fit into the pool or not.
- // txMaxSize = 4 * txSlotSize // 128KB
- // Quorum - value above is not used. instead, ChainConfig.TransactionSizeLimit is used
- )
- var (
- // ErrAlreadyKnown is returned if the transactions is already contained
- // within the pool.
- ErrAlreadyKnown = errors.New("already known")
- // ErrInvalidSender is returned if the transaction contains an invalid signature.
- ErrInvalidSender = errors.New("invalid sender")
- // ErrUnderpriced is returned if a transaction's gas price is below the minimum
- // configured for the transaction pool.
- ErrUnderpriced = errors.New("transaction underpriced")
- // ErrTxPoolOverflow is returned if the transaction pool is full and can't accpet
- // another remote transaction.
- ErrTxPoolOverflow = errors.New("txpool is full")
- // 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")
- // 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 no one 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")
- // ErrInvalidGasPrice is returned if gas price is disabled, but a gas price
- // is specified in the transaction
- ErrInvalidGasPrice = errors.New("Gas price not 0")
- // ErrEtherValueUnsupported is returned if a transaction specifies an Ether Value
- // for a private Quorum transaction.
- ErrEtherValueUnsupported = errors.New("ether value is not supported for private transactions")
- )
- 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
- pendingDiscardMeter = metrics.NewRegisteredMeter("txpool/pending/discard", nil)
- pendingReplaceMeter = metrics.NewRegisteredMeter("txpool/pending/replace", nil)
- pendingRateLimitMeter = metrics.NewRegisteredMeter("txpool/pending/ratelimit", nil) // Dropped due to rate limiting
- pendingNofundsMeter = metrics.NewRegisteredMeter("txpool/pending/nofunds", nil) // Dropped due to out-of-funds
- // Metrics for the queued pool
- queuedDiscardMeter = metrics.NewRegisteredMeter("txpool/queued/discard", nil)
- queuedReplaceMeter = metrics.NewRegisteredMeter("txpool/queued/replace", nil)
- queuedRateLimitMeter = metrics.NewRegisteredMeter("txpool/queued/ratelimit", nil) // Dropped due to rate limiting
- queuedNofundsMeter = metrics.NewRegisteredMeter("txpool/queued/nofunds", nil) // Dropped due to out-of-funds
- queuedEvictionMeter = metrics.NewRegisteredMeter("txpool/queued/eviction", nil) // Dropped due to lifetime
- // General tx metrics
- knownTxMeter = metrics.NewRegisteredMeter("txpool/known", nil)
- validTxMeter = metrics.NewRegisteredMeter("txpool/valid", nil)
- invalidTxMeter = metrics.NewRegisteredMeter("txpool/invalid", nil)
- underpricedTxMeter = metrics.NewRegisteredMeter("txpool/underpriced", nil)
- overflowedTxMeter = metrics.NewRegisteredMeter("txpool/overflowed", nil)
- pendingGauge = metrics.NewRegisteredGauge("txpool/pending", nil)
- queuedGauge = metrics.NewRegisteredGauge("txpool/queued", nil)
- localGauge = metrics.NewRegisteredGauge("txpool/local", nil)
- slotsGauge = metrics.NewRegisteredGauge("txpool/slots", 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, mps.PrivateStateRepository, 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 {
- 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
- }
- if conf.AccountSlots < 1 {
- log.Warn("Sanitizing invalid txpool account slots", "provided", conf.AccountSlots, "updated", DefaultTxPoolConfig.AccountSlots)
- conf.AccountSlots = DefaultTxPoolConfig.AccountSlots
- }
- if conf.GlobalSlots < 1 {
- log.Warn("Sanitizing invalid txpool global slots", "provided", conf.GlobalSlots, "updated", DefaultTxPoolConfig.GlobalSlots)
- conf.GlobalSlots = DefaultTxPoolConfig.GlobalSlots
- }
- if conf.AccountQueue < 1 {
- log.Warn("Sanitizing invalid txpool account queue", "provided", conf.AccountQueue, "updated", DefaultTxPoolConfig.AccountQueue)
- conf.AccountQueue = DefaultTxPoolConfig.AccountQueue
- }
- if conf.GlobalQueue < 1 {
- log.Warn("Sanitizing invalid txpool global queue", "provided", conf.GlobalQueue, "updated", DefaultTxPoolConfig.GlobalQueue)
- conf.GlobalQueue = DefaultTxPoolConfig.GlobalQueue
- }
- if conf.Lifetime < 1 {
- log.Warn("Sanitizing invalid txpool lifetime", "provided", conf.Lifetime, "updated", DefaultTxPoolConfig.Lifetime)
- conf.Lifetime = DefaultTxPoolConfig.Lifetime
- }
- 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
- signer types.Signer
- mu sync.RWMutex
- istanbul bool // Fork indicator whether we are in the istanbul stage.
- eip2718 bool // Fork indicator whether we are using EIP-2718 type transactions.
- currentState *state.StateDB // Current state in the blockchain head
- pendingNonces *txNoncer // 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
- chainHeadCh chan ChainHeadEvent
- chainHeadSub event.Subscription
- reqResetCh chan *txpoolResetRequest
- reqPromoteCh chan *accountSet
- queueTxEventCh chan *types.Transaction
- reorgDoneCh chan chan struct{}
- reorgShutdownCh chan struct{} // requests shutdown of scheduleReorgLoop
- wg sync.WaitGroup // tracks loop, scheduleReorgLoop
- }
- type txpoolResetRequest struct {
- oldHead, newHead *types.Header
- }
- // 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.LatestSigner(chainconfig),
- 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),
- reqResetCh: make(chan *txpoolResetRequest),
- reqPromoteCh: make(chan *accountSet),
- queueTxEventCh: make(chan *types.Transaction),
- reorgDoneCh: make(chan chan struct{}),
- reorgShutdownCh: make(chan struct{}),
- gasPrice: new(big.Int).SetUint64(config.PriceLimit),
- }
- pool.locals = newAccountSet(pool.signer)
- for _, addr := range config.Locals {
- log.Info("Setting new local account", "address", addr)
- pool.locals.add(addr)
- }
- pool.priced = newTxPricedList(pool.all)
- pool.reset(nil, chain.CurrentBlock().Header())
- // Start the reorg loop early so it can handle requests generated during journal loading.
- pool.wg.Add(1)
- go pool.scheduleReorgLoop()
- // 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 and start the main event loop.
- pool.chainHeadSub = pool.chain.SubscribeChainHeadEvent(pool.chainHeadCh)
- 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()
- var (
- prevPending, prevQueued, prevStales int
- // Start the stats reporting and transaction eviction tickers
- report = time.NewTicker(statsReportInterval)
- evict = time.NewTicker(evictionInterval)
- journal = time.NewTicker(pool.config.Rejournal)
- // Track the previous head headers for transaction reorgs
- head = pool.chain.CurrentBlock()
- )
- defer report.Stop()
- defer evict.Stop()
- defer journal.Stop()
- for {
- select {
- // Handle ChainHeadEvent
- case ev := <-pool.chainHeadCh:
- if ev.Block != nil {
- pool.requestReset(head.Header(), ev.Block.Header())
- head = ev.Block
- }
- // System shutdown.
- case <-pool.chainHeadSub.Err():
- close(pool.reorgShutdownCh)
- 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 {
- list := pool.queue[addr].Flatten()
- for _, tx := range list {
- pool.removeTx(tx.Hash(), true)
- }
- queuedEvictionMeter.Mark(int64(len(list)))
- }
- }
- 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()
- }
- }
- }
- }
- // 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) {
- //Quorum
- if pool.chainconfig.IsQuorum && !pool.chainconfig.IsGasPriceEnabled(pool.chain.CurrentBlock().Header().Number) {
- log.Info("Transaction pool price threshold not updated as gasPrice is not enabled")
- return
- }
- //End-Quorum
- pool.mu.Lock()
- defer pool.mu.Unlock()
- pool.gasPrice = price
- for _, tx := range pool.priced.Cap(price) {
- pool.removeTx(tx.Hash(), false)
- }
- log.Info("Transaction pool price threshold updated", "price", price)
- }
- // Nonce returns the next nonce of an account, with all transactions executable
- // by the pool already applied on top.
- func (pool *TxPool) Nonce(addr common.Address) uint64 {
- pool.mu.RLock()
- defer pool.mu.RUnlock()
- return pool.pendingNonces.get(addr)
- }
- // 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 {
- // Accept only legacy transactions until EIP-2718/2930 activates.
- if !pool.eip2718 && tx.Type() != types.LegacyTxType {
- return ErrTxTypeNotSupported
- }
- // Quorum
- sizeLimit := pool.chainconfig.GetTransactionSizeLimit(pool.chain.CurrentBlock().Number())
- // Reject transactions over defined size to prevent DOS attacks
- if float64(tx.Size()) > float64(sizeLimit*1024) {
- return ErrOversizedData
- }
- // End Quorum
- // 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
- }
- if pool.chainconfig.IsQuorum {
- // Quorum
- if tx.IsPrivacyMarker() {
- innerTx, _, _, _ := private.FetchPrivateTransaction(tx.Data())
- if innerTx != nil {
- if err := pool.validateTx(innerTx, local); err != nil {
- return err
- }
- }
- }
- // Quorum
- // Reject transaction if gas price is disabled, but gas price is specified
- if (!pool.chainconfig.IsGasPriceEnabled(pool.chain.CurrentBlock().Header().Number)) && tx.GasPriceIntCmp(common.Big0) != 0 {
- return ErrInvalidGasPrice
- }
- // Ether value is not currently supported on private transactions
- if tx.IsPrivate() && (len(tx.Data()) == 0 || tx.Value().Sign() != 0) {
- return ErrEtherValueUnsupported
- }
- // Quorum - check if the sender account is authorized to perform the transaction
- if err := pcore.CheckAccountPermission(tx.From(), tx.To(), tx.Value(), tx.Data(), tx.Gas(), tx.GasPrice()); err != nil {
- return err
- }
- }
- if !pool.chainconfig.IsQuorum || pool.chainconfig.IsGasPriceEnabled(pool.chain.CurrentBlock().Header().Number) {
- // Drop non-local transactions under our own minimal accepted gas price
- local = local || pool.locals.contains(from)
- if !local && tx.GasPriceIntCmp(pool.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
- }
- // Ensure the transaction has more gas than the basic tx fee.
- intrGas, err := IntrinsicGas(tx.Data(), tx.AccessList(), tx.To() == nil, true, pool.istanbul)
- 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 transaction if its price is higher.
- //
- // 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) (replaced bool, err 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)
- knownTxMeter.Mark(1)
- return false, ErrAlreadyKnown
- }
- // Make the local flag. If it's from local source or it's from the network but
- // the sender is marked as local previously, treat it as the local transaction.
- isLocal := local || pool.locals.containsTx(tx)
- // If the transaction fails basic validation, discard it
- if err := pool.validateTx(tx, isLocal); err != nil {
- log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
- invalidTxMeter.Mark(1)
- return false, err
- }
- // If the transaction pool is full, discard underpriced transactions
- if uint64(pool.all.Count()+numSlots(tx)) > pool.config.GlobalSlots+pool.config.GlobalQueue {
- // If the new transaction is underpriced, don't accept it
- if (!pool.chainconfig.IsQuorum || pool.chainconfig.IsGasPriceEnabled(pool.chain.CurrentBlock().Header().Number)) && !isLocal && pool.priced.Underpriced(tx) {
- log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
- underpricedTxMeter.Mark(1)
- return false, ErrUnderpriced
- }
- // New transaction is better than our worse ones, make room for it.
- // If it's a local transaction, forcibly discard all available transactions.
- // Otherwise if we can't make enough room for new one, abort the operation.
- drop, success := pool.priced.Discard(pool.all.Slots()-int(pool.config.GlobalSlots+pool.config.GlobalQueue)+numSlots(tx), isLocal)
- // Special case, we still can't make the room for the new remote one.
- if !isLocal && !success {
- log.Trace("Discarding overflown transaction", "hash", hash)
- overflowedTxMeter.Mark(1)
- return false, ErrTxPoolOverflow
- }
- // Kick out the underpriced remote transactions.
- for _, tx := range drop {
- log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice())
- underpricedTxMeter.Mark(1)
- pool.removeTx(tx.Hash(), false)
- }
- }
- // Try to replace an existing transaction in the pending pool
- 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 {
- pendingDiscardMeter.Mark(1)
- return false, ErrReplaceUnderpriced
- }
- // New transaction is better, replace old one
- if old != nil {
- pool.all.Remove(old.Hash())
- pool.priced.Removed(1)
- pendingReplaceMeter.Mark(1)
- }
- pool.all.Add(tx, isLocal)
- pool.priced.Put(tx, isLocal)
- pool.journalTx(from, tx)
- pool.queueTxEvent(tx)
- log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())
- // Successful promotion, bump the heartbeat
- pool.beats[from] = time.Now()
- return old != nil, nil
- }
- // New transaction isn't replacing a pending one, push into queue
- replaced, err = pool.enqueueTx(hash, tx, isLocal, true)
- if err != nil {
- return false, err
- }
- // Mark local addresses and journal local transactions
- if local && !pool.locals.contains(from) {
- log.Info("Setting new local account", "address", from)
- pool.locals.add(from)
- pool.priced.Removed(pool.all.RemoteToLocals(pool.locals)) // Migrate the remotes if it's marked as local first time.
- }
- if isLocal {
- localGauge.Inc(1)
- }
- pool.journalTx(from, tx)
- log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
- return replaced, 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, local bool, addAll bool) (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
- queuedDiscardMeter.Mark(1)
- return false, ErrReplaceUnderpriced
- }
- // Discard any previous transaction and mark this
- if old != nil {
- pool.all.Remove(old.Hash())
- pool.priced.Removed(1)
- queuedReplaceMeter.Mark(1)
- } else {
- // Nothing was replaced, bump the queued counter
- queuedGauge.Inc(1)
- }
- // If the transaction isn't in lookup set but it's expected to be there,
- // show the error log.
- if pool.all.Get(hash) == nil && !addAll {
- log.Error("Missing transaction in lookup set, please report the issue", "hash", hash)
- }
- if addAll {
- pool.all.Add(tx, local)
- pool.priced.Put(tx, local)
- }
- // If we never record the heartbeat, do it right now.
- if _, exist := pool.beats[from]; !exist {
- pool.beats[from] = time.Now()
- }
- 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(1)
- pendingDiscardMeter.Mark(1)
- return false
- }
- // Otherwise discard any previous transaction and mark this
- if old != nil {
- pool.all.Remove(old.Hash())
- pool.priced.Removed(1)
- pendingReplaceMeter.Mark(1)
- } else {
- // Nothing was replaced, bump the pending counter
- pendingGauge.Inc(1)
- }
- // Set the potentially new pending nonce and notify any subsystems of the new tx
- pool.pendingNonces.set(addr, tx.Nonce()+1)
- // Successful promotion, bump the heartbeat
- pool.beats[addr] = time.Now()
- return true
- }
- // AddLocals enqueues a batch of transactions into the pool if they are valid, marking the
- // senders as a local ones, ensuring they go around the local pricing constraints.
- //
- // This method is used to add transactions from the RPC API and performs synchronous pool
- // reorganization and event propagation.
- func (pool *TxPool) AddLocals(txs []*types.Transaction) []error {
- return pool.addTxs(txs, !pool.config.NoLocals, true)
- }
- // AddLocal enqueues a single local transaction into the pool if it is valid. This is
- // a convenience wrapper aroundd AddLocals.
- func (pool *TxPool) AddLocal(tx *types.Transaction) error {
- errs := pool.AddLocals([]*types.Transaction{tx})
- return errs[0]
- }
- // 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.
- //
- // This method is used to add transactions from the p2p network and does not wait for pool
- // reorganization and internal event propagation.
- func (pool *TxPool) AddRemotes(txs []*types.Transaction) []error {
- return pool.addTxs(txs, false, false)
- }
- // This is like AddRemotes, but waits for pool reorganization. Tests use this method.
- func (pool *TxPool) AddRemotesSync(txs []*types.Transaction) []error {
- return pool.addTxs(txs, false, true)
- }
- // This is like AddRemotes with a single transaction, but waits for pool reorganization. Tests use this method.
- func (pool *TxPool) addRemoteSync(tx *types.Transaction) error {
- errs := pool.AddRemotesSync([]*types.Transaction{tx})
- return errs[0]
- }
- // AddRemote enqueues a single transaction into the pool if it is valid. This is a convenience
- // wrapper around AddRemotes.
- //
- // Deprecated: use AddRemotes
- func (pool *TxPool) AddRemote(tx *types.Transaction) error {
- errs := pool.AddRemotes([]*types.Transaction{tx})
- return errs[0]
- }
- // addTxs attempts to queue a batch of transactions if they are valid.
- func (pool *TxPool) addTxs(txs []*types.Transaction, local, sync bool) []error {
- // Filter out known ones without obtaining the pool lock or recovering signatures
- var (
- errs = make([]error, len(txs))
- news = make([]*types.Transaction, 0, len(txs))
- )
- for i, tx := range txs {
- // If the transaction is known, pre-set the error slot
- if pool.all.Get(tx.Hash()) != nil {
- errs[i] = ErrAlreadyKnown
- knownTxMeter.Mark(1)
- continue
- }
- // Exclude transactions with invalid signatures as soon as
- // possible and cache senders in transactions before
- // obtaining lock
- _, err := types.Sender(pool.signer, tx)
- if err != nil {
- errs[i] = ErrInvalidSender
- invalidTxMeter.Mark(1)
- continue
- }
- // Accumulate all unknown transactions for deeper processing
- news = append(news, tx)
- }
- if len(news) == 0 {
- return errs
- }
- // Process all the new transaction and merge any errors into the original slice
- pool.mu.Lock()
- newErrs, dirtyAddrs := pool.addTxsLocked(news, local)
- pool.mu.Unlock()
- var nilSlot = 0
- for _, err := range newErrs {
- for errs[nilSlot] != nil {
- nilSlot++
- }
- errs[nilSlot] = err
- nilSlot++
- }
- // Reorg the pool internals if needed and return
- done := pool.requestPromoteExecutables(dirtyAddrs)
- if sync {
- <-done
- }
- return errs
- }
- // addTxsLocked attempts to queue a batch of transactions if they are valid.
- // The transaction pool lock must be held.
- func (pool *TxPool) addTxsLocked(txs []*types.Transaction, local bool) ([]error, *accountSet) {
- dirty := newAccountSet(pool.signer)
- errs := make([]error, len(txs))
- for i, tx := range txs {
- replaced, err := pool.add(tx, local)
- errs[i] = err
- if err == nil && !replaced {
- dirty.addTx(tx)
- }
- }
- validTxMeter.Mark(int64(len(dirty.accounts)))
- return errs, dirty
- }
- // Status returns the status (unknown/pending/queued) of a batch of transactions
- // identified by their hashes.
- func (pool *TxPool) Status(hashes []common.Hash) []TxStatus {
- status := make([]TxStatus, len(hashes))
- for i, hash := range hashes {
- tx := pool.Get(hash)
- if tx == nil {
- continue
- }
- from, _ := types.Sender(pool.signer, tx) // already validated
- pool.mu.RLock()
- if txList := pool.pending[from]; txList != nil && txList.txs.items[tx.Nonce()] != nil {
- status[i] = TxStatusPending
- } else if txList := pool.queue[from]; txList != nil && txList.txs.items[tx.Nonce()] != nil {
- status[i] = TxStatusQueued
- }
- // implicit else: the tx may have been included into a block between
- // checking pool.Get and obtaining the lock. In that case, TxStatusUnknown is correct
- pool.mu.RUnlock()
- }
- 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)
- }
- // Has returns an indicator whether txpool has a transaction cached with the
- // given hash.
- func (pool *TxPool) Has(hash common.Hash) bool {
- return pool.all.Get(hash) != nil
- }
- // 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(1)
- }
- if pool.locals.contains(addr) {
- localGauge.Dec(1)
- }
- // 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)
- }
- // Postpone any invalidated transactions
- for _, tx := range invalids {
- // Internal shuffle shouldn't touch the lookup set.
- pool.enqueueTx(tx.Hash(), tx, false, false)
- }
- // Update the account nonce if needed
- pool.pendingNonces.setIfLower(addr, tx.Nonce())
- // Reduce the pending counter
- pendingGauge.Dec(int64(1 + len(invalids)))
- return
- }
- }
- // Transaction is in the future queue
- if future := pool.queue[addr]; future != nil {
- if removed, _ := future.Remove(tx); removed {
- // Reduce the queued counter
- queuedGauge.Dec(1)
- }
- if future.Empty() {
- delete(pool.queue, addr)
- delete(pool.beats, addr)
- }
- }
- }
- // requestReset requests a pool reset to the new head block.
- // The returned channel is closed when the reset has occurred.
- func (pool *TxPool) requestReset(oldHead *types.Header, newHead *types.Header) chan struct{} {
- select {
- case pool.reqResetCh <- &txpoolResetRequest{oldHead, newHead}:
- return <-pool.reorgDoneCh
- case <-pool.reorgShutdownCh:
- return pool.reorgShutdownCh
- }
- }
- // requestPromoteExecutables requests transaction promotion checks for the given addresses.
- // The returned channel is closed when the promotion checks have occurred.
- func (pool *TxPool) requestPromoteExecutables(set *accountSet) chan struct{} {
- select {
- case pool.reqPromoteCh <- set:
- return <-pool.reorgDoneCh
- case <-pool.reorgShutdownCh:
- return pool.reorgShutdownCh
- }
- }
- // queueTxEvent enqueues a transaction event to be sent in the next reorg run.
- func (pool *TxPool) queueTxEvent(tx *types.Transaction) {
- select {
- case pool.queueTxEventCh <- tx:
- case <-pool.reorgShutdownCh:
- }
- }
- // scheduleReorgLoop schedules runs of reset and promoteExecutables. Code above should not
- // call those methods directly, but request them being run using requestReset and
- // requestPromoteExecutables instead.
- func (pool *TxPool) scheduleReorgLoop() {
- defer pool.wg.Done()
- var (
- curDone chan struct{} // non-nil while runReorg is active
- nextDone = make(chan struct{})
- launchNextRun bool
- reset *txpoolResetRequest
- dirtyAccounts *accountSet
- queuedEvents = make(map[common.Address]*txSortedMap)
- )
- for {
- // Launch next background reorg if needed
- if curDone == nil && launchNextRun {
- // Run the background reorg and announcements
- go pool.runReorg(nextDone, reset, dirtyAccounts, queuedEvents)
- // Prepare everything for the next round of reorg
- curDone, nextDone = nextDone, make(chan struct{})
- launchNextRun = false
- reset, dirtyAccounts = nil, nil
- queuedEvents = make(map[common.Address]*txSortedMap)
- }
- select {
- case req := <-pool.reqResetCh:
- // Reset request: update head if request is already pending.
- if reset == nil {
- reset = req
- } else {
- reset.newHead = req.newHead
- }
- launchNextRun = true
- pool.reorgDoneCh <- nextDone
- case req := <-pool.reqPromoteCh:
- // Promote request: update address set if request is already pending.
- if dirtyAccounts == nil {
- dirtyAccounts = req
- } else {
- dirtyAccounts.merge(req)
- }
- launchNextRun = true
- pool.reorgDoneCh <- nextDone
- case tx := <-pool.queueTxEventCh:
- // Queue up the event, but don't schedule a reorg. It's up to the caller to
- // request one later if they want the events sent.
- addr, _ := types.Sender(pool.signer, tx)
- if _, ok := queuedEvents[addr]; !ok {
- queuedEvents[addr] = newTxSortedMap()
- }
- queuedEvents[addr].Put(tx)
- case <-curDone:
- curDone = nil
- case <-pool.reorgShutdownCh:
- // Wait for current run to finish.
- if curDone != nil {
- <-curDone
- }
- close(nextDone)
- return
- }
- }
- }
- // runReorg runs reset and promoteExecutables on behalf of scheduleReorgLoop.
- func (pool *TxPool) runReorg(done chan struct{}, reset *txpoolResetRequest, dirtyAccounts *accountSet, events map[common.Address]*txSortedMap) {
- defer close(done)
- var promoteAddrs []common.Address
- if dirtyAccounts != nil && reset == nil {
- // Only dirty accounts need to be promoted, unless we're resetting.
- // For resets, all addresses in the tx queue will be promoted and
- // the flatten operation can be avoided.
- promoteAddrs = dirtyAccounts.flatten()
- }
- pool.mu.Lock()
- if reset != nil {
- // Reset from the old head to the new, rescheduling any reorged transactions
- pool.reset(reset.oldHead, reset.newHead)
- // Nonces were reset, discard any events that became stale
- for addr := range events {
- events[addr].Forward(pool.pendingNonces.get(addr))
- if events[addr].Len() == 0 {
- delete(events, addr)
- }
- }
- // Reset needs promote for all addresses
- promoteAddrs = make([]common.Address, 0, len(pool.queue))
- for addr := range pool.queue {
- promoteAddrs = append(promoteAddrs, addr)
- }
- }
- // Check for pending transactions for every account that sent new ones
- promoted := pool.promoteExecutables(promoteAddrs)
- // If a new block appeared, validate the pool of pending transactions. This will
- // remove any transaction that has been included in the block or was invalidated
- // because of another transaction (e.g. higher gas price).
- if reset != nil {
- pool.demoteUnexecutables()
- }
- // Ensure pool.queue and pool.pending sizes stay within the configured limits.
- pool.truncatePending()
- pool.truncateQueue()
- // Update all accounts to the latest known pending nonce
- for addr, list := range pool.pending {
- highestPending := list.LastElement()
- pool.pendingNonces.set(addr, highestPending.Nonce()+1)
- }
- pool.mu.Unlock()
- // Notify subsystems for newly added transactions
- for _, tx := range promoted {
- addr, _ := types.Sender(pool.signer, tx)
- if _, ok := events[addr]; !ok {
- events[addr] = newTxSortedMap()
- }
- events[addr].Put(tx)
- }
- if len(events) > 0 {
- var txs []*types.Transaction
- for _, set := range events {
- txs = append(txs, set.Flatten()...)
- }
- pool.txFeed.Send(NewTxsEvent{txs})
- }
- }
- // 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())
- )
- if rem == nil {
- // This can happen if a setHead is performed, where we simply discard the old
- // head from the chain.
- // If that is the case, we don't have the lost transactions any more, and
- // there's nothing to add
- if newNum >= oldNum {
- // If we reorged to a same or higher number, then it's not a case of setHead
- log.Warn("Transaction pool reset with missing oldhead",
- "old", oldHead.Hash(), "oldnum", oldNum, "new", newHead.Hash(), "newnum", newNum)
- return
- }
- // If the reorg ended up on a lower number, it's indicative of setHead being the cause
- log.Debug("Skipping transaction reset caused by setHead",
- "old", oldHead.Hash(), "oldnum", oldNum, "new", newHead.Hash(), "newnum", newNum)
- // We still need to update the current state s.th. the lost transactions can be readded by the user
- } else {
- 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.pendingNonces = newTxNoncer(statedb)
- pool.currentMaxGas = newHead.GasLimit
- // Inject any transactions discarded due to reorgs
- log.Debug("Reinjecting stale transactions", "count", len(reinject))
- senderCacher.recover(pool.signer, reinject)
- pool.addTxsLocked(reinject, false)
- // Update all fork indicator by next pending block number.
- next := new(big.Int).Add(newHead.Number, big.NewInt(1))
- pool.istanbul = pool.chainconfig.IsIstanbul(next)
- pool.eip2718 = pool.chainconfig.IsBerlin(next)
- }
- // 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) []*types.Transaction {
- isQuorum := pool.chainconfig.IsQuorum
- // Init delayed since tx pool could have been started before any state sync
- if isQuorum && pool.pendingNonces == nil {
- pool.reset(nil, nil)
- }
- // Track the promoted transactions to broadcast them at once
- var promoted []*types.Transaction
- // 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)
- forwards := list.Forward(pool.currentState.GetNonce(addr))
- for _, tx := range forwards {
- hash := tx.Hash()
- pool.all.Remove(hash)
- }
- log.Trace("Removed old queued transactions", "count", len(forwards))
- var drops types.Transactions
- if !isQuorum || pool.chainconfig.IsGasPriceEnabled(pool.chain.CurrentBlock().Header().Number) {
- // 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()
- pool.all.Remove(hash)
- }
- log.Trace("Removed unpayable queued transactions", "count", len(drops))
- queuedNofundsMeter.Mark(int64(len(drops)))
- }
- // Gather all executable transactions and promote them
- readies := list.Ready(pool.pendingNonces.get(addr))
- for _, tx := range readies {
- hash := tx.Hash()
- log.Trace("Promoting queued transaction", "hash", hash)
- if pool.promoteTx(addr, hash, tx) {
- promoted = append(promoted, tx)
- }
- }
- log.Trace("Promoted queued transactions", "count", len(promoted))
- queuedGauge.Dec(int64(len(readies)))
- // Drop all transactions over the allowed limit
- var caps types.Transactions
- if !pool.locals.contains(addr) {
- caps = list.Cap(int(pool.config.AccountQueue))
- for _, tx := range caps {
- hash := tx.Hash()
- pool.all.Remove(hash)
- log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
- }
- queuedRateLimitMeter.Mark(int64(len(caps)))
- }
- // Mark all the items dropped as removed
- pool.priced.Removed(len(forwards) + len(drops) + len(caps))
- queuedGauge.Dec(int64(len(forwards) + len(drops) + len(caps)))
- if pool.locals.contains(addr) {
- localGauge.Dec(int64(len(forwards) + len(drops) + len(caps)))
- }
- // Delete the entire queue entry if it became empty.
- if list.Empty() {
- delete(pool.queue, addr)
- delete(pool.beats, addr)
- }
- }
- return promoted
- }
- // truncatePending removes transactions from the pending queue if the pool is above the
- // pending limit. The algorithm tries to reduce transaction counts by an approximately
- // equal number for all for accounts with many pending transactions.
- func (pool *TxPool) truncatePending() {
- pending := uint64(0)
- for _, list := range pool.pending {
- pending += uint64(list.Len())
- }
- if pending <= pool.config.GlobalSlots {
- return
- }
- 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]]
- caps := list.Cap(list.Len() - 1)
- for _, tx := range caps {
- // Drop the transaction from the global pools too
- hash := tx.Hash()
- pool.all.Remove(hash)
- // Update the account nonce to the dropped transaction
- pool.pendingNonces.setIfLower(offenders[i], tx.Nonce())
- log.Trace("Removed fairness-exceeding pending transaction", "hash", hash)
- }
- pool.priced.Removed(len(caps))
- pendingGauge.Dec(int64(len(caps)))
- if pool.locals.contains(offenders[i]) {
- localGauge.Dec(int64(len(caps)))
- }
- 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]
- caps := list.Cap(list.Len() - 1)
- for _, tx := range caps {
- // Drop the transaction from the global pools too
- hash := tx.Hash()
- pool.all.Remove(hash)
- // Update the account nonce to the dropped transaction
- pool.pendingNonces.setIfLower(addr, tx.Nonce())
- log.Trace("Removed fairness-exceeding pending transaction", "hash", hash)
- }
- pool.priced.Removed(len(caps))
- pendingGauge.Dec(int64(len(caps)))
- if pool.locals.contains(addr) {
- localGauge.Dec(int64(len(caps)))
- }
- pending--
- }
- }
- }
- pendingRateLimitMeter.Mark(int64(pendingBeforeCap - pending))
- }
- // truncateQueue drops the oldes transactions in the queue if the pool is above the global queue limit.
- func (pool *TxPool) truncateQueue() {
- queued := uint64(0)
- for _, list := range pool.queue {
- queued += uint64(list.Len())
- }
- if queued <= pool.config.GlobalQueue {
- return
- }
- // 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
- queuedRateLimitMeter.Mark(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--
- queuedRateLimitMeter.Mark(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)
- olds := list.Forward(nonce)
- for _, tx := range olds {
- hash := tx.Hash()
- pool.all.Remove(hash)
- log.Trace("Removed old pending transaction", "hash", hash)
- }
- // Drop all transactions that are too costly (low balance or out of gas), and queue any invalids back for later
- 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(len(olds) + len(drops))
- pendingNofundsMeter.Mark(int64(len(drops)))
- for _, tx := range invalids {
- hash := tx.Hash()
- log.Trace("Demoting pending transaction", "hash", hash)
- // Internal shuffle shouldn't touch the lookup set.
- pool.enqueueTx(hash, tx, false, false)
- }
- pendingGauge.Dec(int64(len(olds) + len(drops) + len(invalids)))
- if pool.locals.contains(addr) {
- localGauge.Dec(int64(len(olds) + len(drops) + len(invalids)))
- }
- // If there's a gap in front, alert (should never happen) and postpone all transactions
- if list.Len() > 0 && list.txs.Get(nonce) == nil {
- gapped := list.Cap(0)
- for _, tx := range gapped {
- hash := tx.Hash()
- log.Error("Demoting invalidated transaction", "hash", hash)
- // Internal shuffle shouldn't touch the lookup set.
- pool.enqueueTx(hash, tx, false, false)
- }
- pendingGauge.Dec(int64(len(gapped)))
- // This might happen in a reorg, so log it to the metering
- blockReorgInvalidatedTx.Mark(int64(len(gapped)))
- }
- // Delete the entire pending entry if it became empty.
- if list.Empty() {
- delete(pool.pending, 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, addrs ...common.Address) *accountSet {
- as := &accountSet{
- accounts: make(map[common.Address]struct{}),
- signer: signer,
- }
- for _, addr := range addrs {
- as.add(addr)
- }
- return as
- }
- // 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
- }
- func (as *accountSet) empty() bool {
- return len(as.accounts) == 0
- }
- // 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
- }
- // addTx adds the sender of tx into the set.
- func (as *accountSet) addTx(tx *types.Transaction) {
- if addr, err := types.Sender(as.signer, tx); err == nil {
- as.add(addr)
- }
- }
- // 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
- }
- // merge adds all addresses from the 'other' set into 'as'.
- func (as *accountSet) merge(other *accountSet) {
- for addr := range other.accounts {
- as.accounts[addr] = struct{}{}
- }
- as.cache = nil
- }
- // 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.
- //
- // This lookup set combines the notion of "local transactions", which is useful
- // to build upper-level structure.
- type txLookup struct {
- slots int
- lock sync.RWMutex
- locals map[common.Hash]*types.Transaction
- remotes map[common.Hash]*types.Transaction
- }
- // newTxLookup returns a new txLookup structure.
- func newTxLookup() *txLookup {
- return &txLookup{
- locals: make(map[common.Hash]*types.Transaction),
- remotes: make(map[common.Hash]*types.Transaction),
- }
- }
- // Range calls f on each key and value present in the map. The callback passed
- // should return the indicator whether the iteration needs to be continued.
- // Callers need to specify which set (or both) to be iterated.
- func (t *txLookup) Range(f func(hash common.Hash, tx *types.Transaction, local bool) bool, local bool, remote bool) {
- t.lock.RLock()
- defer t.lock.RUnlock()
- if local {
- for key, value := range t.locals {
- if !f(key, value, true) {
- return
- }
- }
- }
- if remote {
- for key, value := range t.remotes {
- if !f(key, value, false) {
- return
- }
- }
- }
- }
- // 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()
- if tx := t.locals[hash]; tx != nil {
- return tx
- }
- return t.remotes[hash]
- }
- // GetLocal returns a transaction if it exists in the lookup, or nil if not found.
- func (t *txLookup) GetLocal(hash common.Hash) *types.Transaction {
- t.lock.RLock()
- defer t.lock.RUnlock()
- return t.locals[hash]
- }
- // GetRemote returns a transaction if it exists in the lookup, or nil if not found.
- func (t *txLookup) GetRemote(hash common.Hash) *types.Transaction {
- t.lock.RLock()
- defer t.lock.RUnlock()
- return t.remotes[hash]
- }
- // Count returns the current number of transactions in the lookup.
- func (t *txLookup) Count() int {
- t.lock.RLock()
- defer t.lock.RUnlock()
- return len(t.locals) + len(t.remotes)
- }
- // LocalCount returns the current number of local transactions in the lookup.
- func (t *txLookup) LocalCount() int {
- t.lock.RLock()
- defer t.lock.RUnlock()
- return len(t.locals)
- }
- // RemoteCount returns the current number of remote transactions in the lookup.
- func (t *txLookup) RemoteCount() int {
- t.lock.RLock()
- defer t.lock.RUnlock()
- return len(t.remotes)
- }
- // Slots returns the current number of slots used in the lookup.
- func (t *txLookup) Slots() int {
- t.lock.RLock()
- defer t.lock.RUnlock()
- return t.slots
- }
- // Add adds a transaction to the lookup.
- func (t *txLookup) Add(tx *types.Transaction, local bool) {
- t.lock.Lock()
- defer t.lock.Unlock()
- t.slots += numSlots(tx)
- slotsGauge.Update(int64(t.slots))
- if local {
- t.locals[tx.Hash()] = tx
- } else {
- t.remotes[tx.Hash()] = tx
- }
- }
- // Remove removes a transaction from the lookup.
- func (t *txLookup) Remove(hash common.Hash) {
- t.lock.Lock()
- defer t.lock.Unlock()
- tx, ok := t.locals[hash]
- if !ok {
- tx, ok = t.remotes[hash]
- }
- if !ok {
- log.Error("No transaction found to be deleted", "hash", hash)
- return
- }
- t.slots -= numSlots(tx)
- slotsGauge.Update(int64(t.slots))
- delete(t.locals, hash)
- delete(t.remotes, hash)
- }
- // RemoteToLocals migrates the transactions belongs to the given locals to locals
- // set. The assumption is held the locals set is thread-safe to be used.
- func (t *txLookup) RemoteToLocals(locals *accountSet) int {
- t.lock.Lock()
- defer t.lock.Unlock()
- var migrated int
- for hash, tx := range t.remotes {
- if locals.containsTx(tx) {
- t.locals[hash] = tx
- delete(t.remotes, hash)
- migrated += 1
- }
- }
- return migrated
- }
- // numSlots calculates the number of slots needed for a single transaction.
- func numSlots(tx *types.Transaction) int {
- return int((tx.Size() + txSlotSize - 1) / txSlotSize)
- }
- // Quorum
- // helper function to return chainHeadChannel size
- func GetChainHeadChannleSize() int {
- return chainHeadChanSize
- }
|