core/txpool/legacypool: add overflowpool for txs

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/heap.go

@@ -0,0 +1,171 @@
+package legacypool
+
+import (
+	"container/heap"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+// txHeapItem implements the Interface interface (https://pkg.go.dev/container/heap#Interface) of heap so that it can be heapified
+type txHeapItem struct {
+	tx        *types.Transaction
+	timestamp int64 // Unix timestamp (nanoseconds) of when the transaction was added
+	index     int
+}
+
+type txHeap []*txHeapItem
+
+func (h txHeap) Len() int { return len(h) }
+func (h txHeap) Less(i, j int) bool {
+	return h[i].timestamp < h[j].timestamp
+}
+func (h txHeap) Swap(i, j int) {
+	if i < 0 || j < 0 || i >= len(h) || j >= len(h) {
+		return // Silently fail if indices are out of bounds
+	}
+	h[i], h[j] = h[j], h[i]
+	if h[i] != nil {
+		h[i].index = i
+	}
+	if h[j] != nil {
+		h[j].index = j
+	}
+}
+
+func (h *txHeap) Push(x interface{}) {
+	item, ok := x.(*txHeapItem)
+	if !ok {
+		return
+	}
+	n := len(*h)
+	item.index = n
+	*h = append(*h, item)
+}
+
+func (h *txHeap) Pop() interface{} {
+	old := *h
+	n := len(old)
+	if n == 0 {
+		return nil // Return nil if the heap is empty
+	}
+	item := old[n-1]
+	old[n-1] = nil // avoid memory leak
+	*h = old[0 : n-1]
+	if item != nil {
+		item.index = -1 // for safety
+	}
+	return item
+}
+
+type TxOverflowPoolHeap struct {
+	txHeap    txHeap
+	index     map[common.Hash]*txHeapItem
+	mu        sync.RWMutex
+	maxSize   uint64
+	totalSize int
+}
+
+func NewTxOverflowPoolHeap(estimatedMaxSize uint64) *TxOverflowPoolHeap {
+	return &TxOverflowPoolHeap{
+		txHeap:  make(txHeap, 0, estimatedMaxSize),
+		index:   make(map[common.Hash]*txHeapItem, estimatedMaxSize),
+		maxSize: estimatedMaxSize,
+	}
+}
+
+func (tp *TxOverflowPoolHeap) Add(tx *types.Transaction) {
+	tp.mu.Lock()
+	defer tp.mu.Unlock()
+
+	if _, exists := tp.index[tx.Hash()]; exists {
+		// Transaction already in pool, ignore
+		return
+	}
+
+	if uint64(len(tp.txHeap)) >= tp.maxSize {
+		// Remove the oldest transaction to make space
+		oldestItem, ok := heap.Pop(&tp.txHeap).(*txHeapItem)
+		if !ok || oldestItem == nil {
+			return
+		}
+		delete(tp.index, oldestItem.tx.Hash())
+		tp.totalSize -= numSlots(oldestItem.tx)
+		OverflowPoolGauge.Dec(1)
+	}
+
+	item := &txHeapItem{
+		tx:        tx,
+		timestamp: time.Now().UnixNano(),
+	}
+	heap.Push(&tp.txHeap, item)
+	tp.index[tx.Hash()] = item
+	tp.totalSize += numSlots(tx)
+	OverflowPoolGauge.Inc(1)
+}
+
+func (tp *TxOverflowPoolHeap) Get(hash common.Hash) (*types.Transaction, bool) {
+	tp.mu.RLock()
+	defer tp.mu.RUnlock()
+	if item, ok := tp.index[hash]; ok {
+		return item.tx, true
+	}
+	return nil, false
+}
+
+func (tp *TxOverflowPoolHeap) Remove(hash common.Hash) {
+	tp.mu.Lock()
+	defer tp.mu.Unlock()
+	if item, ok := tp.index[hash]; ok {
+		heap.Remove(&tp.txHeap, item.index)
+		delete(tp.index, hash)
+		tp.totalSize -= numSlots(item.tx)
+		OverflowPoolGauge.Dec(1)
+	}
+}
+
+func (tp *TxOverflowPoolHeap) Flush(n int) []*types.Transaction {
+	tp.mu.Lock()
+	defer tp.mu.Unlock()
+	if n > tp.txHeap.Len() {
+		n = tp.txHeap.Len()
+	}
+	txs := make([]*types.Transaction, n)
+	for i := 0; i < n; i++ {
+		item, ok := heap.Pop(&tp.txHeap).(*txHeapItem)
+		if !ok || item == nil {
+			continue
+		}
+		txs[i] = item.tx
+		delete(tp.index, item.tx.Hash())
+		tp.totalSize -= numSlots(item.tx)
+	}
+
+	OverflowPoolGauge.Dec(int64(n))
+	return txs
+}
+
+func (tp *TxOverflowPoolHeap) Len() int {
+	tp.mu.RLock()
+	defer tp.mu.RUnlock()
+	return tp.txHeap.Len()
+}
+
+func (tp *TxOverflowPoolHeap) Size() int {
+	tp.mu.RLock()
+	defer tp.mu.RUnlock()
+	return tp.totalSize
+}
+
+func (tp *TxOverflowPoolHeap) PrintTxStats() {
+	tp.mu.RLock()
+	defer tp.mu.RUnlock()
+	for _, item := range tp.txHeap {
+		tx := item.tx
+		fmt.Printf("Hash: %s, Timestamp: %d, GasFeeCap: %s, GasTipCap: %s\n",
+			tx.Hash().String(), item.timestamp, tx.GasFeeCap().String(), tx.GasTipCap().String())
+	}
+}