feat(miner): account fetch limit (#939)

* feat(worker): try to limit the number of txns miner has to deal with (#745)

to reduce the effect of having a huge backlog on performance

* fix(worker): set default account fetch limit (#756)

* fix

* update miner/worker.go

* fix

---------

Co-authored-by: Ömer Faruk Irmak <omerfirmak@gmail.com>
Co-authored-by: Péter Garamvölgyi <peter@scroll.io>

cmd/geth/main.go

@@ -124,6 +124,7 @@ var (
 		utils.MinerRecommitIntervalFlag,
 		utils.MinerNewPayloadTimeout,
 		utils.MinerStoreSkippedTxTracesFlag,
+		utils.MinerMaxAccountsNumFlag,
 		utils.NATFlag,
 		utils.NoDiscoverFlag,
 		utils.DiscoveryV4Flag,

cmd/utils/flags.go

@@ -526,6 +526,13 @@ var (
 	MinerStoreSkippedTxTracesFlag = &cli.BoolFlag{
 		Name:     "miner.storeskippedtxtraces",
 		Usage:    "Store the wrapped traces when storing a skipped tx",
+		Category: flags.MinerCategory,
+	}
+	MinerMaxAccountsNumFlag = &cli.IntFlag{
+		Name:     "miner.maxaccountsnum",
+		Usage:    "Maximum number of accounts that miner will fetch the pending transactions of when building a new block",
+		Value:    math.MaxInt,
+		Category: flags.MinerCategory,
 	}
 
 	// Account settings
@@ -1697,6 +1704,9 @@ func setMiner(ctx *cli.Context, cfg *miner.Config) {
 	if ctx.IsSet(MinerStoreSkippedTxTracesFlag.Name) {
 		cfg.StoreSkippedTxTraces = ctx.Bool(MinerStoreSkippedTxTracesFlag.Name)
 	}
+	if ctx.IsSet(MinerMaxAccountsNumFlag.Name) {
+		cfg.MaxAccountsNum = ctx.Int(MinerMaxAccountsNumFlag.Name)
+	}
 }
 
 func setRequiredBlocks(ctx *cli.Context, cfg *ethconfig.Config) {

core/txpool/blobpool/blobpool.go

@@ -29,6 +29,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/holiman/billy"
+	"github.com/holiman/uint256"
 	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/consensus/misc/eip1559"
 	"github.com/scroll-tech/go-ethereum/consensus/misc/eip4844"
@@ -42,8 +44,6 @@ import (
 	"github.com/scroll-tech/go-ethereum/params"
 	"github.com/scroll-tech/go-ethereum/rlp"
 	"github.com/scroll-tech/go-ethereum/rollup/fees"
-	"github.com/holiman/billy"
-	"github.com/holiman/uint256"
 )
 
 const (
@@ -1381,6 +1381,16 @@ func (p *BlobPool) drop() {
 // Pending retrieves all currently processable transactions, grouped by origin
 // account and sorted by nonce.
 func (p *BlobPool) Pending(enforceTips bool) map[common.Address][]*txpool.LazyTransaction {
+	return p.pendingWithMax(enforceTips, math.MaxInt)
+}
+
+// PendingWithMax works similar to Pending but allows setting an upper limit on how many
+// accounts to return
+func (p *BlobPool) PendingWithMax(enforceTips bool, maxAccountsNum int) map[common.Address][]*txpool.LazyTransaction {
+	return p.pendingWithMax(enforceTips, maxAccountsNum)
+}
+
+func (p *BlobPool) pendingWithMax(enforceTips bool, maxAccountsNum int) map[common.Address][]*txpool.LazyTransaction {
 	// Track the amount of time waiting to retrieve the list of pending blob txs
 	// from the pool and the amount of time actually spent on assembling the data.
 	// The latter will be pretty much moot, but we've kept it to have symmetric
@@ -1411,6 +1421,9 @@ func (p *BlobPool) Pending(enforceTips bool) map[common.Address][]*txpool.LazyTr
 		if len(lazies) > 0 {
 			pending[addr] = lazies
 		}
+		if len(pending) >= maxAccountsNum {
+			break
+		}
 	}
 	return pending
 }

core/txpool/legacypool/legacypool.go

@@ -523,6 +523,16 @@ func (pool *LegacyPool) ContentFrom(addr common.Address) ([]*types.Transaction,
 // transactions and only return those whose **effective** tip is large enough in
 // the next pending execution environment.
 func (pool *LegacyPool) Pending(enforceTips bool) map[common.Address][]*txpool.LazyTransaction {
+	return pool.pendingWithMax(enforceTips, math.MaxInt)
+}
+
+// PendingWithMax works similar to Pending but allows setting an upper limit on how many
+// accounts to return
+func (pool *LegacyPool) PendingWithMax(enforceTips bool, maxAccountsNum int) map[common.Address][]*txpool.LazyTransaction {
+	return pool.pendingWithMax(enforceTips, maxAccountsNum)
+}
+
+func (pool *LegacyPool) pendingWithMax(enforceTips bool, maxAccountsNum int) map[common.Address][]*txpool.LazyTransaction {
 	pool.mu.Lock()
 	defer pool.mu.Unlock()
 
@@ -554,6 +564,9 @@ func (pool *LegacyPool) Pending(enforceTips bool) map[common.Address][]*txpool.L
 				}
 			}
 			pending[addr] = lazies
+			if len(pending) >= maxAccountsNum {
+				break
+			}
 		}
 	}
 	return pending

core/txpool/legacypool/legacypool_test.go

@@ -29,6 +29,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/core"
 	"github.com/scroll-tech/go-ethereum/core/rawdb"
@@ -2628,3 +2630,27 @@ func BenchmarkMultiAccountBatchInsert(b *testing.B) {
 		pool.addRemotesSync([]*types.Transaction{tx})
 	}
 }
+
+func TestPoolPending(t *testing.T) {
+	// Generate a batch of transactions to enqueue into the pool
+	pool, _ := setupPool()
+	defer pool.Close()
+	numTxns := 100
+	batches := make(types.Transactions, numTxns)
+	for i := 0; i < numTxns; i++ {
+		key, _ := crypto.GenerateKey()
+		account := crypto.PubkeyToAddress(key.PublicKey)
+		pool.currentState.AddBalance(account, big.NewInt(1000000))
+		tx := transaction(uint64(0), 100000, key)
+		batches[i] = tx
+	}
+	// Benchmark importing the transactions into the queue
+	for _, tx := range batches {
+		pool.addRemotesSync([]*types.Transaction{tx})
+	}
+
+	assert.Len(t, pool.Pending(false), numTxns)
+
+	maxAccounts := 10
+	assert.Len(t, pool.PendingWithMax(false, maxAccounts), maxAccounts)
+}

core/txpool/subpool.go

@@ -110,6 +110,8 @@ type SubPool interface {
 	// account and sorted by nonce.
 	Pending(enforceTips bool) map[common.Address][]*LazyTransaction
 
+	PendingWithMax(enforceTips bool, maxAccountsNum int) map[common.Address][]*LazyTransaction
+
 	// SubscribeTransactions subscribes to new transaction events. The subscriber
 	// can decide whether to receive notifications only for newly seen transactions
 	// or also for reorged out ones.

core/txpool/txpool.go

@@ -318,6 +318,16 @@ func (p *TxPool) Pending(enforceTips bool) map[common.Address][]*LazyTransaction
 	return txs
 }
 
+func (p *TxPool) PendingWithMax(enforceTips bool, maxAccountsNum int) map[common.Address][]*LazyTransaction {
+	txs := make(map[common.Address][]*LazyTransaction)
+	for _, subpool := range p.subpools {
+		for addr, set := range subpool.PendingWithMax(enforceTips, maxAccountsNum) {
+			txs[addr] = set
+		}
+	}
+	return txs
+}
+
 // SubscribeTransactions registers a subscription for new transaction events,
 // supporting feeding only newly seen or also resurrected transactions.
 func (p *TxPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool) event.Subscription {

miner/miner.go

@@ -59,6 +59,7 @@ type Config struct {
 	NewPayloadTimeout time.Duration // The maximum time allowance for creating a new payload
 
 	StoreSkippedTxTraces bool // Whether store the wrapped traces when storing a skipped tx
+	MaxAccountsNum       int  // Maximum number of accounts that miner will fetch the pending transactions of when building a new block
 }
 
 // DefaultConfig contains default settings for miner.

miner/miner_test.go

@@ -19,6 +19,7 @@ package miner
 
 import (
 	"errors"
+	"math"
 	"math/big"
 	"testing"
 	"time"
@@ -302,6 +303,7 @@ func createMiner(t *testing.T) (*Miner, *event.TypeMux, func(skipMiner bool)) {
 	// Create Ethash config
 	config := Config{
 		Etherbase:      common.HexToAddress("123456789"),
+		MaxAccountsNum: math.MaxInt,
 	}
 	// Create chainConfig
 	chainDB := rawdb.NewMemoryDatabase()

miner/scroll_worker.go

@@ -19,6 +19,7 @@ package miner
 import (
 	"bytes"
 	"errors"
+	"math"
 	"math/big"
 	"sync"
 	"sync/atomic"
@@ -162,6 +163,12 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus
 	// Subscribe events for blockchain
 	worker.chainHeadSub = eth.BlockChain().SubscribeChainHeadEvent(worker.chainHeadCh)
 
+	// Sanitize account fetch limit.
+	if worker.config.MaxAccountsNum == 0 {
+		log.Warn("Sanitizing miner account fetch limit", "provided", worker.config.MaxAccountsNum, "updated", math.MaxInt)
+		worker.config.MaxAccountsNum = math.MaxInt
+	}
+
 	worker.wg.Add(1)
 	go worker.mainLoop()
 
@@ -382,7 +389,7 @@ func (w *worker) startNewPipeline(timestamp int64) {
 
 	tidyPendingStart := time.Now()
 	// Fill the block with all available pending transactions.
-	pending := w.eth.TxPool().Pending(false)
+	pending := w.eth.TxPool().PendingWithMax(false, w.config.MaxAccountsNum)
 	// Split the pending transactions into locals and remotes
 	localTxs, remoteTxs := make(map[common.Address][]*txpool.LazyTransaction), pending
 	for _, account := range w.eth.TxPool().Locals() {

miner/worker.go

@@ -353,6 +353,12 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus
 	}
 	worker.newpayloadTimeout = newpayloadTimeout
 
+	// Sanitize account fetch limit.
+	if worker.config.MaxAccountsNum == 0 {
+		log.Warn("Sanitizing miner account fetch limit", "provided", worker.config.MaxAccountsNum, "updated", math.MaxInt)
+		worker.config.MaxAccountsNum = math.MaxInt
+	}
+
 	worker.wg.Add(4)
 	go worker.mainLoop()
 	go worker.newWorkLoop(recommit)
@@ -1420,7 +1426,7 @@ func (w *worker) fillTransactions(interrupt *atomic.Int32, env *environment) err
 	}
 
 	tidyPendingStart := time.Now()
-	pending := w.eth.TxPool().Pending(true)
+	pending := w.eth.TxPool().PendingWithMax(true, w.config.MaxAccountsNum)
 
 	// Split the pending transactions into locals and remotes.
 	localTxs, remoteTxs := make(map[common.Address][]*txpool.LazyTransaction), pending

miner/worker_test.go

@@ -17,6 +17,7 @@
 package miner
 
 import (
+	"math"
 	"math/big"
 	"testing"
 	"time"
@@ -69,6 +70,7 @@ var (
 	testConfig = &Config{
 		Recommit:       time.Second,
 		GasCeil:        params.GenesisGasLimit,
+		MaxAccountsNum: math.MaxInt,
 	}
 )