crates/nornir-bench/src/telemetry.rs
//! Low-overhead CPU/memory telemetry for a single bench unit.
//!
//! A [`Sampler`] spawns one background thread that, ~once a second, reads four
//! cheap `/proc` files — `/proc/stat` (aggregate + per-core jiffy counters),
//! `/proc/self/stat` (**this process's** own CPU time), `/proc/meminfo` (system
//! memory), and `/proc/self/status` (this process's peak RSS) — and folds the
//! deltas into running min/avg/max accumulators. When the bench unit finishes
//! the caller calls [`Sampler::stop`], which joins the thread and returns a
//! [`Telemetry`] summary.
//!
//! Why `/proc` and not a perf tool: the whole point is **non-perturbation**.
//! Each sample is a handful of small synchronous file reads on the *sampler*
//! thread (not the bench thread) and a couple of integer subtractions — well
//! under a millisecond, once a second. No `perf`, no `ptrace`.
//!
//! ## HOST-WIDE vs PROCESS-SCOPED — read this before trusting a number
//!
//! `/proc/stat` counts **the whole box**, every process on it. On a shared or
//! loaded machine its numbers say nothing whatsoever about the bench:
//!
//! > Measured on oden (32 cores, loadavg ~36) 2026-08-03: the strictly
//! > single-threaded arm `gunnar.walk_1m` reported **`cores_busy_max` 32/32 and
//! > `cpu_pct_max` 99.9%** while the benchmark process itself occupied
//! > **0.56 of one core**. The neighbours' load was being published as the
//! > bench's own parallelism.
//!
//! So the host-wide family (`cpu_pct_*`, `cores_busy_*`) is **ambient context
//! only** — how busy the box was while we measured — and MUST NOT be used to
//! judge whether a bench used its cores. Everything named `proc_*` /
//! `child_cpu_ms` is derived from `/proc/self/stat` and is host-independent:
//! that is what a saturation gate reads.
//!
//! ## What is computed
//! HOST-WIDE (ambient — the whole box, not us):
//! - **CPU% aggregate** — `100 * (1 - idle_delta/total_delta)` from
//! `/proc/stat`'s first (`cpu`) line across a sampling interval. avg + max.
//! - **cores-busy** — of the N per-core (`cpu0`, `cpu1`, …) lines, how many had
//! per-core utilisation **> 50%** over the interval. avg (rounded) + max.
//!
//! PROCESS-SCOPED (us — the number that means something):
//! - **`proc_cores_avg`** — **task-clock ÷ wall**: `(utime+stime+cutime+cstime)`
//! deltas from `/proc/self/stat` (fields 14/15/16/17, in clock ticks, divided
//! by `sysconf(_SC_CLK_TCK)`) over the wall-clock window. 1.0 means "this
//! bench occupied exactly one core"; 12.0 means twelve. Also published live
//! (see [`Sampler::proc_cores_now`]) so a long-running bench names its own
//! occupancy in its progress line.
//! - **`proc_cores_max`** — the same ratio, peak over the 1 Hz intervals.
//! - **`proc_cpu_ms`** — CPU time consumed by **this process's own threads**.
//! - **`child_cpu_ms`** — CPU time of **child processes reaped during the
//! window** (`cutime`+`cstime`). Kept as its own field on purpose: see the
//! caveat below.
//! - **peak RSS** — the max `VmHWM` seen in `/proc/self/status` (MB).
//! - **mem%** — peak `(MemTotal-MemAvailable)/MemTotal` from `/proc/meminfo`.
//! - **elapsed_ms** — wall time from `start` to `stop`.
//!
//! ### What `proc_*` does NOT include
//! `cutime`/`cstime` are only credited to us when a child is **reaped**
//! (`wait`). So an arm whose work happens in a child process that is still
//! running at `stop()` — a long-lived spawned server, a `git` daemon left up —
//! contributes **nothing**, and `proc_cores_avg` UNDER-reports exactly as badly
//! as `/proc/stat` over-reports. Detached/orphaned children are never counted at
//! all. An arm that spawns-and-waits (the common shape: `Command::status()`,
//! `output()`) IS counted, via `child_cpu_ms`. When `child_cpu_ms` is a large
//! share of the total, read the number as "process tree, as reaped", not "this
//! process".
//!
//! Symmetrically, the scope is the whole **process**, not the bench's threads:
//! anything else running in the same process during the window (a background
//! worker, another test in the same test binary) is charged to it. The bench
//! harness runs its arms one at a time in one process, so in production this is
//! exactly the number wanted; in a multi-threaded `cargo test` it is not, which
//! is why the tests below serialize.
//!
//! ### Units too short to measure
//! The tick counters have `1/USER_HZ` (usually 10 ms) granularity, so over a
//! few-millisecond unit the CPU delta can only be 0 or 1 tick — "0 cores" or
//! "4 cores", neither of them a measurement. Below 20 ticks of wall time the
//! `proc_*` fields are `None` rather than a number with several hundred percent
//! of quantisation error. (Seen for real: `autonom.function_scan` in preview
//! mode, 2.4 ms wall, reported `0.00` cores — which a gate would have read as a
//! hard RED.)
//!
//! Non-Linux (no `/proc`) degrades gracefully: the reads fail, so the host-wide
//! numbers are zero and every `proc_*` field is `None` — "not measured", which
//! is deliberately distinct from a measured `0.0` so a gate can tell "no data"
//! from "used no cores". The sampler never panics and never affects the result.
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
use std::sync::OnceLock;
use std::time::{Duration, Instant};
/// The per-core utilisation threshold (fraction, 0..1) above which a core is
/// counted "busy" for the cores-busy metric.
const BUSY_THRESHOLD: f64 = 0.50;
/// Sampling period. 1 Hz keeps overhead negligible while still resolving a
/// multi-second bench's load profile.
const SAMPLE_PERIOD: Duration = Duration::from_millis(1000);
/// Summary telemetry for one bench unit. All values are best-effort; on a host
/// without `/proc` the host-wide numbers are zero and the `proc_*` ones `None`.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct Telemetry {
/// Number of logical cores the sampler observed in `/proc/stat`.
pub n_cores: u32,
/// **HOST-WIDE.** Aggregate CPU utilisation of the whole box across all
/// cores, percent (0..100) — every process on the machine, not just this
/// one. Ambient context; never a statement about the bench.
pub cpu_pct_avg: f64,
pub cpu_pct_max: f64,
/// **HOST-WIDE.** How many of the machine's cores were >50% busy *for any
/// reason*, averaged over the samples / peak. On a shared box this is the
/// neighbours' load. Use [`Telemetry::proc_cores_max`] to judge the bench.
pub cores_busy_avg: f64,
pub cores_busy_max: u32,
/// Peak resident set size of *this* process, in MB.
pub mem_peak_mb: f64,
/// Peak system memory utilisation, percent (0..100).
pub mem_pct_max: f64,
/// Wall-clock duration of the sampled unit.
pub elapsed_ms: f64,
/// How many interval samples contributed to the averages.
pub samples: u32,
// ── PROCESS-SCOPED (host-independent). `None` = not measured. ──────────
/// CPU time consumed by **this process's own threads** over the window
/// (`utime`+`stime` delta), in ms. Excludes child processes.
pub proc_cpu_ms: Option<f64>,
/// CPU time of **child processes reaped during the window**
/// (`cutime`+`cstime` delta), in ms. Zero for an arm that spawns nothing —
/// and also zero for one whose child is still alive at `stop()`.
pub child_cpu_ms: Option<f64>,
/// **The saturation number: task-clock ÷ wall.** Mean cores occupied by
/// this process (+ its reaped children) over the whole window.
/// `(proc_cpu_ms + child_cpu_ms) / elapsed_ms`. 1.0 = one core.
pub proc_cores_avg: Option<f64>,
/// Peak of the same ratio over the 1 Hz sampling intervals. Never below
/// `proc_cores_avg` (a mean is a lower bound on a peak), so a sub-second
/// unit that produced no interval still reports its true occupancy rather
/// than a blind zero.
pub proc_cores_max: Option<f64>,
// ── THERMAL + ACHIEVED CLOCK. `None` = not measured on this box. ───────
/// Which hwmon sensor the temperature came from, e.g. `k10temp/Tctl` or
/// `coretemp/Package id 0`. Carried so two runs' temperatures are only ever
/// compared when they came from the same sensor.
pub temp_sensor: Option<String>,
/// Mean CPU package temperature over the sampled intervals, °C.
pub temp_c_avg: Option<f64>,
/// Peak CPU package temperature over the sampled intervals, °C.
pub temp_c_max: Option<f64>,
/// Mean ALL-CORE achieved clock over the sampled intervals, MHz — the mean
/// across every logical CPU of `scaling_cur_freq`, then averaged over the
/// samples. This is the clock the run actually got, not the nameplate one.
pub mhz_avg: Option<f64>,
/// Peak of the same all-core mean, MHz.
pub mhz_max: Option<f64>,
/// Minimum of the same all-core mean, MHz. A run whose `mhz_min` is far
/// below its `mhz_avg` spent part of its window clocked down.
pub mhz_min: Option<f64>,
/// How many times the kernel's per-core throttle counters advanced during
/// the window (summed over cores). `Some(0)` = the counters exist and did
/// not move — a positive statement of "no throttling". `None` = this box
/// has no such counters (AMD k10temp does not expose them), which is NOT
/// the same fact and must not be read as zero.
pub throttle_events: Option<u64>,
}
impl Telemetry {
/// Did the CPU throttle during this run? `None` when the box exposes no
/// throttle counters — deliberately tri-state, because "we could not tell"
/// and "it did not throttle" are different readings and a benchmark that
/// conflates them will publish a slow number as if it were a fast box.
pub fn throttled(&self) -> Option<bool> {
self.throttle_events.map(|n| n > 0)
}
}
/// One `/proc/stat` snapshot: aggregate (idle, total) jiffies plus the same per
/// core. Deltas between two snapshots give per-interval utilisation.
struct StatSnapshot {
/// (idle_jiffies, total_jiffies) for the aggregate `cpu` line.
agg: (u64, u64),
/// (idle, total) per logical core, in `cpu0..cpuN` order.
per_core: Vec<(u64, u64)>,
}
/// Live-updated, lock-free handle the sampler thread writes and the progress
/// emitter reads, so `[BENCH]` lines can show the current occupancy.
#[derive(Default)]
struct Live {
cores_busy_now: AtomicU32,
n_cores: AtomicU32,
/// This process's cores-occupied in the most recent interval, in
/// **centi-cores** (`cores * 100`) so it fits a lock-free integer atomic.
/// `u64::MAX` is the "no interval yet" sentinel.
proc_centicores_now: AtomicU64,
}
/// A running telemetry sampler. Construct with [`Sampler::start`], then
/// [`Sampler::stop`] to join and collect the [`Telemetry`].
pub struct Sampler {
stop: Arc<AtomicBool>,
live: Arc<Live>,
handle: Option<std::thread::JoinHandle<Telemetry>>,
start: Instant,
/// `/proc/self/stat` CPU counters at `start()`. The whole-window average is
/// computed from this against a second read in `stop()` — on the CALLING
/// thread, not from the interval loop — so a sub-second unit that never
/// completes a sampling interval still gets an exact task-clock ÷ wall
/// instead of a zero.
cpu_at_start: Option<SelfCpu>,
}
impl Sampler {
/// Start sampling on a background thread. Cheap: spawns one thread that
/// sleeps most of the time.
pub fn start() -> Self {
let stop = Arc::new(AtomicBool::new(false));
let live = Arc::new(Live::default());
live.proc_centicores_now.store(u64::MAX, Ordering::Relaxed);
// Read our own CPU counters BEFORE the clock starts, so the window the
// ratio divides by is never shorter than the CPU time it counts.
let cpu_at_start = read_self_cpu();
let start = Instant::now();
let stop_t = Arc::clone(&stop);
let live_t = Arc::clone(&live);
let handle = std::thread::Builder::new()
.name("bench-telemetry".into())
.spawn(move || sample_loop(stop_t, live_t, start))
.ok();
Sampler { stop, live, handle, start, cpu_at_start }
}
/// **HOST-WIDE.** Count of the machine's cores observed >50% busy in the
/// most recent interval — by any process. Lock-free read for the live
/// progress line. 0 until the first interval completes (~1s in).
pub fn cores_busy_now(&self) -> u32 {
self.live.cores_busy_now.load(Ordering::Relaxed)
}
/// **PROCESS-SCOPED.** Cores this process occupied in the most recent
/// interval (task-clock ÷ wall). `None` until the first interval completes.
pub fn proc_cores_now(&self) -> Option<f64> {
match self.live.proc_centicores_now.load(Ordering::Relaxed) {
u64::MAX => None,
c => Some(c as f64 / 100.0),
}
}
/// Number of logical cores the sampler has seen (0 until first sample).
pub fn n_cores(&self) -> u32 {
self.live.n_cores.load(Ordering::Relaxed)
}
/// Stop sampling, join the thread, and return the summary. Always sets
/// `elapsed_ms` from the wall clock even if the thread never produced a
/// full interval (sub-second bench), and computes the whole-window
/// task-clock ÷ wall from the `start()`/`stop()` pair of
/// `/proc/self/stat` reads.
pub fn stop(mut self) -> Telemetry {
self.stop.store(true, Ordering::Relaxed);
let cpu_at_stop = read_self_cpu();
let elapsed_ms = self.start.elapsed().as_secs_f64() * 1000.0;
let mut t = self
.handle
.take()
.and_then(|h| h.join().ok())
.unwrap_or_default();
t.elapsed_ms = elapsed_ms;
let hz = clock_ticks_per_sec();
// TICK GRANULARITY. `/proc/self/stat` counts in clock ticks — 10 ms at
// the usual `USER_HZ` of 100. Over a 2.4 ms unit the CPU delta can only
// be 0 or 1 tick, i.e. "0 cores" or "4 cores", and neither is a
// measurement. Observed for real: `autonom.function_scan` in preview
// mode (2.4 ms) reported `proc_cores_avg = 0.00` — a number a gate
// would have read as a hard RED. Below `MIN_RESOLVABLE_TICKS` of wall
// the answer is `None`, "too short to measure", not a fabricated zero.
const MIN_RESOLVABLE_TICKS: f64 = 20.0;
let resolvable = elapsed_ms >= MIN_RESOLVABLE_TICKS / hz * 1000.0;
if let (Some(a), Some(b), true) = (self.cpu_at_start, cpu_at_stop, resolvable) {
let proc_ms = b.self_ticks.saturating_sub(a.self_ticks) as f64 / hz * 1000.0;
let child_ms = b.child_ticks.saturating_sub(a.child_ticks) as f64 / hz * 1000.0;
t.proc_cpu_ms = Some(proc_ms);
t.child_cpu_ms = Some(child_ms);
let avg = if elapsed_ms > 0.0 { (proc_ms + child_ms) / elapsed_ms } else { 0.0 };
t.proc_cores_avg = Some(avg);
// A mean is a lower bound on a peak, so clamping up is sound — and
// it is what keeps a sub-second unit (zero intervals sampled) from
// publishing a blind `proc_cores_max = 0`.
t.proc_cores_max = Some(t.proc_cores_max.unwrap_or(0.0).max(avg));
} else {
// No `/proc/self/stat`, or a window too short for the tick counters
// to resolve: say "not measured", never a fake 0.0.
t.proc_cpu_ms = None;
t.child_cpu_ms = None;
t.proc_cores_avg = None;
t.proc_cores_max = None;
}
t
}
}
/// The `/proc/self/stat` CPU counters we care about, in clock ticks.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct SelfCpu {
/// `utime` + `stime` — this process (all its threads).
self_ticks: u64,
/// `cutime` + `cstime` — children **already reaped**. See the module note.
child_ticks: u64,
}
/// `sysconf(_SC_CLK_TCK)` — the kernel's `USER_HZ`, the unit every tick counter
/// in `/proc` is expressed in. Queried, never assumed to be 100: that constant
/// is right on mainstream Linux and wrong on e.g. alpha (1024), and a hardcoded
/// divisor is precisely the kind of silently-wrong number this module exists to
/// stop shipping.
fn clock_ticks_per_sec() -> f64 {
static HZ: OnceLock<f64> = OnceLock::new();
*HZ.get_or_init(|| {
// SAFETY: `sysconf` is a pure query with no arguments to get wrong.
let hz = unsafe { libc::sysconf(libc::_SC_CLK_TCK) };
if hz > 0 { hz as f64 } else { 100.0 }
})
}
/// Parse `utime`/`stime`/`cutime`/`cstime` (fields 14–17, 1-based) out of
/// `/proc/self/stat`.
///
/// Field 2 (`comm`) is the executable name wrapped in parentheses and may itself
/// contain spaces *and* `)` — so the fields are indexed from **after the last
/// `)`**, never by splitting the whole line on whitespace. Splitting naively is
/// the classic `/proc/self/stat` bug and would mis-attribute CPU time for any
/// binary whose name contains a space.
fn read_self_cpu() -> Option<SelfCpu> {
parse_self_stat(&std::fs::read_to_string("/proc/self/stat").ok()?)
}
/// The pure half of [`read_self_cpu`], so the parser can be tested against
/// hostile `comm` values without a matching process on the box.
fn parse_self_stat(text: &str) -> Option<SelfCpu> {
let rest = &text[text.rfind(')')? + 1..];
// `rest` begins at field 3 (`state`), so field N lives at index N-3.
let f: Vec<&str> = rest.split_whitespace().collect();
let at = |n: usize| -> Option<&str> { f.get(n - 3).copied() };
let utime: u64 = at(14)?.parse().ok()?;
let stime: u64 = at(15)?.parse().ok()?;
// cutime/cstime are signed in the kernel's own docs; clamp rather than fail.
let cutime: i64 = at(16)?.parse().ok()?;
let cstime: i64 = at(17)?.parse().ok()?;
Some(SelfCpu {
self_ticks: utime + stime,
child_ticks: (cutime.max(0) as u64) + (cstime.max(0) as u64),
})
}
/// The background loop. Reads an initial `/proc/stat` baseline, then every
/// `SAMPLE_PERIOD` reads again, folds the interval's utilisation into the
/// accumulators, and publishes the live cores-busy count. Exits promptly when
/// `stop` is set (checked at sub-period granularity so `stop()` doesn't block a
/// whole second).
fn sample_loop(stop: Arc<AtomicBool>, live: Arc<Live>, _start: Instant) -> Telemetry {
let mut acc = Telemetry::default();
let mut prev = read_proc_stat();
if let Some(p) = &prev {
let n = p.per_core.len() as u32;
acc.n_cores = n;
live.n_cores.store(n, Ordering::Relaxed);
}
let mut cpu_sum = 0.0_f64;
let mut busy_sum = 0.0_f64;
// Thermal + achieved clock. Sampled on the SAME thread, in the SAME loop,
// at the SAME 1 Hz cadence as everything else — one mechanism, not two. The
// sensor is resolved once (a directory walk) and then read as one file per
// interval; the clock is one file per logical CPU.
let sensor = ThermalSensor::find();
if let Some(s) = &sensor {
acc.temp_sensor = Some(s.label.clone());
}
let mut temp_sum = 0.0_f64;
let mut temp_n = 0u32;
let mut mhz_sum = 0.0_f64;
let mut mhz_n = 0u32;
let throttle_at_start = read_throttle_count();
// Process-scoped interval tracking: our own CPU counters plus the wall
// instant they were read at, so each interval's task-clock ÷ wall uses the
// interval it actually covers (sleep slices are approximate).
let mut prev_self = read_self_cpu();
let mut prev_at = Instant::now();
loop {
// Sleep in small slices so a stop request is honoured quickly.
let mut waited = Duration::ZERO;
while waited < SAMPLE_PERIOD {
if stop.load(Ordering::Relaxed) {
break;
}
let slice = Duration::from_millis(50);
std::thread::sleep(slice);
waited += slice;
}
let stopping = stop.load(Ordering::Relaxed);
if let (Some(p), Some(cur)) = (prev.as_ref(), read_proc_stat()) {
if let Some((cpu_pct, cores_busy)) = interval_utilisation(p, &cur) {
acc.samples += 1;
cpu_sum += cpu_pct;
busy_sum += cores_busy as f64;
acc.cpu_pct_max = acc.cpu_pct_max.max(cpu_pct);
acc.cores_busy_max = acc.cores_busy_max.max(cores_busy);
live.cores_busy_now.store(cores_busy, Ordering::Relaxed);
}
prev = Some(cur);
}
// PROCESS-SCOPED peak. Only intervals long enough to be meaningful
// count: a 50 ms tail slice against a 10 ms tick granularity can
// manufacture a spike of several cores out of one rounded tick.
let now = Instant::now();
let dt = now.duration_since(prev_at).as_secs_f64();
if let (Some(a), Some(b)) = (prev_self, read_self_cpu()) {
if dt >= 0.2 {
let hz = clock_ticks_per_sec();
let ticks = b.self_ticks.saturating_sub(a.self_ticks)
+ b.child_ticks.saturating_sub(a.child_ticks);
let cores = (ticks as f64 / hz) / dt;
acc.proc_cores_max = Some(acc.proc_cores_max.unwrap_or(0.0).max(cores));
live.proc_centicores_now
.store((cores * 100.0).round().max(0.0) as u64, Ordering::Relaxed);
prev_self = Some(b);
prev_at = now;
}
} else {
prev_self = read_self_cpu();
prev_at = now;
}
// Memory: peak RSS of this process + peak system mem%.
if let Some(rss_mb) = read_self_peak_rss_mb() {
acc.mem_peak_mb = acc.mem_peak_mb.max(rss_mb);
}
if let Some(mem_pct) = read_system_mem_pct() {
acc.mem_pct_max = acc.mem_pct_max.max(mem_pct);
}
// Temperature: one file read from the sensor resolved at loop entry.
if let Some(c) = sensor.as_ref().and_then(|s| s.read_c()) {
temp_sum += c;
temp_n += 1;
acc.temp_c_max = Some(acc.temp_c_max.unwrap_or(f64::MIN).max(c));
}
// Achieved clock: the mean of every logical CPU's current frequency.
if let Some(mhz) = read_all_core_mhz() {
mhz_sum += mhz;
mhz_n += 1;
acc.mhz_max = Some(acc.mhz_max.unwrap_or(f64::MIN).max(mhz));
acc.mhz_min = Some(acc.mhz_min.unwrap_or(f64::MAX).min(mhz));
}
if stopping {
break;
}
}
if acc.samples > 0 {
acc.cpu_pct_avg = cpu_sum / acc.samples as f64;
acc.cores_busy_avg = busy_sum / acc.samples as f64;
}
if temp_n > 0 {
acc.temp_c_avg = Some(temp_sum / temp_n as f64);
}
if mhz_n > 0 {
acc.mhz_avg = Some(mhz_sum / mhz_n as f64);
}
// `Some(0)` is the positive statement "the counters exist and did not
// move"; `None` is "this box has no counters". Only a box that had them at
// the start can answer at all.
acc.throttle_events = match (throttle_at_start, read_throttle_count()) {
(Some(a), Some(b)) => Some(b.saturating_sub(a)),
_ => None,
};
acc
}
/// A resolved hwmon temperature input: the file to read and the human label of
/// where it comes from (`k10temp/Tctl`).
///
/// Resolved ONCE per run, by walking `/sys/class/hwmon`. The walk is the
/// expensive part and it must not happen at 1 Hz.
struct ThermalSensor {
path: std::path::PathBuf,
label: String,
}
/// Driver names we accept as "the CPU package", best first. Anything else under
/// `/sys/class/hwmon` (nvme, iwlwifi, the battery) is a different thing and
/// reporting it as the CPU temperature would be a wrong number, not a missing
/// one.
const CPU_HWMON_DRIVERS: &[&str] = &["k10temp", "coretemp", "zenpower", "cpu_thermal"];
/// Within an accepted driver, the input labels that mean "the package", best
/// first. `Tctl` is AMD's control temperature (what `sensors` shows for Ryzen);
/// `Package id 0` is Intel's.
const CPU_TEMP_LABELS: &[&str] = &["Tctl", "Package id 0", "Tdie"];
impl ThermalSensor {
/// Walk `/sys/class/hwmon`, pick the first CPU-package input. Pure file
/// reads — no `sensors(1)`, which the law forbids and which is why these
/// numbers had to be logged by hand alongside a run until now.
fn find() -> Option<ThermalSensor> {
Self::find_under(std::path::Path::new("/sys/class/hwmon"))
}
fn find_under(root: &std::path::Path) -> Option<ThermalSensor> {
let mut dirs: Vec<std::path::PathBuf> = std::fs::read_dir(root)
.ok()?
.filter_map(|e| e.ok())
.map(|e| e.path())
.collect();
dirs.sort();
// Two passes so driver preference beats directory order: a box with
// both k10temp and cpu_thermal must not be decided by hwmon numbering.
for want in CPU_HWMON_DRIVERS {
for d in &dirs {
let name = std::fs::read_to_string(d.join("name"))
.ok()
.map(|s| s.trim().to_string())
.unwrap_or_default();
if name != *want {
continue;
}
if let Some(s) = Self::pick_input(d, &name) {
return Some(s);
}
}
}
None
}
/// Prefer a labelled package input; fall back to `temp1_input`, which is
/// the package on every driver in [`CPU_HWMON_DRIVERS`].
fn pick_input(dir: &std::path::Path, driver: &str) -> Option<ThermalSensor> {
for want in CPU_TEMP_LABELS {
for i in 1..=16u32 {
let lbl = dir.join(format!("temp{i}_label"));
let Ok(l) = std::fs::read_to_string(&lbl) else {
continue;
};
if l.trim() != *want {
continue;
}
let input = dir.join(format!("temp{i}_input"));
if input.exists() {
return Some(ThermalSensor {
path: input,
label: format!("{driver}/{want}"),
});
}
}
}
let input = dir.join("temp1_input");
input.exists().then(|| ThermalSensor {
path: input,
label: format!("{driver}/temp1"),
})
}
/// hwmon reports millidegrees Celsius.
fn read_c(&self) -> Option<f64> {
let raw: f64 = std::fs::read_to_string(&self.path).ok()?.trim().parse().ok()?;
Some(raw / 1000.0)
}
}
/// Mean of every logical CPU's `scaling_cur_freq` (kHz in sysfs), in MHz.
///
/// `scaling_cur_freq` is what the governor last set; on `acpi-cpufreq`/`amd-pstate`
/// it tracks the achieved clock closely enough to show a throttle. Falls back
/// to `/proc/cpuinfo`'s `cpu MHz` lines when cpufreq is absent (a VM, or a
/// kernel without the driver) — which is the file today's runs had to be read
/// by hand.
fn read_all_core_mhz() -> Option<f64> {
if let Some(v) = read_cpufreq_mhz(std::path::Path::new("/sys/devices/system/cpu")) {
return Some(v);
}
let s = std::fs::read_to_string("/proc/cpuinfo").ok()?;
mean_cpuinfo_mhz(&s)
}
fn read_cpufreq_mhz(cpu_root: &std::path::Path) -> Option<f64> {
let entries = std::fs::read_dir(cpu_root).ok()?;
let mut sum = 0.0_f64;
let mut n = 0u32;
for e in entries.filter_map(|e| e.ok()) {
let name = e.file_name().to_string_lossy().into_owned();
let Some(rest) = name.strip_prefix("cpu") else {
continue;
};
if rest.is_empty() || !rest.chars().all(|c| c.is_ascii_digit()) {
continue;
}
let p = e.path().join("cpufreq/scaling_cur_freq");
if let Ok(khz) = std::fs::read_to_string(&p) {
if let Ok(khz) = khz.trim().parse::<f64>() {
sum += khz / 1000.0;
n += 1;
}
}
}
(n > 0).then(|| sum / n as f64)
}
/// Mean of the `cpu MHz` lines in a `/proc/cpuinfo` body.
fn mean_cpuinfo_mhz(s: &str) -> Option<f64> {
let mut sum = 0.0_f64;
let mut n = 0u32;
for line in s.lines() {
if let Some((k, v)) = line.split_once(':') {
if k.trim() == "cpu MHz" {
if let Ok(f) = v.trim().parse::<f64>() {
sum += f;
n += 1;
}
}
}
}
(n > 0).then(|| sum / n as f64)
}
/// Sum of every core's `thermal_throttle/core_throttle_count`. `None` when no
/// core exposes one — Intel does, AMD's k10temp does not.
fn read_throttle_count() -> Option<u64> {
read_throttle_count_under(std::path::Path::new("/sys/devices/system/cpu"))
}
fn read_throttle_count_under(cpu_root: &std::path::Path) -> Option<u64> {
let entries = std::fs::read_dir(cpu_root).ok()?;
let mut total = 0u64;
let mut any = false;
for e in entries.filter_map(|e| e.ok()) {
let name = e.file_name().to_string_lossy().into_owned();
let Some(rest) = name.strip_prefix("cpu") else {
continue;
};
if rest.is_empty() || !rest.chars().all(|c| c.is_ascii_digit()) {
continue;
}
let p = e.path().join("thermal_throttle/core_throttle_count");
if let Ok(v) = std::fs::read_to_string(&p) {
if let Ok(v) = v.trim().parse::<u64>() {
total += v;
any = true;
}
}
}
any.then_some(total)
}
/// Compute (aggregate CPU%, cores-busy) for the interval between two snapshots.
/// Returns `None` if the total jiffy delta is zero (no time passed / read
/// glitch) so a degenerate interval doesn't pollute the averages.
fn interval_utilisation(prev: &StatSnapshot, cur: &StatSnapshot) -> Option<(f64, u32)> {
let (pi, pt) = prev.agg;
let (ci, ct) = cur.agg;
let total_d = ct.saturating_sub(pt);
if total_d == 0 {
return None;
}
let idle_d = ci.saturating_sub(pi);
let cpu_pct = (1.0 - (idle_d as f64 / total_d as f64)) * 100.0;
let cpu_pct = cpu_pct.clamp(0.0, 100.0);
let mut busy = 0u32;
let n = prev.per_core.len().min(cur.per_core.len());
for i in 0..n {
let (cpi, cpt) = prev.per_core[i];
let (cci, cct) = cur.per_core[i];
let td = cct.saturating_sub(cpt);
if td == 0 {
continue;
}
let id = cci.saturating_sub(cpi);
let util = 1.0 - (id as f64 / td as f64);
if util > BUSY_THRESHOLD {
busy += 1;
}
}
Some((cpu_pct, busy))
}
/// Parse `/proc/stat`'s `cpu` (aggregate) and `cpuN` (per-core) lines into a
/// [`StatSnapshot`]. Each line is
/// `cpu user nice system idle iowait irq softirq steal guest guest_nice`;
/// idle = field 4 (`idle`) + field 5 (`iowait`); total = sum of all fields.
/// Returns `None` on non-Linux / read error.
fn read_proc_stat() -> Option<StatSnapshot> {
let text = std::fs::read_to_string("/proc/stat").ok()?;
let mut agg = None;
let mut per_core = Vec::new();
for line in text.lines() {
if !line.starts_with("cpu") {
// /proc/stat lists all cpu lines first; once we leave them we're done.
if agg.is_some() {
break;
}
continue;
}
let mut it = line.split_whitespace();
let label = it.next()?; // "cpu" or "cpuN"
let nums: Vec<u64> = it.filter_map(|f| f.parse::<u64>().ok()).collect();
if nums.len() < 5 {
continue;
}
let total: u64 = nums.iter().sum();
// idle + iowait
let idle = nums[3] + nums[4];
if label == "cpu" {
agg = Some((idle, total));
} else {
per_core.push((idle, total));
}
}
let agg = agg?;
Some(StatSnapshot { agg, per_core })
}
/// Peak resident set (`VmHWM`) of the current process, in MB, from
/// `/proc/self/status`. `VmHWM` is a high-water mark the kernel maintains, so a
/// 1 Hz sampler never misses a transient peak between reads.
fn read_self_peak_rss_mb() -> Option<f64> {
let text = std::fs::read_to_string("/proc/self/status").ok()?;
for line in text.lines() {
if let Some(rest) = line.strip_prefix("VmHWM:") {
// `VmHWM: 123456 kB`
let kb: f64 = rest.split_whitespace().next()?.parse().ok()?;
return Some(kb / 1024.0);
}
}
None
}
/// System memory utilisation percent from `/proc/meminfo`:
/// `(MemTotal - MemAvailable) / MemTotal * 100`.
fn read_system_mem_pct() -> Option<f64> {
let text = std::fs::read_to_string("/proc/meminfo").ok()?;
let mut total = None;
let mut avail = None;
for line in text.lines() {
if let Some(rest) = line.strip_prefix("MemTotal:") {
total = rest.split_whitespace().next().and_then(|v| v.parse::<f64>().ok());
} else if let Some(rest) = line.strip_prefix("MemAvailable:") {
avail = rest.split_whitespace().next().and_then(|v| v.parse::<f64>().ok());
}
if total.is_some() && avail.is_some() {
break;
}
}
let (t, a) = (total?, avail?);
if t <= 0.0 {
return None;
}
Some(((t - a) / t * 100.0).clamp(0.0, 100.0))
}
/// Reserved metric-key prefix under which a unit's telemetry rides inside
/// [`crate::BenchResult::metrics`]. The harness writes these so the
/// numbers survive the bench-binary → nornir stdout-JSON hop; the warehouse
/// writer then lifts them out into the dedicated `bench_telemetry` table.
pub const TELEM_PREFIX: &str = "telem_";
/// Metric key carrying the **process-scoped** peak occupancy (task-clock ÷
/// wall, peak over the sampling intervals). Named as a constant because it is
/// the one telemetry key a release gate is allowed to judge a bench on, and
/// because its ABSENCE is meaningful: rows written before 2026-08-03 have no
/// process-scoped measurement at all and must be skipped, not read as zero.
pub const PROC_CORES_MAX_KEY: &str = "telem_proc_cores_max";
/// Fold a [`Telemetry`] into a result's metric map under the [`TELEM_PREFIX`]
/// keys, so it travels with the [`crate::BenchResult`] through JSON and
/// into the warehouse.
///
/// The host-wide keys (`cpu_pct_*`, `cores_busy_*`) keep the exact names and
/// meaning they have always had, so every `bench_history.jsonl` row ever
/// written stays readable and comparable. The process-scoped truth arrives as
/// NEW keys (`proc_cpu_ms`, `child_cpu_ms`, `proc_cores_avg`,
/// `proc_cores_max`) rather than by redefining a column older rows filled with
/// something else — a silent redefinition would make old and new rows look
/// alike while meaning different things. `None` (not measured) is written as
/// no key at all, never as `0`.
pub fn inject_into_metrics(
metrics: &mut serde_json::Map<String, serde_json::Value>,
t: &Telemetry,
) {
let j = |f: f64| serde_json::json!(f);
metrics.insert(format!("{TELEM_PREFIX}cpu_pct_avg"), j(round2(t.cpu_pct_avg)));
metrics.insert(format!("{TELEM_PREFIX}cpu_pct_max"), j(round2(t.cpu_pct_max)));
metrics.insert(format!("{TELEM_PREFIX}cores_busy_avg"), j(round2(t.cores_busy_avg)));
metrics.insert(format!("{TELEM_PREFIX}cores_busy_max"), serde_json::json!(t.cores_busy_max));
metrics.insert(format!("{TELEM_PREFIX}n_cores"), serde_json::json!(t.n_cores));
metrics.insert(format!("{TELEM_PREFIX}mem_peak_mb"), j(round2(t.mem_peak_mb)));
metrics.insert(format!("{TELEM_PREFIX}mem_pct_max"), j(round2(t.mem_pct_max)));
metrics.insert(format!("{TELEM_PREFIX}elapsed_ms"), j(round2(t.elapsed_ms)));
// Process-scoped. Absent when unmeasurable — a missing key is honest, a
// zero is a lie a gate would read as "used no cores".
if let Some(v) = t.proc_cpu_ms {
metrics.insert(format!("{TELEM_PREFIX}proc_cpu_ms"), j(round2(v)));
}
if let Some(v) = t.child_cpu_ms {
metrics.insert(format!("{TELEM_PREFIX}child_cpu_ms"), j(round2(v)));
}
if let Some(v) = t.proc_cores_avg {
metrics.insert(format!("{TELEM_PREFIX}proc_cores_avg"), j(round2(v)));
}
if let Some(v) = t.proc_cores_max {
metrics.insert(PROC_CORES_MAX_KEY.to_string(), j(round2(v)));
}
// Thermal + achieved clock. Same rule: a value that was not measured gets
// NO key. A missing `temp_c_max` means "this box has no CPU hwmon sensor";
// a `0.0` would mean the CPU was at freezing point.
if let Some(v) = &t.temp_sensor {
metrics.insert(
format!("{TELEM_PREFIX}temp_sensor"),
serde_json::Value::String(v.clone()),
);
}
if let Some(v) = t.temp_c_avg {
metrics.insert(format!("{TELEM_PREFIX}temp_c_avg"), j(round2(v)));
}
if let Some(v) = t.temp_c_max {
metrics.insert(format!("{TELEM_PREFIX}temp_c_max"), j(round2(v)));
}
if let Some(v) = t.mhz_avg {
metrics.insert(format!("{TELEM_PREFIX}mhz_avg"), j(round2(v)));
}
if let Some(v) = t.mhz_max {
metrics.insert(format!("{TELEM_PREFIX}mhz_max"), j(round2(v)));
}
if let Some(v) = t.mhz_min {
metrics.insert(format!("{TELEM_PREFIX}mhz_min"), j(round2(v)));
}
if let Some(v) = t.throttle_events {
metrics.insert(format!("{TELEM_PREFIX}throttle_events"), serde_json::json!(v));
}
}
/// Reconstruct a [`Telemetry`] from a result's metric map (the inverse of
/// [`inject_into_metrics`]). Returns `None` if the result carries no telemetry
/// keys (e.g. a legacy run, or a `_st` row we chose not to sample). The
/// warehouse writer uses this to populate `bench_telemetry`.
pub fn from_metrics(metrics: &serde_json::Map<String, serde_json::Value>) -> Option<Telemetry> {
let get = |k: &str| -> Option<f64> {
metrics.get(&format!("{TELEM_PREFIX}{k}")).and_then(|v| v.as_f64())
};
// Require at least one telemetry key to be present.
if !metrics.keys().any(|k| k.starts_with(TELEM_PREFIX)) {
return None;
}
Some(Telemetry {
n_cores: get("n_cores").unwrap_or(0.0) as u32,
cpu_pct_avg: get("cpu_pct_avg").unwrap_or(0.0),
cpu_pct_max: get("cpu_pct_max").unwrap_or(0.0),
cores_busy_avg: get("cores_busy_avg").unwrap_or(0.0),
cores_busy_max: get("cores_busy_max").unwrap_or(0.0) as u32,
mem_peak_mb: get("mem_peak_mb").unwrap_or(0.0),
mem_pct_max: get("mem_pct_max").unwrap_or(0.0),
elapsed_ms: get("elapsed_ms").unwrap_or(0.0),
samples: 0,
// Stay `None` on a legacy row: these keys did not exist before
// 2026-08-03, and `unwrap_or(0.0)` here would turn "we never measured
// this" into "this bench used zero cores" — a RED a gate cannot
// distinguish from a real one.
proc_cpu_ms: get("proc_cpu_ms"),
child_cpu_ms: get("child_cpu_ms"),
proc_cores_avg: get("proc_cores_avg"),
proc_cores_max: get("proc_cores_max"),
// Thermal/clock keys did not exist before 2026-09-20. Absent stays
// `None` for the same reason as the process-scoped family above.
temp_sensor: metrics
.get(&format!("{TELEM_PREFIX}temp_sensor"))
.and_then(|v| v.as_str())
.map(str::to_string),
temp_c_avg: get("temp_c_avg"),
temp_c_max: get("temp_c_max"),
mhz_avg: get("mhz_avg"),
mhz_max: get("mhz_max"),
mhz_min: get("mhz_min"),
throttle_events: metrics
.get(&format!("{TELEM_PREFIX}throttle_events"))
.and_then(|v| v.as_u64()),
})
}
fn round2(f: f64) -> f64 {
(f * 100.0).round() / 100.0
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn proc_stat_parses_aggregate_and_cores() {
// The parser runs against the real /proc on Linux; assert it sees the
// aggregate line and at least one core. On non-Linux this is None and we
// skip (the sampler degrades gracefully there).
if let Some(snap) = read_proc_stat() {
assert!(snap.agg.1 > 0, "total jiffies should be > 0");
assert!(!snap.per_core.is_empty(), "expected at least one cpuN line");
}
}
#[test]
fn interval_utilisation_counts_busy_cores() {
// Synthetic two snapshots: core 0 fully busy (no idle delta), core 1
// fully idle (all delta is idle). Expect cpu% ~50 aggregate, 1 busy core.
let prev = StatSnapshot {
agg: (100, 200),
per_core: vec![(50, 100), (50, 100)],
};
let cur = StatSnapshot {
// aggregate: +100 idle, +200 total → 50% busy
agg: (200, 400),
// core0: +0 idle, +100 total → 100% busy
// core1: +100 idle, +100 total → 0% busy
per_core: vec![(50, 200), (150, 200)],
};
let (cpu_pct, busy) = interval_utilisation(&prev, &cur).unwrap();
assert!((cpu_pct - 50.0).abs() < 0.01, "cpu_pct={cpu_pct}");
assert_eq!(busy, 1, "exactly one core should be >50% busy");
}
#[test]
fn zero_interval_is_ignored() {
let snap = StatSnapshot { agg: (10, 20), per_core: vec![(5, 10)] };
let same = StatSnapshot { agg: (10, 20), per_core: vec![(5, 10)] };
assert!(interval_utilisation(&snap, &same).is_none());
}
#[test]
fn metrics_roundtrip() {
let t = Telemetry {
n_cores: 8,
cpu_pct_avg: 73.25,
cpu_pct_max: 99.5,
cores_busy_avg: 5.5,
cores_busy_max: 8,
mem_peak_mb: 1234.5,
mem_pct_max: 42.0,
elapsed_ms: 1500.0,
samples: 3,
proc_cpu_ms: Some(4500.0),
child_cpu_ms: Some(120.0),
proc_cores_avg: Some(3.08),
proc_cores_max: Some(5.75),
temp_sensor: Some("k10temp/Tctl".into()),
temp_c_avg: Some(56.0),
temp_c_max: Some(64.1),
mhz_avg: Some(4167.0),
mhz_max: Some(4425.0),
mhz_min: Some(3900.0),
throttle_events: Some(0),
};
let mut m = serde_json::Map::new();
inject_into_metrics(&mut m, &t);
let back = from_metrics(&m).expect("telemetry keys present");
assert_eq!(back.temp_sensor.as_deref(), Some("k10temp/Tctl"));
assert!((back.temp_c_avg.unwrap() - 56.0).abs() < 0.01);
assert!((back.temp_c_max.unwrap() - 64.1).abs() < 0.01);
assert!((back.mhz_avg.unwrap() - 4167.0).abs() < 0.01);
assert!((back.mhz_min.unwrap() - 3900.0).abs() < 0.01);
// `Some(0)` is "the counters exist and did not move" — a positive
// no-throttling statement. It must survive the round trip as `false`,
// never collapse to `None`.
assert_eq!(back.throttled(), Some(false));
assert_eq!(back.n_cores, 8);
assert_eq!(back.cores_busy_max, 8);
assert!((back.cpu_pct_avg - 73.25).abs() < 0.01);
assert!((back.mem_peak_mb - 1234.5).abs() < 0.01);
assert!((back.proc_cores_max.unwrap() - 5.75).abs() < 0.01);
assert!((back.proc_cores_avg.unwrap() - 3.08).abs() < 0.01);
assert!((back.proc_cpu_ms.unwrap() - 4500.0).abs() < 0.01);
assert!((back.child_cpu_ms.unwrap() - 120.0).abs() < 0.01);
// No telemetry keys → None.
assert!(from_metrics(&serde_json::Map::new()).is_none());
}
/// HISTORY COMPAT: a row written before the process-scoped fields existed
/// must still parse, and must report the new fields as `None` — NOT as a
/// measured `0.0`, which a saturation gate would read as a real RED.
#[test]
fn legacy_row_without_proc_keys_reads_as_not_measured() {
let legacy: serde_json::Map<String, serde_json::Value> = serde_json::from_str(
r#"{"telem_n_cores":32,"telem_cpu_pct_avg":99.9,"telem_cpu_pct_max":99.95,
"telem_cores_busy_avg":31.5,"telem_cores_busy_max":32,
"telem_mem_peak_mb":900.0,"telem_mem_pct_max":40.0,
"telem_elapsed_ms":5000.0}"#,
)
.unwrap();
let t = from_metrics(&legacy).expect("legacy telemetry row must still parse");
assert_eq!(t.cores_busy_max, 32, "host-wide column keeps its old meaning");
assert_eq!(t.proc_cores_max, None, "absent ≠ zero");
assert_eq!(t.proc_cores_avg, None);
assert_eq!(t.proc_cpu_ms, None);
assert_eq!(t.child_cpu_ms, None);
}
/// `/proc/self/stat`'s `comm` field is `(name)` and the name may contain
/// spaces AND `)`. Splitting the whole line on whitespace shifts every
/// index and silently reports another field's number as CPU time.
#[test]
fn self_stat_parser_survives_a_hostile_comm() {
// Fields: 1 pid, 2 comm, 3 state, then 4..; 14=utime 15=stime
// 16=cutime 17=cstime.
let line = "1234 (evil ) name) S 1 1 1 0 -1 4194304 100 0 0 0 700 300 40 20 \
20 0 5 0 900 0 0 0 0";
let c = parse_self_stat(line).expect("hostile comm must parse");
assert_eq!(c.self_ticks, 1000, "utime 700 + stime 300");
assert_eq!(c.child_ticks, 60, "cutime 40 + cstime 20");
// The naive parse — split the WHOLE line — would land on other fields.
let naive: Vec<&str> = line.split_whitespace().collect();
assert_ne!(
naive[13].parse::<u64>().unwrap_or(0) + naive[14].parse::<u64>().unwrap_or(0),
c.self_ticks,
"test is only meaningful if the naive parse actually differs"
);
}
/// A unit too short for the tick counters to resolve must report "not
/// measured", not `0.00` cores. Without this, `autonom.function_scan`
/// (2.4 ms in preview mode) published a measured-looking zero that a
/// saturation gate reads as a hard RED.
#[test]
fn a_unit_shorter_than_the_tick_resolution_is_not_measured() {
let t = Sampler::start().stop(); // ~microseconds
assert!(t.elapsed_ms < 20.0 / clock_ticks_per_sec() * 1000.0, "precondition");
assert_eq!(t.proc_cores_avg, None, "elapsed={:.3}ms", t.elapsed_ms);
assert_eq!(t.proc_cores_max, None);
assert_eq!(t.proc_cpu_ms, None);
// …and a unit that IS long enough does get a number, so the guard above
// is not just "always None".
let long = spin(1, 400);
assert!(long.proc_cores_avg.is_some(), "a 400ms unit must be measurable");
}
#[test]
fn clk_tck_is_queried_and_sane() {
let hz = clock_ticks_per_sec();
assert!(hz > 0.0 && hz <= 10_000.0, "implausible CLK_TCK {hz}");
}
/// Spin `n` threads for `ms`, return the telemetry the sampler saw.
///
/// Takes [`crate::serial_guard`]: `/proc/self/stat` is **process**-scoped,
/// so a concurrently-running test that spins 32 threads in this same test
/// binary is charged to our window too. (Observed, not theorised: without
/// the guard the 1-thread arm below read **17.25 cores** on oden because
/// `sampler_runs_and_reports_on_a_busy_workload` was spinning alongside it.)
/// The bench harness itself runs its arms one at a time in one process, so
/// this is a test-harness artefact rather than a property of the metric —
/// but it is the same caveat as the child-process one: the number is
/// "everything this PROCESS did", no finer.
fn spin(n: usize, ms: u64) -> Telemetry {
let _g = crate::serial_guard();
let sampler = Sampler::start();
let stop = Arc::new(AtomicBool::new(false));
let mut handles = Vec::new();
for _ in 0..n {
let s = Arc::clone(&stop);
handles.push(std::thread::spawn(move || {
let mut x = 0u64;
while !s.load(Ordering::Relaxed) {
// Burn CPU; black-box so it isn't optimised away.
x = x.wrapping_mul(2654435761).wrapping_add(1);
std::hint::black_box(x);
}
}));
}
std::thread::sleep(Duration::from_millis(ms));
stop.store(true, Ordering::Relaxed);
for h in handles {
let _ = h.join();
}
sampler.stop()
}
/// This thread's own CPU time, in ms (`CLOCK_THREAD_CPUTIME_ID`).
fn thread_cpu_ms() -> f64 {
let mut ts = libc::timespec { tv_sec: 0, tv_nsec: 0 };
// SAFETY: `ts` is a valid, writable timespec; the clock id is a constant.
let rc = unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &mut ts) };
assert_eq!(rc, 0, "clock_gettime(CLOCK_THREAD_CPUTIME_ID)");
ts.tv_sec as f64 * 1000.0 + ts.tv_nsec as f64 / 1e6
}
/// Spin `n` threads until EACH has burned `cpu_ms` of its OWN CPU time, and
/// return the telemetry the sampler saw.
///
/// [`spin`] stops on the WALL clock, so how much CPU an arm gets depends on
/// the box: at oden load 100 to 400 (ODEN-CI round 1, 2026-10-25) its 4-thread
/// arm read 1.87 cores and went red. This one stops on each thread's own CPU
/// clock, so the work charged to the process is the same on an idle box and
/// a loaded one; only the wall time it takes moves.
fn spin_cpu(n: usize, cpu_ms: f64) -> Telemetry {
let _g = crate::serial_guard();
let sampler = Sampler::start();
let handles: Vec<_> = (0..n)
.map(|_| {
std::thread::spawn(move || {
let start = thread_cpu_ms();
let mut x = 0u64;
while thread_cpu_ms() - start < cpu_ms {
for _ in 0..10_000 {
x = x.wrapping_mul(2654435761).wrapping_add(1);
std::hint::black_box(x);
}
}
})
})
.collect();
for h in handles {
let _ = h.join();
}
sampler.stop()
}
#[test]
fn sampler_runs_and_reports_on_a_busy_workload() {
// INJECT-AND-ASSERT: start the sampler, spin a couple of threads in a
// tight loop for ~2.2s (long enough for ≥2 sampling intervals), then
// assert the sampler observed real cores and plausible non-zero load.
let n = std::thread::available_parallelism().map(|x| x.get()).unwrap_or(2).max(2);
let t = spin(n, 2200);
// On Linux we get real /proc data; assert it's plausible and non-zero.
if read_proc_stat().is_some() {
assert!(t.n_cores >= 1, "n_cores={}", t.n_cores);
assert!(t.samples >= 1, "expected ≥1 interval, got {}", t.samples);
assert!(t.cpu_pct_max > 0.0, "cpu_pct_max should be >0 under load");
assert!(
t.cores_busy_max >= 1,
"at least one core should be >50% busy under an N-thread spin, got {}",
t.cores_busy_max
);
assert!(t.mem_peak_mb > 0.0, "peak RSS should be >0");
}
assert!(t.elapsed_ms >= 2000.0, "elapsed_ms={}", t.elapsed_ms);
}
/// **THE GUARD THIS MODULE EXISTS FOR (LAW 2).**
///
/// A saturation metric earns its name only by telling a serial arm apart
/// from a parallel one. `/proc/stat`-derived `cores_busy_max` cannot: on a
/// loaded box it reports ~all cores for BOTH, which is how a "peak ≥ 0.9 ×
/// cores" gate came to pass forever on an arm with no parallelism at all.
///
/// So: run one thread, then run four, on the same box in the same test, and
/// require the process-scoped number to MOVE. One arm alone proves nothing —
/// the discrimination is the assertion.
#[test]
fn proc_cores_distinguishes_serial_from_parallel() {
if read_self_cpu().is_none() {
return; // no /proc/self/stat — nothing to assert
}
// A fixed amount of CPU work per thread, not a fixed wall time (see
// `spin_cpu`): the assertions below hold at any host load, and on a box
// with fewer than four cores too, because they read CPU time, not wall.
const BURN_MS: f64 = 600.0;
let one = spin_cpu(1, BURN_MS);
let four = spin_cpu(4, BURN_MS);
let one_cpu = one.proc_cpu_ms.expect("measured");
let four_cpu = four.proc_cpu_ms.expect("measured");
// One tick of slack per thread: /proc/self/stat counts in 10 ms ticks.
let tick = 1000.0 / clock_ticks_per_sec();
// RED SIDE: a single-threaded arm can never occupy two cores, whatever
// the neighbours are doing. This is the assertion the host-wide metric fails.
let one_avg = one.proc_cores_avg.expect("measured");
assert!(
one_avg < 2.0,
"single-threaded arm reported {one_avg:.2} cores — a process-scoped \
metric must not see the box's other load (host-wide said \
{}/{} cores busy)",
one.cores_busy_max,
one.n_cores
);
assert!(
one_cpu >= BURN_MS - 2.0 * tick,
"a thread that burned {BURN_MS} ms of its own CPU must be charged it, got {one_cpu:.0} ms"
);
// GREEN SIDE: four threads of the same work are charged four times the CPU.
assert!(
four_cpu >= 4.0 * (BURN_MS - 2.0 * tick),
"4 threads × {BURN_MS} ms of CPU were charged only {four_cpu:.0} ms"
);
assert!(
four_cpu > one_cpu * 2.5,
"saturation metric failed to DISTINGUISH: 1-thread={one_cpu:.0} ms \
4-thread={four_cpu:.0} ms of CPU — a metric that reports the same for \
both is the bug, not the fix"
);
// The occupancy is that CPU over the wall it took, and the peak is never
// below the whole-window mean.
for t in [&one, &four] {
let cpu = t.proc_cpu_ms.unwrap() + t.child_cpu_ms.unwrap();
let avg = t.proc_cores_avg.unwrap();
assert!((avg - cpu / t.elapsed_ms).abs() < 1e-9, "cores_avg is cpu/wall");
assert!(t.proc_cores_max.unwrap() >= avg - 1e-9);
}
}
/// The RED PROBE, kept runnable so the claim can be re-checked rather than
/// believed:
///
/// ```text
/// cargo test -p nornir-bench --lib -- --ignored --nocapture probe_host_wide
/// ```
///
/// Runs a 1-thread and a 4-thread arm back to back and prints both metrics
/// for each. On a loaded box the host-wide pair is identical (~N/N, ~100%)
/// for both arms — the gate input that cannot report red — while the
/// process-scoped pair separates them.
///
/// `#[ignore]` on purpose: its *point* is what the HOST is doing, so on an
/// idle machine there is nothing to demonstrate and asserting on it would be
/// a flake. The always-on guard is
/// [`proc_cores_distinguishes_serial_from_parallel`]; this one is the
/// evidence you paste.
#[test]
#[ignore]
fn probe_host_wide_vs_process_scoped() {
let one = spin(1, 4000);
let four = spin(4, 4000);
println!("\n loadavg: {}", std::fs::read_to_string("/proc/loadavg").unwrap_or_default().trim());
println!(
" {:<10} | {:>18} | {:>18} | {:>16}",
"arm", "HOST cores_busy_max", "HOST cpu_pct_max", "PROC cores_avg"
);
for (name, t) in [("1 thread", &one), ("4 threads", &four)] {
println!(
" {:<10} | {:>18} | {:>17.2}% | {:>16.2}",
name,
format!("{}/{}", t.cores_busy_max, t.n_cores),
t.cpu_pct_max,
t.proc_cores_avg.unwrap_or(f64::NAN),
);
}
let gate_frac = 0.9;
for (name, t) in [("1 thread", &one), ("4 threads", &four)] {
let old_pass = (t.cores_busy_max as f64) >= gate_frac * t.n_cores as f64;
let new_pass =
t.proc_cores_max.unwrap_or(0.0) >= gate_frac * t.n_cores as f64;
println!(
" gate(peak >= 0.9*cores) on {name:<10}: OLD={} NEW={}",
if old_pass { "PASS" } else { "RED " },
if new_pass { "PASS" } else { "RED " },
);
}
println!();
}
/// The child-process caveat, asserted rather than merely documented: an arm
/// whose work happens in a **reaped** child is counted (via `child_cpu_ms`),
/// and `proc_cpu_ms` alone would have missed it.
/// How long the child burns in [`reaped_child_cpu_is_counted_and_attributed`].
const CHILD_BURN_ENV: &str = "NORNIR_BENCH_TEST_BURN_CPU_MS";
/// Not a test of its own: the CHILD of
/// [`reaped_child_cpu_is_counted_and_attributed`]. It burns `CHILD_BURN_ENV`
/// ms of its own CPU and exits; without the variable it does nothing.
#[test]
#[ignore = "the child of reaped_child_cpu_is_counted_and_attributed; runs only with its env var"]
fn child_burns_a_fixed_cpu_budget() {
let Some(ms) = std::env::var(CHILD_BURN_ENV).ok().and_then(|v| v.parse::<f64>().ok())
else {
return;
};
let start = thread_cpu_ms();
let mut x = 0u64;
while thread_cpu_ms() - start < ms {
for _ in 0..10_000 {
x = x.wrapping_mul(2654435761).wrapping_add(1);
std::hint::black_box(x);
}
}
}
/// The child-process caveat, asserted rather than merely documented: an arm
/// whose work happens in a **reaped** child is counted (via `child_cpu_ms`),
/// and `proc_cpu_ms` alone would have missed it.
///
/// The child burns a fixed CPU budget (this test binary, re-run as
/// [`child_burns_a_fixed_cpu_budget`]), not two seconds of wall clock: a `sh`
/// loop on `date` got 230 ms of CPU in its two seconds at oden load 100 to
/// 400 (ODEN-CI round 1, 2026-10-25) and went red.
#[test]
fn reaped_child_cpu_is_counted_and_attributed() {
if read_self_cpu().is_none() {
return;
}
const BURN_MS: f64 = 1000.0;
let exe = std::env::current_exe().expect("the test binary's own path");
let _g = crate::serial_guard(); // see `spin`: /proc/self/stat is process-wide
let sampler = Sampler::start();
let status = std::process::Command::new(&exe)
.args([
"--exact",
"telemetry::tests::child_burns_a_fixed_cpu_budget",
"--ignored",
"--test-threads=1",
"-q",
])
.env(CHILD_BURN_ENV, BURN_MS.to_string())
.stdout(std::process::Stdio::null())
.status()
.expect("re-run the test binary as the child");
let t = sampler.stop();
assert!(status.success(), "the burning child failed: {status}");
let child = t.child_cpu_ms.expect("measured");
let own = t.proc_cpu_ms.expect("measured");
let tick = 1000.0 / clock_ticks_per_sec();
assert!(
child >= BURN_MS - 2.0 * tick,
"a REAPED child that burned {BURN_MS} ms of CPU must show up in child_cpu_ms, \
got {child:.0}ms (own={own:.0}ms) — this is the under-report \
/proc/self/stat's utime+stime alone would have hidden"
);
let avg = t.proc_cores_avg.expect("measured");
assert!(
(avg - (own + child) / t.elapsed_ms).abs() < 1e-9,
"process-tree occupancy is (own + child) / wall, got {avg:.3}"
);
}
}