From 3474c4724e929f2006cb591c890798970e38a32d Mon Sep 17 00:00:00 2001 From: andyzhang2023 <147463846+andyzhang2023@users.noreply.github.com> Date: Thu, 9 Jan 2025 16:26:37 +0800 Subject: [PATCH] Txpool opt async priced (#246) Co-authored-by: andyzhang2023 --- cmd/geth/main.go | 1 + cmd/utils/flags.go | 9 + core/txpool/legacypool/async_priced_list.go | 192 +++++++++++++ core/txpool/legacypool/legacypool.go | 31 +- core/txpool/legacypool/legacypool_test.go | 303 +++++++++++++++++++- core/txpool/legacypool/list.go | 38 ++- miner/worker.go | 2 +- 7 files changed, 553 insertions(+), 23 deletions(-) create mode 100644 core/txpool/legacypool/async_priced_list.go diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 784be6d7f3..e45efc9e25 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -79,6 +79,7 @@ var ( utils.TxPoolRejournalFlag, utils.TxPoolPriceLimitFlag, utils.TxPoolPriceBumpFlag, + utils.TxPoolEnableAsyncPricedFlag, utils.TxPoolAccountSlotsFlag, utils.TxPoolGlobalSlotsFlag, utils.TxPoolAccountQueueFlag, diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index ac46716bc6..ef009ac3bb 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -410,6 +410,12 @@ var ( Value: ethconfig.Defaults.TxPool.PriceBump, Category: flags.TxPoolCategory, } + TxPoolEnableAsyncPricedFlag = &cli.BoolFlag{ + Name: "txpool.asyncpriced", + Usage: "enable async-priced-sorted list for txpool", + Value: false, + Category: flags.TxPoolCategory, + } TxPoolAccountSlotsFlag = &cli.Uint64Flag{ Name: "txpool.accountslots", Usage: "Minimum number of executable transaction slots guaranteed per account", @@ -1723,6 +1729,9 @@ func setTxPool(ctx *cli.Context, cfg *legacypool.Config) { if ctx.IsSet(TxPoolPriceBumpFlag.Name) { cfg.PriceBump = ctx.Uint64(TxPoolPriceBumpFlag.Name) } + if ctx.IsSet(TxPoolEnableAsyncPricedFlag.Name) { + cfg.EnableAsyncPriced = ctx.Bool(TxPoolEnableAsyncPricedFlag.Name) + } if ctx.IsSet(TxPoolAccountSlotsFlag.Name) { cfg.AccountSlots = ctx.Uint64(TxPoolAccountSlotsFlag.Name) } diff --git a/core/txpool/legacypool/async_priced_list.go b/core/txpool/legacypool/async_priced_list.go new file mode 100644 index 0000000000..9ac5c31247 --- /dev/null +++ b/core/txpool/legacypool/async_priced_list.go @@ -0,0 +1,192 @@ +package legacypool + +import ( + "math/big" + "sync" + "sync/atomic" + + "github.com/ethereum/go-ethereum/core/types" +) + +var _ pricedListInterface = &asyncPricedList{} + +type addEvent struct { + tx *types.Transaction + local bool +} + +type asyncPricedList struct { + priced *pricedList + floatingLowest atomic.Value + urgentLowest atomic.Value + baseFee atomic.Value + mu sync.Mutex + + // events + quit chan struct{} + reheap chan struct{} + add chan *addEvent + remove chan int + setBaseFee chan *big.Int +} + +func newAsyncPricedList(all *lookup) *asyncPricedList { + a := &asyncPricedList{ + priced: newPricedList(all), + quit: make(chan struct{}), + reheap: make(chan struct{}), + add: make(chan *addEvent), + remove: make(chan int), + setBaseFee: make(chan *big.Int), + } + go a.run() + return a +} + +// run is a loop that handles async operations: +// - reheap: reheap the whole priced list, to get the lowest gas price +// - put: add a transaction to the priced list +// - remove: remove transactions from the priced list +// - discard: remove transactions to make room for new ones +func (a *asyncPricedList) run() { + var reheap bool + var newOnes []*types.Transaction + var toRemove int = 0 + // current loop state + var currentDone chan struct{} = nil + var baseFee *big.Int = nil + for { + if currentDone == nil { + currentDone = make(chan struct{}) + go a.handle(reheap, newOnes, toRemove, baseFee, currentDone) + reheap, newOnes, toRemove, baseFee = false, nil, 0, nil + } + select { + case <-a.reheap: + reheap = true + + case add := <-a.add: + newOnes = append(newOnes, add.tx) + + case remove := <-a.remove: + toRemove += remove + + case baseFee = <-a.setBaseFee: + + case <-currentDone: + currentDone = nil + + case <-a.quit: + // Wait for current run to finish. + if currentDone != nil { + <-currentDone + } + return + } + } +} + +func (a *asyncPricedList) handle(reheap bool, newOnes []*types.Transaction, toRemove int, baseFee *big.Int, finished chan struct{}) { + defer close(finished) + a.mu.Lock() + defer a.mu.Unlock() + // add new transactions to the priced list + for _, tx := range newOnes { + a.priced.Put(tx, false) + } + // remove staled transactions from the priced list + a.priced.Removed(toRemove) + // reheap if needed + if reheap { + a.priced.Reheap() + // set the lowest priced transaction when reheap is done + var emptyTx *types.Transaction = nil + if len(a.priced.floating.list) > 0 { + a.floatingLowest.Store(a.priced.floating.list[0]) + } else { + a.floatingLowest.Store(emptyTx) + } + if len(a.priced.urgent.list) > 0 { + a.urgentLowest.Store(a.priced.urgent.list[0]) + } else { + a.urgentLowest.Store(emptyTx) + } + } + if baseFee != nil { + a.baseFee.Store(baseFee) + a.priced.SetBaseFee(baseFee) + } +} + +func (a *asyncPricedList) Staled() int { + // the Staled() of pricedList is thread-safe, so we don't need to lock here + return a.priced.Staled() +} + +func (a *asyncPricedList) Put(tx *types.Transaction, local bool) { + a.add <- &addEvent{tx, local} +} + +func (a *asyncPricedList) Removed(count int) { + a.remove <- count +} + +func (a *asyncPricedList) Underpriced(tx *types.Transaction) bool { + var urgentLowest, floatingLowest *types.Transaction = nil, nil + ul, fl := a.urgentLowest.Load(), a.floatingLowest.Load() + if ul != nil { + // be careful that ul might be nil + urgentLowest = ul.(*types.Transaction) + } + if fl != nil { + // be careful that fl might be nil + floatingLowest = fl.(*types.Transaction) + } + a.mu.Lock() + defer a.mu.Unlock() + return (urgentLowest == nil || a.priced.urgent.cmp(urgentLowest, tx) >= 0) && + (floatingLowest == nil || a.priced.floating.cmp(floatingLowest, tx) >= 0) && + (floatingLowest != nil || urgentLowest != nil) +} + +// Disacard cleans staled transactions to make room for new ones +func (a *asyncPricedList) Discard(slots int, force bool) (types.Transactions, bool) { + a.mu.Lock() + defer a.mu.Unlock() + return a.priced.Discard(slots, force) +} + +func (a *asyncPricedList) NeedReheap(currHead *types.Header) bool { + return false +} + +func (a *asyncPricedList) Reheap() { + a.reheap <- struct{}{} +} + +func (a *asyncPricedList) SetBaseFee(baseFee *big.Int) { + a.setBaseFee <- baseFee + a.reheap <- struct{}{} +} + +func (a *asyncPricedList) SetHead(currHead *types.Header) { + //do nothing +} + +func (a *asyncPricedList) GetBaseFee() *big.Int { + baseFee := a.baseFee.Load() + if baseFee == nil { + return big.NewInt(0) + } + return baseFee.(*big.Int) +} + +func (a *asyncPricedList) Close() { + close(a.quit) +} + +func (a *asyncPricedList) TxCount() int { + a.mu.Lock() + defer a.mu.Unlock() + return a.priced.TxCount() +} diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go index d69f6fe9d4..c82b2eeb73 100644 --- a/core/txpool/legacypool/legacypool.go +++ b/core/txpool/legacypool/legacypool.go @@ -153,7 +153,8 @@ type BlockChain interface { // Config are the configuration parameters of the transaction pool. type Config struct { - EnableCache bool // enable pending cache for mining. Set as true only --mine option is enabled + EnableAsyncPriced bool // enable async pricedlist. Set as true only --txpool.enableasyncpriced option is enabled + EnableCache bool // enable pending cache for mining. Set as true only --mine option is enabled Locals []common.Address // Addresses that should be treated by default as local NoLocals bool // Whether local transaction handling should be disabled @@ -238,6 +239,9 @@ func (config *Config) sanitize() Config { log.Warn("Sanitizing invalid txpool reannounce time", "provided", conf.ReannounceTime, "updated", time.Minute) conf.ReannounceTime = time.Minute } + if config.EnableAsyncPriced { + log.Info("Enabling async pricedlist") + } // log to inform user if the cache is enabled or not if conf.EnableCache { log.Info("legacytxpool Pending Cache is enabled") @@ -276,7 +280,7 @@ type LegacyPool struct { queue map[common.Address]*list // Queued but non-processable transactions beats map[common.Address]time.Time // Last heartbeat from each known account all *lookup // All transactions to allow lookups - priced *pricedList // All transactions sorted by price + priced pricedListInterface // All transactions sorted by price pendingCounter int queueCounter int @@ -333,7 +337,11 @@ func New(config Config, chain BlockChain) *LegacyPool { pool.locals.add(addr) pool.pendingCache.markLocal(addr) } - pool.priced = newPricedList(pool.all) + if config.EnableAsyncPriced { + pool.priced = newAsyncPricedList(pool.all) + } else { + pool.priced = newPricedList(pool.all) + } if (!config.NoLocals || config.JournalRemote) && config.Journal != "" { pool.journal = newTxJournal(config.Journal) @@ -419,6 +427,7 @@ func (pool *LegacyPool) loop() { defer evict.Stop() defer journal.Stop() defer reannounce.Stop() + defer pool.priced.Close() // Notify tests that the init phase is done close(pool.initDoneCh) @@ -433,7 +442,7 @@ func (pool *LegacyPool) loop() { pool.mu.RLock() pending, queued := pool.stats() pool.mu.RUnlock() - stales := int(pool.priced.stales.Load()) + stales := pool.priced.Staled() if pending != prevPending || queued != prevQueued || stales != prevStales { log.Debug("Transaction pool status report", "executable", pending, "queued", queued, "stales", stales) @@ -882,16 +891,6 @@ func (pool *LegacyPool) add(tx *types.Transaction, local bool) (replaced bool, e } // If the transaction pool is full, discard underpriced transactions if uint64(pool.all.Slots()+numSlots(tx)) > pool.config.GlobalSlots+pool.config.GlobalQueue { - currHead := pool.currentHead.Load() - if currHead != nil && currHead.BaseFee != nil && pool.priced.NeedReheap(currHead) { - if pool.chainconfig.IsLondon(new(big.Int).Add(currHead.Number, big.NewInt(1))) { - baseFee := eip1559.CalcBaseFee(pool.chainconfig, currHead, currHead.Time+1) - pool.priced.SetBaseFee(baseFee) - } - pool.priced.Reheap() - pool.priced.currHead = currHead - } - // If the new transaction is underpriced, don't accept it if !isLocal && pool.priced.Underpriced(tx) { log.Trace("Discarding underpriced transaction", "hash", hash, "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap()) @@ -1509,11 +1508,13 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest, if reset != nil { pool.demoteUnexecutables(demoteAddrs) demoteTimer.UpdateSince(t0) - var pendingBaseFee = pool.priced.urgent.baseFee + var pendingBaseFee = pool.priced.GetBaseFee() if reset.newHead != nil { if pool.chainconfig.IsLondon(new(big.Int).Add(reset.newHead.Number, big.NewInt(1))) { pendingBaseFee = eip1559.CalcBaseFee(pool.chainconfig, reset.newHead, reset.newHead.Time+1) pool.priced.SetBaseFee(pendingBaseFee) + } else { + pool.priced.Reheap() } } // Update all accounts to the latest known pending nonce diff --git a/core/txpool/legacypool/legacypool_test.go b/core/txpool/legacypool/legacypool_test.go index b587f3676d..8e37e3908a 100644 --- a/core/txpool/legacypool/legacypool_test.go +++ b/core/txpool/legacypool/legacypool_test.go @@ -197,7 +197,11 @@ func validatePoolInternals(pool *LegacyPool) error { return fmt.Errorf("total transaction count %d != %d pending + %d queued", total, pending, queued) } pool.priced.Reheap() - priced, remote := pool.priced.urgent.Len()+pool.priced.floating.Len(), pool.all.RemoteCount() + if pool.config.EnableAsyncPriced { + // sleep a bit to wait for the reheap to finish + time.Sleep(50 * time.Millisecond) + } + priced, remote := pool.priced.TxCount(), pool.all.RemoteCount() if priced != remote { return fmt.Errorf("total priced transaction count %d != %d", priced, remote) } @@ -1873,6 +1877,122 @@ func TestUnderpricing(t *testing.T) { } } +func TestAsyncUnderpricing(t *testing.T) { + t.Parallel() + + // Create the pool to test the pricing enforcement with + statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil) + blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) + + config := testTxPoolConfig + config.GlobalSlots = 2 + config.GlobalQueue = 2 + config.EnableAsyncPriced = true + + pool := New(config, blockchain) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + defer pool.Close() + + // Keep track of transaction events to ensure all executables get announced + events := make(chan core.NewTxsEvent, 32) + sub := pool.txFeed.Subscribe(events) + defer sub.Unsubscribe() + + // Create a number of test accounts and fund them + keys := make([]*ecdsa.PrivateKey, 5) + for i := 0; i < len(keys); i++ { + keys[i], _ = crypto.GenerateKey() + testAddBalance(pool, crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000)) + } + // Generate and queue a batch of transactions, both pending and queued + txs := types.Transactions{} + + txs = append(txs, pricedTransaction(0, 100000, big.NewInt(1), keys[0])) + txs = append(txs, pricedTransaction(1, 100000, big.NewInt(2), keys[0])) + + txs = append(txs, pricedTransaction(1, 100000, big.NewInt(1), keys[1])) + + ltx := pricedTransaction(0, 100000, big.NewInt(1), keys[2]) + + // Import the batch and that both pending and queued transactions match up + pool.addRemotes(txs) + pool.addLocal(ltx) + + // sleep a bit to wait for priced transactions to be processed in parallel + time.Sleep(50 * time.Millisecond) + + pending, queued := pool.Stats() + if pending != 3 { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 3) + } + if queued != 1 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 1) + } + if err := validateEvents(events, 3); err != nil { + t.Fatalf("original event firing failed: %v", err) + } + if err := validatePoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } + // Ensure that adding an underpriced transaction on block limit fails + if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1), keys[1])); !errors.Is(err, txpool.ErrUnderpriced) { + t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, txpool.ErrUnderpriced) + } + // Replace a future transaction with a future transaction + if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(2), keys[1])); err != nil { // +K1:1 => -K1:1 => Pend K0:0, K0:1, K2:0; Que K1:1 + t.Fatalf("failed to add well priced transaction: %v", err) + } + // Ensure that adding high priced transactions drops cheap ones, but not own + if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(3), keys[1])); err != nil { // +K1:0 => -K1:1 => Pend K0:0, K0:1, K1:0, K2:0; Que - + t.Fatalf("failed to add well priced transaction: %v", err) + } + if err := pool.addRemoteSync(pricedTransaction(2, 100000, big.NewInt(4), keys[1])); err != nil { // +K1:2 => -K0:0 => Pend K1:0, K2:0; Que K0:1 K1:2 + t.Fatalf("failed to add well priced transaction: %v", err) + } + if err := pool.addRemote(pricedTransaction(3, 100000, big.NewInt(5), keys[1])); err != nil { // +K1:3 => -K0:1 => Pend K1:0, K2:0; Que K1:2 K1:3 + t.Fatalf("failed to add well priced transaction: %v", err) + } + // Ensure that replacing a pending transaction with a future transaction fails + if err := pool.addRemote(pricedTransaction(5, 100000, big.NewInt(6), keys[1])); err != txpool.ErrFutureReplacePending { + t.Fatalf("adding future replace transaction error mismatch: have %v, want %v", err, txpool.ErrFutureReplacePending) + } + pending, queued = pool.Stats() + if pending != 2 { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2) + } + if queued != 2 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 2) + } + if err := validateEvents(events, 2); err != nil { + t.Fatalf("additional event firing failed: %v", err) + } + if err := validatePoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } + // Ensure that adding local transactions can push out even higher priced ones + ltx = pricedTransaction(1, 100000, big.NewInt(0), keys[2]) + if err := pool.addLocal(ltx); err != nil { + t.Fatalf("failed to append underpriced local transaction: %v", err) + } + ltx = pricedTransaction(0, 100000, big.NewInt(0), keys[3]) + if err := pool.addLocal(ltx); err != nil { + t.Fatalf("failed to add new underpriced local transaction: %v", err) + } + pending, queued = pool.Stats() + if pending != 3 { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 3) + } + if queued != 1 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 1) + } + if err := validateEvents(events, 2); err != nil { + t.Fatalf("local event firing failed: %v", err) + } + if err := validatePoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } +} + // Tests that more expensive transactions push out cheap ones from the pool, but // without producing instability by creating gaps that start jumping transactions // back and forth between queued/pending. @@ -1941,6 +2061,186 @@ func TestStableUnderpricing(t *testing.T) { } } +// Tests that more expensive transactions push out cheap ones from the pool, but +// without producing instability by creating gaps that start jumping transactions +// back and forth between queued/pending. +func TestAsyncStableUnderpricing(t *testing.T) { + t.Parallel() + + // Create the pool to test the pricing enforcement with + statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil) + blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) + + config := testTxPoolConfig + config.GlobalSlots = 128 + config.GlobalQueue = 0 + config.EnableAsyncPriced = true + + pool := New(config, blockchain) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + defer pool.Close() + + // Keep track of transaction events to ensure all executables get announced + events := make(chan core.NewTxsEvent, 32) + sub := pool.txFeed.Subscribe(events) + defer sub.Unsubscribe() + + // Create a number of test accounts and fund them + keys := make([]*ecdsa.PrivateKey, 2) + for i := 0; i < len(keys); i++ { + keys[i], _ = crypto.GenerateKey() + testAddBalance(pool, crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000)) + } + // Fill up the entire queue with the same transaction price points + txs := types.Transactions{} + for i := uint64(0); i < config.GlobalSlots; i++ { + txs = append(txs, pricedTransaction(i, 100000, big.NewInt(1), keys[0])) + } + pool.addRemotesSync(txs) + + pending, queued := pool.Stats() + if pending != int(config.GlobalSlots) { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, config.GlobalSlots) + } + if queued != 0 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 0) + } + if err := validateEvents(events, int(config.GlobalSlots)); err != nil { + t.Fatalf("original event firing failed: %v", err) + } + if err := validatePoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } + // Ensure that adding high priced transactions drops a cheap, but doesn't produce a gap + if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(3), keys[1])); err != nil { + t.Fatalf("failed to add well priced transaction: %v", err) + } + pending, queued = pool.Stats() + if pending != int(config.GlobalSlots) { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, config.GlobalSlots) + } + if queued != 0 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 0) + } + if err := validateEvents(events, 1); err != nil { + t.Fatalf("additional event firing failed: %v", err) + } + if err := validatePoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } +} + +// Tests that when the pool reaches its global transaction limit, underpriced +// transactions (legacy & dynamic fee) are gradually shifted out for more +// expensive ones and any gapped pending transactions are moved into the queue. +// +// Note, local transactions are never allowed to be dropped. +func TestAsyncUnderpricingDynamicFee(t *testing.T) { + t.Parallel() + + pool, _ := setupPoolWithConfig(eip1559Config) + defer pool.Close() + + pool.config.GlobalSlots = 2 + pool.config.GlobalQueue = 2 + pool.config.EnableAsyncPriced = true + + // Keep track of transaction events to ensure all executables get announced + events := make(chan core.NewTxsEvent, 32) + sub := pool.txFeed.Subscribe(events) + defer sub.Unsubscribe() + + // Create a number of test accounts and fund them + keys := make([]*ecdsa.PrivateKey, 4) + for i := 0; i < len(keys); i++ { + keys[i], _ = crypto.GenerateKey() + testAddBalance(pool, crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000)) + } + + // Generate and queue a batch of transactions, both pending and queued + txs := types.Transactions{} + + txs = append(txs, dynamicFeeTx(0, 100000, big.NewInt(3), big.NewInt(2), keys[0])) + txs = append(txs, pricedTransaction(1, 100000, big.NewInt(2), keys[0])) + txs = append(txs, dynamicFeeTx(1, 100000, big.NewInt(2), big.NewInt(1), keys[1])) + + ltx := dynamicFeeTx(0, 100000, big.NewInt(2), big.NewInt(1), keys[2]) + + // Import the batch and that both pending and queued transactions match up + pool.addRemotes(txs) // Pend K0:0, K0:1; Que K1:1 + pool.addLocal(ltx) // +K2:0 => Pend K0:0, K0:1, K2:0; Que K1:1 + + pending, queued := pool.Stats() + if pending != 3 { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 3) + } + if queued != 1 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 1) + } + if err := validateEvents(events, 3); err != nil { + t.Fatalf("original event firing failed: %v", err) + } + if err := validatePoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } + + // Ensure that adding an underpriced transaction fails + tx := dynamicFeeTx(0, 100000, big.NewInt(2), big.NewInt(1), keys[1]) + if err := pool.addRemote(tx); !errors.Is(err, txpool.ErrUnderpriced) { // Pend K0:0, K0:1, K2:0; Que K1:1 + t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, txpool.ErrUnderpriced) + } + + // Ensure that adding high priced transactions drops cheap ones, but not own + tx = pricedTransaction(0, 100000, big.NewInt(2), keys[1]) + if err := pool.addRemote(tx); err != nil { // +K1:0, -K1:1 => Pend K0:0, K0:1, K1:0, K2:0; Que - + t.Fatalf("failed to add well priced transaction: %v", err) + } + + tx = pricedTransaction(1, 100000, big.NewInt(3), keys[1]) + if err := pool.addRemoteSync(tx); err != nil { // +K1:2, -K0:1 => Pend K0:0 K1:0, K2:0; Que K1:2 + t.Fatalf("failed to add well priced transaction: %v", err) + } + tx = dynamicFeeTx(2, 100000, big.NewInt(4), big.NewInt(1), keys[1]) + if err := pool.addRemoteSync(tx); err != nil { // +K1:3, -K1:0 => Pend K0:0 K2:0; Que K1:2 K1:3 + t.Fatalf("failed to add well priced transaction: %v", err) + } + pending, queued = pool.Stats() + if pending != 2 { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2) + } + if queued != 2 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 2) + } + if err := validateEvents(events, 2); err != nil { + t.Fatalf("additional event firing failed: %v", err) + } + if err := validatePoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } + // Ensure that adding local transactions can push out even higher priced ones + ltx = dynamicFeeTx(1, 100000, big.NewInt(0), big.NewInt(0), keys[2]) + if err := pool.addLocal(ltx); err != nil { + t.Fatalf("failed to append underpriced local transaction: %v", err) + } + ltx = dynamicFeeTx(0, 100000, big.NewInt(0), big.NewInt(0), keys[3]) + if err := pool.addLocal(ltx); err != nil { + t.Fatalf("failed to add new underpriced local transaction: %v", err) + } + pending, queued = pool.Stats() + if pending != 3 { + t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 3) + } + if queued != 1 { + t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 1) + } + if err := validateEvents(events, 2); err != nil { + t.Fatalf("local event firing failed: %v", err) + } + if err := validatePoolInternals(pool); err != nil { + t.Fatalf("pool internal state corrupted: %v", err) + } +} + // Tests that when the pool reaches its global transaction limit, underpriced // transactions (legacy & dynamic fee) are gradually shifted out for more // expensive ones and any gapped pending transactions are moved into the queue. @@ -2097,7 +2397,6 @@ func TestDualHeapEviction(t *testing.T) { add(false) for baseFee = 0; baseFee <= 1000; baseFee += 100 { pool.priced.SetBaseFee(big.NewInt(int64(baseFee))) - pool.priced.Reheap() add(true) check(highCap, "fee cap") add(false) diff --git a/core/txpool/legacypool/list.go b/core/txpool/legacypool/list.go index 6b823a4a73..181fadb470 100644 --- a/core/txpool/legacypool/list.go +++ b/core/txpool/legacypool/list.go @@ -537,6 +537,21 @@ func (h *priceHeap) Pop() interface{} { return x } +var _ pricedListInterface = (*pricedList)(nil) + +type pricedListInterface interface { + Put(tx *types.Transaction, local bool) + Removed(count int) + Underpriced(tx *types.Transaction) bool + Discard(slots int, force bool) (types.Transactions, bool) + Reheap() + SetBaseFee(baseFee *big.Int) + GetBaseFee() *big.Int + Staled() int + TxCount() int + Close() +} + // pricedList is a price-sorted heap to allow operating on transactions pool // contents in a price-incrementing way. It's built upon the all transactions // in txpool but only interested in the remote part. It means only remote transactions @@ -549,7 +564,6 @@ func (h *priceHeap) Pop() interface{} { // better candidates for inclusion while in other cases (at the top of the baseFee peak) // the floating heap is better. When baseFee is decreasing they behave similarly. type pricedList struct { - currHead *types.Header // Current block header for effective tip calculation // Number of stale price points to (re-heap trigger). stales atomic.Int64 @@ -571,6 +585,10 @@ func newPricedList(all *lookup) *pricedList { } } +func (l *pricedList) Staled() int { + return int(l.stales.Load()) +} + // Put inserts a new transaction into the heap. func (l *pricedList) Put(tx *types.Transaction, local bool) { if local { @@ -668,10 +686,6 @@ func (l *pricedList) Discard(slots int, force bool) (types.Transactions, bool) { return drop, true } -func (l *pricedList) NeedReheap(currHead *types.Header) bool { - return l.currHead == nil || currHead == nil || currHead.Hash().Cmp(l.currHead.Hash()) != 0 -} - // Reheap forcibly rebuilds the heap based on the current remote transaction set. func (l *pricedList) Reheap() { l.reheapMu.Lock() @@ -703,4 +717,18 @@ func (l *pricedList) Reheap() { // necessary to call right before SetBaseFee when processing a new block. func (l *pricedList) SetBaseFee(baseFee *big.Int) { l.urgent.baseFee = baseFee + l.Reheap() +} + +// GetBaseFee returns the current base fee used for sorting the urgent heap. +func (l *pricedList) GetBaseFee() *big.Int { + return l.urgent.baseFee +} + +func (l *pricedList) TxCount() int { + return len(l.urgent.list) + len(l.floating.list) +} + +func (l *pricedList) Close() { + //do nothing } diff --git a/miner/worker.go b/miner/worker.go index 3ea2bfc76b..5cf742bcd5 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -1296,7 +1296,7 @@ func (w *worker) fillTransactions(interrupt *atomic.Int32, env *environment) err pendingBlobTxs := w.eth.TxPool().Pending(filter) packFromTxpoolTimer.UpdateSince(start) - log.Debug("packFromTxpoolTimer", "duration", common.PrettyDuration(time.Since(start)), "hash", env.header.Hash()) + log.Debug("packFromTxpoolTimer", "duration", common.PrettyDuration(time.Since(start)), "hash", env.header.Hash(), "txs", len(pendingPlainTxs)) // Split the pending transactions into locals and remotes. localPlainTxs, remotePlainTxs := make(map[common.Address][]*txpool.LazyTransaction), pendingPlainTxs