New higher capacity Txpool

cmd/geth/main.go

@@ -91,6 +91,8 @@ var (
 		utils.TxPoolGlobalSlotsFlag,
 		utils.TxPoolAccountQueueFlag,
 		utils.TxPoolGlobalQueueFlag,
+		utils.TxPoolPool2SlotsFlag,
+		utils.TxPoolPool3SlotsFlag,
 		utils.TxPoolLifetimeFlag,
 		utils.TxPoolReannounceTimeFlag,
 		utils.BlobPoolDataDirFlag,

cmd/utils/flags.go

@@ -450,6 +450,18 @@ var (
 		Value:    ethconfig.Defaults.TxPool.GlobalQueue,
 		Category: flags.TxPoolCategory,
 	}
+	TxPoolPool2SlotsFlag = &cli.Uint64Flag{
+		Name:     "txpool.pool2slots",
+		Usage:    "Maximum number of transaction slots in pool 2",
+		Value:    ethconfig.Defaults.TxPool.Pool2Slots,
+		Category: flags.TxPoolCategory,
+	}
+	TxPoolPool3SlotsFlag = &cli.Uint64Flag{
+		Name:     "txpool.pool3slots",
+		Usage:    "Maximum number of transaction slots in pool 3",
+		Value:    ethconfig.Defaults.TxPool.Pool3Slots,
+		Category: flags.TxPoolCategory,
+	}
 	TxPoolLifetimeFlag = &cli.DurationFlag{
 		Name:     "txpool.lifetime",
 		Usage:    "Maximum amount of time non-executable transaction are queued",
@@ -1762,6 +1774,12 @@ func setTxPool(ctx *cli.Context, cfg *legacypool.Config) {
 	if ctx.IsSet(TxPoolGlobalQueueFlag.Name) {
 		cfg.GlobalQueue = ctx.Uint64(TxPoolGlobalQueueFlag.Name)
 	}
+	if ctx.IsSet(TxPoolPool2SlotsFlag.Name) {
+		cfg.Pool2Slots = ctx.Uint64(TxPoolPool2SlotsFlag.Name)
+	}
+	if ctx.IsSet(TxPoolPool3SlotsFlag.Name) {
+		cfg.Pool3Slots = ctx.Uint64(TxPoolPool3SlotsFlag.Name)
+	}
 	if ctx.IsSet(TxPoolLifetimeFlag.Name) {
 		cfg.Lifetime = ctx.Duration(TxPoolLifetimeFlag.Name)
 	}
@@ -2292,6 +2310,8 @@ func EnableNodeInfo(poolConfig *legacypool.Config, nodeInfo *p2p.NodeInfo) Setup
 			"GlobalSlots":  poolConfig.GlobalSlots,
 			"AccountQueue": poolConfig.AccountQueue,
 			"GlobalQueue":  poolConfig.GlobalQueue,
+			"Pool2Slots":   poolConfig.Pool2Slots,
+			"Pool3Slots":   poolConfig.Pool3Slots,
 			"Lifetime":     poolConfig.Lifetime,
 		})
 	}

core/events.go

@@ -22,7 +22,13 @@ import (
 )
 
 // NewTxsEvent is posted when a batch of transactions enters the transaction pool.
-type NewTxsEvent struct{ Txs []*types.Transaction }
+type NewTxsEvent struct {
+	Txs []*types.Transaction
+	// Static bool is Whether to send to only Static peer or not.
+	// This is because at high traffic we still want to broadcast transactions to at least some peers so that we
+	// minimize the transaction lost.
+	Static bool
+}
 
 // ReannoTxsEvent is posted when a batch of local pending transactions exceed a specified duration.
 type ReannoTxsEvent struct{ Txs []*types.Transaction }

core/txpool/errors.go

@@ -30,6 +30,10 @@ var (
 	// configured for the transaction pool.
 	ErrUnderpriced = errors.New("transaction underpriced")
 
+	// ErrUnderpriced is returned if a transaction's gas price is below the minimum
+	// configured for the transaction pool.
+	ErrUnderpricedTransferredtoAnotherPool = errors.New("transaction underpriced, so it is either in pool2 or pool3")
+
 	// 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")

core/txpool/legacypool/buffer.go

@@ -0,0 +1,111 @@
+package legacypool
+
+import (
+	containerList "container/list"
+	"fmt"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+type LRUBuffer struct {
+	capacity int
+	buffer   *containerList.List
+	index    map[common.Hash]*containerList.Element
+	mu       sync.Mutex
+	size     int // Total number of slots used
+}
+
+func NewLRUBuffer(capacity int) *LRUBuffer {
+	return &LRUBuffer{
+		capacity: capacity,
+		buffer:   containerList.New(),
+		index:    make(map[common.Hash]*containerList.Element),
+		size:     0, // Initialize size to 0
+	}
+}
+
+func (lru *LRUBuffer) Add(tx *types.Transaction) {
+	lru.mu.Lock()
+	defer lru.mu.Unlock()
+
+	if elem, ok := lru.index[tx.Hash()]; ok {
+		lru.buffer.MoveToFront(elem)
+		return
+	}
+
+	txSlots := numSlots(tx)
+
+	// Remove elements until there is enough capacity
+	for lru.size+txSlots > lru.capacity && lru.buffer.Len() > 0 {
+		back := lru.buffer.Back()
+		removedTx := back.Value.(*types.Transaction)
+		lru.buffer.Remove(back)
+		delete(lru.index, removedTx.Hash())
+		lru.size -= numSlots(removedTx) // Decrease size by the slots of the removed transaction
+	}
+
+	elem := lru.buffer.PushFront(tx)
+	lru.index[tx.Hash()] = elem
+	lru.size += txSlots // Increase size by the slots of the new transaction
+}
+
+func (lru *LRUBuffer) Get(hash common.Hash) (*types.Transaction, bool) {
+	lru.mu.Lock()
+	defer lru.mu.Unlock()
+
+	if elem, ok := lru.index[hash]; ok {
+		lru.buffer.MoveToFront(elem)
+		return elem.Value.(*types.Transaction), true
+	}
+	return nil, false
+}
+
+func (lru *LRUBuffer) Flush(maxTransactions int) []*types.Transaction {
+	lru.mu.Lock()
+	defer lru.mu.Unlock()
+
+	txs := make([]*types.Transaction, 0, maxTransactions)
+	count := 0
+	for count < maxTransactions && lru.buffer.Len() > 0 {
+		back := lru.buffer.Back()
+		removedTx := back.Value.(*types.Transaction)
+		txs = append(txs, removedTx)
+		lru.buffer.Remove(back)
+		delete(lru.index, removedTx.Hash())
+		lru.size -= numSlots(removedTx) // Decrease size by the slots of the removed transaction
+		count++
+	}
+	return txs
+}
+
+// New method to get the current size of the buffer in terms of slots
+func (lru *LRUBuffer) Size() int {
+	lru.mu.Lock()
+	defer lru.mu.Unlock()
+	return lru.size
+}
+
+// New iterator method to iterate over all transactions, ONLY used for printing and debugging
+func (lru *LRUBuffer) iterate() <-chan *types.Transaction {
+	ch := make(chan *types.Transaction)
+	go func() {
+		lru.mu.Lock()
+		defer lru.mu.Unlock()
+		defer close(ch)
+
+		for e := lru.buffer.Front(); e != nil; e = e.Next() {
+			ch <- e.Value.(*types.Transaction)
+		}
+	}()
+	return ch
+}
+
+func (lru *LRUBuffer) PrintTxStats() {
+	// Iterating over the transactions
+	for tx := range lru.iterate() {
+		// Print transaction details or process them as needed
+		fmt.Println(tx.Hash().String(), tx.GasFeeCap().String(), tx.GasTipCap().String())
+	}
+}

core/txpool/legacypool/buffer_test.go

@@ -0,0 +1,126 @@
+package legacypool
+
+import (
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+// Helper function to create a dummy transaction of specified size
+func createDummyTransaction(size int) *types.Transaction {
+	data := make([]byte, size)
+	return types.NewTransaction(0, common.Address{}, nil, 0, nil, data)
+}
+
+func TestNewLRUBuffer(t *testing.T) {
+	capacity := 10
+	lru := NewLRUBuffer(capacity)
+	if lru.capacity != capacity {
+		t.Errorf("expected capacity %d, got %d", capacity, lru.capacity)
+	}
+	if lru.buffer.Len() != 0 {
+		t.Errorf("expected buffer length 0, got %d", lru.buffer.Len())
+	}
+	if len(lru.index) != 0 {
+		t.Errorf("expected index length 0, got %d", len(lru.index))
+	}
+	if lru.size != 0 {
+		t.Errorf("expected size 0, got %d", lru.size)
+	}
+}
+
+func TestAddAndGet(t *testing.T) {
+	lru := NewLRUBuffer(10)
+
+	tx1 := createDummyTransaction(500)
+	tx2 := createDummyTransaction(1500)
+
+	lru.Add(tx1)
+	lru.Add(tx2)
+
+	if lru.Size() != 2 {
+		t.Errorf("expected size 2, got %d", lru.Size())
+	}
+
+	retrievedTx, ok := lru.Get(tx1.Hash())
+	if !ok || retrievedTx.Hash() != tx1.Hash() {
+		t.Errorf("failed to retrieve tx1")
+	}
+
+	retrievedTx, ok = lru.Get(tx2.Hash())
+	if !ok || retrievedTx.Hash() != tx2.Hash() {
+		t.Errorf("failed to retrieve tx2")
+	}
+}
+
+func TestBufferCapacity(t *testing.T) {
+	lru := NewLRUBuffer(2) // Capacity in slots
+
+	tx1 := createDummyTransaction(500)  // 1 slot
+	tx2 := createDummyTransaction(1500) // 1 slot
+	tx3 := createDummyTransaction(1000) // 1 slot
+
+	lru.Add(tx1)
+	lru.Add(tx2)
+
+	if lru.Size() != 2 {
+		t.Errorf("expected size 2, got %d", lru.Size())
+	}
+
+	lru.Add(tx3)
+
+	if lru.Size() != 2 {
+		t.Errorf("expected size 2 after adding tx3, got %d", lru.Size())
+	}
+
+	if _, ok := lru.Get(tx1.Hash()); ok {
+		t.Errorf("expected tx1 to be evicted")
+	}
+}
+
+func TestFlush(t *testing.T) {
+	lru := NewLRUBuffer(10)
+
+	tx1 := createDummyTransaction(500)
+	tx2 := createDummyTransaction(1500)
+	tx3 := createDummyTransaction(1000)
+
+	lru.Add(tx1)
+	lru.Add(tx2)
+	lru.Add(tx3)
+
+	flushedTxs := lru.Flush(2)
+
+	if len(flushedTxs) != 2 {
+		t.Errorf("expected to flush 2 transactions, got %d", len(flushedTxs))
+	}
+
+	expectedSize := 1
+	actualSize := lru.Size()
+	if expectedSize != actualSize {
+		t.Errorf("expected size after flush %d, got %d", expectedSize, actualSize)
+	}
+}
+
+func TestSize(t *testing.T) {
+	lru := NewLRUBuffer(10)
+
+	tx1 := createDummyTransaction(500)  // 1 slot
+	tx2 := createDummyTransaction(1500) // 2 slots
+
+	lru.Add(tx1)
+	if lru.Size() != 1 {
+		t.Errorf("expected size 1, got %d", lru.Size())
+	}
+
+	lru.Add(tx2)
+	if lru.Size() != 2 {
+		t.Errorf("expected size 2, got %d", lru.Size())
+	}
+
+	lru.Flush(1)
+	if lru.Size() != 1 {
+		t.Errorf("expected size 1 after flush, got %d", lru.Size())
+	}
+}