subms/
lib.rs

1//! `subms` - tiny std-only perf harness. Records timed samples per stage and
2//! emits a stable JSON shape consumed by [submillisecond.com](https://submillisecond.com).
3//!
4//! # Pipeline
5//!
6//! ```text
7//! recipe -> SubMsPerfHarness -> SubMsBenchSummary -> { print, assert, JSON }
8//! ```
9//!
10//! [`summarize`] turns the raw harness into a typed [`SubMsBenchSummary`].
11//! [`print_summary`], [`assert_p99_under`], and [`summary_to_json`] are
12//! presenters / asserters that consume the summary - they never recompute stats.
13//!
14//! # Example
15//!
16//! ```
17//! use subms::{SubMsPerfHarness, summarize, print_summary, summary_to_json};
18//!
19//! let mut h = SubMsPerfHarness::new("lsm-tree", "rust");
20//! h.input("entries", &50_000.to_string());
21//! h.input("bloom_mode", "on");
22//! h.add_meta("sstables", "46");
23//!
24//! let put = h.stage("put", 50_000);
25//! for _ in 0..50_000 {
26//!     put.time(|| { /* work under test */ });
27//! }
28//!
29//! let summary = summarize(&h);
30//! print_summary(&summary, &mut std::io::stdout()).unwrap();
31//! summary_to_json(&summary, &mut std::io::stdout()).unwrap();
32//! ```
33//!
34//! # JSON shape (stable; matches the Java sibling jar)
35//!
36//! ```text
37//! {
38//!   "workload": "lsm-tree",
39//!   "lang": "rust",
40//!   "timestamp": "2026-05-13T20:24:38Z",
41//!   "inputs":  { "<k>": "<v>", ... },
42//!   "meta":    { "<k>": "<v>", ... },
43//!   "stages": {
44//!     "<name>": {
45//!       "count": <int>,
46//!       "p50_ns": <int>, "p99_ns": <int>, "p999_ns": <int>, "max_ns": <int>,
47//!       "mean_ns": <int>,
48//!       "samples_ns": [<int>, ...]
49//!     }
50//!   }
51//! }
52//! ```
53
54pub mod bench;
55pub mod bench_config;
56pub mod bench_loops;
57pub mod env;
58pub mod feature;
59pub mod growth;
60pub mod observer;
61pub mod params;
62pub mod recipe;
63mod stats; // private - internal to bench summary computation only
64pub mod summary;
65pub mod timer;
66pub mod util;
67
68pub use bench::{
69    DEFAULT_REGRESSION_THRESHOLD_PCT, SubMsBenchAssertion, assert_p99_under, contended_warmup,
70    diff_summary, diff_summary_with, diff_to_json, format_ns, print_diff, print_summary,
71    print_sweep, run_bench, run_sweep, summarize, summarize_lean, summarize_skipping,
72    summarize_sweep, summarize_windowed, summary_to_json, sweep_to_json,
73};
74pub use bench_config::{SubMsBenchConfig, SubMsCpuPin};
75pub use bench_loops::{bench_indexed_op, bench_keyed_op, bench_templated_op};
76pub use feature::{
77    Json, SubMsFeatureCategory, SubMsFeatureManifest, SubMsP99Source, SubMsStageClass,
78    classify_feature, parse_json, roll_up_stages,
79};
80pub use growth::{
81    GROWTH_VERSION, SubMsGrowthClass, SubMsGrowthRecipe, SubMsGrowthReport, SubMsGrowthRound,
82    SubMsGrowthVerdict, assert_growth_holds, grow, growth_to_json,
83};
84
85// NB: percentile / mean / stddev / cdf_buckets / jitter_score are NOT
86// re-exported from `subms`. They're computed internally to build the
87// JSON summary but the public-API surface of the bench harness should
88// stay small. Anyone wanting rich stats (percentile_sweep, tail
89// analysis, KS, Cohen's d, bootstrap CIs, ...) should add
90// `subms-stats = "0.5"` directly. Recipes should depend on `subms`
91// only and read percentiles off the SubMsStageSummary.
92pub use params::{parse_bool, parse_string, parse_u64, parse_usize};
93pub use recipe::{SubMsBenchParams, SubMsRecipe, benchmark};
94pub use summary::{
95    SubMsBenchDiff, SubMsBenchSummary, SubMsBenchSweep, SubMsMetricDiff, SubMsStageDiff,
96    SubMsStageSummary,
97};
98// SubMsPacedStage is defined inline in this file - re-export it from the root.
99pub use env::{SubMsAppEnv, SubMsAppRegion, env_bool, env_f64, env_i64, env_or, env_str, env_u64};
100pub use observer::{ObservationCtx, SubMsObserver, SubMsStageKind};
101pub use timer::{SubMsTick, SubMsTimer, SubMsTimerCheckpoint};
102pub use util::SubMsLcg;
103
104use std::collections::BTreeMap;
105use std::io::{self, Write};
106use std::sync::Arc;
107use std::thread;
108use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
109
110/// Per-stage sample buffer + recorder. Optionally annotated with a
111/// [`SubMsStageKind`] that sibling adapters (e.g. `subms-otel`) use to pick
112/// histogram bucket boundaries.
113pub struct SubMsStage {
114    name: String,
115    samples: Vec<u64>,
116    kind: SubMsStageKind,
117    // Cheap clones of the harness identity + observer registration so each
118    // recorded sample can build an ObservationCtx without borrowing the
119    // harness back. None of these allocate on the hot path - the Arc clones
120    // happen once at stage construction.
121    workload: Arc<str>,
122    lang: Arc<str>,
123    observer: Option<Arc<dyn SubMsObserver>>,
124}
125
126impl SubMsStage {
127    fn new(
128        name: &str,
129        capacity: usize,
130        workload: Arc<str>,
131        lang: Arc<str>,
132        observer: Option<Arc<dyn SubMsObserver>>,
133    ) -> Self {
134        Self {
135            name: name.to_string(),
136            samples: Vec::with_capacity(capacity),
137            kind: SubMsStageKind::Unspecified,
138            workload,
139            lang,
140            observer,
141        }
142    }
143
144    /// Annotate this stage's kind so observers can pick fitting histogram
145    /// buckets. Default is [`SubMsStageKind::Unspecified`]. Chainable.
146    pub fn with_kind(&mut self, kind: SubMsStageKind) -> &mut Self {
147        self.kind = kind;
148        self
149    }
150
151    /// Record an explicit duration in nanoseconds. Also fires the registered
152    /// observer (if any).
153    pub fn record(&mut self, ns: u64) {
154        self.samples.push(ns);
155        if let Some(obs) = &self.observer {
156            let ctx = ObservationCtx {
157                workload: &self.workload,
158                lang: &self.lang,
159                stage: &self.name,
160                stage_kind: self.kind,
161            };
162            obs.on_record(&ctx, ns);
163        }
164    }
165    /// Time a closure and record its duration.
166    pub fn time<F: FnOnce() -> R, R>(&mut self, f: F) -> R {
167        let t0 = Instant::now();
168        let r = f();
169        self.record(t0.elapsed().as_nanos() as u64);
170        r
171    }
172
173    /// Warm, then record `measured` timed samples of `op`. Runs `op` for
174    /// `warmup` untimed iterations first, then times `measured` more. `op`
175    /// receives the iteration index on both passes (warmup: `0..warmup`,
176    /// measured: `0..measured`); index a shorter input with `i % len`.
177    ///
178    /// On this AOT-compiled side the warmup mainly primes caches and the
179    /// branch predictor. The Java counterpart `warmThenTime` carries the
180    /// real weight: it drives HotSpot to C2 (and lets escape analysis elide
181    /// short-lived allocations) before any sample is recorded, without which
182    /// a JIT-cold low-iteration stage reads orders of magnitude slow. The two
183    /// harnesses expose the method symmetrically so a bench reads the same in
184    /// either language.
185    pub fn warm_then_time<F: FnMut(usize)>(&mut self, warmup: usize, measured: usize, mut op: F) {
186        for i in 0..warmup {
187            op(i);
188        }
189        for i in 0..measured {
190            let t0 = Instant::now();
191            op(i);
192            self.record(t0.elapsed().as_nanos() as u64);
193        }
194    }
195
196    /// Wrap the stage in a coordinated-omission-corrected paced recorder. Each
197    /// [`SubMsPacedStage::time`] call blocks until its intended slot, runs the
198    /// workload, then records latency from the *intended* start time, which
199    /// folds queue delay into the per-op number - the correction
200    /// `HdrHistogram` exists for.
201    ///
202    /// ```ignore
203    /// let mut h = SubMsPerfHarness::new("queue", "rust");
204    /// let stage = h.stage("offer", 100_000);
205    /// let mut paced = stage.with_pacing(10_000.0); // target 10k ops/sec
206    /// for _ in 0..100_000 { paced.time(|| do_work()); }
207    /// ```
208    pub fn with_pacing(&mut self, target_ops_per_second: f64) -> SubMsPacedStage<'_> {
209        SubMsPacedStage::new(self, target_ops_per_second)
210    }
211
212    pub fn name(&self) -> &str {
213        &self.name
214    }
215    pub fn samples(&self) -> &[u64] {
216        &self.samples
217    }
218}
219
220/// Coordinated-omission-corrected stage wrapper. Each [`SubMsPacedStage::time`]
221/// call blocks until its intended slot, runs the workload, then records the
222/// latency from the *intended* start time to end-of-op (so queue delay is
223/// reflected in the per-op latency, not silently dropped).
224///
225/// Use for benches that simulate constant-throughput arrivals - queues, rate
226/// limiters, anything where "if the system stalls, late ops should still count
227/// as slow". Java counterpart: `SubMsPerfHarness.SubMsPacedStage`.
228pub struct SubMsPacedStage<'a> {
229    stage: &'a mut SubMsStage,
230    interval_ns: u64,
231    started_at: Instant,
232    op_index: u64,
233}
234
235impl<'a> SubMsPacedStage<'a> {
236    fn new(stage: &'a mut SubMsStage, target_ops_per_second: f64) -> Self {
237        assert!(
238            target_ops_per_second > 0.0,
239            "target_ops_per_second must be > 0"
240        );
241        let interval_ns = ((1_000_000_000.0 / target_ops_per_second) as u64).max(1);
242        Self {
243            stage,
244            interval_ns,
245            started_at: Instant::now(),
246            op_index: 0,
247        }
248    }
249
250    /// Time the closure; latency is end-of-op minus *intended* start.
251    pub fn time<F: FnOnce() -> R, R>(&mut self, f: F) -> R {
252        let intended_start =
253            self.started_at + Duration::from_nanos(self.op_index * self.interval_ns);
254        let now = Instant::now();
255        if now < intended_start {
256            thread::sleep(intended_start - now);
257        }
258        let r = f();
259        let end = Instant::now();
260        let corrected_latency = end.duration_since(intended_start).as_nanos() as u64;
261        self.stage.record(corrected_latency);
262        self.op_index += 1;
263        r
264    }
265
266    pub fn op_index(&self) -> u64 {
267        self.op_index
268    }
269    pub fn interval_ns(&self) -> u64 {
270        self.interval_ns
271    }
272}
273
274/// A workload run. Owns raw samples + metadata only. Analysis and serialisation
275/// live in [`crate::bench`] - call [`summarize`] to lift this into a
276/// [`SubMsBenchSummary`].
277pub struct SubMsPerfHarness {
278    // Arc<str> so each Stage holds a cheap clone of the harness identity
279    // without per-call string allocation when an observer is registered.
280    workload: Arc<str>,
281    lang: Arc<str>,
282    inputs: BTreeMap<String, String>,
283    meta: BTreeMap<String, String>,
284    stages: Vec<SubMsStage>,
285    observer: Option<Arc<dyn SubMsObserver>>,
286    sample_cap: usize,
287}
288
289impl SubMsPerfHarness {
290    pub fn new(workload: &str, lang: &str) -> Self {
291        // The harness is the instrument, so a capture that does not name its
292        // version is only reconstructible by luck - the same recipe measured by
293        // two harness releases is two experiments. Compile-time, so it cannot
294        // drift from the crate it shipped with.
295        let mut meta = BTreeMap::new();
296        meta.insert(
297            "harness_version".to_string(),
298            env!("CARGO_PKG_VERSION").to_string(),
299        );
300        Self {
301            workload: Arc::from(workload),
302            lang: Arc::from(lang),
303            inputs: BTreeMap::new(),
304            meta,
305            stages: Vec::new(),
306            observer: None,
307            sample_cap: 500,
308        }
309    }
310
311    /// Max points kept in each stage's emitted `samples_ns` timeline. Default
312    /// 500; [`crate::benchmark`] sets it from [`crate::SubMsBenchParams::sample_cap`].
313    /// Clamped to at least 1.
314    pub fn set_sample_cap(&mut self, cap: usize) -> &mut Self {
315        self.sample_cap = cap.max(1);
316        self
317    }
318
319    /// The configured `samples_ns` downsample cap (see [`Self::set_sample_cap`]).
320    pub fn sample_cap(&self) -> usize {
321        self.sample_cap
322    }
323
324    pub fn input(&mut self, key: &str, value: &str) -> &mut Self {
325        self.inputs.insert(key.to_string(), value.to_string());
326        self
327    }
328
329    /// Set a meta field. Renamed from {@code meta} so the {@link Self::meta}
330    /// getter can keep its symmetric name with Java's getter.
331    pub fn add_meta(&mut self, key: &str, value: &str) -> &mut Self {
332        self.meta.insert(key.to_string(), value.to_string());
333        self
334    }
335
336    /// Create a stage; record samples via [`SubMsStage::time`] or [`SubMsStage::record`].
337    pub fn stage(&mut self, name: &str, capacity: usize) -> &mut SubMsStage {
338        let stage = SubMsStage::new(
339            name,
340            capacity,
341            Arc::clone(&self.workload),
342            Arc::clone(&self.lang),
343            self.observer.as_ref().map(Arc::clone),
344        );
345        self.stages.push(stage);
346        self.stages.last_mut().unwrap()
347    }
348
349    /// Register an observer to receive every recorded sample and the
350    /// post-bench summary. Replaces any existing observer; updates already-
351    /// created stages so they fire the new observer too. Returns self for
352    /// builder-style chaining.
353    pub fn with_observer(mut self, observer: Arc<dyn SubMsObserver>) -> Self {
354        self.set_observer(Some(observer));
355        self
356    }
357
358    /// Mutable setter for late wiring. `None` clears the observer.
359    pub fn set_observer(&mut self, observer: Option<Arc<dyn SubMsObserver>>) -> &mut Self {
360        for stage in self.stages.iter_mut() {
361            stage.observer = observer.as_ref().map(Arc::clone);
362        }
363        self.observer = observer;
364        self
365    }
366
367    /// Read the currently-registered observer, if any. Mostly for tests.
368    pub fn observer(&self) -> Option<&Arc<dyn SubMsObserver>> {
369        self.observer.as_ref()
370    }
371
372    /// Borrow a previously-created stage by name.
373    pub fn stage_mut(&mut self, name: &str) -> Option<&mut SubMsStage> {
374        self.stages.iter_mut().find(|s| s.name == name)
375    }
376
377    pub fn stage_by_name(&self, name: &str) -> Option<&SubMsStage> {
378        self.stages.iter().find(|s| s.name == name)
379    }
380
381    pub fn stages(&self) -> &[SubMsStage] {
382        &self.stages
383    }
384
385    pub fn workload(&self) -> &str {
386        &self.workload
387    }
388    pub fn lang(&self) -> &str {
389        &self.lang
390    }
391    pub fn inputs(&self) -> &BTreeMap<String, String> {
392        &self.inputs
393    }
394    pub fn meta(&self) -> &BTreeMap<String, String> {
395        &self.meta
396    }
397
398    /// ISO-8601 seconds-precision timestamp captured at call time. Matches the
399    /// on-disk JSON's `timestamp` field.
400    pub fn timestamp(&self) -> String {
401        iso8601_now()
402    }
403
404    /// Back-compat: summarise + emit JSON in the standard subms JSON shape. New
405    /// code should call [`summarize`] then [`summary_to_json`] so the
406    /// analyser is explicit.
407    pub fn write_json<W: Write>(&self, out: &mut W) -> io::Result<()> {
408        summary_to_json(&summarize(self), out)
409    }
410
411    /// Drop a stage if you never recorded into it.
412    pub fn discard_stage(&mut self, name: &str) {
413        self.stages.retain(|s| s.name != name);
414    }
415}
416
417fn iso8601_now() -> String {
418    let d = SystemTime::now()
419        .duration_since(UNIX_EPOCH)
420        .unwrap_or_default();
421    let secs = d.as_secs() as i64;
422    let mut year = 1970i64;
423    let mut days = secs / 86_400;
424    let rem = secs % 86_400;
425    let hour = rem / 3600;
426    let minute = (rem % 3600) / 60;
427    let second = rem % 60;
428    while days >= year_days(year) {
429        days -= year_days(year);
430        year += 1;
431    }
432    let mut month = 1u32;
433    for m in 1..=12 {
434        let dm = month_days(year, m);
435        if days < dm as i64 {
436            month = m;
437            break;
438        }
439        days -= dm as i64;
440    }
441    let day = (days + 1) as u32;
442    format!(
443        "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z",
444        year, month, day, hour, minute, second
445    )
446}
447
448fn year_days(y: i64) -> i64 {
449    if (y % 4 == 0 && y % 100 != 0) || (y % 400 == 0) {
450        366
451    } else {
452        365
453    }
454}
455fn month_days(y: i64, m: u32) -> u32 {
456    match m {
457        1 | 3 | 5 | 7 | 8 | 10 | 12 => 31,
458        4 | 6 | 9 | 11 => 30,
459        2 => {
460            if (y % 4 == 0 && y % 100 != 0) || (y % 400 == 0) {
461                29
462            } else {
463                28
464            }
465        }
466        _ => 0,
467    }
468}
469
470/// Parse stdin `key=value` lines into a flat map. Skips blank lines and `#` comments.
471pub fn read_stdin_kv() -> BTreeMap<String, String> {
472    use std::io::BufRead;
473    let mut m = BTreeMap::new();
474    let stdin = io::stdin();
475    for line in stdin.lock().lines().map_while(Result::ok) {
476        let line = line.trim();
477        if line.is_empty() || line.starts_with('#') {
478            continue;
479        }
480        if let Some((k, v)) = line.split_once('=') {
481            m.insert(k.trim().to_string(), v.trim().to_string());
482        }
483    }
484    m
485}
486
487#[cfg(test)]
488#[path = "subms_tests.rs"]
489mod tests;
subms/lib.rs

subms/
lib.rs
