core/txpool/legacypool: add overflowpool for txs (#2660)

cmd/geth/main.go

@@ -92,6 +92,7 @@ var (
 		utils.TxPoolGlobalSlotsFlag,
 		utils.TxPoolAccountQueueFlag,
 		utils.TxPoolGlobalQueueFlag,
+		utils.TxPoolOverflowPoolSlotsFlag,
 		utils.TxPoolLifetimeFlag,
 		utils.TxPoolReannounceTimeFlag,
 		utils.BlobPoolDataDirFlag,

cmd/utils/flags.go

@@ -458,6 +458,12 @@ var (
 		Value:    ethconfig.Defaults.TxPool.GlobalQueue,
 		Category: flags.TxPoolCategory,
 	}
+	TxPoolOverflowPoolSlotsFlag = &cli.Uint64Flag{
+		Name:     "txpool.overflowpoolslots",
+		Usage:    "Maximum number of transaction slots in overflow pool",
+		Value:    ethconfig.Defaults.TxPool.OverflowPoolSlots,
+		Category: flags.TxPoolCategory,
+	}
 	TxPoolLifetimeFlag = &cli.DurationFlag{
 		Name:     "txpool.lifetime",
 		Usage:    "Maximum amount of time non-executable transaction are queued",
@@ -1789,6 +1795,9 @@ func setTxPool(ctx *cli.Context, cfg *legacypool.Config) {
 	if ctx.IsSet(TxPoolGlobalQueueFlag.Name) {
 		cfg.GlobalQueue = ctx.Uint64(TxPoolGlobalQueueFlag.Name)
 	}
+	if ctx.IsSet(TxPoolOverflowPoolSlotsFlag.Name) {
+		cfg.OverflowPoolSlots = ctx.Uint64(TxPoolOverflowPoolSlotsFlag.Name)
+	}
 	if ctx.IsSet(TxPoolLifetimeFlag.Name) {
 		cfg.Lifetime = ctx.Duration(TxPoolLifetimeFlag.Name)
 	}
@@ -2310,16 +2319,17 @@ func EnableNodeInfo(poolConfig *legacypool.Config, nodeInfo *p2p.NodeInfo) Setup
 	return func() {
 		// register node info into metrics
 		metrics.NewRegisteredLabel("node-info", nil).Mark(map[string]interface{}{
-			"Enode":        nodeInfo.Enode,
-			"ENR":          nodeInfo.ENR,
-			"ID":           nodeInfo.ID,
-			"PriceLimit":   poolConfig.PriceLimit,
-			"PriceBump":    poolConfig.PriceBump,
-			"AccountSlots": poolConfig.AccountSlots,
-			"GlobalSlots":  poolConfig.GlobalSlots,
-			"AccountQueue": poolConfig.AccountQueue,
-			"GlobalQueue":  poolConfig.GlobalQueue,
-			"Lifetime":     poolConfig.Lifetime,
+			"Enode":             nodeInfo.Enode,
+			"ENR":               nodeInfo.ENR,
+			"ID":                nodeInfo.ID,
+			"PriceLimit":        poolConfig.PriceLimit,
+			"PriceBump":         poolConfig.PriceBump,
+			"AccountSlots":      poolConfig.AccountSlots,
+			"GlobalSlots":       poolConfig.GlobalSlots,
+			"AccountQueue":      poolConfig.AccountQueue,
+			"GlobalQueue":       poolConfig.GlobalQueue,
+			"OverflowPoolSlots": poolConfig.OverflowPoolSlots,
+			"Lifetime":          poolConfig.Lifetime,
 		})
 	}
 }

core/txpool/legacypool/legacypool.go

@@ -19,6 +19,7 @@ package legacypool
 
 import (
 	"errors"
+	"fmt"
 	"math"
 	"math/big"
 	"sort"
@@ -99,10 +100,11 @@ var (
 	// that this number is pretty low, since txpool reorgs happen very frequently.
 	dropBetweenReorgHistogram = metrics.NewRegisteredHistogram("txpool/dropbetweenreorg", nil, metrics.NewExpDecaySample(1028, 0.015))
 
-	pendingGauge = metrics.NewRegisteredGauge("txpool/pending", nil)
-	queuedGauge  = metrics.NewRegisteredGauge("txpool/queued", nil)
-	localGauge   = metrics.NewRegisteredGauge("txpool/local", nil)
-	slotsGauge   = metrics.NewRegisteredGauge("txpool/slots", 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)
+	OverflowPoolGauge = metrics.NewRegisteredGauge("txpool/overflowpool", nil)
 
 	reheapTimer = metrics.NewRegisteredTimer("txpool/reheap", nil)
 )
@@ -133,10 +135,11 @@ type Config struct {
 	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
+	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
+	OverflowPoolSlots uint64 // Maximum number of transaction slots in overflow pool
 
 	Lifetime       time.Duration // Maximum amount of time non-executable transaction are queued
 	ReannounceTime time.Duration // Duration for announcing local pending transactions again
@@ -150,10 +153,11 @@ var DefaultConfig = Config{
 	PriceLimit: 1,
 	PriceBump:  10,
 
-	AccountSlots: 16,
-	GlobalSlots:  4096 + 1024, // urgent + floating queue capacity with 4:1 ratio
-	AccountQueue: 64,
-	GlobalQueue:  1024,
+	AccountSlots:      16,
+	GlobalSlots:       4096 + 1024, // urgent + floating queue capacity with 4:1 ratio
+	AccountQueue:      64,
+	GlobalQueue:       1024,
+	OverflowPoolSlots: 0,
 
 	Lifetime:       3 * time.Hour,
 	ReannounceTime: 10 * 365 * 24 * time.Hour,
@@ -235,6 +239,8 @@ type LegacyPool struct {
 	all     *lookup                      // All transactions to allow lookups
 	priced  *pricedList                  // All transactions sorted by price
 
+	localBufferPool *TxOverflowPool // Local buffer transactions
+
 	reqResetCh      chan *txpoolResetRequest
 	reqPromoteCh    chan *accountSet
 	queueTxEventCh  chan *types.Transaction
@@ -272,6 +278,7 @@ func New(config Config, chain BlockChain) *LegacyPool {
 		reorgDoneCh:     make(chan chan struct{}),
 		reorgShutdownCh: make(chan struct{}),
 		initDoneCh:      make(chan struct{}),
+		localBufferPool: NewTxOverflowPoolHeap(config.OverflowPoolSlots),
 	}
 	pool.locals = newAccountSet(pool.signer)
 	for _, addr := range config.Locals {
@@ -408,7 +415,6 @@ func (pool *LegacyPool) loop() {
 					if !pool.locals.contains(addr) {
 						continue
 					}
-
 					for _, tx := range list.Flatten() {
 						// Default ReannounceTime is 10 years, won't announce by default.
 						if time.Since(tx.Time()) < pool.config.ReannounceTime {
@@ -517,6 +523,17 @@ func (pool *LegacyPool) Stats() (int, int) {
 	return pool.stats()
 }
 
+func (pool *LegacyPool) statsOverflowPool() int {
+	pool.mu.RLock()
+	defer pool.mu.RUnlock()
+
+	if pool.localBufferPool == nil {
+		return 0
+	}
+
+	return pool.localBufferPool.Size()
+}
+
 // stats retrieves the current pool stats, namely the number of pending and the
 // number of queued (non-executable) transactions.
 func (pool *LegacyPool) stats() (int, int) {
@@ -831,6 +848,8 @@ func (pool *LegacyPool) add(tx *types.Transaction, local bool) (replaced bool, e
 			}
 		}
 
+		pool.addToOverflowPool(drop, isLocal)
+
 		// Kick out the underpriced remote transactions.
 		for _, tx := range drop {
 			log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap())
@@ -887,6 +906,29 @@ func (pool *LegacyPool) add(tx *types.Transaction, local bool) (replaced bool, e
 	return replaced, nil
 }
 
+func (pool *LegacyPool) addToOverflowPool(drop types.Transactions, isLocal bool) {
+	// calculate total number of slots in drop. Accordingly add them to OverflowPool (if there is space)
+	availableSlotsOverflowPool := pool.availableSlotsOverflowPool()
+	if availableSlotsOverflowPool > 0 {
+		// transfer availableSlotsOverflowPool number of transactions slots from drop to OverflowPool
+		currentSlotsUsed := 0
+		for i, tx := range drop {
+			txSlots := numSlots(tx)
+			if currentSlotsUsed+txSlots <= availableSlotsOverflowPool {
+				from, _ := types.Sender(pool.signer, tx)
+				pool.localBufferPool.Add(tx)
+				log.Debug("adding to OverflowPool", "transaction", tx.Hash().String(), "from", from.String())
+				currentSlotsUsed += txSlots
+			} else {
+				log.Debug("not all got added to OverflowPool", "totalAdded", i+1)
+				return
+			}
+		}
+	} else {
+		log.Debug("adding to OverflowPool unsuccessful", "availableSlotsOverflowPool", availableSlotsOverflowPool)
+	}
+}
+
 // isGapped reports whether the given transaction is immediately executable.
 func (pool *LegacyPool) isGapped(from common.Address, tx *types.Transaction) bool {
 	// Short circuit if transaction falls within the scope of the pending list
@@ -1333,7 +1375,6 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest,
 		reorgDurationTimer.Update(time.Since(t0))
 	}(time.Now())
 	defer close(done)
-
 	var promoteAddrs []common.Address
 	if dirtyAccounts != nil && reset == nil {
 		// Only dirty accounts need to be promoted, unless we're resetting.
@@ -1391,6 +1432,9 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest,
 	pool.changesSinceReorg = 0 // Reset change counter
 	pool.mu.Unlock()
 
+	// Transfer transactions from OverflowPool to MainPool for new block import
+	pool.transferTransactions()
+
 	// Notify subsystems for newly added transactions
 	for _, tx := range promoted {
 		addr, _ := types.Sender(pool.signer, tx)
@@ -2038,3 +2082,50 @@ func (t *lookup) RemotesBelowTip(threshold *big.Int) types.Transactions {
 func numSlots(tx *types.Transaction) int {
 	return int((tx.Size() + txSlotSize - 1) / txSlotSize)
 }
+
+// transferTransactions moves transactions from OverflowPool to MainPool
+func (pool *LegacyPool) transferTransactions() {
+	// Fail fast if the overflow pool is empty
+	if pool.localBufferPool.Size() == 0 {
+		return
+	}
+
+	maxMainPoolSize := int(pool.config.GlobalSlots + pool.config.GlobalQueue)
+	// Use pool.all.Slots() to get the total slots used by all transactions
+	currentMainPoolSize := pool.all.Slots()
+	if currentMainPoolSize >= maxMainPoolSize {
+		return
+	}
+
+	extraSlots := maxMainPoolSize - currentMainPoolSize
+	extraTransactions := (extraSlots + 3) / 4 // Since a transaction can take up to 4 slots
+	log.Debug("Will attempt to transfer from OverflowPool to MainPool", "transactions", extraTransactions)
+	txs := pool.localBufferPool.Flush(extraTransactions)
+	if len(txs) == 0 {
+		return
+	}
+
+	pool.Add(txs, true, false)
+}
+
+func (pool *LegacyPool) availableSlotsOverflowPool() int {
+	maxOverflowPoolSize := int(pool.config.OverflowPoolSlots)
+	availableSlots := maxOverflowPoolSize - pool.localBufferPool.Size()
+	if availableSlots > 0 {
+		return availableSlots
+	}
+	return 0
+}
+
+func (pool *LegacyPool) PrintTxStats() {
+	for _, l := range pool.pending {
+		for _, transaction := range l.txs.items {
+			from, _ := types.Sender(pool.signer, transaction)
+			fmt.Println("from: ", from, " Pending:", transaction.Hash().String(), transaction.GasFeeCap(), transaction.GasTipCap())
+		}
+	}
+
+	pool.localBufferPool.PrintTxStats()
+	fmt.Println("length of all: ", pool.all.Slots())
+	fmt.Println("----------------------------------------------------")
+}