execution: add --fcu.timeout and --fcu.background.prune flags (#18723)

trying out some new flags for getting the gigagas execution client
benchmarks to work

Author: taratorio

cmd/utils/flags.go

@@ -1103,6 +1103,16 @@ var (
 		Name:  "experimental.always-generate-changesets",
 		Usage: "Allows to override changesets generation logic",
 	}
+	FcuTimeoutFlag = cli.DurationFlag{
+		Name:  "fcu.timeout",
+		Usage: "FCU timeout before it switches to being process async (use 0 to disable)",
+		Value: ethconfig.Defaults.FcuTimeout,
+	}
+	FcuBackgroundPruneFlag = cli.BoolFlag{
+		Name:  "fcu.background.prune",
+		Usage: "Enables background pruning post fcu",
+		Value: ethconfig.Defaults.FcuBackgroundPrune,
+	}
 	// ErigonDB geometry settings
 	ErigonDBStepSizeFlag = cli.Uint64Flag{
 		Name:  "erigondb.override.stepsize",
@@ -1919,7 +1929,8 @@ func SetEthConfig(ctx *cli.Context, nodeConfig *nodecfg.Config, cfg *ethconfig.C
 
 	cfg.ErigonDBStepSize = ctx.Int(ErigonDBStepSizeFlag.Name)
 	cfg.ErigonDBStepsInFrozenFile = ctx.Int(ErigonDBStepsInFrozenFileFlag.Name)
-
+	cfg.FcuTimeout = ctx.Duration(FcuTimeoutFlag.Name)
+	cfg.FcuBackgroundPrune = ctx.Bool(FcuBackgroundPruneFlag.Name)
 	if ctx.IsSet(RPCGlobalGasCapFlag.Name) {
 		cfg.RPCGasCap = ctx.Uint64(RPCGlobalGasCapFlag.Name)
 	}

execution/engineapi/engine_block_downloader/block_downloader.go

@@ -38,10 +38,6 @@ import (
 	"github.com/erigontech/erigon/node/gointerfaces/executionproto"
 )
 
-const (
-	forkchoiceTimeoutMillis = 5000
-)
-
 type Status int
 
 const (
@@ -80,14 +76,16 @@ func NewEngineBlockDownloader(
 	if err != nil {
 		panic(fmt.Errorf("failed to create badHeaders cache: %w", err))
 	}
+	// the block downloader has fcuTimeout=0 to avoid having to deal with async fcu
+	chainRW := eth1_chain_reader.NewChainReaderEth1(config, executionClient, 0 /* fcuTimeout */)
 	return &EngineBlockDownloader{
 		backgroundCtx: ctx,
 		db:            db,
 		status:        s,
 		syncCfg:       syncCfg,
 		logger:        logger,
 		blockReader:   blockReader,
-		chainRW:       eth1_chain_reader.NewChainReaderEth1(config, executionClient, forkchoiceTimeoutMillis),
+		chainRW:       chainRW,
 		bbd:           bbd,
 		badHeaders:    badHeaders,
 	}

execution/engineapi/engine_server.go

@@ -83,8 +83,6 @@ type EngineServer struct {
 	printPectraBanner bool
 }
 
-const DefaultFcuTimeout = 1 * time.Second
-
 func NewEngineServer(
 	logger log.Logger,
 	config *chain.Config,
@@ -96,9 +94,6 @@ func NewEngineServer(
 	txPool txpoolproto.TxpoolClient,
 	fcuTimeout time.Duration,
 ) *EngineServer {
-	if fcuTimeout == 0 {
-		fcuTimeout = DefaultFcuTimeout
-	}
 	chainRW := eth1_chain_reader.NewChainReaderEth1(config, executionService, uint64(fcuTimeout.Milliseconds()))
 	srv := &EngineServer{
 		logger:            logger,

execution/engineapi/engine_server_test.go

@@ -112,7 +112,8 @@ func TestGetBlobsV1(t *testing.T) {
 
 	executionRpc := direct.NewExecutionClientDirect(mockSentry.Eth1ExecutionService)
 	eth := rpcservices.NewRemoteBackend(nil, mockSentry.DB, mockSentry.BlockReader)
-	engineServer := NewEngineServer(mockSentry.Log, mockSentry.ChainConfig, executionRpc, nil, false, false, true, txPool, DefaultFcuTimeout)
+	fcuTimeout := ethconfig.Defaults.FcuTimeout
+	engineServer := NewEngineServer(mockSentry.Log, mockSentry.ChainConfig, executionRpc, nil, false, false, true, txPool, fcuTimeout)
 	ctx, cancel := context.WithCancel(ctx)
 	var eg errgroup.Group
 	t.Cleanup(func() {
@@ -162,7 +163,8 @@ func TestGetBlobsV2(t *testing.T) {
 
 	executionRpc := direct.NewExecutionClientDirect(mockSentry.Eth1ExecutionService)
 	eth := rpcservices.NewRemoteBackend(nil, mockSentry.DB, mockSentry.BlockReader)
-	engineServer := NewEngineServer(mockSentry.Log, mockSentry.ChainConfig, executionRpc, nil, false, false, true, txPool, DefaultFcuTimeout)
+	fcuTimeout := ethconfig.Defaults.FcuTimeout
+	engineServer := NewEngineServer(mockSentry.Log, mockSentry.ChainConfig, executionRpc, nil, false, false, true, txPool, fcuTimeout)
 	ctx, cancel := context.WithCancel(ctx)
 	var eg errgroup.Group
 	t.Cleanup(func() {

execution/eth1/ethereum_execution.go

@@ -21,7 +21,6 @@ import (
 	"errors"
 	"math/big"
 	"strings"
-	"sync/atomic"
 	"time"
 
 	"golang.org/x/sync/semaphore"
@@ -120,8 +119,7 @@ type EthereumExecutionModule struct {
 	// rules engine
 	engine rules.Engine
 
-	doingPostForkchoice atomic.Bool
-
+	fcuBackgroundPrune bool
 	// metrics for average mgas/sec
 	avgMgasSec float64
 
@@ -137,6 +135,7 @@ func NewEthereumExecutionModule(blockReader services.FullBlockReader, db kv.Temp
 	logger log.Logger, engine rules.Engine,
 	syncCfg ethconfig.Sync,
 	ctx context.Context,
+	fcuBackgroundPrune bool,
 ) *EthereumExecutionModule {
 	return &EthereumExecutionModule{
 		blockReader:         blockReader,
@@ -155,6 +154,7 @@ func NewEthereumExecutionModule(blockReader services.FullBlockReader, db kv.Temp
 		engine:              engine,
 		syncCfg:             syncCfg,
 		bacgroundCtx:        ctx,
+		fcuBackgroundPrune:  fcuBackgroundPrune,
 	}
 }
 

execution/eth1/forkchoice.go

@@ -180,7 +180,12 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
 		}, false)
 		return
 	}
-	defer e.semaphore.Release(1)
+	shouldReleaseSema := true
+	defer func() {
+		if shouldReleaseSema {
+			e.semaphore.Release(1)
+		}
+	}()
 
 	defer UpdateForkChoiceDuration(time.Now())
 
@@ -602,8 +607,22 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
 			e.logger.Info("head updated", logArgs...)
 		}
 	}
-	if *headNumber >= startPruneFrom {
-		e.runPostForkchoiceInBackground(initialCycle)
+	if e.fcuBackgroundPrune {
+		shouldReleaseSema = false // pass on semaphore to background goroutine doing post fcu processing
+		go func() {
+			defer e.semaphore.Release(1)
+			err := e.runPostForkchoice(*headNumber, initialCycle)
+			if err != nil {
+				e.logger.Error("run post fork choice in background", "error", err)
+			}
+		}()
+	} else {
+		err := e.runPostForkchoice(*headNumber, initialCycle)
+		if err != nil {
+			e.logger.Error("run post fork choice in background", "error", err)
+			sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, stateFlushingInParallel)
+			return
+		}
 	}
 
 	sendForkchoiceReceiptWithoutWaiting(outcomeCh, &executionproto.ForkChoiceReceipt{
@@ -613,18 +632,9 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
 	}, stateFlushingInParallel)
 }
 
-func (e *EthereumExecutionModule) runPostForkchoiceInBackground(initialCycle bool) {
-	if !e.doingPostForkchoice.CompareAndSwap(false, true) {
-		return
-	}
-	go func() {
-		defer e.doingPostForkchoice.Store(false)
-		var timings []interface{}
-		// Wait for semaphore to be available
-		if e.semaphore.Acquire(e.bacgroundCtx, 1) != nil {
-			return
-		}
-		defer e.semaphore.Release(1)
+func (e *EthereumExecutionModule) runPostForkchoice(headNumber uint64, initialCycle bool) error {
+	var timings []interface{}
+	if headNumber >= startPruneFrom {
 		pruneStart := time.Now()
 		defer UpdateForkChoicePruneDuration(pruneStart)
 		if err := e.db.Update(e.bacgroundCtx, func(tx kv.RwTx) error {
@@ -634,16 +644,14 @@ func (e *EthereumExecutionModule) runPostForkchoiceInBackground(initialCycle boo
 			if pruneTimings := e.executionPipeline.PrintTimings(); len(pruneTimings) > 0 {
 				timings = append(timings, pruneTimings...)
 			}
-
 			return nil
 		}); err != nil {
-			e.logger.Error("runPostForkchoiceInBackground", "error", err)
-			return
+			return err
 		}
-
-		if len(timings) > 0 {
-			timings = append(timings, "initialCycle", initialCycle)
-			e.logger.Info("Timings: Post-Forkchoice", timings...)
-		}
-	}()
+	}
+	if len(timings) > 0 {
+		timings = append(timings, "initialCycle", initialCycle)
+		e.logger.Info("Timings: Post-Forkchoice", timings...)
+	}
+	return nil
 }

execution/tests/mock/mock_sentry.go

@@ -548,7 +548,7 @@ func MockWithEverything(tb testing.TB, gspec *types.Genesis, key *ecdsa.PrivateK
 	pipelineStages := stageloop.NewPipelineStages(mock.Ctx, db, &cfg, mock.sentriesClient, mock.Notifications, snapDownloader, mock.BlockReader, blockRetire, nil, forkValidator, tracer)
 	mock.posStagedSync = stagedsync.New(cfg.Sync, pipelineStages, stagedsync.PipelineUnwindOrder, stagedsync.PipelinePruneOrder, logger, stages.ModeApplyingBlocks)
 
-	mock.Eth1ExecutionService = eth1.NewEthereumExecutionModule(mock.BlockReader, mock.DB, mock.posStagedSync, forkValidator, mock.ChainConfig, assembleBlockPOS, nil, mock.Notifications.Accumulator, mock.Notifications.RecentLogs, mock.Notifications.StateChangesConsumer, logger, engine, cfg.Sync, ctx)
+	mock.Eth1ExecutionService = eth1.NewEthereumExecutionModule(mock.BlockReader, mock.DB, mock.posStagedSync, forkValidator, mock.ChainConfig, assembleBlockPOS, nil, mock.Notifications.Accumulator, mock.Notifications.RecentLogs, mock.Notifications.StateChangesConsumer, logger, engine, cfg.Sync, ctx, cfg.FcuBackgroundPrune)
 
 	mock.sentriesClient.Hd.StartPoSDownloader(mock.Ctx, sendHeaderRequest, penalize)
 

node/cli/default_flags.go

@@ -43,7 +43,8 @@ var DefaultFlags = []cli.Flag{
 	&PruneBlocksDistanceFlag,
 	&PruneModeFlag,
 	&utils.KeepExecutionProofsFlag,
-
+	&utils.FcuTimeoutFlag,
+	&utils.FcuBackgroundPruneFlag,
 	&BatchSizeFlag,
 	&BodyCacheLimitFlag,
 	&DatabaseVerbosityFlag,

node/eth/backend.go

@@ -990,12 +990,12 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 
 	pipelineStages := stageloop.NewPipelineStages(ctx, backend.chainDB, config, backend.sentriesClient, backend.notifications, backend.downloaderClient, blockReader, blockRetire, backend.silkworm, backend.forkValidator, tracer)
 	backend.pipelineStagedSync = stagedsync.New(config.Sync, pipelineStages, stagedsync.PipelineUnwindOrder, stagedsync.PipelinePruneOrder, logger, stages.ModeApplyingBlocks)
-	backend.eth1ExecutionServer = eth1.NewEthereumExecutionModule(blockReader, backend.chainDB, backend.pipelineStagedSync, backend.forkValidator, chainConfig, assembleBlockPOS, hook, backend.notifications.Accumulator, backend.notifications.RecentLogs, backend.notifications.StateChangesConsumer, logger, backend.engine, config.Sync, ctx)
+	backend.eth1ExecutionServer = eth1.NewEthereumExecutionModule(blockReader, backend.chainDB, backend.pipelineStagedSync, backend.forkValidator, chainConfig, assembleBlockPOS, hook, backend.notifications.Accumulator, backend.notifications.RecentLogs, backend.notifications.StateChangesConsumer, logger, backend.engine, config.Sync, ctx, config.FcuBackgroundPrune)
 	executionRpc := direct.NewExecutionClientDirect(backend.eth1ExecutionServer)
 
 	var executionEngine executionclient.ExecutionEngine
 
-	executionEngine, err = executionclient.NewExecutionClientDirect(eth1_chain_reader.NewChainReaderEth1(chainConfig, executionRpc, 1000), txPoolRpcClient)
+	executionEngine, err = executionclient.NewExecutionClientDirect(eth1_chain_reader.NewChainReaderEth1(chainConfig, executionRpc, uint64(config.FcuTimeout.Milliseconds())), txPoolRpcClient)
 	if err != nil {
 		return nil, err
 	}

node/ethconfig/config.go

@@ -117,6 +117,8 @@ var Defaults = Config{
 
 	ErigonDBStepSize:          config3.DefaultStepSize,
 	ErigonDBStepsInFrozenFile: config3.DefaultStepsInFrozenFile,
+	FcuTimeout:                1 * time.Second,
+	FcuBackgroundPrune:        true,
 }
 
 const DefaultChainDBPageSize = 16 * datasize.KB
@@ -280,7 +282,8 @@ type Config struct {
 	ErigonDBStepsInFrozenFile int
 
 	// fork choice update timeout
-	FcuTimeout time.Duration
+	FcuTimeout         time.Duration
+	FcuBackgroundPrune bool
 }
 
 type Sync struct {