metrale_bench/benchmarks/spec_cost/
mod.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! 2026-10-04: Speculative step cost calibration: on the serve it is pointed at, the wall time
4//! and GPU-rail energy of ONE speculative scheduler step at each batch width `n`, for the
5//! draft depth `k` the serve runs, split into its draft (propose) and verify shares. A
6//! measurement tool for the speculative cost model, not a gate: `gate::coverage::NOT_REQUIRED`
7//! lists it.
8//!
9//! Per width, ascending (`wave.rs`): `n` concurrent streaming essay requests, each with its
10//! own prompt identity; once every stream has delivered a token and `settle_s` has passed,
11//! two `/metrics` snapshots `window_s` apart and the rail energy between them; then the
12//! streams are closed and the serve drained to idle. The arithmetic, and its proportional
13//! energy split, are in `cell.rs`.
14//!
15//! The pins (the benchmark's definition, not parameters):
16//! - Prompt: `wave::PROMPT_TOKENS` of `stats::make_prompt` filler, then the concurrency
17//!   sweep's essay ask (`concurrency::ESSAY_TASK`), which keeps a model writing.
18//! - Request: temperature 0, seed 0, presence and frequency penalty 0, and
19//!   `reasoning_effort: "none"` (thinking off), as in `decode_floor`: speculative dispatch is
20//!   off inside `<think>` unless `METRALE_DFLASH_SPEC_THINK=1`.
21//! - A speculative serve (`k` > 0) must run with `--telemetry basic` or `kernel`, which
22//!   exports the scheduler phase histogram; without it the run fails.
23//!
24//! Owner: bench, spec-cost.
25//! Invariants: the run verdict is never a pass: info when at least one width was measured,
26//! INCONCLUSIVE otherwise.
27
28use crate::hardware::Sensitivity;
29use std::collections::BTreeMap;
30use std::future::Future;
31use std::time::{Duration, Instant};
32
33use anyhow::{Context, Result, bail};
34
35use crate::benchmark::{Benchmark, BenchmarkDescriptor};
36use crate::hardware::energy_sampler::EnergyMeter;
37use crate::http;
38use crate::metadata::PluginMetadata;
39use crate::params::{ParamKind, ParamSpec, ParamValue, ParamValues};
40use crate::plugin::{Plugin, PluginHandle};
41use crate::result::{
42    BenchmarkResult, Cell, CellStyle, Column, LogLine, ResultTable, RunStatus, Verdict,
43};
44
45pub mod acceptance;
46mod cell;
47mod prom;
48pub mod table_input;
49mod wave;
50
51use cell::CellVerdict;
52
53const SUMMARY: &str = "Speculative step cost calibration: wall time and GPU-rail energy of one \
54                       speculative step per batch width, split into draft and verify";
55pub const METADATA: PluginMetadata = PluginMetadata::metrale(SUMMARY);
56
57pub const DESCRIPTOR: BenchmarkDescriptor = BenchmarkDescriptor {
58    id: "spec-cost",
59    name: "Speculative step cost calibration",
60    summary: SUMMARY,
61    detail: "For each batch width n (param widths, ascending), sends n concurrent streaming \
62             essay requests (distinct prompts, temperature 0, seed 0, thinking off, max_tokens \
63             = osl). Once every stream has a token and settle_s has passed, it reads /metrics \
64             twice, window_s apart, and integrates the GPU rail over the window. Per width it \
65             records n{n}_steps, n{n}_verify_ms, n{n}_verify_j, n{n}_draft_ms, n{n}_draft_j, \
66             n{n}_tok_per_step, n{n}_wall_ms, n{n}_nvml_j and n{n}_vacuous, plus k. Steps and \
67             their time come from the scheduler phase histogram (phase step_mtp, its propose \
68             share from phase propose), so a speculative serve needs --telemetry basic; at \
69             k = 0 a step is one token per sequence. Joules are the bench's GPU-rail integral \
70             over the window; n{n}_nvml_j is the change in the serve's NVML energy counter per \
71             step, a cross-check that repeats less well. Energy is split between draft and \
72             verify in proportion to time (an assumption, not a measurement). A width is \
73             vacuous when a stream ends \
74             inside the window, fewer than 20 steps were seen, or the serve did not time the \
75             propose phase. Informational: no bounds.",
76    duration_hint: "~2 min",
77    expected_secs: 120,
78    updated: "2026-10-04",
79    intended_for: None,
80    threshold_params: &[],
81    needs_confirmation: false,
82    // 2026-10-04: Step time and joules per step are rates: a throttled box reports a cost the
83    // code did not cause.
84    sensitivity: Sensitivity::Speed,
85    ctor: || Box::new(SpecCost::default()),
86};
87
88/// 2026-10-04: The `k` value that means "not stated"; `configure` refuses it, so the caller
89/// must state the depth the serve runs.
90const UNSTATED_K: i64 = -1;
91/// 2026-10-04: The deepest draft the serve's verify arms take (K4 = three drafts).
92const MAX_K: i64 = 3;
93
94#[derive(Default)]
95pub struct SpecCost {
96    handle: Option<PluginHandle>,
97    started: Option<Instant>,
98    timeout: Duration,
99    k: u8,
100    widths: Vec<usize>,
101    osl: usize,
102    window: Duration,
103    settle: Duration,
104    probed: bool,
105    /// 2026-10-04: The probe's `/metrics` page, the start of the acceptance counts.
106    first_scrape: Option<prom::Scrape>,
107    rows: Vec<(usize, CellVerdict)>,
108    /// 2026-10-04: GPU-rail sampling, started after the probe (idle baseline).
109    energy: EnergyMeter,
110}
111
112impl SpecCost {
113    fn handle(&self) -> Result<&PluginHandle> {
114        self.handle.as_ref().context("benchmark was not loaded")
115    }
116
117    fn elapsed(&self) -> Duration {
118        self.started.map(|s| s.elapsed()).unwrap_or_default()
119    }
120
121    fn table(&self) -> ResultTable {
122        let mut t = ResultTable::new(
123            "SPECULATIVE STEP COST",
124            vec![
125                Column::right("n", 4),
126                Column::right("Steps", 7),
127                Column::right("Verify ms", 10),
128                Column::right("Verify J", 9),
129                Column::right("Draft ms", 9),
130                Column::right("Draft J", 8),
131                Column::right("Tok/step", 9),
132            ],
133        );
134        let joules = |v: Option<f64>| v.map(|v| format!("{v:.3}")).unwrap_or_else(|| "—".into());
135        for (n, verdict) in &self.rows {
136            let row = match verdict {
137                CellVerdict::Vacuous(_) => {
138                    let mut row = vec![
139                        Cell::new(n.to_string()),
140                        Cell::styled("vacuous", CellStyle::Warn),
141                    ];
142                    row.extend((0..5).map(|_| Cell::new("—")));
143                    row
144                }
145                CellVerdict::Measured(c) => vec![
146                    Cell::new(n.to_string()),
147                    Cell::new(format!("{:.0}", c.steps)),
148                    Cell::styled(format!("{:.2}", c.verify_ms), CellStyle::Accent),
149                    Cell::new(joules(c.verify_j)),
150                    Cell::new(format!("{:.2}", c.draft_ms)),
151                    Cell::new(joules(c.draft_j)),
152                    Cell::new(format!("{:.2}", c.tok_per_step)),
153                ],
154            };
155            t.push(row);
156        }
157        t
158    }
159
160    fn finish(&self, mut metrics: BTreeMap<String, f64>) -> BenchmarkResult {
161        metrics.insert(cell::KEY_K.to_string(), f64::from(self.k));
162        for (n, verdict) in &self.rows {
163            cell::record(*n, verdict, &mut metrics);
164        }
165        self.energy.metrics(&mut metrics);
166        let measured = self
167            .rows
168            .iter()
169            .filter(|(_, v)| matches!(v, CellVerdict::Measured(_)))
170            .count();
171        let verdict = if measured == 0 {
172            Verdict::fail("INCONCLUSIVE: every width was vacuous")
173        } else {
174            Verdict::info(format!(
175                "{measured} of {} widths measured at k = {} — a calibration table, no bounds",
176                self.rows.len(),
177                self.k
178            ))
179        };
180        let total = self.widths.len() as u64;
181        BenchmarkResult {
182            status: RunStatus::Completed,
183            ..BenchmarkResult::running("done", self.elapsed())
184        }
185        .with_progress(total, total)
186        .with_table(self.table())
187        .with_metrics(metrics)
188        .with_verdict(verdict)
189    }
190}
191
192impl Plugin for SpecCost {
193    fn metadata(&self) -> &'static PluginMetadata {
194        &METADATA
195    }
196
197    fn load(&mut self, handle: PluginHandle) -> impl Future<Output = Result<()>> + Send {
198        self.handle = Some(handle);
199        self.started = Some(Instant::now());
200        async { Ok(()) }
201    }
202}
203
204impl Benchmark for SpecCost {
205    fn descriptor(&self) -> &'static BenchmarkDescriptor {
206        &DESCRIPTOR
207    }
208
209    fn parameters(&self) -> Vec<ParamSpec> {
210        vec![
211            ParamSpec::new(
212                "k",
213                "Draft depth",
214                "Drafts per speculative step the serve runs (0 = a serve without \
215                 --speculative). Required: the default -1 means not stated and is refused.",
216                ParamKind::Int {
217                    min: UNSTATED_K,
218                    max: MAX_K,
219                },
220                ParamValue::Int(UNSTATED_K),
221            ),
222            ParamSpec::new(
223                "widths",
224                "Batch widths",
225                "Concurrent streams per measured window, strictly ascending.",
226                ParamKind::IntList { min: 1, max: 256 },
227                ParamValue::IntList(vec![1, 2, 4, 8, 16]),
228            ),
229            ParamSpec::new(
230                "osl",
231                "Output budget",
232                "max_tokens of every stream. Must outlast settle + window at the serve's \
233                 per-stream rate, or the width is vacuous.",
234                ParamKind::Int {
235                    min: 64,
236                    max: 16384,
237                },
238                ParamValue::Int(1024),
239            ),
240            ParamSpec::new(
241                "window_s",
242                "Window",
243                "Seconds between the two /metrics snapshots of a width.",
244                ParamKind::Int { min: 1, max: 120 },
245                ParamValue::Int(8),
246            ),
247            ParamSpec::new(
248                "settle_s",
249                "Settle",
250                "Seconds to wait after every stream has its first token before the window \
251                 opens.",
252                ParamKind::Int { min: 0, max: 60 },
253                ParamValue::Int(3),
254            ),
255            ParamSpec::new(
256                "request_timeout_s",
257                "Request timeout",
258                "Seconds before a single stream, or a /metrics read, is abandoned. \
259                 Transport-side only.",
260                ParamKind::Int { min: 30, max: 3600 },
261                ParamValue::Int(300),
262            ),
263        ]
264    }
265
266    fn configure(&mut self, values: &ParamValues) -> Result<()> {
267        let specs = self.parameters();
268        values.validate_against(&specs)?;
269        let k = values.int("k")?;
270        if k == UNSTATED_K {
271            bail!(
272                "k: state the draft depth the serve runs with --param k=N (0..={MAX_K}; 0 for a \
273                 serve without --speculative); it has no default"
274            );
275        }
276        self.k = u8::try_from(k).context("k is out of range")?;
277        let widths = values
278            .int_list("widths")?
279            .iter()
280            .map(|&w| usize::try_from(w).context("widths: a width is negative"))
281            .collect::<Result<Vec<_>>>()?;
282        if widths.windows(2).any(|p| p[0] >= p[1]) {
283            bail!("widths: {widths:?} is not strictly ascending");
284        }
285        self.widths = widths;
286        self.osl = values.usize("osl")?;
287        self.window = Duration::from_secs(values.usize("window_s")? as u64);
288        self.settle = Duration::from_secs(values.usize("settle_s")? as u64);
289        self.timeout = Duration::from_secs(values.usize("request_timeout_s")? as u64);
290        self.probed = false;
291        self.first_scrape = None;
292        self.rows.clear();
293        self.energy = EnergyMeter::default();
294        Ok(())
295    }
296
297    async fn next(&mut self) -> Result<BenchmarkResult> {
298        let handle = self.handle()?.clone();
299        handle.check_cancelled()?;
300        let total = self.widths.len() as u64;
301
302        if !self.probed {
303            self.probed = true;
304            http::probe(handle.target(), Duration::from_secs(10))
305                .await
306                .context("endpoint probe failed — check the target URL and port")?;
307            self.first_scrape = Some(wave::scrape_metrics(handle.target(), self.timeout).await?);
308            // 2026-10-04: Nothing is in flight yet, so the idle baseline is taken now.
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                    "{} · k {} · widths {:?} · ~{} prompt tokens · osl {} · settle {}s · \
316                     window {}s",
317                    handle.target().base_url,
318                    self.k,
319                    self.widths,
320                    wave::PROMPT_TOKENS,
321                    self.osl,
322                    self.settle.as_secs(),
323                    self.window.as_secs(),
324                ))));
325        }
326
327        let Some(&n) = self.widths.get(self.rows.len()) else {
328            if let Some(line) = self.energy.stop().await {
329                handle.log(line.level, line.text);
330            }
331            let last = wave::scrape_metrics(handle.target(), self.timeout).await?;
332            let first = self
333                .first_scrape
334                .take()
335                .context("no scrape before the first width")?;
336            let mut acceptance = BTreeMap::new();
337            acceptance::record(&first, &last, &mut acceptance)?;
338            return Ok(self.finish(acceptance));
339        };
340        handle.status(format!("width {n}: {n} concurrent streams"));
341        let outcome = self.measure_width(n).await?;
342        let mut line = match &outcome.verdict {
343            CellVerdict::Vacuous(why) => LogLine::warn(format!("n {n}: vacuous — {why}")),
344            CellVerdict::Measured(c) => LogLine::info(format!(
345                "n {n}: {:.0} steps · step {:.2} ms (verify {:.2}, draft {:.2}) · wall {:.2} ms \
346                 · {:.2} tok/step",
347                c.steps, c.step_ms, c.verify_ms, c.draft_ms, c.wall_ms, c.tok_per_step
348            )),
349        };
350        for failure in &outcome.failures {
351            line.text.push_str(&format!(" · stream failed: {failure}"));
352        }
353        self.rows.push((n, outcome.verdict));
354        let done = self.rows.len() as u64;
355        handle.progress(done, total);
356        Ok(BenchmarkResult::running("timed", self.elapsed())
357            .with_progress(done, total)
358            .with_table(self.table())
359            .log_line(line))
360    }
361}
362
363#[cfg(test)]
364#[path = "spec_cost_tests.rs"]
365mod tests;
metrale_bench/benchmarks/spec_cost/mod.rs

metrale_bench/benchmarks/spec_cost/
mod.rs
