open-standards/proof/proof-rs/crates/proof-cascade/src/meter.rs
Transaction Science 60bcca7594 proof-cascade: the meter reads its own process only once the measuring thread has left its core
measure_marginal's process-counter rung brackets its window with two
reads of this process's CPU energy (proc_pid_rusage ri_energy_nj). The
kernel adds a thread's energy to that count only when the thread leaves
a core, and a process's read of its own count does not take the reading
thread off its core. So the first read left up to a scheduling quantum
of what the measuring thread spent before the window (the liveness
probe's burn, the caller's own work) to land inside it, and the figure
is sealed HwShunt.

Measured on an M3 Ultra (macOS 27.0.1) with the meter's own read at each
end of a window on the reading thread, against the tests' own reader,
which sleeps before each read until the process's context-switch count
rose; three runs of 40 windows of each kind:
- 1 ms of the reading thread's CPU: all 40 read no advance, every run;
- a 2 ms sleep straight after 8 ms of CPU: all 40 read the burn before
  them, every run, 40 to 48 mJ for a sleep.

Each of the measurement's two reads now sleeps (50 us, then 200 us,
1 ms, 5 ms) until the measuring thread's own energy count (proc_pidinfo
PROC_PIDTHREADCOUNTS, read only in the one layout known: 88 bytes, two
performance levels here) rose across a sleep, and only then reads the
process. A read whose thread was not seen to leave gives the measurement
no figure, by the path an unreadable counter already takes: the work
does not run, measure_marginal returns None and the caller keeps its
estimate. The process counter is Unavailable where the thread's count
does not read (proof --measure says so), and the liveness probe still
samples with no sleep. The rule and the layout decoding are pure and
compiled on every platform.

The measurement was already one window around a batch: a second of
calls on every spare core, two reads in all. Its workers have exited and
been joined before the closing read, and a thread's exit puts its last
time on a core into the count before the join returns (measured: a 5 ms
burn read the same at the join whether or not the thread slept first),
so only the measuring thread needed taking off its core.

With it, on that host: quiet, every window read 98% or more of the
tests' window and no sleep more than 1% of the window that held the
burn; under 40 busy loops (110 runs), 82% or more and 4% or less. A
read costs about 80 us quiet and 190 us at the median under the busy
loops, against under 1 us.

process_window_read_nj is a doc-hidden hook, like the crate's other
test hooks: tests/process_energy_short_span.rs, the test that fails
without the check, needs a process of its own. The published
conformance vectors are unchanged (gen-vectors into a scratch directory
is byte-identical to conformance/v1).

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-12 20:50:53 -04:00

5824 lines
281 KiB
Rust
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

//! Real hardware energy measurement — the `HwShunt` backend the schema's
//! `Provenance::HwShunt` promises. Returns the **marginal** microjoules of one
//! `work` call, run on every spare core, from the first of these sources that
//! is live on this host ([`meter_source`](crate::meter::meter_source)):
//!
//! 1. **This process's CPU energy counter** (macOS): `proc_pid_rusage(getpid(),
//! RUSAGE_INFO_V6)`'s `ri_energy_nj`, the cumulative CPU energy the kernel
//! attributes to this process's threads — public libproc API, no privilege.
//! The counter is read before the workers start and after they are all
//! joined, so its advance covers every call counted — and whatever else
//! this process's threads burn meanwhile, which in a quiet caller is the
//! workers' start-up and the sleeping caller, a few microjoules — while
//! other processes are excluded by construction: no idle baseline to
//! subtract. Each of the two reads is taken only once the calling thread
//! is seen to have left its core (below).
//! 2. **IOReport's Energy Model** (Apple Silicon, read directly via its C API —
//! no `macmon`, no sudo): SoC power bracketed busy against idle.
//! 3. **RAPL** `energy_uj` (Linux): package power bracketed busy against idle.
//!
//! Returns `None` where no source is live — the caller keeps its
//! honestly-tagged estimate rather than inventing a number. No
//! elevated privileges and no external binaries.
//!
//! **A figure from this meter is a hardware reading, `HwShunt`, from every
//! source** ([`MeterSource::provenance`](crate::meter::MeterSource::provenance)).
//! IOReport's Energy Model and RAPL are vendor energy counters, and the
//! process counter is the kernel's accounting of the CPU's own energy
//! counters per thread — none is a shunt on the supply, and a vendor or kernel
//! counter's reading is a real reading: what it lacks is independent
//! provenance, not accuracy. The grade holds only while the counter is
//! demonstrably live for the domain the figure claims (SPEC §5): the
//! implementation must have observed the counter covering that domain advance
//! under load. A counter that is present but not live — frozen, or republished
//! in lumps — must not produce a hardware-graded figure. Every source here is
//! probed by the same rule before it is used, and a source that fails it is
//! never selected.
//!
//! **The domain differs by source, and a figure claims only its own**
//! ([`MeterSource::domain`](crate::meter::MeterSource::domain)). The process
//! counter is THIS process's CPU energy: not other processes, not child
//! processes, not GPU, ANE or DRAM energy, and everything this process's
//! threads burn during the window — so a caller must not run other work in
//! the process while it meters. IOReport's figure is SoC energy (CPU + GPU +
//! other domains, `counts_once`) over idle; RAPL's, package energy over idle.
//!
//! **On macOS, the process counter is used only while it is live**
//! ([`process_energy_status`](crate::meter::process_energy_status)): it must
//! keep advancing while the probe burns CPU — in at least five separate sample
//! intervals, in at least 80% of the intervals observed, over at least 300 ms
//! of wall time — the sustained-advance rule every probe here applies. Measured
//! on macOS 27.0.1 for an unprivileged reader, it is live on an M5 Max and on
//! an M3 Ultra (the probe calls it `Live`, and an independent reader saw it
//! change at every one of 29–36 reads 25 ms apart): on the M5 Max it changes
//! about every 0.5–0.7 ms under load, holds near 1–2 mJ/s idle, and keeps the
//! energy of threads that have exited.
//!
//! **A read of this process's own counter is taken only once the reading
//! thread is seen to have left its core.** The kernel adds a thread's energy
//! to its process's count only when the thread leaves a core — a context
//! switch, or the end of its quantum, about 10 ms of its CPU — and a
//! process's read of its own count does not take the reading thread off its
//! core: its CPU time is current at the read, its energy is not. So a window
//! bracketed by two plain reads holds, at its start, up to a quantum of what
//! the reading thread spent BEFORE the window (the liveness probe's burn,
//! say, or the caller's own work), and lacks, at its end, what that thread
//! spent since it last left a core. Measured on an M3 Ultra (macOS 27.0.1,
//! three runs of 40 windows of each kind, against the tests' own reader that
//! sleeps before each of its reads): a window over 1 ms of the reading
//! thread's CPU read no advance in all 40, every run; a window over a 2 ms
//! sleep opened straight after 8 ms of that thread's CPU read the burn
//! before it in all 40, every run, 40 to 48 mJ for a sleep.
//!
//! So before each read of its window the measuring thread sleeps (50 µs,
//! then 200 µs, 1 ms, 5 ms) until its own energy count
//! (`proc_pidinfo`'s `PROC_PIDTHREADCOUNTS`, by its thread id) rose across a
//! sleep — which it does only when the kernel adds that thread's energy to
//! the process's — and only then reads the process. A sleep alone is no
//! proof: one whose timer fires before the thread has blocked returns with
//! the thread still on its core; and the process's context-switch count is
//! none either, since any other thread's switch moves it. A read whose
//! thread was not seen to leave is no read: the measurement returns `None`,
//! and the caller keeps its estimate — never a figure short of the window's
//! tail, or holding what was spent before it, graded `HwShunt`. With it, the
//! same windows on the same host: quiet, all 40 of the 1 ms windows read 98%
//! or more of the tests' own window around them, and all 40 of the sleeps
//! under 1% of the window that held the burn before them; under 40 busy
//! loops (110 runs), every window read 82% or more, and no sleep more than
//! 4%. A read costs about 80 µs on the quiet host (the sleep and the wait
//! for a core after it) and 190 µs at the median under the busy loops (0.9 ms
//! at the median of the worst run, 50 ms the worst single read), against
//! under 1 µs for the read alone, and all four sleeps (6.25 ms) and their
//! waits when it fails. The measurement takes two, around a window of a
//! second over every spare core: one window around the whole batch of
//! calls, not one around each call. The workers have exited and been joined
//! before the closing read, and a thread's exit puts its last time on a core
//! into the count before the join returns (measured: a 5 ms burn read the
//! same at the join whether or not the thread slept before it exited), so
//! the window holds every call it counts. The counter is used only where the
//! measuring thread's own count reads, in the one layout known, and the
//! liveness probe samples with no sleep: it asks only whether the count
//! keeps advancing under a burn, which it does at every quantum.
//!
//! **IOReport counts as reachable only while every CPU energy rail is live**
//! ([`ioreport_status`](crate::meter::ioreport_status)). A subscription can
//! open while no CPU rail tracks the work: measured on macOS 27.0.1 for an
//! unprivileged reader, the CPU rails of an M5 Max and an M3 Ultra hold their
//! value under minutes of full CPU load and move only in rare batch refreshes
//! (both M3 Ultra dies held for at least 22 minutes, then stepped by ~1.3 kJ
//! each within one 100 ms sample, and an M5 Max's `CPU Energy` was seen to
//! publish one refresh in two steps 64 ms apart), while the M5 Max's GPU rail
//! keeps advancing. Bracketing CPU work against idle on those rails brackets
//! GPU activity, nothing, or a refresh. So the interface is used only when
//! every CPU rail — `CPU Energy`, or each die's `DIE_<n>_CPU Energy` on a
//! multi-die part such as the M3 Ultra — keeps advancing on its own under the
//! same rule; otherwise it is declined. Every rail, not one: the power reading
//! sums every CPU rail, so a die whose rail held is a die the figure would
//! claim without reading it.
//!
//! **On Linux, RAPL counts as reachable only while every package counter is
//! live** ([`rapl_status`](crate::meter::rapl_status)). The same probe loop and
//! the same rule run over the `energy_uj` counters of the RAPL zones named
//! `package-<n>`, while the probe burns CPU; RAPL is used only when EVERY
//! package counter keeps advancing — in at least five separate sample
//! intervals, in at least 80% of the intervals observed, over at least 300 ms
//! — because the figure sums all of them. A counter that holds, steps in a
//! lump, or goes down (a wrap) in an interval did not advance in it. The
//! figure is the package zones' energy alone: a `psys` zone covers the
//! package too, so summing it beside one would count that energy twice, and
//! its liveness is not probed. `energy_uj` is root-only on current kernels,
//! so an unprivileged run usually reads no zone: `Unavailable`, and `None`.
//!
//! The chains are short on purpose. Each source's production status is one
//! expression — the generic probe loop (`liveness::probe_watching`) on the
//! real parts (the reader, the thread-CPU tick and the clocks) and the pure
//! status of its run — cached once per process: `host_status` for IOReport
//! ([`ioreport_status`](crate::meter::ioreport_status), gated by `admits`),
//! `process_host_status` for the process counter
//! ([`process_energy_status`](crate::meter::process_energy_status)) and
//! `rapl_host_status` for RAPL ([`rapl_status`](crate::meter::rapl_status)),
//! both gated by `admits_counter`, which admits only `Live`. The selection
//! ([`meter_source`](crate::meter::meter_source)) takes the first admitted
//! source in order, and every metering path — [`measure_marginal`](crate::meter::measure_marginal),
//! `power_over`, `rapl_power_over` — goes through its source's own gate. The
//! decisions all live in pure code that the tests run on scripted fakes, and
//! the tests also run the production compositions on the real FFI.
//!
//! The brackets are measurements with stated bounds — SoC/package power
//! resolved to the marginal delta over an idle machine ÷ iterations — exactly
//! the discipline the WAI sink meter uses, including running the burst on
//! every core so the signal clears a loaded machine's noisy idle baseline.
//! `work` must be cheap, pure, and idempotent (e.g. the real T0 decider on a
//! fixed goal).
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;
/// The hardware source a figure from this meter came from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MeterSource {
/// This process's CPU energy counter (macOS `proc_pid_rusage`
/// `ri_energy_nj`), read around the work: no idle subtraction.
ProcessEnergy,
/// IOReport's Energy Model (Apple Silicon): SoC power, busy over idle.
IoReport,
/// RAPL's package counters (Linux): package power, busy over idle.
Rapl,
}
impl MeterSource {
/// The grade of a figure from this source: `HwShunt` for every source — a
/// vendor counter's (IOReport, RAPL) or the kernel's per-process
/// accounting of the CPU's energy counters, read only while live (SPEC
/// §5; the owner's decision for the process counter).
pub fn provenance(self) -> proof_schema::Provenance {
proof_schema::Provenance::HwShunt
}
/// The domain a figure from this source claims, and what it leaves out.
pub fn domain(self) -> &'static str {
match self {
MeterSource::ProcessEnergy => {
"this process's CPU energy, as the kernel attributes it to this process's threads \
(not other processes, not child processes, not GPU, ANE or DRAM energy)"
}
MeterSource::IoReport => "SoC energy (CPU, GPU and other Energy Model domains) over the idle baseline",
MeterSource::Rapl => "CPU package energy (every RAPL package zone) over the idle baseline",
}
}
}
/// One figure from this meter: the marginal energy of one `work` call, and the
/// source that metered it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Measurement {
pub uj: u64,
pub source: MeterSource,
}
impl Measurement {
/// The figure as a schema reading: `measured_uj` present, graded by its
/// source ([`MeterSource::provenance`]).
pub fn energy(&self) -> proof_schema::Energy {
proof_schema::Energy { measured_uj: Some(self.uj), estimated_uj: self.uj, provenance: self.source.provenance() }
}
}
/// Marginal energy of one `work()` call, from the first live hardware source
/// ([`meter_source`]), and which source it was. A figure is a hardware reading,
/// `HwShunt`, and is only ever produced through counters the liveness probe
/// saw keep advancing under load (see the module doc).
///
/// - `None` if no source is live ([`meter_source`] is `None`: a counter that
/// is not live yields `None` before any window runs, and `work` never runs),
/// if the selected source gave no reading, or if `work` never completed a
/// call. The caller keeps its own estimate.
/// - From the process counter, `Some` is the counter's advance across the
/// calls ÷ the calls; `Some(0)` means under half a microjoule a call, below
/// this API's resolution — not a zero-energy call.
/// - From a bracket (IOReport, RAPL), `Some(0)` means the busy window did not
/// draw more than the idle one: the signal stayed inside the idle noise.
/// That is not a measured zero-energy call, and a caller must not report it
/// as one (`proof --measure` says the signal stayed inside the idle noise).
///
/// `work` runs on every spare core, so keep it pure + `Copy`. Anything else
/// this process runs during the window is counted by the process counter.
pub fn measure_marginal<F>(work: F) -> Option<Measurement>
where
F: Fn() + Send + Sync + Copy + 'static,
{
let source = meter_source()?;
let uj = match source {
MeterSource::ProcessEnergy => process_marginal_uj(work)?,
MeterSource::IoReport | MeterSource::Rapl => bracket_marginal_uj(work)?,
};
Some(Measurement { uj, source })
}
/// [`measure_marginal`]'s figure alone, in microjoules.
pub fn measure_marginal_uj<F>(work: F) -> Option<u64>
where
F: Fn() + Send + Sync + Copy + 'static,
{
measure_marginal(work).map(|m| m.uj)
}
/// The bracket rung alone — IOReport on macOS, RAPL on Linux — gated by its
/// own interface's cached status, whatever source ranks above it.
fn bracket_marginal_uj<F>(work: F) -> Option<u64>
where
F: Fn() + Send + Sync + Copy + 'static,
{
measure_gated(bracket_admitted(), work, idle_power, busy_power)
}
/// Whether the bracket rung's interface is admitted: IOReport `Live` on macOS,
/// RAPL `Live` on Linux, never elsewhere. Each probe runs once per process.
fn bracket_admitted() -> bool {
#[cfg(target_os = "macos")]
{
admits(ioreport_status())
}
#[cfg(target_os = "linux")]
{
admits_counter(rapl_status())
}
#[cfg(not(any(target_os = "macos", target_os = "linux")))]
{
false
}
}
/// [`bracket_marginal_uj`] with the selection's answer and the power readers
/// passed in. The selection is taken first: an interface that is not
/// available (one [`admits`] or [`admits_counter`] turns away) returns `None`
/// before any power is read or `work` runs.
fn measure_gated<F>(
available: bool,
work: F,
idle_power: impl FnOnce() -> Option<f64>,
busy_power: impl FnOnce() -> Option<f64>,
) -> Option<u64>
where
F: Fn() + Send + Sync + Copy + 'static,
{
// An interface that opens but cannot see CPU work is declined up front.
if !available {
return None;
}
let (idle_w, busy_w, window, runs) = bracket_with(work, idle_power, busy_power)?;
let marginal_w = (busy_w - idle_w).max(0.0);
if runs == 0 {
return None;
}
Some((marginal_w * window / runs as f64 * 1e6).round() as u64)
}
/// The source [`measure_marginal`] meters with on this host: the first one
/// that is live, in order — on macOS this process's CPU energy counter
/// ([`process_energy_status`]), then IOReport ([`ioreport_status`]); on Linux
/// RAPL ([`rapl_status`]); elsewhere none. A readable counter that is not live
/// is never selected. The first call in a process pays the liveness probe of
/// each source it has to consult (about 0.3 s of burning CPU on the calling
/// thread for a live process counter; for IOReport its own Energy Model
/// subscription and about a second); each answer is cached for the process,
/// and a source ranked below a live one is not probed here.
pub fn meter_source() -> Option<MeterSource> {
#[cfg(target_os = "macos")]
{
macos_source(process_energy_status(), ioreport_status)
}
#[cfg(target_os = "linux")]
{
admits_counter(rapl_status()).then_some(MeterSource::Rapl)
}
#[cfg(not(any(target_os = "macos", target_os = "linux")))]
{
None
}
}
/// The macOS selection, pure: the process counter if it is live; otherwise
/// IOReport if its CPU rail is live; otherwise none. IOReport's status is
/// asked only when the process counter is not live.
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
fn macos_source(process: CounterStatus, ioreport: impl FnOnce() -> IoReportStatus) -> Option<MeterSource> {
if admits_counter(process) {
Some(MeterSource::ProcessEnergy)
} else if admits(ioreport()) {
Some(MeterSource::IoReport)
} else {
None
}
}
/// Whether any hardware source is live on this machine: [`meter_source`] is
/// `Some`. On macOS that is this process's CPU energy counter or IOReport with
/// a live CPU energy rail; on Linux, RAPL with every package counter live. A
/// readable counter that is not live does not count.
pub fn hardware_meter_available() -> bool {
meter_source().is_some()
}
/// Why the IOReport interface is or is not used on this host.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IoReportStatus {
/// IOReport could not be opened or sampled: no subscription to its Energy
/// Model group could be opened (not macOS, not Apple Silicon, running
/// under Rosetta, or no such group), or that subscription opened but never
/// yielded two consecutive samples within the probe's five budgets. Never
/// used.
Unavailable,
/// The Energy Model subscription opened and was sampled, but no
/// `CPU Energy` rail exists in it — an unfamiliar channel naming. Never
/// used.
CpuRailAbsent,
/// A CPU energy rail exists, but not every one kept advancing while CPU
/// was burned: at least one CPU rail (on a multi-die part, one die's) did
/// not advance in five separate sample intervals that were at least 80% of
/// those observed, over at least 300 ms. A rail that stepped once, or in a
/// few steps close together, did not track the work either — that is a
/// batch refresh. The platform is not exposing live energy for the whole
/// CPU to this process, so bracketing CPU work on it would measure
/// GPU-only energy, nothing, a refresh, or some dies' energy as the whole
/// CPU's. Observed on macOS 27.0.1 on both an M5 Max and an M3 Ultra.
CpuRailFrozen,
/// Every CPU energy rail kept advancing, each one on its own, while the
/// probe burned CPU: in at least five separate sample intervals, in at
/// least 80% of the intervals observed, over at least 300 ms of wall time.
/// IOReport is used.
Live,
}
/// How long (wall time) the liveness probe watches the CPU energy rail before
/// concluding it is frozen. Before each re-sample the probe burns ~25 ms of
/// the calling thread's own CPU time (at most 500 ms of wall time per tick).
/// It samples its own subscription to the Energy Model group alone — 364
/// channels on an M5 Max, 645 on an M3 Ultra, ~6–9 ms of thread CPU a sample,
/// where a sample of every channel (~11,900 / ~14,300) costs ~85–125 ms — so on
/// a quiet host this buys about thirty intervals, and fewer on a loaded one,
/// where a live rail needs five, spanning at least 300 ms; a live rail on a
/// quiet host is called live after about ten. If fewer than five intervals
/// were observed when it runs out, the probe keeps sampling, up to five
/// budgets. A live rail slower than that fails safe: declined, and the caller
/// keeps its own estimate. Paid once per process; see
/// [`ioreport_status`].
#[cfg(target_os = "macos")]
pub const CPU_RAIL_LIVENESS_BUDGET: std::time::Duration = std::time::Duration::from_millis(1000);
/// How many times this process has run the production liveness probe,
/// [`host_status`] — once at most. It counts runs, not cache lookups: the
/// counter is `host_status`'s first statement, and `host_status` is the only
/// production caller of the probe. A process-wide static (not a
/// thread-local), so a test that asks from many threads at once can pin it.
#[cfg(target_os = "macos")]
static REAL_PROBE_RUNS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
/// How many times this process has run the production IOReport liveness probe
/// (always 0 off macOS). Exposed for the once-per-process test in
/// `tests/ioreport_once.rs`; not part of the meter's interface.
#[doc(hidden)]
pub fn real_probe_runs() -> usize {
#[cfg(target_os = "macos")]
{
REAL_PROBE_RUNS.load(Ordering::SeqCst)
}
#[cfg(not(target_os = "macos"))]
{
0
}
}
/// Set by [`simulate_no_energy_model_group`]: the probe's opener then finds no
/// Energy Model group to subscribe to.
#[cfg(target_os = "macos")]
static NO_ENERGY_MODEL_GROUP: AtomicBool = AtomicBool::new(false);
/// A test hook, not a setting: from now on this process's liveness probe
/// finds no Energy Model group to subscribe to, as on a host that has none. A
/// test in a process of its own (`tests/ioreport_unopenable.rs`) uses it to
/// drive the whole production chain — probe, status, cache, selection,
/// metering — down its unopenable path, which neither Mac that runs the tests
/// can reach otherwise: the status must be `Unavailable` and every meter must
/// decline. It takes effect only before the first use of [`ioreport_status`],
/// which caches its answer for the process. It can only turn the meter off.
/// A no-op off macOS.
#[doc(hidden)]
pub fn simulate_no_energy_model_group() {
#[cfg(target_os = "macos")]
NO_ENERGY_MODEL_GROUP.store(true, Ordering::SeqCst);
}
/// The production liveness probe's answer, and the record of the run that
/// decided it (every read, its ticks, and what the rule was handed), so tests
/// check the run that really made the decision.
#[cfg(target_os = "macos")]
#[derive(Debug)]
struct HostStatus {
status: IoReportStatus,
#[cfg_attr(not(test), allow(dead_code))]
run: liveness::ProbeRun,
}
#[cfg(target_os = "macos")]
impl HostStatus {
/// A run and its status: always [`status_of`] of that same run.
fn of(run: liveness::ProbeRun) -> Self {
Self { status: status_of(&run), run }
}
}
/// THE production composition: the generic probe loop
/// ([`liveness::probe_with`]) on the real parts — the probe's own Energy Model
/// subscription and its reads, the thread-CPU tick, this process's wall clock
/// and the thread CPU clock — and the pure status of its run. One expression,
/// no branch: every decision is in `probe_with` and [`status_of`], which the
/// tests run on scripted fakes, and the tests also run this very function
/// (uncached), on the real FFI and on scripted channel reads. Counts itself
/// first; the only production caller of `probe_with`, reached in production
/// only through the cache ([`host`]).
#[cfg(target_os = "macos")]
fn host_status() -> HostStatus {
REAL_PROBE_RUNS.fetch_add(1, Ordering::SeqCst);
HostStatus::of(liveness::probe_with(
macos_ioreport::EnergyModelSub::open,
macos_ioreport::EnergyModelSub::read,
probe_clock::spin_tick,
probe_clock::wall_clock,
probe_clock::thread_cpu_time,
CPU_RAIL_LIVENESS_BUDGET,
))
}
/// The once-per-process cache of [`host_status`]: one process-wide
/// `OnceLock`, so concurrent first callers wait for the one probe instead of
/// each running their own.
#[cfg(target_os = "macos")]
fn host() -> &'static HostStatus {
static HOST: std::sync::OnceLock<HostStatus> = std::sync::OnceLock::new();
HOST.get_or_init(host_status)
}
/// This host's IOReport status, probed once per process and cached.
///
/// On macOS, [`meter_source`] (and through it [`measure_marginal`]) and
/// `power_over` use IOReport only when `admits` this — only when it is
/// [`IoReportStatus::Live`] — and the selection asks for it only when this
/// process's CPU energy counter, ranked first, is not live. Exposed so a
/// caller can say *why* IOReport is not used instead of guessing. Always
/// `Unavailable` off macOS; it does not describe Linux RAPL.
pub fn ioreport_status() -> IoReportStatus {
#[cfg(target_os = "macos")]
{
host().status
}
#[cfg(not(target_os = "macos"))]
{
IoReportStatus::Unavailable
}
}
/// The status a probe run earns: `Unavailable` when its Energy Model
/// subscription did not open, or opened but never yielded two consecutive
/// samples (no verdict) — never `Live`, and never `CpuRailAbsent`, which needs
/// a sampled subscription with no CPU rail in it; otherwise its verdict, one to
/// one.
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
fn status_of(run: &liveness::ProbeRun) -> IoReportStatus {
match (run.opened, run.verdict) {
(false, _) | (true, None) => IoReportStatus::Unavailable,
(true, Some(liveness::CpuRail::Absent)) => IoReportStatus::CpuRailAbsent,
(true, Some(liveness::CpuRail::Frozen)) => IoReportStatus::CpuRailFrozen,
(true, Some(liveness::CpuRail::Live)) => IoReportStatus::Live,
}
}
/// Whether a status lets the IOReport meter be used: only `Live`. Every
/// selection path on macOS gates through this, on the cached status.
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
fn admits(status: IoReportStatus) -> bool {
matches!(status, IoReportStatus::Live)
}
/// Whether an energy counter other than IOReport's — this process's CPU
/// energy counter on macOS ([`process_energy_status`]), RAPL's package
/// counters on Linux ([`rapl_status`]) — is live for the domain a figure from
/// it would claim (SPEC §5). Decided by the same probe loop and the same
/// sustained-advance rule as [`IoReportStatus`], once per process.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CounterStatus {
/// The counter could not be read: no such interface on this platform, no
/// permission to read it (RAPL's `energy_uj` is root-only on current
/// kernels), or never two consecutive reads within the probe's five
/// budgets. Never used. (The process counter is never `Absent`: it is one
/// counter, read or not. It is `Unavailable` too where the measuring
/// thread's own energy count does not read: each read of a window is
/// checked against that count, so without it no window could be read.)
Unavailable,
/// The interface was read, but holds no counter for the domain a figure
/// would claim (RAPL: zones, but no `package-<n>` zone). Never used.
Absent,
/// The counter did not keep advancing while the probe burned CPU — for
/// RAPL, at least one package counter did not — it held, or moved in a
/// few steps close together. A figure from it would not be a live
/// reading. Never used.
Frozen,
/// The counter kept advancing while the probe burned CPU — for RAPL,
/// every package counter did: in at least five separate sample
/// intervals, in at least 80% of the intervals observed, over at least
/// 300 ms of wall time. Used.
Live,
}
/// The status a probe run of a [`CounterStatus`] source earns: `Unavailable`
/// when nothing could be opened, or nothing was read twice in a row (no
/// verdict); otherwise its verdict, one to one.
#[cfg_attr(not(any(target_os = "macos", target_os = "linux")), allow(dead_code))]
fn counter_status_of(run: &liveness::ProbeRun) -> CounterStatus {
match (run.opened, run.verdict) {
(false, _) | (true, None) => CounterStatus::Unavailable,
(true, Some(liveness::CpuRail::Absent)) => CounterStatus::Absent,
(true, Some(liveness::CpuRail::Frozen)) => CounterStatus::Frozen,
(true, Some(liveness::CpuRail::Live)) => CounterStatus::Live,
}
}
/// Whether a [`CounterStatus`] lets its source meter: only `Live`.
#[cfg_attr(not(any(target_os = "macos", target_os = "linux")), allow(dead_code))]
fn admits_counter(status: CounterStatus) -> bool {
matches!(status, CounterStatus::Live)
}
/// How long (wall time) the RAPL liveness probe watches the package counters
/// before concluding one is frozen: the IOReport probe's budget, ticks and
/// caps. RAPL counters update about every millisecond, so on a quiet host a
/// live set is called live at the tenth or so ~25 ms interval.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const RAPL_LIVENESS_BUDGET: std::time::Duration = std::time::Duration::from_millis(1000);
/// The RAPL liveness probe on its parts: `zones` reads every top-level
/// powercap zone (sysfs, on Linux), and the tick and clocks are as for
/// IOReport. It opens when a read finds any zone, and hands the rule each
/// read's package counters ([`rapl::read_of`]), every one of which must keep
/// advancing ([`rapl::PACKAGES`]). A read that finds no zone is a failed read.
/// Generic, so the tests run this very composition on scripted zones on every
/// platform.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn rapl_probe_with(
zones: impl Fn() -> Vec<rapl::Zone>,
tick: impl FnMut() -> probe_clock::TickOutcome,
wall: impl Fn() -> std::time::Duration,
cpu: impl Fn() -> Option<std::time::Duration>,
budget: std::time::Duration,
) -> liveness::ProbeRun {
liveness::probe_watching(
rapl::PACKAGES,
|| (!zones().is_empty()).then_some(()),
|_: &()| {
let z = zones();
(!z.is_empty()).then(|| rapl::read_of(&z))
},
tick,
wall,
cpu,
budget,
)
}
/// How many times this process has run the production RAPL liveness probe —
/// once at most (always 0 off Linux). Exposed for the once-per-process test in
/// `tests/rapl_once.rs`; not part of the meter's interface.
#[doc(hidden)]
pub fn real_rapl_probe_runs() -> usize {
#[cfg(target_os = "linux")]
{
REAL_RAPL_PROBE_RUNS.load(Ordering::SeqCst)
}
#[cfg(not(target_os = "linux"))]
{
0
}
}
#[cfg(target_os = "linux")]
static REAL_RAPL_PROBE_RUNS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
/// THE production RAPL composition: [`rapl_probe_with`] on the real sysfs
/// zones, the thread-CPU tick and the clocks, and the status of its run.
/// Counts itself first; reached in production only through the cache in
/// [`rapl_status`].
#[cfg(target_os = "linux")]
fn rapl_host_status() -> CounterStatus {
REAL_RAPL_PROBE_RUNS.fetch_add(1, Ordering::SeqCst);
counter_status_of(&rapl_probe_with(
rapl_zones,
probe_clock::spin_tick,
probe_clock::wall_clock,
probe_clock::thread_cpu_time,
RAPL_LIVENESS_BUDGET,
))
}
/// This host's RAPL status, probed once per process and cached: whether every
/// RAPL package counter (`intel-rapl:<n>` zones named `package-<n>`) kept
/// advancing while the probe burned CPU. On Linux, [`meter_source`] (and
/// through it [`measure_marginal`]) and `rapl_power_over` use RAPL only when
/// this is [`CounterStatus::Live`]. The first call in a process pays the probe (about
/// `RAPL_LIVENESS_BUDGET` of burning CPU and re-reading sysfs on the calling
/// thread when the counters are readable; nothing when they are not). Always
/// `Unavailable` off Linux.
pub fn rapl_status() -> CounterStatus {
#[cfg(target_os = "linux")]
{
static RAPL: std::sync::OnceLock<CounterStatus> = std::sync::OnceLock::new();
*RAPL.get_or_init(rapl_host_status)
}
#[cfg(not(target_os = "linux"))]
{
CounterStatus::Unavailable
}
}
/// Run `work` on every spare core for a window, bracketed against idle, and
/// return `(idle_w, busy_w, window_s, total_runs)`. The power readings come
/// from `idle_power` and `busy_power`; `None` if either gives none. The
/// workers are stopped and joined before the busy reading is looked at, so a
/// busy window with no reading returns `None` without leaving them spinning for
/// the rest of the process.
fn bracket_with<F>(
work: F,
idle_power: impl FnOnce() -> Option<f64>,
busy_power: impl FnOnce() -> Option<f64>,
) -> Option<(f64, f64, f64, u64)>
where
F: Fn() + Send + Sync + Copy + 'static,
{
let workers = spare_cores();
let idle_w = idle_power()?;
let stop = Arc::new(AtomicBool::new(false));
let iters = Arc::new(AtomicU64::new(0));
let handles: Vec<_> = (0..workers)
.map(|_| {
let (stop, iters) = (stop.clone(), iters.clone());
std::thread::spawn(move || {
while !stop.load(Ordering::Relaxed) {
work();
iters.fetch_add(1, Ordering::Relaxed);
}
})
})
.collect();
let t0 = Instant::now();
let busy_w = busy_power();
let window = t0.elapsed().as_secs_f64();
let runs = iters.load(Ordering::Relaxed);
stop.store(true, Ordering::Relaxed);
for h in handles {
let _ = h.join();
}
Some((idle_w, busy_w?, window, runs))
}
// ---------------------------------------------------------------------------
// This process's CPU energy counter (macOS): its liveness, cached once per
// process, and the measurement that reads it around the work.
// ---------------------------------------------------------------------------
/// How long (wall time) the process-counter liveness probe watches the counter
/// before concluding it is frozen: the IOReport probe's budget, ticks and caps.
/// The counter changes every ~0.5–0.7 ms under load on an M5 Max (macOS
/// 27.0.1), so a live counter is called live at the first interval past 300 ms.
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
const PROCESS_LIVENESS_BUDGET: std::time::Duration = std::time::Duration::from_millis(1000);
/// How long the process-counter measurement runs `work` on every spare core.
/// The counter is read before the workers start and after they are joined,
/// so the window only sets how many calls the figure averages over.
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
const PROCESS_WINDOW: std::time::Duration = std::time::Duration::from_millis(1000);
/// The name the process counter has in the sample the liveness rule is handed.
const PROCESS_RAIL: &str = "process CPU energy";
/// Whether a sample channel is the process counter.
fn is_process_rail(name: &str) -> bool {
name == PROCESS_RAIL
}
/// The liveness rule over the process counter: the one counter must keep
/// advancing, by the same sustained-advance rule as every other source.
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
const PROCESS_COUNTER: liveness::Watched = liveness::Watched { pick: is_process_rail };
/// One read of the process counter (cumulative nJ) as the rule is handed it.
fn process_read(nj: u64) -> liveness::Read {
liveness::Read {
sample: vec![(PROCESS_RAIL.to_string(), i64::try_from(nj).unwrap_or(i64::MAX))],
listed: liveness::Listed { channels: 1, group_only: true },
}
}
/// The process-counter liveness probe on its parts: `counter` reads this
/// process's cumulative CPU energy (nJ), and the tick and clocks are as for
/// IOReport. It opens when the counter reads, and hands the rule every read
/// ([`PROCESS_COUNTER`]). A read that fails is a failed read. Generic, so the
/// tests run this very composition on scripted counters on every platform.
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
fn process_probe_with(
counter: impl Fn() -> Option<u64>,
tick: impl FnMut() -> probe_clock::TickOutcome,
wall: impl Fn() -> std::time::Duration,
cpu: impl Fn() -> Option<std::time::Duration>,
budget: std::time::Duration,
) -> liveness::ProbeRun {
liveness::probe_watching(
PROCESS_COUNTER,
|| counter().map(|_| ()),
|_: &()| counter().map(process_read),
tick,
wall,
cpu,
budget,
)
}
/// How many times this process has run the production process-counter
/// liveness probe — once at most (always 0 off macOS). Exposed for the
/// once-per-process tests; not part of the meter's interface.
#[doc(hidden)]
pub fn real_process_probe_runs() -> usize {
#[cfg(target_os = "macos")]
{
REAL_PROCESS_PROBE_RUNS.load(Ordering::SeqCst)
}
#[cfg(not(target_os = "macos"))]
{
0
}
}
#[cfg(target_os = "macos")]
static REAL_PROCESS_PROBE_RUNS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
/// THE production process-counter composition: [`process_probe_with`] on the
/// real counter (`proc_pid_rusage` for this pid), the thread-CPU tick and the
/// clocks, and the status of its run. Counts itself first; reached in
/// production only through the cache in [`process_energy_status`].
#[cfg(target_os = "macos")]
fn process_host_status() -> CounterStatus {
REAL_PROCESS_PROBE_RUNS.fetch_add(1, Ordering::SeqCst);
process_status_where(macos_process::thread_energy_nj().is_some(), || {
counter_status_of(&process_probe_with(
macos_process::energy_nj,
probe_clock::spin_tick,
probe_clock::wall_clock,
probe_clock::thread_cpu_time,
PROCESS_LIVENESS_BUDGET,
))
})
}
/// The process counter's status on a host where the measuring thread's own
/// energy count reads, or does not: what `probe` finds where it does, and
/// `Unavailable`, with no probe run, where it does not — every read of a
/// window is checked against that count ([`own_read`]), so a counter there
/// could never meter.
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
fn process_status_where(thread_count_reads: bool, probe: impl FnOnce() -> CounterStatus) -> CounterStatus {
if thread_count_reads {
probe()
} else {
CounterStatus::Unavailable
}
}
/// This host's process-counter status, probed once per process and cached:
/// whether this process's cumulative CPU energy counter (`proc_pid_rusage`,
/// `RUSAGE_INFO_V6`, `ri_energy_nj`) kept advancing while the probe burned
/// CPU. On macOS [`meter_source`] selects it first when this is
/// [`CounterStatus::Live`]. The first call pays the probe (about 0.3 s of
/// burning CPU on the calling thread for a live counter, up to about a second
/// for a frozen one). Always `Unavailable` off macOS.
pub fn process_energy_status() -> CounterStatus {
#[cfg(target_os = "macos")]
{
static PROCESS: std::sync::OnceLock<CounterStatus> = std::sync::OnceLock::new();
*PROCESS.get_or_init(process_host_status)
}
#[cfg(not(target_os = "macos"))]
{
CounterStatus::Unavailable
}
}
/// A test hook, not a setting: from now on this process's CPU energy counter
/// reads as the value it held at its first read after this call, as a frozen
/// counter would. A test in a process of its own uses it to drive the whole
/// production chain — probe, status, cache, selection, metering — down the
/// frozen path, which no Mac that runs the tests reaches otherwise: the
/// status must be `Frozen` and the process counter must never be selected.
/// It takes effect only before the first use of [`process_energy_status`],
/// which caches its answer for the process. It can only turn the meter off: a
/// counter that holds is never live. A no-op off macOS.
#[doc(hidden)]
pub fn simulate_frozen_process_energy() {
#[cfg(target_os = "macos")]
macos_process::FROZEN.store(true, Ordering::SeqCst);
}
/// The read [`measure_marginal`] takes of this process's CPU energy counter at
/// each end of its window, in nanojoules, where that counter is live: `None`
/// off macOS, where it is not live, and where the calling thread was not seen
/// to leave its core before the read ([`own_read`]). Exposed for the
/// short-span test, which brackets work of its own with two of these reads in
/// a process of its own; not part of the meter's interface.
#[doc(hidden)]
pub fn process_window_read_nj() -> Option<u64> {
#[cfg(target_os = "macos")]
{
if !admits_counter(process_energy_status()) {
return None;
}
macos_process::window_read_nj()
}
#[cfg(not(target_os = "macos"))]
{
None
}
}
/// The process-counter rung: `work` on every spare core, metered by this
/// process's CPU energy counter — only while it is live (its own gate, on the
/// cached status). `None` off macOS.
fn process_marginal_uj<F>(work: F) -> Option<u64>
where
F: Fn() + Send + Sync + Copy + 'static,
{
#[cfg(target_os = "macos")]
{
if !admits_counter(process_energy_status()) {
return None;
}
process_marginal_with(work, macos_process::window_read_nj, PROCESS_WINDOW)
}
#[cfg(not(target_os = "macos"))]
{
let _ = work;
None
}
}
/// How many workers run `work`: every core but the calling thread's, at
/// least one, at most twelve.
fn spare_cores() -> usize {
std::thread::available_parallelism().map(|n| n.get().saturating_sub(1).clamp(1, 12)).unwrap_or(6)
}
/// Run `work` on every spare core for `window`, reading `counter` — this
/// process's cumulative CPU energy in nJ — before the workers start and after
/// every one of them has been joined, and return the energy per call in µJ,
/// rounded to the nearest: the counter's advance ÷ every call that completed.
/// Reading after the join puts the energy of every counted call, and only
/// those calls' window, in the figure; a call still running when `window`
/// ends finishes and is counted. `None` if either read gives nothing (the
/// work never runs when the first one does) — the call failed, or, in
/// production, the calling thread was not seen to leave its core before the
/// read ([`own_read`]) — if the counter went down, or if no call completed.
/// Generic over the counter, so the tests drive it on a scripted counter.
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
fn process_marginal_with<F>(work: F, counter: impl Fn() -> Option<u64>, window: std::time::Duration) -> Option<u64>
where
F: Fn() + Send + Sync + Copy + 'static,
{
let before = counter()?;
let stop = Arc::new(AtomicBool::new(false));
let calls = Arc::new(AtomicU64::new(0));
let handles: Vec<_> = (0..spare_cores())
.map(|_| {
let (stop, calls) = (stop.clone(), calls.clone());
std::thread::spawn(move || {
while !stop.load(Ordering::Relaxed) {
work();
calls.fetch_add(1, Ordering::Relaxed);
}
})
})
.collect();
std::thread::sleep(window);
stop.store(true, Ordering::Relaxed);
for h in handles {
let _ = h.join();
}
let runs = calls.load(Ordering::SeqCst);
let after = counter()?;
if runs == 0 {
return None;
}
let nj = u128::from(after.checked_sub(before)?);
let runs = u128::from(runs);
u64::try_from((nj + runs * 500) / (runs * 1000)).ok()
}
// ---------------------------------------------------------------------------
// Counter liveness: the decision, the probe loop and the record it keeps,
// what an unopenable subscription means, which channels a read keeps, and
// which rails the power reading sums — kept apart from the FFI so all of it is
// pure, tested code on every platform. The macOS probe only supplies the
// subscription, the raw channel reads, the CPU burn and the clocks; the Linux
// RAPL probe, the sysfs zone reads and the same burn and clocks.
// ---------------------------------------------------------------------------
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
mod liveness {
use std::time::Duration;
/// In how many separate sample intervals each watched rail must advance, on
/// its own, to count as live. Two is not enough — a batch refresh can be published
/// in two steps — and neither is any count on its own: see [`RailWatch`].
pub(super) const MIN_LIVE_ADVANCES: u32 = 5;
/// How much wall time, from the first sample, the observation must span
/// before a rail can count as live: a refresh published in steps over a
/// few tens of milliseconds cannot fill it.
pub(super) const MIN_LIVE_WINDOW: Duration = Duration::from_millis(300);
/// The probe's hard cap, in budgets, while it waits for enough intervals.
pub(super) const MAX_BUDGETS: u32 = 5;
/// Whether `advances` advancing intervals of one rail are at least 80% of
/// the `intervals` observed so far.
fn sustained(advances: u32, intervals: u32) -> bool {
u64::from(advances) * 5 >= u64::from(intervals) * 4
}
/// What the liveness probe saw of the CPU energy rail.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum CpuRail {
/// Every CPU energy rail kept advancing, each one on its own, while
/// the probing thread burned CPU: in at least [`MIN_LIVE_ADVANCES`]
/// separate sample intervals, in at least 80% of the intervals
/// observed, over at least [`MIN_LIVE_WINDOW`] of wall time.
Live,
/// A CPU energy rail exists, but not every one kept advancing like
/// that: at least one held its value, or moved in one or a few steps
/// close together (a batch refresh, not a live reading).
Frozen,
/// No CPU energy channel exists in the Energy Model group.
Absent,
}
/// One sample as the rule reads it: every Energy Model channel's name and
/// its ABSOLUTE raw integer, in channel order. Channels that are not CPU
/// rails ride along and are ignored by the rule.
pub(super) type Sample = Vec<(String, i64)>;
/// One channel of an IOReport sample as the FFI read it: its group and its
/// name (`None` where the string could not be read), and its raw integer.
pub(super) type Channel = (Option<String>, Option<String>, i64);
/// What one IOReport sample listed before the Energy Model channels were
/// picked out of it: how many channels, and whether every one of them is in
/// the Energy Model group. The probe subscribes to that group alone, so each
/// of its samples lists exactly the group's channels (364 on an M5 Max, 645
/// on an M3 Ultra) and nothing else; a subscription to every channel lists
/// ~11,900 / ~14,300 of them, in 186 / 255 groups.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg_attr(not(test), allow(dead_code))]
pub(super) struct Listed {
pub(super) channels: usize,
pub(super) group_only: bool,
}
/// One read of the probe's subscription: the sample the rule is handed, and
/// what the IOReport sample it came from listed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct Read {
pub(super) sample: Sample,
pub(super) listed: Listed,
}
/// A read, from the channels one IOReport sample listed, each as (group,
/// name, raw integer) the way the FFI read them. The sample keeps the
/// channels of the Energy Model group (any case) that have a name, in
/// order, each with ITS OWN raw integer. `listed` counts every channel, of
/// any group, and says whether all of them are in the Energy Model group (a
/// channel whose group could not be read is not).
pub(super) fn read_from(channels: impl IntoIterator<Item = Channel>) -> Read {
let mut sample = Sample::new();
let mut listed = Listed { channels: 0, group_only: true };
for (group, name, raw) in channels {
let in_group = group.as_deref().is_some_and(|g| g.eq_ignore_ascii_case("Energy Model"));
listed.channels += 1;
listed.group_only &= in_group;
if let (true, Some(name)) = (in_group, name) {
sample.push((name, raw));
}
}
Read { sample, listed }
}
/// A rail's identity from one sample to the next: its channel name, and
/// its ordinal among the channels of that sample with the same name. Names
/// alone are not unique: an M3 Ultra's Energy Model group lists 264 of its
/// 645 names twice (e.g. `ECPUDTL00`..`22`, once per die, with no `DIE_`
/// prefix). Its CPU aggregates are unique today, but two same-named rails
/// must stay two rails, each compared only with its own previous value:
/// keyed by name alone, one would be compared with the other, and a larger
/// value held beside a smaller one would read as an advance in every
/// interval.
pub(super) type RailKey = (String, u32);
/// What the rule was handed in one successful sample, keyed as the rule
/// keys it: every CPU rail, and every `GPU Energy` channel, each with its
/// raw integer. The probe keeps one per successful sample, in order, so a
/// test can compare the values the real probe decided on with an
/// independent reading, and check that its reads were fresh (the GPU rail
/// moves where the CPU rails are frozen).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[cfg_attr(not(test), allow(dead_code))]
pub(super) struct Seen {
pub(super) cpu: Vec<(RailKey, i64)>,
pub(super) gpu: Vec<(RailKey, i64)>,
}
/// Which channels of a sample a [`RailWatch`] watches. EVERY one of them
/// seen must keep advancing for the source to be live: each source's
/// figure sums every counter it watches (IOReport's CPU rails, one per die
/// on a multi-die part; RAPL's package counters), and a counter that held
/// is a domain that figure would claim without reading it.
#[derive(Debug, Clone, Copy)]
pub(super) struct Watched {
/// Whether a channel is one of the watched counters.
pub(super) pick: fn(&str) -> bool,
}
/// The IOReport watch: the CPU energy rails (`CPU Energy`, `DIE_<n>_CPU
/// Energy`), every one of which must keep advancing: the power reading
/// sums them all.
pub(super) const CPU_RAILS: Watched = Watched { pick: is_cpu_energy_rail };
/// The liveness rule, fed one whole sample at a time, with the wall-clock
/// reading at which it was taken. It watches the channels its [`Watched`]
/// picks — by default IOReport's CPU energy rails ([`CPU_RAILS`]); the
/// same rule decides RAPL's package counters ([`super::rapl::PACKAGES`]).
///
/// The source is live only when EVERY watched rail keeps advancing, each
/// one on its own, and at least one was seen: after some interval, all
/// three of these hold at once for every rail seen in any sample, the
/// first included —
///
/// - it advanced in at least [`MIN_LIVE_ADVANCES`] separate intervals;
/// - those advancing intervals are at least 80% of all the intervals
/// observed so far;
/// - the observation so far spans at least [`MIN_LIVE_WINDOW`] of wall
/// time, from the first sample to this one.
///
/// Every rail, because the figure sums every rail it watches: on an M3
/// Ultra `DIE_0_CPU Energy` and `DIE_1_CPU Energy` both go into the CPU
/// figure, so one die's live rail cannot vouch for a die that held. A rail
/// missing from some samples is still a rail: it advanced in none of the
/// intervals it was missing from.
///
/// The rule is evaluated after every interval, and the source is live from
/// the first interval at which all three hold for every rail. An interval is two
/// CONSECUTIVE samples and an advance is the rail's raw integer being
/// larger than in the sample just before. The watch keeps the previous
/// sample itself, so the comparison cannot drift to a fixed baseline.
///
/// Why so much: on macOS 27.0.1 the CPU rails are not permanently frozen
/// for an unprivileged reader; they are refreshed in rare batches. An M3
/// Ultra's `DIE_0_` / `DIE_1_CPU Energy` held for at least 22 minutes, then
/// stepped by +1,344,869 and +1,391,637 mJ in a single 100 ms sample and
/// held again; an M5 Max's `CPU Energy` did not change in 13,614 samples
/// over 25 minutes, yet had moved from 480,083,160 to 541,893,638 mJ earlier
/// that hour. And a refresh is not always one step: a watcher sampling the
/// Energy Model group every ~16 ms on an M5 Max saw `CPU Energy` go
/// 2,643,843,378 -> 2,705,607,863 (+61,764,485) and then, 64 ms later,
/// -> 2,705,611,298 (+3,435), with `MCPU0`, `MCPU1` and `PCPU` each stepping
/// twice too. The probe's intervals are ~31 ms, so a window holding such a
/// refresh sees the same rail advance in two separate intervals — which a
/// "two advances" rule called live, selecting the meter for the life of the
/// process, after which every later refresh lands whole in whichever
/// bracket straddles it. A live rail advances in nearly every interval
/// (the M5 Max GPU rail changes every ~28 ms), for as long as it is
/// watched; a refresh advances in a few intervals close together and then
/// holds. So the rule asks for many advances (a few steps are not enough),
/// in nearly every interval (steps followed by a hold fall under 80%), over
/// a window a refresh does not fill (to pass, it would have to keep
/// stepping in four of every five intervals for 300 ms — which is what a
/// live rail does). Compared with a fixed baseline instead of the previous
/// sample, one jump would read as an advance in every later interval.
/// Counting per rail keeps two dies' separate steps from adding up to a
/// false "live".
#[derive(Debug)]
pub(super) struct RailWatch {
/// Which channels are watched.
watched: Watched,
/// Sample intervals observed so far (successful samples after the first).
intervals: u32,
/// The wall-clock reading of the first successful sample: where the
/// observation window starts. `None` until the first sample.
first_at: Option<Duration>,
/// The CPU rails of the previous successful sample, with their raw
/// integers. `None` until the first sample.
prev: Option<Vec<(RailKey, i64)>>,
/// Every watched rail seen in any sample, the first included, with the
/// number of intervals in which it advanced.
rails: Vec<(RailKey, u32)>,
/// Whether every rail seen has met the rule after some interval. Once
/// set, it stays set: the rule declares Live as soon as all three
/// conditions hold for every rail, and the probe stops sampling there.
live: bool,
/// What each successful sample handed the rule, in order.
seen: Vec<Seen>,
}
impl Default for RailWatch {
/// A watch of IOReport's CPU energy rails ([`CPU_RAILS`]).
fn default() -> Self {
Self::new(CPU_RAILS)
}
}
impl RailWatch {
/// A fresh watch of the channels `watched` picks.
pub(super) fn new(watched: Watched) -> Self {
Self { watched, intervals: 0, first_at: None, prev: None, rails: Vec::new(), live: false, seen: Vec::new() }
}
/// Fold in the next sample, taken at wall-clock reading `at`. The first
/// one is the baseline and starts the window; each later one closes an
/// interval against the sample just before it, and then the rule is
/// evaluated. A rail advances at most once per interval, and only if it
/// was in both samples.
pub(super) fn observe(&mut self, sample: &[(String, i64)], at: Duration) {
let cur = keyed(sample, self.watched.pick);
self.seen.push(Seen { cpu: cur.clone(), gpu: keyed(sample, is_gpu_energy_rail) });
for (key, _) in &cur {
if !self.rails.iter().any(|(k, _)| k == key) {
self.rails.push((key.clone(), 0));
}
}
let first_at = *self.first_at.get_or_insert(at);
if let Some(prev) = self.prev.take() {
self.intervals += 1;
for (key, now) in &cur {
// Keys are unique within a sample, so this is one advance
// per rail per interval at most.
if prev.iter().any(|(k, before)| k == key && now > before) {
if let Some((_, n)) = self.rails.iter_mut().find(|(k, _)| k == key) {
*n += 1;
}
}
}
let window = at.saturating_sub(first_at);
let meets = |(_, n): &(RailKey, u32)| *n >= MIN_LIVE_ADVANCES && sustained(*n, self.intervals);
let met = !self.rails.is_empty() && self.rails.iter().all(meets);
self.live = self.live || (window >= MIN_LIVE_WINDOW && met);
}
self.prev = Some(cur);
}
/// Sample intervals observed so far.
pub(super) fn intervals(&self) -> u32 {
self.intervals
}
/// Has every watched rail kept advancing — met the rule after some
/// interval?
pub(super) fn is_live(&self) -> bool {
self.live
}
/// Keep sampling? Until the rails have proved live, or `budget` has elapsed
/// **and** at least [`MIN_LIVE_ADVANCES`] intervals were observed, so a
/// live rail could have shown that many advances — slow IOReport
/// samples must not make a live rail look frozen — but never past
/// [`MAX_BUDGETS`] budgets.
pub(super) fn wants_more(&self, elapsed: Duration, budget: Duration) -> bool {
!self.is_live()
&& (elapsed < budget
|| (self.intervals < MIN_LIVE_ADVANCES && elapsed < budget * MAX_BUDGETS))
}
/// The verdict, or `None` if not a single interval could be observed
/// (the caller reports IOReport as unavailable).
pub(super) fn verdict(&self) -> Option<CpuRail> {
if self.intervals == 0 {
None
} else if self.is_live() {
Some(CpuRail::Live)
} else if self.rails.is_empty() {
Some(CpuRail::Absent)
} else {
Some(CpuRail::Frozen)
}
}
}
/// The channels of one sample that `pick` accepts, keyed by name and
/// ordinal among the same-named channels, in channel order.
fn keyed(sample: &[(String, i64)], pick: fn(&str) -> bool) -> Vec<(RailKey, i64)> {
let mut rails: Vec<(RailKey, i64)> = Vec::new();
for (name, raw) in sample {
if pick(name) {
let ordinal = rails.iter().filter(|((n, _), _)| n == name).count() as u32;
rails.push(((name.clone(), ordinal), *raw));
}
}
rails
}
/// One read of the probe's subscription, as the probe recorded it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg_attr(not(test), allow(dead_code))]
pub(super) struct SampleCall {
/// The probe's wall clock when the read began, and when it returned.
pub(super) start: Duration,
pub(super) end: Duration,
/// The probing thread's CPU clock just before and just after the read;
/// `None` where it could not be read.
pub(super) cpu_before: Option<Duration>,
pub(super) cpu_after: Option<Duration>,
/// What the IOReport sample listed, for a read that returned one;
/// `None` for a read that returned no sample.
pub(super) listed: Option<Listed>,
}
/// What one run of the liveness probe did: its verdict, and a record of how
/// it got there — every read, every tick, and what each successful read
/// handed the rule — so tests check how the real probe sampled and what it
/// decided on, not only what it decided. Outside tests only `opened` and
/// the verdict are read.
#[derive(Debug, Clone, Default)]
#[cfg_attr(not(test), allow(dead_code))]
pub(super) struct ProbeRun {
pub(super) verdict: Option<CpuRail>,
pub(super) intervals: u32,
/// Whether the subscription opened.
pub(super) opened: bool,
/// How many times the probe ticked (burned CPU before a re-read).
pub(super) ticks: u32,
/// What each of those ticks did, in order. `ticks` is its length.
pub(super) tick_outcomes: Vec<crate::meter::probe_clock::TickOutcome>,
/// Every read, in order, the baseline included.
pub(super) calls: Vec<SampleCall>,
/// What each successful read handed the rule, in order: one per
/// `calls` entry whose `listed` is `Some`.
pub(super) seen: Vec<Seen>,
}
#[cfg_attr(not(test), allow(dead_code))]
impl ProbeRun {
/// The probing thread's CPU time consumed in each interval between two
/// consecutive successful reads, outside the reads themselves: from the
/// end of one successful read to the start of the next. One entry per
/// interval; `None` for an interval where the CPU clock could not be
/// read.
pub(super) fn interval_cpu(&self) -> Vec<Option<Duration>> {
let mut out = Vec::new();
let mut prev_end: Option<Option<Duration>> = None;
for c in self.calls.iter().filter(|c| c.listed.is_some()) {
if let Some(end) = prev_end {
out.push(match (end, c.cpu_before) {
(Some(end), Some(start)) => start.checked_sub(end),
_ => None,
});
}
prev_end = Some(c.cpu_after);
}
out
}
/// The wall time each interval between two consecutive successful reads
/// spanned, outside the reads themselves. The companion to
/// [`ProbeRun::interval_cpu`]: a tick stops on the CPU target OR the
/// wall cap, so the CPU figure alone cannot say whether a short tick was
/// starved into the cap or stopped early.
pub(super) fn interval_wall(&self) -> Vec<Duration> {
let mut out = Vec::new();
let mut prev_end: Option<Duration> = None;
for c in self.calls.iter().filter(|c| c.listed.is_some()) {
if let Some(end) = prev_end {
out.push(c.start.saturating_sub(end));
}
prev_end = Some(c.end);
}
out
}
/// The median thread CPU time one read took (the upper one of an even
/// count), over the reads whose CPU clock could be read; `None` if
/// there are none.
pub(super) fn median_read_cpu(&self) -> Option<Duration> {
let mut cpu: Vec<Duration> =
self.calls.iter().filter_map(|c| c.cpu_after?.checked_sub(c.cpu_before?)).collect();
cpu.sort();
cpu.get(cpu.len() / 2).copied()
}
/// The median wall time one read took; logged, never asserted.
pub(super) fn median_read_wall(&self) -> Option<Duration> {
let mut wall: Vec<Duration> = self.calls.iter().map(|c| c.end.saturating_sub(c.start)).collect();
wall.sort();
wall.get(wall.len() / 2).copied()
}
}
/// The whole probe, generic over its parts — how the subscription is opened
/// and read, how the thread burns CPU between reads, and its wall and CPU
/// clocks — so this very loop, and the record it keeps, run on scripted
/// fakes in the tests. `host_status` runs it on the real parts.
///
/// Opens the subscription, takes a baseline read, then ticks and re-reads
/// for as long as [`RailWatch::wants_more`] says to. Each successful read
/// is handed to the rule with the wall-clock reading taken when it
/// returned (its record's `end`). A failed read (`None`) is not an
/// interval: the next successful one is compared with the last successful
/// one. A subscription that cannot be opened is never read or
/// ticked and gives no verdict; so does one that opens but never yields two
/// consecutive samples. Neither is `Absent`: that needs sampled intervals
/// with no CPU rail in them. The subscription is dropped (released) on
/// return.
pub(super) fn probe_with<S>(
open: impl FnOnce() -> Option<S>,
read: impl FnMut(&S) -> Option<Read>,
tick: impl FnMut() -> crate::meter::probe_clock::TickOutcome,
wall: impl Fn() -> Duration,
cpu: impl Fn() -> Option<Duration>,
budget: Duration,
) -> ProbeRun {
probe_watching(CPU_RAILS, open, read, tick, wall, cpu, budget)
}
/// [`probe_with`], watching the channels `watched` picks: the same loop,
/// record and rule for every source this meter probes.
pub(super) fn probe_watching<S>(
watched: Watched,
open: impl FnOnce() -> Option<S>,
mut read: impl FnMut(&S) -> Option<Read>,
mut tick: impl FnMut() -> crate::meter::probe_clock::TickOutcome,
wall: impl Fn() -> Duration,
cpu: impl Fn() -> Option<Duration>,
budget: Duration,
) -> ProbeRun {
let mut run = ProbeRun::default();
let Some(sub) = open() else {
return run;
};
run.opened = true;
let started = wall();
let mut watch = RailWatch::new(watched);
loop {
let (start, cpu_before) = (wall(), cpu());
let got = read(&sub);
let (cpu_after, end) = (cpu(), wall());
run.calls.push(SampleCall { start, end, cpu_before, cpu_after, listed: got.as_ref().map(|r| r.listed) });
if let Some(got) = got {
watch.observe(&got.sample, end);
}
if !watch.wants_more(wall().saturating_sub(started), budget) {
break;
}
run.tick_outcomes.push(tick());
run.ticks += 1;
}
run.verdict = watch.verdict();
run.intervals = watch.intervals();
run.seen = watch.seen;
run
}
/// Whether a channel is a CPU energy rail: its name, after stripping a
/// `DIE_<n>_` prefix, is `CPU Energy` (any case). Not the per-cluster or
/// per-core sub-rails (`PCPU`, `MCPU*`, `PACC_*`, `DIE_0_PACC0_CPU0`).
pub(super) fn is_cpu_energy_rail(name: &str) -> bool {
strip_die_prefix(name).eq_ignore_ascii_case("CPU Energy")
}
/// Whether a channel is the GPU energy aggregate: `GPU Energy` (any case,
/// after an optional `DIE_<n>_` prefix). Not a liveness signal — the probe
/// records it only so tests can see its reads were fresh.
pub(super) fn is_gpu_energy_rail(name: &str) -> bool {
strip_die_prefix(name).eq_ignore_ascii_case("GPU Energy")
}
/// Whether the power reading sums this Energy Model rail, so that energy is
/// counted once. The aggregates `CPU Energy`, `GPU Energy` and `ANE Energy`
/// (optionally `DIE_<n>_`-prefixed) are summed; their summands are not. A
/// summand is any other rail whose name says CPU, GPU or ANE, and also the
/// CPU summands whose names do not: per-cluster accumulators (`PACC*`,
/// `EACC*`, `MACC*`) and CPU power managers (`*CPM*`, with their `_SRAM`
/// rails). From absolute samples of the Energy Model group on each chip: on
/// an M5 Max, `CPU Energy` = `MCPU0` + `MCPU1` + `PCPU`, and `PCPU` =
/// `PACC_0..5` + `PCPU0..5_SRAM` + `PCPM` + `PCPM_SRAM`; on an M3 Ultra,
/// each `DIE_<n>_CPU Energy` = its `EACC_CPU` + `PACC0_CPU` + `PACC1_CPU`,
/// each the sum of its cores, their SRAM and its `_CPM` + `_CPM_SRAM`
/// rails (all to within 1-8 mJ). Summing `PACC_*` or `*_CPM` beside
/// `CPU Energy` would count that CPU energy twice. Neither chip has an
/// `ANE Energy` aggregate, and `ANE0` (`ANE0_0`/`ANE0_1`) is dropped as a
/// summand, so ANE energy is not summed at all.
pub(super) fn counts_once(name: &str) -> bool {
let base = strip_die_prefix(name);
if base.eq_ignore_ascii_case("CPU Energy")
|| base.eq_ignore_ascii_case("GPU Energy")
|| base.eq_ignore_ascii_case("ANE Energy")
{
return true;
}
let u = base.to_ascii_uppercase();
if u.contains("CPU") || u.contains("GPU") || u.contains("ANE") {
return false;
}
let cpu_summand = ["PACC", "EACC", "MACC"].iter().any(|p| u.starts_with(p)) || u.contains("CPM");
!cpu_summand
}
/// Strip an optional `DIE_<n>_` prefix, `<n>` one or more digits, `DIE_`
/// in any case (multi-die parts like the M-series Ultra report per-die
/// aggregates).
pub(super) fn strip_die_prefix(name: &str) -> &str {
let b = name.as_bytes();
if b.len() > 4 && b[..4].eq_ignore_ascii_case(b"DIE_") {
// The first four bytes are ASCII, so index 4 is a char boundary.
let rest = &name[4..];
let digits = rest.bytes().take_while(u8::is_ascii_digit).count();
if digits > 0 {
if let Some(r) = rest[digits..].strip_prefix('_') {
return r;
}
}
}
name
}
#[cfg(test)]
pub(super) mod tests {
use super::*;
use std::cell::Cell;
const BUDGET: Duration = Duration::from_millis(1000);
const TICK: Duration = Duration::from_millis(25);
/// The probe's interval on a quiet M5 Max: one ~25 ms tick and a
/// ~6 ms read of the Energy Model group.
const QUIET: Duration = Duration::from_millis(31);
/// An interval on a loaded host.
const LOADED: Duration = Duration::from_millis(100);
fn ms(n: u64) -> Duration {
Duration::from_millis(n)
}
fn sample(pairs: &[(&str, i64)]) -> Sample {
pairs.iter().map(|(n, v)| (n.to_string(), *v)).collect()
}
/// Feed `samples` to a fresh watch, sample `i` at wall time `i * every`,
/// and return the watch and whether it was live after each sample.
fn watch(every: Duration, samples: &[Vec<(&str, i64)>]) -> (RailWatch, Vec<bool>) {
let mut w = RailWatch::default();
let mut live = Vec::new();
for (i, s) in samples.iter().enumerate() {
w.observe(&sample(s), every * i as u32);
live.push(w.is_live());
}
(w, live)
}
/// The first sample after which the watch was live, if any.
fn first_live(live: &[bool]) -> Option<usize> {
live.iter().position(|l| *l)
}
/// A read of a subscription to the Energy Model group alone, listing
/// exactly these channels.
pub(in super::super) fn read_of(pairs: &[(&str, i64)]) -> Read {
Read { sample: sample(pairs), listed: Listed { channels: pairs.len(), group_only: true } }
}
/// The (key, raw) values the rule keys out of `pairs` with `pick`: the
/// test's own keying, by name and ordinal.
pub(in super::super) fn expect_keyed(pairs: &[(&str, i64)], pick: impl Fn(&str) -> bool) -> Vec<(RailKey, i64)> {
let mut out: Vec<(RailKey, i64)> = Vec::new();
for (name, raw) in pairs.iter().filter(|(n, _)| pick(n)) {
let ordinal = out.iter().filter(|((n, _), _)| n == name).count() as u32;
out.push(((name.to_string(), ordinal), *raw));
}
out
}
/// What one scripted [`probe_with`] did.
#[derive(Debug)]
struct Run {
verdict: Option<CpuRail>,
intervals: u32,
/// How many times it asked for a read, the baseline included.
samples: u32,
/// The scripted clock when it stopped.
took: Duration,
run: ProbeRun,
}
/// Run [`probe_with`] on a subscription that opens, scripted reads
/// (`script(n)` is the n-th read, from 0), and a scripted clock that only
/// the tick advances, by `tick`.
fn run(tick: Duration, mut script: impl FnMut(u32) -> Option<Vec<(&'static str, i64)>>) -> Run {
let clock = Cell::new(Duration::ZERO);
let calls = Cell::new(0u32);
let run = probe_with(
|| Some(()),
|_: &()| {
let n = calls.get();
calls.set(n + 1);
script(n).map(|s| read_of(&s))
},
|| {
clock.set(clock.get() + tick);
crate::meter::probe_clock::TickOutcome { cpu: None, spins: 1 }
},
|| clock.get(),
|| None,
BUDGET,
);
assert_eq!(run.calls.len(), calls.get() as usize, "one record per read");
Run { verdict: run.verdict, intervals: run.intervals, samples: calls.get(), took: clock.get(), run }
}
#[test]
fn only_the_cpu_energy_aggregate_is_a_cpu_rail() {
assert!(is_cpu_energy_rail("CPU Energy"));
assert!(is_cpu_energy_rail("cpu energy"));
assert!(is_cpu_energy_rail("DIE_0_CPU Energy"));
assert!(is_cpu_energy_rail("DIE_1_CPU Energy"));
assert!(is_cpu_energy_rail("DIE_12_CPU Energy"));
assert!(is_cpu_energy_rail("die_1_CPU Energy"));
assert!(!is_cpu_energy_rail("DIE__CPU Energy"), "DIE_<n>_ needs a digit");
assert!(!is_cpu_energy_rail("DIE_x_CPU Energy"));
assert!(!is_cpu_energy_rail("GPU Energy"));
assert!(!is_cpu_energy_rail("DIE_0_GPU Energy"));
assert!(!is_cpu_energy_rail("ANE Energy"));
assert!(!is_cpu_energy_rail("PCPU"));
assert!(!is_cpu_energy_rail("MCPU0"));
assert!(!is_cpu_energy_rail("PACC_0"));
assert!(!is_cpu_energy_rail("DIE_0_PACC0_CPU0"));
assert!(!is_cpu_energy_rail("ECPUDTL00"));
assert!(is_gpu_energy_rail("GPU Energy"));
assert!(is_gpu_energy_rail("DIE_1_GPU Energy"));
assert!(!is_gpu_energy_rail("GPU0"));
assert!(!is_gpu_energy_rail("GPU0_0"));
assert!(!is_gpu_energy_rail("CPU Energy"));
}
/// A rail that moves in one step and then holds — a batch refresh caught
/// mid-probe — is not live, however large the step.
#[test]
fn a_single_jump_is_a_batch_refresh_not_a_live_rail() {
let script: Vec<_> =
[100, 100, 1_444_969, 1_444_969, 1_444_969, 1_444_969].iter().map(|v| vec![("CPU Energy", *v)]).collect();
let (w, live) = watch(LOADED, &script);
assert_eq!(w.intervals(), 5);
assert_eq!(first_live(&live), None, "one refresh counted as a live rail");
assert_eq!(w.verdict(), Some(CpuRail::Frozen));
}
/// Two advances of the same rail — here the second die, with a hold
/// between them — are not live, however far apart; the same rail
/// advancing in five of six intervals over 600 ms is, and not before.
/// The first die advances at every interval, so the verdict is the
/// second die's: every die must keep advancing.
#[test]
fn two_advances_of_the_same_rail_are_not_live_and_sustained_ones_are() {
let script: Vec<_> = [20, 21, 21, 22, 23, 24, 25, 26]
.iter()
.enumerate()
.map(|(i, v)| vec![("DIE_0_CPU Energy", 10 + i as i64), ("DIE_1_CPU Energy", *v)])
.collect();
let (w, live) = watch(LOADED, &script[..4]);
assert_eq!((w.intervals(), w.verdict()), (3, Some(CpuRail::Frozen)), "two advances over 300 ms: {live:?}");
let (w, live) = watch(LOADED, &script);
// Advances in intervals 1, 3, 4, 5, 6: the fifth closes at sample
// 6 (600 ms), where 5 of 6 intervals is over 80%.
assert_eq!(first_live(&live), Some(6), "{live:?}");
assert_eq!(w.verdict(), Some(CpuRail::Live));
}
/// A rail that goes DOWN between consecutive samples — a reset or a
/// wrap — did not advance. The rail that drops here has already
/// advanced in four intervals over 400 ms, so counting the drop would
/// make it live.
#[test]
fn a_decrease_between_consecutive_samples_is_not_an_advance() {
let values = [20, 21, 22, 23, 24, 3];
let script: Vec<_> =
values.iter().map(|v| vec![("DIE_0_CPU Energy", 100), ("DIE_1_CPU Energy", *v)]).collect();
let (w, live) = watch(LOADED, &script);
assert_eq!(w.intervals(), 5);
assert_eq!(first_live(&live), None, "a decrease counted as an advance");
assert_eq!(w.verdict(), Some(CpuRail::Frozen));
// The same through the loop: four advances, a drop, then a hold
// for the whole budget.
let r = run(LOADED, |n| Some(vec![("CPU Energy", values[n.min(5) as usize])]));
assert_eq!(r.verdict, Some(CpuRail::Frozen), "{r:?}");
assert_eq!((r.intervals, r.took), (10, BUDGET), "{r:?}");
}
/// A rail absent from the previous sample only sets its baseline: its
/// first appearance is not an advance, whatever its value. The rail
/// that appears here then advances in four intervals over 500 ms, so
/// counting its appearance would make it live — both for a new name and
/// for a second channel with a name already present.
#[test]
fn a_rail_new_in_this_sample_only_sets_its_baseline() {
for name in ["DIE_1_CPU Energy", "CPU Energy"] {
let mut script = vec![vec![("CPU Energy", 1)]];
for v in 5..=9 {
script.push(vec![("CPU Energy", 1), (name, v)]);
}
let (w, live) = watch(LOADED, &script);
assert_eq!(w.intervals(), 5);
assert_eq!(first_live(&live), None, "{name}: a rail's first appearance counted as an advance");
assert_eq!(w.verdict(), Some(CpuRail::Frozen));
}
}
/// A read keeps each Energy Model channel with its OWN raw integer, in
/// order, and drops other groups and unnamed channels — no rail is
/// given another rail's value, a sum, or zero — and it says what the
/// IOReport sample listed: every channel, of any group, and whether all
/// of them are in the Energy Model group.
#[test]
fn a_read_keeps_each_energy_model_channel_with_its_own_value_and_says_what_was_listed() {
let em = || Some("Energy Model".to_string());
let name = |n: &str| Some(n.to_string());
let r = read_from(vec![
(em(), name("DIE_0_CPU Energy"), 9_494_336),
(name("CPU Stats"), name("CPU Energy"), 1),
(name("energy model"), name("DIE_1_CPU Energy"), 10_212_656),
(em(), None, 5),
(None, name("GPU Energy"), 7),
(em(), name("GPU Energy"), 14_809_813),
]);
assert_eq!(
r.sample,
sample(&[("DIE_0_CPU Energy", 9_494_336), ("DIE_1_CPU Energy", 10_212_656), ("GPU Energy", 14_809_813)]),
"each Energy Model channel keeps its own raw integer"
);
assert_eq!(r.listed, Listed { channels: 6, group_only: false }, "a channel of another group was listed");
let r = read_from(vec![(em(), name("CPU Energy"), 3), (em(), None, 4), (name("ENERGY MODEL"), name("GPU0"), 5)]);
assert_eq!(r.listed, Listed { channels: 3, group_only: true }, "every channel is in the group");
assert_eq!(r.sample, sample(&[("CPU Energy", 3), ("GPU0", 5)]));
let r = read_from(vec![(em(), name("CPU Energy"), 3), (None, name("GPU Energy"), 4)]);
assert_eq!(r.listed, Listed { channels: 2, group_only: false }, "an unreadable group is not the group");
assert_eq!(read_from(Vec::new()).listed, Listed { channels: 0, group_only: true });
}
/// No observed interval at all is no verdict; intervals with no CPU
/// channel in them are `Absent`.
#[test]
fn no_interval_is_no_verdict_and_no_cpu_rail_is_absent() {
let mut w = RailWatch::default();
assert_eq!(w.verdict(), None);
w.observe(&sample(&[("CPU Energy", 1)]), BUDGET);
assert_eq!(w.verdict(), None, "a baseline alone is not an interval");
let script: Vec<_> = (0..8).map(|n| vec![("GPU Energy", n), ("PCPU", n)]).collect();
let (w, _) = watch(LOADED, &script);
assert_eq!(w.intervals(), 7);
assert_eq!(w.verdict(), Some(CpuRail::Absent), "a moving GPU rail and a CPU sub-rail are no CPU rail");
}
/// (a) The M5 Max's refresh: `CPU Energy` holds at 480,083,160 mJ,
/// steps once to 541,893,638 and holds, while `GPU Energy` advances in
/// every sample. Frozen. Compared with the FIRST sample instead of the
/// previous one, every interval after the step would read as an advance.
#[test]
fn watch_m5_max_refresh_is_frozen() {
let r = run(TICK, |n| {
let cpu = if n < 2 { 480_083_160 } else { 541_893_638 };
Some(vec![("CPU Energy", cpu), ("GPU Energy", 851_622_003 + 28_000_000 * i64::from(n))])
});
assert_eq!(r.verdict, Some(CpuRail::Frozen), "{r:?}");
// One interval per 25 ms tick until the 1 s budget is spent.
assert_eq!((r.intervals, r.samples, r.took), (40, 41, BUDGET), "{r:?}");
assert_eq!(r.run.ticks, 40, "one tick before every re-read");
}
/// (b) The M3 Ultra's refresh: both dies step once, in the same sample,
/// among unprefixed per-die sub-rails listed once per die and a GPU
/// rail that moves now and then. Frozen.
#[test]
fn watch_m3_ultra_two_die_refresh_is_frozen() {
let r = run(TICK, |n| {
let (d0, d1) = if n < 7 { (9_494_336, 10_212_656) } else { (10_839_205, 11_604_293) };
let n = i64::from(n);
Some(vec![
("ECPUDTL00", 5_000 + n),
("DIE_0_CPU Energy", d0),
("ECPUDTL00", 7_000 + n),
("DIE_1_CPU Energy", d1),
("GPU Energy", 14_809_813 * (n / 5)),
])
});
assert_eq!(r.verdict, Some(CpuRail::Frozen), "{r:?}");
assert_eq!(r.intervals, 40, "{r:?}");
}
/// (c) Two channels both named `CPU Energy`, holding at different
/// values and stepping together once, are two rails, each compared with
/// its own previous value: keyed by name alone, the larger would read
/// as an advance over the smaller in every interval. Frozen.
#[test]
fn watch_two_same_named_rails_stepping_together_once_is_frozen() {
let r = run(TICK, |n| {
let step = if n < 3 { 0 } else { 1_300_000 };
Some(vec![("CPU Energy", 9_494_336 + step), ("CPU Energy", 10_212_656 + step)])
});
assert_eq!(r.verdict, Some(CpuRail::Frozen), "{r:?}");
assert_eq!(r.intervals, 40, "{r:?}");
}
/// (d) A rail that advances in every interval, at the quiet host's
/// ~31 ms, is Live — but not before five advances AND 300 ms: it has
/// five advances at 155 ms and is still not live until the tenth
/// interval closes at 310 ms, where the watch stops sampling. Once
/// declared, Live stays: a later hold does not undo it.
#[test]
fn watch_a_rail_advancing_every_interval_is_live_only_after_five_advances_and_300_ms() {
let r = run(QUIET, |n| Some(vec![("CPU Energy", 1_000 + i64::from(n)), ("GPU Energy", 5)]));
assert_eq!(r.verdict, Some(CpuRail::Live), "{r:?}");
assert_eq!((r.intervals, r.samples, r.took), (10, 11, QUIET * 10), "{r:?}");
// Eleven advances, then a hold long enough (from the 14th interval)
// to take the share under 80%.
let script: Vec<_> = (0..20).map(|n| vec![("CPU Energy", 1_000 + n.min(11))]).collect();
let (w, live) = watch(QUIET, &script);
assert!(!live[5], "five advances in 155 ms counted as live");
assert!(!live[9], "nine advances in 279 ms counted as live");
assert_eq!(first_live(&live), Some(10), "{live:?}");
assert!(live[10..].iter().all(|l| *l), "Live was undone by a later hold: {live:?}");
assert_eq!(w.verdict(), Some(CpuRail::Live));
}
/// (*) The refresh observed on an M5 Max (macOS 27.0.1), as the probe
/// sees it at ~31 ms an interval: `CPU Energy` steps +61,764,485 and,
/// one interval later, +3,435, then holds — and `MCPU0`, `MCPU1` and
/// `PCPU` step twice with it, while the GPU rail keeps moving. Two
/// advancing intervals of the same rail, which the "two advances" rule
/// called Live. Frozen, after the whole budget.
#[test]
fn watch_the_observed_two_step_refresh_is_frozen() {
const A: i64 = 2_643_843_378;
let step = |n: u32, first: i64, second: i64| match n {
0..=2 => 0,
3 => first,
_ => first + second,
};
let r = run(QUIET, |n| {
Some(vec![
("MCPU0", 91_000_000 + step(n, 20_000_000, 1_000)),
("MCPU1", 92_000_000 + step(n, 21_000_000, 1_100)),
("PCPU", 400_000_000 + step(n, 20_764_485, 1_335)),
("CPU Energy", A + step(n, 61_764_485, 3_435)),
("GPU Energy", 851_622_003 + 28_000_000 * i64::from(n)),
])
});
assert_eq!(r.verdict, Some(CpuRail::Frozen), "a two-step refresh counted as a live rail: {r:?}");
assert_eq!((r.intervals, r.samples, r.took), (33, 34, QUIET * 33), "{r:?}");
// The rule was handed the refresh exactly as observed.
let cpu: Vec<i64> = r.run.seen.iter().map(|s| s.cpu[0].1).collect();
assert_eq!(&cpu[..6], &[A, A, A, 2_705_607_863, 2_705_611_298, 2_705_611_298]);
}
/// (*) A refresh published in three, four or five steps within its
/// first 150 ms (30 ms intervals) — consecutive, or spread out — then
/// holding: Frozen. Five steps even meet the count, but never with
/// 300 ms of window at 80%.
#[test]
fn watch_a_refresh_in_three_to_five_steps_at_the_start_is_frozen() {
const EVERY: Duration = Duration::from_millis(30);
let patterns: [&[u32]; 5] = [&[1, 2, 3], &[1, 2, 3, 4], &[1, 2, 3, 4, 5], &[1, 3, 5], &[2, 3, 4, 5]];
for steps in patterns {
let r = run(EVERY, move |n| {
let stepped = steps.iter().filter(|&&s| s <= n).count() as i64;
Some(vec![("CPU Energy", 2_643_843_378 + 1_000_000 * stepped), ("GPU Energy", i64::from(n))])
});
assert_eq!(r.verdict, Some(CpuRail::Frozen), "a refresh in steps {steps:?} counted as live: {r:?}");
assert_eq!(r.intervals, 34, "steps {steps:?}: {r:?}");
}
}
/// (*) The same two-step refresh in the M3 Ultra's shape (the two
/// steps were seen on an M5 Max; the M3 Ultra's dies were seen
/// stepping once, in one sample): `DIE_0_CPU Energy` and
/// `DIE_1_CPU Energy` each step twice within 64 ms (in consecutive
/// intervals, and two intervals apart), among unprefixed per-die
/// sub-rails listed once per die and a GPU rail that moves now and
/// then. Frozen.
#[test]
fn watch_m3_ultra_two_dies_each_stepping_twice_is_frozen() {
let r = run(QUIET, |n| {
let d0 = 9_494_336 + if n >= 3 { 1_344_000 } else { 0 } + if n >= 4 { 869 } else { 0 };
let d1 = 10_212_656 + if n >= 3 { 1_391_000 } else { 0 } + if n >= 5 { 637 } else { 0 };
let n = i64::from(n);
Some(vec![
("ECPUDTL00", 5_000 + n),
("DIE_0_CPU Energy", d0),
("ECPUDTL00", 7_000 + n),
("DIE_1_CPU Energy", d1),
("GPU Energy", 14_809_813 * (n / 5)),
])
});
assert_eq!(r.verdict, Some(CpuRail::Frozen), "{r:?}");
assert_eq!(r.intervals, 33, "{r:?}");
}
/// A live rail that misses one interval in every five advances in 80%
/// of them: Live, at the tenth interval (310 ms), where it has 8 of 10
/// — exactly 80%. One that misses two in every five (60%) is never
/// Live, over the whole budget.
#[test]
fn watch_a_live_rail_missing_one_interval_in_five_is_live_and_two_in_five_is_not() {
let misses_one = |n: u32| {
let advances = n - n / 5; // no advance in intervals 5, 10, 15, ...
Some(vec![("CPU Energy", 1_000 + i64::from(advances))])
};
let r = run(QUIET, misses_one);
assert_eq!(r.verdict, Some(CpuRail::Live), "{r:?}");
assert_eq!((r.intervals, r.took), (10, QUIET * 10), "{r:?}");
let misses_two = |n: u32| {
let advances = n - n / 5 - (n + 2) / 5; // none in intervals 3, 5, 8, 10, ...
Some(vec![("CPU Energy", 1_000 + i64::from(advances))])
};
let r = run(QUIET, misses_two);
assert_eq!(r.verdict, Some(CpuRail::Frozen), "a rail advancing in 60% of intervals counted as live: {r:?}");
assert_eq!(r.intervals, 33, "{r:?}");
}
/// A rail that missed two of its first five intervals is not yet Live
/// — even at intervals 7, 8 and 9 of a loaded host, where it has five,
/// six and seven advances and 700–900 ms of window — until its share
/// of advancing intervals reaches 80%, at the tenth (8 of 10).
#[test]
fn watch_a_rail_missing_two_of_its_first_five_intervals_is_not_yet_live() {
// No advance in intervals 2 and 4, then one in every interval.
let values = [0, 1, 1, 2, 2, 3, 4, 5, 6, 7, 8, 9];
let script: Vec<_> = values.iter().map(|v| vec![("CPU Energy", 1_000 + *v)]).collect();
let (w, live) = watch(LOADED, &script);
assert_eq!(&live[..10], &[false; 10], "Live below 80% of intervals: {live:?}");
assert_eq!(first_live(&live), Some(10), "{live:?}");
assert_eq!(w.verdict(), Some(CpuRail::Live));
}
/// A live rail on a loaded host, sampled every 100 ms, has 300 ms of
/// window after three intervals but is Live only at the fifth advance
/// (500 ms), where the probe stops.
#[test]
fn watch_a_live_rail_on_a_loaded_host_is_live_after_five_intervals() {
let r = run(LOADED, |n| Some(vec![("CPU Energy", 1_000 + i64::from(n))]));
assert_eq!(r.verdict, Some(CpuRail::Live), "{r:?}");
assert_eq!((r.intervals, r.samples, r.took), (5, 6, LOADED * 5), "{r:?}");
let script: Vec<_> = (0..6).map(|n| vec![("CPU Energy", 1_000 + n)]).collect();
let (_, live) = watch(LOADED, &script);
assert_eq!(live, [false, false, false, false, false, true], "Live before five advances");
}
/// (e) Samples that keep failing are retried up to the five-budget cap
/// and give no verdict — IOReport unavailable — with or without a
/// baseline.
#[test]
fn watch_failing_samples_run_to_the_cap_with_no_verdict() {
let r = run(TICK, |_| None);
assert_eq!((r.verdict, r.intervals), (None, 0), "{r:?}");
assert_eq!(r.took, BUDGET * MAX_BUDGETS, "{r:?}");
assert_eq!(r.samples, 1 + (BUDGET * MAX_BUDGETS).as_millis() as u32 / 25, "{r:?}");
assert!(r.run.seen.is_empty() && r.run.calls.iter().all(|c| c.listed.is_none()), "{r:?}");
let r = run(TICK, |n| (n == 0).then(|| vec![("CPU Energy", 1)]));
assert_eq!((r.verdict, r.intervals, r.took), (None, 0, BUDGET * MAX_BUDGETS), "{r:?}");
}
/// (f) Past the budget, the watch keeps sampling while it has fewer
/// than five intervals and stops once it has five (Frozen). Samples
/// fail until the clock is past the budget: at 1.2 s one interval, and
/// on through 2.4 s (four) → keep going; at 2.8 s five → stop.
#[test]
fn watch_past_the_budget_stops_only_with_five_intervals() {
let r = run(ms(400), |n| (n == 0 || n >= 3).then(|| vec![("CPU Energy", 7)]));
assert_eq!(r.verdict, Some(CpuRail::Frozen), "{r:?}");
assert_eq!((r.intervals, r.samples, r.took), (5, 8, ms(2800)), "{r:?}");
// A failed sample between two good ones is not an interval, and the
// next good one is compared with the last good one: a rail read
// every 50 ms is Live at the sixth interval (300 ms).
let r = run(TICK, |n| (n % 2 == 0).then(|| vec![("CPU Energy", 1_000 + i64::from(n))]));
assert_eq!(r.verdict, Some(CpuRail::Live), "{r:?}");
assert_eq!((r.intervals, r.samples, r.took), (6, 13, ms(300)), "{r:?}");
}
/// (f') The cap: a watch that cannot get five intervals stops at five
/// budgets, Frozen — even when its rail advanced in every interval it
/// did get, since three advances are not a live rail.
#[test]
fn watch_short_of_five_intervals_stops_at_the_cap() {
// Only every fourth read succeeds: reads at 0, 1.6, 3.2 and 4.8 s,
// three intervals; the next read, at 5.2 s, fails and the cap ends it.
let r = run(ms(400), |n| (n % 4 == 0).then(|| vec![("CPU Energy", 1_000 + i64::from(n))]));
assert_eq!(r.verdict, Some(CpuRail::Frozen), "{r:?}");
assert_eq!((r.intervals, r.samples, r.took), (3, 14, ms(5200)), "{r:?}");
assert!(r.took >= BUDGET * MAX_BUDGETS, "stopped before the cap: {r:?}");
}
/// (g) Two dies advancing in alternate intervals — each in half of
/// them — do not add up to a live rail: liveness is per rail, not
/// summed. Through the whole loop, so a sum placed anywhere between the
/// read and the rule advances in every interval, reads as live, and
/// fails here; and the record shows each die's own values, as read.
#[test]
fn watch_two_dies_advancing_in_alternate_intervals_is_frozen() {
let script = |n: u32| {
vec![
("DIE_0_CPU Energy", 100 + 50 * i64::from(n.div_ceil(2))),
("DIE_1_CPU Energy", 200 + 50 * i64::from(n / 2)),
("GPU Energy", 1_000 + i64::from(n)),
]
};
let r = run(TICK, |n| Some(script(n)));
assert_eq!(r.verdict, Some(CpuRail::Frozen), "{r:?}");
assert_eq!(r.intervals, 40, "{r:?}");
assert_eq!(r.run.seen.len(), 41, "one record per successful read");
for (n, seen) in r.run.seen.iter().enumerate() {
let s = script(n as u32);
assert_eq!(seen.cpu, expect_keyed(&s, |c| c.ends_with("CPU Energy")), "read {n}: what the rule was handed");
assert_eq!(seen.gpu, expect_keyed(&s, |c| c == "GPU Energy"), "read {n}: the GPU rail as read");
}
}
// ── Every CPU rail: the power reading sums each die's rail ──────────
/// The two CPU rails of an M3 Ultra.
const DIES: (&str, &str) = ("DIE_0_CPU Energy", "DIE_1_CPU Energy");
/// A rail's values over 32 intervals: +7 in interval `i` (from 1)
/// exactly when `advances(i)`, held otherwise.
fn rail_where(advances: impl Fn(usize) -> bool) -> Vec<i64> {
let mut v = vec![1_000_000i64];
for i in 1..=32 {
let last = v[i - 1];
v.push(if advances(i) { last + 7 } else { last });
}
v
}
/// The two dies at these values, sample by sample, beside a GPU rail
/// that moves at every sample.
fn dies(v0: &[i64], v1: &[i64]) -> Vec<Vec<(&'static str, i64)>> {
assert_eq!(v0.len(), v1.len());
(0..v0.len()).map(|i| vec![(DIES.0, v0[i]), (DIES.1, v1[i]), ("GPU Energy", i as i64)]).collect()
}
/// (a) Both dies keep advancing: Live, at the interval where the
/// slower die is, never before, whichever die that is. Both at every
/// interval: the tenth, where the window reaches 300 ms at ~31 ms an
/// interval, as for one rail. One missing an interval in five: the
/// tenth too (eight of ten). One missing its first three: the
/// fifteenth (twelve of fifteen), though the other alone was ready at
/// the tenth.
#[test]
fn watch_two_dies_that_both_keep_advancing_is_live() {
let every = rail_where(|_| true);
for (what, slower, at) in [
("at every interval", rail_where(|_| true), 10),
("missing one interval in five", rail_where(|i| i % 5 != 0), 10),
("missing its first three intervals", rail_where(|i| i > 3), 15),
] {
for samples in [dies(&every, &slower), dies(&slower, &every)] {
let (w, live) = watch(QUIET, &samples);
assert_eq!(first_live(&live), Some(at), "one die {what}: {w:?}");
assert!(live[at..].iter().all(|l| *l), "one die {what}: Live was undone: {live:?}");
assert_eq!(w.verdict(), Some(CpuRail::Live), "one die {what}");
assert_eq!(w.rails.len(), 2, "two dies, two rails: {w:?}");
}
}
}
/// (b) One die keeps advancing at every interval while the other does
/// not: Frozen, and never Live at any interval, whichever die it is.
/// The other held for the whole budget (its energy is in the CPU
/// figure, unread), stepped once, stepped five times in a refresh, or
/// advanced in three intervals of five. The one-rail rule called each
/// of these Live at the tenth interval. The same for two channels both
/// named `CPU Energy`.
#[test]
fn watch_one_die_that_keeps_advancing_beside_a_frozen_one_is_frozen() {
let every = rail_where(|_| true);
for (what, other) in [
("held", rail_where(|_| false)),
("stepped once", rail_where(|i| i == 4)),
("stepped five times in a refresh", rail_where(|i| i <= 5)),
("advanced in three intervals of five", rail_where(|i| i % 5 != 0 && i % 5 != 4)),
] {
for samples in [dies(&every, &other), dies(&other, &every)] {
let (w, live) = watch(QUIET, &samples);
assert_eq!(first_live(&live), None, "the other die {what}: {w:?}");
assert_eq!(w.verdict(), Some(CpuRail::Frozen), "the other die {what}");
assert!(w.rails.iter().any(|(_, n)| *n == 32), "one die advanced in every interval: {w:?}");
}
}
let held = rail_where(|_| false);
let twins: Vec<Vec<(&str, i64)>> =
every.iter().zip(&held).map(|(a, b)| vec![("CPU Energy", *a), ("CPU Energy", *b)]).collect();
let (w, live) = watch(QUIET, &twins);
assert_eq!(first_live(&live), None, "twin channels, one held: {w:?}");
assert_eq!(w.rails, [(("CPU Energy".to_string(), 0), 32), (("CPU Energy".to_string(), 1), 0)]);
assert_eq!(w.verdict(), Some(CpuRail::Frozen));
}
/// (c) One die that keeps advancing, and no other die in any sample:
/// a part whose CPU figure sums that one rail. Live at the tenth
/// interval, exactly as `CPU Energy` alone is.
#[test]
fn watch_one_die_alone_that_keeps_advancing_is_live() {
let samples: Vec<Vec<(&str, i64)>> =
rail_where(|_| true).iter().enumerate().map(|(i, v)| vec![(DIES.0, *v), ("GPU Energy", i as i64)]).collect();
let (w, live) = watch(QUIET, &samples);
assert_eq!(first_live(&live), Some(10), "{w:?}");
assert_eq!(w.rails, [((DIES.0.to_string(), 0), 32)]);
assert_eq!(w.verdict(), Some(CpuRail::Live));
}
/// A die in only some of the samples is a rail the source needs,
/// judged over every interval: it advanced in none of those it was
/// missing from. DIE_1 in the first sample alone, in the first five
/// alone, or from the third on and holding: Frozen, never Live. DIE_1
/// from the fifth sample on, advancing at every interval after: Live
/// only once its advances are 80% of every interval, at the twentieth
/// (sixteen of twenty), though DIE_0 was ready at the tenth.
#[test]
fn watch_a_die_in_only_some_samples_is_a_rail_the_watch_needs() {
let v = rail_where(|_| true);
let held = 9_742_153;
let script = |die1: &dyn Fn(usize) -> Option<i64>| -> Vec<Vec<(&'static str, i64)>> {
(0..=32)
.map(|i| {
let mut s = vec![(DIES.0, v[i])];
if let Some(x) = die1(i) {
s.push((DIES.1, x));
}
s
})
.collect()
};
for (what, samples) in [
("in the first sample alone", script(&|i| (i == 0).then_some(held))),
("in the first five samples alone", script(&|i| (i < 5).then(|| v[i]))),
("from the third sample on, holding", script(&|i| (i >= 2).then_some(held))),
] {
let (w, live) = watch(QUIET, &samples);
assert_eq!(first_live(&live), None, "DIE_1 {what}: {w:?}");
assert_eq!(w.verdict(), Some(CpuRail::Frozen), "DIE_1 {what}");
}
let (w, _) = watch(QUIET, &script(&|i| (i == 0).then_some(held)));
assert_eq!(
w.rails,
[((DIES.0.to_string(), 0), 32), ((DIES.1.to_string(), 0), 0)],
"the first sample's rails are rails"
);
let (w, live) = watch(QUIET, &script(&|i| (i >= 4).then(|| v[i])));
assert_eq!(first_live(&live), Some(20), "DIE_1 from the fifth sample on, advancing: {w:?}");
assert_eq!(w.verdict(), Some(CpuRail::Live));
}
/// (a), (b) and (c) through the whole probe loop, with a GPU rail
/// moving at every read. Both dies advancing at every read: Live at
/// the tenth interval, where the probe stops. DIE_0 advancing and
/// DIE_1 held: Frozen, after the whole budget. DIE_0 alone, advancing:
/// Live at the tenth.
#[test]
fn the_probe_needs_every_die_to_keep_advancing() {
let r = run(QUIET, |n| {
let n = i64::from(n);
Some(vec![(DIES.0, 100 + n), (DIES.1, 900 + 2 * n), ("GPU Energy", n)])
});
assert_eq!(r.verdict, Some(CpuRail::Live), "both dies: {r:?}");
assert_eq!((r.intervals, r.samples), (10, 11), "both dies: {r:?}");
let r = run(QUIET, |n| {
let n = i64::from(n);
Some(vec![(DIES.0, 100 + n), (DIES.1, 9_742_153), ("GPU Energy", n)])
});
assert_eq!(r.verdict, Some(CpuRail::Frozen), "DIE_1 held: {r:?}");
assert!(r.took >= BUDGET, "DIE_1 held: stopped at {:?}, before the budget", r.took);
let r = run(QUIET, |n| Some(vec![(DIES.0, 100 + i64::from(n)), ("GPU Energy", i64::from(n))]));
assert_eq!(r.verdict, Some(CpuRail::Live), "DIE_0 alone: {r:?}");
assert_eq!((r.intervals, r.samples), (10, 11), "DIE_0 alone: {r:?}");
}
/// The stopping rule on its own: within the budget until a rail proves
/// live; past the budget only until a live rail could have advanced in
/// five intervals; and never past five budgets.
#[test]
fn the_probe_samples_until_live_or_long_enough_to_tell() {
let budget = Duration::from_millis(1000);
let mut w = RailWatch::default();
assert!(w.wants_more(Duration::ZERO, budget));
assert!(w.wants_more(budget / 2, budget), "within budget: keep watching");
// Slow samples: past the budget with fewer than five intervals, keep going.
assert!(w.wants_more(budget * 2, budget), "no interval yet, past budget: keep going");
for i in 0..5 {
w.observe(&sample(&[("CPU Energy", 1)]), ms(400) * i);
assert!(w.wants_more(budget * 2, budget), "{} interval(s), past budget: keep going", w.intervals());
}
assert_eq!(w.intervals(), 4);
assert!(w.wants_more(budget * 5 - ms(1), budget), "four intervals, just under the cap");
assert!(!w.wants_more(budget * 5, budget), "hard cap: five budgets");
w.observe(&sample(&[("CPU Energy", 1)]), ms(2000));
assert!(w.wants_more(budget / 2, budget), "five intervals, within budget: keep watching");
assert!(!w.wants_more(budget, budget), "five intervals, budget spent: frozen");
assert!(!RailWatch::default().wants_more(budget * 5, budget), "hard cap with no interval");
assert!(!RailWatch::default().wants_more(budget * 50, budget), "hard cap with no interval");
// Advances short of the rule are not a reason to stop early: two,
// or five in 155 ms.
let (two, _) = watch(QUIET, &[vec![("CPU Energy", 1)], vec![("CPU Energy", 2)], vec![("CPU Energy", 3)]]);
assert!(two.wants_more(Duration::ZERO, budget), "two advances: keep watching");
let script: Vec<_> = (1..=6).map(|v| vec![("CPU Energy", v)]).collect();
let (five, _) = watch(QUIET, &script);
assert!(five.wants_more(ms(155), budget), "five advances in 155 ms: keep watching");
let script: Vec<_> = (1..=11).map(|v| vec![("CPU Energy", v)]).collect();
let (live, _) = watch(QUIET, &script);
assert!(!live.wants_more(ms(310), budget), "live: stop at once");
}
/// The record keeps, for every read, the wall and CPU clocks around it
/// and what its sample listed; for every successful read, what the
/// rule was handed. A failed read records its clocks and no sample.
/// Each re-read is preceded by exactly one tick. The rule is handed
/// each successful read's `end`: the window runs from the first read's
/// end (8 ms), so the rail, advancing at every successful read, is
/// Live at read 10 (330 ms of window), not read 9 (297 ms).
#[test]
fn the_probe_records_every_read_and_what_the_rule_was_handed() {
let clock = Cell::new(Duration::ZERO);
let cpu = Cell::new(Duration::ZERO);
let n = Cell::new(0u32);
let run = probe_with(
|| Some(()),
|_: &()| {
let k = n.get();
n.set(k + 1);
clock.set(clock.get() + ms(8));
cpu.set(cpu.get() + ms(7));
// The third read fails; the rail advances at every other read.
(k != 2).then(|| read_of(&[("CPU Energy", 5 + i64::from(k)), ("GPU Energy", i64::from(k))]))
},
|| {
clock.set(clock.get() + TICK);
cpu.set(cpu.get() + TICK);
crate::meter::probe_clock::TickOutcome { cpu: Some(TICK), spins: 1 }
},
|| clock.get(),
|| Some(cpu.get()),
BUDGET,
);
assert_eq!(run.verdict, Some(CpuRail::Live));
assert_eq!((run.intervals, run.ticks, run.calls.len()), (9, 10, 11));
let listed = Some(Listed { channels: 2, group_only: true });
assert_eq!(
&run.calls[..4],
&[
SampleCall { start: ms(0), end: ms(8), cpu_before: Some(ms(0)), cpu_after: Some(ms(7)), listed },
SampleCall { start: ms(33), end: ms(41), cpu_before: Some(ms(32)), cpu_after: Some(ms(39)), listed },
SampleCall {
start: ms(66),
end: ms(74),
cpu_before: Some(ms(64)),
cpu_after: Some(ms(71)),
listed: None,
},
SampleCall { start: ms(99), end: ms(107), cpu_before: Some(ms(96)), cpu_after: Some(ms(103)), listed },
]
);
for (k, call) in run.calls.iter().enumerate() {
let k = k as u64;
assert_eq!(
*call,
SampleCall {
start: ms(33 * k),
end: ms(33 * k + 8),
cpu_before: Some(ms(32 * k)),
cpu_after: Some(ms(32 * k + 7)),
listed: (k != 2).then_some(Listed { channels: 2, group_only: true }),
},
"read {k}"
);
}
let key = |name: &str| (name.to_string(), 0u32);
let seen: Vec<Seen> = [0i64, 1, 3, 4, 5, 6, 7, 8, 9, 10]
.iter()
.map(|k| Seen { cpu: vec![(key("CPU Energy"), 5 + k)], gpu: vec![(key("GPU Energy"), *k)] })
.collect();
assert_eq!(run.seen, seen);
// Between successful reads, outside the reads: 32 - 7, then 96 - 39
// (the failed read's 7 ms and a second tick lie in between), then
// 25 ms each.
let mut cpu = vec![Some(ms(25)), Some(ms(57))];
cpu.extend([Some(ms(25)); 7]);
assert_eq!(run.interval_cpu(), cpu);
assert_eq!(run.median_read_cpu(), Some(ms(7)));
assert_eq!(run.median_read_wall(), Some(ms(8)));
assert_eq!(ProbeRun::default().median_read_cpu(), None);
assert!(ProbeRun::default().interval_cpu().is_empty());
}
/// The power reading sums each aggregate once and none of its summands,
/// on the real Energy Model channel names of both chips (macOS 27.0.1).
/// Every channel whose name does not say CPU, GPU or ANE is listed for
/// both chips — so a rule that admits one more CPU summand, or drops a
/// domain, fails here — beside samples of the rails that do.
#[test]
fn counts_once_sums_each_aggregate_once_on_both_chips() {
// M5 Max: summed. `CPU Energy`, `GPU Energy`, and every domain that
// is no aggregate's summand.
let m5_summed = [
"CPU Energy",
"GPU Energy",
"AFR0",
"AMCC0",
"AVE0",
"DCS0",
"DISP0",
"DISPEXT0",
"DRAM0",
"FAB0",
"ISP0",
"MSR0",
"PCIe Port 0 Energy",
"PCIe Port 1 Energy",
"apciec0 Energy",
"apciec1 Energy",
"apciec2 Energy",
];
// M5 Max: summands of `CPU Energy`. `CPU Energy` = `MCPU0` + `MCPU1`
// + `PCPU`; `MCPUn` = its cores + their SRAM + `MCPMn` +
// `MCPMn_SRAM`; `PCPU` = `PACC_0..5` + `PCPU0..5_SRAM` + `PCPM` +
// `PCPM_SRAM`. The first twelve are every summand whose name does
// not say CPU.
let m5_summands = [
"PACC_0",
"PACC_1",
"PACC_2",
"PACC_3",
"PACC_4",
"PACC_5",
"MCPM0",
"MCPM1",
"PCPM",
"MCPM0_SRAM",
"MCPM1_SRAM",
"PCPM_SRAM",
"MCPU0",
"MCPU1",
"PCPU",
"MCPU0_3",
"MCPU1_5_SRAM",
"PCPU2_SRAM",
"MCPU0DTL0e",
"PCPUDTL513",
"GPU0",
"ANE0",
];
// M3 Ultra: summed.
let m3_summed = [
"DIE_0_CPU Energy",
"DIE_1_CPU Energy",
"GPU Energy",
"AMCC0_0",
"AMCC0_1",
"AVE0_0",
"AVE0_1",
"DCS0_0",
"DCS0_1",
"DISP0_0",
"DISP0_1",
"DISPEXT0_0",
"DISPEXT0_1",
"DRAM0_0",
"DRAM0_1",
"ISP0_0",
"ISP0_1",
"MSR0_0",
"MSR0_1",
"PCIe Port 0 Energy",
"PCIe Port 1 Energy",
"PCIe Port 2 Energy",
"PCIe Port 3 Energy",
"apciec0 Energy",
"apciec1 Energy",
"apciec2 Energy",
"apciec3 Energy",
"apciec4 Energy",
"apciec5 Energy",
];
// M3 Ultra: summands. `DIE_n_CPU Energy` = its `EACC_CPU` +
// `PACC0_CPU` + `PACC1_CPU`, each = its cores + their SRAM + its
// `_CPM` + `_CPM_SRAM`. The first twelve are every summand whose name
// does not say CPU.
let m3_summands = [
"DIE_0_EACC_CPM",
"DIE_0_EACC_CPM_SRAM",
"DIE_0_PACC0_CPM",
"DIE_0_PACC0_CPM_SRAM",
"DIE_0_PACC1_CPM",
"DIE_0_PACC1_CPM_SRAM",
"DIE_1_EACC_CPM",
"DIE_1_EACC_CPM_SRAM",
"DIE_1_PACC0_CPM",
"DIE_1_PACC0_CPM_SRAM",
"DIE_1_PACC1_CPM",
"DIE_1_PACC1_CPM_SRAM",
"DIE_0_EACC_CPU",
"DIE_1_PACC0_CPU",
"DIE_0_PACC1_CPU",
"DIE_0_PACC1_CPU3",
"DIE_1_EACC_CPU2_SRAM",
"ECPUDTL00",
"GPU0_0",
"GPU CS0_0",
"GPU SRAM0_0",
"GPU CS SRAM0_0",
"ANE0_0",
"ANE0_1",
];
for name in m5_summed.iter().chain(&m3_summed) {
assert!(counts_once(name), "{name} is not summed");
}
for name in m5_summands {
assert!(!counts_once(name), "{name}, a summand of an aggregate on an M5 Max, is summed beside it");
}
for name in m3_summands {
assert!(!counts_once(name), "{name}, a summand of an aggregate on an M3 Ultra, is summed beside it");
}
// The same rule for the aggregate the chips do not list, and for a
// per-cluster accumulator of a kind neither lists.
assert!(counts_once("ANE Energy"));
assert!(!counts_once("MACC_0"));
}
}
}
// ---------------------------------------------------------------------------
// The probes' clocks and their CPU burn, shared by every liveness probe on the
// platforms that have one (IOReport on macOS, RAPL on Linux).
// ---------------------------------------------------------------------------
#[cfg(any(target_os = "macos", target_os = "linux"))]
#[allow(unsafe_code)]
mod probe_clock {
use std::os::raw::{c_int, c_long};
use std::time::{Duration, Instant};
unsafe extern "C" {
fn clock_gettime(clock_id: c_int, tp: *mut Timespec) -> c_int;
}
/// `struct timespec`: `time_t` (a `long` on both platforms), `long`.
#[repr(C)]
struct Timespec {
tv_sec: c_long,
tv_nsec: c_long,
}
/// `CLOCK_THREAD_CPUTIME_ID`, the calling thread's CPU time: 16 in macOS's
/// `<_time.h>`, 3 in Linux's `<linux/time.h>`.
#[cfg(target_os = "macos")]
const CLOCK_THREAD_CPUTIME_ID: c_int = 16;
#[cfg(target_os = "linux")]
const CLOCK_THREAD_CPUTIME_ID: c_int = 3;
/// Thread CPU time a probe burns before each liveness re-sample.
pub(super) const PROBE_TICK: Duration = Duration::from_millis(25);
/// The most wall time one tick may take to burn [`PROBE_TICK`] of CPU
/// time. Only a thread starved of the CPU for that long is cut short.
const TICK_WALL_CAP: Duration = Duration::from_millis(500);
/// Wall time since this process first asked: the probes' wall clock. One
/// process-wide epoch, so a test can place its own reads on the same
/// timeline as a probe's.
pub(super) fn wall_clock() -> Duration {
static EPOCH: std::sync::OnceLock<Instant> = std::sync::OnceLock::new();
EPOCH.get_or_init(Instant::now).elapsed()
}
/// This thread's CPU time, from `clock_gettime(CLOCK_THREAD_CPUTIME_ID)`;
/// `None` if the clock cannot be read.
pub(super) fn thread_cpu_time() -> Option<Duration> {
let mut ts = Timespec { tv_sec: 0, tv_nsec: 0 };
// SAFETY: ts is a valid, writable timespec.
let rc = unsafe { clock_gettime(CLOCK_THREAD_CPUTIME_ID, &mut ts) };
let (secs, nanos) = (u64::try_from(ts.tv_sec).ok()?, u32::try_from(ts.tv_nsec).ok()?);
(rc == 0 && nanos < 1_000_000_000).then(|| Duration::new(secs, nanos))
}
/// One probe tick: burn [`PROBE_TICK`] of this thread's CPU time — CPU
/// time, not wall time, so a loaded host that runs this thread only part
/// of the time still gets the full burn — giving up after
/// [`TICK_WALL_CAP`] of wall time. Without a readable thread CPU clock it
/// spins [`PROBE_TICK`] of wall time. The burn cannot manufacture `Live`
/// — only a counter's own raw integer that keeps advancing can — so a
/// short burn errs only toward declining, the safe side.
pub(super) fn spin_tick() -> TickOutcome {
burn_thread_cpu(PROBE_TICK, TICK_WALL_CAP)
}
/// What one tick did.
///
/// `spins` is why this type exists. A tick stops on either of two
/// conditions — the CPU target or the wall cap — and from the outside the
/// two outcomes that matter are not separable by time: a spin the host
/// starved into the cap and a tick that simply *slept* both show a long wall
/// span against little CPU. Nor does CPU-per-turn separate them, measured:
/// a turn that blocks costs the thread MORE CPU than a turn of arithmetic,
/// because its context switch is charged to the thread. The only honest
/// separator is whether the loop ran at all, and only the loop can report
/// that.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub(super) struct TickOutcome {
/// The thread CPU the tick saw itself burn, where the clock is readable.
pub(super) cpu: Option<Duration>,
/// Turns of the spin loop. Zero means the tick never span.
pub(super) spins: u64,
}
fn burn_thread_cpu(cpu: Duration, wall_cap: Duration) -> TickOutcome {
let wall = Instant::now();
let mut acc: u64 = 0;
let mut spins: u64 = 0;
let Some(start) = thread_cpu_time() else {
while wall.elapsed() < cpu {
spins += 1;
acc = std::hint::black_box(acc.wrapping_add(1));
}
std::hint::black_box(acc);
return TickOutcome { cpu: None, spins };
};
let mut used = Duration::ZERO;
while wall.elapsed() < wall_cap {
spins += 1;
for _ in 0..4096 {
acc = std::hint::black_box(acc.wrapping_add(1));
}
match thread_cpu_time() {
Some(now) => {
used = now.saturating_sub(start);
if used >= cpu {
break;
}
}
None => break,
}
}
std::hint::black_box(acc);
TickOutcome { cpu: Some(used), spins }
}
}
// ---------------------------------------------------------------------------
// A process's read of its own energy counter: when it may be taken, and the
// one byte layout that check depends on. Kept apart from the FFI so both are
// pure, tested code on every platform; the macOS module only supplies the
// thread's own count, the sleep and the read. Why the check exists is in the
// module doc ("A read of this process's own counter…").
// ---------------------------------------------------------------------------
#[cfg_attr(not(target_os = "macos"), allow(dead_code))]
mod own_read {
use std::time::Duration;
/// The sleeps the measuring thread takes, in turn, before a read of its
/// own process, until one is seen to take it off its core. They grow, so
/// a sleep that returned without blocking is followed by one that much
/// more surely blocks; 6.25 ms in all on a read that is never seen to.
pub(super) const SLEEPS: [Duration; 4] =
[Duration::from_micros(50), Duration::from_micros(200), Duration::from_millis(1), Duration::from_millis(5)];
/// Whether the measuring thread was seen to leave its core: its own
/// energy count (`count`) rose across one of `sleeps`, each taken in turn
/// through `sleep`. The kernel moves that count only when it adds the
/// thread's energy to its process's, so a rise across a sleep is the
/// thread having left its core after all it did before the sleep. A
/// count that held across every sleep, or went down, shows no such
/// thing, and one that cannot be read ends the wait at once.
pub(super) fn thread_left_its_core(
sleeps: &[Duration],
mut count: impl FnMut() -> Option<u64>,
mut sleep: impl FnMut(Duration),
) -> bool {
for &pause in sleeps {
let Some(before) = count() else { return false };
sleep(pause);
let Some(after) = count() else { return false };
if after > before {
return true;
}
}
false
}
/// One read at an end of the measurement's window: `counter`, this
/// process's cumulative CPU energy, read only once the measuring thread
/// was seen to leave its core ([`thread_left_its_core`] over [`SLEEPS`]).
/// `None` where it was not — the counter is then not read at all — or
/// the counter's own read gives nothing.
pub(super) fn window_read(
count: impl FnMut() -> Option<u64>,
sleep: impl FnMut(Duration),
counter: impl FnOnce() -> Option<u64>,
) -> Option<u64> {
if thread_left_its_core(&SLEEPS, count, sleep) {
counter()
} else {
None
}
}
/// `PROC_PIDTHREADCOUNTS`, the `proc_pidinfo` flavor that returns one
/// thread's counts, asked by its 64-bit thread id: in the kernel's
/// `<sys/proc_info_private.h>` (macOS 13 and later), not the SDK's
/// public header, so its result is read in one layout only
/// ([`thread_energy_nj`]).
pub(super) const THREAD_COUNTS_FLAVOR: i32 = 34;
/// The bytes a read of a thread's counts gives the kernel room for: the
/// header and twelve performance levels.
pub(super) const THREAD_COUNTS_BYTES: usize = 8 + 12 * 40;
/// A thread's cumulative CPU energy in nanojoules, summed over the CPU
/// performance levels, from the `wrote` bytes the call says it wrote
/// into `counts`: a `uint16_t` count of levels and six bytes of padding,
/// then for each level five `uint64_t` — instructions, cycles, user
/// time, system time, energy (nJ).
///
/// `None` unless the call wrote exactly the 8-byte header and one 40-byte
/// record for each level the header names, at least one. A failed call
/// and a result of any other size are a layout this code does not know:
/// no count, never a number read from the wrong place. An M5 Max and an
/// M3 Ultra (macOS 27.0.1) write 88 bytes, two levels.
pub(super) fn thread_energy_nj(counts: &[u8], wrote: i32) -> Option<u64> {
let levels = usize::from(u16::from_ne_bytes([*counts.first()?, *counts.get(1)?]));
let bytes = 8 + levels * 40;
if levels == 0 || bytes > counts.len() || usize::try_from(wrote).ok() != Some(bytes) {
return None;
}
let energy = (0..levels).map(|level| {
let at = 8 + level * 40 + 32;
let mut word = [0u8; 8];
word.copy_from_slice(&counts[at..at + 8]);
u64::from_ne_bytes(word)
});
Some(energy.fold(0, u64::saturating_add))
}
#[cfg(test)]
mod tests {
use super::*;
use std::cell::{Cell, RefCell};
/// The measuring thread is seen to leave its core only when its own
/// count rose across a sleep; the sleeps are taken in turn until one
/// does. A count that held, went down or could not be read is no
/// such sight, and an unreadable count takes no further sleep.
#[test]
fn the_thread_is_seen_to_leave_its_core_only_by_its_own_count_rising() {
let sleeps = [Duration::from_micros(50), Duration::from_micros(200), Duration::from_millis(1)];
let seen = |counts: &[Option<u64>]| {
let mut counts = counts.iter().copied();
let slept = RefCell::new(Vec::new());
let left = thread_left_its_core(&sleeps, || counts.next().flatten(), |pause| slept.borrow_mut().push(pause));
(left, slept.into_inner())
};
assert_eq!(seen(&[Some(70), Some(71)]), (true, vec![sleeps[0]]), "rose across the first sleep");
assert_eq!(
seen(&[Some(70), Some(70), Some(70), Some(70), Some(70), Some(5_000)]),
(true, sleeps.to_vec()),
"held across two sleeps, rose across the third"
);
assert_eq!(seen(&[Some(70); 6]), (false, sleeps.to_vec()), "held across every sleep");
assert_eq!(seen(&[Some(70), Some(60), Some(60), Some(50), Some(50), Some(40)]), (false, sleeps.to_vec()), "went down");
assert_eq!(seen(&[None]), (false, vec![]), "unreadable before any sleep");
assert_eq!(seen(&[Some(70), None]), (false, vec![sleeps[0]]), "unreadable after a sleep");
assert!(!thread_left_its_core(&[], || Some(1), |_| panic!("slept")), "no sleep, nothing seen");
assert_eq!(SLEEPS.iter().sum::<Duration>(), Duration::from_micros(6_250));
}
/// A window's read is the counter's only once the thread was seen to
/// leave its core. Where its count never rises the read is `None`
/// after every sleep, and the counter is not read at all; where its
/// count cannot be read, `None` at once. A counter that gives
/// nothing after a thread seen to leave is `None` too.
#[test]
fn a_window_read_whose_thread_is_not_seen_to_leave_its_core_is_no_read() {
let run = |counts: &[Option<u64>], counter: Option<u64>| {
let mut counts = counts.iter().copied().cycle();
let slept = RefCell::new(Vec::new());
let read = Cell::new(false);
let r = window_read(
|| counts.next().flatten(),
|pause| slept.borrow_mut().push(pause),
|| {
read.set(true);
counter
},
);
(r, slept.into_inner(), read.get())
};
assert_eq!(run(&[Some(3), Some(8)], Some(123_456)), (Some(123_456), vec![SLEEPS[0]], true));
assert_eq!(run(&[Some(3)], Some(123_456)), (None, SLEEPS.to_vec(), false), "a count that never rose");
assert_eq!(run(&[None], Some(123_456)), (None, vec![], false), "a count that cannot be read");
assert_eq!(run(&[Some(3), Some(8)], None), (None, vec![SLEEPS[0]], true), "the counter gave nothing");
}
/// A thread's counts as the kernel writes them: `levels` in the first
/// two bytes, six of padding, five u64 a level, the fifth its energy.
fn counts(levels: u16, energy_nj: &[u64]) -> Vec<u8> {
let mut bytes = vec![0u8; THREAD_COUNTS_BYTES];
bytes[..2].copy_from_slice(&levels.to_ne_bytes());
for (level, nj) in energy_nj.iter().enumerate() {
for (field, word) in [71u64, 72, 73, 74, *nj].into_iter().enumerate() {
let at = 8 + level * 40 + field * 8;
bytes[at..at + 8].copy_from_slice(&word.to_ne_bytes());
}
}
bytes
}
/// The energy is each level's fifth u64, summed, and a result of any
/// size but the header and one record a level is no count.
#[test]
fn a_threads_energy_is_read_only_from_the_layout_known() {
assert_eq!(THREAD_COUNTS_FLAVOR, 34);
assert_eq!(thread_energy_nj(&counts(2, &[2_500_000, 40]), 88), Some(2_500_040), "two levels, 88 bytes");
assert_eq!(thread_energy_nj(&counts(1, &[9]), 48), Some(9));
assert_eq!(thread_energy_nj(&counts(3, &[1, 20, 300]), 128), Some(321));
assert_eq!(thread_energy_nj(&counts(2, &[u64::MAX, 2]), 88), Some(u64::MAX), "the sum saturates");
let two = counts(2, &[2_500_000, 40]);
for (wrote, why) in [(-1, "the call failed"), (0, "nothing written"), (48, "one record"), (80, "short"), (96, "long")] {
assert_eq!(thread_energy_nj(&two, wrote), None, "{why}");
}
assert_eq!(thread_energy_nj(&counts(0, &[]), 8), None, "no performance level");
assert_eq!(thread_energy_nj(&counts(13, &[]), 8 + 13 * 40), None, "more levels than the buffer holds");
assert_eq!(thread_energy_nj(&[2], 88), None, "no header");
}
}
}
// ---------------------------------------------------------------------------
// macOS — this process's CPU energy counter: `proc_pid_rusage(getpid(),
// RUSAGE_INFO_V6)`, public libproc API, readable by an unprivileged process.
// ---------------------------------------------------------------------------
#[cfg(target_os = "macos")]
#[allow(unsafe_code)]
mod macos_process {
use super::own_read;
use std::os::raw::{c_int, c_void};
use std::sync::atomic::{AtomicBool, Ordering};
/// `struct rusage_info_v6` from `<sys/resource.h>` (macOS 27.0 SDK), field
/// for field: 16 bytes of UUID then 56 `uint64_t`, 464 bytes, with
/// `ri_energy_nj` at byte 336. The kernel writes the whole struct for
/// `RUSAGE_INFO_V6`, so the buffer must be at least this large; a test
/// pins the size and the offset against the header's own numbers.
#[repr(C)]
#[allow(dead_code)]
pub(super) struct RusageInfoV6 {
ri_uuid: [u8; 16],
ri_user_time: u64,
ri_system_time: u64,
ri_pkg_idle_wkups: u64,
ri_interrupt_wkups: u64,
ri_pageins: u64,
ri_wired_size: u64,
ri_resident_size: u64,
ri_phys_footprint: u64,
ri_proc_start_abstime: u64,
ri_proc_exit_abstime: u64,
ri_child_user_time: u64,
ri_child_system_time: u64,
ri_child_pkg_idle_wkups: u64,
ri_child_interrupt_wkups: u64,
ri_child_pageins: u64,
ri_child_elapsed_abstime: u64,
ri_diskio_bytesread: u64,
ri_diskio_byteswritten: u64,
ri_cpu_time_qos_default: u64,
ri_cpu_time_qos_maintenance: u64,
ri_cpu_time_qos_background: u64,
ri_cpu_time_qos_utility: u64,
ri_cpu_time_qos_legacy: u64,
ri_cpu_time_qos_user_initiated: u64,
ri_cpu_time_qos_user_interactive: u64,
ri_billed_system_time: u64,
ri_serviced_system_time: u64,
ri_logical_writes: u64,
ri_lifetime_max_phys_footprint: u64,
ri_instructions: u64,
ri_cycles: u64,
ri_billed_energy: u64,
ri_serviced_energy: u64,
ri_interval_max_phys_footprint: u64,
ri_runnable_time: u64,
ri_flags: u64,
ri_user_ptime: u64,
ri_system_ptime: u64,
ri_pinstructions: u64,
ri_pcycles: u64,
/// The CPU energy the kernel attributes to this process's threads,
/// cumulative, in nanojoules — exited threads included.
pub(super) ri_energy_nj: u64,
ri_penergy_nj: u64,
ri_secure_time_in_system: u64,
ri_secure_ptime_in_system: u64,
ri_neural_footprint: u64,
ri_lifetime_max_neural_footprint: u64,
ri_interval_max_neural_footprint: u64,
ri_conclave_footprint: u64,
ri_page_wait_time_mach: u64,
ri_page_cache_hits: u64,
ri_reserved: [u64; 6],
}
/// `RUSAGE_INFO_V6` (`<sys/resource.h>`).
const RUSAGE_INFO_V6: c_int = 6;
unsafe extern "C" {
/// `<libproc.h>`: `int proc_pid_rusage(int pid, int flavor,
/// rusage_info_t *buffer)`; 0 on success.
fn proc_pid_rusage(pid: c_int, flavor: c_int, buffer: *mut c_void) -> c_int;
/// `<libproc.h>`: `int proc_pidinfo(int pid, int flavor, uint64_t arg,
/// void *buffer, int buffersize)`; the bytes written, or 0 or less.
fn proc_pidinfo(pid: c_int, flavor: c_int, arg: u64, buffer: *mut c_void, buffersize: c_int) -> c_int;
/// `<pthread.h>`: the calling thread.
fn pthread_self() -> *mut c_void;
/// `<pthread.h>`: `int pthread_threadid_np(pthread_t thread,
/// uint64_t *thread_id)`; 0 on success.
fn pthread_threadid_np(thread: *mut c_void, thread_id: *mut u64) -> c_int;
}
/// Set by [`super::simulate_frozen_process_energy`].
pub(super) static FROZEN: AtomicBool = AtomicBool::new(false);
/// This process's cumulative CPU energy in nanojoules: `ri_energy_nj` of
/// `proc_pid_rusage(<this pid>, RUSAGE_INFO_V6)`. `None` if the call
/// fails. Under [`super::simulate_frozen_process_energy`], the value of
/// the first read after it, forever.
pub(super) fn energy_nj() -> Option<u64> {
if FROZEN.load(Ordering::SeqCst) {
static HELD: std::sync::OnceLock<Option<u64>> = std::sync::OnceLock::new();
return *HELD.get_or_init(read_energy_nj);
}
read_energy_nj()
}
fn read_energy_nj() -> Option<u64> {
let pid = c_int::try_from(std::process::id()).ok()?;
let mut info = std::mem::MaybeUninit::<RusageInfoV6>::zeroed();
// SAFETY: the buffer is a zeroed, writable `rusage_info_v6`, the size
// the kernel writes for RUSAGE_INFO_V6; it is read only on success,
// and every field is a plain integer.
let rc = unsafe { proc_pid_rusage(pid, RUSAGE_INFO_V6, info.as_mut_ptr().cast()) };
(rc == 0).then(|| unsafe { info.assume_init() }.ri_energy_nj)
}
/// The calling thread's own cumulative CPU energy in nanojoules: the
/// count a window's read watches to see the thread leave its core.
/// `None` if the thread's id or its counts cannot be read, or the counts
/// come back in a layout [`own_read::thread_energy_nj`] does not know.
pub(super) fn thread_energy_nj() -> Option<u64> {
let pid = c_int::try_from(std::process::id()).ok()?;
let mut thread = 0u64;
// SAFETY: `thread` is a writable u64, and `pthread_self` is the
// calling thread, alive for the call.
if unsafe { pthread_threadid_np(pthread_self(), &mut thread) } != 0 {
return None;
}
let mut counts = [0u8; own_read::THREAD_COUNTS_BYTES];
let size = c_int::try_from(counts.len()).ok()?;
// SAFETY: `counts` is writable for `size` bytes, which the call is
// given; the kernel writes plain integers and no more than that.
let wrote = unsafe { proc_pidinfo(pid, own_read::THREAD_COUNTS_FLAVOR, thread, counts.as_mut_ptr().cast(), size) };
own_read::thread_energy_nj(&counts, wrote)
}
/// The read the measurement takes at each end of its window: this
/// process's counter ([`energy_nj`]), read only once the calling thread
/// was seen to leave its core ([`own_read::window_read`]). `None` where
/// it was not, or the counter's read fails.
pub(super) fn window_read_nj() -> Option<u64> {
own_read::window_read(thread_energy_nj, std::thread::sleep, energy_nj)
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::{Duration, Instant};
/// The header's numbers, as the tests' own literals: `rusage_info_v6`
/// is 16 bytes of UUID and 56 `uint64_t` (464 bytes), and
/// `ri_energy_nj` is the 41st `uint64_t` (byte 16 + 40 × 8 = 336).
#[test]
fn the_rusage_struct_matches_the_header() {
assert_eq!(std::mem::size_of::<RusageInfoV6>(), 464, "sizeof(struct rusage_info_v6)");
assert_eq!(std::mem::offset_of!(RusageInfoV6, ri_energy_nj), 336, "offsetof(ri_energy_nj)");
assert_eq!(std::mem::offset_of!(RusageInfoV6, ri_penergy_nj), 344, "offsetof(ri_penergy_nj)");
}
// ── The tests' own reading of the counter ──────────────────────────
#[allow(clashing_extern_declarations)]
unsafe extern "C" {
#[link_name = "proc_pid_rusage"]
fn oracle_proc_pid_rusage(pid: c_int, flavor: c_int, buffer: *mut [u64; 58]) -> c_int;
#[link_name = "getpid"]
fn oracle_getpid() -> c_int;
}
/// This process's `ri_energy_nj`, read by the tests themselves: their
/// own binding, their own pid (`getpid`), flavor 6, and a raw 464-byte
/// buffer read at byte 336 (the 43rd `uint64_t` of the buffer).
fn oracle_energy_nj() -> u64 {
let mut buf = [0u64; 58];
// SAFETY: buf is 464 writable bytes, the size of rusage_info_v6.
let rc = unsafe { oracle_proc_pid_rusage(oracle_getpid(), 6, &mut buf) };
assert_eq!(rc, 0, "the tests' own proc_pid_rusage(getpid(), 6) failed");
buf[42]
}
fn burn(d: Duration) {
let t = Instant::now();
let mut acc = 0u64;
while t.elapsed() < d {
acc = std::hint::black_box(acc.wrapping_add(1));
}
}
/// The FFI value pin: the production reader reads THIS process's
/// counter, fresh, every time. Each round brackets one production read
/// between two of the tests' own reads; the production value must lie
/// between them (the counter only goes up). Between rounds this thread
/// burns 50 ms, so the counter must advance, and the production reads
/// with it. A reader of another pid, a constant, or a stale first read
/// fails here.
#[test]
fn the_reader_reads_this_processs_counter_as_the_oracle_does() {
let mut prod = Vec::new();
for round in 1..=3 {
let lo = oracle_energy_nj();
let v = energy_nj().expect("the production reader reads this process's counter");
let hi = oracle_energy_nj();
eprintln!("process ffi pin: round={round} oracle_before={lo} production={v} oracle_after={hi}");
assert!(lo <= v && v <= hi, "round {round}: the production read {v} is outside the oracle's [{lo}, {hi}]");
prod.push(v);
burn(Duration::from_millis(50));
}
assert!(prod.windows(2).all(|w| w[1] > w[0]), "the production reads did not advance across 50 ms burns: {prod:?}");
}
/// On this host the calling thread's own count reads, in the layout
/// known, and a window's read sees the thread leave its core: the
/// path both of the measurement's reads take. The read is this
/// process's counter, between two of the tests' own reads of it.
#[test]
fn a_window_read_sees_the_measuring_thread_leave_its_core_here() {
let thread = thread_energy_nj();
eprintln!("process window read: this thread's own energy count: {thread:?} nJ");
assert!(thread.is_some(), "this thread's own energy count does not read here");
let lo = oracle_energy_nj();
let v = window_read_nj().expect("the measuring thread was not seen to leave its core");
let hi = oracle_energy_nj();
assert!(lo <= v && v <= hi, "the window's read {v} is outside the oracle's [{lo}, {hi}]");
}
}
}
// ---------------------------------------------------------------------------
// Linux RAPL: the energy between two reads of the powercap zones, and what
// the liveness rule is handed from one read, kept apart from the sysfs reads
// so they are pure, tested functions on every platform.
// ---------------------------------------------------------------------------
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
mod rapl {
use super::liveness::{Listed, Read, Sample, Watched};
/// One powercap zone (`/sys/class/powercap/intel-rapl:<index>`), read once.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct Zone {
pub(super) index: u32,
/// Whether the zone is a package (its `name` is `package-<n>`), not a
/// platform (`psys`) or other top-level zone. Only package zones are
/// summed, and only their liveness is probed: `psys` covers the
/// package too, so summing it beside one counts that energy twice.
pub(super) package: bool,
/// The cumulative `energy_uj` counter.
pub(super) energy_uj: u64,
/// `max_energy_range_uj`, the value this zone's counter wraps at, if
/// it could be read.
pub(super) max_energy_range_uj: Option<u64>,
}
/// Whether a top-level zone's `name` file names a package: `package-<n>`,
/// `<n>` one or more digits.
pub(super) fn is_package_name(name: &str) -> bool {
name.trim().strip_prefix("package-").is_some_and(|n| !n.is_empty() && n.bytes().all(|b| b.is_ascii_digit()))
}
/// The package zones of one read, in zone order: the zones whose energy a
/// RAPL figure sums.
pub(super) fn packages(zones: &[Zone]) -> Vec<Zone> {
zones.iter().filter(|z| z.package).copied().collect()
}
/// The name a package zone's counter has in the sample the liveness rule
/// is handed: its sysfs zone, `intel-rapl:<index>`.
pub(super) fn rail_name(index: u32) -> String {
format!("intel-rapl:{index}")
}
/// Whether a sample channel is a RAPL package counter: the only channels
/// [`read_of`] puts in a sample.
pub(super) fn is_package_rail(name: &str) -> bool {
name.strip_prefix("intel-rapl:").is_some_and(|n| !n.is_empty() && n.bytes().all(|b| b.is_ascii_digit()))
}
/// The liveness rule over RAPL: live only when EVERY package counter
/// keeps advancing (the figure sums all of them), by the same
/// sustained-advance rule as IOReport's CPU rails.
pub(super) const PACKAGES: Watched = Watched { pick: is_package_rail };
/// What one read of the zones hands the liveness rule: each PACKAGE
/// zone's raw `energy_uj`, under its sysfs name, in zone order. Other
/// zones are left out — they are neither summed nor probed — so a read
/// with zones but no package hands the rule an empty sample (no rail:
/// `Absent`). `listed` counts the counters handed over.
pub(super) fn read_of(zones: &[Zone]) -> Read {
let sample: Sample = packages(zones)
.iter()
.map(|z| (rail_name(z.index), i64::try_from(z.energy_uj).unwrap_or(i64::MAX)))
.collect();
let listed = Listed { channels: sample.len(), group_only: true };
Read { sample, listed }
}
/// One zone's energy between two reads, modulo its own wrap. A counter that
/// went down wrapped at `max_range`; without a readable (non-zero) range, or
/// with a first read above it, the delta is unknown and `None`. Undercounts
/// a wrap by at most one counter unit (the counter's last value before 0 is
/// `max_range` less one unit), and cannot see more than one wrap in a window
/// — minutes at package power, against this meter's ~1 s windows.
pub(super) fn zone_delta_uj(e0: u64, e1: u64, max_range: Option<u64>) -> Option<u64> {
if e1 >= e0 {
return Some(e1 - e0);
}
let max = max_range.filter(|&m| m > 0)?;
max.checked_sub(e0)?.checked_add(e1)
}
/// The energy between two reads of the same zones: the sum of per-zone
/// deltas, each taken modulo that zone's own `max_energy_range_uj`. Each
/// zone wraps at its own range, so wrapping the SUM of the counters turns
/// one zone's wrap into ~1.8e19 uJ. `None` — no figure, never an invented
/// one — if the zone sets differ or a zone's delta is unknown.
pub(super) fn delta_uj(before: &[Zone], after: &[Zone]) -> Option<u64> {
if before.len() != after.len() {
return None;
}
let mut total = 0u64;
for (b, a) in before.iter().zip(after) {
if b.index != a.index {
return None;
}
let range = a.max_energy_range_uj.or(b.max_energy_range_uj);
total = total.checked_add(zone_delta_uj(b.energy_uj, a.energy_uj, range)?)?;
}
Some(total)
}
#[cfg(test)]
mod tests {
use super::*;
/// A package zone's range as the kernel reports it (2^32 units of
/// 61.035 uJ).
const RANGE: u64 = 262_143_328_850;
fn zone(index: u32, energy_uj: u64, max: Option<u64>) -> Zone {
Zone { index, package: true, energy_uj, max_energy_range_uj: max }
}
#[test]
fn no_wrap_is_the_plain_sum_of_zone_deltas() {
let before = [zone(0, 1_000, Some(RANGE)), zone(1, 5_000, Some(65_532_610_987))];
let after = [zone(0, 3_000, Some(RANGE)), zone(1, 9_000, Some(65_532_610_987))];
assert_eq!(delta_uj(&before, &after), Some(2_000 + 4_000));
assert_eq!(delta_uj(&before, &before), Some(0));
}
/// One zone wraps at its own range while the other does not: its delta
/// is taken modulo ITS range. Wrapping the sum instead gives ~1.8e19.
#[test]
fn a_wrap_in_one_zone_is_taken_modulo_that_zones_range() {
let before = [zone(0, RANGE - 1_000, Some(RANGE)), zone(1, 5_000, Some(65_532_610_987))];
let after = [zone(0, 500, Some(RANGE)), zone(1, 9_000, Some(65_532_610_987))];
assert_eq!(delta_uj(&before, &after), Some(1_500 + 4_000));
let summed = (500u64 + 9_000).wrapping_sub(RANGE - 1_000 + 5_000);
assert!(summed > 1e18 as u64, "the summed-wrap figure this replaces: {summed}");
}
/// A zone whose range cannot be read: fine while it does not wrap; once
/// it wraps its delta is unknown, and so is the total.
#[test]
fn a_zone_with_an_unreadable_range_gives_no_figure_only_when_it_wraps() {
let before = [zone(0, 1_000, None), zone(1, 5_000, Some(RANGE))];
let after = [zone(0, 4_000, None), zone(1, 6_000, Some(RANGE))];
assert_eq!(delta_uj(&before, &after), Some(3_000 + 1_000));
let wrapped = [zone(0, 10, None), zone(1, 6_000, Some(RANGE))];
assert_eq!(zone_delta_uj(1_000, 10, None), None, "a wrap with an unreadable range");
assert_eq!(delta_uj(&before, &wrapped), None);
assert_eq!(zone_delta_uj(1_000, 10, Some(0)), None, "a zero range is unreadable");
}
/// A package zone is named `package-<n>` (the sysfs file ends in a
/// newline); `psys`, `dram`, `core` and malformed names are not.
#[test]
fn only_a_package_name_is_a_package() {
for name in ["package-0", "package-1\n", "package-12"] {
assert!(is_package_name(name), "{name:?}");
}
for name in ["psys", "dram", "core", "uncore", "package-", "package-x", "package-0a", "Package-0", ""] {
assert!(!is_package_name(name), "{name:?}");
}
assert!(is_package_rail("intel-rapl:0") && is_package_rail("intel-rapl:12"));
assert!(!is_package_rail("intel-rapl:") && !is_package_rail("intel-rapl:0:1") && !is_package_rail("CPU Energy"));
}
/// A read hands the rule each PACKAGE zone's own raw counter, under
/// its sysfs zone, in zone order; a `psys` zone is left out, so a read
/// with no package hands over an empty sample.
#[test]
fn a_read_hands_the_rule_each_package_counter_as_read() {
let psys = Zone { index: 1, package: false, energy_uj: 9_999_999, max_energy_range_uj: Some(RANGE) };
let zones = [zone(0, 1_000, Some(RANGE)), psys, zone(2, 7_000, Some(RANGE))];
let r = read_of(&zones);
assert_eq!(r.sample, vec![("intel-rapl:0".to_string(), 1_000), ("intel-rapl:2".to_string(), 7_000)]);
assert_eq!(r.listed, Listed { channels: 2, group_only: true });
assert_eq!(packages(&zones), vec![zones[0], zones[2]], "only package zones are summed");
let none = read_of(&[psys]);
assert!(none.sample.is_empty() && none.listed.channels == 0, "{none:?}");
}
#[test]
fn mismatched_or_inconsistent_reads_give_no_figure() {
let one = [zone(0, 1_000, Some(RANGE))];
let two = [zone(0, 2_000, Some(RANGE)), zone(1, 5_000, Some(RANGE))];
assert_eq!(delta_uj(&one, &two), None, "a zone appeared");
assert_eq!(delta_uj(&one, &[zone(1, 2_000, Some(RANGE))]), None, "a different zone");
assert_eq!(zone_delta_uj(RANGE + 5, 3, Some(RANGE)), None, "first read above the range");
}
}
}
// ---------------------------------------------------------------------------
// Platform power readings (watts), averaged over a short window.
// ---------------------------------------------------------------------------
#[cfg(target_os = "macos")]
fn idle_power() -> Option<f64> {
macos_ioreport::power_over(std::time::Duration::from_millis(1600))
}
#[cfg(target_os = "macos")]
fn busy_power() -> Option<f64> {
macos_ioreport::power_over(std::time::Duration::from_millis(2800))
}
// Apple Silicon power via the IOReport Energy Model C API — energy delta over a
// timed window ÷ window = watts, the same shape as the Linux RAPL path. No sudo,
// no external binary. Ported from jouleclaw-energy's validated `apple_ioreport`.
// Used only while the CPU energy rail is live: see `host_status`.
#[cfg(target_os = "macos")]
#[allow(non_camel_case_types, unsafe_code)]
mod macos_ioreport {
use super::liveness::{counts_once, read_from, Channel, Read};
use std::os::raw::{c_int, c_long, c_void};
use std::time::Duration;
type CFTypeRef = *const c_void;
type CFStringRef = *const c_void;
type CFArrayRef = *const c_void;
type CFDictionaryRef = *const c_void;
type CFMutableDictionaryRef = *mut c_void;
type IOReportSubscriptionRef = *const c_void;
const K_CFSTRING_ENCODING_UTF8: u32 = 0x0800_0100;
unsafe extern "C" {
fn IOReportCopyAllChannels(a: u64, b: u64) -> CFMutableDictionaryRef;
fn IOReportCopyChannelsInGroup(
group: CFStringRef,
subgroup: CFStringRef,
a: u64,
b: u64,
c: u64,
) -> CFMutableDictionaryRef;
fn IOReportCreateSubscription(
allowed: *const c_void,
desired_channels: CFMutableDictionaryRef,
subscribed_channels: *mut CFMutableDictionaryRef,
channel_id: u64,
options: CFTypeRef,
) -> IOReportSubscriptionRef;
fn IOReportCreateSamples(
iorsub: IOReportSubscriptionRef,
chan_dict: CFMutableDictionaryRef,
options: CFTypeRef,
) -> CFDictionaryRef;
fn IOReportCreateSamplesDelta(
previous: CFDictionaryRef,
current: CFDictionaryRef,
options: CFTypeRef,
) -> CFDictionaryRef;
fn IOReportChannelGetGroup(item: CFTypeRef) -> CFStringRef;
fn IOReportChannelGetChannelName(item: CFTypeRef) -> CFStringRef;
fn IOReportSimpleGetIntegerValue(item: CFTypeRef, _v: c_int) -> c_long;
fn IOReportChannelGetUnitLabel(item: CFTypeRef) -> CFStringRef;
fn CFRelease(cf: CFTypeRef);
fn CFArrayGetCount(the_array: CFArrayRef) -> c_long;
fn CFArrayGetValueAtIndex(the_array: CFArrayRef, idx: c_long) -> CFTypeRef;
fn CFDictionaryGetValue(the_dict: CFDictionaryRef, key: *const c_void) -> *const c_void;
fn CFStringCreateWithCString(
allocator: *mut c_void,
cstr: *const i8,
encoding: u32,
) -> CFStringRef;
fn CFStringGetCString(
the_string: CFStringRef,
buffer: *mut i8,
buffer_size: c_long,
encoding: u32,
) -> bool;
}
#[link(name = "IOReport", kind = "dylib")]
#[link(name = "IOKit", kind = "framework")]
#[link(name = "CoreFoundation", kind = "framework")]
unsafe extern "C" {}
/// Open an IOReport subscription over all channels. `None` if unavailable.
/// Returns `(sub, subscribed_channels)`; caller must `CFRelease` both.
unsafe fn open() -> Option<(IOReportSubscriptionRef, CFMutableDictionaryRef)> {
let all = IOReportCopyAllChannels(0, 0);
if all.is_null() {
return None;
}
let mut subscribed: CFMutableDictionaryRef = std::ptr::null_mut();
let sub = IOReportCreateSubscription(
std::ptr::null(),
all,
&mut subscribed as *mut _,
0,
std::ptr::null(),
);
CFRelease(all as CFTypeRef);
if sub.is_null() || subscribed.is_null() {
if !sub.is_null() {
CFRelease(sub);
}
if !subscribed.is_null() {
CFRelease(subscribed as CFTypeRef);
}
return None;
}
Some((sub, subscribed))
}
/// The liveness probe's own subscription: the Energy Model group alone
/// (`IOReportCopyChannelsInGroup`), not every channel. Its CPU rails are the
/// same counters, with the same absolute values, an all-channel subscription
/// reads, but a sample of the group (364 channels on an M5 Max, 645 on an M3
/// Ultra) costs ~6–9 ms of thread CPU, where one of every channel (~11,900
/// and ~14,300) costs ~85–125 ms — so the probe really re-samples after
/// every ~25 ms of CPU burn instead of spending its budget in IOReport.
/// Caller must `CFRelease` both (the probe holds them in an
/// `EnergyModelSub`, which does).
unsafe fn open_energy_model() -> Option<(IOReportSubscriptionRef, CFMutableDictionaryRef)> {
let group = CFStringCreateWithCString(
std::ptr::null_mut(),
c"Energy Model".as_ptr() as *const _,
K_CFSTRING_ENCODING_UTF8,
);
if group.is_null() {
return None;
}
let desired = IOReportCopyChannelsInGroup(group, std::ptr::null(), 0, 0, 0);
CFRelease(group);
if desired.is_null() {
return None;
}
let mut subscribed: CFMutableDictionaryRef = std::ptr::null_mut();
let sub = IOReportCreateSubscription(
std::ptr::null(),
desired,
&mut subscribed as *mut _,
0,
std::ptr::null(),
);
CFRelease(desired as CFTypeRef);
if sub.is_null() || subscribed.is_null() {
if !sub.is_null() {
CFRelease(sub);
}
if !subscribed.is_null() {
CFRelease(subscribed as CFTypeRef);
}
return None;
}
Some((sub, subscribed))
}
/// The probe's own Energy Model subscription, released on drop: the real
/// opener and reader `host_status` hands the probe loop.
pub(super) struct EnergyModelSub {
sub: IOReportSubscriptionRef,
ch: CFMutableDictionaryRef,
}
impl EnergyModelSub {
/// Subscribe to the Energy Model group alone ([`open_energy_model`]).
/// `None` when there is no such group to subscribe to — or when a test
/// has made this process ([`super::simulate_no_energy_model_group`]) or,
/// in this crate's own tests, this thread (`seam`) behave as if there
/// were none.
pub(super) fn open() -> Option<Self> {
if super::NO_ENERGY_MODEL_GROUP.load(std::sync::atomic::Ordering::SeqCst) {
return None;
}
#[cfg(test)]
if let Some(opens) = seam::opens() {
return opens.then_some(Self { sub: std::ptr::null(), ch: std::ptr::null_mut() });
}
// SAFETY: open_energy_model returns owned, non-null refs or None.
unsafe { open_energy_model() }.map(|(sub, ch)| Self { sub, ch })
}
/// One read: the channels one IOReport sample of this subscription
/// listed ([`energy_model_channels`]), turned by the pure [`read_from`]
/// into the sample the rule is handed and what was listed. `None` if
/// IOReport returned no sample.
pub(super) fn read(&self) -> Option<Read> {
self.channels().map(read_from)
}
/// The raw channels of one sample; in this crate's own tests, a
/// scripted list instead when this thread has one (`seam`).
fn channels(&self) -> Option<Vec<Channel>> {
#[cfg(test)]
if let Some(scripted) = seam::scripted() {
return scripted;
}
if self.sub.is_null() {
return None;
}
// SAFETY: sub and ch are live, non-null refs until drop.
unsafe { energy_model_channels(self.sub, self.ch) }
}
}
impl Drop for EnergyModelSub {
fn drop(&mut self) {
// SAFETY: both were created by open_energy_model and are released
// once; a test's scripted subscription holds neither.
unsafe {
if !self.sub.is_null() {
CFRelease(self.sub);
}
if !self.ch.is_null() {
CFRelease(self.ch as CFTypeRef);
}
}
}
}
/// Every channel one sample of `sub` listed, as (group, name, ABSOLUTE raw
/// integer), in channel order — raw, before unit conversion, so a small
/// advance in a fine unit cannot truncate to zero. `None` if IOReport
/// returns no sample. The FFI only reads; which channels count, and how,
/// is pure code ([`read_from`] and the rule). No decision is made here.
unsafe fn energy_model_channels(sub: IOReportSubscriptionRef, ch: CFMutableDictionaryRef) -> Option<Vec<Channel>> {
let smp = IOReportCreateSamples(sub, ch, std::ptr::null());
if smp.is_null() {
return None;
}
let mut channels = Vec::new();
let key = CFStringCreateWithCString(
std::ptr::null_mut(),
c"IOReportChannels".as_ptr() as *const _,
K_CFSTRING_ENCODING_UTF8,
);
if !key.is_null() {
let arr = CFDictionaryGetValue(smp, key as *const _) as CFArrayRef;
CFRelease(key as CFTypeRef);
if !arr.is_null() {
for i in 0..CFArrayGetCount(arr) {
let item = CFArrayGetValueAtIndex(arr, i);
if item.is_null() {
continue;
}
channels.push((
cfstring_to_string(IOReportChannelGetGroup(item)),
cfstring_to_string(IOReportChannelGetChannelName(item)),
IOReportSimpleGetIntegerValue(item, 0) as i64,
));
}
}
}
CFRelease(smp);
Some(channels)
}
/// Average SoC power (watts) over `d`: two Energy-Model samples bracketing a
/// sleep, energy delta ÷ window. A hardware reading — the bracket built on
/// it is graded `HwShunt` — and so produced only while the CPU energy rail
/// is live: [`super::ioreport_status`] is `Live`, meaning the probe saw that
/// rail advance in at least five sample intervals, in at least 80% of them,
/// over at least 300 ms, while it burned CPU. `None` if IOReport is
/// unavailable, its CPU energy rail is not live (`admits` declines the
/// cached status), or the signal is empty.
pub fn power_over(d: Duration) -> Option<f64> {
// Declined before sampling: on a frozen CPU rail the summed delta below
// would be GPU energy, or zero, attributed to CPU work. IOReport's own
// gate, whatever source ranks above it.
if !super::admits(super::ioreport_status()) {
return None;
}
unsafe {
let (sub, ch) = open()?;
let s0 = IOReportCreateSamples(sub, ch, std::ptr::null());
if s0.is_null() {
CFRelease(sub);
CFRelease(ch as CFTypeRef);
return None;
}
std::thread::sleep(d);
let s1 = IOReportCreateSamples(sub, ch, std::ptr::null());
let uj = if s1.is_null() {
0
} else {
let delta = IOReportCreateSamplesDelta(s0, s1, std::ptr::null());
let u = if delta.is_null() {
0
} else {
let u = sum_energy_uj(delta);
CFRelease(delta);
u
};
CFRelease(s1);
u
};
CFRelease(s0);
CFRelease(sub);
CFRelease(ch as CFTypeRef);
let secs = d.as_secs_f64();
if secs > 0.0 && uj > 0 {
Some(uj as f64 / 1e6 / secs)
} else {
None
}
}
}
/// The energy in one delta sample: every Energy Model rail that
/// [`counts_once`] admits, converted from its unit to microjoules.
unsafe fn sum_energy_uj(delta: CFDictionaryRef) -> u64 {
let key = CFStringCreateWithCString(
std::ptr::null_mut(),
b"IOReportChannels\0".as_ptr() as *const _,
K_CFSTRING_ENCODING_UTF8,
);
if key.is_null() {
return 0;
}
let arr = CFDictionaryGetValue(delta, key as *const _) as CFArrayRef;
CFRelease(key as CFTypeRef);
if arr.is_null() {
return 0;
}
let count = CFArrayGetCount(arr);
let mut total_uj: u64 = 0;
for i in 0..count {
let item = CFArrayGetValueAtIndex(arr, i);
if item.is_null() {
continue;
}
if !cfstring_eq_ic(IOReportChannelGetGroup(item), "energy model") {
continue;
}
let raw = IOReportSimpleGetIntegerValue(item, 0);
if raw <= 0 {
continue;
}
let name = match cfstring_to_string(IOReportChannelGetChannelName(item)) {
Some(s) => s,
None => continue,
};
if !counts_once(&name) {
continue;
}
let unit = cfstring_to_string(IOReportChannelGetUnitLabel(item)).unwrap_or_default();
total_uj = total_uj.saturating_add(energy_to_uj(raw, &unit));
}
total_uj
}
fn energy_to_uj(raw: c_long, unit: &str) -> u64 {
let r = (raw.max(0)) as f64;
let uj = match unit.trim().to_ascii_lowercase().as_str() {
"pj" => r / 1_000_000.0,
"nj" => r / 1_000.0,
"uj" | "µj" => r,
"mj" => r * 1_000.0,
"j" => r * 1_000_000.0,
_ => r / 1_000.0,
};
uj as u64
}
unsafe fn cfstring_to_string(s: CFStringRef) -> Option<String> {
if s.is_null() {
return None;
}
let mut buf = [0i8; 256];
let ok =
CFStringGetCString(s, buf.as_mut_ptr(), buf.len() as c_long, K_CFSTRING_ENCODING_UTF8);
if !ok {
return None;
}
let bytes: &[u8] = std::slice::from_raw_parts(buf.as_ptr() as *const u8, buf.len());
let end = bytes.iter().position(|&b| b == 0).unwrap_or(bytes.len());
let s = std::str::from_utf8(&bytes[..end]).ok()?.trim();
if s.is_empty() { None } else { Some(s.to_string()) }
}
unsafe fn cfstring_eq_ic(s: CFStringRef, target: &str) -> bool {
cfstring_to_string(s).map(|x| x.eq_ignore_ascii_case(target)).unwrap_or(false)
}
/// This crate's own tests' stand-ins for the probe's two FFI calls, per
/// thread, so a test can run the REAL production composition
/// (`host_status`) on a group that cannot be opened, or on scripted channel
/// reads, with the real tick and clocks. Compiled into this crate's own
/// tests only; on every thread the default is the real FFI.
#[cfg(test)]
pub(super) mod seam {
use super::Channel;
use std::cell::RefCell;
/// How the probe's opener and reads behave on this thread.
pub(in super::super) enum Mode {
/// The Energy Model group cannot be opened.
Deny,
/// The group opens (no real subscription), and each read lists the
/// script's next channels; `None` from the script is a failed read.
Script(Box<dyn FnMut() -> Option<Vec<Channel>>>),
}
thread_local! {
static MODE: RefCell<Option<Mode>> = const { RefCell::new(None) };
}
/// `None`: the real opener. `Some(false)`: cannot open. `Some(true)`:
/// a scripted subscription opens.
pub(super) fn opens() -> Option<bool> {
MODE.with(|m| match &*m.borrow() {
None => None,
Some(Mode::Deny) => Some(false),
Some(Mode::Script(_)) => Some(true),
})
}
/// The next scripted read on this thread, or `None` for the real FFI.
pub(super) fn scripted() -> Option<Option<Vec<Channel>>> {
MODE.with(|m| match &mut *m.borrow_mut() {
Some(Mode::Script(next)) => Some(next()),
_ => None,
})
}
/// Run `f` with this thread's opener and reads in `mode`, then return
/// to the real FFI, also when `f` panics.
pub(in super::super) fn with<T>(mode: Mode, f: impl FnOnce() -> T) -> T {
struct Restore;
impl Drop for Restore {
fn drop(&mut self) {
MODE.with(|m| *m.borrow_mut() = None);
}
}
MODE.with(|m| *m.borrow_mut() = Some(mode));
let _restore = Restore;
f()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::meter::liveness::tests::expect_keyed;
use crate::meter::probe_clock::{spin_tick, thread_cpu_time, wall_clock};
use std::time::Instant;
use crate::meter::liveness::{CpuRail, Listed, ProbeRun, RailKey};
use crate::meter::{
admits, bracket_admitted, bracket_marginal_uj, host, host_status, ioreport_status, meter_source, status_of,
HostStatus, IoReportStatus, MeterSource, CPU_RAIL_LIVENESS_BUDGET,
};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
/// The secondary group-only pin: the median thread CPU time of the
/// probe's reads stays under this. A read of the Energy Model group
/// alone costs ~6–9 ms of thread CPU; one of every channel, ~85–125 ms.
/// Thread CPU, not wall time: another process sampling IOReport makes a
/// read wait, not burn. The test's own literal.
const MAX_MEDIAN_READ_CPU: Duration = Duration::from_millis(50);
/// The probe's burn, as the tests see it: at least this much of the
/// probing thread's CPU in every interval, and per tick at least
/// [`TICK_CPU`] unless the tick ran into [`TICK_WALL`]. The tests' own
/// literals, not the implementation's constants.
const TICK_CPU: Duration = Duration::from_millis(25);
const TICK_WALL: Duration = Duration::from_millis(500);
/// Real, cheap, pure CPU work for the metering tests.
fn cpu_work() {
std::hint::black_box((0..4000u64).fold(0u64, |a, b| a.wrapping_mul(31).wrapping_add(b)));
}
// ── The tests' own thread CPU clock ──────────────────────────────────
/// `CLOCK_THREAD_CPUTIME_ID` in macOS's `<time.h>`: 16. Declared here,
/// apart from the production constant, and read through the tests' own
/// binding, so a wrong production clock cannot hide behind the same
/// wrong value in the tests.
const TEST_THREAD_CPU_CLOCK: c_int = 16;
#[repr(C)]
struct TestTimespec {
tv_sec: i64,
tv_nsec: c_long,
}
#[allow(clashing_extern_declarations)]
unsafe extern "C" {
#[link_name = "clock_gettime"]
fn test_clock_gettime(clock_id: c_int, tp: *mut TestTimespec) -> c_int;
}
/// This thread's CPU time, read by the tests themselves.
fn test_thread_cpu() -> Duration {
let mut ts = TestTimespec { tv_sec: 0, tv_nsec: 0 };
// SAFETY: ts is a valid, writable timespec.
let rc = unsafe { test_clock_gettime(TEST_THREAD_CPU_CLOCK, &mut ts) };
assert_eq!(rc, 0, "the tests' own thread CPU clock does not read");
Duration::new(ts.tv_sec as u64, ts.tv_nsec as u32)
}
/// Burn at least `d` of this thread's CPU, by the tests' own clock.
fn burn(d: Duration) -> Duration {
let start = test_thread_cpu();
let mut acc: u64 = 0;
loop {
for _ in 0..4096 {
acc = std::hint::black_box(acc.wrapping_add(1));
}
let spent = test_thread_cpu() - start;
if spent >= d {
return spent;
}
}
}
/// F3, clock identity: the production thread CPU clock is THIS thread's
/// CPU time, checked against the tests' own reading of it. It does not
/// move while the thread sleeps (so it is no wall clock), nor while
/// another thread of this process burns CPU and this one sleeps (so it
/// is not process CPU), and it moves with this thread's own burn.
#[test]
fn the_thread_cpu_clock_is_this_threads_cpu_time() {
let prod = || thread_cpu_time().expect("the production thread CPU clock reads");
let p0 = prod();
std::thread::sleep(Duration::from_millis(100));
let slept = prod() - p0;
let p0 = prod();
let helper = std::thread::spawn(|| burn(Duration::from_millis(100)));
let helper_burned = helper.join().expect("the burning helper panicked");
let while_helper_burned = prod() - p0;
let (t0, p0) = (test_thread_cpu(), prod());
burn(Duration::from_millis(30));
let (p1, t1) = (prod(), test_thread_cpu());
let (own_prod, own_test) = (p1 - p0, t1 - t0);
eprintln!(
"ioreport clock: slept 100ms -> {slept:?}; helper burned {helper_burned:?} -> {while_helper_burned:?}; \
own burn {own_test:?} (tests' clock) -> {own_prod:?} (production clock)"
);
assert!(
slept < Duration::from_millis(5),
"the production thread CPU clock advanced {slept:?} while this thread slept 100 ms: it is not this \
thread's CPU time (a wall clock, or the process's CPU while other threads ran)"
);
assert!(helper_burned >= Duration::from_millis(100), "the helper burned only {helper_burned:?}");
assert!(
while_helper_burned < Duration::from_millis(5),
"the production thread CPU clock advanced {while_helper_burned:?} while another thread burned \
{helper_burned:?} and this one slept: it counts the process's CPU, not this thread's"
);
assert!(
own_prod >= Duration::from_millis(25) && own_prod <= own_test,
"this thread burned {own_test:?} by the tests' clock; the production clock moved {own_prod:?}"
);
}
/// The probe's wall clock is this process's monotonic clock: across a
/// 50 ms sleep it advances at least 50 ms, and no more than `Instant`
/// does around it. A stopped, scaled or skewed clock would change how
/// long the probe watches, and where the freshness pin places the
/// watcher's reads. (No bound on the host: a sleep never returns early,
/// and the nested reading can never exceed the outer one.)
#[test]
fn the_probe_wall_clock_is_the_monotonic_clock() {
let i0 = Instant::now();
let w0 = wall_clock();
std::thread::sleep(Duration::from_millis(50));
let w1 = wall_clock();
let i1 = Instant::now();
let (w, i) = (w1 - w0, i1 - i0);
assert!(
w >= Duration::from_millis(50) && w <= i,
"across a 50 ms sleep the probe's wall clock advanced {w:?}, Instant {i:?} around it"
);
}
/// R1, the tick, on the tests' own clock: each production tick
/// consumes at least 25 ms of this thread's CPU time, unless it ran
/// into its 500 ms wall-clock cap — and never more CPU than wall time.
#[test]
fn each_tick_burns_the_probe_tick_of_thread_cpu_time() {
for i in 0..5 {
let (w0, c0) = (Instant::now(), test_thread_cpu());
spin_tick();
let (c1, w1) = (test_thread_cpu(), Instant::now());
let (cpu, wall) = (c1 - c0, w1 - w0);
eprintln!("ioreport tick: tick {i} burned {cpu:?} of thread CPU time in {wall:?}");
assert!(
cpu >= TICK_CPU || wall >= TICK_WALL,
"tick {i} burned {cpu:?} of thread CPU (tests' clock) in {wall:?}: under {TICK_CPU:?}, and it \
did not run into the {TICK_WALL:?} wall cap"
);
assert!(cpu <= wall, "tick {i}: {cpu:?} of thread CPU in {wall:?} of wall time");
}
}
// ── The oracle: an independent read of every channel ──────────────
/// A channel the oracle read: its group, name and ABSOLUTE raw integer.
#[derive(Debug, Clone)]
struct OracleRail {
group: String,
name: String,
value: i64,
}
/// One read of every channel by the oracle.
#[derive(Debug)]
struct OracleRead {
/// How many channels the sample listed, and how many of them are in
/// the Energy Model group.
group: Group,
/// Every channel whose name ends with `CPU Energy`, any group.
cpu: Vec<OracleRail>,
/// Every Energy Model channel named `GPU Energy`.
gpu: Vec<OracleRail>,
}
/// How many channels the Energy Model group has, and how many there are
/// in all, as the oracle counted them in one all-channel sample.
#[derive(Debug, Clone, Copy)]
struct Group {
energy_model: usize,
all: usize,
}
impl OracleRead {
/// The Energy Model CPU rails, keyed by name and ordinal: the
/// oracle's own keying.
fn cpu_keyed(&self) -> Vec<(RailKey, i64)> {
em_keyed(&self.cpu)
}
fn gpu_keyed(&self) -> Vec<(RailKey, i64)> {
em_keyed(&self.gpu)
}
}
fn em_keyed(rails: &[OracleRail]) -> Vec<(RailKey, i64)> {
let mut keyed: Vec<(RailKey, i64)> = Vec::new();
for r in rails.iter().filter(|r| r.group.eq_ignore_ascii_case("Energy Model")) {
let ordinal = keyed.iter().filter(|((n, _), _)| *n == r.name).count() as u32;
keyed.push(((r.name.clone(), ordinal), r.value));
}
keyed
}
fn keys(values: &[(RailKey, i64)]) -> Vec<RailKey> {
values.iter().map(|(k, _)| k.clone()).collect()
}
/// The oracle's own subscription, over every channel (not the probe's
/// Energy Model one, and not through `open()`), released on drop.
struct OracleSub {
sub: IOReportSubscriptionRef,
ch: CFMutableDictionaryRef,
}
impl OracleSub {
fn open() -> Option<Self> {
unsafe {
let all = IOReportCopyAllChannels(0, 0);
if all.is_null() {
return None;
}
let mut ch: CFMutableDictionaryRef = std::ptr::null_mut();
let sub = IOReportCreateSubscription(std::ptr::null(), all, &mut ch as *mut _, 0, std::ptr::null());
CFRelease(all as CFTypeRef);
if sub.is_null() || ch.is_null() {
if !sub.is_null() {
CFRelease(sub);
}
if !ch.is_null() {
CFRelease(ch as CFTypeRef);
}
return None;
}
Some(Self { sub, ch })
}
}
/// One read of every channel: the channel counts, and the ABSOLUTE
/// cumulative values of the CPU and GPU energy channels, in channel
/// order.
fn read(&self) -> Option<OracleRead> {
unsafe {
let smp = IOReportCreateSamples(self.sub, self.ch, std::ptr::null());
if smp.is_null() {
return None;
}
let key = CFStringCreateWithCString(
std::ptr::null_mut(),
c"IOReportChannels".as_ptr() as *const _,
K_CFSTRING_ENCODING_UTF8,
);
let mut read = OracleRead { group: Group { energy_model: 0, all: 0 }, cpu: Vec::new(), gpu: Vec::new() };
if !key.is_null() {
let arr = CFDictionaryGetValue(smp, key as *const _) as CFArrayRef;
CFRelease(key as CFTypeRef);
if !arr.is_null() {
for i in 0..CFArrayGetCount(arr) {
let it = CFArrayGetValueAtIndex(arr, i);
if it.is_null() {
continue;
}
read.group.all += 1;
let group = cfstring_to_string(IOReportChannelGetGroup(it)).unwrap_or_default();
let name = cfstring_to_string(IOReportChannelGetChannelName(it)).unwrap_or_default();
let em = group.eq_ignore_ascii_case("Energy Model");
read.group.energy_model += usize::from(em);
let rail = |group: String, name: String| OracleRail {
group,
name,
value: IOReportSimpleGetIntegerValue(it, 0) as i64,
};
if name.ends_with("CPU Energy") {
read.cpu.push(rail(group, name));
} else if em && name == "GPU Energy" {
read.gpu.push(rail(group, name));
}
}
}
}
CFRelease(smp);
Some(read)
}
}
}
impl Drop for OracleSub {
fn drop(&mut self) {
unsafe {
CFRelease(self.sub);
CFRelease(self.ch as CFTypeRef);
}
}
}
/// What the independent oracle saw on this host.
#[derive(Debug, PartialEq)]
enum Oracle {
NoIoReport,
NoCpuRail,
CpuRailMoved(bool),
}
/// The oracle's bar: a CPU rail it calls moving changed at no fewer
/// than this many of its reads (and at 80% of them). The tests' own
/// literal.
const ORACLE_MIN_CHANGES: u32 = 5;
/// The most the oracle extends its window, in budgets, while short of
/// [`ORACLE_MIN_CHANGES`] reads. Its reads of every channel slow down
/// under IOReport contention (with an all-channel sampler running
/// beside the tests on a loaded M5 Max it needed 6 s for 5 of them),
/// and a floor of 5 reads needs room for 2.5 times the reads a floor of
/// 2 did under a cap of 3 budgets. Costs nothing while reads are fast:
/// the window is extended only while reads are too few.
const ORACLE_MAX_BUDGETS: u32 = 8;
/// The oracle's count, kept apart from its reads so the tests can feed
/// it scripted ones: how many reads it compared with the read before,
/// and per CPU rail (in channel order) at how many of them the rail's
/// absolute value had changed.
#[derive(Debug)]
struct OracleTally {
reads: u32,
changes: Vec<u32>,
}
impl OracleTally {
fn new(rails: usize) -> Self {
Self { reads: 0, changes: vec![0; rails] }
}
/// One more read, `now`, against the read before it, `before`: the
/// rails' absolute values, in channel order.
fn add(&mut self, before: &[i64], now: &[i64]) {
self.reads += 1;
for (count, (now, before)) in self.changes.iter_mut().zip(now.iter().zip(before)) {
*count += u32::from(now != before);
}
}
/// The oracle's verdict so far: EVERY CPU rail it reads, and at
/// least one, changed at no fewer than [`ORACLE_MIN_CHANGES`]
/// reads, and at at least 80% of them — what a Live status claims.
/// One die that keeps moving does not vouch for one that held.
fn moved(&self) -> bool {
!self.changes.is_empty()
&& self.changes.iter().all(|&c| c >= ORACLE_MIN_CHANGES && c * 5 >= self.reads * 4)
}
}
/// The oracle's bar on scripted reads, so a change to it is caught on
/// any host. A refresh, however it is published — one step, or the two
/// steps 64 ms apart observed on an M5 Max, which the oracle's reads of
/// every channel (0.1–0.8 s apart) see in one read or two — is never
/// "moved", however many reads follow; nor is a rail that changed at
/// four reads of four. A rail changing at every read is, from its fifth
/// read; one changing at 80% of them is, and at 70% is not.
#[test]
fn the_oracle_never_calls_a_refresh_moving() {
let tally = |reads: &[i64]| {
let mut t = OracleTally::new(1);
let mut moved = Vec::new();
for pair in reads.windows(2) {
t.add(&pair[..1], &pair[1..]);
moved.push(t.moved());
}
(t, moved)
};
const A: i64 = 2_643_843_378;
let (b, c) = (A + 61_764_485, A + 61_764_485 + 3_435);
for (what, reads) in [
("one step", vec![A, A, c, c, c, c, c, c, c, c]),
("two steps in consecutive reads", vec![A, A, b, c, c, c, c, c, c, c]),
("two steps at the first two reads", vec![A, b, c, c, c, c, c]),
("two steps, just two reads", vec![A, b, c]),
] {
let (t, moved) = tally(&reads);
assert!(!moved.iter().any(|m| *m), "{what}: the oracle called a refresh moving: {t:?} {moved:?}");
}
let (t, moved) = tally(&[1, 2, 3, 4, 5]);
assert!(!t.moved(), "four changes of four reads: {t:?}");
assert_eq!(moved, [false; 4]);
let (t, _) = tally(&[1, 2, 3, 4, 5, 6]);
assert!(t.moved(), "a rail changing at every read, five reads: {t:?}");
// 8 changes of 10 reads (80%), and 7 of 10 (70%).
let (t, _) = tally(&[1, 2, 3, 4, 5, 5, 6, 7, 8, 9, 9]);
assert!(t.moved(), "80% of reads: {t:?}");
let (t, _) = tally(&[1, 2, 3, 3, 4, 5, 5, 6, 7, 7, 8]);
assert!(!t.moved(), "70% of reads: {t:?}");
// Two dies: moved only when both kept changing.
let dies = |d0: &[i64], d1: &[i64]| {
let mut t = OracleTally::new(2);
for i in 1..d0.len() {
t.add(&[d0[i - 1], d1[i - 1]], &[d0[i], d1[i]]);
}
t
};
let (moving, held) = ([1, 2, 3, 4, 5, 6], [9, 9, 9, 9, 9, 9]);
assert!(dies(&moving, &moving).moved(), "both dies changing at every read");
assert!(!dies(&moving, &held).moved(), "one die changing at every read, the other held");
assert!(!dies(&held, &moving).moved(), "the other die held");
assert!(!OracleTally::new(0).moved(), "no rail");
}
/// One oracle run: its verdict, how many reads it compared, whether any
/// CPU rail it read sits in a group named `Energy Model`, and the
/// channel counts of its first read.
#[derive(Debug)]
struct OracleRun {
verdict: Oracle,
reads: u32,
energy_model_rail: bool,
group: Option<Group>,
}
/// An oracle that shares only the raw FFI declarations and the CFString
/// helper (`cfstring_to_string`) with the liveness probe. It opens its
/// own subscription over every channel ([`OracleSub`]), reads each CPU
/// rail's ABSOLUTE cumulative value with its own comparison loop (not
/// `liveness::probe_with` or `RailWatch`), matches names with
/// `ends_with("CPU Energy")` in any group rather than the probe's group
/// filter and die-prefix stripping, and burns CPU on a helper thread for
/// the whole of its window while it reads.
///
/// `CpuRailMoved(true)` only if every CPU rail's absolute value changed
/// at [`ORACLE_MIN_CHANGES`] or more of its reads AND at at least 80%
/// of them. A live rail changes at nearly every read (the oracle's reads
/// of every channel are 0.1–0.8 s apart); a batch refresh changes a
/// rail at one read, or two when it is published in two steps (the M5
/// Max's `CPU Energy`: two steps 64 ms apart), and then holds. So a
/// refresh landing in both the probe's and the oracle's window cannot
/// vouch for a false Live: it gives the oracle too few changes, and
/// too small a share of its reads. Its own literals, not the
/// implementation's constants, so a change to one cannot silently move
/// the other.
///
/// Evidence: the window is `budget`, extended (to at most
/// [`ORACLE_MAX_BUDGETS`] budgets) only while the oracle has fewer than
/// [`ORACLE_MIN_CHANGES`] reads after its baseline — as many as a live
/// rail needs to meet the bar. Fewer at the end is a test FAILURE,
/// never `CpuRailMoved(false)`: too few reads cannot show a rail
/// keeping on moving.
fn oracle_cpu_rail(budget: Duration) -> OracleRun {
let none = |verdict, group| OracleRun { verdict, reads: 0, energy_model_rail: false, group };
let Some(sub) = OracleSub::open() else {
return none(Oracle::NoIoReport, None);
};
let Some(first) = sub.read() else {
return none(Oracle::NoIoReport, None);
};
let (group, v0) = (first.group, first.cpu);
if v0.is_empty() {
return none(Oracle::NoCpuRail, Some(group));
}
let energy_model_rail = v0.iter().any(|r| r.group.eq_ignore_ascii_case("Energy Model"));
// The burn runs on its own thread for the WHOLE window: this
// subscription samples every channel, 0.1–0.8 s a sample on an M5
// Max, so spinning only between samples would burn a small share of
// the window.
let stop = Arc::new(AtomicBool::new(false));
let burner = {
let stop = stop.clone();
std::thread::spawn(move || {
let mut acc: u64 = 0;
while !stop.load(Ordering::Relaxed) {
acc = std::hint::black_box(acc.wrapping_add(1));
}
})
};
let started = Instant::now();
let values = |rails: &[OracleRail]| rails.iter().map(|r| r.value).collect::<Vec<i64>>();
let mut tally = OracleTally::new(v0.len());
let mut prev = values(&v0);
while !tally.moved()
&& (started.elapsed() < budget || tally.reads < ORACLE_MIN_CHANGES)
&& started.elapsed() < budget * ORACLE_MAX_BUDGETS
{
let Some(cur) = sub.read().map(|r| values(&r.cpu)) else {
continue;
};
if cur.len() != prev.len() {
continue;
}
tally.add(&prev, &cur);
prev = cur;
}
stop.store(true, Ordering::Relaxed);
let _ = burner.join();
let took = started.elapsed();
let (reads, moved) = (tally.reads, tally.moved());
eprintln!("ioreport oracle: {reads} reads in {took:?}, changes per CPU rail {:?}", tally.changes);
assert!(
reads >= ORACLE_MIN_CHANGES,
"the oracle got only {reads} read(s) of every channel in {took:?} (window {budget:?}, extended \
to at most {ORACLE_MAX_BUDGETS}x while short of {ORACLE_MIN_CHANGES} reads): too few for a live \
CPU rail to show {ORACLE_MIN_CHANGES} changes, so this run has no evidence either way"
);
OracleRun { verdict: Oracle::CpuRailMoved(moved), reads, energy_model_rail, group: Some(group) }
}
// ── Pins on how a real probe run sampled ─────────────────────────────
/// F1 and R2, on one real probe run whose subscription opened (and, for
/// one that did not: no verdict, no read, no tick). F1: every sample it
/// read listed the Energy Model group alone — every channel in that
/// group, as many as the group has, far fewer than every channel — and
/// the median thread CPU of its reads stays under 50 ms. R2: every
/// tick's spin loop ran, and every interval either reached the tick of
/// CPU or spent the wall cap. It is not a CPU floor: a tick stops on
/// either condition, so a floor bounds the host's scheduler rather than
/// the probe. The reads themselves carry no wall-clock bound either —
/// another process sampling IOReport makes this probe wait, never
/// changes what it listed or what it burned. Returns the structural
/// line to log.
fn assert_sampling(run: &ProbeRun, who: &str, group: &Group) -> String {
if !run.opened {
assert_eq!(run.verdict, None, "{who}: no Energy Model subscription, yet a verdict");
assert!(run.calls.is_empty(), "{who}: read a subscription that did not open");
assert_eq!(run.ticks, 0, "{who}: ticked on a subscription that did not open");
return format!("ioreport structure: who={who} opened=false reads=0 ticks=0");
}
assert!(!run.calls.is_empty(), "{who}: the subscription opened, yet no read ran");
assert_eq!(run.ticks as usize, run.calls.len() - 1, "{who}: not one tick before every re-read");
let listed: Vec<Listed> = run.calls.iter().filter_map(|c| c.listed).collect();
assert!(!listed.is_empty(), "{who}: none of its {} reads returned a sample", run.calls.len());
for (i, l) in listed.iter().enumerate() {
assert!(
l.group_only,
"{who}: sample {i} listed {} channels, some outside the Energy Model group: the probe's \
subscription is not the group alone",
l.channels
);
assert_eq!(
l.channels, group.energy_model,
"{who}: sample {i} listed {} channels; the Energy Model group has {} (every channel: {})",
l.channels, group.energy_model, group.all
);
assert!(l.channels < group.all, "{who}: sample {i} listed {} channels of {}", l.channels, group.all);
}
let median = run
.median_read_cpu()
.unwrap_or_else(|| panic!("{who}: the thread CPU clock could not be read around any read"));
assert!(
median < MAX_MEDIAN_READ_CPU,
"{who}: the median read cost {median:?} of thread CPU (>= {MAX_MEDIAN_READ_CPU:?}) over {} reads: \
the probe is not sampling the Energy Model group alone (~6-9 ms a read; every channel: ~85-125 ms)",
run.calls.len()
);
let cpu = run.interval_cpu();
let wall = run.interval_wall();
assert_eq!(cpu.len(), run.intervals as usize, "{who}: one CPU figure per interval");
assert_eq!(wall.len(), cpu.len(), "{who}: one wall figure per interval");
assert_eq!(
run.tick_outcomes.len(),
run.ticks as usize,
"{who}: every tick reported an outcome"
);
// The claim is that the probe burns CPU before each re-read, and a
// CPU floor cannot check it. A tick stops on the CPU target OR the
// {TICK_WALL:?} wall cap, so a host that gives this thread a sliver
// of a core caps the tick with little CPU to show — this assertion
// used to require half of {TICK_CPU:?} and failed on the macOS
// runner that way, under other work on the same machine. Nor does
// CPU-per-turn separate a starved spin from a tick that slept:
// measured, a blocking turn costs the thread MORE CPU than an
// arithmetic one, because its context switch is charged here. So the
// tick reports whether its loop ran at all, which no load can forge.
for (i, out) in run.tick_outcomes.iter().enumerate() {
assert!(
out.spins > 0,
"{who}: tick {i} of {} reported {} turns of its spin loop: the probe is not burning CPU \
before each re-read, it waited or skipped (outcome {out:?})",
run.ticks,
out.spins
);
}
for (i, (c, w)) in cpu.iter().zip(&wall).enumerate() {
let c = c.unwrap_or_else(|| panic!("{who}: interval {i}: this thread's CPU clock could not be read"));
assert!(
c >= TICK_CPU || *w + Duration::from_millis(1) >= TICK_WALL,
"{who}: interval {i} of {} consumed {c:?} of the probing thread's CPU time between reads in \
{w:?} of wall time: short of the {TICK_CPU:?} tick, and short of the {TICK_WALL:?} wall cap, \
so neither stopping condition had been met",
run.intervals
);
}
format!(
"ioreport structure: who={who} reads={} sampled={} group_only=true channels={} energy_model_group={} \
all_channels={} median_read_cpu={median:?} median_read_wall={:?} min_interval_cpu={:?} ticks={}",
run.calls.len(),
listed.len(),
listed[0].channels,
group.energy_model,
group.all,
run.median_read_wall(),
cpu.iter().flatten().min(),
run.ticks
)
}
/// The tests' own reading of a probe record, with their own loop and
/// their own literals (not `RailWatch` or its constants): the intervals
/// between consecutive recorded samples, each placed at the wall time
/// its closing read returned, and the first interval, if any, after
/// which EVERY CPU rail seen in any sample (the first included) had
/// risen in at least 5 intervals, in at least 80% of those so far,
/// over at least 300 ms from the first sample.
/// The run's verdict must agree with it, and a Live run must have
/// stopped at that very interval.
fn assert_verdict_matches_record(run: &ProbeRun, who: &str) {
let at: Vec<Duration> = run.calls.iter().filter(|c| c.listed.is_some()).map(|c| c.end).collect();
assert_eq!(run.seen.len(), at.len(), "{who}: one record per sample");
let mut rises: Vec<(RailKey, u32)> =
run.seen.first().map(|s| s.cpu.iter().map(|(k, _)| (k.clone(), 0)).collect()).unwrap_or_default();
let mut first_live = None;
for (i, pair) in run.seen.windows(2).enumerate() {
for (key, now) in &pair[1].cpu {
let rose = pair[0].cpu.iter().any(|(k, before)| k == key && now > before);
match rises.iter_mut().find(|(k, _)| k == key) {
Some((_, n)) => *n += u32::from(rose),
None => rises.push((key.clone(), u32::from(rose))),
}
}
let intervals = i as u32 + 1;
let window = at[i + 1].saturating_sub(at[0]);
if first_live.is_none()
&& window >= Duration::from_millis(300)
&& !rises.is_empty()
&& rises.iter().all(|(_, n)| *n >= 5 && *n * 5 >= intervals * 4)
{
first_live = Some(intervals);
}
}
let intervals = run.seen.len().saturating_sub(1);
assert_eq!(run.intervals as usize, intervals, "{who}: intervals, against its own record");
if intervals > 0 {
assert_eq!(
run.verdict == Some(CpuRail::Live),
first_live.is_some(),
"{who}: verdict {:?}, yet by its own record the CPU rails rose {rises:?} in {intervals} intervals \
over {:?}",
run.verdict,
at[intervals].saturating_sub(at[0])
);
if let Some(n) = first_live {
assert_eq!(n as usize, intervals, "{who}: Live by interval {n}, yet it sampled on to {intervals}");
}
} else {
assert_eq!(run.verdict, None, "{who}: a verdict with no interval");
}
}
/// The property that matters: the IOReport meter is **never** available
/// when its CPU rail is frozen, because then it would bracket GPU-only
/// energy, zero, or a batch refresh as the energy of CPU work. "The
/// IOReport meter" is the bracket rung's own gate (`bracket_admitted`),
/// which `power_over` and the bracket go through whatever source ranks
/// above IOReport, and the selection never names IOReport unless it
/// is admitted.
/// Deterministic on any host — it asserts the meter's decision against
/// what the rail actually did, and every branch asserts something.
///
/// The oracle gets three times the probe's budget, so a rail too slow for
/// the probe but moving within the oracle's window is the only case that
/// may be declined while live — and that fails safe: no hardware figure,
/// never an over-claimed one. The cached run that made the decision must
/// also have sampled the Energy Model group alone and burned CPU in
/// every interval ([`assert_sampling`]), its status must be the status
/// of that run, and its verdict must agree with its own record.
#[test]
fn the_ioreport_meter_is_never_available_on_a_frozen_cpu_rail() {
let status = ioreport_status();
let cached = host();
let run = &cached.run;
let selected = bracket_admitted();
let oracle = oracle_cpu_rail(CPU_RAIL_LIVENESS_BUDGET * 3);
// Recorded so a log of a macOS run (manual, or a CI job that runs
// this crate's lib tests with --nocapture) says which branch below
// was actually asserted, and what the probe's sampling looked like.
eprintln!(
"ioreport liveness: status={status:?} oracle={:?} selected={selected} probe_intervals={} \
opened={} median_read_cpu={:?} min_interval_cpu={:?} oracle_reads={}",
oracle.verdict,
run.intervals,
run.opened,
run.median_read_cpu(),
run.interval_cpu().iter().flatten().min(),
oracle.reads
);
assert_eq!(status, cached.status, "ioreport_status() is not the cached status");
assert_eq!(cached.status, status_of(run), "the cached status is not the status of the cached run");
assert_eq!(selected, admits(status), "the selection is not admits(the cached status)");
let group = oracle.group.unwrap_or(Group { energy_model: 0, all: 0 });
eprintln!("{}", assert_sampling(run, "the cached probe", &group));
assert_verdict_matches_record(run, "the cached probe");
if oracle.energy_model_rail {
assert!(run.opened, "the oracle read a CPU rail in the Energy Model group; the probe could not subscribe to it");
}
match oracle.verdict {
Oracle::NoIoReport => {
assert_eq!(status, IoReportStatus::Unavailable);
assert!(!selected, "a hardware meter was selected with no IOReport");
}
Oracle::NoCpuRail => {
let want =
if run.intervals > 0 { IoReportStatus::CpuRailAbsent } else { IoReportStatus::Unavailable };
assert_eq!(status, want, "no CPU rail anywhere, probe intervals {}", run.intervals);
assert!(!selected, "selected an IOReport meter that has no CPU rail");
}
Oracle::CpuRailMoved(false) => {
assert!(
!selected,
"IOReport meter available although not every CPU rail kept moving under {:?} of \
CPU burn — it would bracket GPU-only energy, zero, or a refresh as the energy of \
CPU work",
CPU_RAIL_LIVENESS_BUDGET * 3
);
// A probe that sampled the rail calls it frozen; one that
// could not sample at all says Unavailable. Both decline.
let want =
if run.intervals > 0 { IoReportStatus::CpuRailFrozen } else { IoReportStatus::Unavailable };
assert_eq!(status, want, "probe intervals {}", run.intervals);
// Declined end to end: no figure comes out of a frozen rail,
// and the selection never names it.
assert_eq!(power_over(Duration::from_millis(200)), None, "watts read from a frozen CPU rail");
assert_eq!(bracket_marginal_uj(cpu_work), None, "energy metered from a frozen CPU rail");
assert_ne!(meter_source(), Some(MeterSource::IoReport), "a frozen CPU rail was selected");
}
Oracle::CpuRailMoved(true) => {
// Live rails. The probe had a third of the oracle's time:
// selected if its own record shows every CPU rail kept
// rising (checked above), declined — the safe side — if
// not. It sampled a rail either way.
if oracle.energy_model_rail {
assert!(
matches!(status, IoReportStatus::Live | IoReportStatus::CpuRailFrozen),
"the oracle saw every Energy Model CPU rail keep moving; the probe said {status:?}"
);
} else {
assert_ne!(status, IoReportStatus::Live, "Live with no CPU rail in the Energy Model group");
}
}
}
// The IOReport rung is admitted exactly when Live.
assert_eq!(
selected,
status == IoReportStatus::Live,
"bracket_admitted() disagrees with ioreport_status() = {status:?}"
);
}
/// The dangerous direction on its own: the probe never reports a live rail
/// the oracle cannot see keep moving. The REAL production composition
/// runs fresh here (`host_status()`, uncached), and the oracle runs
/// unconditionally, so every outcome is checked against it, not only
/// Live. The fresh run must also pass [`assert_sampling`] and agree with
/// its own record. Logs how long the probe took, how many intervals it
/// saw, and its median read cost.
#[test]
fn the_liveness_probe_never_reports_a_frozen_rail_as_live() {
let t = Instant::now();
let fresh = host_status();
let probe_took = t.elapsed();
let run = &fresh.run;
let status = ioreport_status();
let oracle = oracle_cpu_rail(CPU_RAIL_LIVENESS_BUDGET * 3);
eprintln!(
"ioreport probe: probe={:?} fresh_status={:?} status={status:?} oracle={:?} probe_took={probe_took:?} \
intervals={} opened={} reads={} median_read_cpu={:?} min_interval_cpu={:?} oracle_reads={}",
run.verdict,
fresh.status,
oracle.verdict,
run.intervals,
run.opened,
run.calls.len(),
run.median_read_cpu(),
run.interval_cpu().iter().flatten().min(),
oracle.reads
);
assert_eq!(fresh.status, status_of(run), "host_status()'s status is not the status of its run");
let group = oracle.group.unwrap_or(Group { energy_model: 0, all: 0 });
eprintln!("{}", assert_sampling(run, "the fresh probe", &group));
assert_verdict_matches_record(run, "the fresh probe");
if oracle.energy_model_rail {
assert!(run.opened, "the oracle read a CPU rail in the Energy Model group; the probe could not subscribe to it");
}
match (run.verdict, status) {
(Some(CpuRail::Live), _) | (_, IoReportStatus::Live) => assert_eq!(
oracle.verdict,
Oracle::CpuRailMoved(true),
"probe said Live (probe={:?}, status={status:?}), the oracle did not see every CPU rail keep \
moving",
run.verdict
),
(Some(CpuRail::Frozen), _) => assert!(
matches!(oracle.verdict, Oracle::CpuRailMoved(_)),
"the probe saw a CPU rail, the oracle found none: {:?}",
oracle.verdict
),
(Some(CpuRail::Absent), _) => assert_eq!(
oracle.verdict,
Oracle::NoCpuRail,
"the probe found no CPU rail, the oracle found one"
),
(None, _) => assert!(
!oracle.energy_model_rail,
"the probe could not open or sample the Energy Model group, yet the oracle read a CPU rail \
in it: {:?}",
oracle.verdict
),
}
}
/// The FFI value pin: the probe's reader reads each CPU rail's raw
/// integer exactly as the oracle's independent absolute read of the
/// same rail does, and lists the Energy Model group alone. Each round
/// brackets one read between two oracle reads. The CPU and GPU keys the
/// read hands the rule must be the oracle's Energy Model keys. Where the
/// oracle saw a rail hold across the bracket, the read must EQUAL it;
/// where it moved (the GPU rail usually does), the read must lie
/// between the two oracle reads — so a constant, a stale first read, a
/// sum of rails, a dropped rail or the wrong rail fails here, wherever
/// the rail it touches is distinguishable on this host.
#[test]
fn the_probe_sampler_reads_each_cpu_rail_as_the_oracle_does() {
let Some(oracle) = OracleSub::open() else {
eprintln!("ioreport ffi pin: IOReport does not open on this host; nothing to compare");
assert_ne!(ioreport_status(), IoReportStatus::Live, "Live with no IOReport");
return;
};
let first = oracle.read().expect("the oracle's subscription yields a sample");
if first.cpu_keyed().is_empty() {
eprintln!("ioreport ffi pin: no CPU Energy channel in the Energy Model group; nothing to compare");
assert_ne!(ioreport_status(), IoReportStatus::Live, "Live with no Energy Model CPU rail");
return;
}
let probe = EnergyModelSub::open()
.expect("the oracle reads a CPU rail in the Energy Model group; the probe cannot subscribe to it");
for round in 1..=3 {
let before = oracle.read().expect("oracle read before the probe's read");
let read = probe.read().expect("the probe's subscription yields a sample");
let after = oracle.read().expect("oracle read after the probe's read");
assert!(read.listed.group_only, "round {round}: the probe's read listed channels outside the group");
assert_eq!(
read.listed.channels, before.group.energy_model,
"round {round}: the probe's read listed {} channels; the Energy Model group has {}",
read.listed.channels, before.group.energy_model
);
let cpu = expect_keyed(&read.sample.iter().map(|(n, v)| (n.as_str(), *v)).collect::<Vec<_>>(), |n| {
n.ends_with("CPU Energy")
});
let gpu = expect_keyed(&read.sample.iter().map(|(n, v)| (n.as_str(), *v)).collect::<Vec<_>>(), |n| {
n == "GPU Energy"
});
assert_eq!(keys(&cpu), keys(&before.cpu_keyed()), "round {round}: the CPU rails read, against the oracle's");
assert_eq!(keys(&gpu), keys(&before.gpu_keyed()), "round {round}: the GPU rails read, against the oracle's");
for (what, probed, (b, a)) in
[("cpu", &cpu, (before.cpu_keyed(), after.cpu_keyed())), ("gpu", &gpu, (before.gpu_keyed(), after.gpu_keyed()))]
{
for ((key, value), ((_, lo), (_, hi))) in probed.iter().zip(b.iter().zip(a.iter())) {
let held = lo == hi;
eprintln!(
"ioreport ffi pin: round={round} {what}={}#{} oracle_before={lo} probe={value} \
oracle_after={hi} held={held}",
key.0, key.1
);
if held {
assert_eq!(
value, lo,
"round {round}: {key:?} held at {lo} across the oracle's reads, yet the probe read {value}"
);
} else {
assert!(
lo.min(hi) <= value && value <= lo.max(hi),
"round {round}: {key:?} moved {lo} -> {hi} across the oracle's reads, yet the probe \
read {value}, outside that range"
);
}
}
}
}
}
/// The raw reader reads the subscription it is handed, every time.
/// Handed the probe's own Energy Model subscription and an all-channel
/// one (`open()`) in turn, it lists the group's channels, then every
/// channel, then the group's again — as many as the oracle counts in
/// each. A reader that hands back an earlier read (stale for the
/// process) fails here whether or not any channel moves, which the GPU
/// freshness pin cannot promise on a host whose GPU is idle.
#[test]
fn the_raw_reader_reads_the_subscription_it_is_given_every_time() {
let Some(oracle) = OracleSub::open() else {
eprintln!("ioreport raw reader: IOReport does not open on this host; nothing to compare");
assert_ne!(ioreport_status(), IoReportStatus::Live, "Live with no IOReport");
return;
};
let group = oracle.read().expect("the oracle's subscription yields a sample").group;
let probe = EnergyModelSub::open().expect("the oracle reads IOReport; the probe's group subscription does not open");
// SAFETY: open returns owned, non-null refs, released below.
let (sub, ch) = unsafe { open() }.expect("the probe's group opens, a subscription to every channel does not");
let mut listed = Vec::new();
for _ in 0..3 {
// SAFETY: both subscriptions are live until released below.
let g = unsafe { energy_model_channels(probe.sub, probe.ch) }.expect("the group subscription yields a sample");
let a = unsafe { energy_model_channels(sub, ch) }.expect("the all-channel subscription yields a sample");
listed.push((g.len(), a.len()));
}
// SAFETY: released once, here.
unsafe {
CFRelease(sub);
CFRelease(ch as CFTypeRef);
}
eprintln!(
"ioreport raw reader: (group, every channel) listed {listed:?}; the oracle counts ({}, {})",
group.energy_model, group.all
);
for (i, (g, a)) in listed.iter().enumerate() {
assert_eq!(
(*g, *a),
(group.energy_model, group.all),
"read {i}: the raw reader listed {g} channels for the group and {a} for every channel; the oracle \
counts {} and {}: it did not read the subscription it was handed",
group.energy_model,
group.all
);
}
}
/// A GPU Energy watcher for the freshness pin: a subscription of its
/// own to the Energy Model group, opened here with the raw FFI (not the
/// probe's code), read back to back on a thread of its own. Each read is
/// placed on the probe's timeline (`wall_clock`), so a change it saw can
/// be placed strictly between two of the probe's reads.
struct GpuWatch {
stop: Arc<AtomicBool>,
handle: std::thread::JoinHandle<Vec<GpuRead>>,
}
/// One read by the watcher: when it began and returned, on the probe's
/// clock, and the `GPU Energy` values, in channel order.
#[derive(Debug, Clone)]
struct GpuRead {
start: Duration,
end: Duration,
gpu: Vec<i64>,
}
impl GpuWatch {
fn start() -> Self {
let stop = Arc::new(AtomicBool::new(false));
let handle = {
let stop = stop.clone();
std::thread::spawn(move || unsafe {
let group = CFStringCreateWithCString(
std::ptr::null_mut(),
c"Energy Model".as_ptr() as *const _,
K_CFSTRING_ENCODING_UTF8,
);
assert!(!group.is_null(), "the watcher's group name");
let desired = IOReportCopyChannelsInGroup(group, std::ptr::null(), 0, 0, 0);
CFRelease(group);
assert!(!desired.is_null(), "the watcher cannot list the Energy Model group");
let mut ch: CFMutableDictionaryRef = std::ptr::null_mut();
let sub = IOReportCreateSubscription(std::ptr::null(), desired, &mut ch as *mut _, 0, std::ptr::null());
CFRelease(desired as CFTypeRef);
assert!(!sub.is_null() && !ch.is_null(), "the watcher cannot subscribe to the Energy Model group");
let key = CFStringCreateWithCString(
std::ptr::null_mut(),
c"IOReportChannels".as_ptr() as *const _,
K_CFSTRING_ENCODING_UTF8,
);
let mut reads = Vec::new();
while !stop.load(Ordering::SeqCst) {
let start = wall_clock();
let smp = IOReportCreateSamples(sub, ch, std::ptr::null());
let end = wall_clock();
if smp.is_null() {
continue;
}
let arr = CFDictionaryGetValue(smp, key as *const _) as CFArrayRef;
let mut gpu = Vec::new();
if !arr.is_null() {
for i in 0..CFArrayGetCount(arr) {
let it = CFArrayGetValueAtIndex(arr, i);
if !it.is_null()
&& cfstring_to_string(IOReportChannelGetChannelName(it)).as_deref() == Some("GPU Energy")
{
gpu.push(IOReportSimpleGetIntegerValue(it, 0) as i64);
}
}
}
CFRelease(smp);
reads.push(GpuRead { start, end, gpu });
}
CFRelease(key);
CFRelease(sub);
CFRelease(ch as CFTypeRef);
reads
})
};
Self { stop, handle }
}
fn stop(self) -> Vec<GpuRead> {
self.stop.store(true, Ordering::SeqCst);
self.handle.join().expect("the GPU watcher panicked")
}
}
/// F4, what the REAL production composition hands the rule. The very
/// function the cache runs, `host_status()`, runs fresh (uncached)
/// between two all-channel oracle reads, while a GPU watcher reads the
/// group on its own thread. For every sample the run handed the rule:
/// its CPU and GPU keys are the oracle's Energy Model keys; a rail the
/// oracle saw hold was handed exactly the oracle's value, and a rail
/// that moved a value within the oracle's two reads. Freshness: for
/// every rise of GPU Energy the watcher saw strictly between two of the
/// probe's reads, the GPU values handed to the rule for those two reads
/// must differ — a sampler that returns a first read forever (per probe
/// or per process) fails here. Asserted whenever the watcher saw such a
/// rise — on an M5 Max usually every ~28 ms, but an idle GPU holds, and
/// on an M3 Ultra it moves only sporadically — and the log line says how
/// many were checked, so a run whose GPU held says so. A reader that is
/// stale for the whole process is also caught without any moving
/// channel ([`the_raw_reader_reads_the_subscription_it_is_given_every_time`]).
/// The run must also pass [`assert_sampling`].
#[test]
fn the_production_probe_hands_the_rule_what_the_oracle_reads() {
let Some(oracle) = OracleSub::open() else {
eprintln!("ioreport freshness: IOReport does not open on this host; nothing to compare");
assert_ne!(ioreport_status(), IoReportStatus::Live, "Live with no IOReport");
return;
};
let before = oracle.read().expect("the oracle's read before the probe");
let watch = GpuWatch::start();
let fresh = host_status();
let watched = watch.stop();
let after = oracle.read().expect("the oracle's read after the probe");
let run = &fresh.run;
assert_eq!(fresh.status, status_of(run), "host_status()'s status is not the status of its run");
eprintln!("{}", assert_sampling(run, "host_status()", &before.group));
assert_verdict_matches_record(run, "host_status()");
if before.cpu_keyed().is_empty() {
assert_ne!(fresh.status, IoReportStatus::Live, "Live with no Energy Model CPU rail");
}
let (cpu_keys, gpu_keys) = (keys(&before.cpu_keyed()), keys(&before.gpu_keyed()));
for (k, s) in run.seen.iter().enumerate() {
assert_eq!(keys(&s.cpu), cpu_keys, "sample {k}: the CPU rails handed to the rule, against the oracle's");
assert_eq!(keys(&s.gpu), gpu_keys, "sample {k}: the GPU rails handed to the rule, against the oracle's");
}
for (what, bracket) in [
("cpu", before.cpu_keyed().into_iter().zip(after.cpu_keyed()).collect::<Vec<_>>()),
("gpu", before.gpu_keyed().into_iter().zip(after.gpu_keyed()).collect::<Vec<_>>()),
] {
for ((key, lo), (_, hi)) in &bracket {
let handed: Vec<i64> = run
.seen
.iter()
.map(|s| {
let rails = if what == "cpu" { &s.cpu } else { &s.gpu };
rails.iter().find(|(k, _)| k == key).map(|(_, v)| *v).unwrap_or(i64::MIN)
})
.collect();
let (min, max) = (handed.iter().min(), handed.iter().max());
eprintln!(
"ioreport freshness: {what}={}#{} oracle_before={lo} oracle_after={hi} handed={} values in \
[{min:?}, {max:?}]",
key.0,
key.1,
handed.len()
);
for (k, v) in handed.iter().enumerate() {
if lo == hi {
assert_eq!(v, lo, "sample {k}: {key:?} held at {lo} across the probe, yet it handed the rule {v}");
} else {
assert!(
lo.min(hi) <= v && v <= lo.max(hi),
"sample {k}: {key:?} moved {lo} -> {hi} across the probe, yet it handed the rule {v}"
);
}
}
}
}
// Freshness, on the watcher's rises of GPU Energy.
let sampled: Vec<(Duration, Duration, Vec<i64>)> = run
.calls
.iter()
.filter(|c| c.listed.is_some())
.zip(&run.seen)
.map(|(c, s)| (c.start, c.end, s.gpu.iter().map(|(_, v)| *v).collect()))
.collect();
let (mut rises, mut checked) = (0usize, 0usize);
for pair in watched.windows(2) {
let (x, y) = (&pair[0], &pair[1]);
let rose = x.gpu.len() == y.gpu.len()
&& x.gpu.iter().zip(&y.gpu).all(|(a, b)| b >= a)
&& x.gpu != y.gpu;
if !rose {
continue;
}
rises += 1;
// The last probe read that returned before the watcher's first
// read began, and the first that began after its second returned.
let i = sampled.iter().rposition(|(_, end, _)| *end <= x.start);
let j = sampled.iter().position(|(start, _, _)| *start >= y.end);
let (Some(i), Some(j)) = (i, j) else {
continue;
};
checked += 1;
assert_ne!(
sampled[i].2, sampled[j].2,
"GPU Energy rose {:?} -> {:?} between the probe's reads {i} and {j} (the watcher read it at \
{:?} and {:?}), yet both handed the rule the same GPU value: the probe's reads are stale",
x.gpu, y.gpu, x.start, y.end
);
}
eprintln!(
"ioreport freshness: watcher_reads={} gpu_rises={rises} checked_between_probe_reads={checked} \
probe_samples={}{}",
watched.len(),
sampled.len(),
if checked == 0 { " (not asserted: the watcher saw GPU Energy hold across the probe's reads)" } else { "" }
);
}
// ── The production composition on scripted reads ──────────────────
/// Channels as the probe's own subscription lists them: all in the
/// Energy Model group.
fn em_channels(pairs: &[(&str, i64)]) -> Vec<Channel> {
pairs.iter().map(|(n, v)| (Some("Energy Model".to_string()), Some(n.to_string()), *v)).collect()
}
/// The REAL production composition, `host_status()`, run on this thread
/// with every channel read scripted — read `n` lists `script(n)` — and
/// the real tick and clocks. Anything between the raw channel read and
/// the rule runs as in production. Checks that every one of the run's
/// real ticks actually span, which is what "burned CPU" means here and
/// the only form of it a loaded host cannot forge — see
/// [`assert_sampling`] for why a CPU floor, aggregate or per interval,
/// bounds the scheduler rather than the probe.
fn host_status_scripted(script: fn(u32) -> Vec<(&'static str, i64)>) -> HostStatus {
let mut n = 0u32;
let next = move || {
let channels = em_channels(&script(n));
n += 1;
Some(channels)
};
let c0 = test_thread_cpu();
let hs = seam::with(seam::Mode::Script(Box::new(next)), host_status);
let burned = test_thread_cpu() - c0;
assert!(hs.run.ticks > 0, "a scripted run that never ticked: {:?}", hs.run);
// Per tick, that its spin loop ran — not an aggregate CPU floor.
// A floor of half a tick per tick is a bound on what the host's
// scheduler allowed, not on what the probe did: every tick can hit
// its wall cap on a loaded box and the total falls under the line
// while the probe behaved correctly.
for (i, out) in hs.run.tick_outcomes.iter().enumerate() {
assert!(
out.spins > 0,
"host_status() tick {i} of {} reported {} turns of its spin loop: it did not burn CPU \
(outcome {out:?}, this thread burned {burned:?} over the whole call)",
hs.run.ticks,
out.spins
);
}
hs
}
/// Every sample the scripted run handed the rule is the script's, as
/// the tests key it: no rail summed, zeroed, dropped, stale or shifted.
fn assert_handed_the_script(run: &ProbeRun, script: fn(u32) -> Vec<(&'static str, i64)>) {
assert_eq!(run.seen.len(), run.calls.len(), "every scripted read succeeds");
for (n, seen) in run.seen.iter().enumerate() {
let s = script(n as u32);
assert_eq!(seen.cpu, expect_keyed(&s, |c| c.ends_with("CPU Energy")), "read {n}: the CPU rails handed to the rule");
assert_eq!(seen.gpu, expect_keyed(&s, |c| c == "GPU Energy"), "read {n}: the GPU rail handed to the rule");
}
}
fn scripted_group(script: fn(u32) -> Vec<(&'static str, i64)>) -> Group {
Group { energy_model: script(0).len(), all: usize::MAX }
}
/// A single batch jump of the M5 Max's `CPU Energy`, beside a GPU rail
/// that moves every read.
fn m5_refresh(n: u32) -> Vec<(&'static str, i64)> {
let cpu = if n < 3 { 480_083_160 } else { 541_893_638 };
vec![("MCPU0", 7), ("CPU Energy", cpu), ("PCPU", 9), ("GPU Energy", 851_622_003 + 28_000_000 * i64::from(n))]
}
/// A CPU rail that holds at one value, as both Macs' rails do between
/// refreshes, beside a GPU rail that moves every read.
fn constant(n: u32) -> Vec<(&'static str, i64)> {
vec![("CPU Energy", 2_588_973_125), ("GPU Energy", 126_266_724_972_887 + i64::from(n))]
}
/// One CPU rail advancing at every read, for as long as it is read.
fn advancing(n: u32) -> Vec<(&'static str, i64)> {
vec![("CPU Energy", 1_000 + i64::from(n)), ("GPU Energy", 5)]
}
/// The refresh observed on an M5 Max (macOS 27.0.1), published in two
/// steps: `CPU Energy` +61,764,485 at read 3, +3,435 at read 4, then
/// held; `MCPU0`, `MCPU1` and `PCPU` step twice with it, and the GPU
/// rail moves every read.
fn two_step_refresh(n: u32) -> Vec<(&'static str, i64)> {
let step = |first: i64, second: i64| match n {
0..=2 => 0,
3 => first,
_ => first + second,
};
vec![
("MCPU0", 91_000_000 + step(20_000_000, 1_000)),
("MCPU1", 92_000_000 + step(21_000_000, 1_100)),
("PCPU", 400_000_000 + step(20_764_485, 1_335)),
("CPU Energy", 2_643_843_378 + step(61_764_485, 3_435)),
("GPU Energy", 851_622_003 + 28_000_000 * i64::from(n)),
]
}
/// An M3 Ultra's two dies advancing in alternate intervals — each in
/// half of them, together in every one — among a per-die sub-rail
/// listed once per die and a moving GPU rail.
fn two_dies(n: u32) -> Vec<(&'static str, i64)> {
vec![
("ECPUDTL00", 5_000),
("DIE_0_CPU Energy", 9_494_336 + 1_344 * i64::from(n.div_ceil(2))),
("ECPUDTL00", 7_000),
("DIE_1_CPU Energy", 10_212_656 + 1_391 * i64::from(n / 2)),
("GPU Energy", 14_809_813 + i64::from(n)),
]
}
/// An M3 Ultra's two dies both advancing at every read, among a
/// per-die sub-rail listed once per die and a moving GPU rail.
fn two_live_dies(n: u32) -> Vec<(&'static str, i64)> {
vec![
("ECPUDTL00", 5_000),
("DIE_0_CPU Energy", 9_494_336 + 1_344 * i64::from(n)),
("ECPUDTL00", 7_000),
("DIE_1_CPU Energy", 10_212_656 + 1_391 * i64::from(n)),
("GPU Energy", 14_809_813 + i64::from(n)),
]
}
/// An M3 Ultra's `DIE_0_CPU Energy` advancing at every read while
/// `DIE_1_CPU Energy` holds, beside a moving GPU rail.
fn one_live_die_one_held(n: u32) -> Vec<(&'static str, i64)> {
vec![
("ECPUDTL00", 5_000),
("DIE_0_CPU Energy", 9_494_336 + 1_344 * i64::from(n)),
("ECPUDTL00", 7_000),
("DIE_1_CPU Energy", 10_212_656),
("GPU Energy", 14_809_813 + i64::from(n)),
]
}
/// `host_status()` on an Energy Model group that cannot be opened: no
/// read, no tick, no verdict, `Unavailable` — never `Live`, never
/// `CpuRailAbsent` — and the meter is not admitted.
#[test]
fn host_status_on_an_unopenable_group_is_unavailable_and_never_reads_or_ticks() {
let hs = seam::with(seam::Mode::Deny, host_status);
eprintln!("ioreport scripted: unopenable -> status={:?} run={:?}", hs.status, hs.run);
assert!(!hs.run.opened, "an unopenable group opened");
assert!(hs.run.calls.is_empty(), "read a subscription that did not open");
assert_eq!(hs.run.ticks, 0, "ticked on a subscription that did not open");
assert_eq!(hs.run.verdict, None, "a verdict with no subscription");
assert_eq!(hs.status, IoReportStatus::Unavailable, "an unopenable Energy Model group");
assert!(!admits(hs.status), "the meter admitted with no subscription");
}
/// `host_status()` on a single batch jump of the CPU rail: Frozen,
/// declined, the real tick burned in every interval, and the rule was
/// handed exactly what was read.
#[test]
fn host_status_calls_a_single_batch_jump_frozen() {
let hs = host_status_scripted(m5_refresh);
let line = assert_sampling(&hs.run, "scripted single jump", &scripted_group(m5_refresh));
eprintln!("ioreport scripted: single jump -> status={:?}; {line}", hs.status);
assert_eq!(hs.run.verdict, Some(CpuRail::Frozen), "one refresh counted as a live rail");
assert_eq!(hs.status, IoReportStatus::CpuRailFrozen);
assert!(!admits(hs.status), "a batch refresh admitted the meter");
assert_handed_the_script(&hs.run, m5_refresh);
}
/// `host_status()` on a CPU rail that never moves: Frozen for the whole
/// budget, declined, the real tick burned in every interval, and the
/// rule was handed exactly what was read.
#[test]
fn host_status_calls_a_constant_rail_frozen() {
let hs = host_status_scripted(constant);
let line = assert_sampling(&hs.run, "scripted constant", &scripted_group(constant));
eprintln!("ioreport scripted: constant -> status={:?}; {line}", hs.status);
assert_eq!(hs.run.verdict, Some(CpuRail::Frozen), "a rail that never moved");
assert_eq!(hs.status, IoReportStatus::CpuRailFrozen);
assert!(!admits(hs.status), "a constant rail admitted the meter");
assert!(hs.run.intervals >= 5, "frozen on {} interval(s)", hs.run.intervals);
assert_handed_the_script(&hs.run, constant);
assert_verdict_matches_record(&hs.run, "scripted constant");
}
/// `host_status()` on a CPU rail that keeps advancing, at every read:
/// Live, admitted — and only once the rail has advanced in at least
/// five intervals over at least 300 ms of the real wall clock (the
/// tests' own literals), at the first interval where that holds, where
/// it stops at once.
#[test]
fn host_status_calls_a_sustained_advance_of_one_rail_live() {
let hs = host_status_scripted(advancing);
let run = &hs.run;
let window = run.calls.last().map(|c| c.end).unwrap_or_default().saturating_sub(run.calls[0].end);
let line = assert_sampling(run, "scripted sustained advance", &scripted_group(advancing));
eprintln!(
"ioreport scripted: sustained advance -> status={:?} intervals={} window={window:?}; {line}",
hs.status, run.intervals
);
assert_eq!(run.verdict, Some(CpuRail::Live), "a rail advancing at every read");
assert_eq!(hs.status, IoReportStatus::Live);
assert!(admits(hs.status), "a live rail not admitted");
assert!(run.intervals >= 5, "Live after {} interval(s)", run.intervals);
assert!(window >= Duration::from_millis(300), "Live after {window:?} of window");
assert_eq!((run.calls.len(), run.ticks as usize), (run.intervals as usize + 1, run.intervals as usize));
assert_verdict_matches_record(run, "scripted sustained advance");
assert_handed_the_script(run, advancing);
// Covered per tick by `assert_sampling` above; here only that every
// interval's CPU was readable at all. The old `>= TICK_CPU / 2` was
// a bound on the host's scheduler, not on the probe.
let cpu = run.interval_cpu();
assert!(cpu.iter().all(Option::is_some), "an interval's thread CPU clock could not be read: {cpu:?}");
assert!(
run.tick_outcomes.iter().all(|t| t.spins > 0),
"a real tick did not spin: {:?}",
run.tick_outcomes
);
}
/// `host_status()` on the two-step refresh observed on an M5 Max: the
/// same rail advances in two consecutive intervals and then holds.
/// `CpuRailFrozen`, declined, after the whole budget, with the real
/// tick burned in every interval and the rule handed exactly what was
/// read. The "two advances" rule called this Live.
#[test]
fn host_status_calls_a_two_step_refresh_frozen() {
let hs = host_status_scripted(two_step_refresh);
let line = assert_sampling(&hs.run, "scripted two-step refresh", &scripted_group(two_step_refresh));
eprintln!("ioreport scripted: two-step refresh -> status={:?}; {line}", hs.status);
assert_eq!(hs.run.verdict, Some(CpuRail::Frozen), "a two-step refresh counted as a live rail");
assert_eq!(hs.status, IoReportStatus::CpuRailFrozen);
assert!(!admits(hs.status), "a two-step refresh admitted the meter");
assert!(hs.run.intervals >= 5, "frozen on {} interval(s)", hs.run.intervals);
assert_handed_the_script(&hs.run, two_step_refresh);
assert_verdict_matches_record(&hs.run, "scripted two-step refresh");
}
/// `host_status()` on two dies advancing in alternate intervals: each
/// die in half of them, so Frozen. A sum of the CPU rails placed
/// anywhere between the raw read and the rule advances in every
/// interval, reads as live and fails here, on any host; so does any
/// rail handed a value other than its own.
#[test]
fn host_status_calls_two_dies_advancing_in_alternate_intervals_frozen() {
let hs = host_status_scripted(two_dies);
let line = assert_sampling(&hs.run, "scripted two dies", &scripted_group(two_dies));
eprintln!("ioreport scripted: two dies -> status={:?}; {line}", hs.status);
assert_eq!(hs.run.verdict, Some(CpuRail::Frozen), "two dies' alternate advances added up to a live rail");
assert_eq!(hs.status, IoReportStatus::CpuRailFrozen);
assert_handed_the_script(&hs.run, two_dies);
assert_verdict_matches_record(&hs.run, "scripted two dies");
}
/// (a) `host_status()` on two dies that both advance at every read:
/// Live, admitted, at the first interval where both have risen in at
/// least five intervals over at least 300 ms of the real wall clock.
#[test]
fn host_status_calls_two_dies_that_keep_advancing_live() {
let hs = host_status_scripted(two_live_dies);
let line = assert_sampling(&hs.run, "scripted two live dies", &scripted_group(two_live_dies));
eprintln!("ioreport scripted: two live dies -> status={:?} intervals={}; {line}", hs.status, hs.run.intervals);
assert_eq!(hs.run.verdict, Some(CpuRail::Live), "two dies that kept advancing");
assert_eq!(hs.status, IoReportStatus::Live);
assert!(admits(hs.status), "two live dies not admitted");
assert!(hs.run.intervals >= 5, "Live after {} interval(s)", hs.run.intervals);
assert_handed_the_script(&hs.run, two_live_dies);
assert_verdict_matches_record(&hs.run, "scripted two live dies");
}
/// (b) `host_status()` on `DIE_0_CPU Energy` advancing at every read
/// while `DIE_1_CPU Energy` holds: `CpuRailFrozen`, declined, after
/// the whole budget. The power reading sums both dies, so a Live here
/// would bracket one die's energy as the whole CPU's. The one-rail
/// rule called this Live within ~300 ms.
#[test]
fn host_status_calls_one_live_die_beside_a_frozen_one_frozen() {
let hs = host_status_scripted(one_live_die_one_held);
let line = assert_sampling(&hs.run, "scripted one live die", &scripted_group(one_live_die_one_held));
eprintln!("ioreport scripted: one live die, one held -> status={:?}; {line}", hs.status);
assert_eq!(hs.run.verdict, Some(CpuRail::Frozen), "one live die vouched for a held one");
assert_eq!(hs.status, IoReportStatus::CpuRailFrozen);
assert!(!admits(hs.status), "one live die beside a held one admitted the meter");
assert!(hs.run.intervals >= 5, "frozen on {} interval(s)", hs.run.intervals);
assert_handed_the_script(&hs.run, one_live_die_one_held);
assert_verdict_matches_record(&hs.run, "scripted one live die");
}
#[test]
fn energy_to_uj_scales_by_unit() {
assert_eq!(energy_to_uj(5, "mJ"), 5_000);
assert_eq!(energy_to_uj(5_000, "nJ"), 5);
}
/// `power_over` yields watts only through a live CPU rail, and a
/// plausible figure when it does; a declined IOReport yields `None`.
#[test]
fn power_over_declines_unless_the_cpu_rail_is_live() {
let w = power_over(Duration::from_millis(300));
if bracket_admitted() {
assert!(
matches!(w, Some(w) if w > 0.0 && w < 500.0),
"a live CPU rail gave implausible SoC watts: {w:?}"
);
} else {
assert_eq!(
w,
None,
"power_over produced watts although IOReport is declined ({:?})",
ioreport_status()
);
}
}
/// Every gated IOReport entry point takes its status from the cache.
/// Once this thread has the status, the IOReport rung's gate costs it
/// under 10 ms of its own CPU (tests' clock), and on a declined host so
/// do `power_over` and the bracket, which must give no figure. Re-running the
/// probe by any route costs at least six reads of the Energy Model
/// group and five 25 ms ticks of this thread's CPU, and on a frozen
/// rail about a second of it. Thread CPU, not wall time, so host load and
/// other IOReport readers cannot fail it. On a live host `power_over`
/// and the bracket do real sampling work, so there only the gate is
/// timed and the status must be `Live`.
#[test]
fn the_gated_entry_points_read_the_cached_status() {
const WARM: Duration = Duration::from_millis(10);
let status = ioreport_status();
let c0 = test_thread_cpu();
let selected = bracket_admitted();
let selection = test_thread_cpu() - c0;
eprintln!("ioreport warm: status={status:?} selected={selected} selection burned {selection:?}");
assert!(selection < WARM, "bracket_admitted() burned {selection:?} of this thread's CPU after the status was cached");
assert_eq!(selected, admits(status), "the selection is not admits(the cached status)");
if selected {
assert_eq!(status, IoReportStatus::Live, "selected on {status:?}");
return;
}
let c0 = test_thread_cpu();
let watts = power_over(Duration::from_millis(200));
let power = test_thread_cpu() - c0;
let c0 = test_thread_cpu();
let figure = bracket_marginal_uj(cpu_work);
let metering = test_thread_cpu() - c0;
eprintln!("ioreport warm: declined power_over={watts:?} burned {power:?}; bracket={figure:?} burned {metering:?}");
assert_eq!((watts, figure), (None, None), "a declined meter ({status:?}) gave a figure");
assert!(power < WARM, "a declined power_over burned {power:?} of this thread's CPU: it probed again");
assert!(metering < WARM, "a declined bracket burned {metering:?} of this thread's CPU: it probed again");
}
/// When the meter IS selected, a multi-core CPU burst meters as non-zero
/// energy through a rail that moves; when it is not, the status says why
/// and no figure comes out. No branch goes unasserted.
#[test]
fn a_selected_ioreport_meter_meters_cpu_work() {
if bracket_admitted() {
assert_eq!(ioreport_status(), IoReportStatus::Live, "selected on a CPU rail that is not live");
assert_eq!(
host_status().status,
IoReportStatus::Live,
"the CPU rail stopped advancing after the meter was selected"
);
let uj = bracket_marginal_uj(cpu_work);
assert!(
matches!(uj, Some(u) if u > 0),
"a live CPU rail metered a multi-core CPU burst as {uj:?}"
);
} else {
assert_ne!(
ioreport_status(),
IoReportStatus::Live,
"status Live but the IOReport meter was not selected"
);
assert_eq!(bracket_marginal_uj(cpu_work), None, "a declined IOReport meter produced a figure");
}
}
}
}
// ---------------------------------------------------------------------------
// Linux — RAPL package energy counters. Power = energy delta over a timed
// window, used only while every package counter is live (`rapl_status`).
// `energy_uj` is root-only on mainline kernels since late 2020 (commit
// 949dd0104c49, "powercap: restrict energy meter to root access", in 5.10 and
// backported to stable: the CVE-2020-8694 / PLATYPUS mitigation). An
// unprivileged process therefore usually reads no zone, and the meter declines
// unless it runs as root or an administrator has granted read access.
// ---------------------------------------------------------------------------
#[cfg(target_os = "linux")]
fn idle_power() -> Option<f64> {
rapl_power_over(std::time::Duration::from_millis(700))
}
#[cfg(target_os = "linux")]
fn busy_power() -> Option<f64> {
rapl_power_over(std::time::Duration::from_millis(1400))
}
/// Average package power over `d`: the RAPL package zones' energy deltas,
/// each modulo its own wrap ([`rapl::delta_uj`]), ÷ the window. A hardware
/// reading — the bracket built on it is graded `HwShunt` — and so produced
/// only while every package counter is live ([`rapl_status`] is `Live`).
/// `None` if RAPL is declined, no package zone is readable, or the delta is
/// unknown. Other top-level zones (`psys`) are not summed: `psys` covers the
/// package too, and its liveness is not probed.
#[cfg(target_os = "linux")]
fn rapl_power_over(d: std::time::Duration) -> Option<f64> {
// Declined before reading: a counter that is readable but not live must
// not produce a hardware-graded figure.
if !admits_counter(rapl_status()) {
return None;
}
let z0 = rapl::packages(&rapl_zones());
if z0.is_empty() {
return None;
}
let t = Instant::now();
std::thread::sleep(d);
let z1 = rapl::packages(&rapl_zones());
let dt = t.elapsed().as_secs_f64();
Some(rapl::delta_uj(&z0, &z1)? as f64 / 1e6 / dt)
}
/// Every readable top-level RAPL zone (`intel-rapl:0`..`7`): whether it is a
/// package (its `name`), its `energy_uj` counter, and the
/// `max_energy_range_uj` it wraps at, when that is readable.
#[cfg(target_os = "linux")]
fn rapl_zones() -> Vec<rapl::Zone> {
let text = |path: String| std::fs::read_to_string(path).ok();
let read = |path: String| text(path)?.trim().parse::<u64>().ok();
(0..8u32)
.filter_map(|index| {
let base = format!("/sys/class/powercap/intel-rapl:{index}");
Some(rapl::Zone {
index,
package: text(format!("{base}/name")).is_some_and(|n| rapl::is_package_name(&n)),
energy_uj: read(format!("{base}/energy_uj"))?,
max_energy_range_uj: read(format!("{base}/max_energy_range_uj")),
})
})
.collect()
}
#[cfg(not(any(target_os = "macos", target_os = "linux")))]
fn idle_power() -> Option<f64> {
None
}
#[cfg(not(any(target_os = "macos", target_os = "linux")))]
fn busy_power() -> Option<f64> {
None
}
#[cfg(test)]
mod tests {
use super::*;
use liveness::tests::read_of;
use liveness::{probe_with, CpuRail, ProbeRun};
use std::cell::Cell;
use std::time::Duration;
/// How many times [`eighty_ms_of_work`] has run in this process.
static WORK_CALLS: AtomicU64 = AtomicU64::new(0);
/// Real CPU work: ~80 ms of it on the calling thread, per call.
fn eighty_ms_of_work() {
WORK_CALLS.fetch_add(1, Ordering::SeqCst);
let t = Instant::now();
let mut acc: u64 = 0;
while t.elapsed() < Duration::from_millis(80) {
acc = std::hint::black_box(acc.wrapping_mul(31).wrapping_add(1));
}
}
/// A busy window with no reading gives no figure AND stops its workers:
/// they must not keep burning every core for the rest of the process.
#[test]
fn a_busy_window_with_no_reading_stops_its_workers() {
static CALLS: AtomicU64 = AtomicU64::new(0);
fn counted_work() {
CALLS.fetch_add(1, Ordering::SeqCst);
std::hint::black_box((0..4000u64).fold(0u64, |a, b| a.wrapping_mul(31).wrapping_add(b)));
}
// The busy reading waits until the workers have really run (not a
// fixed sleep, which a loaded host could outlast), then gives none.
let r = bracket_with(counted_work, || Some(1.0), || {
let t = Instant::now();
while CALLS.load(Ordering::SeqCst) == 0 && t.elapsed() < Duration::from_secs(10) {
std::thread::sleep(Duration::from_millis(1));
}
None
});
assert_eq!(r, None);
let after = CALLS.load(Ordering::SeqCst);
assert!(after > 0, "the busy window never ran the work in 10 s");
std::thread::sleep(Duration::from_millis(60));
assert_eq!(CALLS.load(Ordering::SeqCst), after, "workers kept running after the bracket returned");
}
/// A run as the probe loop leaves it, for the pure mapping.
fn run_of(opened: bool, verdict: Option<CpuRail>) -> ProbeRun {
ProbeRun { opened, verdict, ..ProbeRun::default() }
}
/// No verdict — the probe's Energy Model subscription could not be opened,
/// or opened but never yielded two consecutive samples — is `Unavailable`:
/// never `Live`, never `CpuRailAbsent`. A run whose subscription did not
/// open is `Unavailable` whatever verdict it carries.
#[test]
fn no_verdict_is_unavailable() {
assert_eq!(status_of(&run_of(true, None)), IoReportStatus::Unavailable);
for verdict in [None, Some(CpuRail::Live), Some(CpuRail::Frozen), Some(CpuRail::Absent)] {
assert_eq!(
status_of(&run_of(false, verdict)),
IoReportStatus::Unavailable,
"a subscription that did not open, verdict {verdict:?}"
);
}
}
/// An opened run's verdict maps one-to-one onto the public status.
#[test]
fn the_verdict_maps_onto_the_status_one_to_one() {
assert_eq!(status_of(&run_of(true, Some(CpuRail::Absent))), IoReportStatus::CpuRailAbsent);
assert_eq!(status_of(&run_of(true, Some(CpuRail::Frozen))), IoReportStatus::CpuRailFrozen);
assert_eq!(status_of(&run_of(true, Some(CpuRail::Live))), IoReportStatus::Live);
}
/// Only a live CPU rail admits the IOReport meter.
#[test]
fn only_live_is_admitted() {
assert!(admits(IoReportStatus::Live), "a live CPU rail must admit the meter");
assert!(!admits(IoReportStatus::CpuRailFrozen), "a frozen CPU rail admitted the meter");
assert!(!admits(IoReportStatus::CpuRailAbsent), "no CPU rail admitted the meter");
assert!(!admits(IoReportStatus::Unavailable), "no IOReport admitted the meter");
}
/// The metering path with the status injected: it meters only when
/// [`admits`] the status. Declined, it returns `None` without reading power
/// or running the work; admitted, it brackets the work between the two
/// readings.
#[test]
fn only_a_live_status_meters_and_a_declined_one_never_runs_work() {
static CALLS: AtomicU64 = AtomicU64::new(0);
fn counted_work() {
CALLS.fetch_add(1, Ordering::SeqCst);
std::hint::black_box((0..4000u64).fold(0u64, |a, b| a.wrapping_mul(31).wrapping_add(b)));
}
for status in [
IoReportStatus::Unavailable,
IoReportStatus::CpuRailAbsent,
IoReportStatus::CpuRailFrozen,
IoReportStatus::Live,
] {
let before = CALLS.load(Ordering::SeqCst);
let readings = Cell::new(0u32);
let r = measure_gated(
admits(status),
counted_work,
|| {
readings.set(readings.get() + 1);
Some(1.0)
},
|| {
readings.set(readings.get() + 1);
// A worker counts a run when its call RETURNS, so wait until
// more calls have started than there can be workers (at most
// 12): one worker has then started a second call, so its
// first has been counted.
let t = Instant::now();
while CALLS.load(Ordering::SeqCst) - before <= 12 && t.elapsed() < Duration::from_secs(10) {
std::thread::sleep(Duration::from_millis(1));
}
Some(3.0)
},
);
let ran = CALLS.load(Ordering::SeqCst) - before;
if status == IoReportStatus::Live {
assert!(r.is_some() && ran > 0 && readings.get() == 2, "{status:?}: {r:?}, {ran} runs, {} readings", readings.get());
} else {
assert_eq!((r, ran, readings.get()), (None, 0, 0), "{status:?} metered");
}
}
}
/// Read `n` of a scripted run: its channels, or `None` for a failed read.
type Script = fn(u32) -> Option<Vec<(&'static str, i64)>>;
/// The production composition — `status_of(&probe_with(open, read, tick,
/// wall, cpu, budget))`, as `host_status` composes it — on fakes: an opener
/// that opens or not, scripted reads, a tick that advances a scripted wall
/// clock and CPU clock by 25 ms and counts itself. Returns the status, the
/// run, and how many reads and ticks the fakes saw.
fn composed(opens: bool, script: Script) -> (IoReportStatus, ProbeRun, u32, u32) {
let (clock, cpu) = (Cell::new(Duration::ZERO), Cell::new(Duration::ZERO));
let (reads, ticks) = (Cell::new(0u32), Cell::new(0u32));
let run = probe_with(
|| opens.then_some(()),
|_: &()| {
let n = reads.get();
reads.set(n + 1);
script(n).map(|s| read_of(&s))
},
|| {
ticks.set(ticks.get() + 1);
clock.set(clock.get() + Duration::from_millis(25));
cpu.set(cpu.get() + Duration::from_millis(25));
crate::meter::probe_clock::TickOutcome {
cpu: Some(Duration::from_millis(25)),
spins: 1,
}
},
|| clock.get(),
|| Some(cpu.get()),
Duration::from_millis(1000),
);
(status_of(&run), run, reads.get(), ticks.get())
}
/// The composition on an Energy Model group that cannot be opened: no
/// read, no tick, no verdict, `Unavailable`.
#[test]
fn the_composition_on_an_unopenable_group_reads_nothing_and_is_unavailable() {
let (status, run, reads, ticks) = composed(false, |_| Some(vec![("CPU Energy", 1)]));
assert_eq!((reads, ticks), (0, 0), "an unopened subscription was read or ticked");
assert!(!run.opened && run.calls.is_empty() && run.ticks == 0 && run.verdict.is_none(), "{run:?}");
assert_eq!(status, IoReportStatus::Unavailable);
assert!(!admits(status));
}
/// The composition on a single batch jump: Frozen, declined.
#[test]
fn the_composition_on_a_single_batch_jump_is_frozen() {
let (status, run, reads, ticks) = composed(true, |n| {
Some(vec![("CPU Energy", if n < 4 { 480_083_160 } else { 541_893_638 }), ("GPU Energy", i64::from(n))])
});
assert_eq!(run.verdict, Some(CpuRail::Frozen), "{run:?}");
assert_eq!(status, IoReportStatus::CpuRailFrozen);
assert!(!admits(status));
assert_eq!((reads, ticks, run.intervals), (41, 40, 40));
}
/// The composition on a rail that advances at every read: Live once it
/// has advanced in five intervals over 300 ms — at 25 ms an interval, the
/// twelfth — and admitted.
#[test]
fn the_composition_on_a_sustained_advance_is_live() {
let (status, run, reads, ticks) = composed(true, |n| Some(vec![("CPU Energy", 1_000 + i64::from(n))]));
assert_eq!(run.verdict, Some(CpuRail::Live), "{run:?}");
assert_eq!(status, IoReportStatus::Live);
assert!(admits(status));
assert_eq!((reads, ticks, run.intervals), (13, 12, 12));
}
/// The composition on the observed two-step refresh: Frozen, declined,
/// after the whole budget.
#[test]
fn the_composition_on_a_two_step_refresh_is_frozen() {
let (status, run, reads, ticks) = composed(true, |n| {
let step = match n {
0..=2 => 0,
3 => 61_764_485,
_ => 61_764_485 + 3_435,
};
Some(vec![("CPU Energy", 2_643_843_378 + step), ("GPU Energy", i64::from(n))])
});
assert_eq!(run.verdict, Some(CpuRail::Frozen), "{run:?}");
assert_eq!(status, IoReportStatus::CpuRailFrozen);
assert!(!admits(status));
assert_eq!((reads, ticks, run.intervals), (41, 40, 40));
}
/// The composition on two dies advancing in alternate intervals: Frozen —
/// liveness is per rail, and nothing between the read and the rule sums
/// them (their sum advances in every interval).
#[test]
fn the_composition_on_two_dies_advancing_in_alternate_intervals_is_frozen() {
let (status, run, _, _) = composed(true, |n| {
Some(vec![
("DIE_0_CPU Energy", 100 + 50 * i64::from(n.div_ceil(2))),
("DIE_1_CPU Energy", 200 + 50 * i64::from(n / 2)),
])
});
assert_eq!(run.verdict, Some(CpuRail::Frozen), "{run:?}");
assert_eq!(status, IoReportStatus::CpuRailFrozen);
}
// ── The process counter: the composition, the measurement, the order ──
/// Read `n` of a scripted process counter (cumulative nJ), or `None`.
type CounterScript = fn(u32) -> Option<u64>;
/// [`process_probe_with`] — the composition `process_host_status` runs —
/// on a scripted counter, with a tick that advances a scripted wall clock
/// and CPU clock by 25 ms and counts itself. Returns the status, the run,
/// and how many times the counter was read and the probe ticked.
fn process_composed(script: CounterScript) -> (CounterStatus, ProbeRun, u32, u32) {
let (clock, cpu) = (Cell::new(Duration::ZERO), Cell::new(Duration::ZERO));
let (reads, ticks) = (Cell::new(0u32), Cell::new(0u32));
let run = process_probe_with(
|| {
let n = reads.get();
reads.set(n + 1);
script(n)
},
|| {
ticks.set(ticks.get() + 1);
clock.set(clock.get() + Duration::from_millis(25));
cpu.set(cpu.get() + Duration::from_millis(25));
crate::meter::probe_clock::TickOutcome {
cpu: Some(Duration::from_millis(25)),
spins: 1,
}
},
|| clock.get(),
|| Some(cpu.get()),
PROCESS_LIVENESS_BUDGET,
);
(counter_status_of(&run), run, reads.get(), ticks.get())
}
/// A process counter that advances at every read is Live at the twelfth
/// 25 ms interval (five advances, 80%, 300 ms) and admitted, and the rule
/// was handed each value as read. The open is one read.
#[test]
fn process_a_counter_advancing_at_every_read_is_live() {
let (status, run, reads, ticks) = process_composed(|n| Some(48_564_621 + 43_868_912 * u64::from(n)));
assert_eq!(status, CounterStatus::Live, "{run:?}");
assert!(admits_counter(status));
assert_eq!((reads, ticks, run.intervals), (1 + 13, 12, 12));
let handed: Vec<i64> = run.seen.iter().map(|s| s.cpu[0].1).collect();
assert_eq!(handed, (1..=13).map(|n| 48_564_621 + 43_868_912 * n).collect::<Vec<i64>>());
}
/// A counter that holds, steps once, or steps twice close together is
/// Frozen after the whole budget, and never admitted.
#[test]
fn process_a_constant_or_stepping_counter_is_frozen() {
for (what, script) in [
("constant", (|_| Some(48_564_621)) as CounterScript),
("one step", |n| Some(if n < 6 { 48_564_621 } else { 920_000_000 })),
("two steps", |n| Some(48_564_621 + if n >= 6 { 61_764_485 } else { 0 } + if n >= 8 { 3_435 } else { 0 })),
("four steps then a hold", |n| Some(48_564_621 + 1_000 * u64::from(n.min(4)))),
] {
let (status, run, reads, ticks) = process_composed(script);
assert_eq!(status, CounterStatus::Frozen, "{what}: {run:?}");
assert!(!admits_counter(status), "{what}");
assert_eq!((reads, ticks, run.intervals), (1 + 41, 40, 40), "{what}");
}
}
/// A counter that cannot be read is `Unavailable` with no probe read and
/// no tick; one that reads at the open and never after gives no verdict,
/// `Unavailable`, after the five-budget cap.
#[test]
fn process_an_unreadable_counter_is_unavailable() {
let (status, run, reads, ticks) = process_composed(|_| None);
assert_eq!(status, CounterStatus::Unavailable);
assert!(!run.opened && run.calls.is_empty(), "{run:?}");
assert_eq!((reads, ticks), (1, 0));
let (status, run, _, ticks) = process_composed(|n| (n == 0).then_some(1));
assert_eq!(status, CounterStatus::Unavailable, "{run:?}");
assert!(run.opened && run.verdict.is_none());
assert_eq!(ticks, 200, "five budgets of 25 ms ticks");
}
/// The measurement on a scripted counter that each call of `work` raises
/// by exactly 7,000 nJ: the figure is exactly 7 uJ a call, which holds
/// only if the counter is read before the first call and after the last
/// one is counted. Reading it while workers still run, or counting calls
/// before they are joined, misses calls on one side and not the other.
#[test]
fn process_the_measurement_is_the_counters_advance_over_every_call() {
static NJ: AtomicU64 = AtomicU64::new(1_000_000);
fn seven_uj() {
std::hint::black_box((0..2000u64).fold(0u64, |a, b| a.wrapping_mul(31).wrapping_add(b)));
NJ.fetch_add(7_000, Ordering::SeqCst);
}
let reads = std::sync::Mutex::new(Vec::new());
let r = process_marginal_with(
seven_uj,
|| {
let v = NJ.load(Ordering::SeqCst);
reads.lock().unwrap().push(v);
Some(v)
},
Duration::from_millis(200),
);
let reads = reads.into_inner().unwrap();
assert_eq!(reads.len(), 2, "one read before, one after: {reads:?}");
assert_eq!(r, Some(7), "7,000 nJ a call is 7 uJ: reads {reads:?}");
// Rounded to the nearest microjoule: 7,499 nJ is 7, 7,500 is 8.
static NJ2: AtomicU64 = AtomicU64::new(0);
fn just_under() {
NJ2.fetch_add(7_499, Ordering::SeqCst);
}
assert_eq!(process_marginal_with(just_under, || Some(NJ2.load(Ordering::SeqCst)), Duration::from_millis(50)), Some(7));
static NJ3: AtomicU64 = AtomicU64::new(0);
fn half_up() {
NJ3.fetch_add(7_500, Ordering::SeqCst);
}
assert_eq!(process_marginal_with(half_up, || Some(NJ3.load(Ordering::SeqCst)), Duration::from_millis(50)), Some(8));
}
/// A host on which the measuring thread's own energy count does not read
/// has no process counter: `Unavailable`, which never meters, and the
/// liveness probe is not run. Where the count reads, the status is the
/// probe's.
#[test]
fn process_no_counter_where_the_threads_own_count_does_not_read() {
let status = process_status_where(false, || panic!("probed a counter no window could be read on"));
assert_eq!(status, CounterStatus::Unavailable);
assert!(!admits_counter(status));
for probed in [CounterStatus::Live, CounterStatus::Frozen, CounterStatus::Unavailable] {
assert_eq!(process_status_where(true, || probed), probed);
}
}
/// The production window read on a thread never seen to leave its core
/// gives the measurement no figure, and the work never runs: the same
/// `None` an unreadable counter gives, never a window read from a count
/// that may lack what the thread spent.
#[test]
fn process_a_window_whose_thread_is_not_seen_to_leave_its_core_gives_no_figure() {
static CALLS: AtomicU64 = AtomicU64::new(0);
fn counted() {
CALLS.fetch_add(1, Ordering::SeqCst);
}
let read = Cell::new(0u32);
// The thread's count reads, and never moves; the counter would read.
let never_seen = || {
own_read::window_read(
|| Some(7),
|_| {},
|| {
read.set(read.get() + 1);
Some(1_000_000)
},
)
};
assert_eq!(process_marginal_with(counted, never_seen, Duration::from_millis(50)), None);
assert_eq!((CALLS.load(Ordering::SeqCst), read.get()), (0, 0), "the work ran, or the counter was read");
}
/// No figure — never an invented one — from a counter that cannot be read
/// (and then the work never runs) or that went down.
#[test]
fn process_an_unreadable_or_falling_counter_gives_no_figure() {
static CALLS: AtomicU64 = AtomicU64::new(0);
fn counted() {
CALLS.fetch_add(1, Ordering::SeqCst);
}
assert_eq!(process_marginal_with(counted, || None, Duration::from_millis(50)), None);
assert_eq!(CALLS.load(Ordering::SeqCst), 0, "the work ran although the counter could not be read");
let n = Cell::new(0u32);
let falling = || {
n.set(n.get() + 1);
Some(if n.get() == 1 { 9_000_000 } else { 1_000 })
};
assert_eq!(process_marginal_with(counted, falling, Duration::from_millis(50)), None, "a counter that went down");
let m = Cell::new(0u32);
let fails_after = || {
m.set(m.get() + 1);
(m.get() == 1).then_some(5)
};
assert_eq!(process_marginal_with(counted, fails_after, Duration::from_millis(50)), None, "the second read failed");
}
/// The macOS order: the process counter if it is live — and then
/// IOReport is never asked — else IOReport if its CPU rail is live, else
/// nothing. A process counter that is not live never wins.
#[test]
fn the_process_counter_ranks_first_and_only_when_live() {
let never = || -> IoReportStatus { panic!("IOReport was asked although the process counter is live") };
assert_eq!(macos_source(CounterStatus::Live, never), Some(MeterSource::ProcessEnergy));
for process in [CounterStatus::Unavailable, CounterStatus::Absent, CounterStatus::Frozen] {
assert_eq!(macos_source(process, || IoReportStatus::Live), Some(MeterSource::IoReport), "{process:?}");
for ioreport in [IoReportStatus::Unavailable, IoReportStatus::CpuRailAbsent, IoReportStatus::CpuRailFrozen] {
assert_eq!(macos_source(process, || ioreport), None, "{process:?} / {ioreport:?}");
}
}
}
/// Every source grades its figure `HwShunt`, with a measured figure, so
/// the schema's coupling holds; each names its own domain, and the
/// process counter's says what it leaves out.
#[test]
fn every_source_grades_hwshunt_and_names_its_domain() {
for source in [MeterSource::ProcessEnergy, MeterSource::IoReport, MeterSource::Rapl] {
let e = Measurement { uj: 42, source }.energy();
assert_eq!((e.measured_uj, e.provenance), (Some(42), proof_schema::Provenance::HwShunt), "{source:?}");
assert!(e.check_provenance().is_ok(), "{source:?}");
}
let d = MeterSource::ProcessEnergy.domain();
for left_out in ["other processes", "child processes", "GPU", "ANE", "DRAM"] {
assert!(d.contains(left_out), "the process counter's domain does not say it leaves out {left_out}: {d}");
}
}
/// The process counter is a macOS interface: off macOS its status says
/// so, no probe runs, and the hook is a no-op.
#[cfg(not(target_os = "macos"))]
#[test]
fn process_energy_is_unavailable_off_macos() {
simulate_frozen_process_energy();
assert_eq!(process_energy_status(), CounterStatus::Unavailable);
assert_eq!(real_process_probe_runs(), 0);
assert_ne!(meter_source(), Some(MeterSource::ProcessEnergy));
}
// ── RAPL: the production composition on scripted zones ────────────────
/// Read `n` of a scripted RAPL run: every top-level zone it lists, as
/// (index, is a package, `energy_uj`).
type ZoneScript = fn(u32) -> Vec<(u32, bool, u64)>;
/// [`rapl_probe_with`] — the composition `rapl_host_status` runs — on
/// scripted zones, with a tick that advances a scripted wall clock and CPU
/// clock by 25 ms and counts itself. Returns the status, the run, and how
/// many times the zones were read and the probe ticked.
fn rapl_composed(script: ZoneScript) -> (CounterStatus, ProbeRun, u32, u32) {
let (clock, cpu) = (Cell::new(Duration::ZERO), Cell::new(Duration::ZERO));
let (reads, ticks) = (Cell::new(0u32), Cell::new(0u32));
let run = rapl_probe_with(
|| {
let n = reads.get();
reads.set(n + 1);
script(n)
.into_iter()
.map(|(index, package, energy_uj)| rapl::Zone {
index,
package,
energy_uj,
max_energy_range_uj: Some(262_143_328_850),
})
.collect()
},
|| {
ticks.set(ticks.get() + 1);
clock.set(clock.get() + Duration::from_millis(25));
cpu.set(cpu.get() + Duration::from_millis(25));
crate::meter::probe_clock::TickOutcome {
cpu: Some(Duration::from_millis(25)),
spins: 1,
}
},
|| clock.get(),
|| Some(cpu.get()),
RAPL_LIVENESS_BUDGET,
);
(counter_status_of(&run), run, reads.get(), ticks.get())
}
/// A package counter that advances at every read is Live once it has
/// advanced in five intervals over 300 ms — at 25 ms an interval, the
/// twelfth — and admitted. The open is one read of the zones; the probe
/// then reads 13 times.
#[test]
fn rapl_a_package_counter_advancing_at_every_read_is_live() {
let (status, run, reads, ticks) = rapl_composed(|n| vec![(0, true, 5_000 + 900 * u64::from(n))]);
assert_eq!(run.verdict, Some(CpuRail::Live), "{run:?}");
assert_eq!(status, CounterStatus::Live);
assert!(admits_counter(status));
assert_eq!((reads, ticks, run.intervals), (1 + 13, 12, 12));
let handed: Vec<i64> = run.seen.iter().map(|s| s.cpu[0].1).collect();
assert_eq!(handed, (1..=13).map(|n| 5_000 + 900 * n).collect::<Vec<i64>>(), "the counter as read");
}
/// A package counter that holds, or steps once and holds, is Frozen after
/// the whole budget and declined: the rule RAPL had none of before.
#[test]
fn rapl_a_constant_or_once_stepping_package_counter_is_frozen() {
for (what, script) in [
("constant", (|_| vec![(0, true, 42_000_000)]) as ZoneScript),
("one step", |n| vec![(0, true, if n < 6 { 42_000_000 } else { 43_344_869 })]),
("two steps", |n| vec![(0, true, 42_000_000 + if n >= 6 { 1_000_000 } else { 0 } + if n >= 8 { 3_435 } else { 0 })]),
] {
let (status, run, reads, ticks) = rapl_composed(script);
assert_eq!(status, CounterStatus::Frozen, "{what}: {run:?}");
assert!(!admits_counter(status), "{what}");
assert_eq!((reads, ticks, run.intervals), (1 + 41, 40, 40), "{what}");
}
}
/// The figure sums every package counter, so EVERY one must keep
/// advancing: with one advancing and the other holding, RAPL is Frozen,
/// and with both advancing it is Live. Two packages advancing in alternate
/// intervals (each in half of them) are not.
#[test]
fn rapl_every_package_counter_must_keep_advancing() {
let one_holds: ZoneScript = |n| vec![(0, true, 1_000 + u64::from(n)), (1, true, 77)];
let (status, run, _, _) = rapl_composed(one_holds);
assert_eq!(status, CounterStatus::Frozen, "one package counter held: {run:?}");
assert_eq!(run.intervals, 40, "watched for the whole budget: {run:?}");
let both: ZoneScript = |n| vec![(0, true, 1_000 + u64::from(n)), (1, true, 9_000 + 2 * u64::from(n))];
let (status, run, reads, _) = rapl_composed(both);
assert_eq!(status, CounterStatus::Live, "{run:?}");
assert_eq!(reads, 1 + 13);
let alternate: ZoneScript =
|n| vec![(0, true, 100 + 50 * u64::from(n.div_ceil(2))), (1, true, 200 + 50 * u64::from(n / 2))];
let (status, run, _, _) = rapl_composed(alternate);
assert_eq!(status, CounterStatus::Frozen, "alternate advances added up to live: {run:?}");
}
/// `psys` is neither summed nor probed: a moving `psys` beside a holding
/// package is Frozen, and zones with no package at all are Absent.
#[test]
fn rapl_psys_is_not_a_package_counter() {
let (status, run, _, _) = rapl_composed(|n| vec![(0, true, 5_000), (1, false, 8_000 + 1_000 * u64::from(n))]);
assert_eq!(status, CounterStatus::Frozen, "a moving psys made the package live: {run:?}");
let (status, run, _, _) = rapl_composed(|n| vec![(1, false, 8_000 + 1_000 * u64::from(n))]);
assert_eq!(status, CounterStatus::Absent, "{run:?}");
assert!(!admits_counter(status));
}
/// A package counter that wraps (goes down) did not advance in that
/// interval: one that wraps and then holds is Frozen, whatever came
/// before the wrap within the first 300 ms.
#[test]
fn rapl_a_wrap_is_not_an_advance() {
let (status, run, _, _) =
rapl_composed(|n| vec![(0, true, if n < 5 { 262_143_328_000 + 100 * u64::from(n) } else { 10 })]);
assert_eq!(status, CounterStatus::Frozen, "{run:?}");
}
/// No readable zone — `energy_uj` is root-only on current kernels — is
/// `Unavailable`: the open reads the zones once, and the probe never
/// reads or ticks. Zones that are readable at the open and never after
/// give no verdict: `Unavailable`, after the five-budget cap.
#[test]
fn rapl_unreadable_zones_are_unavailable() {
let (status, run, reads, ticks) = rapl_composed(|_| Vec::new());
assert_eq!(status, CounterStatus::Unavailable);
assert!(!run.opened && run.calls.is_empty(), "{run:?}");
assert_eq!((reads, ticks), (1, 0));
assert!(!admits_counter(status));
let (status, run, _, ticks) = rapl_composed(|n| if n == 0 { vec![(0, true, 1)] } else { Vec::new() });
assert_eq!(status, CounterStatus::Unavailable, "{run:?}");
assert!(run.opened && run.verdict.is_none());
assert_eq!(ticks, 200, "five budgets of 25 ms ticks");
}
/// The counter status maps one to one from an opened run's verdict; a
/// run that did not open, or gave no verdict, is `Unavailable`; and only
/// `Live` admits the source, so a declined RAPL never runs the work.
#[test]
fn rapl_only_a_live_status_meters() {
assert_eq!(counter_status_of(&run_of(true, Some(CpuRail::Live))), CounterStatus::Live);
assert_eq!(counter_status_of(&run_of(true, Some(CpuRail::Frozen))), CounterStatus::Frozen);
assert_eq!(counter_status_of(&run_of(true, Some(CpuRail::Absent))), CounterStatus::Absent);
assert_eq!(counter_status_of(&run_of(true, None)), CounterStatus::Unavailable);
for verdict in [None, Some(CpuRail::Live), Some(CpuRail::Frozen), Some(CpuRail::Absent)] {
assert_eq!(counter_status_of(&run_of(false, verdict)), CounterStatus::Unavailable, "{verdict:?}");
}
static CALLS: AtomicU64 = AtomicU64::new(0);
fn counted_work() {
CALLS.fetch_add(1, Ordering::SeqCst);
}
for status in [CounterStatus::Unavailable, CounterStatus::Absent, CounterStatus::Frozen] {
assert!(!admits_counter(status), "{status:?} admitted");
let readings = Cell::new(0u32);
let r = measure_gated(
admits_counter(status),
counted_work,
|| {
readings.set(readings.get() + 1);
Some(1.0)
},
|| {
readings.set(readings.get() + 1);
Some(3.0)
},
);
assert_eq!((r, CALLS.load(Ordering::SeqCst), readings.get()), (None, 0, 0), "{status:?} metered");
}
assert!(admits_counter(CounterStatus::Live));
}
/// RAPL is a Linux interface: off Linux its status says so, and no RAPL
/// probe ever runs.
#[cfg(not(target_os = "linux"))]
#[test]
fn rapl_status_is_unavailable_off_linux() {
assert_eq!(rapl_status(), CounterStatus::Unavailable);
assert_eq!(real_rapl_probe_runs(), 0);
}
/// The declined path meters real work honestly. This meter's fallback is no
/// figure at all: on a host whose interface is declined, real CPU work (a
/// direct call burns 80 ms) put through `measure_marginal_uj` gives `None`
/// — not `Some(0)`, not a floored figure — before any bracket runs, so the
/// work is never called and the caller keeps its own estimate.
/// Where an interface is usable, the bracket really runs the work. How
/// long the call took is logged, not asserted: a loaded host can stall
/// any thread, and the work-call count is the pin.
#[test]
fn a_declined_meter_gives_real_work_no_figure_and_runs_no_bracket() {
let t = Instant::now();
eighty_ms_of_work();
let direct = t.elapsed();
assert!(direct >= Duration::from_millis(80), "the work did not run: {direct:?}");
let available = hardware_meter_available();
let before = WORK_CALLS.load(Ordering::SeqCst);
let t = Instant::now();
let r = measure_marginal_uj(eighty_ms_of_work);
let took = t.elapsed();
let runs = WORK_CALLS.load(Ordering::SeqCst) - before;
eprintln!(
"declined path: available={available} ioreport_status={:?} figure={r:?} work_runs={runs} \
took={took:?}",
ioreport_status()
);
if available {
assert!(runs > 0, "a usable meter answered {r:?} without running the work");
assert!(r.is_some(), "a usable hardware interface returned no reading");
} else {
assert_eq!(r, None, "a declined meter gave real work a figure");
assert_eq!(runs, 0, "a declined meter ran a bracket ({runs} calls of the work)");
}
}
/// A figure exactly when a usable interface exists: `None` — never a number —
/// where none is (including a declined IOReport), and a plausible per-call
/// figure where one is. Replaces a version that asserted only inside
/// `if let Some`, so a `None` passed silently on every host.
#[test]
fn meter_is_total_or_nothing() {
let r = measure_marginal_uj(|| {
std::hint::black_box((0..4000u64).fold(0u64, |a, b| a.wrapping_mul(31).wrapping_add(b)));
});
if hardware_meter_available() {
let uj = r.expect("a usable hardware interface returned no reading");
assert!(uj < 100_000_000, "implausible per-call energy {uj} uj");
} else {
assert_eq!(r, None, "no usable hardware interface ({:?}), yet a figure came out", ioreport_status());
}
}
/// IOReport is an Apple Silicon interface: off macOS the status says so.
#[cfg(not(target_os = "macos"))]
#[test]
fn ioreport_status_is_unavailable_off_macos() {
assert_eq!(ioreport_status(), IoReportStatus::Unavailable);
assert_eq!(real_probe_runs(), 0);
#[cfg(not(target_os = "linux"))]
assert!(!hardware_meter_available(), "no interface is implemented for this OS");
}
}