From 6c1f1e1b396465c6ea4f5b8ad66f032904df0744 Mon Sep 17 00:00:00 2001
From: Mile Druzijanic <mile@dragonflydb.io>
Date: Wed, 22 Jul 2026 11:09:37 +0200
Subject: [PATCH] feat: dfly_bench implementation and tests

---
 README.md                                     |  51 ++-
 bench/commands/results.go                     |   6 +-
 bench/commands/runner.go                      |  15 +
 bench/commands/runner_dfly.go                 | 323 ++++++++++++++++++
 bench/dflybench.go                            | 195 +++++++++++
 bench/dflybench_test.go                       | 109 ++++++
 bench/replication/dflybench.go                | 100 ------
 bench/replication/scenario.go                 |   3 +-
 benchmarks/GET/dfly_bench/GET_reproduce.md    | 115 +++++++
 benchmarks/GET/{ => memtier}/GET_reproduce.md |  30 ++
 benchmarks/GET/{ => memtier}/ena/after.txt    |   0
 benchmarks/GET/{ => memtier}/ena/before.txt   |   0
 .../GET/{ => memtier}/ena/timeseries.txt      |   0
 .../GET/{ => memtier}/mpstat/get-mpstat.log   |   0
 benchmarks/SET/dfly_bench/SET_reproduce.md    |  90 +++++
 benchmarks/SET/{ => memtier}/SET_reproduce.md |  20 ++
 cmd/root.go                                   |   1 +
 cmd/run_commands_dfly.go                      | 282 +++++++++++++++
 cmd/setup.go                                  |  11 +
 infra/state.go                                |   4 +
 20 files changed, 1252 insertions(+), 103 deletions(-)
 create mode 100644 bench/commands/runner_dfly.go
 create mode 100644 bench/dflybench.go
 create mode 100644 bench/dflybench_test.go
 create mode 100644 benchmarks/GET/dfly_bench/GET_reproduce.md
 rename benchmarks/GET/{ => memtier}/GET_reproduce.md (81%)
 rename benchmarks/GET/{ => memtier}/ena/after.txt (100%)
 rename benchmarks/GET/{ => memtier}/ena/before.txt (100%)
 rename benchmarks/GET/{ => memtier}/ena/timeseries.txt (100%)
 rename benchmarks/GET/{ => memtier}/mpstat/get-mpstat.log (100%)
 create mode 100644 benchmarks/SET/dfly_bench/SET_reproduce.md
 rename benchmarks/SET/{ => memtier}/SET_reproduce.md (57%)
 create mode 100644 cmd/run_commands_dfly.go

diff --git a/README.md b/README.md
index 2eddf7c..e395d1e 100644
--- a/README.md
+++ b/README.md
@@ -56,6 +56,7 @@ examples.
 | [`setup`](#setup) | Provision an isolated VPC + EC2 fleet and pre-install every engine. | mutates |
 | [`run`](#run) | Execute a `memtier_benchmark` pass (N trials, all engines) against an existing setup. | mutates |
 | [`run-commands`](#run-commands) | Benchmark one Redis command (SET, GET, LPUSH, ...) per invocation against an existing setup. | mutates |
+| [`run-commands-dfly`](#run-commands-dfly) | Same as `run-commands`, but drives load with `dfly_bench` instead of memtier. | mutates |
 | [`run-replication`](#run-replication) | Execute the Dragonfly master/replica sweep against a setup with `--replicas >= 1`. | mutates |
 | [`quickstart`](#quickstart) | One-shot: `setup` + `run` + `destroy` in a single invocation. | mutates |
 | [`compare`](#compare) | Diff two result files (auto-detects memtier vs replication) and exit non-zero on regression. | read-only |
@@ -623,6 +624,54 @@ benchmark. 100M x 256B assumes a ~128GB server; on smaller instances
 lower it proportionally or the engines will evict or OOM during
 preload.
 
+### `run-commands-dfly`
+
+The `dfly_bench` counterpart to `run-commands`: it drives the same
+per-command flush/preload/warmup/trials cycle, but with Dragonfly's own
+load generator instead of memtier. `dfly_bench` is built from the
+`dragonflydb/dragonfly` source and installed on the client at `dfbench
+setup` time (alongside memtier); override the build with
+`--dfly-bench-ref`. Results use the same schema as `run-commands` (tagged
+`tool="dfly_bench"`) and land in `~/.dfbench/runs/<id>/commands/<timestamp>/`,
+so a memtier file and a `dfly_bench` file for the same command diff with
+`dfbench compare`.
+
+Because `dfly_bench` uses its own key prefix (`--key_prefix`, default
+`key:`), read commands (GET, ...) are preloaded by `dfly_bench` itself
+(sequential distribution, one connection per thread covering the whole
+`--key-maximum` range) rather than reusing a memtier preload.
+
+| Flag | Default | Notes |
+| ---- | ------- | ----- |
+| `--id` | (required) | Setup previously created by `dfbench setup`. |
+| `--command` | (required) | Single command name, e.g. `SET`, `GET`, `LRANGE`. |
+| `--engines` | `dragonfly` | Subset of the setup's installed engines (`dfly_bench` speaks RESP, so Redis/Valkey work too). |
+| `--dfly-threads` | `32` | `dfly_bench --proactor_threads` (analogous to memtier threads). Total connections = threads * conns. |
+| `--dfly-conns` | `5` | `dfly_bench -c`, connections per thread (analogous to memtier clients). |
+| `--dfly-pipeline` | `20` | `dfly_bench --pipeline`: max pending requests per connection. |
+| `--dfly-data-size` | `256` | `dfly_bench -d`: value size in bytes (`__data__` placeholders). |
+| `--key-maximum` | `100000000` | Key range (`--key_maximum`). Structured preloads use a built-in 1M-key range. |
+| `--key-dist` | `U` | `dfly_bench --key_dist` for the measured pass: `U` (uniform, the memtier random `R` equivalent), `N`, `Z`, `S`. |
+| `--qps` | `0` | `dfly_bench --qps` per connection; `0` = closed-loop max load (coordinated omission). |
+| `--test-time` | `300` | Seconds of measured load per trial. |
+| `--warmup-time` | `10` | Seconds of discarded warmup load per engine. `0` disables. |
+| `--trials` | `3` | Measured trials per engine (median with [min..max]). |
+| `--dfly-bench-ref` | `main` | Git ref of `dragonflydb/dragonfly` to build `dfly_bench` from if it isn't already installed. |
+| `--dragonfly_num_shards`, `--dragonfly_conn_use_incoming_cpu`, `--dragonfly_version` | (as in `run-commands`) | Server-side Dragonfly tuning; identical semantics to `run-commands`. |
+| `--redis_io_threads`, `--valkey_io_threads`, `--redis_pin_network`, `--valkey_pin_network` | (as in `run-commands`) | Redis/Valkey tuning. |
+| `--output` | `table` | `table` / `json` / `csv`. |
+
+`dfly_bench`'s `--json_out_file` writes a memtier-compatible latency
+report, so the same parser produces the throughput / p50 / p99 / p99.9 /
+avg columns for both tools.
+
+```bash
+dfbench run-commands-dfly --id percmd \
+  --engines dragonfly \
+  --command GET \
+  --key-maximum 1000000 --test-time 15 --trials 1
+```
+
 ### `run-replication`
 
 Execute the Dragonfly master/replica benchmark scenario against an
@@ -1106,7 +1155,7 @@ their keyspace can empty mid-trial. A `results.csv` with one row per
 | `run-replication` errors: `ERR Out of memory` during preload | Default `--preload-keys` and `--value-size` exceed the instance's memory budget. | Set `--master-flags "--maxmemory=25gb"` (or larger), lower `--preload-keys`, or shrink `--value-size`. Dragonfly will evict rather than OOM. |
 | Maxload case appears to run at ~20 req/sec | `dfly_bench --qps` defaults to 20 when omitted. dfbench explicitly emits `--qps=0` for `maxload`; if you see this, verify `--dfly-ref` resolves to a revision whose `dfly_bench` treats 0 as unbounded. | See `BuildDflyBenchArgs` in [`bench/replication/dflybench.go`](bench/replication/dflybench.go). |
 | `run` refuses to start with `setup "foo" is not ready (status="benchmarking")` | Previous run crashed or was killed before the deferred status-restore ran. | Rare now; `run` and `run-replication` wrap their context with `signal.NotifyContext(Interrupt, SIGTERM)` so `Ctrl+C` unwinds cleanly. If it happens: `dfbench destroy --id ...` and start over. |
-| `dfly_bench` build fails on the client: missing `helio/blaze.sh` | Submodule population failed. | `InstallDflyBench` falls back to a full clone + `git submodule update --init --recursive`. If it still fails, check the error output for missing apt packages (`bison`, `libboost-fiber-dev`, `libunwind-dev`). |
+| `dfly_bench` build fails on the client: missing `helio/blaze.sh` | Submodule population failed. | `InstallDflyBench` in [`bench/dflybench.go`](bench/dflybench.go) falls back to a full clone + `git submodule update --init --recursive`. If it still fails, check the error output for missing apt packages (`bison`, `libboost-fiber-dev`, `libunwind-dev`). |
 | `destroy --id foo` returns `setup "foo" not found` | Typo, or state file moved / deleted. | `dfbench list` to see what IDs exist. Old versions leaked the internal path; the current message is deliberately terse. |
 
 ---
diff --git a/bench/commands/results.go b/bench/commands/results.go
index 97156c3..6f88042 100644
--- a/bench/commands/results.go
+++ b/bench/commands/results.go
@@ -19,7 +19,11 @@ const ScenarioTag = "commands"
 // CommandRunResults is the results.json schema for one dfbench
 // run-commands invocation.
 type CommandRunResults struct {
-	Scenario       string          `json:"scenario"` // always "commands"
+	Scenario string `json:"scenario"` // always "commands"
+	// Tool is the load generator that produced these numbers: "memtier"
+	// (dfbench run-commands) or "dfly_bench" (dfbench run-commands-dfly).
+	// It lets 'dfbench compare' diff a memtier file against a dfly_bench one.
+	Tool           string          `json:"tool,omitempty"`
 	RunID          string          `json:"run_id"`
 	SetupID        string          `json:"setup_id"`
 	CreatedAt      time.Time       `json:"created_at"`
diff --git a/bench/commands/runner.go b/bench/commands/runner.go
index d48a5b9..7e14300 100644
--- a/bench/commands/runner.go
+++ b/bench/commands/runner.go
@@ -60,6 +60,20 @@ type Config struct {
 	// dfbench setup installed. Redis/Valkey ignore it.
 	DflyVersion string
 
+	// The fields below are only read by RunDfly (dfbench run-commands-dfly).
+	// The memtier Run ignores them.
+
+	// KeyDist is dfly_bench --key_dist: U (uniform, the memtier "R"
+	// equivalent), N (normal), Z (zipfian), S (sequential). Empty defaults
+	// to U for the measured pass.
+	KeyDist string
+	// QPS is dfly_bench --qps, a per-connection rate cap; 0 means closed-loop
+	// max load (coordinated omission).
+	QPS int
+	// DflyBenchRef is the dragonflydb/dragonfly git ref used to (re)build
+	// dfly_bench on the client if it isn't already installed.
+	DflyBenchRef string
+
 	OutputDir string
 }
 
@@ -102,6 +116,7 @@ func Run(
 
 	run := &CommandRunResults{
 		Scenario:       ScenarioTag,
+		Tool:           "memtier",
 		RunID:          time.Now().UTC().Format("20060102-150405"),
 		SetupID:        state.ID,
 		CreatedAt:      time.Now().UTC(),
diff --git a/bench/commands/runner_dfly.go b/bench/commands/runner_dfly.go
new file mode 100644
index 0000000..4597556
--- /dev/null
+++ b/bench/commands/runner_dfly.go
@@ -0,0 +1,323 @@
+package commands
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/dragonflydb/benchmarking/bench"
+	"github.com/dragonflydb/benchmarking/engine"
+	"github.com/dragonflydb/benchmarking/infra"
+	"github.com/dragonflydb/benchmarking/remote"
+)
+
+// RunDfly benchmarks a single command against an existing dfbench setup
+// using dfly_bench instead of memtier. It mirrors Run: each engine goes
+// through its own flush -> preload -> warmup -> trials cycle, and results
+// land in the same CommandRunResults schema (tagged tool="dfly_bench") so a
+// dfly_bench run and a memtier run can be diffed with 'dfbench compare'.
+//
+// dfly_bench uses its own key prefix (--key_prefix, default "key:"), so the
+// preload must run through dfly_bench too; a memtier SET preload would write
+// keys dfly_bench never reads.
+func RunDfly(
+	ctx context.Context,
+	out io.Writer,
+	state *infra.SetupState,
+	cfg *Config,
+	preinstalledVersions map[string]string,
+) (*CommandRunResults, error) {
+	if cfg.Command.Name == "" {
+		return nil, fmt.Errorf("no command configured")
+	}
+	if len(cfg.Engines) == 0 {
+		return nil, fmt.Errorf("no engines configured")
+	}
+
+	serverSSH := remote.NewSSHClient(state.SSH.PrivateKeyPath, state.Server.PublicIP, state.SSH.User)
+	clientSSH := remote.NewSSHClient(state.SSH.PrivateKeyPath, state.Client.PublicIP, state.SSH.User)
+	defer serverSSH.Close()
+	defer clientSSH.Close()
+
+	fmt.Fprintln(out, "[1/3] Checking SSH connectivity...")
+	if err := serverSSH.WaitForReady(ctx, 30*time.Second); err != nil {
+		return nil, fmt.Errorf("server SSH: %w", err)
+	}
+	if err := clientSSH.WaitForReady(ctx, 30*time.Second); err != nil {
+		return nil, fmt.Errorf("client SSH: %w", err)
+	}
+
+	// dfly_bench is installed at setup time, but ensure it's present for
+	// setups created before that; the installer no-ops if it's on PATH.
+	fmt.Fprintf(out, "      Ensuring dfly_bench is installed (ref=%s)...\n", cfg.DflyBenchRef)
+	if err := bench.InstallDflyBench(ctx, clientSSH, cfg.DflyBenchRef); err != nil {
+		return nil, fmt.Errorf("install dfly_bench: %w", err)
+	}
+
+	run := &CommandRunResults{
+		Scenario:       ScenarioTag,
+		Tool:           "dfly_bench",
+		RunID:          time.Now().UTC().Format("20060102-150405"),
+		SetupID:        state.ID,
+		CreatedAt:      time.Now().UTC(),
+		Status:         "completed",
+		Region:         state.Config.Region,
+		AWSAccount:     state.AWSAccount,
+		ServerInstance: state.Server.InstanceType,
+		ClientInstance: state.Client.InstanceType,
+		Memtier: Metadata{
+			Threads:            cfg.Threads,
+			Clients:            cfg.Clients,
+			Pipeline:           cfg.Pipeline,
+			DataSize:           cfg.DataSize,
+			KeyMaximum:         cfg.KeyMaximum,
+			TestTime:           cfg.TestTime,
+			WarmupTime:         cfg.WarmupTime,
+			DistinctClientSeed: cfg.DistinctClientSeed,
+			IOThreads:          cfg.IOThreads,
+			NumShards:          cfg.NumShards,
+			ConnUseIncomingCPU: cfg.ConnUseIncomingCPU,
+			PinNetwork:         cfg.PinNetwork,
+		},
+		Trials: cfg.Trials,
+	}
+
+	spec := cfg.Command
+	run.Commands = append(run.Commands, CommandResult{
+		Command:     spec.Name,
+		Template:    spec.Template,
+		Approximate: spec.Drains,
+	})
+	result := &run.Commands[0]
+
+	outDir := filepath.Join(cfg.OutputDir, "dflybench")
+	if err := os.MkdirAll(outDir, 0o755); err != nil {
+		return nil, fmt.Errorf("create dflybench dir: %w", err)
+	}
+
+	fmt.Fprintf(out, "[2/3] Benchmarking %s with dfly_bench (%d engines x %d trials)...\n",
+		spec.Name, len(cfg.Engines), cfg.Trials)
+
+	for ei, engineName := range cfg.Engines {
+		fmt.Fprintf(out, "\n  -> [%d/%d] Engine %s\n", ei+1, len(cfg.Engines), engineName)
+
+		eng, err := engine.New(engineName)
+		if err != nil {
+			return run, err
+		}
+		if n := cfg.IOThreads[engineName]; n > 0 {
+			if tunable, ok := eng.(interface{ SetIOThreads(int) }); ok {
+				tunable.SetIOThreads(n)
+			}
+		}
+		if shardable, ok := eng.(interface{ SetNumShards(int) }); ok {
+			shardable.SetNumShards(cfg.NumShards)
+		}
+		if connable, ok := eng.(interface{ SetConnUseIncomingCPU(bool) }); ok {
+			connable.SetConnUseIncomingCPU(cfg.ConnUseIncomingCPU)
+		}
+		if pin, ok := cfg.PinNetwork[engineName]; ok {
+			if pinnable, ok := eng.(interface{ SetPinNetwork(bool) }); ok {
+				pinnable.SetPinNetwork(pin)
+			}
+		}
+
+		if versioner, ok := eng.(interface{ SetInstallVersion(string) }); ok && cfg.DflyVersion != "" {
+			versioner.SetInstallVersion(cfg.DflyVersion)
+			fmt.Fprintf(out, "    Installing %s %s...\n", engineName, cfg.DflyVersion)
+			if err := eng.Install(ctx, serverSSH); err != nil {
+				return run, fmt.Errorf("install %s %s: %w", engineName, cfg.DflyVersion, err)
+			}
+		}
+
+		fmt.Fprintf(out, "    Starting %s...\n", engineName)
+		if err := eng.Start(ctx, serverSSH); err != nil {
+			return run, fmt.Errorf("start %s: %w", engineName, err)
+		}
+
+		version := eng.Version()
+		if version == "" {
+			version = preinstalledVersions[engineName]
+		}
+
+		trials, engErr := runCommandDfly(ctx, out, cfg, spec, eng, serverSSH, clientSSH,
+			state.Server.PrivateIP, outDir, engineName)
+		if engErr == nil {
+			if len(trials) > 0 {
+				result.Engines = append(result.Engines, bench.Aggregate(engineName, version, trials))
+			} else {
+				fmt.Fprintf(out, "      No successful trials for %s/%s\n", engineName, spec.Name)
+			}
+		}
+
+		fmt.Fprintf(out, "\n    Stopping %s...\n", engineName)
+		if err := eng.Stop(ctx, serverSSH); err != nil {
+			fmt.Fprintf(out, "    Warning: stop %s: %v\n", engineName, err)
+		}
+		if engErr != nil {
+			return run, engErr
+		}
+
+		if ei < len(cfg.Engines)-1 {
+			fmt.Fprintf(out, "    Cooldown (%s)...\n", enginePause)
+			sleepBetween(enginePause)
+		}
+	}
+
+	run.CompletedAt = time.Now().UTC()
+
+	fmt.Fprintln(out, "\n[3/3] Writing results...")
+	if err := persistResults(cfg.OutputDir, run); err != nil {
+		return run, fmt.Errorf("persist results: %w", err)
+	}
+	fmt.Fprintf(out, "      results.json + results.csv -> %s\n", cfg.OutputDir)
+
+	return run, nil
+}
+
+// runCommandDfly drives the flush/preload/warmup/trials cycle for a single
+// (engine, command) pair using dfly_bench and returns the parsed trials.
+func runCommandDfly(
+	ctx context.Context,
+	out io.Writer,
+	cfg *Config,
+	spec CommandSpec,
+	eng engine.Engine,
+	serverSSH, clientSSH *remote.SSHClient,
+	serverHost, outDir, engineName string,
+) ([]bench.TrialResult, error) {
+	keyMax := cfg.KeyMaximum
+	if spec.KeyMaximum > 0 {
+		keyMax = spec.KeyMaximum
+	}
+
+	keyDist := cfg.KeyDist
+	if keyDist == "" {
+		keyDist = "U"
+	}
+
+	measureCfg := &bench.DflyBenchConfig{
+		ServerHost:      serverHost,
+		ServerPort:      eng.Port(),
+		ProactorThreads: cfg.Threads,
+		ConnsPerThread:  cfg.Clients,
+		Pipeline:        cfg.Pipeline,
+		DataSize:        cfg.DataSize,
+		Command:         spec.Template,
+		KeyDist:         keyDist,
+		KeyMaximum:      keyMax,
+		TestTime:        cfg.TestTime,
+		QPS:             cfg.QPS,
+	}
+
+	reset := func(phase string) error {
+		if err := eng.Flush(ctx, serverSSH); err != nil {
+			fmt.Fprintf(out, "      Warning: flush before %s: %v\n", phase, err)
+		}
+		if spec.Preload == nil {
+			return nil
+		}
+		fmt.Fprintf(out, "      Preloading (%s, %d keys x %d items)...\n",
+			spec.Preload.Template, keyMax, spec.Preload.Items())
+		preloadCfg := buildDflyPreloadConfig(cfg, spec, serverHost, eng.Port(), keyMax)
+		label := fmt.Sprintf("%s-%s-preload", engineName, spec.Name)
+		if _, _, err := bench.RunDflyBench(ctx, clientSSH, preloadCfg, label); err != nil {
+			return fmt.Errorf("preload %s/%s: %w", engineName, spec.Name, err)
+		}
+		return nil
+	}
+
+	if err := reset("warmup"); err != nil {
+		return nil, err
+	}
+
+	if cfg.WarmupTime > 0 {
+		fmt.Fprintf(out, "      Warming up (%ds, results discarded)...\n", cfg.WarmupTime)
+		warmCfg := *measureCfg
+		warmCfg.TestTime = cfg.WarmupTime
+		label := fmt.Sprintf("%s-%s-warmup", engineName, spec.Name)
+		if _, _, err := bench.RunDflyBench(ctx, clientSSH, &warmCfg, label); err != nil {
+			return nil, fmt.Errorf("warmup %s/%s: %w", engineName, spec.Name, err)
+		}
+		if spec.Mutates {
+			if err := reset("trial 1"); err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	var trials []bench.TrialResult
+	for trial := 1; trial <= cfg.Trials; trial++ {
+		fmt.Fprintf(out, "      Trial %d/%d...\n", trial, cfg.Trials)
+		if trial > 1 {
+			if spec.Mutates {
+				if err := reset(fmt.Sprintf("trial %d", trial)); err != nil {
+					return nil, err
+				}
+			}
+			sleepBetween(trialPause)
+		}
+
+		label := fmt.Sprintf("%s-%s-t%d", engineName, spec.Name, trial)
+		jsonData, stdout, err := bench.RunDflyBench(ctx, clientSSH, measureCfg, label)
+		if err != nil {
+			return nil, fmt.Errorf("benchmark %s/%s trial %d: %w", engineName, spec.Name, trial, err)
+		}
+
+		rawPath := filepath.Join(outDir, label+".json")
+		if werr := os.WriteFile(rawPath, jsonData, 0o644); werr != nil {
+			fmt.Fprintf(out, "      Warning: save raw JSON: %v\n", werr)
+		}
+
+		parsed, perr := bench.ParseWithFallback(engineName, jsonData, stdout)
+		if perr != nil {
+			fmt.Fprintf(out, "      Warning: parse trial %d: %v\n", trial, perr)
+			continue
+		}
+		trials = append(trials, bench.TrialResult{
+			Trial:      trial,
+			Throughput: parsed.Throughput,
+			P50:        parsed.P50,
+			P99:        parsed.P99,
+			P999:       parsed.P999,
+			AvgLatency: parsed.AvgLatency,
+			RawJSON:    jsonData,
+		})
+		fmt.Fprintf(out, "      ✓ trial %d: %.0f ops/s, p99=%.3fms, avg=%.3fms\n",
+			trial, parsed.Throughput, parsed.P99, parsed.AvgLatency)
+	}
+
+	return trials, nil
+}
+
+// buildDflyPreloadConfig builds the dfly_bench config that populates the
+// keyspace before measuring. It uses the sequential ("S") key distribution
+// with one connection per thread so each proactor thread walks its own
+// key subrange exactly once, covering [0, keyMax) with keyMax*items total
+// writes (no per-connection overlap).
+func buildDflyPreloadConfig(cfg *Config, spec CommandSpec, host string, port, keyMax int) *bench.DflyBenchConfig {
+	totalOps := keyMax * spec.Preload.Items()
+	perThread := totalOps / cfg.Threads
+	if totalOps%cfg.Threads != 0 {
+		perThread++
+	}
+	if perThread < 1 {
+		perThread = 1
+	}
+	return &bench.DflyBenchConfig{
+		ServerHost:      host,
+		ServerPort:      port,
+		ProactorThreads: cfg.Threads,
+		ConnsPerThread:  1,
+		Pipeline:        cfg.Pipeline,
+		DataSize:        cfg.DataSize,
+		Command:         spec.Preload.Template,
+		KeyDist:         "S",
+		KeyMaximum:      keyMax,
+		Requests:        perThread,
+		QPS:             0,
+	}
+}
diff --git a/bench/dflybench.go b/bench/dflybench.go
new file mode 100644
index 0000000..4014a50
--- /dev/null
+++ b/bench/dflybench.go
@@ -0,0 +1,195 @@
+package bench
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/dragonflydb/benchmarking/remote"
+)
+
+// DflyBenchConfig maps to dfly_bench CLI flags. Total connections is
+// ProactorThreads * ConnsPerThread, mirroring memtier's Threads * Clients.
+type DflyBenchConfig struct {
+	ServerHost string
+	ServerPort int
+	// ProactorThreads is dfly_bench --proactor_threads (analogous to
+	// memtier --threads).
+	ProactorThreads int
+	// ConnsPerThread is dfly_bench -c, connections per thread (analogous to
+	// memtier --clients).
+	ConnsPerThread int
+	Pipeline       int
+	DataSize       int // value size in bytes (-d)
+
+	// Command is the arbitrary command template, e.g.
+	// "SET __key__ __data__". dfly_bench shares memtier's __key__/__data__
+	// placeholders, so the same CommandSpec templates work for both tools.
+	Command string
+	// KeyDist is dfly_bench --key_dist: U (uniform), N (normal), Z
+	// (zipfian), S (sequential). Defaults to U (≈ memtier's random pattern).
+	KeyDist string
+	// KeyMaximum sets --key_maximum when > 0.
+	KeyMaximum int
+	// TestTime runs for a fixed number of seconds (--test_time) instead of a
+	// fixed request count. 0 falls back to Requests (--n, per connection).
+	TestTime int
+	// Requests is --n, the number of requests per connection. Used only when
+	// TestTime == 0 (preload passes).
+	Requests int
+	// QPS is dfly_bench --qps, a per-connection rate cap. 0 means
+	// coordinated-omission / closed-loop, i.e. maximum load.
+	QPS int
+}
+
+// InstallDflyBench builds dfly_bench from the Dragonfly repo at the git ref
+// dflyRef (branch, tag, or sha) and installs it on the given host, skipping
+// the build if a binary is already on PATH. Pinning the ref matters because
+// dfly_bench's output format has changed over time; the memtier-compatible
+// --json_out_file report this package parses is only present on recent refs.
+func InstallDflyBench(ctx context.Context, ssh *remote.SSHClient, dflyRef string) error {
+	if dflyRef == "" {
+		dflyRef = "main"
+	}
+	// A shallow clone can leave the helio/ submodule empty, so submodules
+	// are initialized explicitly. `git clone --branch` only accepts
+	// branch/tag names, so a full clone + checkout is the fallback for shas.
+	script := fmt.Sprintf(`set -e
+if command -v dfly_bench >/dev/null 2>&1; then
+  echo "dfly_bench already installed: $(dfly_bench --version 2>&1 | head -1 || true)"
+  exit 0
+fi
+
+sudo sed -i -E '/^[[:space:]]*(deb|#|$)/!d' /etc/apt/sources.list 2>/dev/null || true
+sudo apt-get update -qq
+# Full Dragonfly build toolchain: dfly_bench is built out of the same
+# tree as the server so it needs the same deps.
+# Deps mirror Dragonfly's official build-from-source instructions:
+# https://github.com/dragonflydb/dragonfly#building-from-source
+# libboost-fiber-dev is the headline Boost package that pulls in the
+# rest of Boost Fiber + dependencies the helio third_party cmake needs.
+sudo apt-get install -y -qq \
+  git cmake ninja-build build-essential bison \
+  autoconf-archive libtool libssl-dev \
+  libboost-fiber-dev \
+  pkg-config zlib1g-dev libunwind-dev
+
+cd /tmp
+rm -rf dragonfly
+
+# Try the fast path first (branch/tag + shallow submodules, ~10-20s).
+# Fall back to a full clone + explicit checkout so --dfly-ref can be a
+# sha too.
+if git clone --depth 1 --shallow-submodules --recurse-submodules \
+     --branch %q https://github.com/dragonflydb/dragonfly.git 2>/dev/null; then
+  echo "shallow clone at branch/tag %q"
+else
+  echo "falling back to full clone to resolve ref %q"
+  git clone https://github.com/dragonflydb/dragonfly.git
+  cd dragonfly
+  git checkout %q
+  git submodule update --init --recursive
+  cd /tmp
+fi
+cd dragonfly
+
+# Belt and suspenders: ensure submodules are populated even if the
+# shallow --recurse-submodules clone skipped them for some reason.
+if [ ! -f helio/blaze.sh ]; then
+  git submodule update --init --recursive
+fi
+if [ ! -f helio/blaze.sh ]; then
+  echo "ERROR: helio/blaze.sh missing after submodule init; check repo state"
+  ls -la helio/ || true
+  exit 1
+fi
+
+# Build only the dfly_bench target; skip the server to keep the build
+# under ~3 minutes on the client instance.
+./helio/blaze.sh -release
+cd build-opt
+ninja dfly_bench
+
+# dfly_bench lands at either ./dfly_bench or ./tools/dfly_bench depending
+# on the Dragonfly revision; install whichever ninja produced.
+if [ -x ./dfly_bench ]; then
+  sudo install -m 0755 ./dfly_bench /usr/local/bin/dfly_bench
+elif [ -x ./tools/dfly_bench ]; then
+  sudo install -m 0755 ./tools/dfly_bench /usr/local/bin/dfly_bench
+else
+  echo "ERROR: dfly_bench binary not found after build"
+  find . -name dfly_bench -type f -executable 2>/dev/null || true
+  exit 1
+fi
+echo "dfly_bench installed: $(dfly_bench --version 2>&1 | head -1 || true)"
+`, dflyRef, dflyRef, dflyRef, dflyRef)
+
+	out, err := ssh.Execute(ctx, script)
+	if err != nil {
+		return fmt.Errorf("install dfly_bench: %w\n%s", err, out)
+	}
+	fmt.Fprintf(os.Stderr, "      %s\n", lastNonEmptyLine(out))
+	return nil
+}
+
+// RunDflyBench runs dfly_bench against the target server and pulls the
+// memtier-compatible JSON results file back over SSH. Stdout is captured too
+// so callers can reuse the plain-text fallback parser if needed.
+func RunDflyBench(ctx context.Context, ssh *remote.SSHClient, cfg *DflyBenchConfig, label string) ([]byte, string, error) {
+	jsonFile := fmt.Sprintf("/tmp/dflybench-results-%s.json", label)
+	cmd := BuildDflyBenchCmd(cfg, jsonFile)
+
+	fmt.Fprintf(os.Stderr, "      Running: %s\n", cmd)
+
+	// dfly_bench writes its periodic progress to the console; tee it to the
+	// operator's terminal while capturing it for the fallback parser.
+	var captured bytes.Buffer
+	tee := io.MultiWriter(os.Stderr, &captured)
+	stderr, err := ssh.ExecuteStream(ctx, cmd, tee)
+	if err != nil {
+		return nil, captured.String(), fmt.Errorf("dfly_bench failed: %w\nstderr: %s", err, stderr)
+	}
+
+	data, err := ssh.CopyFrom(ctx, jsonFile)
+	if err != nil {
+		return nil, captured.String(), fmt.Errorf("read dfly_bench results: %w", err)
+	}
+
+	return data, captured.String(), nil
+}
+
+// BuildDflyBenchCmd renders the dfly_bench command string. Flag names are
+// dfly_bench-native (--h/--p/-c/-d/--pipeline/--key_maximum/...), but the
+// --command template and connection math line up with memtier so the two
+// tools can be compared on the same workload.
+func BuildDflyBenchCmd(cfg *DflyBenchConfig, jsonFile string) string {
+	cmd := fmt.Sprintf(
+		`dfly_bench --h=%s --p=%d --proactor_threads=%d -c=%d `,
+		cfg.ServerHost, cfg.ServerPort, cfg.ProactorThreads, cfg.ConnsPerThread,
+	)
+
+	if cfg.TestTime > 0 {
+		cmd += fmt.Sprintf(`--test_time=%d `, cfg.TestTime)
+	} else {
+		cmd += fmt.Sprintf(`--n=%d `, cfg.Requests)
+	}
+
+	keyDist := cfg.KeyDist
+	if keyDist == "" {
+		keyDist = "U"
+	}
+	cmd += fmt.Sprintf(`--command=%q --key_dist=%s -d %d --pipeline=%d `,
+		cfg.Command, keyDist, cfg.DataSize, cfg.Pipeline)
+
+	if cfg.KeyMaximum > 0 {
+		cmd += fmt.Sprintf(`--key_maximum=%d `, cfg.KeyMaximum)
+	}
+
+	// --qps is always emitted: 0 means coordinated omission (closed-loop max
+	// load), whereas omitting it uses dfly_bench's built-in default of 20/sec.
+	cmd += fmt.Sprintf(`--qps=%d `, cfg.QPS)
+	cmd += fmt.Sprintf(`--json_out_file=%s`, jsonFile)
+	return cmd
+}
diff --git a/bench/dflybench_test.go b/bench/dflybench_test.go
new file mode 100644
index 0000000..04f9302
--- /dev/null
+++ b/bench/dflybench_test.go
@@ -0,0 +1,109 @@
+package bench
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestBuildDflyBenchCmd_SET(t *testing.T) {
+	cfg := &DflyBenchConfig{
+		ServerHost:      "10.0.0.1",
+		ServerPort:      6379,
+		ProactorThreads: 32,
+		ConnsPerThread:  5,
+		Pipeline:        30,
+		DataSize:        128,
+		Command:         "SET __key__ __data__",
+		KeyMaximum:      100_000_000,
+		TestTime:        300,
+		QPS:             0,
+	}
+	got := BuildDflyBenchCmd(cfg, "/tmp/out.json")
+	want := "dfly_bench --h=10.0.0.1 --p=6379 --proactor_threads=32 -c=5 " +
+		"--test_time=300 " +
+		`--command="SET __key__ __data__" --key_dist=U -d 128 --pipeline=30 ` +
+		"--key_maximum=100000000 " +
+		"--qps=0 " +
+		"--json_out_file=/tmp/out.json"
+	if got != want {
+		t.Fatalf("SET command drift:\n got: %q\nwant: %q", got, want)
+	}
+	// dfly_bench --command mode drives its own key distribution; --ratio
+	// must not appear.
+	if strings.Contains(got, "--ratio") {
+		t.Fatalf("command mode must not emit --ratio: %q", got)
+	}
+}
+
+func TestBuildDflyBenchCmd_GET(t *testing.T) {
+	cfg := &DflyBenchConfig{
+		ServerHost:      "10.0.0.1",
+		ServerPort:      6379,
+		ProactorThreads: 32,
+		ConnsPerThread:  5,
+		Pipeline:        120,
+		DataSize:        128,
+		Command:         "GET __key__",
+		KeyDist:         "U",
+		KeyMaximum:      100_000_000,
+		TestTime:        60,
+		QPS:             0,
+	}
+	got := BuildDflyBenchCmd(cfg, "/tmp/get.json")
+	want := "dfly_bench --h=10.0.0.1 --p=6379 --proactor_threads=32 -c=5 " +
+		"--test_time=60 " +
+		`--command="GET __key__" --key_dist=U -d 128 --pipeline=120 ` +
+		"--key_maximum=100000000 " +
+		"--qps=0 " +
+		"--json_out_file=/tmp/get.json"
+	if got != want {
+		t.Fatalf("GET command drift:\n got: %q\nwant: %q", got, want)
+	}
+}
+
+func TestBuildDflyBenchCmd_PreloadRequestBounded(t *testing.T) {
+	// TestTime == 0 => request-bounded (--n) preload with sequential keys.
+	cfg := &DflyBenchConfig{
+		ServerHost:      "10.0.0.1",
+		ServerPort:      6379,
+		ProactorThreads: 32,
+		ConnsPerThread:  1,
+		Pipeline:        30,
+		DataSize:        128,
+		Command:         "SET __key__ __data__",
+		KeyDist:         "S",
+		KeyMaximum:      100_000_000,
+		Requests:        3_125_000,
+		QPS:             0,
+	}
+	got := BuildDflyBenchCmd(cfg, "/tmp/preload.json")
+	if !strings.Contains(got, "--n=3125000") {
+		t.Fatalf("expected request-bounded preload (--n=3125000), got: %q", got)
+	}
+	if strings.Contains(got, "--test_time") {
+		t.Fatalf("TestTime=0 must not emit --test_time: %q", got)
+	}
+	if !strings.Contains(got, "--key_dist=S") {
+		t.Fatalf("expected sequential preload (--key_dist=S), got: %q", got)
+	}
+	if !strings.Contains(got, "-c=1") {
+		t.Fatalf("expected one connection per thread for preload (-c=1), got: %q", got)
+	}
+}
+
+func TestBuildDflyBenchCmd_DefaultsKeyDistToUniform(t *testing.T) {
+	cfg := &DflyBenchConfig{
+		ServerHost:      "10.0.0.1",
+		ServerPort:      6379,
+		ProactorThreads: 4,
+		ConnsPerThread:  10,
+		Pipeline:        20,
+		DataSize:        256,
+		Command:         "GET __key__",
+		TestTime:        30,
+	}
+	got := BuildDflyBenchCmd(cfg, "/tmp/out.json")
+	if !strings.Contains(got, "--key_dist=U") {
+		t.Fatalf("empty KeyDist should default to U, got: %q", got)
+	}
+}
diff --git a/bench/replication/dflybench.go b/bench/replication/dflybench.go
index c492a58..917a11f 100644
--- a/bench/replication/dflybench.go
+++ b/bench/replication/dflybench.go
@@ -8,7 +8,6 @@ import (
 	"os"
 	"regexp"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/dragonflydb/benchmarking/remote"
@@ -22,96 +21,6 @@ var (
 	p99Pattern = regexp.MustCompile(`p99_lat\(us\):\s*(\d+)`)
 )
 
-// InstallDflyBench builds dfly_bench from the Dragonfly repo at the git ref
-// dflyRef (branch, tag, or sha) and installs it on the given host, skipping
-// the build if a binary is already on PATH. Pinning the ref matters because
-// dfly_bench's stdout format has changed and the parsing regexes are coupled
-// to it.
-func InstallDflyBench(ctx context.Context, ssh *remote.SSHClient, dflyRef string) error {
-	if dflyRef == "" {
-		dflyRef = "main"
-	}
-	// A shallow clone can leave the helio/ submodule empty, so submodules
-	// are initialized explicitly. `git clone --branch` only accepts
-	// branch/tag names, so a full clone + checkout is the fallback for shas.
-	script := fmt.Sprintf(`set -e
-if command -v dfly_bench >/dev/null 2>&1; then
-  echo "dfly_bench already installed: $(dfly_bench --version 2>&1 | head -1 || true)"
-  exit 0
-fi
-
-sudo sed -i -E '/^[[:space:]]*(deb|#|$)/!d' /etc/apt/sources.list 2>/dev/null || true
-sudo apt-get update -qq
-# Full Dragonfly build toolchain: dfly_bench is built out of the same
-# tree as the server so it needs the same deps.
-# Deps mirror Dragonfly's official build-from-source instructions:
-# https://github.com/dragonflydb/dragonfly#building-from-source
-# libboost-fiber-dev is the headline Boost package that pulls in the
-# rest of Boost Fiber + dependencies the helio third_party cmake needs.
-sudo apt-get install -y -qq \
-  git cmake ninja-build build-essential bison \
-  autoconf-archive libtool libssl-dev \
-  libboost-fiber-dev \
-  pkg-config zlib1g-dev libunwind-dev
-
-cd /tmp
-rm -rf dragonfly
-
-# Try the fast path first (branch/tag + shallow submodules, ~10-20s).
-# Fall back to a full clone + explicit checkout so --dfly-ref can be a
-# sha too.
-if git clone --depth 1 --shallow-submodules --recurse-submodules \
-     --branch %q https://github.com/dragonflydb/dragonfly.git 2>/dev/null; then
-  echo "shallow clone at branch/tag %q"
-else
-  echo "falling back to full clone to resolve ref %q"
-  git clone https://github.com/dragonflydb/dragonfly.git
-  cd dragonfly
-  git checkout %q
-  git submodule update --init --recursive
-  cd /tmp
-fi
-cd dragonfly
-
-# Belt and suspenders: ensure submodules are populated even if the
-# shallow --recurse-submodules clone skipped them for some reason.
-if [ ! -f helio/blaze.sh ]; then
-  git submodule update --init --recursive
-fi
-if [ ! -f helio/blaze.sh ]; then
-  echo "ERROR: helio/blaze.sh missing after submodule init; check repo state"
-  ls -la helio/ || true
-  exit 1
-fi
-
-# Build only the dfly_bench target; skip the server to keep the build
-# under ~3 minutes on the client instance.
-./helio/blaze.sh -release
-cd build-opt
-ninja dfly_bench
-
-# dfly_bench lands at either ./dfly_bench or ./tools/dfly_bench depending
-# on the Dragonfly revision; install whichever ninja produced.
-if [ -x ./dfly_bench ]; then
-  sudo install -m 0755 ./dfly_bench /usr/local/bin/dfly_bench
-elif [ -x ./tools/dfly_bench ]; then
-  sudo install -m 0755 ./tools/dfly_bench /usr/local/bin/dfly_bench
-else
-  echo "ERROR: dfly_bench binary not found after build"
-  find . -name dfly_bench -type f -executable 2>/dev/null || true
-  exit 1
-fi
-echo "dfly_bench installed: $(dfly_bench --version 2>&1 | head -1 || true)"
-`, dflyRef, dflyRef, dflyRef, dflyRef)
-
-	out, err := ssh.Execute(ctx, script)
-	if err != nil {
-		return fmt.Errorf("install dfly_bench: %w\n%s", err, out)
-	}
-	fmt.Fprintf(os.Stderr, "      %s\n", lastNonEmptyLine(out))
-	return nil
-}
-
 // BuildDflyBenchArgs renders the dfly_bench command string for one case.
 // Mode is either "preload" (populate the keyspace) or "load" (sustained
 // traffic for cfg.TestTime). Flag choices mirror
@@ -217,12 +126,3 @@ func ParseDflyBenchLine(line string) (rps int, p99us int, matchedRPS, matchedP99
 	}
 	return
 }
-
-func lastNonEmptyLine(s string) string {
-	s = strings.TrimRight(s, "\n")
-	idx := strings.LastIndex(s, "\n")
-	if idx < 0 {
-		return strings.TrimSpace(s)
-	}
-	return strings.TrimSpace(s[idx+1:])
-}
diff --git a/bench/replication/scenario.go b/bench/replication/scenario.go
index 102f341..71fe726 100644
--- a/bench/replication/scenario.go
+++ b/bench/replication/scenario.go
@@ -9,6 +9,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/dragonflydb/benchmarking/bench"
 	"github.com/dragonflydb/benchmarking/infra"
 	"github.com/dragonflydb/benchmarking/remote"
 )
@@ -39,7 +40,7 @@ func Run(ctx context.Context, state *infra.SetupState, cfg *ReplicationConfig, o
 	defer replicaSSH.Close()
 
 	fmt.Fprintf(out, "[1/4] Ensuring dfly_bench is installed on client (ref=%s)...\n", cfg.DflyRef)
-	if err := InstallDflyBench(ctx, clientSSH, cfg.DflyRef); err != nil {
+	if err := bench.InstallDflyBench(ctx, clientSSH, cfg.DflyRef); err != nil {
 		return nil, fmt.Errorf("install dfly_bench: %w", err)
 	}
 
diff --git a/benchmarks/GET/dfly_bench/GET_reproduce.md b/benchmarks/GET/dfly_bench/GET_reproduce.md
new file mode 100644
index 0000000..fa23d50
--- /dev/null
+++ b/benchmarks/GET/dfly_bench/GET_reproduce.md
@@ -0,0 +1,115 @@
+# GET benchmark (Dragonfly) with dfly_bench — reproduce & analysis
+
+Reproduction of the 128-byte GET workload on `m7g.8xlarge` (Graviton3, 32 vCPU)
+using Dragonfly's own load generator, `dfly_bench`, instead of memtier. At the
+same 160 connections, GET throughput and tail latency trade off directly against
+the pipeline depth:
+
+- **Unthrottled, deep pipeline (120):** ~13.5M ops/s but p99.9 blows up to ~33 ms.
+- **Recommended (pipeline 30, unthrottled):** **10.31M ops/s at p99.9 = 0.996 ms** —
+  best tail, still above 10M ops/s.
+- **Pipeline 30 throttled to ~11M:** 11.82M ops/s at p99.9 = 7.67 ms.
+- **Engine comparison (pipeline 30, unthrottled):** Dragonfly **9.88M ops/s** vs
+  valkey **1.61M** and redis **1.48M** — Dragonfly is ~6.1–6.7x the throughput at a
+  fraction of the latency (p99.9 1.12 ms vs 8.79 ms / 6.96 ms).
+
+All configs are CPU-bound on the server (Dragonfly saturates its shards); the tail
+is governed by how many requests are kept in flight (`conns x pipeline`).
+
+## Setup
+
+Provisioned with `dfbench setup` (installs both memtier and dfly_bench on the
+client; `dfly_bench` is built from `dragonflydb/dragonfly` at `--dfly-bench-ref`,
+default `main`):
+
+```
+dfbench setup --id maxqps2 --tune-network --ubuntu-version 24.04 \
+  --server-instance m7g.8xlarge  --server-arch arm64 \
+  --client-instance c6gn.8xlarge --client-arch arm64 \
+  --engines dragonfly,redis,valkey --availability-zone us-east-1c
+```
+
+Note: pick an availability zone that has `c6gn.8xlarge` capacity (AWS suggested
+`us-east-1b/c/d/f`; `us-east-1a`/`1e` were unavailable at run time). Server and
+client share a cluster placement group, so both instance types must have capacity
+in the chosen AZ.
+
+| Component | Value |
+| --------- | ----- |
+| Server | m7g.8xlarge (arm64), Ubuntu 24.04, `--tune-network` |
+| Client | c6gn.8xlarge (arm64) |
+| Dragonfly build | `s3://df-use1-pub/v2/dragonfly-aarch64.tar.gz` (`i1.39.4-32-gcc0f424a`) |
+| Dragonfly flags | `--num_shards 16`, `--conn_use_incoming_cpu` |
+| Load generator | `dfly_bench` (ref `main`) |
+| Value size | 128 B |
+| Keyspace | 100,000,000 keys (`--key_maximum`) |
+| Connections | 160 (`--dfly-threads 32` x `--dfly-conns 5`) |
+| Key distribution | `U` (uniform, ≈ memtier random pattern `R`) |
+
+## Test command (recommended config: pipeline 30, unthrottled)
+
+```
+dfbench run-commands-dfly --id maxqps2 --engines dragonfly,redis,valkey --command GET \
+  --dragonfly_version s3://df-use1-pub/v2/dragonfly-aarch64.tar.gz \
+  --dragonfly_num_shards 16 \
+  --key-maximum 100000000 --key-dist U \
+  --dfly-threads 32 --dfly-conns 5 --dfly-pipeline 30 --dfly-data-size 128 \
+  --qps 0 --test-time 60 --warmup-time 10 --trials 1
+```
+
+`--qps 0` is closed-loop max load (coordinated omission). To pin throughput near a
+target rate instead, set `--qps` per connection = target_total / 160 (e.g.
+`--qps 68750` ≈ 11M ops/s total). Redis/Valkey are the preinstalled builds started
+with `io_threads=32` (auto); only Dragonfly is reinstalled from `--dragonfly_version`.
+
+## Engine comparison (dragonfly vs redis vs valkey)
+
+Latest 3-engine run (benchmark `20260721-114528`, pipeline 30, `--qps 0`, 160
+connections, 60s trial, value size 128B). Redis 8.8.0 and Valkey 9.1.0 with
+`io_threads=32`; each engine is preloaded to 100M keys by dfly_bench, then measured
+back-to-back with a 30s cooldown between engines.
+
+| Engine | Throughput (median) | p50 | p99 | p99.9 | Avg latency |
+| ------ | ------------------- | --- | --- | ----- | ----------- |
+| **dragonfly** | **9.88M ops/s** | 0.457 ms | 0.785 ms | 1.123 ms | 0.463 ms |
+| valkey | 1.61M ops/s | 2.544 ms | 7.885 ms | 8.791 ms | 2.959 ms |
+| redis  | 1.48M ops/s | 3.250 ms | 6.076 ms | 6.956 ms | 3.218 ms |
+
+**Takeaway:** at the same client config Dragonfly delivers ~6.1x Valkey and ~6.7x
+Redis throughput while keeping p99.9 near 1 ms (vs ~7–9 ms). Redis/Valkey are
+single-shard event loops that cannot spread a uniform GET load across all 32 cores
+the way Dragonfly's 16 shards do, so they saturate a small number of threads well
+below the box's capacity. (Dragonfly here reads 9.88M ops/s vs the 10.31M in the
+pipeline sweep below — the two runs are on different EC2 instances from separate
+`dfbench setup`s, so expect ~5% run-to-run variance.)
+
+## Results (Dragonfly, pipeline sweep)
+
+Server: m7g.8xlarge (arm64) | Client: c6gn.8xlarge (arm64) | value size 128B, 160 connections, 60s trial
+
+| Pipeline | Rate       | Throughput | p50      | p99      | p99.9      | Avg latency |
+| -------- | ---------- | ---------- | -------- | -------- | ---------- | ----------- |
+| 120      | `--qps 0`  | 13.47M ops/s | 1.064 ms | 7.435 ms | 33.307 ms | 1.344 ms |
+| 80       | `--qps 0`  | 12.98M ops/s | 0.780 ms | 3.971 ms | 15.049 ms | 0.928 ms |
+| 30       | `--qps 68750` (~11M) | 11.82M ops/s | 1.064 ms | 3.669 ms | 7.666 ms | 1.173 ms |
+| **30**   | **`--qps 0`** | **10.31M ops/s** | **0.431 ms** | **0.765 ms** | **0.996 ms** | **0.442 ms** |
+
+**Reading the sweep:** with 160 connections at max load, tail latency scales with
+in-flight depth (`conns x pipeline`). Dropping the pipeline from 120 to 30 cuts
+p99.9 by ~33x (33.3 ms -> 1.0 ms) while only trimming throughput ~24% (13.5M ->
+10.3M), because Dragonfly is CPU-bound and 160x30 still keeps every shard busy.
+Throttling pipeline-30 up toward 11M (`--qps 68750`) recovers throughput but
+re-inflates the tail to ~7.7 ms — still under 10 ms, if you want the extra ops/s.
+
+## Comparison with memtier
+
+The memtier reproduction (see [../memtier/GET_reproduce.md](../memtier/GET_reproduce.md))
+measured Dragonfly at **11.63M ops/s, p99.9 = 7.97 ms** with pipeline 120. dfly_bench
+at pipeline 120 pushes harder (13.5M ops/s) but with a much longer tail, so the two
+tools are not directly comparable at identical pipeline settings — dfly_bench's
+closed-loop driver keeps deeper queues. For a memtier-like tail, use a lower
+pipeline (30) or throttle with `--qps`.
+
+Both tools' results share the same schema (this run is tagged `tool="dfly_bench"`),
+so a memtier GET file and a dfly_bench GET file diff with `dfbench compare`.
+
diff --git a/benchmarks/GET/GET_reproduce.md b/benchmarks/GET/memtier/GET_reproduce.md
similarity index 81%
rename from benchmarks/GET/GET_reproduce.md
rename to benchmarks/GET/memtier/GET_reproduce.md
index 0c5fbde..6319c37 100644
--- a/benchmarks/GET/GET_reproduce.md
+++ b/benchmarks/GET/memtier/GET_reproduce.md
@@ -51,6 +51,36 @@ command execution, so they cannot use the 32 cores the way Dragonfly's shards do
 collapse): throughput drops ~18% and latency ~2.5×, while server CPU utilization also drops — a
 sign of too much in-flight work, not a capacity gain. Use `--memtier-clients 5`.
 
+### Same run with dfly_bench
+
+`dfbench run-commands-dfly` reproduces this GET workload with Dragonfly's own
+load generator instead of memtier. dfly_bench is installed on the client at
+`dfbench setup` time (alongside memtier); override the build with
+`--dfly-bench-ref`.
+
+```
+dfbench run-commands-dfly --id maxqps2 --engines dragonfly --command GET \
+  --dragonfly_version s3://df-use1-pub/v2/dragonfly-aarch64.tar.gz \
+  --dragonfly_num_shards 16 \
+  --dfly-threads 32 --dfly-conns 5 --dfly-pipeline 120 --dfly-data-size 128 \
+  --key-maximum 100000000 --key-dist U --qps 0 \
+  --test-time 60 --warmup-time 10 --trials 1
+```
+
+Notes:
+- **dfly_bench self-preloads.** dfly_bench uses its own key prefix
+  (`--key_prefix`, default `key:`), which differs from memtier's keys, so GET's
+  SET preload is run by dfly_bench (sequential distribution, one connection per
+  thread covering the whole `--key_maximum` range) rather than reused from a
+  memtier preload.
+- **Flag mapping:** `--dfly-threads` -> `--proactor_threads`, `--dfly-conns` ->
+  `-c` (32x5 = 160 connections), `--key-dist U` approximates memtier's random
+  key pattern `R`, `--qps 0` = closed-loop max load.
+- Results use the same schema (tagged `tool="dfly_bench"`), so
+  `dfbench compare <memtier-get.json> <dflybench-get.json>` diffs the two tools.
+- dfly_bench's `--json_out_file` writes a memtier-compatible latency report, so
+  the same parser produces the throughput / p50 / p99 / p99.9 / avg columns.
+
 ## Bottleneck: CPU, not network (verified)
 
 Two independent measurements taken during the same 60s measured trial.
diff --git a/benchmarks/GET/ena/after.txt b/benchmarks/GET/memtier/ena/after.txt
similarity index 100%
rename from benchmarks/GET/ena/after.txt
rename to benchmarks/GET/memtier/ena/after.txt
diff --git a/benchmarks/GET/ena/before.txt b/benchmarks/GET/memtier/ena/before.txt
similarity index 100%
rename from benchmarks/GET/ena/before.txt
rename to benchmarks/GET/memtier/ena/before.txt
diff --git a/benchmarks/GET/ena/timeseries.txt b/benchmarks/GET/memtier/ena/timeseries.txt
similarity index 100%
rename from benchmarks/GET/ena/timeseries.txt
rename to benchmarks/GET/memtier/ena/timeseries.txt
diff --git a/benchmarks/GET/mpstat/get-mpstat.log b/benchmarks/GET/memtier/mpstat/get-mpstat.log
similarity index 100%
rename from benchmarks/GET/mpstat/get-mpstat.log
rename to benchmarks/GET/memtier/mpstat/get-mpstat.log
diff --git a/benchmarks/SET/dfly_bench/SET_reproduce.md b/benchmarks/SET/dfly_bench/SET_reproduce.md
new file mode 100644
index 0000000..f739a6e
--- /dev/null
+++ b/benchmarks/SET/dfly_bench/SET_reproduce.md
@@ -0,0 +1,90 @@
+# SET benchmark (Dragonfly vs Redis vs Valkey) with dfly_bench — reproduce & results
+
+Reproduction of the 128-byte SET workload on `m7g.8xlarge` (Graviton3, 32 vCPU)
+using Dragonfly's own load generator, `dfly_bench`, instead of memtier, across all
+three engines at 160 connections / pipeline 30 / max load:
+
+- **dragonfly 7.78M ops/s** (p99.9 11.5 ms) — ~7.4x Redis and ~7.4x Valkey.
+- **redis 1.06M ops/s** (p99.9 13.4 ms), **valkey 1.05M ops/s** (p99.9 15.8 ms).
+- **Near-perfect memtier parity:** dfly_bench SET matches the memtier SET numbers
+  within ~2% (Dragonfly 7.78M vs 7.94M), because SET is ingress-bandwidth-bound at
+  ~15 Gbps, so both load generators hit the same network ceiling regardless of the
+  client-side driver.
+
+## Setup
+
+Provisioned with `dfbench setup` (installs both memtier and dfly_bench on the
+client; `dfly_bench` is built from `dragonflydb/dragonfly` at `--dfly-bench-ref`,
+default `main`):
+
+```
+dfbench setup --id maxqps2 --tune-network --ubuntu-version 24.04 \
+  --server-instance m7g.8xlarge  --server-arch arm64 \
+  --client-instance c6gn.8xlarge --client-arch arm64 \
+  --engines dragonfly,redis,valkey --availability-zone us-east-1c
+```
+
+Note: pick an availability zone that has `c6gn.8xlarge` capacity (AWS suggested
+`us-east-1b/c/d/f`). Server and client share a cluster placement group, so both
+instance types must have capacity in the chosen AZ.
+
+| Component | Value |
+| --------- | ----- |
+| Server | m7g.8xlarge (arm64), Ubuntu 24.04, `--tune-network` |
+| Client | c6gn.8xlarge (arm64) |
+| Dragonfly build | `s3://df-use1-pub/v2/dragonfly-aarch64.tar.gz` (`i1.39.4-32-gcc0f424a`) |
+| Dragonfly flags | `--num_shards 16`, `--conn_use_incoming_cpu` |
+| Redis / Valkey | Redis 8.8.0 / Valkey 9.1.0, `io_threads=8` |
+| Load generator | `dfly_bench` (ref `main`) |
+| Value size | 128 B |
+| Keyspace | 100,000,000 keys (`--key_maximum`) |
+| Connections | 160 (`--dfly-threads 32` x `--dfly-conns 5`) |
+| Key distribution | `U` (uniform, ≈ memtier random pattern `R`) |
+
+## Test command
+
+```
+dfbench run-commands-dfly --id maxqps2 --engines dragonfly,redis,valkey --command SET \
+  --dragonfly_version s3://df-use1-pub/v2/dragonfly-aarch64.tar.gz \
+  --dragonfly_num_shards 16 --dragonfly_conn_use_incoming_cpu=true \
+  --key-maximum 100000000 --key-dist U \
+  --dfly-threads 32 --dfly-conns 5 --dfly-pipeline 30 --dfly-data-size 128 \
+  --qps 0 --test-time 300 --warmup-time 10 --trials 1 \
+  --redis_io_threads 8 --valkey_io_threads 8
+```
+
+## Results
+
+Latest 3-engine run (benchmark `20260721-115805`, pipeline 30, `--qps 0`, 160
+connections, 1 x 300s trial, value size 128B).
+
+| Engine | Throughput (median) | p50 | p99 | p99.9 | Avg latency |
+| ------ | ------------------- | --- | --- | ----- | ----------- |
+| **dragonfly** | **7.78M ops/s** | 0.483 ms | 3.271 ms | 11.505 ms | 0.600 ms |
+| redis  | 1.06M ops/s | 3.896 ms | 7.571 ms | 13.354 ms | 4.534 ms |
+| valkey | 1.05M ops/s | 4.461 ms | 11.501 ms | 15.771 ms | 4.541 ms |
+
+**Takeaway:** Dragonfly delivers ~7.4x the write throughput of Redis/Valkey. Redis
+and Valkey are effectively single-threaded on the command path, so more `io_threads`
+does not lift SET much; Dragonfly spreads writes across its 16 shards. All three
+engines' RPS is capped by the server's ~15 Gbps ingress budget at this value size
+(the throttling shows up as periodic dips in the redis/valkey RPS trace).
+
+## Comparison with memtier
+
+memtier SET reproduction (see [../memtier/SET_reproduce.md](../memtier/SET_reproduce.md)):
+
+| Engine | memtier throughput | dfly_bench throughput | Δ |
+| ------ | ------------------ | --------------------- | - |
+| dragonfly | 7.94M ops/s | 7.78M ops/s | -2.0% |
+| redis     | 1.06M ops/s | 1.06M ops/s | ~0% |
+| valkey    | 1.07M ops/s | 1.05M ops/s | -1.9% |
+
+Unlike GET (where dfly_bench's closed-loop driver pushes noticeably harder than
+memtier), **SET reproduces almost exactly** across the two tools. SET is
+bandwidth-bound rather than CPU/queue-bound, so both drivers converge on the same
+ceiling and the tail latencies line up too (Dragonfly p99.9 11.5 ms vs memtier
+10.5 ms).
+
+Both tools' results share the same schema (this run is tagged `tool="dfly_bench"`),
+so `dfbench compare <memtier-set.json> <dflybench-set.json>` diffs them directly.
diff --git a/benchmarks/SET/SET_reproduce.md b/benchmarks/SET/memtier/SET_reproduce.md
similarity index 57%
rename from benchmarks/SET/SET_reproduce.md
rename to benchmarks/SET/memtier/SET_reproduce.md
index e2d30d0..7ca3655 100644
--- a/benchmarks/SET/SET_reproduce.md
+++ b/benchmarks/SET/memtier/SET_reproduce.md
@@ -25,3 +25,23 @@ Server: m7g.8xlarge (arm64) | Client: c6gn.8xlarge (arm64)
 | dragonfly | 7.94M ops/s         | 0.503 ms | 2.127 ms  | 10.495 ms | 0.592 ms    |
 | redis     | 1.06M ops/s         | 3.935 ms | 7.583 ms  | 12.735 ms | 4.511 ms    |
 | valkey    | 1.07M ops/s         | 4.479 ms | 10.815 ms | 15.039 ms | 4.493 ms    |
+
+### Same run with dfly_bench
+
+`dfbench run-commands-dfly` reproduces the same workload with Dragonfly's own
+load generator instead of memtier. dfly_bench is installed on the client at
+`dfbench setup` time (alongside memtier); override the build with
+`--dfly-bench-ref`.
+
+```
+dfbench run-commands-dfly --id test --engines dragonfly --command SET \
+  --dragonfly_num_shards 16 \
+  --dfly-threads 32 --dfly-conns 5 --dfly-pipeline 30 --dfly-data-size 128 \
+  --key-maximum 100000000 --qps 0 \
+  --test-time 300 --trials 1 --dragonfly_version s3://df-use1-pub/v2/dragonfly-aarch64.tar.gz
+```
+
+Flag mapping vs memtier: `--dfly-threads` -> `--proactor_threads`, `--dfly-conns`
+-> `-c` (so 32x5 = 160 connections, same as memtier threadsxclients), `--qps 0`
+= closed-loop max load. Results use the same schema (tagged `tool="dfly_bench"`),
+so `dfbench compare <memtier-set.json> <dflybench-set.json>` diffs the two tools.
diff --git a/cmd/root.go b/cmd/root.go
index 085814e..da9da72 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -23,6 +23,7 @@ Teardown is a single DeleteStack call.`,
 		newSetupCommand(),
 		newRunCommand(),
 		newRunCommandsCommand(),
+		newRunCommandsDflyCommand(),
 		newRunReplicationCommand(),
 		newCompareCommand(),
 		newQuickstartCommand(),
diff --git a/cmd/run_commands_dfly.go b/cmd/run_commands_dfly.go
new file mode 100644
index 0000000..bddf980
--- /dev/null
+++ b/cmd/run_commands_dfly.go
@@ -0,0 +1,282 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"github.com/dragonflydb/benchmarking/bench/commands"
+	"github.com/dragonflydb/benchmarking/infra"
+)
+
+// newRunCommandsDflyCommand builds `dfbench run-commands-dfly`, the
+// dfly_bench counterpart to run-commands. It drives the same per-command
+// flush/preload/warmup/trials cycle but with Dragonfly's own load
+// generator, so the two tools can be compared on identical workloads. The
+// existing memtier run-commands flow is untouched.
+func newRunCommandsDflyCommand() *cobra.Command {
+	var (
+		id           string
+		enginesFlag  string
+		commandFlag  string
+		threads      int
+		conns        int
+		pipeline     int
+		dataSize     int
+		keyMaximum   int
+		keyDist      string
+		qps          int
+		testTime     int
+		warmupTime   int
+		trials       int
+		dflyBenchRef string
+		output       string
+
+		// Dragonfly-specific server tuning (identical to run-commands).
+		dflyNumShards    int
+		dflyConnIncoming bool
+		dflyVersion      string
+		redisIOThreads   int
+		valkeyIOThreads  int
+		redisPinNetwork  bool
+		valkeyPinNetwork bool
+	)
+
+	cmd := &cobra.Command{
+		Use:   "run-commands-dfly",
+		Short: "Benchmark a single Redis command with dfly_bench against an existing setup",
+		Long: `Runs a dedicated dfly_bench pass for one Redis command (SET, GET,
+INCR, LPUSH, ...) against every selected engine, mirroring the memtier
+'dfbench run-commands' flow but using Dragonfly's own load generator.
+
+dfly_bench uses its own key prefix (--key_prefix, default "key:"), so
+read commands (GET, ...) are preloaded by dfly_bench itself rather than
+memtier. For each engine:
+
+  1. FLUSHALL, then preload the keyspace with dfly_bench if the command
+     reads existing data.
+  2. Optional time-boxed warmup pass (results discarded).
+  3. --trials measured trials of --test-time seconds each.
+
+Results use the same schema as run-commands (tagged tool="dfly_bench")
+and land under ~/.dfbench/runs/<id>/commands/<timestamp>/, so a memtier
+file and a dfly_bench file for the same command diff with 'dfbench
+compare'.`,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			if id == "" {
+				return fmt.Errorf("--id is required; use 'dfbench list' to see available setups")
+			}
+			engines, ok := parseEngines(enginesFlag)
+			if !ok {
+				return fmt.Errorf("invalid --engines value %q: must be comma-separated list of dragonfly,redis,valkey", enginesFlag)
+			}
+			spec, err := commands.Lookup(commandFlag)
+			if err != nil {
+				return fmt.Errorf("invalid --command: %w", err)
+			}
+			if trials < 1 {
+				return fmt.Errorf("--trials must be >= 1, got %d", trials)
+			}
+			if testTime <= 0 {
+				return fmt.Errorf("--test-time must be > 0 seconds, got %d", testTime)
+			}
+			if warmupTime < 0 {
+				return fmt.Errorf("--warmup-time must be >= 0 seconds, got %d", warmupTime)
+			}
+			if keyMaximum <= 0 {
+				return fmt.Errorf("--key-maximum must be > 0, got %d", keyMaximum)
+			}
+			tuning, err := resolveEngineTuning(cmd.Flags())
+			if err != nil {
+				return err
+			}
+			if !ValidOutput[output] {
+				return fmt.Errorf("invalid --output %q: must be one of table|json|csv", output)
+			}
+
+			cfg := &commands.Config{
+				SetupID:            id,
+				Command:            spec,
+				Engines:            engines,
+				Threads:            threads,
+				Clients:            conns,
+				Pipeline:           pipeline,
+				DataSize:           dataSize,
+				KeyMaximum:         keyMaximum,
+				KeyDist:            keyDist,
+				QPS:                qps,
+				TestTime:           testTime,
+				WarmupTime:         warmupTime,
+				Trials:             trials,
+				DflyBenchRef:       dflyBenchRef,
+				IOThreads:          map[string]int{"redis": tuning.redisIO, "valkey": tuning.valkeyIO},
+				NumShards:          tuning.numShards,
+				ConnUseIncomingCPU: tuning.connIncoming,
+				PinNetwork:         map[string]bool{"redis": tuning.redisPin, "valkey": tuning.valkeyPin},
+				DflyVersion:        tuning.version,
+			}
+			return runCommandsDfly(cmd.Context(), cfg, output)
+		},
+	}
+
+	f := cmd.Flags()
+	f.StringVar(&id, "id", "", "Setup ID previously created by 'dfbench setup' (required)")
+	f.StringVar(&enginesFlag, "engines", "dragonfly",
+		"Comma-separated list of engines to benchmark (must be a subset of the setup's installed engines)")
+	f.StringVar(&commandFlag, "command", "",
+		"Command to benchmark (required), one of: "+strings.Join(commands.SupportedNames(), ","))
+	f.IntVar(&threads, "dfly-threads", 32, "dfly_bench --proactor_threads (analogous to memtier threads)")
+	f.IntVar(&conns, "dfly-conns", 5, "dfly_bench -c, connections per thread (analogous to memtier clients)")
+	f.IntVar(&pipeline, "dfly-pipeline", 20, "dfly_bench --pipeline: max pending requests per connection")
+	f.IntVar(&dataSize, "dfly-data-size", 256, "dfly_bench -d: value size in bytes (fills __data__ placeholders)")
+	f.IntVar(&keyMaximum, "key-maximum", 100_000_000,
+		"Key range (--key_maximum). Size it to 30-90% of server memory. Commands with multi-item structured preloads use a built-in 1M-key range instead.")
+	f.StringVar(&keyDist, "key-dist", "U",
+		"dfly_bench --key_dist for the measured pass: U (uniform, the memtier random equivalent), N (normal), Z (zipfian), S (sequential)")
+	f.IntVar(&qps, "qps", 0, "dfly_bench --qps per connection; 0 means closed-loop max load (coordinated omission)")
+	f.IntVar(&testTime, "test-time", 300, "Seconds of measured load per trial")
+	f.IntVar(&warmupTime, "warmup-time", 10, "Seconds of discarded warmup load per (engine, command); 0 disables")
+	f.IntVar(&trials, "trials", 3, "Number of measured trials per (engine, command) (>=1)")
+	f.StringVar(&dflyBenchRef, "dfly-bench-ref", "main",
+		"Git ref (branch/tag/sha) of dragonflydb/dragonfly to build dfly_bench from if it isn't already installed on the client")
+
+	// Dragonfly-specific server tuning (ignored by Redis/Valkey).
+	f.IntVar(&dflyNumShards, "dragonfly_num_shards", -1,
+		"Dragonfly --num_shards: -1 auto-derives VCPUS - 2*NUM_RX_QUEUES on the server, 0 uses Dragonfly's default, >0 sets it explicitly.")
+	f.BoolVar(&dflyConnIncoming, "dragonfly_conn_use_incoming_cpu", true,
+		"Run Dragonfly with --conn_use_incoming_cpu. Pair with 'dfbench setup --tune-network'. Disable with =false.")
+	f.StringVar(&dflyVersion, "dragonfly_version", "",
+		"Reinstall Dragonfly at this build before benchmarking: a release tag, 'latest', a full http(s):// or s3:// tarball URL, or a local path. Empty keeps whatever 'dfbench setup' installed.")
+
+	// Redis/Valkey tuning, split per engine (ignored by Dragonfly).
+	f.IntVar(&redisIOThreads, "redis_io_threads", 0,
+		"Override --io-threads for Redis (0 = one per vCPU).")
+	f.IntVar(&valkeyIOThreads, "valkey_io_threads", 0,
+		"Override --io-threads for Valkey (0 = one per vCPU).")
+	f.BoolVar(&redisPinNetwork, "redis_pin_network", true,
+		"Best-effort taskset pinning for Redis onto CPU 0 plus the network-I/O CPUs. Disable with =false.")
+	f.BoolVar(&valkeyPinNetwork, "valkey_pin_network", true,
+		"Best-effort taskset pinning for Valkey onto CPU 0 plus the network-I/O CPUs. Disable with =false.")
+
+	f.StringVar(&output, "output", "table", "Stdout format: table|json|csv (results.json is always written)")
+
+	return cmd
+}
+
+// runCommandsDfly wraps commands.RunDfly with state locking, status
+// transitions and result emission, mirroring runCommands.
+func runCommandsDfly(ctx context.Context, cfg *commands.Config, output string) error {
+	state, err := infra.LoadState(cfg.SetupID)
+	if err != nil {
+		return fmt.Errorf("load setup %q: %w (use 'dfbench list' to see available setups)", cfg.SetupID, err)
+	}
+	if state.Status != infra.StatusReady {
+		return fmt.Errorf("setup %q is not ready (status=%q); only a ready setup can be benchmarked", cfg.SetupID, state.Status)
+	}
+	if err := requireEnginesInstalled(state, cfg.Engines); err != nil {
+		return err
+	}
+
+	runDir, err := infra.RunDir(cfg.SetupID)
+	if err != nil {
+		return err
+	}
+	lock, err := infra.AcquireLock(runDir)
+	if err != nil {
+		return err
+	}
+	defer lock.Release()
+
+	bmID := time.Now().UTC().Format("20060102-150405")
+	startedAt := time.Now()
+	cfg.OutputDir = filepath.Join(runDir, "commands", bmID)
+
+	logTo := logWriter(output)
+
+	fmt.Fprintln(logTo)
+	fmt.Fprintln(logTo, "╔══════════════════════════════════════════════════════╗")
+	fmt.Fprintln(logTo, "║              dfbench run-commands-dfly               ║")
+	fmt.Fprintln(logTo, "╚══════════════════════════════════════════════════════╝")
+	fmt.Fprintln(logTo)
+	fmt.Fprintf(logTo, "  Setup:        %s\n", cfg.SetupID)
+	fmt.Fprintf(logTo, "  Benchmark ID: %s\n", bmID)
+	fmt.Fprintf(logTo, "  Engines:      %s\n", strings.Join(cfg.Engines, ", "))
+	fmt.Fprintf(logTo, "  Command:      %s (%s)\n", cfg.Command.Name, cfg.Command.Template)
+	fmt.Fprintf(logTo, "  Server:       %s (%s) %s\n", state.Server.InstanceType, state.Server.Arch, state.Server.PublicIP)
+	fmt.Fprintf(logTo, "  Client:       %s (%s) %s\n", state.Client.InstanceType, state.Client.Arch, state.Client.PublicIP)
+	fmt.Fprintf(logTo, "  Trials:       %d x %ds (warmup=%ds)\n", cfg.Trials, cfg.TestTime, cfg.WarmupTime)
+	fmt.Fprintf(logTo, "  dfly_bench:   threads=%d conns=%d pipeline=%d data-size=%d key-maximum=%d key-dist=%s qps=%d\n",
+		cfg.Threads, cfg.Clients, cfg.Pipeline, cfg.DataSize, cfg.KeyMaximum, cfg.KeyDist, cfg.QPS)
+	fmt.Fprintf(logTo, "  dfly-bench-ref: %s\n", cfg.DflyBenchRef)
+	fmt.Fprintf(logTo, "  IO threads:   %s (dragonfly unaffected)\n", ioThreadsDesc(cfg.IOThreads))
+	fmt.Fprintf(logTo, "  Dragonfly:    num_shards=%s conn_use_incoming_cpu=%t\n", numShardsDesc(cfg.NumShards), cfg.ConnUseIncomingCPU)
+	if cfg.DflyVersion != "" {
+		fmt.Fprintf(logTo, "  Dfly version: %s (reinstalled before run; redis/valkey unaffected)\n", cfg.DflyVersion)
+	}
+	fmt.Fprintf(logTo, "  Pin network:  %s (redis/valkey taskset; dragonfly unaffected)\n", pinNetworkDesc(cfg.PinNetwork))
+	fmt.Fprintf(logTo, "  Output dir:   %s\n", cfg.OutputDir)
+	fmt.Fprintln(logTo)
+
+	state.Status = infra.StatusBenchmarking
+	_ = state.Save()
+
+	bmRecord := infra.BenchmarkRun{
+		ID:        bmID,
+		StartedAt: startedAt,
+		Status:    "running",
+		Engines:   cfg.Engines,
+		Trials:    cfg.Trials,
+	}
+
+	defer func() {
+		state.Status = infra.StatusReady
+		_ = state.Save()
+	}()
+
+	ctx, stopSignals := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
+	defer stopSignals()
+
+	preinstalled := make(map[string]string, len(state.Engines))
+	for _, e := range state.Engines {
+		preinstalled[e.Name] = e.Version
+	}
+
+	results, runErr := commands.RunDfly(ctx, logTo, state, cfg, preinstalled)
+	if runErr != nil {
+		bmRecord.Status = "failed"
+		bmRecord.CompletedAt = time.Now()
+		state.Benchmarks = append(state.Benchmarks, bmRecord)
+		return runErr
+	}
+
+	bmRecord.CompletedAt = time.Now()
+	bmRecord.Status = "completed"
+	bmRecord.ResultsPath = filepath.Join("commands", bmID, "results.json")
+	state.Benchmarks = append(state.Benchmarks, bmRecord)
+
+	switch output {
+	case "json":
+		if err := commands.WriteJSON(os.Stdout, results); err != nil {
+			fmt.Fprintf(logTo, "  Warning: emit JSON: %v\n", err)
+		}
+	case "csv":
+		if err := commands.WriteCSV(os.Stdout, results); err != nil {
+			fmt.Fprintf(logTo, "  Warning: emit CSV: %v\n", err)
+		}
+	default:
+		commands.PrintTable(os.Stdout, results)
+	}
+
+	fmt.Fprintf(logTo, "  Results: %s\n", filepath.Join(cfg.OutputDir, "results.json"))
+	fmt.Fprintln(logTo)
+	fmt.Fprintf(logTo, "  Setup %q is still running. Destroy with: dfbench destroy --id %s\n", cfg.SetupID, cfg.SetupID)
+
+	return nil
+}
diff --git a/cmd/setup.go b/cmd/setup.go
index 4238594..8a38da0 100644
--- a/cmd/setup.go
+++ b/cmd/setup.go
@@ -43,6 +43,10 @@ type SetupFlags struct {
 	Replicas        int
 	ReplicaInstance string
 	ReplicaArch     string
+
+	// DflyBenchRef is the dragonflydb/dragonfly git ref used to build
+	// dfly_bench on the client (installed alongside memtier).
+	DflyBenchRef string
 }
 
 func newSetupCommand() *cobra.Command {
@@ -128,6 +132,8 @@ by repeated 'dfbench run --id <id>' invocations until you run
 	f.IntVar(&flags.Replicas, "replicas", 0, "Number of replica EC2 instances to provision for dfbench run-replication (0 disables)")
 	f.StringVar(&flags.ReplicaInstance, "replica-instance", "", "EC2 instance type for each replica (defaults to --server-instance)")
 	f.StringVar(&flags.ReplicaArch, "replica-arch", "", "CPU architecture for each replica: x86_64 | arm64 (defaults to --server-arch)")
+	f.StringVar(&flags.DflyBenchRef, "dfly-bench-ref", "main",
+		"Git ref (branch/tag/sha) of dragonflydb/dragonfly used when building dfly_bench on the client, installed alongside memtier for 'dfbench run-commands-dfly'")
 
 	return cmd
 }
@@ -268,6 +274,7 @@ func runSetup(ctx context.Context, flags *SetupFlags) error {
 			ReplicaInstance:  flags.ReplicaInstance,
 			ReplicaArch:      flags.ReplicaArch,
 			TuneNetwork:      flags.TuneNetwork,
+			DflyBenchRef:     flags.DflyBenchRef,
 		},
 		StackName: stackName,
 		Server: infra.InstanceInfo{
@@ -353,6 +360,10 @@ func runSetup(ctx context.Context, flags *SetupFlags) error {
 	if err := bench.InstallMemtier(ctx, clientSSH); err != nil {
 		return markFailed(fmt.Errorf("install memtier: %w", err))
 	}
+	fmt.Fprintf(out, "      -> Installing dfly_bench on client (ref=%s)...\n", flags.DflyBenchRef)
+	if err := bench.InstallDflyBench(ctx, clientSSH, flags.DflyBenchRef); err != nil {
+		return markFailed(fmt.Errorf("install dfly_bench: %w", err))
+	}
 
 	fmt.Fprintf(out, "[6/%d] Pre-installing engines on server: %s\n", totalSteps, strings.Join(flags.Engines, ", "))
 	for _, name := range flags.Engines {
diff --git a/infra/state.go b/infra/state.go
index 058e745..9efbb65 100644
--- a/infra/state.go
+++ b/infra/state.go
@@ -60,6 +60,10 @@ type SetupConfig struct {
 	// TuneNetwork records whether net_tune.sh was applied to the server
 	// during setup (IRQ/RPS/XPS affinity for high-throughput runs).
 	TuneNetwork bool `json:"tune_network,omitempty"`
+	// DflyBenchRef records the dragonflydb/dragonfly git ref used to build
+	// dfly_bench on the client at setup time. Empty means the historical
+	// default (no dfly_bench installed / "main").
+	DflyBenchRef string `json:"dfly_bench_ref,omitempty"`
 }
 
 type InstanceInfo struct {