From c4f3a21c3476a11f00e6ed0828d2038b5130b055 Mon Sep 17 00:00:00 2001 From: rob thijssen Date: Thu, 9 Jul 2026 17:57:00 +0300 Subject: [PATCH] feat(bench): concurrency-sweep knee report + /api/concurrency (#137 T3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Turns the #89 concurrency-burst data into a capacity verdict. Adds Store::concurrency() — per (target, model) at the latest build, the throughput / p95-TTFT / queue-wait / reject-rate curve across burst widths, plus the derived knee: the highest concurrency the model sustains before it sheds load or its p95 TTFT tail stretches past 2× the lightest-load baseline. This is the data-backed justification for a max_in_flight ceiling. Surfaced three ways: - `helexa-bench report --concurrency` (md/json) - GET /api/concurrency — the bench UI's live-capacity view - example config enables the sweep [1,2,4,8,16] to bracket beast's max_in_flight=8 so the knee is visible Tests: the knee lands on the last sustainable level; None when even the lightest level sheds. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_012GaW6F2j3zsQ6yABKPsXXe --- crates/helexa-bench/src/api.rs | 11 +++ crates/helexa-bench/src/main.rs | 12 +++ crates/helexa-bench/src/report.rs | 54 ++++++++++- crates/helexa-bench/src/store.rs | 152 ++++++++++++++++++++++++++++++ helexa-bench.example.toml | 11 ++- 5 files changed, 235 insertions(+), 5 deletions(-) diff --git a/crates/helexa-bench/src/api.rs b/crates/helexa-bench/src/api.rs index 14d9bc0a..3fc92a3d 100644 --- a/crates/helexa-bench/src/api.rs +++ b/crates/helexa-bench/src/api.rs @@ -36,6 +36,7 @@ pub fn api_routes(state: ApiState) -> Router { .route("/api/dimensions", get(dimensions)) .route("/api/summary", get(summary)) .route("/api/scaling", get(scaling)) + .route("/api/concurrency", get(concurrency)) .route("/api/swap", get(swap)) .route("/api/capability", get(capability)) .route("/api/series", get(series)) @@ -85,6 +86,16 @@ async fn scaling( store.scaling().map(Json).map_err(err500) } +/// Concurrency sweep per (target, model) — throughput / p95 TTFT / reject-rate +/// across burst widths, plus the sustainable-concurrency knee (#137). The UI's +/// live-capacity view. +async fn concurrency( + State(s): State, +) -> Result>, ApiError> { + let store = s.lock().await; + store.concurrency().map(Json).map_err(err500) +} + /// Cold-load / model-swap costs per (target, model) — reload latency + cold /// first-request (#90). async fn swap(State(s): State) -> Result>, ApiError> { diff --git a/crates/helexa-bench/src/main.rs b/crates/helexa-bench/src/main.rs index d6529c41..e447d28c 100644 --- a/crates/helexa-bench/src/main.rs +++ b/crates/helexa-bench/src/main.rs @@ -94,6 +94,11 @@ enum Command { /// scores, with per-model median. #[arg(long)] capability: bool, + /// Render the concurrency-sweep view (#137): throughput / p95 TTFT / + /// reject-rate across burst widths per model, with the sustainable- + /// concurrency knee. + #[arg(long)] + concurrency: bool, }, } @@ -205,6 +210,7 @@ async fn run(cli: Cli) -> Result<()> { scaling, swap, capability, + concurrency, } => { let db_path = match db { Some(p) => p, @@ -217,6 +223,12 @@ async fn run(cli: Cli) -> Result<()> { Format::Md => report::render_capability_markdown(&runs), Format::Json => report::render_capability_json(&runs)?, } + } else if concurrency { + let curves = store.concurrency()?; + match format { + Format::Md => report::render_concurrency_markdown(&curves), + Format::Json => report::render_concurrency_json(&curves)?, + } } else if swap { let costs = store.swap_costs()?; match format { diff --git a/crates/helexa-bench/src/report.rs b/crates/helexa-bench/src/report.rs index 7e94bf42..80d07409 100644 --- a/crates/helexa-bench/src/report.rs +++ b/crates/helexa-bench/src/report.rs @@ -3,7 +3,7 @@ //! doc: engine, model, prompt tok, TTFT (s), decode tok/s, total (s), //! plus the build SHA each cell was measured against. -use crate::store::{CapabilityRun, ReportRow, ScalingCurve, SwapCost}; +use crate::store::{CapabilityRun, ConcurrencyCurve, ReportRow, ScalingCurve, SwapCost}; use anyhow::Result; pub fn render_markdown(rows: &[ReportRow]) -> String { @@ -136,6 +136,58 @@ pub fn render_scaling_json(curves: &[ScalingCurve]) -> Result { Ok(serde_json::to_string_pretty(curves)?) } +/// Concurrency-sweep view (#137): one block per (target, model) with the +/// throughput / latency-tail / shedding curve across burst widths, then the +/// knee — the max sustainable concurrency, the data-backed `max_in_flight`. +pub fn render_concurrency_markdown(curves: &[ConcurrencyCurve]) -> String { + let mut out = String::new(); + for c in curves { + let gpu = c.gpu.as_deref().unwrap_or(""); + out.push_str(&format!( + "### {} · {} (`{}`{})\n\n", + c.target_name, + c.model_id, + c.git_sha, + if gpu.is_empty() { + String::new() + } else { + format!(", {gpu}") + }, + )); + out.push_str("| N | decode tok/s | p95 TTFT (s) | queue wait (ms) | reject % | n |\n"); + out.push_str("|---:|---:|---:|---:|---:|---:|\n"); + for p in &c.points { + let reject = p + .reject_rate + .map(|r| format!("{:.0}%", r * 100.0)) + .unwrap_or_else(|| "—".into()); + out.push_str(&format!( + "| {} | {} | {} | {} | {} | {} |\n", + p.concurrency, + fmt_opt(p.decode_tps, 1), + fmt_opt(p.ttft_p95_s, 2), + fmt_opt(p.queue_wait_ms, 0), + reject, + p.samples, + )); + } + match c.knee_concurrency { + Some(k) => out.push_str(&format!( + "\nmax sustainable concurrency: **{k}** \ + (no shedding, p95 TTFT within 2× of the lightest-load baseline)\n\n", + )), + None => out.push_str( + "\nmax sustainable concurrency: — (sheds or breaks even at the lightest level)\n\n", + ), + } + } + out +} + +pub fn render_concurrency_json(curves: &[ConcurrencyCurve]) -> Result { + Ok(serde_json::to_string_pretty(curves)?) +} + /// Cold-load / model-swap cost view (#90): reload latency + cold /// first-request per model. pub fn render_swap_markdown(costs: &[SwapCost]) -> String { diff --git a/crates/helexa-bench/src/store.rs b/crates/helexa-bench/src/store.rs index 65946190..668d9326 100644 --- a/crates/helexa-bench/src/store.rs +++ b/crates/helexa-bench/src/store.rs @@ -431,6 +431,58 @@ impl Store { Ok(out) } + /// Concurrency sweep (#137): per (target, model) at the latest build, the + /// aggregate serving behaviour across `concurrency:` burst levels, plus + /// the derived knee — the max sustainable concurrency before the model + /// sheds load or its p95 TTFT tail breaks. The data-backed justification + /// for a `max_in_flight` ceiling. + pub fn concurrency(&self) -> Result> { + use std::collections::BTreeMap; + // Reuse the aggregated report cells; the concurrency: rows are the + // per-burst-width measurement points. + let mut by_model: BTreeMap<(String, String), Vec> = BTreeMap::new(); + for r in self.report_rows()? { + if r.scenario_id.starts_with("concurrency:") { + by_model + .entry((r.target_name.clone(), r.model_id.clone())) + .or_default() + .push(r); + } + } + let mut out = Vec::new(); + for ((target_name, model_id), mut rows) in by_model { + rows.sort_by_key(|r| r.concurrency.unwrap_or(0)); + let points: Vec = rows + .iter() + .map(|r| { + let n = r.concurrency.unwrap_or(0); + let reject_rate = match r.rejected_median { + Some(rej) if n > 0 => Some(rej / n as f64), + _ => None, + }; + ConcurrencyPoint { + concurrency: n as u32, + decode_tps: r.decode_tps_median, + ttft_p95_s: r.ttft_p95_load_s, + queue_wait_ms: r.queue_wait_ms_median, + reject_rate, + samples: r.samples, + } + }) + .collect(); + let knee_concurrency = concurrency_knee(&points); + out.push(ConcurrencyCurve { + target_name, + model_id, + git_sha: rows.first().map(|r| r.git_sha.clone()).unwrap_or_default(), + gpu: rows.iter().find_map(|r| r.gpu.clone()), + points, + knee_concurrency, + }); + } + Ok(out) + } + /// Cold-load / model-swap costs (#90): per (target, model) at the latest /// build, the median unload→reload latency and the cold first-request /// latency after reload (the `scenario_id = "swap"` rows). @@ -1007,6 +1059,68 @@ pub struct ScalingPoint { pub samples: usize, } +/// One point on a concurrency sweep for a (target, model) at its latest +/// build (#137): the aggregate serving behaviour at `concurrency` N +/// simultaneous streams. +#[derive(Debug, Clone, serde::Serialize)] +pub struct ConcurrencyPoint { + /// Burst width — the number of simultaneous streams. + pub concurrency: u32, + /// Aggregate node throughput (total tokens / burst window), tok/s. + pub decode_tps: Option, + /// Within-burst p95 time-to-first-token — where saturation first bites. + pub ttft_p95_s: Option, + /// Median admission queue wait, ms. + pub queue_wait_ms: Option, + /// Fraction of the burst shed by admission (`rejected / concurrency`). + /// `> 0` means the model is at capacity for this N. + pub reject_rate: Option, + pub samples: usize, +} + +/// A concurrency sweep for a (target, model) at its latest build (#137): +/// the points ordered by burst width, plus the derived knee — the highest +/// concurrency the model sustains before it sheds load or its latency tail +/// breaks. This is the data-backed justification for a `max_in_flight` +/// ceiling. +#[derive(Debug, Clone, serde::Serialize)] +pub struct ConcurrencyCurve { + pub target_name: String, + pub model_id: String, + pub git_sha: String, + pub gpu: Option, + pub points: Vec, + /// Max sustainable concurrency: the largest level that shed nothing and + /// held p95 TTFT within [`KNEE_TTFT_FACTOR`] of the lightest-load + /// baseline. `None` when even the smallest level sheds or lacks a TTFT. + pub knee_concurrency: Option, +} + +/// How far the p95 TTFT tail may stretch past the lightest-load baseline +/// before a concurrency level counts as "broken" for the knee (#137). +const KNEE_TTFT_FACTOR: f64 = 2.0; + +/// Highest sustainable concurrency from an ascending-ordered sweep: the +/// largest level that shed nothing and held p95 TTFT within +/// [`KNEE_TTFT_FACTOR`] of the lightest-load baseline. Degradation is +/// monotonic in N, so the scan stops at the first level that breaks. +fn concurrency_knee(points: &[ConcurrencyPoint]) -> Option { + let baseline = points.iter().find_map(|p| p.ttft_p95_s)?; + let mut knee = None; + for p in points { + let shed = p.reject_rate.unwrap_or(0.0) > 0.0; + let within_tail = p + .ttft_p95_s + .map(|t| t <= baseline * KNEE_TTFT_FACTOR) + .unwrap_or(false); + if shed || !within_tail { + break; + } + knee = Some(p.concurrency); + } + knee +} + /// Group by (target, model, scenario), keep only the latest SHA's rows /// (latest = the SHA of the last-inserted row, since input is id-ordered), /// and median each metric. @@ -1313,6 +1427,44 @@ mod tests { assert_eq!(row.ttft_p95_load_s, Some(0.9)); } + #[test] + fn concurrency_sweep_finds_the_knee() { + let s = Store::open_in_memory().unwrap(); + // A sweep where 1,2,4 hold but 8 sheds and blows the p95 tail. + for (n, ttft, rej) in [(1u32, 0.2, 0u32), (2, 0.25, 0), (4, 0.35, 0), (8, 0.9, 2)] { + let mut r = rec("beast", "sha", "m", &format!("concurrency:{n}"), true); + r.concurrency = Some(n); + r.ttft_p95_s = Some(ttft); + r.rejected = Some(rej); + s.insert_run(&r).unwrap(); + } + let curves = s.concurrency().unwrap(); + assert_eq!(curves.len(), 1); + let c = &curves[0]; + // Ordered ascending by burst width. + assert_eq!( + c.points.iter().map(|p| p.concurrency).collect::>(), + vec![1, 2, 4, 8] + ); + // N=8 shed 2 of 8 → reject_rate 0.25. + assert_eq!(c.points[3].reject_rate, Some(0.25)); + // Knee = 4: highest level with no shedding and p95 TTFT within 2× of + // the lightest-load baseline (0.2 → 0.4; 0.35 holds, 0.9 breaks). + assert_eq!(c.knee_concurrency, Some(4)); + } + + #[test] + fn concurrency_knee_is_none_when_lightest_level_sheds() { + let s = Store::open_in_memory().unwrap(); + let mut r = rec("beast", "sha", "m", "concurrency:2", true); + r.concurrency = Some(2); + r.ttft_p95_s = Some(0.3); + r.rejected = Some(1); + s.insert_run(&r).unwrap(); + let curves = s.concurrency().unwrap(); + assert_eq!(curves[0].knee_concurrency, None); + } + #[test] fn capability_runs_store_artifact_and_accept_scores() { let s = Store::open_in_memory().unwrap(); diff --git a/helexa-bench.example.toml b/helexa-bench.example.toml index 542570e4..a7f4040d 100644 --- a/helexa-bench.example.toml +++ b/helexa-bench.example.toml @@ -32,10 +32,13 @@ prompt_sizes = [128, 4096] max_tokens = 256 # Concurrency / agentic-load scenarios (#89): one concurrency: scenario # per level, each firing N simultaneous streams to characterize the real -# a0/hermes/opencode fan-out. Empty by default — enable deliberately, since -# a burst puts genuine simultaneous load on the serving fleet. -# concurrency_levels = [2, 4, 8] -# concurrency_prompt_tokens = 512 +# a0/hermes/opencode fan-out. Enable deliberately — a burst puts genuine +# simultaneous load on the serving fleet. Bracket the model's max_in_flight +# (beast serves 8) so `helexa-bench report --concurrency` (or GET +# /api/concurrency) can locate the sustainable-concurrency knee (#137): the +# highest N before the model sheds load or its p95 TTFT tail breaks. +concurrency_levels = [1, 2, 4, 8, 16] +concurrency_prompt_tokens = 512 # Capability probes (#91): each runs a fixed prompt and stores the full # output for quality scoring — the reasoning/planning axis the speed