metrale_bench/benchmarks/
concurrency.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! 2026-09-26: The concurrency sweep: for every (ISL × concurrency) cell, send
4//! `conc` streaming requests at once and report client TTFT, TPOT and E2E
5//! percentiles plus the batch's aggregate output tok/s. Each `next()` runs one
6//! cell. Every request's delivered evidence is kept, and a cell that is
7//! vacuous (`concurrency_vacuity.rs`), errored or cache-uncontrolled is left
8//! out of the gate metrics.
9//!
10//! Owner: bench (concurrency).
11//! Invariants: the per-rung and peak throughput keys in [`ConcurrencySweep`]'s
12//! metrics come only from cells for which `CellRow::comparable` holds.
13
14use crate::hardware::Sensitivity;
15use crate::hardware::energy::EnergyWindow;
16use crate::hardware::energy_sampler::EnergyMeter;
17use crate::http::{GapSample, GapStats};
18use std::collections::BTreeMap;
19use std::future::Future;
20use std::time::{Duration, Instant};
21
22use anyhow::{Context, Result, bail};
23use serde_json::json;
24
25use crate::benchmark::{Benchmark, BenchmarkDescriptor};
26use crate::benchmarks::stats::{self, Percentiles, PromptMode};
27use crate::http;
28use crate::metadata::PluginMetadata;
29use crate::params::{ParamKind, ParamSpec, ParamValue, ParamValues};
30use crate::plugin::{Plugin, PluginHandle};
31use crate::result::{
32    BenchmarkResult, Cell, CellStyle, Column, LogLine, ResultTable, RunStatus, Stat,
33};
34use cache::{
35    SSM_CACHE_SLOTS_KEY, WARM_CACHE_FLOOR, cache_is_uncontrolled, slots_needed, warm_cache_capable,
36};
37pub use descriptors::{DESCRIPTOR, DFLASH2_DESCRIPTOR, METADATA, MOE_DESCRIPTOR};
38use descriptors::{PEAK_FLOOR, RUNGS};
39use prompt::{Fixture, prompt_plan};
40use report::evidence_line;
41use vacuity::{Delivery, VACUITY_FLOOR};
42
43/// 2026-09-26: The natural fixture's ask, appended after the ISL filler.
44const CODE_TASK: &str = "Ignore the reference text above. Task: write a complete, \
45    production-quality MinHeap class in Python with insert, peek_min, extract_min, \
46    decrease_key and heapify methods — full docstrings, input validation and a worked \
47    usage example — followed by a unit-test class covering every method, including \
48    empty-heap, duplicate-key and single-element edge cases. Write every method and \
49    every test out in full; do not summarize or elide any code.";
50
51/// 2026-09-26: The published ladder's ask: `SUFFIX_ESSAY` in
52/// `bench/ladder38/harness_w55_conc_ladder.py`, byte for byte.
53/// `concurrency_essay_tests.rs` pins whole prompts against SHA-256 digests of
54/// that harness's output.
55///
56/// Kept here rather than as a `stats::PromptMode` variant: `gate::coverage`
57/// excludes this file from every non-concurrency gate's invalidation set but
58/// does not exclude `stats.rs`, so a prompt added there would invalidate every
59/// gate.
60pub(crate) const ESSAY_TASK: &str = " Using the text above only as a starting point, write a long, \
61    richly detailed essay that keeps introducing new specifics, examples and vocabulary. \
62    Never repeat a sentence or paraphrase one you have already written. \
63    Do not summarise and do not stop early.";
64
65/// 2026-09-26: The harness's `NONCE_WIDTH`. A fixed-width nonce keeps every
66/// request's prompt the same length.
67const ESSAY_NONCE_WIDTH: usize = 6;
68
69/// 2026-09-26: `req NNNNNN` for request `i`, taken modulo
70/// `10^ESSAY_NONCE_WIDTH` so it never widens, as the harness's `make_prompt`
71/// renders its nonce.
72fn essay_nonce_tag(i: usize) -> String {
73    let modulus = 10usize.pow(ESSAY_NONCE_WIDTH as u32);
74    format!("req {:0width$}", i % modulus, width = ESSAY_NONCE_WIDTH)
75}
76
77/// 2026-09-26: What one completed request delivered, as parsed by
78/// `http::ChatOutcome` from the stream.
79#[derive(Clone, Debug, Default)]
80struct RequestEvidence {
81    completion_tokens: usize,
82    prompt_tokens: usize,
83    cached_prompt_tokens: usize,
84    finish_reason: Option<String>,
85    server_ttft_ms: Option<f64>,
86    server_tps: Option<f64>,
87    /// 2026-09-26: `usage.completion_tokens_details.accepted_prediction_tokens`.
88    accepted_prediction_tokens: Option<usize>,
89}
90
91#[derive(Default)]
92struct CellRow {
93    isl: usize,
94    conc: usize,
95    ttft: Percentiles,
96    tpot: Percentiles,
97    /// 2026-09-26: Server-clock ITL; empty when the server does not report
98    /// `usage.decode_time_ms`.
99    server_tpot: Percentiles,
100    e2e_p50: Option<f64>,
101    throughput: f64,
102    /// 2026-09-26: `throughput`'s numerator and the J/token denominator.
103    tokens: usize,
104    errors: usize,
105    requests: Vec<RequestEvidence>,
106    vacuous: bool,
107    cache_uncontrolled: bool,
108    gaps: Option<GapStats>,
109    energy: Option<EnergyWindow>,
110}
111
112#[derive(Default)]
113pub struct ConcurrencySweep {
114    handle: Option<PluginHandle>,
115    cells: Vec<(usize, usize)>,
116    cursor: usize,
117    osl: usize,
118    warmup: usize,
119    fixture: Fixture,
120    timeout: Duration,
121    rows: Vec<CellRow>,
122    started: Option<Instant>,
123    probed: bool,
124    /// 2026-09-26: Verdict floors in tok/s; with all at 0.0 an error-free
125    /// sweep gets an info verdict.
126    floors: verdict::Floors,
127    /// 2026-09-26: Started after the endpoint probe, with the idle baseline
128    /// taken before the first cell; one window per measured batch.
129    energy: EnergyMeter,
130}
131
132impl ConcurrencySweep {
133    fn handle(&self) -> Result<&PluginHandle> {
134        self.handle.as_ref().context("benchmark was not loaded")
135    }
136
137    /// 2026-09-26: One verdict-floor spec. Floors decide only the verdict and
138    /// cannot change a measured number.
139    fn floor_spec(key: &'static str, label: &'static str) -> ParamSpec {
140        ParamSpec::new(
141            key,
142            label,
143            "Run-verdict floor on this rung's aggregate tok/s (comparable cells only). 0 \
144             disables (a standalone run reports an info verdict); under --pull-request-gate \
145             it is auto-filled from the variant's BENCH.toml `min` bound. Vacuous or errored \
146             sweeps never PASS regardless.",
147            ParamKind::Float {
148                min: 0.0,
149                max: 100_000.0,
150            },
151            // 2026-09-26: 0.0 is the documented off state, not an implicit bar.
152            ParamValue::Float(0.0),
153        )
154    }
155
156    fn elapsed(&self) -> Duration {
157        self.started.map(|s| s.elapsed()).unwrap_or_default()
158    }
159
160    /// 2026-09-26: `Err` for a transport failure and for a stream the server
161    /// failed (an in-band error frame, or no terminal frame; see
162    /// `http::chat_stream`). A completed request with zero tokens is `Ok`.
163    async fn one(&self, isl: usize, prefix_tag: String) -> Result<http::ChatOutcome> {
164        let handle = self.handle()?;
165        let target = handle.target();
166        let body = self.request_body(&target.model, isl, &prefix_tag);
167        http::chat_stream(target, &body, self.timeout).await
168    }
169}
170
171impl Plugin for ConcurrencySweep {
172    fn metadata(&self) -> &'static PluginMetadata {
173        &METADATA
174    }
175
176    fn load(&mut self, handle: PluginHandle) -> impl Future<Output = Result<()>> + Send {
177        self.handle = Some(handle);
178        self.started = Some(Instant::now());
179        async { Ok(()) }
180    }
181}
182
183impl Benchmark for ConcurrencySweep {
184    fn descriptor(&self) -> &'static BenchmarkDescriptor {
185        &DESCRIPTOR
186    }
187
188    fn parameters(&self) -> Vec<ParamSpec> {
189        let mut specs = vec![
190            ParamSpec::new(
191                "concurrencies",
192                "Concurrency levels",
193                "How many requests are in flight at once, one sweep column each.",
194                ParamKind::IntList { min: 1, max: 256 },
195                ParamValue::IntList(vec![1, 2, 4, 8, 16, 32]),
196            ),
197            ParamSpec::new(
198                "isls",
199                "Input lengths",
200                "Prompt sizes in tokens. Must fit inside the server's --max-seq-len with the output.",
201                ParamKind::IntList {
202                    min: 16,
203                    max: 131_072,
204                },
205                ParamValue::IntList(vec![128, 512, 1024, 2048]),
206            ),
207            ParamSpec::new(
208                "osl",
209                "Output tokens",
210                "Max tokens per request.",
211                ParamKind::Int { min: 1, max: 8192 },
212                ParamValue::Int(128),
213            ),
214            ParamSpec::new(
215                "warmup",
216                "Warm-up rounds",
217                "Unmeasured rounds per cell. Each round runs every exact prompt in the measured \
218                 batch so prefix-cache state is controlled before timing.",
219                ParamKind::Int { min: 0, max: 8 },
220                ParamValue::Int(1),
221            ),
222            ParamSpec::new(
223                "prompt_mode",
224                "Prompt mode",
225                // 2026-09-26: No mode forces the output budget (the server does
226                // not accept `ignore_eos`); short completions are caught by the
227                // vacuity rule.
228                "natural (default) poses a code-generation task that reliably fills the output \
229                 budget; count appends a counting instruction the model may still stop early on; \
230                 essay sends the published ladder38 request (bench/ladder38/harness_w55_conc_ladder.py \
231                 essay mode: the long-essay ask, a fixed-width request nonce, presence/frequency \
232                 penalty pinned to 0.0) byte for byte, so a cell can be compared with the published \
233                 ISL 128 / OSL 1024 ladder. None forces the budget — under-budget cells are \
234                 flagged vacuous.",
235                ParamKind::Choice(&["natural", "count", "essay"]),
236                ParamValue::Text("natural".into()),
237            ),
238            ParamSpec::new(
239                "request_timeout_s",
240                "Request timeout",
241                "Seconds before a single request is abandoned and counted as an error.",
242                ParamKind::Int { min: 10, max: 3600 },
243                ParamValue::Int(600),
244            ),
245        ];
246        // 2026-09-26: One floor per entry of `RUNGS`, plus the peak.
247        specs.extend(
248            RUNGS
249                .iter()
250                .map(|(_, key, _, label)| Self::floor_spec(key, label)),
251        );
252        specs.push(Self::floor_spec(PEAK_FLOOR.0, PEAK_FLOOR.2));
253        specs
254    }
255
256    fn configure(&mut self, values: &ParamValues) -> Result<()> {
257        let specs = self.parameters();
258        values.validate_against(&specs)?;
259        let concurrencies = values.int_list("concurrencies")?.to_vec();
260        let isls = values.int_list("isls")?.to_vec();
261        // 2026-09-26: ISL-major: a full concurrency curve at one prompt size
262        // before the next size.
263        self.cells = isls
264            .iter()
265            .flat_map(|isl| {
266                concurrencies
267                    .iter()
268                    .map(move |c| (*isl as usize, *c as usize))
269            })
270            .collect();
271        self.osl = values.usize("osl")?;
272        self.warmup = values.usize("warmup")?;
273        self.fixture = Fixture::parse(values.text("prompt_mode")?)
274            .context("prompt_mode must be natural, count or essay")?;
275        self.timeout = Duration::from_secs(values.usize("request_timeout_s")? as u64);
276        let mut per_c = Vec::with_capacity(RUNGS.len());
277        for (c, key, _, _) in RUNGS {
278            per_c.push((c, values.float(key)?));
279        }
280        self.floors = verdict::Floors {
281            per_c,
282            peak: values.float(PEAK_FLOOR.0)?,
283        };
284        self.cursor = 0;
285        self.rows.clear();
286        // 2026-09-26: A fresh meter per configuration; dropping a live sampler
287        // kills its child process (`kill_on_drop`).
288        self.energy = EnergyMeter::default();
289        Ok(())
290    }
291
292    async fn next(&mut self) -> Result<BenchmarkResult> {
293        let handle = self.handle()?.clone();
294        handle.check_cancelled()?;
295
296        // 2026-09-26: Probe first, so a wrong port fails at once rather than
297        // as a sweep of transport errors.
298        if !self.probed {
299            self.probed = true;
300            http::probe(handle.target(), Duration::from_secs(10))
301                .await
302                .context("endpoint probe failed — check the target URL and port")?;
303            let total = self.cells.len() as u64;
304            if total == 0 {
305                bail!("no cells to run — check the concurrency and input-length lists");
306            }
307            // 2026-09-26: Nothing is in flight yet, so the idle baseline
308            // precedes the first measured window.
309            for line in self.energy.start(handle.target()).await {
310                handle.log(line.level, line.text);
311            }
312            return Ok(BenchmarkResult::running("probe", self.elapsed())
313                .with_progress(0, total)
314                .log_line(LogLine::info(format!(
315                    "{} · model {} · {total} cells",
316                    handle.target().base_url,
317                    handle.target().model
318                ))));
319        }
320
321        if self.cursor >= self.cells.len() {
322            let errors: usize = self.rows.iter().map(|r| r.errors).sum();
323            let vacuous = self.rows.iter().filter(|r| r.vacuous).count();
324            let cache_uncontrolled = self.rows.iter().filter(|r| r.cache_uncontrolled).count();
325            let non_mtp_arm = self.rows.iter().filter(|r| r.arm_is_not_mtp()).count();
326            // 2026-09-26: Stopped before `metrics()`, which records its cost.
327            let sampler_cost_line = self.energy.stop().await;
328            // 2026-09-26: The verdict reads the same metrics map the record
329            // carries.
330            let metrics = self.metrics();
331            let verdict = verdict::sweep_verdict(
332                &metrics,
333                self.rows.len(),
334                errors,
335                verdict::Exclusions {
336                    vacuous,
337                    cache_uncontrolled,
338                    non_mtp_arm,
339                },
340                VACUITY_FLOOR * 100.0,
341                &self.floors,
342            );
343            let mut frame = BenchmarkResult {
344                status: RunStatus::Completed,
345                ..BenchmarkResult::running("done", self.elapsed())
346            }
347            .with_progress(self.cells.len() as u64, self.cells.len() as u64)
348            .with_summary(self.summary())
349            .with_table(self.table())
350            .with_metrics(metrics)
351            .with_verdict(verdict);
352            if let Some(line) = sampler_cost_line {
353                frame = frame.log_line(line);
354            }
355            let unmeasured = self.rows.iter().filter(|r| r.tpot.p50.is_none()).count();
356            if unmeasured > 0 {
357                frame = frame.log_line(LogLine::warn(format!(
358                    "TPOT unmeasured in {unmeasured} cell(s): the endpoint sent the whole reply \
359                     in one SSE delta, so there is no inter-token interval to time. Raise the \
360                     output-token budget to measure decode."
361                )));
362            }
363            return Ok(frame);
364        }
365
366        let (isl, conc) = self.cells[self.cursor];
367        let row = self.run_cell(isl, conc).await?;
368        let line = LogLine::info(format!(
369            "isl {isl} conc {conc}: ttft p50 {} ms · tpot p50 {} ms (server clock {} ms) · \
370             {:.1} tok/s{}",
371            stats::fmt_ms(row.ttft.p50),
372            stats::fmt_ms(row.tpot.p50),
373            stats::fmt_ms(row.server_tpot.p50),
374            row.throughput,
375            if row.vacuous { " (vacuous)" } else { "" }
376        ));
377        self.rows.push(row);
378        self.cursor += 1;
379        handle.progress(self.cursor as u64, self.cells.len() as u64);
380        Ok(
381            BenchmarkResult::running(format!("isl {isl} · conc {conc}"), self.elapsed())
382                .with_progress(self.cursor as u64, self.cells.len() as u64)
383                .with_summary(self.summary())
384                .with_table(self.table())
385                .log_line(line),
386        )
387    }
388}
389
390#[path = "concurrency_instruments.rs"]
391mod instruments;
392
393#[path = "concurrency_verdict.rs"]
394mod verdict;
395
396#[path = "concurrency_vacuity.rs"]
397mod vacuity;
398
399#[path = "concurrency_descriptors.rs"]
400mod descriptors;
401
402#[path = "concurrency_prompt.rs"]
403mod prompt;
404
405#[path = "concurrency_cache.rs"]
406mod cache;
407
408#[path = "concurrency_cell.rs"]
409mod cell;
410
411#[path = "concurrency_report.rs"]
412mod report;
413
414#[cfg(test)]
415#[path = "concurrency_tests.rs"]
416mod concurrency_tests;
417
418#[cfg(test)]
419#[path = "concurrency_verdict_tests.rs"]
420mod concurrency_verdict_tests;
421
422#[cfg(test)]
423#[path = "concurrency_vacuity_tests.rs"]
424mod concurrency_vacuity_tests;
425
426#[cfg(test)]
427#[path = "concurrency_moe_tests.rs"]
428mod concurrency_moe_tests;
metrale_bench/benchmarks/concurrency.rs

metrale_bench/benchmarks/
concurrency.rs
