From fa2e940659838f222c8d6db3db71e610d06ec056 Mon Sep 17 00:00:00 2001 From: Julian Meyer Date: Mon, 27 Jul 2026 14:13:05 -0700 Subject: [PATCH] feat(load-test): separate benchmark setup and measurement Co-authored-by: Amp Amp-Thread-ID: https://ampcode.com/threads/T-019fa48a-da77-7350-ba7c-64e3d06edc76 --- runner/network/sequencer_benchmark.go | 70 ++++++- runner/payload/loadtest/load_test_worker.go | 179 +++++++++++++++--- .../payload/loadtest/load_test_worker_test.go | 130 +++++++++++-- runner/payload/worker/types.go | 5 + 4 files changed, 335 insertions(+), 49 deletions(-) diff --git a/runner/network/sequencer_benchmark.go b/runner/network/sequencer_benchmark.go index 5562df2..13e517a 100644 --- a/runner/network/sequencer_benchmark.go +++ b/runner/network/sequencer_benchmark.go @@ -28,20 +28,30 @@ import ( const gracefulWorkerShutdownTimeout = 90 * time.Second type benchmarkRunController struct { - maxBlocks int - completion payloadworker.CompletionWorker + maxBlocks int + completion payloadworker.CompletionWorker + measurement payloadworker.MeasurementCompletionWorker } func newBenchmarkRunController(transactionWorker payloadworker.Worker, params benchtypes.RunParams) benchmarkRunController { completion, ok := transactionWorker.(payloadworker.CompletionWorker) if ok { - return benchmarkRunController{completion: completion} + measurement, _ := transactionWorker.(payloadworker.MeasurementCompletionWorker) + return benchmarkRunController{completion: completion, measurement: measurement} } return benchmarkRunController{maxBlocks: params.NumBlocks} } func (c benchmarkRunController) shouldStop(nextBlockIndex uint64) (bool, error) { if c.completion != nil { + if c.measurement != nil { + select { + case <-c.measurement.MeasurementDone(): + return true, c.completion.Err() + default: + return false, nil + } + } select { case <-c.completion.Done(): return true, c.completion.Err() @@ -56,6 +66,10 @@ func (c benchmarkRunController) usesWorkerCompletion() bool { return c.completion != nil } +func (c benchmarkRunController) separatesMeasurementCompletion() bool { + return c.measurement != nil +} + type sequencerBenchmark struct { log log.Logger sequencerClient types.ExecutionClient @@ -308,7 +322,12 @@ func (nb *sequencerBenchmark) Run(ctx context.Context, metricsCollector metrics. blockIndex++ } - if !runController.usesWorkerCompletion() { + if runController.separatesMeasurementCompletion() { + if err := nb.settleCompletionWorker(benchmarkCtx, transactionWorker, consensusClient, pendingTxs, runController.completion); err != nil { + errChan <- err + return + } + } else if !runController.usesWorkerCompletion() { if err := nb.settleGracefulWorkerShutdown(benchmarkCtx, transactionWorker, consensusClient, pendingTxs); err != nil { errChan <- err return @@ -341,6 +360,49 @@ func (nb *sequencerBenchmark) Run(ctx context.Context, metricsCollector metrics. } } +func (nb *sequencerBenchmark) settleCompletionWorker( + ctx context.Context, + transactionWorker payloadworker.Worker, + consensusClient *consensus.SequencerConsensusClient, + pendingTxs int, + completion payloadworker.CompletionWorker, +) error { + settlementCtx, cancel := context.WithTimeout(ctx, gracefulWorkerShutdownTimeout) + defer cancel() + + settlementBlock := 0 + for { + select { + case <-completion.Done(): + nb.log.Info("Load-test worker finished draining", "settlement_blocks", settlementBlock) + return completion.Err() + case <-settlementCtx.Done(): + return errors.Wrapf( + settlementCtx.Err(), + "timed out waiting for load-test worker to drain after %d settlement blocks", + settlementBlock, + ) + default: + } + + var err error + _, pendingTxs, err = nb.proposeBlock( + settlementCtx, + transactionWorker, + consensusClient, + nil, + uint64(settlementBlock+1), + pendingTxs, + true, + false, + ) + if err != nil { + return errors.Wrap(err, "failed to propose load-test settlement block") + } + settlementBlock++ + } +} + func (nb *sequencerBenchmark) proposeBlock( ctx context.Context, transactionWorker payloadworker.Worker, diff --git a/runner/payload/loadtest/load_test_worker.go b/runner/payload/loadtest/load_test_worker.go index 54d94e9..5a2117f 100644 --- a/runner/payload/loadtest/load_test_worker.go +++ b/runner/payload/loadtest/load_test_worker.go @@ -24,8 +24,7 @@ import ( // LoadTestPayloadDefinition is the YAML payload params for the load-test type. // The load-test workload itself lives in a native base-load-tester config file; -// benchmark mode overlays the RPC fields it must control and overlays target_gps -// only when the benchmark matrix specifies one. +// benchmark mode overlays the RPC and block fields it must control. type LoadTestPayloadDefinition struct { ConfigFile string `yaml:"config_file"` Network string `yaml:"network"` @@ -44,12 +43,17 @@ type loadTestPayloadWorker struct { mempool *mempool.StaticWorkloadMempool cmd *exec.Cmd done chan struct{} - startOnce sync.Once + measurementDone chan struct{} + processOnce sync.Once + armOnce sync.Once + finishOnce sync.Once shutdownOnce sync.Once waitErrMu sync.Mutex waitErr error + armErr error sourceConfigPath string renderedConfigPath string + controlDir string outputPath string } @@ -86,6 +90,7 @@ func NewLoadTestPayloadWorker( configOverrides: params.LoadTestConfigOverrides, mempool: mp, done: make(chan struct{}), + measurementDone: make(chan struct{}), sourceConfigPath: sourceConfigPath, outputPath: outputPath, } @@ -104,12 +109,16 @@ func (w *loadTestPayloadWorker) Setup(ctx context.Context) error { } w.renderedConfigPath = configPath - w.log.Info("Prepared load test", "binary", w.loadTestBin, "config", configPath) - return nil + if err := w.start(ctx); err != nil { + return err + } + readyFile := filepath.Join(w.controlDir, "ready") + w.log.Info("Waiting for load test readiness", "ready_file", readyFile) + return w.waitForFile(ctx, readyFile, "load test exited before becoming ready") } func (w *loadTestPayloadWorker) start(ctx context.Context) error { - w.startOnce.Do(func() { + w.processOnce.Do(func() { if w.renderedConfigPath == "" { w.finish(errors.New("load-test config has not been prepared")) return @@ -117,7 +126,13 @@ func (w *loadTestPayloadWorker) start(ctx context.Context) error { w.log.Info("Starting load test", "binary", w.loadTestBin, "config", w.renderedConfigPath) - cmd := exec.CommandContext(ctx, w.loadTestBin, w.renderedConfigPath) + cmd := exec.CommandContext( + ctx, + w.loadTestBin, + "--separate-setup", w.controlDir, + "--block-gas-limit", strconv.FormatUint(w.gasLimit, 10), + w.renderedConfigPath, + ) cmd.Stdout = os.Stdout cmd.Stderr = os.Stdout cmd.Env = append(os.Environ(), fmt.Sprintf("FUNDER_KEY=%s", w.prefundSK)) @@ -134,9 +149,29 @@ func (w *loadTestPayloadWorker) start(ctx context.Context) error { return } w.cmd = cmd + if w.measurementDone == nil { + w.measurementDone = make(chan struct{}) + } go func() { w.finish(cmd.Wait()) }() + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 24*time.Hour) + defer cancel() + err := w.waitForFile( + ctx, + filepath.Join(w.controlDir, "finished"), + "load test exited before measurement completed", + ) + if err != nil { + w.waitErrMu.Lock() + if w.waitErr == nil { + w.waitErr = err + } + w.waitErrMu.Unlock() + } + close(w.measurementDone) + }() }) return w.Err() @@ -174,6 +209,10 @@ func (w *loadTestPayloadWorker) Done() <-chan struct{} { return w.done } +func (w *loadTestPayloadWorker) MeasurementDone() <-chan struct{} { + return w.measurementDone +} + func (w *loadTestPayloadWorker) Err() error { w.waitErrMu.Lock() defer w.waitErrMu.Unlock() @@ -181,10 +220,14 @@ func (w *loadTestPayloadWorker) Err() error { } func (w *loadTestPayloadWorker) finish(err error) { - w.waitErrMu.Lock() - w.waitErr = err - w.waitErrMu.Unlock() - close(w.done) + w.finishOnce.Do(func() { + w.waitErrMu.Lock() + if w.waitErr == nil { + w.waitErr = err + } + w.waitErrMu.Unlock() + close(w.done) + }) } func (w *loadTestPayloadWorker) Stop(ctx context.Context) error { @@ -213,15 +256,71 @@ func (w *loadTestPayloadWorker) Stop(ctx context.Context) error { w.log.Warn("failed to remove load-test config", "path", w.renderedConfigPath, "err", err) } } + if w.controlDir != "" { + if err := os.RemoveAll(w.controlDir); err != nil { + w.log.Warn("failed to remove load-test control directory", "path", w.controlDir, "err", err) + } + } return nil } func (w *loadTestPayloadWorker) SendTxs(ctx context.Context, _ int) (int, error) { - if err := w.start(ctx); err != nil { - return 0, err + w.armOnce.Do(func() { + startFile := filepath.Join(w.controlDir, "start") + if err := publishFile(startFile); err != nil { + w.armErr = errors.Wrap(err, "failed to arm load test") + return + } + startedFile := filepath.Join(w.controlDir, "started") + w.log.Info("Armed load test; waiting for start acknowledgement", "started_file", startedFile) + w.armErr = w.waitForFile(ctx, startedFile, "load test exited before acknowledging start") + }) + return 0, w.armErr +} + +func (w *loadTestPayloadWorker) waitForFile(ctx context.Context, path, exitMessage string) error { + if _, hasDeadline := ctx.Deadline(); !hasDeadline { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + } + ticker := time.NewTicker(50 * time.Millisecond) + defer ticker.Stop() + for { + if _, err := os.Stat(path); err == nil { + return nil + } else if !os.IsNotExist(err) { + return errors.Wrap(err, "failed to inspect load-test handshake file") + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-w.Done(): + if err := w.Err(); err != nil { + return errors.Wrap(err, exitMessage) + } + return errors.New(exitMessage) + case <-ticker.C: + } } - return 0, nil +} + +func publishFile(path string) error { + tmp, err := os.CreateTemp(filepath.Dir(path), ".load-test-handshake-*") + if err != nil { + return err + } + tmpPath := tmp.Name() + if err := tmp.Close(); err != nil { + _ = os.Remove(tmpPath) + return err + } + if err := os.Rename(tmpPath, path); err != nil { + _ = os.Remove(tmpPath) + return err + } + return nil } func resolveConfigFilePath(benchmarkConfigPath string, loadTestConfigPath string) (string, error) { @@ -258,9 +357,8 @@ func (w *loadTestPayloadWorker) buildConfig() (*yaml.Node, error) { flashblocksURL = "ws://localhost:7111" } setMappingValue(config, "flashblocks_ws", stringNode(flashblocksURL)) - if w.blockTime > 0 && w.gasLimit > 0 { - targetGPS := w.gasLimit / uint64(w.blockTime.Seconds()) - setMappingValue(config, "target_gps", uintNode(targetGPS)) + if mappingUintValue(config, "mempool_target_blocks") == 0 { + setMappingValue(config, "mempool_target_blocks", uintNode(3)) } for key, value := range w.configOverrides { node, err := nodeFromValue(value) @@ -269,6 +367,7 @@ func (w *loadTestPayloadWorker) buildConfig() (*yaml.Node, error) { } setMappingValue(config, key, node) } + setMappingValue(config, "target_gps", nullNode()) return config, nil } @@ -298,6 +397,16 @@ func setMappingValue(mapping *yaml.Node, key string, value *yaml.Node) { mapping.Content = append(mapping.Content, stringNode(key), value) } +func mappingUintValue(mapping *yaml.Node, key string) uint64 { + for i := 0; i < len(mapping.Content)-1; i += 2 { + if mapping.Content[i].Value == key { + value, _ := strconv.ParseUint(mapping.Content[i+1].Value, 10, 64) + return value + } + } + return 0 +} + func stringNode(value string) *yaml.Node { return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: value} } @@ -322,6 +431,10 @@ func nodeFromValue(value interface{}) (*yaml.Node, error) { return doc.Content[0], nil } +func nullNode() *yaml.Node { + return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!null", Value: "null"} +} + func stringSequenceNode(values ...string) *yaml.Node { node := &yaml.Node{Kind: yaml.SequenceNode, Tag: "!!seq"} for _, value := range values { @@ -331,10 +444,25 @@ func stringSequenceNode(values ...string) *yaml.Node { } // writeConfig generates a temporary YAML config file for the load-test binary -// with benchmark-controlled RPC, timing, and report fields. +// with benchmark-controlled RPC, block, handshake, and report fields. func (w *loadTestPayloadWorker) writeConfig() (string, error) { + tmpFile, err := os.CreateTemp("", "load-test-config-*.yaml") + if err != nil { + return "", errors.Wrap(err, "failed to create temp config file") + } + configPath := tmpFile.Name() + if err := tmpFile.Close(); err != nil { + return "", errors.Wrap(err, "failed to close temp config file") + } + w.controlDir, err = os.MkdirTemp("", "load-test-control-*") + if err != nil { + _ = os.Remove(configPath) + return "", errors.Wrap(err, "failed to create load-test control directory") + } + config, err := w.buildConfig() if err != nil { + _ = os.Remove(configPath) return "", err } data, err := yaml.Marshal(config) @@ -342,25 +470,16 @@ func (w *loadTestPayloadWorker) writeConfig() (string, error) { return "", errors.Wrap(err, "failed to marshal load-test config") } - tmpFile, err := os.CreateTemp("", "load-test-config-*.yaml") - if err != nil { - return "", errors.Wrap(err, "failed to create temp config file") - } - - if _, err := tmpFile.Write(data); err != nil { - _ = tmpFile.Close() + if err := os.WriteFile(configPath, data, 0600); err != nil { + _ = os.Remove(configPath) return "", errors.Wrap(err, "failed to write temp config file") } - if err := tmpFile.Close(); err != nil { - return "", errors.Wrap(err, "failed to close temp config file") - } - w.log.Info("Generated load-test config", "source_config", w.sourceConfigPath, "gas_limit", w.gasLimit, "block_time", w.blockTime, ) - return tmpFile.Name(), nil + return configPath, nil } diff --git a/runner/payload/loadtest/load_test_worker_test.go b/runner/payload/loadtest/load_test_worker_test.go index 6a33f0a..eee13b0 100644 --- a/runner/payload/loadtest/load_test_worker_test.go +++ b/runner/payload/loadtest/load_test_worker_test.go @@ -85,7 +85,7 @@ transactions: "transaction_submission_rpcs:\n - http://sequencer.example", "query_rpc: http://sequencer.example", "flashblocks_ws: ws://benchmark-flashblocks.example", - "target_gps: 75000000", + "mempool_target_blocks: 3", "duration: \"60s\"", "chain_id: 8453", "sender_count: 250", @@ -105,13 +105,16 @@ transactions: "standalone-submitter.invalid", "standalone-query.invalid", "standalone-flashblocks.invalid", - "target_gps: 123", + "block_gas_limit:", + "ready_file:", + "start_file:", + "started_file:", } { require.NotContains(t, output, oldValue) } } -func TestBuildConfigPreservesNativeTargetGPSWhenGasLimitOrBlockTimeIsZero(t *testing.T) { +func TestBuildConfigRemovesTargetGPSAndPreservesExplicitMempoolMultiplier(t *testing.T) { configPath := filepath.Join(t.TempDir(), "load-test.yaml") err := os.WriteFile(configPath, []byte(` transaction_submission_rpcs: @@ -119,6 +122,7 @@ transaction_submission_rpcs: query_rpc: "http://standalone-query.invalid" flashblocks_ws: "ws://standalone-flashblocks.invalid" target_gps: 123 +mempool_target_blocks: 5 duration: "60s" transactions: - weight: 100 @@ -139,7 +143,8 @@ transactions: require.NoError(t, err) output := string(encoded) - require.Contains(t, output, "target_gps: 123") + require.Contains(t, output, "target_gps: null") + require.Contains(t, output, "mempool_target_blocks: 5") require.Contains(t, output, "duration: \"60s\"") } @@ -178,13 +183,13 @@ transactions: require.NoError(t, err) output := string(encoded) - require.Contains(t, output, "target_gps: 75000000") + require.Contains(t, output, "target_gps: null") require.Contains(t, output, "seed: 654790") require.Contains(t, output, "fresh_recipient_ratio: 1") require.NotContains(t, output, "seed: 654789") } -func TestSetupPreparesConfigWithoutStartingProcess(t *testing.T) { +func TestSetupStartsProcessAndFirstSendTxsCompletesHandshake(t *testing.T) { configPath := filepath.Join(t.TempDir(), "load-test.yaml") err := os.WriteFile(configPath, []byte(` transaction_submission_rpcs: @@ -197,27 +202,122 @@ transactions: `), 0644) require.NoError(t, err) + helper := writeHelper(t, ` +control=$2 +ready="$control/ready" +start="$control/start" +started="$control/started" +finished="$control/finished" +touch "$ready" +while [ ! -e "$start" ]; do sleep 0.01; done +touch "$started" +touch "$finished" +trap 'exit 0' INT TERM +while :; do sleep 1; done +`) worker := &loadTestPayloadWorker{ log: log.New(), + loadTestBin: helper, elRPCURL: "http://sequencer.example", sourceConfigPath: configPath, done: make(chan struct{}), } - t.Cleanup(func() { - if worker.renderedConfigPath != "" { - require.NoError(t, os.Remove(worker.renderedConfigPath)) - } - }) + t.Cleanup(func() { require.NoError(t, worker.Stop(context.Background())) }) require.NoError(t, worker.Setup(context.Background())) require.NotEmpty(t, worker.renderedConfigPath) - require.Nil(t, worker.cmd) + require.NotNil(t, worker.cmd) + require.FileExists(t, filepath.Join(worker.controlDir, "ready")) + require.NoFileExists(t, filepath.Join(worker.controlDir, "start")) + count, err := worker.SendTxs(context.Background(), 0) + require.NoError(t, err) + require.Zero(t, count) + require.FileExists(t, filepath.Join(worker.controlDir, "start")) + require.FileExists(t, filepath.Join(worker.controlDir, "started")) select { - case <-worker.Done(): - t.Fatal("load-test worker should not be done before it starts") - default: + case <-worker.MeasurementDone(): + case <-time.After(time.Second): + t.Fatal("measurement completion was not signaled") } + + // The barrier is one-shot; subsequent calls return immediately. + _, err = worker.SendTxs(context.Background(), 0) + require.NoError(t, err) +} + +func TestSetupFailsWhenProcessExitsBeforeReady(t *testing.T) { + worker := testWorker(t, writeHelper(t, `exit 7`)) + err := worker.Setup(context.Background()) + require.Error(t, err) + require.Contains(t, err.Error(), "before becoming ready") + require.NoError(t, worker.Stop(context.Background())) +} + +func TestSetupHonorsContextCancellation(t *testing.T) { + worker := testWorker(t, writeHelper(t, `while :; do sleep 1; done`)) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + err := worker.Setup(ctx) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.NoError(t, worker.Stop(context.Background())) +} + +func TestSendTxsHonorsContextWhileWaitingForStarted(t *testing.T) { + worker := testWorker(t, writeHelper(t, ` +control=$2 +ready="$control/ready" +touch "$ready" +trap 'exit 0' INT TERM +while :; do sleep 1; done +`)) + require.NoError(t, worker.Setup(context.Background())) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + _, err := worker.SendTxs(ctx, 0) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.FileExists(t, filepath.Join(worker.controlDir, "start")) + require.NoError(t, worker.Stop(context.Background())) +} + +func TestCleanExitBeforeFinishedFailsMeasurement(t *testing.T) { + worker := testWorker(t, writeHelper(t, ` +control=$2 +touch "$control/ready" +while [ ! -e "$control/start" ]; do sleep 0.01; done +touch "$control/started" +sleep 0.1 +exit 0 +`)) + require.NoError(t, worker.Setup(context.Background())) + _, err := worker.SendTxs(context.Background(), 0) + require.NoError(t, err) + + select { + case <-worker.MeasurementDone(): + case <-time.After(time.Second): + t.Fatal("measurement failure was not signaled") + } + require.ErrorContains(t, worker.Err(), "exited before measurement completed") + require.NoError(t, worker.Stop(context.Background())) +} + +func testWorker(t *testing.T, binary string) *loadTestPayloadWorker { + t.Helper() + configPath := filepath.Join(t.TempDir(), "load-test.yaml") + require.NoError(t, os.WriteFile(configPath, []byte("transactions: []\n"), 0644)) + return &loadTestPayloadWorker{ + log: log.New(), loadTestBin: binary, elRPCURL: "http://sequencer.example", + sourceConfigPath: configPath, done: make(chan struct{}), + } +} + +func writeHelper(t *testing.T, body string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "helper.sh") + require.NoError(t, os.WriteFile(path, []byte("#!/bin/sh\nset -eu\n"+body), 0755)) + return path } func TestResolveConfigFilePath(t *testing.T) { diff --git a/runner/payload/worker/types.go b/runner/payload/worker/types.go index 942df00..596548b 100644 --- a/runner/payload/worker/types.go +++ b/runner/payload/worker/types.go @@ -32,3 +32,8 @@ type CompletionWorker interface { Done() <-chan struct{} Err() error } + +// MeasurementCompletionWorker signals when measured load generation ends, before final draining. +type MeasurementCompletionWorker interface { + MeasurementDone() <-chan struct{} +}