op-challenger: Add CLI option for max concurrency

op-challenger/challenger.go

@@ -11,6 +11,9 @@ import (
 
 // Main is the programmatic entry-point for running op-challenger
 func Main(ctx context.Context, logger log.Logger, cfg *config.Config) error {
+	if err := cfg.Check(); err != nil {
+		return err
+	}
 	service, err := fault.NewService(ctx, logger, cfg)
 	if err != nil {
 		return fmt.Errorf("failed to create the fault service: %w", err)

op-challenger/challenger_test.go

+package op_challenger
+
+import (
+	"context"
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/op-challenger/config"
+	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMainShouldReturnErrorWhenConfigInvalid(t *testing.T) {
+	cfg := &config.Config{}
+	err := Main(context.Background(), testlog.Logger(t, log.LvlInfo), cfg)
+	require.ErrorIs(t, err, cfg.Check())
+}

op-challenger/cmd/main_test.go

@@ -147,6 +147,28 @@ func TestAgreeWithProposedOutput(t *testing.T) {
 	})
 }
 
+func TestMaxConcurrency(t *testing.T) {
+	t.Run("Valid", func(t *testing.T) {
+		expected := uint(345)
+		cfg := configForArgs(t, addRequiredArgs(config.TraceTypeAlphabet, "--max-concurrency", "345"))
+		require.Equal(t, expected, cfg.MaxConcurrency)
+	})
+
+	t.Run("Invalid", func(t *testing.T) {
+		verifyArgsInvalid(
+			t,
+			"invalid value \"abc\" for flag -max-concurrency",
+			addRequiredArgs(config.TraceTypeAlphabet, "--max-concurrency", "abc"))
+	})
+
+	t.Run("Zero", func(t *testing.T) {
+		verifyArgsInvalid(
+			t,
+			"max-concurrency must not be 0",
+			addRequiredArgs(config.TraceTypeAlphabet, "--max-concurrency", "0"))
+	})
+}
+
 func TestCannonBin(t *testing.T) {
 	t.Run("NotRequiredForAlphabetTrace", func(t *testing.T) {
 		configForArgs(t, addRequiredArgsExcept(config.TraceTypeAlphabet, "--cannon-bin"))

op-challenger/config/config.go

@@ -3,6 +3,7 @@ package config
 import (
 	"errors"
 	"fmt"
+	"runtime"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -16,6 +17,7 @@ import (
 var (
 	ErrMissingTraceType              = errors.New("missing trace type")
 	ErrMissingDatadir                = errors.New("missing datadir")
+	ErrMaxConcurrencyZero            = errors.New("max concurrency must not be 0")
 	ErrMissingCannonL2               = errors.New("missing cannon L2")
 	ErrMissingCannonBin              = errors.New("missing cannon bin")
 	ErrMissingCannonServer           = errors.New("missing cannon server")
@@ -93,6 +95,7 @@ type Config struct {
 	GameWindow              time.Duration    // Maximum time duration to look for games to progress
 	AgreeWithProposedOutput bool             // Temporary config if we agree or disagree with the posted output
 	Datadir                 string           // Data Directory
+	MaxConcurrency          uint             // Maximum number of threads to use when progressing games
 
 	TraceType TraceType // Type of trace
 
@@ -124,6 +127,7 @@ func NewConfig(
 	return Config{
 		L1EthRpc:           l1EthRpc,
 		GameFactoryAddress: gameFactoryAddress,
+		MaxConcurrency:     uint(runtime.NumCPU()),
 
 		AgreeWithProposedOutput: agreeWithProposedOutput,
 
@@ -153,6 +157,9 @@ func (c Config) Check() error {
 	if c.Datadir == "" {
 		return ErrMissingDatadir
 	}
+	if c.MaxConcurrency == 0 {
+		return ErrMaxConcurrencyZero
+	}
 	if c.TraceType == TraceTypeCannon {
 		if c.CannonBin == "" {
 			return ErrMissingCannonBin

op-challenger/config/config_test.go

 package config
 
 import (
+	"runtime"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -104,6 +105,19 @@ func TestDatadirRequired(t *testing.T) {
 	require.ErrorIs(t, config.Check(), ErrMissingDatadir)
 }
 
+func TestMaxConcurrency(t *testing.T) {
+	t.Run("Required", func(t *testing.T) {
+		config := validConfig(TraceTypeAlphabet)
+		config.MaxConcurrency = 0
+		require.ErrorIs(t, config.Check(), ErrMaxConcurrencyZero)
+	})
+
+	t.Run("DefaultToNumberOfCPUs", func(t *testing.T) {
+		config := validConfig(TraceTypeAlphabet)
+		require.EqualValues(t, runtime.NumCPU(), config.MaxConcurrency)
+	})
+}
+
 func TestCannonL2Required(t *testing.T) {
 	config := validConfig(TraceTypeCannon)
 	config.CannonL2 = ""

op-challenger/fault/scheduler/scheduler.go

@@ -14,7 +14,7 @@ var ErrBusy = errors.New("busy scheduling previous update")
 type Scheduler struct {
 	logger         log.Logger
 	coordinator    *coordinator
-	maxConcurrency int
+	maxConcurrency uint
 	scheduleQueue  chan []common.Address
 	jobQueue       chan job
 	resultQueue    chan job
@@ -22,7 +22,7 @@ type Scheduler struct {
 	cancel         func()
 }
 
-func NewScheduler(logger log.Logger, disk DiskManager, maxConcurrency int, createPlayer PlayerCreator) *Scheduler {
+func NewScheduler(logger log.Logger, disk DiskManager, maxConcurrency uint, createPlayer PlayerCreator) *Scheduler {
 	// Size job and results queues to be fairly small so backpressure is applied early
 	// but with enough capacity to keep the workers busy
 	jobQueue := make(chan job, maxConcurrency*2)
@@ -46,7 +46,7 @@ func (s *Scheduler) Start(ctx context.Context) {
 	ctx, cancel := context.WithCancel(ctx)
 	s.cancel = cancel
 
-	for i := 0; i < s.maxConcurrency; i++ {
+	for i := uint(0); i < s.maxConcurrency; i++ {
 		s.wg.Add(1)
 		go progressGames(ctx, s.jobQueue, s.resultQueue, &s.wg)
 	}

op-challenger/fault/service.go

@@ -20,9 +20,6 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
-// TODO(CLI-4342): Make this a cli option
-const maxConcurrency = 4
-
 type Loader interface {
 	FetchAbsolutePrestateHash(ctx context.Context) ([]byte, error)
 }
@@ -79,7 +76,7 @@ func NewService(ctx context.Context, logger log.Logger, cfg *config.Config) (*Se
 	sched := scheduler.NewScheduler(
 		logger,
 		disk,
-		maxConcurrency,
+		cfg.MaxConcurrency,
 		func(addr common.Address, dir string) (scheduler.GamePlayer, error) {
 			return NewGamePlayer(ctx, logger, cfg, dir, addr, txMgr, client)
 		})

op-challenger/flags/flags.go

@@ -2,6 +2,7 @@ package flags
 
 import (
 	"fmt"
+	"runtime"
 	"strings"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -63,6 +64,12 @@ var (
 		EnvVars: prefixEnvVars("DATADIR"),
 	}
 	// Optional Flags
+	MaxConcurrencyFlag = &cli.UintFlag{
+		Name:    "max-concurrency",
+		Usage:   "Maximum number of threads to use when progressing games",
+		EnvVars: prefixEnvVars("MAX_CONCURRENCY"),
+		Value:   uint(runtime.NumCPU()),
+	}
 	AlphabetFlag = &cli.StringFlag{
 		Name:    "alphabet",
 		Usage:   "Correct Alphabet Trace (alphabet trace type only)",
@@ -128,6 +135,7 @@ var requiredFlags = []cli.Flag{
 
 // optionalFlags is a list of unchecked cli flags
 var optionalFlags = []cli.Flag{
+	MaxConcurrencyFlag,
 	AlphabetFlag,
 	GameAllowlistFlag,
 	CannonNetworkFlag,
@@ -220,6 +228,10 @@ func NewConfigFromCLI(ctx *cli.Context) (*config.Config, error) {
 
 	traceTypeFlag := config.TraceType(strings.ToLower(ctx.String(TraceTypeFlag.Name)))
 
+	maxConcurrency := ctx.Uint(MaxConcurrencyFlag.Name)
+	if maxConcurrency == 0 {
+		return nil, fmt.Errorf("%v must not be 0", MaxConcurrencyFlag.Name)
+	}
 	return &config.Config{
 		// Required Flags
 		L1EthRpc:                ctx.String(L1EthRpcFlag.Name),
@@ -227,6 +239,7 @@ func NewConfigFromCLI(ctx *cli.Context) (*config.Config, error) {
 		GameFactoryAddress:      gameFactoryAddress,
 		GameAllowlist:           allowedGames,
 		GameWindow:              ctx.Duration(GameWindowFlag.Name),
+		MaxConcurrency:          maxConcurrency,
 		AlphabetTrace:           ctx.String(AlphabetFlag.Name),
 		CannonNetwork:           ctx.String(CannonNetworkFlag.Name),
 		CannonRollupConfigPath:  ctx.String(CannonRollupConfigFlag.Name),