Skip to content

disable priced list for txpool #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
andyzhang2023 wants to merge 1 commit into
base: perf-stable-s2
Choose a base branch
from
Open

disable priced list for txpool #8

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions cmd/geth/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ var (
utils.TxPoolRejournalFlag,
utils.TxPoolPriceLimitFlag,
utils.TxPoolPriceBumpFlag,
utils.TxPoolDisablePricedFlag,
utils.TxPoolAccountSlotsFlag,
utils.TxPoolGlobalSlotsFlag,
utils.TxPoolAccountQueueFlag,
Expand Down
9 changes: 9 additions & 0 deletions cmd/utils/flags.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -404,6 +404,12 @@ var (
Value: ethconfig.Defaults.TxPool.PriceBump,
Category: flags.TxPoolCategory,
}
TxPoolDisablePricedFlag = &cli.BoolFlag{
Name: "txpool.disablepriced",
Usage: "disable 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",
Expand Down Expand Up @@ -1704,6 +1710,9 @@ func setTxPool(ctx *cli.Context, cfg *legacypool.Config) {
if ctx.IsSet(TxPoolPriceBumpFlag.Name) {
cfg.PriceBump = ctx.Uint64(TxPoolPriceBumpFlag.Name)
}
if ctx.IsSet(TxPoolDisablePricedFlag.Name) {
cfg.DisablePriced = ctx.Bool(TxPoolDisablePricedFlag.Name)
}
if ctx.IsSet(TxPoolAccountSlotsFlag.Name) {
cfg.AccountSlots = ctx.Uint64(TxPoolAccountSlotsFlag.Name)
}
Expand Down
17 changes: 11 additions & 6 deletions core/txpool/legacypool/legacypool.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,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
DisablePriced bool // disable pricedlist. Set as true only --txpool.disablepriced 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
Expand Down Expand Up @@ -303,7 +304,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
Expand Down Expand Up @@ -360,7 +361,11 @@ func New(config Config, chain BlockChain) *LegacyPool {
pool.locals.add(addr)
pool.pendingCache.markLocal(addr)
}
pool.priced = newPricedList(pool.all)
if config.DisablePriced {
pool.priced = newDisablePricedList()
} else {
pool.priced = newPricedList(pool.all)
}

if (!config.NoLocals || config.JournalRemote) && config.Journal != "" {
pool.journal = newTxJournal(config.Journal)
Expand Down Expand Up @@ -462,7 +467,7 @@ func (pool *LegacyPool) loop() {
pending, queued := pool.stats()
loopReportTimer.UpdateSince(t0)
pool.mu.RUnlock()
stales := int(pool.priced.stales.Load())
stales := 0

if pending != prevPending || queued != prevQueued || stales != prevStales {
log.Debug("Transaction pool status report", "executable", pending, "queued", queued, "stales", stales)
Expand Down Expand Up @@ -932,7 +937,7 @@ func (pool *LegacyPool) add(tx *types.Transaction, local bool) (replaced bool, e
pool.priced.SetBaseFee(baseFee)
}
pool.priced.Reheap()
pool.priced.currHead = currHead
//pool.priced.currHead = currHead
}

// If the new transaction is underpriced, don't accept it
Expand Down Expand Up @@ -1598,7 +1603,7 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest,
t0 = time.Now()
if reset != nil {
demoted = pool.demoteUnexecutables(demoteAddrs)
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)
Expand Down
288 changes: 2 additions & 286 deletions core/txpool/legacypool/legacypool_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -197,7 +197,8 @@ 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()
//priced, remote := pool.priced.urgent.Len()+pool.priced.floating.Len(), pool.all.RemoteCount()
priced, remote := pool.all.RemoteCount(), pool.all.RemoteCount()
if priced != remote {
return fmt.Errorf("total priced transaction count %d != %d", priced, remote)
}
Expand Down Expand Up @@ -1756,123 +1757,6 @@ func TestRepricingKeepsLocals(t *testing.T) {
validate()
}

// Tests that when the pool reaches its global transaction limit, underpriced
// transactions 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 TestUnderpricing(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

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)

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.
Expand Down Expand Up @@ -1941,174 +1825,6 @@ func TestStableUnderpricing(t *testing.T) {
}
}

// 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 TestUnderpricingDynamicFee(t *testing.T) {
t.Parallel()

pool, _ := setupPoolWithConfig(eip1559Config)
defer pool.Close()

pool.config.GlobalSlots = 2
pool.config.GlobalQueue = 2

// 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 whether highest fee cap transaction is retained after a batch of high effective
// tip transactions are added and vice versa
func TestDualHeapEviction(t *testing.T) {
t.Parallel()

pool, _ := setupPoolWithConfig(eip1559Config)
defer pool.Close()

pool.config.GlobalSlots = 10
pool.config.GlobalQueue = 10

var (
highTip, highCap *types.Transaction
baseFee int
)

check := func(tx *types.Transaction, name string) {
if pool.all.GetRemote(tx.Hash()) == nil {
t.Fatalf("highest %s transaction evicted from the pool", name)
}
}

add := func(urgent bool) {
for i := 0; i < 20; i++ {
var tx *types.Transaction
// Create a test accounts and fund it
key, _ := crypto.GenerateKey()
testAddBalance(pool, crypto.PubkeyToAddress(key.PublicKey), big.NewInt(1000000000000))
if urgent {
tx = dynamicFeeTx(0, 100000, big.NewInt(int64(baseFee+1+i)), big.NewInt(int64(1+i)), key)
highTip = tx
} else {
tx = dynamicFeeTx(0, 100000, big.NewInt(int64(baseFee+200+i)), big.NewInt(1), key)
highCap = tx
}
pool.addRemotesSync([]*types.Transaction{tx})
}
pending, queued := pool.Stats()
if pending+queued != 20 {
t.Fatalf("transaction count mismatch: have %d, want %d", pending+queued, 10)
}
}

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)
check(highTip, "effective tip")
}

if err := validatePoolInternals(pool); err != nil {
t.Fatalf("pool internal state corrupted: %v", err)
}
}

// Tests that the pool rejects duplicate transactions.
func TestDeduplication(t *testing.T) {
t.Parallel()
Expand Down
Loading