diff --git a/.gitignore b/.gitignore index 8f3cd37e..30495ddc 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,4 @@ current-user-asks.md __pycache__/ /.hawkeye-bin/ /.claude +/plans/ diff --git a/AGENTS.md b/AGENTS.md index d0866998..85400243 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -111,6 +111,10 @@ git diff origin/main..HEAD \ && echo "REVIEW each hit" || echo "diff clean" ``` +Planning and scratch artifacts are not repository documentation. Keep `plans/`, +implementation logs, and agent working notes out of commits; durable design +documentation belongs under `docs/`. + ## 5) Investigate rocm-cli Before Editing Understand existing patterns first: diff --git a/Cargo.lock b/Cargo.lock index 59dd53de..29ed2780 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -134,6 +134,16 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b0f477b951e452a0b6b4a10b53ccd569042d1d01729b519e02074a9c0958a063" +[[package]] +name = "assert-json-diff" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e4f2b81832e72834d7518d8487a0396a28cc408186a2e8854c0f98011faf12" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "async-broadcast" version = "0.7.2" @@ -958,6 +968,24 @@ version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" +[[package]] +name = "deadpool" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b" +dependencies = [ + "deadpool-runtime", + "lazy_static", + "num_cpus", + "tokio", +] + +[[package]] +name = "deadpool-runtime" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b" + [[package]] name = "deltae" version = "0.3.2" @@ -2366,6 +2394,16 @@ dependencies = [ "libm", ] +[[package]] +name = "num_cpus" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" +dependencies = [ + "hermit-abi", + "libc", +] + [[package]] name = "num_threads" version = "0.1.7" @@ -3260,6 +3298,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tracing", + "wiremock", ] [[package]] @@ -5343,6 +5382,29 @@ dependencies = [ "memchr", ] +[[package]] +name = "wiremock" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08db1edfb05d9b3c1542e521aea074442088292f00b5f28e435c714a98f85031" +dependencies = [ + "assert-json-diff", + "base64", + "deadpool", + "futures", + "http", + "http-body-util", + "hyper", + "hyper-util", + "log", + "once_cell", + "regex", + "serde", + "serde_json", + "tokio", + "url", +] + [[package]] name = "wit-bindgen" version = "0.57.1" diff --git a/MANIFEST.md b/MANIFEST.md index dd85043a..4ea4ad73 100644 --- a/MANIFEST.md +++ b/MANIFEST.md @@ -48,6 +48,7 @@ repository. | apple-native-keyring-store | 1.0.0 | MIT OR Apache-2.0 | | approx | 0.5.1 | Apache-2.0 | | as-any | 0.3.2 | MIT OR Apache-2.0 | +| assert-json-diff | 2.0.2 | MIT | | async-broadcast | 0.7.2 | MIT OR Apache-2.0 | | async-channel | 2.5.0 | Apache-2.0 OR MIT | | async-executor | 1.14.0 | Apache-2.0 OR MIT | @@ -126,6 +127,8 @@ repository. | darling_core | 0.23.0 | MIT | | darling_macro | 0.23.0 | MIT | | data-encoding | 2.11.0 | MIT | +| deadpool | 0.12.3 | MIT OR Apache-2.0 | +| deadpool-runtime | 0.1.4 | MIT OR Apache-2.0 | | deltae | 0.3.2 | MIT | | der | 0.7.10 | Apache-2.0 OR MIT | | deranged | 0.5.8 | MIT OR Apache-2.0 | @@ -274,6 +277,7 @@ repository. | num-iter | 0.1.45 | MIT OR Apache-2.0 | | num-rational | 0.4.2 | MIT OR Apache-2.0 | | num-traits | 0.2.19 | MIT OR Apache-2.0 | +| num_cpus | 1.17.0 | MIT OR Apache-2.0 | | num_threads | 0.1.7 | MIT OR Apache-2.0 | | objc2-core-foundation | 0.3.2 | Zlib OR Apache-2.0 OR MIT | | object | 0.37.3 | Apache-2.0 OR MIT | @@ -543,6 +547,7 @@ repository. | windows_x86_64_msvc | 0.53.1 | MIT OR Apache-2.0 | | winnow | 0.7.15 | MIT | | winnow | 1.0.3 | MIT | +| wiremock | 0.6.5 | MIT/Apache-2.0 | | wit-bindgen | 0.57.1 | Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT | | writeable | 0.6.3 | Unicode-3.0 | | xattr | 1.6.1 | MIT OR Apache-2.0 | diff --git a/apps/rocm/src/dash.rs b/apps/rocm/src/dash.rs index 3ad3d441..efbbe38d 100644 --- a/apps/rocm/src/dash.rs +++ b/apps/rocm/src/dash.rs @@ -43,7 +43,11 @@ pub fn runner_options( ) -> RunnerOptions { let d = &config.dashboard.daemon; RunnerOptions { - bench_csv: d.bench_results_dir.clone(), + bench_csv: Some( + d.bench_results_dir + .clone() + .unwrap_or_else(|| default_bench_csv_path(paths)), + ), enable_docker, image_patterns: None, gpu_tick: Duration::from_secs_f64(d.gpu_tick_secs), @@ -201,6 +205,7 @@ pub fn resolved_args( // The real executor is injected in `run_async` for a live dash; None // here keeps demo/replay/mock behaving exactly as today. tool_executor: None, + bench_results_dir: config.dashboard.daemon.bench_results_dir.clone(), } } @@ -319,6 +324,149 @@ pub fn run_chat(chat_mock: bool) -> Result<()> { rt.block_on(run_async(config, paths, args, None, chat_mock)) } +/// CLI arguments for `rocm bench load`. +pub struct BenchLoadArgs { + pub endpoint: String, + pub model: Option, + pub concurrency: Vec, + pub isl: u32, + pub osl: u32, + pub requests: u32, + pub out: Option, + pub auto_ramp: bool, +} + +/// Default `--out` path for `rocm bench load`: the same file the telemetry +/// daemon tails by default (`DashboardDaemonConfig::bench_results_dir`), so a +/// plain CLI run populates the dashboard's Bench panel without any config +/// edits. Extracted as a pure function of `AppPaths` for testability. +fn default_bench_csv_path(paths: &AppPaths) -> std::path::PathBuf { + paths.data_dir.join("bench").join("results.csv") +} + +fn ensure_bench_csv_parent(csv_path: &std::path::Path) -> Result<()> { + if let Some(parent) = csv_path.parent() + && !parent.as_os_str().is_empty() + { + std::fs::create_dir_all(parent) + .with_context(|| format!("creating bench output dir {}", parent.display()))?; + } + Ok(()) +} + +/// Entry point for `rocm bench load`. +/// +/// Runs a concurrency sweep against a local http:// endpoint and appends one +/// aggregate CSV row per concurrency level. Output defaults to the daemon's +/// tailed `/bench/results.csv` unless `--out` is specified explicitly. +pub fn run_bench(a: BenchLoadArgs) -> Result<()> { + use rocm_dash_daemon::bench_load::{LoadSpec, run_and_append_csv, run_auto_ramp}; + + let BenchLoadArgs { + endpoint, + model, + concurrency, + isl, + osl, + requests, + out, + auto_ramp, + } = a; + + // Reject https:// — no TLS backend compiled in for the load generator. + // Compare on the lowercased scheme so HTTPS:// is also caught. + if endpoint.to_lowercase().starts_with("https://") { + anyhow::bail!( + "rocm bench load supports http:// endpoints only (no TLS backend compiled in)" + ); + } + + // Resolve the model: use the provided value or probe GET {endpoint}/v1/models. + let model = if let Some(m) = model { + m + } else { + let models_url = format!("{}/v1/models", endpoint.trim_end_matches('/')); + let resp = ureq::get(&models_url) + .timeout(std::time::Duration::from_secs(5)) + .call() + .context("fetching /v1/models to detect the default model")?; + let body: serde_json::Value = resp.into_json().context("parsing /v1/models response")?; + rocm_dash_tui::llm::pick_first_model(&body).with_context(|| { + format!("no model found at {endpoint}/v1/models — pass --model explicitly") + })? + }; + + // Resolve the output path: default to the same `/bench/results.csv` + // file the daemon tails. Rows are appended, matching the `CsvBenchTailer` + // semantics. Create parents for explicit and default paths alike. + let csv_path = match out { + Some(path) => path, + None => default_bench_csv_path(&AppPaths::discover()?), + }; + ensure_bench_csv_parent(&csv_path)?; + + let spec = LoadSpec { + endpoint: endpoint.clone(), + model: model.clone(), + input_len: isl, + output_len: osl, + requests, + }; + + println!("mode: raw serving throughput (synthetic prompts) — not agent-workload."); + if auto_ramp { + println!( + "endpoint={endpoint} model={model} mode=auto-ramp isl={isl} osl={osl} requests={requests}" + ); + } else { + println!( + "endpoint={endpoint} model={model} concurrency={} isl={isl} osl={osl} requests={requests}", + concurrency + .iter() + .map(ToString::to_string) + .collect::>() + .join(",") + ); + } + + let rt = build_dashboard_runtime()?; + let rows = if auto_ramp { + rt.block_on(run_auto_ramp(&spec, &csv_path)) + .context("running bench auto-ramp")? + } else { + rt.block_on(run_and_append_csv(&spec, &concurrency, &csv_path)) + .context("running bench load sweep")? + }; + + for row in &rows { + let conc = row + .concurrency + .map_or_else(|| "-".to_string(), |v| v.to_string()); + let gen_tps = row + .gen_tps + .map_or_else(|| "-".to_string(), |v| format!("{v:.1}")); + let prompt_tps = row + .prompt_tps + .map_or_else(|| "-".to_string(), |v| format!("{v:.1}")); + let wall = row + .wall_s + .map_or_else(|| "-".to_string(), |v| format!("{v:.2}")); + let n = row + .n_requests + .map_or_else(|| "-".to_string(), |v| v.to_string()); + println!( + "cell={} concurrency={conc} gen_tps={gen_tps} prompt_tps={prompt_tps} wall={wall}s n={n}", + row.cell + ); + } + println!( + "note: local saturation smoke-test — client-measured throughput, not an official ROCm/AMD benchmark." + ); + println!("wrote {} row(s) to {}", rows.len(), csv_path.display()); + + Ok(()) +} + /// Entry point for `rocm bootstrap setup`. Routes to the same focused Setup host /// as the launcher's "Set up this system" row — the first-run onboarding wizard /// (install ROCm SDK / adopt an existing folder), with no daemon or tab shell. @@ -457,6 +605,36 @@ mod tests { assert!(!opts.enable_docker); } + #[test] + fn runner_options_derives_default_bench_csv_from_current_paths() { + let p = paths(); + let opts = runner_options(&cfg(), &p, false); + assert_eq!(opts.bench_csv, Some(default_bench_csv_path(&p))); + } + + #[test] + fn runner_options_preserves_explicit_bench_csv_override() { + let p = paths(); + let mut config = cfg(); + let explicit = PathBuf::from("/var/rocm/custom-bench.csv"); + config.dashboard.daemon.bench_results_dir = Some(explicit.clone()); + + let opts = runner_options(&config, &p, false); + assert_eq!(opts.bench_csv, Some(explicit)); + } + + #[test] + fn ensure_bench_csv_parent_creates_explicit_output_directory() { + let root = + std::env::temp_dir().join(format!("rocm-cli-bench-parent-{}", std::process::id())); + let csv = root.join("nested").join("results.csv"); + + ensure_bench_csv_parent(&csv).unwrap(); + assert!(csv.parent().unwrap().is_dir()); + + let _ = std::fs::remove_dir_all(root); + } + #[test] fn bootstrap_routes_to_focused_setup() { // `rocm bootstrap setup` must route to the same focused Setup host as the diff --git a/apps/rocm/src/main.rs b/apps/rocm/src/main.rs index c3fc1db8..405c2aa4 100644 --- a/apps/rocm/src/main.rs +++ b/apps/rocm/src/main.rs @@ -400,6 +400,12 @@ rocm logs --search error timeout")] #[arg(long)] chat_mock: bool, }, + /// Saturate a local OpenAI-compatible endpoint and report rough client-side + /// throughput (local smoke-test, NOT an official ROCm/AMD benchmark). + Bench { + #[command(subcommand)] + command: BenchCommand, + }, /// Remove ROCm CLI-managed files from this computer. Uninstall { /// Do not ask for interactive confirmation. @@ -426,6 +432,47 @@ rocm logs --search error timeout")] }, } +#[derive(Subcommand, Debug)] +enum BenchCommand { + /// Run a concurrency sweep (RAW serving throughput, synthetic single-shot + /// requests — not agent-shaped, not comparable to *-agent-bench harnesses). + /// + /// Measures RAW serving throughput (synthetic single-shot requests, vLLM + /// benchmark_serving shape). It does NOT reproduce agent-shaped, multi-turn, + /// long-context tool traffic and is not comparable to the *-agent-bench quality + /// harnesses. + Load { + /// Base URL of the OpenAI-compatible endpoint, e.g. http://127.0.0.1:8000 + #[arg(long, value_name = "URL")] + endpoint: String, + /// Model name (defaults to the first model returned by GET {endpoint}/v1/models) + #[arg(long)] + model: Option, + /// Concurrency levels to sweep, comma-separated + #[arg(long, value_delimiter = ',', default_value = "1,8,32,64", value_parser = clap::value_parser!(u32).range(1..=128))] + concurrency: Vec, + /// Input sequence length in tokens + #[arg(long, default_value_t = 1024, value_parser = clap::value_parser!(u32).range(1..=32768))] + isl: u32, + /// Output sequence length in tokens + #[arg(long, default_value_t = 1024, value_parser = clap::value_parser!(u32).range(1..=32768))] + osl: u32, + /// Requests per concurrency cell + #[arg(long, default_value_t = 128, value_parser = clap::value_parser!(u32).range(1..=10000))] + requests: u32, + /// Output CSV file (default: ~/.rocm/bench/results.csv, the + /// daemon-tailed path that populates the dashboard's Bench panel) + #[arg(long, value_name = "FILE")] + out: Option, + /// Ramp concurrency automatically (1,2,4,8,16,32,64,128), stopping at saturation. + /// + /// Ignores --concurrency when set. Stops early when gen_tps plateaus + /// or the request queue backs up. + #[arg(long)] + auto_ramp: bool, + }, +} + #[derive(Subcommand, Debug)] enum InstallTarget { /// Install TheRock ROCm wheels into a Python folder managed by ROCm CLI. @@ -1567,6 +1614,27 @@ fn dispatch(cli: Cli) -> Result<()> { demo, chat_mock, }) => dash::run(replay, demo, chat_mock), + Some(Command::Bench { command }) => match command { + BenchCommand::Load { + endpoint, + model, + concurrency, + isl, + osl, + requests, + out, + auto_ramp, + } => dash::run_bench(dash::BenchLoadArgs { + endpoint, + model, + concurrency, + isl, + osl, + requests, + out, + auto_ramp, + }), + }, Some(Command::Uninstall { yes, dry_run, @@ -15598,6 +15666,7 @@ fn treat_as_natural_language(args: &[String]) -> bool { "logs", "daemon", "dash", + "bench", "uninstall", "help", "--help", @@ -15631,6 +15700,57 @@ mod tests { Cli::command().debug_assert(); } + #[test] + fn bench_load_rejects_zero_and_unbounded_numeric_arguments() { + for args in [ + [ + "rocm", + "bench", + "load", + "--endpoint", + "http://localhost:8000", + "--concurrency", + "0", + ] + .as_slice(), + [ + "rocm", + "bench", + "load", + "--endpoint", + "http://localhost:8000", + "--isl", + "32769", + ] + .as_slice(), + [ + "rocm", + "bench", + "load", + "--endpoint", + "http://localhost:8000", + "--osl", + "32769", + ] + .as_slice(), + [ + "rocm", + "bench", + "load", + "--endpoint", + "http://localhost:8000", + "--requests", + "10001", + ] + .as_slice(), + ] { + assert!( + Cli::try_parse_from(args).is_err(), + "accepted invalid args: {args:?}" + ); + } + } + fn possible_values_listed_in_help(help: &str) -> Vec { let marker = "[possible values:"; let start = help @@ -19552,6 +19672,7 @@ install therock"; "logs", "daemon", "dash", + "bench", "uninstall", "completions", "help", @@ -19574,6 +19695,29 @@ install therock"; Cli::try_parse_from(["rocm", "comfy", "logs"]).expect("comfy alias should parse"); } + #[test] + fn t5_bench_load_clap_parse_smoke() { + // T5: verify BenchCommand::Load parses correctly including comma-separated concurrency. + let cli = Cli::try_parse_from([ + "rocm", + "bench", + "load", + "--endpoint", + "http://x", + "--concurrency", + "1,8,32,64", + ]) + .expect("rocm bench load should parse"); + match cli.command { + Some(Command::Bench { + command: BenchCommand::Load { concurrency, .. }, + }) => { + assert_eq!(concurrency, vec![1u32, 8, 32, 64]); + } + other => panic!("expected Bench/Load, got {other:?}"), + } + } + #[test] fn setup_reset_cli_output_is_plain_and_persists_first_time_prompt() -> Result<()> { let (_root, paths) = test_paths("setup-reset-cli"); diff --git a/crates/rocm-core/src/lib.rs b/crates/rocm-core/src/lib.rs index afb1e28d..3795987c 100644 --- a/crates/rocm-core/src/lib.rs +++ b/crates/rocm-core/src/lib.rs @@ -3942,7 +3942,9 @@ pub struct DashboardDaemonConfig { pub discovery_tick_secs: f64, #[serde(default = "default_instance_tick_secs")] pub instance_tick_secs: f64, - /// Watch this directory for new normalized benchmark CSVs. + /// Watch this file for new normalized benchmark CSV rows. When unset, the + /// daemon derives `/bench/results.csv` from the current `AppPaths` + /// at startup so machine-specific paths are never persisted in config. #[serde(default, skip_serializing_if = "Option::is_none")] pub bench_results_dir: Option, } @@ -7849,6 +7851,18 @@ Class Name: Display assert_eq!(d.instance_tick(), Duration::from_secs(3)); } + #[test] + fn dashboard_bench_results_path_is_derived_not_persisted() { + let config = DashboardDaemonConfig::default(); + assert_eq!(config.bench_results_dir, None); + + let json = serde_json::to_value(config).unwrap(); + assert!( + json.get("bench_results_dir").is_none(), + "machine-specific derived path must not be serialized" + ); + } + #[test] fn app_paths_expose_telemetry_and_daemon_log_paths() -> Result<()> { let (root, paths) = temp_app_paths("telemetry-paths"); diff --git a/crates/rocm-dash-collectors/Cargo.toml b/crates/rocm-dash-collectors/Cargo.toml index d25d2d67..d6aa631a 100644 --- a/crates/rocm-dash-collectors/Cargo.toml +++ b/crates/rocm-dash-collectors/Cargo.toml @@ -28,3 +28,4 @@ reqwest = { version = "0.12", default-features = false } [dev-dependencies] tokio = { version = "1", features = ["full"] } +wiremock = "0.6" diff --git a/crates/rocm-dash-collectors/src/bench_load.rs b/crates/rocm-dash-collectors/src/bench_load.rs new file mode 100644 index 00000000..d9a9575d --- /dev/null +++ b/crates/rocm-dash-collectors/src/bench_load.rs @@ -0,0 +1,1236 @@ +// Copyright © Advanced Micro Devices, Inc., or its affiliates. +// +// SPDX-License-Identifier: MIT + +//! Concurrency-sweep load generator for local OpenAI-compatible endpoints. +//! +//! Produces one aggregate [`BenchmarkRow`] per concurrency cell and appends +//! them to a CSV file that a running daemon tails via [`CsvBenchTailer`]. +//! Quality fields are left at their defaults (`PassFail::Unknown`). + +use std::fs::OpenOptions; +use std::io::Write; +use std::path::Path; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Instant; + +use reqwest::Client; +use rocm_dash_core::bench_schema::BenchmarkRow; +use rocm_dash_core::traits::InstanceSample; +use serde_json::Value; +use tokio::sync::Semaphore; +use tokio::task::JoinSet; + +/// Timeout for /metrics scrapes (short; must never stall the cell sweep). +const METRICS_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1); + +/// Poll interval for the mid-cell Prometheus poller. +const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(250); + +/// Sentinel stored before any successful paired running/waiting scrape. +const NO_SAMPLE_PAIR: u64 = u64::MAX; + +/// Fixed minimal header written once when a CSV file is new or empty. +pub const CSV_HEADER: &str = "cell,run,concurrency,model,engine,input_len,output_len,\ + n_requests,prompt_tokens,completion_tokens,prompt_tps,gen_tps,wall_s,launcher,\ + max_running_reqs,max_waiting_reqs,ttft_ms,tpot_ms\n"; + +/// Parameters for a single concurrency-level load cell. +#[derive(Debug, Clone)] +pub struct LoadSpec { + /// Base URL of the OpenAI-compatible endpoint, e.g. `http://127.0.0.1:8000`. + pub endpoint: String, + /// Model name to pass in the request body. + pub model: String, + /// Number of input tokens to request (approximated via `max_tokens` prompt). + pub input_len: u32, + /// Number of output tokens to request. + pub output_len: u32, + /// Total number of requests to send at this concurrency level. + pub requests: u32, +} + +/// Aggregate result from one successful or partially-successful response. +struct Outcome { + prompt_tokens: u64, + completion_tokens: u64, +} + +/// Error type for the bench load writer. +#[derive(Debug, thiserror::Error)] +pub enum BenchLoadError { + /// HTTP client construction or send failure. + #[error("http: {0}")] + Http(#[from] reqwest::Error), + /// CSV serialization failure. + #[error("csv: {0}")] + Csv(#[from] csv::Error), + /// File I/O failure. + #[error("io: {0}")] + Io(#[from] std::io::Error), + /// Concurrency must be at least one. + #[error("concurrency must be at least 1 (got {0})")] + InvalidConcurrency(u32), + /// Existing file has a different CSV header; refusing to corrupt it. + #[error( + "refusing to append: {path} has a different header; pass --out or remove the incompatible file" + )] + HeaderMismatch { + /// Path of the file with the conflicting header. + path: String, + }, +} + +/// Build the `/metrics` URL from an OpenAI-compatible endpoint base URL. +/// +/// Returns `None` if the endpoint cannot be parsed (no host:port component). +fn metrics_url(endpoint: &str) -> Option { + let url_base = endpoint.trim_end_matches('/'); + let (scheme, rest) = if let Some(r) = url_base.strip_prefix("https://") { + ("https", r) + } else if let Some(r) = url_base.strip_prefix("http://") { + ("http", r) + } else { + ("http", url_base) + }; + let host_port = rest.split('/').next()?; + Some(format!("{scheme}://{host_port}/metrics")) +} + +/// Scrape Prometheus `/metrics` using the supplied client. +/// +/// Returns `None` on any error (non-vLLM, 404, network failure, parse +/// garbage). Never panics. The caller is responsible for supplying a client +/// with an appropriate timeout. +async fn try_scrape_prom(client: &Client, endpoint: &str) -> Option { + let url = metrics_url(endpoint)?; + let resp = client.get(&url).send().await.ok()?; + if !resp.status().is_success() { + return None; + } + let text = resp.text().await.ok()?; + Some(crate::vllm_prom::parse(&text)) +} + +fn pack_peak_pair(running: u32, waiting: u32) -> u64 { + (u64::from(waiting) << 32) | u64::from(running) +} + +const fn unpack_peak_pair(value: u64) -> (u32, u32) { + (value as u32, (value >> 32) as u32) +} + +fn update_peak_pair(peak: &AtomicU64, running: u32, waiting: u32) { + let candidate = pack_peak_pair(running, waiting); + let _ = peak.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + if current == NO_SAMPLE_PAIR || running > unpack_peak_pair(current).0 { + Some(candidate) + } else { + None + } + }); +} + +fn peak_pair(peak: &AtomicU64) -> Option<(u32, u32)> { + let value = peak.load(Ordering::Relaxed); + (value != NO_SAMPLE_PAIR).then(|| unpack_peak_pair(value)) +} + +struct BenchClients { + post: Client, + metrics: Client, +} + +impl BenchClients { + fn new() -> Result { + Ok(Self { + post: Client::builder() + .timeout(std::time::Duration::from_mins(5)) + .build()?, + metrics: Client::builder().timeout(METRICS_TIMEOUT).build()?, + }) + } +} + +/// Send `spec.requests` POST `/chat/completions` requests with `concurrency` +/// in-flight at a time. +/// +/// Returns one aggregate `BenchmarkRow` with client-side `gen_tps` and +/// `prompt_tps`. Per-request failures are isolated: a non-2xx response or +/// missing `usage` fields excludes that request from the sums but does not +/// abort the cell. +pub async fn run_cell(spec: &LoadSpec, concurrency: u32) -> Result { + run_cell_with_clients(spec, concurrency, &BenchClients::new()?).await +} + +async fn run_cell_with_clients( + spec: &LoadSpec, + concurrency: u32, + clients: &BenchClients, +) -> Result { + if concurrency == 0 { + return Err(BenchLoadError::InvalidConcurrency(concurrency)); + } + let sem = Arc::new(Semaphore::new(concurrency as usize)); + let url = format!("{}/chat/completions", spec.endpoint.trim_end_matches('/')); + + // Before scrape: used only for TTFT/TPOT histogram deltas. + let prom_before = try_scrape_prom(&clients.metrics, &spec.endpoint).await; + + // Keep the queue pair from the sample with the highest running count so + // saturation compares values observed at the same instant. + let peak_queue = Arc::new(AtomicU64::new(NO_SAMPLE_PAIR)); + let stop_flag = Arc::new(AtomicBool::new(false)); + + // Spawn the mid-cell poller before any POST requests so it can observe + // the rising queue depth as requests are dispatched. + let poller = { + let metrics_client = clients.metrics.clone(); + let endpoint = spec.endpoint.clone(); + let peak_queue = Arc::clone(&peak_queue); + let stop_flag = Arc::clone(&stop_flag); + tokio::spawn(async move { + loop { + if stop_flag.load(Ordering::Relaxed) { + break; + } + if let Some(sample) = try_scrape_prom(&metrics_client, &endpoint).await + && let (Some(running), Some(waiting)) = + (sample.running_reqs, sample.waiting_reqs) + { + update_peak_pair(&peak_queue, running, waiting); + } + tokio::time::sleep(POLL_INTERVAL).await; + } + }) + }; + + // Capture makespan BEFORE spawning so the clock includes queue wait time. + let t0 = Instant::now(); + + let mut js: JoinSet> = JoinSet::new(); + for _ in 0..spec.requests { + let client = clients.post.clone(); + let sem = Arc::clone(&sem); + let url = url.clone(); + let model = spec.model.clone(); + let output_len = spec.output_len; + let input_len = spec.input_len; + + js.spawn(async move { + // Named binding: permit is held for the entire request. + let _permit = sem.acquire_owned().await.ok()?; + + let body = serde_json::json!({ + "model": model, + "messages": [{"role": "user", "content": "x".repeat(input_len as usize)}], + "max_tokens": output_len, + "temperature": 0.0, + "stream": false, + }); + + let resp = client + .post(&url) + .header("Content-Type", "application/json") + .body(body.to_string()) + .send() + .await + .ok()?; + + if !resp.status().is_success() { + return None; + } + + let text = resp.text().await.ok()?; + let v: Value = serde_json::from_str(&text).ok()?; + let usage = v.get("usage")?; + + let prompt_tokens = usage.get("prompt_tokens")?.as_u64()?; + let completion_tokens = usage.get("completion_tokens")?; + // Treat missing/zero completion_tokens as failure (excluded from sums). + let completion_tokens = completion_tokens.as_u64()?; + if completion_tokens == 0 { + return None; + } + + Some(Outcome { + prompt_tokens, + completion_tokens, + }) + }); + } + + let mut sum_prompt: u64 = 0; + let mut sum_completion: u64 = 0; + let mut n_success: u32 = 0; + + while let Some(res) = js.join_next().await { + if let Ok(Some(outcome)) = res { + sum_prompt += outcome.prompt_tokens; + sum_completion += outcome.completion_tokens; + n_success += 1; + } + } + + // Stop the poller and wait for it to exit cleanly. + stop_flag.store(true, Ordering::Relaxed); + let _ = poller.await; + + // After scrape: used only for TTFT/TPOT histogram deltas. + let prom_after = try_scrape_prom(&clients.metrics, &spec.endpoint).await; + + let makespan_s = t0.elapsed().as_secs_f64(); + + let gen_tps = if makespan_s > 0.0 && n_success > 0 { + Some(sum_completion as f64 / makespan_s) + } else { + None + }; + let prompt_tps = if makespan_s > 0.0 && n_success > 0 { + Some(sum_prompt as f64 / makespan_s) + } else { + None + }; + + // Running and waiting are retained from one real-time observation. + let (max_running_reqs, max_waiting_reqs) = peak_pair(&peak_queue) + .map_or((None, None), |(running, waiting)| { + (Some(running), Some(waiting)) + }); + + // TTFT/TPOT deltas come from the before/after histogram scrapes (unchanged). + let (_, _, ttft_ms, tpot_ms) = prom_deltas(prom_before.as_ref(), prom_after.as_ref()); + + Ok(BenchmarkRow { + cell: format!("bench-c{concurrency}"), + run: 1, + model: Some(spec.model.clone()), + concurrency: Some(concurrency), + input_len: Some(spec.input_len), + output_len: Some(spec.output_len), + n_requests: Some(n_success), + prompt_tokens: Some(sum_prompt), + completion_tokens: Some(sum_completion), + prompt_tps, + gen_tps, + wall_s: Some(makespan_s), + launcher: Some("rocm bench load (local smoke)".to_string()), + max_running_reqs, + max_waiting_reqs, + ttft_ms, + tpot_ms, + ..Default::default() + }) +} + +/// Compute latency deltas from two Prometheus samples. +/// +/// Returns `((), (), ttft_ms, tpot_ms)` — the first two elements are `None` +/// placeholder for the deprecated peak fields (peaks now come from the +/// mid-cell poller; this function only computes histogram deltas). +/// If either sample is `None`, both latency fields are `None`. +fn prom_deltas( + before: Option<&InstanceSample>, + after: Option<&InstanceSample>, +) -> (Option, Option, Option, Option) { + let (Some(b), Some(a)) = (before, after) else { + return (None, None, None, None); + }; + + let ttft_ms = latency_delta_ms(b.ttft_sum_s, b.ttft_count, a.ttft_sum_s, a.ttft_count); + let tpot_ms = latency_delta_ms(b.tpot_sum_s, b.tpot_count, a.tpot_sum_s, a.tpot_count); + + (None, None, ttft_ms, tpot_ms) +} + +/// Compute `Δsum / Δcount * 1000` (ms). +/// +/// Returns `None` if either input is `None`, if the count delta is not +/// positive (avoids division by zero and nonsense from stale counters), +/// or if the sum delta is negative (counter reset). +fn latency_delta_ms( + sum_before: Option, + count_before: Option, + sum_after: Option, + count_after: Option, +) -> Option { + let delta_sum = sum_after? - sum_before?; + let delta_count = count_after? - count_before?; + if delta_count <= 0.0 || delta_sum < 0.0 { + return None; + } + Some(delta_sum / delta_count * 1000.0) +} + +/// Concurrency levels tried by [`run_auto_ramp`] in order. +pub const RAMP_SEQUENCE: &[u32] = &[1, 2, 4, 8, 16, 32, 64, 128]; + +/// Minimum fractional `gen_tps` improvement to keep ramping. +pub const PLATEAU_GAIN: f64 = 0.05; + +/// Open (or create) `csv_path` once, take an exclusive advisory lock, validate +/// or write the header, then append one newline-terminated row. +/// +/// The lock serializes cooperating `rocm bench load` processes so concurrent +/// first writers cannot both emit the header. `O_APPEND` keeps each row write +/// at the end of the file. +fn append_one_row(row: &BenchmarkRow, csv_path: &Path) -> Result<(), BenchLoadError> { + use std::io::{BufRead, Seek}; + + let mut file = OpenOptions::new() + .create(true) + .read(true) + .append(true) + .open(csv_path)?; + file.lock()?; + + if file.metadata()?.len() == 0 { + file.write_all(CSV_HEADER.as_bytes())?; + } else { + file.seek(std::io::SeekFrom::Start(0))?; + let mut first_line = String::new(); + std::io::BufReader::new(&file).read_line(&mut first_line)?; + if first_line.trim() != CSV_HEADER.trim() { + return Err(BenchLoadError::HeaderMismatch { + path: csv_path.display().to_string(), + }); + } + } + + let line = serialize_row_to_line(row)?; + file.write_all(&line)?; + Ok(()) +} + +/// Run a concurrency sweep and append one aggregate row per cell to `csv_path`. +/// +/// The header is written only when the file is new or empty. Each row is +/// serialized into a `Vec` ending in `\n` and written with a single +/// `write_all` call (O_APPEND safe on regular files). +/// +/// Returns the rows appended (one per concurrency level). +pub async fn run_and_append_csv( + spec: &LoadSpec, + concurrency_levels: &[u32], + csv_path: &Path, +) -> Result, BenchLoadError> { + let clients = BenchClients::new()?; + + let mut rows = Vec::with_capacity(concurrency_levels.len()); + for &conc in concurrency_levels { + let row = run_cell_with_clients(spec, conc, &clients).await?; + append_one_row(&row, csv_path)?; + rows.push(row); + } + + Ok(rows) +} + +/// Decide whether the auto-ramp should stop after `row`. +/// +/// Pure function — no I/O, no side effects — so it can be tested +/// deterministically with hand-built [`BenchmarkRow`] values. +/// +/// Returns `true` when any of the following hold: +/// - `is_last`: the hard cap (last element of [`RAMP_SEQUENCE`]) was reached, +/// - plateau: `prev_gen_tps` is `Some` AND `row.gen_tps` is `Some` AND +/// `gen_tps <= prev * (1.0 + PLATEAU_GAIN)`, +/// - saturation: both `max_running_reqs` and `max_waiting_reqs` are `Some` +/// AND `running > 0` AND `waiting >= running` (queue backed up). +/// The `running > 0` guard prevents a false positive when both fields are +/// observed at rest (zero) before any requests have reached the server. +pub fn should_stop_ramp(prev_gen_tps: Option, row: &BenchmarkRow, is_last: bool) -> bool { + if is_last { + return true; + } + + // Plateau: throughput stopped growing. + if let (Some(prev), Some(cur)) = (prev_gen_tps, row.gen_tps) + && cur <= prev * (1.0 + PLATEAU_GAIN) + { + return true; + } + + // Saturation: the queue is backed up — adding concurrency won't help. + // The `running > 0` guard prevents a false-stop when peaks are both zero + // (observed at rest before requests reach the engine). + if let (Some(running), Some(waiting)) = (row.max_running_reqs, row.max_waiting_reqs) + && running > 0 + && waiting >= running + { + return true; + } + + false +} + +fn next_prev_gen_tps(previous: Option, current: Option) -> Option { + current.or(previous) +} + +/// Run an automatic concurrency ramp over [`RAMP_SEQUENCE`], stopping early +/// when throughput saturates. +/// +/// Each cell is appended to `csv_path` immediately after completion so the +/// daemon tailer shows progress live. Stops after a cell when +/// [`should_stop_ramp`] returns `true`. +/// +/// Returns the rows appended (one per concurrency level run). +pub async fn run_auto_ramp( + spec: &LoadSpec, + csv_path: &Path, +) -> Result, BenchLoadError> { + let mut rows = Vec::new(); + let mut prev_gen_tps: Option = None; + let clients = BenchClients::new()?; + + for &conc in RAMP_SEQUENCE { + let row = run_cell_with_clients(spec, conc, &clients).await?; + append_one_row(&row, csv_path)?; + + let is_last = conc == *RAMP_SEQUENCE.last().unwrap_or(&conc); + let stop = should_stop_ramp(prev_gen_tps, &row, is_last); + prev_gen_tps = next_prev_gen_tps(prev_gen_tps, row.gen_tps); + rows.push(row); + + if stop { + break; + } + } + + Ok(rows) +} + +/// Serialize one `BenchmarkRow` to the 18-column CSV line (with trailing `\n`). +fn serialize_row_to_line(row: &BenchmarkRow) -> Result, BenchLoadError> { + let mut buf: Vec = Vec::new(); + { + let mut wtr = csv::WriterBuilder::new() + .has_headers(false) + .from_writer(&mut buf); + wtr.write_record([ + row.cell.as_str(), + &row.run.to_string(), + &opt_u32(row.concurrency), + opt_str(row.model.as_deref()), + opt_str(row.engine.as_deref()), + &opt_u32(row.input_len), + &opt_u32(row.output_len), + &opt_u32(row.n_requests), + &opt_u64(row.prompt_tokens), + &opt_u64(row.completion_tokens), + &opt_f64(row.prompt_tps), + &opt_f64(row.gen_tps), + &opt_f64(row.wall_s), + opt_str(row.launcher.as_deref()), + &opt_u32(row.max_running_reqs), + &opt_u32(row.max_waiting_reqs), + &opt_f64(row.ttft_ms), + &opt_f64(row.tpot_ms), + ])?; + wtr.flush()?; + } + // csv::Writer ends each record with \n already but we ensure it. + if buf.last() != Some(&b'\n') { + buf.push(b'\n'); + } + Ok(buf) +} + +fn opt_str(v: Option<&str>) -> &str { + v.unwrap_or("") +} + +fn opt_u32(v: Option) -> String { + v.map(|n| n.to_string()).unwrap_or_default() +} + +fn opt_u64(v: Option) -> String { + v.map(|n| n.to_string()).unwrap_or_default() +} + +fn opt_f64(v: Option) -> String { + v.map(|f| format!("{f:.6}")).unwrap_or_default() +} + +#[cfg(test)] +mod tests { + use rocm_dash_core::bench_rollup::{rollup_pass_n, row_verdict}; + use rocm_dash_core::bench_schema::PassFail; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use super::*; + use crate::bench_tail::CsvBenchTailer; + use rocm_dash_core::traits::BenchTailer; + + // ---------- helpers ---------- + + fn tempdir() -> std::path::PathBuf { + use std::sync::atomic::{AtomicU64, Ordering}; + static COUNTER: AtomicU64 = AtomicU64::new(0); + let mut p = std::env::temp_dir(); + let pid = std::process::id(); + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + p.push(format!("rocm-bench-load-{pid}-{n}")); + std::fs::create_dir_all(&p).unwrap(); + p + } + + fn stub_response(prompt_tokens: u64, completion_tokens: u64) -> ResponseTemplate { + ResponseTemplate::new(200).set_body_raw( + format!( + r#"{{"choices":[{{"message":{{"role":"assistant","content":"ok"}}}}], + "usage":{{"prompt_tokens":{prompt_tokens},"completion_tokens":{completion_tokens}}}}}"# + ), + "application/json", + ) + } + + fn make_spec(endpoint: &str) -> LoadSpec { + LoadSpec { + endpoint: endpoint.to_string(), + model: "test-model".to_string(), + input_len: 16, + output_len: 8, + requests: 4, + } + } + + // ---------- helpers: Prometheus stub body ---------- + + fn prom_body(running: u32, waiting: u32, ttft_sum: f64, ttft_count: f64) -> String { + format!( + "vllm:num_requests_running {running}\n\ + vllm:num_requests_waiting {waiting}\n\ + vllm:time_to_first_token_seconds_sum {ttft_sum}\n\ + vllm:time_to_first_token_seconds_count {ttft_count}\n\ + vllm:time_per_output_token_seconds_sum 0\n\ + vllm:time_per_output_token_seconds_count 0\n" + ) + } + + // ---------- T1: run_cell against a stub ---------- + + #[tokio::test] + async fn t1_run_cell_fields_and_tps() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chat/completions")) + .respond_with(stub_response(100, 50)) + .expect(4) + .mount(&server) + .await; + + let spec = make_spec(&server.uri()); + // requests=4, each returns prompt=100 completion=50 + let mut spec4 = spec.clone(); + spec4.requests = 4; + let row = run_cell(&spec4, 2).await.unwrap(); + + assert_eq!(row.cell, "bench-c2"); + assert_eq!(row.run, 1); + assert_eq!(row.concurrency, Some(2)); + assert_eq!(row.n_requests, Some(4)); + assert_eq!(row.completion_tokens, Some(200)); // 4 * 50 + assert_eq!(row.prompt_tokens, Some(400)); // 4 * 100 + // gen_tps divides by measured wall time — just check it's positive + assert!( + row.gen_tps.unwrap_or(0.0) > 0.0, + "gen_tps should be positive" + ); + assert!( + row.prompt_tps.unwrap_or(0.0) > 0.0, + "prompt_tps should be positive" + ); + assert_eq!( + row.launcher.as_deref(), + Some("rocm bench load (local smoke)") + ); + } + + #[tokio::test] + async fn run_cell_rejects_zero_concurrency() { + let spec = make_spec("http://127.0.0.1:1"); + let result = run_cell(&spec, 0).await; + assert!( + matches!(result, Err(BenchLoadError::InvalidConcurrency(0))), + "zero concurrency must fail before any network request: {result:?}" + ); + } + + // ---------- T2: concurrency cap ---------- + // + // wiremock's hyper handler calls respond() under an exclusive write-lock, + // so respond() is serial and cannot measure concurrent overlap. Instead we + // verify the semaphore via total elapsed time: + // + // With N=4, R=16 requests, and a per-response delay of D ms: + // - WITHOUT semaphore: all 16 fire simultaneously → wall ≈ D + // - WITH semaphore N: ceil(16/4)=4 serial batches → wall ≈ 4×D + // + // We assert wall_s > 1.5×D (conservative midpoint), which fails if the + // semaphore is absent because 1 batch × D < 1.5×D. We also assert + // wall_s < 8×D as a sanity upper bound so the test doesn't silently pass + // on a hung server. + + #[tokio::test] + async fn t2_concurrency_cap() { + const DELAY_MS: u64 = 30; + const N: u32 = 4; + const REQUESTS: u32 = 16; + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chat/completions")) + .respond_with( + stub_response(10, 5).set_delay(std::time::Duration::from_millis(DELAY_MS)), + ) + .expect(u64::from(REQUESTS)) + .mount(&server) + .await; + + let mut spec = make_spec(&server.uri()); + spec.requests = REQUESTS; + let row = run_cell(&spec, N).await.unwrap(); + + // Structural check: concurrency column matches N. + assert_eq!(row.concurrency, Some(N)); + assert_eq!( + row.n_requests, + Some(REQUESTS), + "all requests should succeed" + ); + + // Timing check: the semaphore batches requests so wall time is + // proportional to ceil(R/N), not to 1 batch. + let wall_s = row.wall_s.expect("wall_s must be set"); + let delay_s = DELAY_MS as f64 / 1000.0; + // Lower bound: at least 1.5 batches of delay (conservatively) + assert!( + wall_s >= delay_s * 1.5, + "wall_s={wall_s:.3}s < 1.5×delay={:.3}s — semaphore may not be limiting concurrency", + delay_s * 1.5 + ); + // Sanity upper bound: no more than 8 batches (catches hung servers) + assert!( + wall_s < delay_s * 8.0 * f64::from(REQUESTS) / f64::from(N), + "wall_s={wall_s:.3}s looks unreasonably large" + ); + } + + // ---------- T3: CSV round-trip ---------- + + #[tokio::test] + async fn t3_csv_round_trip() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chat/completions")) + .respond_with(stub_response(50, 25)) + .mount(&server) + .await; + + let dir = tempdir(); + let csv_path = dir.join("bench.csv"); + let mut spec = make_spec(&server.uri()); + spec.requests = 2; + + // Append sweep A (concurrency [1]) → drain should return 1 row. + run_and_append_csv(&spec, &[1], &csv_path).await.unwrap(); + let mut tailer = CsvBenchTailer::new(csv_path.clone()); + let rows_a = tailer.drain().unwrap(); + assert_eq!(rows_a.len(), 1, "drain A should return 1 row"); + assert_eq!(rows_a[0].cell, "bench-c1"); + // pass_fail defaults to Unknown (omitted columns default via #[serde(default)]). + assert_eq!(rows_a[0].pass_fail, PassFail::Unknown); + + // Second drain should be empty (no new rows). + let empty = tailer.drain().unwrap(); + assert!(empty.is_empty(), "second drain should be empty"); + + // Append sweep B (concurrency [8]) → drain should return only the new row. + run_and_append_csv(&spec, &[8], &csv_path).await.unwrap(); + let rows_b = tailer.drain().unwrap(); + assert_eq!(rows_b.len(), 1, "drain B should return 1 row"); + assert_eq!(rows_b[0].cell, "bench-c8"); + // pass_fail for a throughput-only row must be Unknown. + assert_eq!(rows_b[0].pass_fail, PassFail::Unknown); + + let _ = std::fs::remove_dir_all(dir); + } + + // ---------- D2: header-mismatch guard ---------- + + #[tokio::test] + async fn d2_header_mismatch_returns_error_without_modifying_file() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chat/completions")) + .respond_with(stub_response(50, 25)) + .mount(&server) + .await; + + let dir = tempdir(); + let csv_path = dir.join("external.csv"); + + // Write a file that starts with a bogus header (simulating an external + // agent-bench CSV with a different column layout). + let bogus_header = "col1,col2,col3\n"; + let original_content = format!("{bogus_header}row1,row2,row3\n"); + std::fs::write(&csv_path, &original_content).unwrap(); + + let spec = make_spec(&server.uri()); + let result = run_and_append_csv(&spec, &[1], &csv_path).await; + + // Must return the HeaderMismatch error. + assert!( + matches!(result, Err(BenchLoadError::HeaderMismatch { .. })), + "expected HeaderMismatch error, got: {result:?}" + ); + + // File must be unmodified. + let content_after = std::fs::read_to_string(&csv_path).unwrap(); + assert_eq!( + content_after, original_content, + "file must not be modified on header mismatch" + ); + + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn concurrent_first_appends_write_one_header() { + const WRITERS: usize = 8; + let dir = tempdir(); + let csv_path = dir.join("concurrent.csv"); + let barrier = Arc::new(std::sync::Barrier::new(WRITERS)); + + let handles: Vec<_> = (0..WRITERS) + .map(|run| { + let csv_path = csv_path.clone(); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + let row = BenchmarkRow { + cell: format!("bench-{run}"), + run: run as u32, + ..Default::default() + }; + barrier.wait(); + append_one_row(&row, &csv_path) + }) + }) + .collect(); + + for handle in handles { + handle.join().unwrap().unwrap(); + } + + let content = std::fs::read_to_string(&csv_path).unwrap(); + assert_eq!( + content + .lines() + .filter(|line| *line == CSV_HEADER.trim()) + .count(), + 1, + "concurrent creators must not duplicate the CSV header" + ); + assert_eq!(content.lines().count(), WRITERS + 1); + + let _ = std::fs::remove_dir_all(dir); + } + + // ---------- T4: Unknown-verdict guard ---------- + + #[test] + fn t4_unknown_verdict_does_not_count_as_pass() { + // A row with only throughput fields populated — quality all default → Unknown. + let row = BenchmarkRow { + cell: "bench-c1".to_string(), + run: 1, + gen_tps: Some(100.0), + concurrency: Some(1), + ..Default::default() + }; + + assert_eq!(row_verdict(&row), PassFail::Unknown); + + let rollup = rollup_pass_n(std::slice::from_ref(&row)); + assert_eq!(rollup.len(), 1); + assert_eq!( + rollup[0].n_passed, 0, + "Unknown verdict must not count as pass" + ); + } + + // ---------- T6: Prometheus poller + before/after → peaks + ttft_ms ---------- + // + // Architecture: peaks come from the mid-cell poller; ttft/tpot come from + // the before/after histogram delta. + // + // Stub layout: + // - First GET /metrics (up_to_n_times=1): before scrape → ttft_sum=10, count=100. + // The poller starts after the before scrape, so it never sees this response. + // - Catch-all GET /metrics: poller + after scrape → running=8, waiting=1, + // ttft_sum=20, count=200. + // + // Expected peaks (from poller): max_running=8, max_waiting=1. + // Expected ttft_ms = (20-10)/(200-100) * 1000 = 100ms. + + #[tokio::test] + async fn t6_prom_poller_populates_peaks_and_ttft() { + let server = MockServer::start().await; + + // /chat/completions returns token data. + Mock::given(method("POST")) + .and(path("/chat/completions")) + .respond_with(stub_response(100, 50)) + .expect(4) + .mount(&server) + .await; + + // Before scrape (first GET /metrics only) — used for ttft/tpot delta origin. + Mock::given(method("GET")) + .and(path("/metrics")) + .respond_with( + wiremock::ResponseTemplate::new(200).set_body_string(prom_body(5, 2, 10.0, 100.0)), + ) + .up_to_n_times(1) + .mount(&server) + .await; + // Catch-all — seen by the poller and the after scrape. + Mock::given(method("GET")) + .and(path("/metrics")) + .respond_with( + wiremock::ResponseTemplate::new(200).set_body_string(prom_body(8, 1, 20.0, 200.0)), + ) + .mount(&server) + .await; + + let mut spec = make_spec(&server.uri()); + spec.requests = 4; + let row = run_cell(&spec, 2).await.unwrap(); + + // Peaks come from the poller (which only sees the catch-all stub). + assert_eq!( + row.max_running_reqs, + Some(8), + "peak running should be 8 (poller)" + ); + assert_eq!( + row.max_waiting_reqs, + Some(1), + "peak waiting should be 1 (poller)" + ); + // ttft_ms is the histogram delta between before and after scrapes. + let ttft = row.ttft_ms.expect("ttft_ms should be Some"); + assert!( + (ttft - 100.0).abs() < 0.01, + "expected ttft_ms≈100 got {ttft}" + ); + // gen_tps must still be computed from client-side measurement. + assert!(row.gen_tps.unwrap_or(0.0) > 0.0, "gen_tps must be positive"); + } + + // ---------- T7: non-vLLM /metrics (404) → new fields None, gen_tps Some ---------- + + #[tokio::test] + async fn t7_non_vllm_metrics_404_new_fields_none_gen_tps_some() { + let server = MockServer::start().await; + + // Normal chat completions succeed. + Mock::given(method("POST")) + .and(path("/chat/completions")) + .respond_with(stub_response(100, 50)) + .mount(&server) + .await; + + // /metrics returns 404 (non-vLLM endpoint). + Mock::given(method("GET")) + .and(path("/metrics")) + .respond_with(wiremock::ResponseTemplate::new(404)) + .mount(&server) + .await; + + let mut spec = make_spec(&server.uri()); + spec.requests = 4; + let row = run_cell(&spec, 2).await.unwrap(); + + assert_eq!( + row.max_running_reqs, None, + "max_running_reqs should be None for 404 /metrics" + ); + assert_eq!( + row.max_waiting_reqs, None, + "max_waiting_reqs should be None for 404 /metrics" + ); + assert_eq!(row.ttft_ms, None, "ttft_ms should be None for 404 /metrics"); + assert_eq!(row.tpot_ms, None, "tpot_ms should be None for 404 /metrics"); + assert!( + row.gen_tps.unwrap_or(0.0) > 0.0, + "gen_tps must still be positive" + ); + } + + // ---------- T8: 18-col CSV round-trip via CsvBenchTailer ---------- + + #[tokio::test] + async fn t8_csv_round_trip_18_cols() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chat/completions")) + .respond_with(stub_response(50, 25)) + .mount(&server) + .await; + // /metrics returns 404 so new fields are None (simpler to assert). + Mock::given(method("GET")) + .and(path("/metrics")) + .respond_with(wiremock::ResponseTemplate::new(404)) + .mount(&server) + .await; + + let dir = tempdir(); + let csv_path = dir.join("bench18.csv"); + let mut spec = make_spec(&server.uri()); + spec.requests = 2; + + // Write one row. + run_and_append_csv(&spec, &[1], &csv_path).await.unwrap(); + + // Verify the header is 18 columns. + let content = std::fs::read_to_string(&csv_path).unwrap(); + let first_line = content.lines().next().expect("file should have a header"); + assert_eq!( + first_line.split(',').count(), + 18, + "header should have 18 columns" + ); + + // Drain via CsvBenchTailer — must deserialize without error. + let mut tailer = CsvBenchTailer::new(csv_path.clone()); + let rows = tailer.drain().unwrap(); + assert_eq!(rows.len(), 1, "should drain 1 row"); + assert_eq!(rows[0].cell, "bench-c1"); + assert_eq!(rows[0].pass_fail, PassFail::Unknown); + // New fields are None (404 /metrics path). + assert_eq!(rows[0].max_running_reqs, None); + assert_eq!(rows[0].ttft_ms, None); + + let _ = std::fs::remove_dir_all(dir); + } + + // ---------- T9: appending to a 14-col Phase-1 file returns HeaderMismatch ---------- + + #[tokio::test] + async fn t9_old_14col_file_returns_header_mismatch() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chat/completions")) + .respond_with(stub_response(50, 25)) + .mount(&server) + .await; + + let dir = tempdir(); + let csv_path = dir.join("phase1.csv"); + + // Write a file with the old 14-col header from Phase 1. + let old_header = "cell,run,concurrency,model,engine,input_len,output_len,\ + n_requests,prompt_tokens,completion_tokens,prompt_tps,gen_tps,wall_s,launcher\n"; + let original = format!( + "{old_header}bench-c1,1,1,m,,16,8,4,200,100,,,0.5,rocm bench load (local smoke)\n" + ); + std::fs::write(&csv_path, &original).unwrap(); + + let spec = make_spec(&server.uri()); + let result = run_and_append_csv(&spec, &[1], &csv_path).await; + + assert!( + matches!(result, Err(BenchLoadError::HeaderMismatch { .. })), + "expected HeaderMismatch for 14-col file, got: {result:?}" + ); + + // File must be unmodified. + let after = std::fs::read_to_string(&csv_path).unwrap(); + assert_eq!(after, original, "14-col file must not be modified"); + + let _ = std::fs::remove_dir_all(dir); + } + + // ---------- T10: auto-ramp plateau — flat gen_tps stops early ---------- + // + // All cells return identical token counts so gen_tps is flat. The plateau + // check fires on the second cell (cur <= prev * 1.05 because cur == prev), + // so the ramp stops at concurrency=2 and never reaches 128. + + #[tokio::test] + async fn t10_auto_ramp_plateau_stops_early() { + let server = MockServer::start().await; + // Same token counts for all requests → flat gen_tps. + Mock::given(method("POST")) + .and(path("/chat/completions")) + .respond_with(stub_response(50, 25)) + .mount(&server) + .await; + // /metrics: 404 so Prometheus fields are None. + Mock::given(method("GET")) + .and(path("/metrics")) + .respond_with(wiremock::ResponseTemplate::new(404)) + .mount(&server) + .await; + + let dir = tempdir(); + let csv_path = dir.join("auto_ramp_plateau.csv"); + let mut spec = make_spec(&server.uri()); + spec.requests = 2; + + let rows = run_auto_ramp(&spec, &csv_path).await.unwrap(); + + // Must have stopped before reaching concurrency=128 (the last element). + assert!( + rows.len() < RAMP_SEQUENCE.len(), + "plateau should have stopped early; got {} rows (RAMP len={})", + rows.len(), + RAMP_SEQUENCE.len() + ); + // Last concurrency must not be 128. + let last_conc = rows.last().and_then(|r| r.concurrency).unwrap_or(0); + assert_ne!(last_conc, 128, "should not have reached concurrency=128"); + // Must have appended at least the first cell. + assert!(!rows.is_empty(), "at least one row must be produced"); + + let _ = std::fs::remove_dir_all(dir); + } + + // ---------- T11: auto-ramp cap — rising gen_tps reaches 128 ---------- + // + // We use a response delay that grows with each call so earlier concurrency + // levels complete fewer tokens per second than later ones. The trick: use a + // wiremock `up_to_n_times` chain of stubs with decreasing delay so the mock + // server delivers progressively faster responses, making gen_tps rise + // monotonically and preventing the plateau check from firing until the last + // element (128) of RAMP_SEQUENCE is reached. + // + // Because accurate per-call timing in a test is fragile, we instead use a + // simpler approach: a single stub that always responds with the same tokens + // but with a very short delay, and set spec.requests = 1 so each cell has + // exactly 1 request. With 1 request per cell and flat token counts the gen_tps + // will be approximately 1/wall which varies by wall time — we can't guarantee + // monotonic growth. + // + // Instead, we use the queue-backed-up exit condition to test the cap: set + // max_waiting >= max_running via Prometheus. But that only works if Prom is up. + // + // Simplest approach: test the cap path directly by verifying that with a + // strictly rising gen_tps signal, the ramp runs all the way to the last + // RAMP_SEQUENCE entry (128). We simulate this by setting requests=1 and using + // a delay that decreases per-cell, ensuring each successive cell is faster. + // + // Since we can't easily make gen_tps strictly increase with a real HTTP mock + // (wall time is non-deterministic), we use a different angle: verify that when + // NO plateau and NO queue-full ever triggers, the ramp hits exactly 128. + // We achieve this by making each request take 0ms (no delay) — but with 1 + // request per cell the gen_tps may still vary. The reliable invariant is: + // the last row's concurrency == 128 when the stop condition never fires early. + // + // We enforce "no early stop" by using enough requests (spec.requests = 64) + // that each cell's gen_tps has a chance to grow (more concurrent = more TPS), + // and by checking the last concurrency rather than exact row count. + + #[test] + fn t11_auto_ramp_hard_cap_stops_at_128() { + let row = row_with_peaks(Some(1_000.0), None, None); + assert!(should_stop_ramp(Some(1.0), &row, true)); + assert_eq!(RAMP_SEQUENCE.last(), Some(&128)); + } + + // ---------- T12: should_stop_ramp — pure-function unit tests ---------- + + fn row_with_peaks( + gen_tps: Option, + max_running_reqs: Option, + max_waiting_reqs: Option, + ) -> BenchmarkRow { + BenchmarkRow { + cell: "bench-c1".to_string(), + run: 1, + gen_tps, + max_running_reqs, + max_waiting_reqs, + ..Default::default() + } + } + + #[test] + fn t12a_plateau_stops_ramp() { + // gen_tps same as prev → cur <= prev * 1.05 → stop. + let row = row_with_peaks(Some(100.0), None, None); + assert!( + should_stop_ramp(Some(100.0), &row, false), + "plateau should stop" + ); + } + + #[test] + fn t12b_rising_gen_tps_continues() { + // gen_tps grew by >5% → continue. + let row = row_with_peaks(Some(120.0), None, None); + assert!( + !should_stop_ramp(Some(100.0), &row, false), + "rising gen_tps should continue" + ); + } + + #[test] + fn t12c_is_last_stops() { + // Hard cap regardless of other fields. + let row = row_with_peaks(Some(200.0), None, None); + assert!( + should_stop_ramp(None, &row, true), + "is_last should always stop" + ); + } + + #[test] + fn t12d_saturation_running8_waiting8_stops() { + // waiting >= running AND running > 0 → saturated. + let row = row_with_peaks(None, Some(8), Some(8)); + assert!( + should_stop_ramp(None, &row, false), + "waiting>=running with running>0 should stop" + ); + } + + #[test] + fn t12e_at_rest_both_zero_does_not_stop() { + // Regression guard for the H1 fix: running=0, waiting=0 must NOT stop. + let row = row_with_peaks(None, Some(0), Some(0)); + assert!( + !should_stop_ramp(None, &row, false), + "running=0,waiting=0 must NOT stop (H1)" + ); + } + + #[test] + fn t12f_failed_cell_preserves_last_successful_throughput() { + assert_eq!(next_prev_gen_tps(Some(100.0), None), Some(100.0)); + assert_eq!(next_prev_gen_tps(Some(100.0), Some(120.0)), Some(120.0)); + } + + #[test] + fn t12f_both_none_peaks_does_not_stop() { + // Non-vLLM endpoint: both peaks are None; no saturation stop. + let row = row_with_peaks(None, None, None); + assert!( + !should_stop_ramp(None, &row, false), + "None peaks must not stop" + ); + } + #[test] + fn peak_pair_keeps_running_and_waiting_from_one_sample() { + let peak = AtomicU64::new(NO_SAMPLE_PAIR); + update_peak_pair(&peak, 2, 5); + update_peak_pair(&peak, 8, 1); + + assert_eq!(peak_pair(&peak), Some((8, 1))); + } +} diff --git a/crates/rocm-dash-collectors/src/bench_tail.rs b/crates/rocm-dash-collectors/src/bench_tail.rs index 7d1dd9e3..308a4427 100644 --- a/crates/rocm-dash-collectors/src/bench_tail.rs +++ b/crates/rocm-dash-collectors/src/bench_tail.rs @@ -33,7 +33,12 @@ impl BenchTailer for CsvBenchTailer { } fn drain(&mut self) -> Result> { - let file = File::open(&self.path)?; + let path = match std::fs::canonicalize(&self.path) { + Ok(path) => path, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => return Err(error.into()), + }; + let file = File::open(path)?; let mut rdr = csv::ReaderBuilder::new() .has_headers(true) .from_reader(BufReader::new(file)); @@ -145,6 +150,15 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + #[test] + fn absent_results_file_is_an_empty_drain() { + let dir = tempdir(); + let path = dir.join("not-created-yet.csv"); + let mut tailer = CsvBenchTailer::new(path); + + assert!(tailer.drain().unwrap().is_empty()); + } + fn tempdir() -> std::path::PathBuf { use std::sync::atomic::{AtomicU64, Ordering}; static COUNTER: AtomicU64 = AtomicU64::new(0); diff --git a/crates/rocm-dash-collectors/src/lib.rs b/crates/rocm-dash-collectors/src/lib.rs index f967a6fa..f5b7bcd8 100644 --- a/crates/rocm-dash-collectors/src/lib.rs +++ b/crates/rocm-dash-collectors/src/lib.rs @@ -8,6 +8,7 @@ //! Stubs return `CollectorError::Unsupported` so the daemon can start with nothing wired. pub mod amd_smi; +pub mod bench_load; pub mod bench_tail; pub mod cgroup; pub mod docker; diff --git a/crates/rocm-dash-core/src/config.rs b/crates/rocm-dash-core/src/config.rs index e9a86d80..c88bd68e 100644 --- a/crates/rocm-dash-core/src/config.rs +++ b/crates/rocm-dash-core/src/config.rs @@ -98,7 +98,8 @@ pub struct DaemonConfig { pub discovery_tick: Duration, #[serde(with = "duration_secs")] pub instance_tick: Duration, - /// Watch this directory for new normalized CSVs. + /// Tail this single CSV file for new normalized benchmark rows (despite + /// the field name, `CsvBenchTailer` opens it as one file, not a directory). pub bench_results_dir: Option, } @@ -179,6 +180,37 @@ fn socket_path( }) } +/// Default CSV file the standalone dashboard daemon tails for normalized +/// benchmark rows. The data-dir override wins; otherwise this legacy config +/// surface uses `$HOME/.rocm/bench/results.csv`. +/// +/// The unified `rocm` command and daemon share `rocm_core::AppPaths`, including +/// managed-root resolution. This standalone config does not read rocm-core's +/// JSON config, so managed-root users must set `ROCM_CLI_DATA_DIR` when launching +/// a standalone dashboard daemon. Normal `rocm daemon` usage is unaffected. +fn default_bench_results_dir() -> Option { + bench_results_dir_from_env( + std::env::var_os("ROCM_CLI_DATA_DIR"), + std::env::var_os("HOME"), + ) +} + +/// Resolve the standalone daemon's tailed bench CSV from explicit environment +/// inputs without mutating process-global variables in tests. +fn bench_results_dir_from_env( + data_dir_override: Option, + home: Option, +) -> Option { + data_dir_override + .filter(|v| !v.is_empty()) + .map(PathBuf::from) + .or_else(|| { + home.filter(|v| !v.is_empty()) + .map(|home| PathBuf::from(home).join(".rocm")) + }) + .map(|data_dir| data_dir.join("bench").join("results.csv")) +} + impl Default for DaemonConfig { fn default() -> Self { Self { @@ -187,7 +219,7 @@ impl Default for DaemonConfig { gpu_tick: Duration::from_secs(1), discovery_tick: Duration::from_secs(5), instance_tick: Duration::from_secs(2), - bench_results_dir: None, + bench_results_dir: default_bench_results_dir(), } } } @@ -484,4 +516,55 @@ theme = "default-dark" let path = socket_path(None, None, Some(String::new()), PathBuf::from("/tmp")); assert_eq!(path, PathBuf::from("/tmp/rocm-user/rocmdashd.sock")); } + + #[test] + fn bench_results_dir_from_env_uses_home_rocm_without_extra_data_segment() { + // EAI-7361: the Bench panel stayed permanently empty because this + // default was `None` — the daemon never tailed any file unless the + // user hand-edited a config file, even after running `rocm bench + // load`. Must resolve to the SAME absolute path `rocm bench load` + // writes to by default: `$HOME/.rocm/bench/results.csv`. Critically + // there is NO `/data/` segment — rocm-core's `default_data_dir()` is + // `$HOME/.rocm` itself, so an earlier `$HOME/.rocm/data/...` default + // would never line up with the producer. + let dir = bench_results_dir_from_env(None, Some("/home/alice".into())); + assert_eq!( + dir, + Some(PathBuf::from("/home/alice/.rocm/bench/results.csv")) + ); + } + + #[test] + fn bench_results_dir_from_env_honors_the_data_dir_override() { + // Mirrors `AppPaths::discover`: `ROCM_CLI_DATA_DIR` wins over `$HOME` + // so the standalone daemon tails exactly what a producer with the same + // override set writes. + let dir = bench_results_dir_from_env( + Some("/mnt/scratch/rocm".into()), + Some("/home/alice".into()), + ); + assert_eq!( + dir, + Some(PathBuf::from("/mnt/scratch/rocm/bench/results.csv")) + ); + } + + #[test] + fn bench_results_dir_from_env_is_none_when_unset_or_empty() { + assert_eq!(bench_results_dir_from_env(None, None), None); + assert_eq!( + bench_results_dir_from_env(Some(String::new().into()), Some(String::new().into())), + None + ); + } + + #[test] + fn daemon_config_default_populates_bench_results_dir_when_home_is_set() { + // Only meaningful where $HOME is actually set (true in this repo's CI + // and dev environments); skip rather than assert a specific value we + // don't control. + if std::env::var_os("HOME").is_some_and(|h| !h.is_empty()) { + assert!(DaemonConfig::default().bench_results_dir.is_some()); + } + } } diff --git a/crates/rocm-dash-daemon/src/lib.rs b/crates/rocm-dash-daemon/src/lib.rs index 62e80b54..8e2aefa8 100644 --- a/crates/rocm-dash-daemon/src/lib.rs +++ b/crates/rocm-dash-daemon/src/lib.rs @@ -20,5 +20,6 @@ pub mod transport; /// Daemon crate version, surfaced in the `Welcome` handshake. pub const DAEMON_VERSION: &str = env!("CARGO_PKG_VERSION"); +pub use rocm_dash_collectors::bench_load; pub use runner::RunnerOptions; pub use server::run; diff --git a/crates/rocm-dash-tui/src/app/mod.rs b/crates/rocm-dash-tui/src/app/mod.rs index 64cb4da9..512ff227 100644 --- a/crates/rocm-dash-tui/src/app/mod.rs +++ b/crates/rocm-dash-tui/src/app/mod.rs @@ -116,6 +116,11 @@ pub struct ResolvedArgs { /// Bin-injected tool-executor seam; None for demo/replay/mock — dash behaves /// as today. Stored here (Phase 2 plumbing); Phase 3 will use it. pub tool_executor: Option, + /// Daemon-tailed bench CSV path (`config.dashboard.daemon.bench_results_dir`). + /// + /// When `Some`, the bench-run form defaults `--out` to this path so appended + /// rows appear live in the bench tab. Adapted by the bin (owns `rocm-core`). + pub bench_results_dir: Option, } type Tui = Terminal>; @@ -567,6 +572,8 @@ pub struct AppState { pub command_screen: Option, /// Config & provider manager overlay (Phase 3 Wave 3). `None` = closed. pub config_manager: Option, + /// Bench-run form overlay. `None` = closed. + pub bench_run: Option, /// Built-in model recipes for the serve wizard's picker. Set from /// `ResolvedArgs` in the event loop; empty by default. pub model_recipes: Vec, @@ -579,6 +586,11 @@ pub struct AppState { /// Bin-injected tool-executor seam. Set from `ResolvedArgs` in the event /// loop; `None` for demo/replay/mock and by default. pub tool_executor: Option, + /// Daemon-tailed bench CSV path from the bin config. + /// + /// Forwarded from [`ResolvedArgs::bench_results_dir`] so the bench-run form + /// can default `--out` to the live-tailed file. `None` when not configured. + pub bench_results_dir: Option, /// Set by a `/quit` (or `/exit`) slash command; the event loop breaks on it. pub(crate) should_quit: bool, /// Edge: a pending executor-backed read-only slash command. Raised by @@ -674,10 +686,12 @@ impl AppState { automations_manager: None, command_screen: None, config_manager: None, + bench_run: None, model_recipes: Vec::new(), runtimes: Vec::new(), automations: Vec::new(), tool_executor: None, + bench_results_dir: None, should_quit: false, slash_tool: None, plan_request: None, @@ -703,6 +717,7 @@ impl AppState { self.automations_manager = None; self.command_screen = None; self.config_manager = None; + self.bench_run = None; self.approval = None; // A fresh overlay starts its console at the top. self.console_scroll = 0; @@ -832,6 +847,7 @@ impl AppState { || self.automations_manager.is_some() || self.command_screen.is_some() || self.config_manager.is_some() + || self.bench_run.is_some() } /// Focused-host exit gate: `true` when a `focus` is active AND its single @@ -909,6 +925,7 @@ impl AppState { .logs_view .as_ref() .is_none_or(|m| m.active_job.is_none()) + // bench_run is always at root when Some (no nested sub-popup or job). } /// Whether an `Esc` keypress should back out of an inline manager: true on @@ -1519,6 +1536,8 @@ async fn event_loop(terminal: &mut Tui, args: &ResolvedArgs) -> color_eyre::Resu // Tool-executor seam (Phase 2 plumbing), injected by the bin; None for // demo/replay/mock. Phase 3 will use it. state.tool_executor = args.tool_executor.clone(); + // Daemon-tailed bench CSV path for the bench-run form's default --out. + state.bench_results_dir = args.bench_results_dir.clone(); // Focused host: open exactly the overlay for the requested flow (Examine // also auto-runs its read-only job). `Focus::Setup` opens the onboarding // overlay — the same wizard `rocm bootstrap setup` routes to. @@ -1923,6 +1942,15 @@ async fn event_loop(terminal: &mut Tui, args: &ResolvedArgs) -> color_eyre::Resu ); crate::jobs::run_effects(fx, &job_tx); } + // Bench-run form, when open, owns all keys. + Some(Ok(CtEvent::Key(k))) if state.bench_run.is_some() => { + let fx = crate::ui::bench_run::on_key( + &mut state.bench_run, + &mut state.jobs, + k, + ); + crate::jobs::run_effects(fx, &job_tx); + } Some(Ok(CtEvent::Key(k))) => { let chat_ctx = ChatKeyCtx { focused: state.chat_focused, @@ -2392,6 +2420,13 @@ fn apply_action(state: &mut AppState, action: KeyAction) -> bool { state.close_overlays(); state.config_manager = Some(crate::ui::config_manager::ConfigManagerState::default()); } + KeyAction::OpenBenchRun => { + let bench_csv = state.bench_results_dir.clone(); + state.close_overlays(); + state.bench_run = Some(crate::ui::bench_run::BenchRunState::new( + bench_csv.as_deref(), + )); + } KeyAction::OpenThemePicker => state.open_theme_picker(), KeyAction::ApplyThemePick => state.apply_theme_pick(), KeyAction::OpenMenu => { @@ -2720,6 +2755,8 @@ pub enum KeyAction { OpenConfig, /// Open the logs overlay (Phase 3 Wave 3). OpenLogs, + /// Open the bench-run form overlay. + OpenBenchRun, } /// Whether a crossterm key event should be acted on. Terminals emit @@ -2937,6 +2974,8 @@ fn handle_key(k: KeyEvent, current: ActiveTab, modal: &Modal, chat: ChatKeyCtx) KeyCode::Char('i') if current == ActiveTab::Observe => KeyAction::OpenInstall, // Logs: browse recent ROCm CLI logs. KeyCode::Char('l') if current == ActiveTab::Observe => KeyAction::OpenLogs, + // Bench-run: launch a bench sweep from the TUI. + KeyCode::Char('b') if current == ActiveTab::Observe => KeyAction::OpenBenchRun, // Runtimes: list/activate/adopt/import ROCm runtimes. KeyCode::Char('r') if current == ActiveTab::Observe => KeyAction::OpenRuntimes, // Onboarding: first-run setup wizard (install / adopt). @@ -3621,6 +3660,41 @@ mod tests { assert!(s.command_screen.is_some() && s.automations_manager.is_none()); apply_action(&mut s, KeyAction::OpenConfig); assert!(s.config_manager.is_some() && s.command_screen.is_none()); + // T13: OpenBenchRun joins the mutual-exclusion set. + apply_action(&mut s, KeyAction::OpenBenchRun); + assert!(s.bench_run.is_some() && s.config_manager.is_none()); + } + + // ---------- T13: bench_run overlay invariants ---------- + + #[test] + fn t13_bench_run_in_has_open_overlay() { + let mut s = AppState::new("t".into(), "default-dark".into()); + assert!(!s.has_open_overlay(), "no overlay initially"); + s.bench_run = Some(crate::ui::bench_run::BenchRunState::new(None)); + assert!( + s.has_open_overlay(), + "bench_run must be in has_open_overlay" + ); + } + + #[test] + fn t13_bench_run_cleared_by_close_overlays() { + let mut s = AppState::new("t".into(), "default-dark".into()); + s.bench_run = Some(crate::ui::bench_run::BenchRunState::new(None)); + s.close_overlays(); + assert!(s.bench_run.is_none(), "close_overlays must clear bench_run"); + } + + #[test] + fn t13_bench_run_at_root() { + let mut s = AppState::new("t".into(), "default-dark".into()); + assert!(s.active_overlay_at_root(), "nothing open → at root"); + s.bench_run = Some(crate::ui::bench_run::BenchRunState::new(None)); + assert!( + s.active_overlay_at_root(), + "bench_run (no sub-popup/job) is always at root" + ); } #[test] @@ -4721,6 +4795,7 @@ mod tests { runtimes: Vec::new(), automations: Vec::new(), tool_executor: None, + bench_results_dir: None, } } diff --git a/crates/rocm-dash-tui/src/ui/bench_run.rs b/crates/rocm-dash-tui/src/ui/bench_run.rs new file mode 100644 index 00000000..d413efce --- /dev/null +++ b/crates/rocm-dash-tui/src/ui/bench_run.rs @@ -0,0 +1,568 @@ +// Copyright © Advanced Micro Devices, Inc., or its affiliates. +// +// SPDX-License-Identifier: MIT + +//! Bench-run form overlay. +//! +//! A small form that collects endpoint, model (optional), concurrency or +//! auto-ramp toggle, and output path, then emits a +//! [`SideEffect::SpawnJob`] for `rocm bench load`. Mirrors the +//! command-screen pattern: pure reducer functions, no I/O. + +use crossterm::event::{KeyCode, KeyEvent}; +use ratatui::Frame; +use ratatui::layout::{Constraint, Direction, Layout, Rect}; +use ratatui::style::{Modifier, Style}; +use ratatui::text::{Line, Span}; +use ratatui::widgets::Paragraph; + +use rocm_dash_core::state::{SideEffect, State, StateEvent}; + +use crate::ui::exec::resolve_exe; +use crate::ui::panel::{self, BoxRole}; +use crate::ui::theme::Theme; + +/// Which field of the form has focus. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +enum Field { + #[default] + Endpoint, + Model, + Concurrency, + Out, +} + +/// Form overlay state. +#[derive(Debug, Clone)] +pub struct BenchRunState { + pub endpoint: String, + pub model: String, + pub concurrency: String, + pub out: String, + /// When `true`, `--auto-ramp` is used instead of `--concurrency`. + pub auto_ramp: bool, + active_field: Field, + pub message: Option, + /// Hint shown in the Out field when it is empty: the daemon-tailed path or + /// a note that live updates need it configured. + default_out_hint: String, + /// When `Some`, this is the default --out path (daemon-tailed CSV). + tailed_path: Option, +} + +impl BenchRunState { + /// Create a new form. + /// + /// `tailed_csv` is the daemon's `bench_results_dir` (from config); when + /// `Some`, it is used as the default `--out` so rows appear live in the + /// bench tab. + pub fn new(tailed_csv: Option<&std::path::Path>) -> Self { + let (tailed_path, default_out_hint) = if let Some(p) = tailed_csv { + let s = p.display().to_string(); + ( + Some(s.clone()), + format!("{s} (daemon-tailed — live updates)"), + ) + } else { + ( + None, + "/bench/results.csv (shared CLI and daemon default)".to_string(), + ) + }; + Self { + endpoint: String::new(), + model: String::new(), + concurrency: "1,8,32,64".to_string(), + out: String::new(), + auto_ramp: false, + active_field: Field::Endpoint, + message: None, + default_out_hint, + tailed_path, + } + } +} + +/// Handle a key while the bench-run form is open. +/// +/// Returns side effects to pump (a `SpawnJob` on submit, empty otherwise). +pub fn on_key(br: &mut Option, jobs: &mut State, key: KeyEvent) -> Vec { + let Some(state) = br.as_mut() else { + return Vec::new(); + }; + + match key.code { + KeyCode::Esc => { + *br = None; + return Vec::new(); + } + KeyCode::Tab => { + state.active_field = match state.active_field { + Field::Endpoint => Field::Model, + Field::Model => Field::Concurrency, + Field::Concurrency => Field::Out, + Field::Out => Field::Endpoint, + }; + state.message = None; + return Vec::new(); + } + KeyCode::BackTab => { + state.active_field = match state.active_field { + Field::Endpoint => Field::Out, + Field::Model => Field::Endpoint, + Field::Concurrency => Field::Model, + Field::Out => Field::Concurrency, + }; + state.message = None; + return Vec::new(); + } + KeyCode::Char('t') + if key + .modifiers + .contains(crossterm::event::KeyModifiers::CONTROL) => + { + // Ctrl+T toggles auto-ramp (works from any field). + state.auto_ramp = !state.auto_ramp; + state.message = None; + return Vec::new(); + } + KeyCode::Enter => { + return try_submit(br, jobs); + } + KeyCode::Backspace => { + match state.active_field { + Field::Endpoint => state.endpoint.pop(), + Field::Model => state.model.pop(), + Field::Concurrency => state.concurrency.pop(), + Field::Out => state.out.pop(), + }; + state.message = None; + return Vec::new(); + } + KeyCode::Char(c) => { + match state.active_field { + Field::Endpoint => state.endpoint.push(c), + Field::Model => state.model.push(c), + Field::Concurrency => state.concurrency.push(c), + Field::Out => state.out.push(c), + } + state.message = None; + return Vec::new(); + } + _ => {} + } + Vec::new() +} + +/// Validate and submit the form as a `SpawnJob`. +fn try_submit(br: &mut Option, jobs: &mut State) -> Vec { + let state = br.as_mut().expect("called only when Some"); + + let endpoint = state.endpoint.trim().to_string(); + if endpoint.is_empty() { + state.message = Some("endpoint is required".to_string()); + return Vec::new(); + } + // Reject https:// — no TLS backend. + if endpoint.to_lowercase().starts_with("https://") { + state.message = + Some("https:// not supported (no TLS backend compiled in); use http://".to_string()); + return Vec::new(); + } + + let cmd = resolve_exe(); + let mut args = vec!["bench".to_string(), "load".to_string()]; + args.push("--endpoint".to_string()); + args.push(endpoint); + + let model = state.model.trim().to_string(); + if !model.is_empty() { + args.push("--model".to_string()); + args.push(model); + } + + if state.auto_ramp { + args.push("--auto-ramp".to_string()); + } else { + let conc = state.concurrency.trim().to_string(); + if !conc.is_empty() { + args.push("--concurrency".to_string()); + args.push(conc); + } + } + + // Resolve the output path: use the explicit Out field, else the tailed path, + // else let the CLI derive its shared `/bench/results.csv` default. + let out = state.out.trim().to_string(); + if !out.is_empty() { + args.push("--out".to_string()); + args.push(out); + } else if let Some(tailed) = state.tailed_path.clone() { + args.push("--out".to_string()); + args.push(tailed); + } + + let job_id = "bench-run".to_string(); + let fx = jobs.apply(StateEvent::StartJob { + id: job_id, + cmd, + args, + }); + if fx.is_empty() { + state.message = Some("a bench run is already running".to_string()); + return fx; + } + // Close the form only after spawning (the job console is shown by the job-bridge). + *br = None; + fx +} + +/// Render the bench-run form. +pub fn draw_bench_run(f: &mut Frame, area: Rect, state: &BenchRunState, theme: &Theme) { + let inner = panel::bento( + f, + area, + Some("Run a bench sweep"), + BoxRole::Primary, + false, + theme, + ); + if inner.height == 0 { + return; + } + + let rows = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Length(1), // title / hint + Constraint::Length(1), // endpoint + Constraint::Length(1), // model + Constraint::Length(1), // concurrency / auto-ramp toggle + Constraint::Length(1), // out + Constraint::Length(1), // message + Constraint::Min(1), // spacer + Constraint::Length(1), // footer + ]) + .split(inner); + + f.render_widget( + Paragraph::new(Line::from(Span::styled( + "Tab next field · Ctrl+T toggle auto-ramp · Enter run · Esc close", + Style::default().fg(theme.muted), + ))), + rows[0], + ); + + render_field( + f, + rows[1], + "endpoint", + &state.endpoint, + "(http://127.0.0.1:8000)", + state.active_field == Field::Endpoint, + theme, + ); + render_field( + f, + rows[2], + "model ", + &state.model, + "(optional — auto-detected from /v1/models)", + state.active_field == Field::Model, + theme, + ); + + // Concurrency row shows auto-ramp status. + let (conc_label, conc_value, conc_hint) = if state.auto_ramp { + ( + "auto-ramp", + "[ON]", + "1,2,4,8,16,32,64,128 — stops at saturation", + ) + } else { + ( + "concurrency", + state.concurrency.as_str(), + "comma-separated, e.g. 1,8,32,64", + ) + }; + let conc_focused = state.active_field == Field::Concurrency; + let conc_style = if conc_focused { + Style::default() + .fg(theme.accent) + .add_modifier(Modifier::BOLD) + } else { + Style::default().fg(theme.fg) + }; + f.render_widget( + Paragraph::new(Line::from(vec![ + Span::styled( + format!("{conc_label:<12} "), + Style::default().fg(theme.muted), + ), + Span::styled(conc_value, conc_style), + Span::styled(format!(" {conc_hint}"), Style::default().fg(theme.muted)), + ])), + rows[3], + ); + + render_field( + f, + rows[4], + "out ", + &state.out, + &state.default_out_hint, + state.active_field == Field::Out, + theme, + ); + + let msg = state.message.as_deref().unwrap_or(""); + f.render_widget( + Paragraph::new(Line::from(Span::styled( + msg, + Style::default().fg(theme.err), + ))), + rows[5], + ); + + f.render_widget( + Paragraph::new(Line::from(Span::styled( + "note: local saturation smoke-test — client-measured throughput, not an official ROCm/AMD benchmark.", + Style::default().fg(theme.muted), + ))), + rows[7], + ); +} + +fn render_field( + f: &mut Frame, + area: Rect, + label: &str, + value: &str, + hint: &str, + focused: bool, + theme: &Theme, +) { + let shown = if value.is_empty() { + hint.to_string() + } else { + value.to_string() + }; + let value_style = if focused { + Style::default() + .fg(theme.accent) + .add_modifier(Modifier::BOLD) + } else if value.is_empty() { + Style::default().fg(theme.muted) + } else { + Style::default().fg(theme.fg) + }; + f.render_widget( + Paragraph::new(Line::from(vec![ + Span::styled(format!("{label:<12} "), Style::default().fg(theme.muted)), + Span::styled(shown, value_style), + ])), + area, + ); +} + +#[cfg(test)] +mod tests { + use super::*; + use crossterm::event::{KeyCode, KeyEvent, KeyModifiers}; + + fn key(c: KeyCode) -> KeyEvent { + KeyEvent::new(c, KeyModifiers::NONE) + } + + fn type_str(br: &mut Option, jobs: &mut State, s: &str) { + for ch in s.chars() { + on_key(br, jobs, key(KeyCode::Char(ch))); + } + } + + /// Build a form pre-filled with a valid endpoint. + fn with_endpoint(endpoint: &str) -> Option { + let mut s = BenchRunState::new(None); + s.endpoint = endpoint.to_string(); + Some(s) + } + + // ---------- T12: submit → exactly one SpawnJob with correct args ---------- + + #[test] + fn t12_submit_emits_spawn_job_with_correct_args() { + // Verify https:// is rejected pre-spawn. + let mut br = with_endpoint("https://example.com"); + let mut jobs = State::default(); + let fx = on_key(&mut br, &mut jobs, key(KeyCode::Enter)); + assert!( + fx.is_empty(), + "https:// must be rejected before spawning, got: {fx:?}" + ); + assert!( + br.as_ref().unwrap().message.is_some(), + "error message must be set on https:// rejection" + ); + + // Valid http:// endpoint → spawns one job with the right args. + let mut br = with_endpoint("http://127.0.0.1:8000"); + let mut jobs = State::default(); + // Tab to model, type a model name. + on_key(&mut br, &mut jobs, key(KeyCode::Tab)); + type_str(&mut br, &mut jobs, "my-model"); + // Tab to concurrency, leave default "1,8,32,64". + on_key(&mut br, &mut jobs, key(KeyCode::Tab)); + // Tab to out, leave empty. + on_key(&mut br, &mut jobs, key(KeyCode::Tab)); + // Back to endpoint field; focus is now Out — press Enter. + let fx = on_key(&mut br, &mut jobs, key(KeyCode::Enter)); + assert_eq!(fx.len(), 1, "exactly one SideEffect must be emitted"); + + match &fx[0] { + SideEffect::SpawnJob { cmd: _, args, .. } => { + assert!( + args.contains(&"bench".to_string()), + "args must include 'bench'" + ); + assert!( + args.contains(&"load".to_string()), + "args must include 'load'" + ); + assert!( + args.contains(&"--endpoint".to_string()), + "args must include '--endpoint'" + ); + assert!( + args.contains(&"http://127.0.0.1:8000".to_string()), + "args must include the endpoint URL" + ); + assert!( + args.contains(&"--model".to_string()), + "args must include '--model'" + ); + assert!( + args.contains(&"my-model".to_string()), + "args must include the model name" + ); + // auto-ramp is off by default — --concurrency should be present. + assert!( + args.contains(&"--concurrency".to_string()), + "args must include '--concurrency' when auto-ramp is off" + ); + assert!( + !args.contains(&"--auto-ramp".to_string()), + "args must NOT include '--auto-ramp' when toggle is off" + ); + } + other => panic!("expected SpawnJob, got {other:?}"), + } + // Form must be closed after submit. + assert!(br.is_none(), "form must close after successful submit"); + } + + #[test] + fn t12_auto_ramp_toggle_replaces_concurrency() { + let mut br = with_endpoint("http://127.0.0.1:8000"); + let mut jobs = State::default(); + // Toggle auto-ramp with Ctrl+T. + on_key( + &mut br, + &mut jobs, + KeyEvent::new(KeyCode::Char('t'), KeyModifiers::CONTROL), + ); + assert!( + br.as_ref().unwrap().auto_ramp, + "auto_ramp should be true after Ctrl+T" + ); + + let fx = on_key(&mut br, &mut jobs, key(KeyCode::Enter)); + assert_eq!(fx.len(), 1, "exactly one SideEffect"); + match &fx[0] { + SideEffect::SpawnJob { args, .. } => { + assert!( + args.contains(&"--auto-ramp".to_string()), + "args must include '--auto-ramp' when toggle is on" + ); + assert!( + !args.contains(&"--concurrency".to_string()), + "args must NOT include '--concurrency' when auto-ramp is on" + ); + } + other => panic!("expected SpawnJob, got {other:?}"), + } + } + + #[test] + fn t12_tailed_path_used_as_default_out() { + use std::path::Path; + let tailed = Path::new("/var/rocm/bench/results.csv"); + let mut br = Some(BenchRunState::new(Some(tailed))); + br.as_mut().unwrap().endpoint = "http://127.0.0.1:8000".to_string(); + let mut jobs = State::default(); + let fx = on_key(&mut br, &mut jobs, key(KeyCode::Enter)); + assert_eq!(fx.len(), 1); + match &fx[0] { + SideEffect::SpawnJob { args, .. } => { + let out_idx = args.iter().position(|a| a == "--out"); + assert!(out_idx.is_some(), "--out must be injected for tailed path"); + let out_val = &args[out_idx.unwrap() + 1]; + assert_eq!( + out_val, "/var/rocm/bench/results.csv", + "tailed path must be used as --out default" + ); + } + other => panic!("expected SpawnJob, got {other:?}"), + } + } + + #[test] + fn duplicate_submit_keeps_form_open_with_message() { + let mut jobs = State::default(); + let mut first = with_endpoint("http://127.0.0.1:8000"); + assert_eq!(on_key(&mut first, &mut jobs, key(KeyCode::Enter)).len(), 1); + + let mut duplicate = with_endpoint("http://127.0.0.1:8000"); + let fx = on_key(&mut duplicate, &mut jobs, key(KeyCode::Enter)); + + assert!(fx.is_empty()); + let state = duplicate.expect("deduplicated submission must keep form open"); + assert!( + state + .message + .as_deref() + .unwrap_or("") + .contains("already running") + ); + } + + #[test] + fn empty_out_hint_describes_shared_default_results_file() { + let state = BenchRunState::new(None); + assert!(state.default_out_hint.contains("bench/results.csv")); + assert!(!state.default_out_hint.contains("")); + } + + #[test] + fn t13_esc_closes_bench_run() { + let mut br = Some(BenchRunState::new(None)); + let mut jobs = State::default(); + let fx = on_key(&mut br, &mut jobs, key(KeyCode::Esc)); + assert!(fx.is_empty()); + assert!(br.is_none(), "Esc must close the form"); + } + + #[test] + fn t13_empty_endpoint_rejected() { + let mut br = Some(BenchRunState::new(None)); + let mut jobs = State::default(); + // endpoint is empty by default. + let fx = on_key(&mut br, &mut jobs, key(KeyCode::Enter)); + assert!(fx.is_empty(), "empty endpoint must be rejected"); + assert!(br.is_some(), "form stays open on validation error"); + assert!( + br.as_ref().unwrap().message.is_some(), + "error message must be set" + ); + } +} diff --git a/crates/rocm-dash-tui/src/ui/mod.rs b/crates/rocm-dash-tui/src/ui/mod.rs index 71c782f2..97990606 100644 --- a/crates/rocm-dash-tui/src/ui/mod.rs +++ b/crates/rocm-dash-tui/src/ui/mod.rs @@ -5,6 +5,7 @@ pub mod approval; pub mod automations_manager; pub mod bench; +pub mod bench_run; pub mod command_screen; pub mod config_manager; pub mod core_bars; @@ -264,6 +265,8 @@ fn draw_active_manager(f: &mut Frame, rect: Rect, state: &AppState, theme: &Them command_screen::draw_command_screen(f, rect, c, &state.jobs, theme); } else if let Some(cm) = &state.config_manager { config_manager::draw_config_manager(f, rect, cm, &state.jobs, theme); + } else if let Some(br) = &state.bench_run { + bench_run::draw_bench_run(f, rect, br, theme); } } @@ -426,6 +429,8 @@ fn draw_footer(f: &mut Frame, area: Rect, state: &AppState, theme: &Theme) -> Ve segs.push(Seg::Sep(" logs ")); segs.push(Seg::Key("s", Some(KeyAction::OpenServices))); segs.push(Seg::Sep(" services ")); + segs.push(Seg::Key("b", Some(KeyAction::OpenBenchRun))); + segs.push(Seg::Sep(" bench ")); } if state.replay.is_some() { segs.push(Seg::Key("Space", Some(KeyAction::ReplayTogglePause))); diff --git a/crates/rocm-dash-tui/src/ui/tabs/bench.rs b/crates/rocm-dash-tui/src/ui/tabs/bench.rs index a6fa6377..8e60a1d5 100644 --- a/crates/rocm-dash-tui/src/ui/tabs/bench.rs +++ b/crates/rocm-dash-tui/src/ui/tabs/bench.rs @@ -25,7 +25,7 @@ pub fn draw(f: &mut Frame, area: Rect, state: &AppState, theme: &Theme) { if state.bench_rows.is_empty() { let inner = panel::bento(f, area, Some("Bench"), BoxRole::Neutral, false, theme); let p = Paragraph::new(Line::from(Span::styled( - "no rows · start daemon with --bench-csv ", + "no rows · run `rocm bench load --endpoint ` to populate the daemon-tailed bench results file · press b to run a sweep", Style::default().fg(theme.muted), ))); f.render_widget(p, inner); @@ -601,6 +601,8 @@ fn build_detail_lines(row: &BenchmarkRow, theme: &Theme) -> Vec> { fmt_opt_u32_si(row.max_waiting_reqs), theme, )); + lines.push(kv_line("ttft_ms", fmt_opt(&row.ttft_ms), theme)); + lines.push(kv_line("tpot_ms", fmt_opt(&row.tpot_ms), theme)); lines.push(kv_line("out_chars", fmt_opt_u64_si(row.out_chars), theme)); lines.push(Line::raw("")); diff --git a/crates/rocm-dash-tui/src/ui/tabs/instances.rs b/crates/rocm-dash-tui/src/ui/tabs/instances.rs index 5e375ee7..3afce901 100644 --- a/crates/rocm-dash-tui/src/ui/tabs/instances.rs +++ b/crates/rocm-dash-tui/src/ui/tabs/instances.rs @@ -911,10 +911,12 @@ mod tests { automations_manager: None, command_screen: None, config_manager: None, + bench_run: None, model_recipes: Vec::new(), runtimes: Vec::new(), automations: Vec::new(), tool_executor: None, + bench_results_dir: None, should_quit: false, slash_tool: None, plan_request: None,