open-standards/sandbox/sandbox-rs/crates/sandbox-meter/tests/tree_energy.rs
Transaction Science 80e254933c
Some checks failed
energy meters on real hardware / RAPL and NVML on jetson-hub (pull_request) Successful in 40s
standards workspaces build + test / jouleclaw-rs (pull_request) Successful in 2m35s
standards workspaces build + test / joule-code-rs (pull_request) Successful in 1m30s
standards workspaces build + test / aac-rs (pull_request) Successful in 2m34s
standards workspaces build + test / joulecontract-rs (pull_request) Successful in 58s
standards workspaces build + test / map-rs (pull_request) Successful in 59s
standards workspaces build + test / mesh-rs (pull_request) Successful in 55s
standards workspaces build + test / openpay (pull_request) Successful in 3m15s
standards workspaces build + test / jouledesk-rs (pull_request) Successful in 55s
standards workspaces build + test / macOS energy backends (pull_request) Successful in 4m31s
no-vanity-metrics / Scan committed Markdown for vanity metrics (pull_request) Successful in 4s
no-vanity-metrics / WAI public text names no comparison (pull_request) Successful in 6s
no-vanity-metrics / WAI public text names no company or product (pull_request) Successful in 4s
standards workspaces build + test / arl-rs (pull_request) Successful in 1m19s
standards workspaces build + test / eoc-rs (pull_request) Successful in 2m6s
standards workspaces build + test / joulehook-rs (pull_request) Has been cancelled
standards workspaces build + test / proof-rs (pull_request) Has been cancelled
standards workspaces build + test / sandbox-rs (pull_request) Has been cancelled
standards workspaces build + test / sense-rs (pull_request) Has been cancelled
standards workspaces build + test / smart-byte-rs (pull_request) Has been cancelled
standards workspaces build + test / joule-ui-rs (pull_request) Has been cancelled
standards workspaces build + test / jcp-rs (pull_request) Has been cancelled
sandbox-meter: a member waiting on the kernel in the middle of its work is not settled
The tree meter took a member for settled when two reads of its counts
agreed with no thread on a core or queued between them. A thread in a wait
that no signal ends (the kernel's own waits: a block being written, a page
being brought in, a lock) passes that rule: it is off its core and folded.
But it runs again when the kernel is done, with nothing arriving, so the
member is in the middle of its work. A stop taken there read the span
short by the rest of that work, whole, with no gap.

Measured (M4 Max, macOS 27.0.1): a server whose reply is followed by
twenty rounds of a 4 KiB write, an fsync and a short burn read short in 24
of 60 round trips, at 16 to 92% of the oracle's advance, each reading
whole. With the rule below it read exact in 60 of 60, and in every one of
72 more under a load past 160.

The rule: once `pti_numrunning` is zero the meter asks each thread's own
run state (`PROC_PIDLISTTHREADS`, then `PROC_PIDTHREADINFO`), and a member
with a thread in `TH_STATE_UNINTERRUPTIBLE` is not settled; it is walked
again, and is `TreeGap::Unsettled` if it never comes to rest inside the
bound. A thread asleep, or blocked reading a pipe, is not in that state (0
of 20 looks each; a thread in an fsync loop, 18 of 60), and none of an
idle host's threads was at 5 looks, so a server waiting for its next
request settles as before. Cost: the start of a span on a tree of 61
processes took 1.73 to 1.75 ms against 1.68 to 1.72 ms.

JCP SPEC 5.1 states the requirement, and names the second cause beside
"still running" in the gap it already has.

Tests, two of which failed on a loaded host and are why this was found:

- `a_worker_reaped_while_its_tree_is_walked_is_never_missed` asserted no
  reading is whole. One round trip of 21 in a CI run read the worker alive
  and off every core, counted its record, and was whole: the worker was
  read, not missed. The test now asserts its name: a whole reading holds
  the worker as a member.
- `a_server_that_reaps_its_worker_is_not_whole` failed its equality with
  the oracle once at a load past 100 on a host short of memory. That is
  what a member read while it waited for a page looks like; the run's
  figures were not kept, so this is the likely cause and not a shown one.
- New: `a_reply_whose_tail_waits_on_the_disk_is_read_once_the_tail_is_done`,
  which fails against the earlier rule on an idle host.
- The oracle's own quiet rule asks the same of each thread, read raw.

Verified on the build Mac: sandbox-meter's tests idle and twelve full runs
under load, and the process-tree tests of jcp-gateway, proof-solvers,
sandbox-proof and joule-code-remediate.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-13 01:03:29 -04:00

969 lines
44 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.

//! The tree meter on real process trees. Each test spawns a server of its
//! own (a process group of its own, killed at the end), drives it with
//! request/response round trips over pipes, and meters each round trip with
//! a [`TreeMeter`] on the server's pid.
//!
//! The oracle shares no code with the meter: the tree's members from `ps`,
//! each member's `proc_pid_rusage(pid, RUSAGE_INFO_V6)` read into a raw
//! buffer and taken at the SDK's byte offsets (energy at 336, start time at
//! 80, exit time at 88, reaped-descendant CPU time at 96 and 104, billed and
//! serviced CPU time at 216 and 224, billed and serviced energy at 264 and
//! 272), read only
//! once the oracle has seen every member quiet on its own terms: no thread on
//! a core or queued (`proc_pidinfo(PROC_PIDTASKINFO)` into a raw buffer,
//! `pti_numrunning` at byte 88) and none in a wait no signal ends (each
//! thread's `pth_run_state`, read raw as well) at two reads 20 ms apart
//! whose energy agrees.
//! A server blocked waiting for its next request does not move its count, so
//! the meter's reading of a round trip must EQUAL the oracle's advance across
//! it, to the nanojoule: a settle that read a member still finishing its
//! reply would read less, and one that read a stale start would read more.
//! The tests log what they saw (`tree meter:`), and run one at a time (a
//! lock) so one test's server is not loading the host under another's.
#![cfg(all(target_os = "macos", target_arch = "aarch64"))]
use sandbox_meter::{SpanEnd, TreeGap, TreeMeter, TreeReading};
use std::io::{BufRead, BufReader, Write};
use std::os::unix::process::CommandExt;
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
use std::sync::{Mutex, MutexGuard};
use std::time::{Duration, Instant};
static ONE_AT_A_TIME: Mutex<()> = Mutex::new(());
fn alone() -> MutexGuard<'static, ()> {
ONE_AT_A_TIME.lock().unwrap_or_else(|e| e.into_inner())
}
/// How long the meter may wait for a tree to go quiet in these tests: far
/// past the production bound, because the host runs at a load of 14 to 260
/// and a reply's tail can wait that long for a core. A member still running
/// after it would fail the test as a gap, which is the point: a whole
/// reading is never a stale one.
const TEST_SETTLE: Duration = Duration::from_secs(2);
/// `pid`'s `rusage_info_v6`, raw: `proc_pid_rusage(pid, 6)` into a buffer
/// larger than the struct's 464 bytes, as 64-bit words.
fn oracle_rusage(pid: i32) -> Option<[u64; 64]> {
let mut buf = [0u64; 64];
// SAFETY: `buf` is writable and larger than `rusage_info_v6`.
let rc = unsafe { libc::proc_pid_rusage(pid, 6, buf.as_mut_ptr() as *mut libc::rusage_info_t) };
(rc == 0).then_some(buf)
}
fn energy_nj(ru: &[u64; 64]) -> u64 {
ru[336 / 8]
}
fn exit_abstime(ru: &[u64; 64]) -> u64 {
ru[88 / 8]
}
fn reaped_cpu_ticks(ru: &[u64; 64]) -> u64 {
ru[96 / 8] + ru[104 / 8]
}
/// `ri_billed_system_time`: the CPU time (mach ticks) of work other
/// processes did on this one's behalf, billed to it.
fn billed_ticks(ru: &[u64; 64]) -> u64 {
ru[216 / 8]
}
/// `ri_serviced_system_time`: the CPU time of work it did for others.
fn serviced_ticks(ru: &[u64; 64]) -> u64 {
ru[224 / 8]
}
/// `ri_billed_energy`, nJ: the energy of work billed to it.
fn billed_nj(ru: &[u64; 64]) -> u64 {
ru[264 / 8]
}
/// `ri_serviced_energy`, nJ: the energy of work it did for others.
fn serviced_nj(ru: &[u64; 64]) -> u64 {
ru[272 / 8]
}
/// `pti_numrunning` of `pid`: `proc_pidinfo(pid, PROC_PIDTASKINFO = 4)` into
/// a raw 96-byte buffer, the int32 at byte 88. `None` if the call fails (an
/// exited process has no task).
fn oracle_running(pid: i32) -> Option<i32> {
let mut buf = [0u8; 96];
// SAFETY: `buf` is writable for the 96 bytes passed.
let n = unsafe { libc::proc_pidinfo(pid, 4, 0, buf.as_mut_ptr().cast(), 96) };
(n == 96).then(|| i32::from_ne_bytes(buf[88..92].try_into().unwrap()))
}
/// Whether every thread of `pid` is at rest by its own run state: none on a
/// core or queued (`TH_STATE_RUNNING`, 1) and none in a wait that no signal
/// ends (`TH_STATE_UNINTERRUPTIBLE`, 4: waiting on the disk, the pager or a
/// kernel lock, to run again without anything arriving).
/// `proc_pidinfo(pid, PROC_PIDLISTTHREADS = 6)` into a raw buffer of 64-bit
/// thread handles, then `proc_pidinfo(pid, PROC_PIDTHREADINFO = 5, handle)`
/// into a raw 112-byte buffer, `pth_run_state` the int32 at byte 24. `None`
/// if the listing fails (an exited process has no task).
fn oracle_at_rest(pid: i32) -> Option<bool> {
let mut handles = [0u64; 512];
// SAFETY: `handles` is writable for the bytes passed.
let n = unsafe { libc::proc_pidinfo(pid, 6, 0, handles.as_mut_ptr().cast(), (handles.len() * 8) as libc::c_int) };
if n <= 0 {
return None;
}
let n = n as usize / 8;
assert!(n < handles.len(), "{pid} has more threads than the oracle's buffer holds");
Some(handles[..n].iter().all(|&handle| {
let mut buf = [0u8; 112];
// SAFETY: `buf` is writable for the 112 bytes passed.
let got = unsafe { libc::proc_pidinfo(pid, 5, handle, buf.as_mut_ptr().cast(), 112) };
// A thread gone since the listing waits on nothing.
got != 112 || !matches!(i32::from_ne_bytes(buf[24..28].try_into().unwrap()), 1 | 4)
}))
}
/// The mach timebase, `(numer, denom)`: ticks × numer / denom = ns.
fn timebase() -> (u64, u64) {
#[repr(C)]
struct Timebase {
numer: u32,
denom: u32,
}
extern "C" {
fn mach_timebase_info(info: *mut Timebase) -> libc::c_int;
}
let mut tb = Timebase { numer: 0, denom: 0 };
// SAFETY: `tb` is writable for the call.
assert_eq!(unsafe { mach_timebase_info(&mut tb) }, 0);
(u64::from(tb.numer), u64::from(tb.denom))
}
/// `pid` and its descendants, from `ps`.
fn ps_tree(root: i32) -> Vec<i32> {
let out = Command::new("/bin/ps").args(["-A", "-o", "pid=,ppid="]).output().expect("ps");
let pairs: Vec<(i32, i32)> = String::from_utf8_lossy(&out.stdout)
.lines()
.filter_map(|l| {
let mut it = l.split_whitespace().map(|w| w.parse::<i32>());
Some((it.next()?.ok()?, it.next()?.ok()?))
})
.collect();
let mut tree = vec![root];
let mut i = 0;
while i < tree.len() {
let p = tree[i];
tree.extend(pairs.iter().filter(|(_, pp)| *pp == p).map(|(c, _)| *c));
i += 1;
}
tree
}
/// `root`'s tree once it has exactly `want` members. Panics after 20 s,
/// naming the tree it last saw.
///
/// A server's reply does not mean its worker is gone. The worker prints and
/// the parent reads that line while the worker still has its work to do and
/// its exit and reaping after that, so a tree walked straight after a reply
/// can still hold it. Waiting for the shape a test assumes is the only way to
/// assert that shape; asserting it on the first walk asserts a race, and that
/// race is real — the count came back 62 instead of 61 in 2 runs of 5 on this
/// host, more often the busier it is.
fn settled_tree(root: i32, want: usize) -> Vec<i32> {
let deadline = Instant::now() + Duration::from_secs(20);
loop {
let tree = ps_tree(root);
if tree.len() == want {
return tree;
}
assert!(
Instant::now() < deadline,
"the tree of {root} never settled to {want} members; last saw {}: {tree:?}",
tree.len()
);
std::thread::sleep(Duration::from_millis(20));
}
}
/// Each of `pids`' rusage once every one of them is quiet by the oracle's
/// rule: no thread running, queued or in a wait no signal ends (or the
/// process exited) at two reads 20 ms apart, with the same energy at both.
/// Panics after 20 s.
fn quiet(pids: &[i32]) -> Vec<[u64; 64]> {
let still = |pid: i32, ru: &[u64; 64]| {
(oracle_running(pid) == Some(0) && oracle_at_rest(pid) == Some(true)) || exit_abstime(ru) != 0
};
let deadline = Instant::now() + Duration::from_secs(20);
loop {
let first: Vec<[u64; 64]> = pids.iter().map(|&p| oracle_rusage(p).expect("oracle read")).collect();
let before = pids.iter().zip(&first).all(|(&p, ru)| still(p, ru));
std::thread::sleep(Duration::from_millis(20));
let second: Vec<[u64; 64]> = pids.iter().map(|&p| oracle_rusage(p).expect("oracle read")).collect();
let after = pids.iter().zip(&second).all(|(&p, ru)| still(p, ru));
let same = first.iter().zip(&second).all(|(a, b)| energy_nj(a) == energy_nj(b));
if before && after && same {
return second;
}
assert!(Instant::now() < deadline, "the oracle never saw {pids:?} quiet");
}
}
/// A server: a child in a process group of its own, driven over its stdin
/// and stdout one line at a time, and killed with its whole group on drop.
struct Server {
child: Child,
stdin: ChildStdin,
stdout: BufReader<ChildStdout>,
}
impl Server {
fn spawn(program: &str, args: &[&str]) -> Server {
let mut child = Command::new(program)
.args(args)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.process_group(0)
.spawn()
.expect("spawn the server");
let stdin = child.stdin.take().unwrap();
let stdout = BufReader::new(child.stdout.take().unwrap());
Server { child, stdin, stdout }
}
fn pid(&self) -> i32 {
self.child.id() as i32
}
/// One line out, and the reply line back.
fn ask(&mut self, line: &str) -> String {
writeln!(self.stdin, "{line}").expect("send");
self.stdin.flush().expect("flush");
self.read_line()
}
fn read_line(&mut self) -> String {
let mut reply = String::new();
self.stdout.read_line(&mut reply).expect("reply");
assert!(!reply.is_empty(), "the server closed its stdout");
reply.trim_end().to_string()
}
fn meter(&self) -> TreeMeter {
TreeMeter::for_process(self.pid() as u32)
.expect("a tree meter for a child of this process (the per-process counter is live on this fleet)")
.with_settle_bound(TEST_SETTLE)
}
}
impl Drop for Server {
fn drop(&mut self) {
// SAFETY: signals this server's own process group.
unsafe { libc::kill(-self.pid(), libc::SIGKILL) };
let _ = self.child.wait();
}
}
fn log(what: &str, rep: usize, r: &TreeReading, oracle_nj: u64) {
eprintln!(
"tree meter: {what} rep {rep}: {} nJ ({} µJ) over {} process(es) {:?}, oracle {oracle_nj} nJ, wall {:?}, gaps {:?}",
r.cpu_nj, r.cpu_uj, r.processes, r.pids, r.wall, r.gaps
);
}
/// A single-process server (awk: each request line is a loop count; it
/// burns that loop, replies, flushes and blocks for the next line), metered
/// over round trips from a bare echo to ~10 ms of CPU: each reading is
/// whole (one process, no GPU client, no Neural Engine memory, settled), not
/// zero, and equal to the oracle's advance across the round trip.
///
/// It is also the control for delegated work: pure CPU work, over which
/// the oracle sees the server's billed and serviced counts hold, carries no
/// `Delegated` gap (each reading is whole) and its figure is the server's
/// own energy, to the nanojoule.
#[test]
fn a_single_process_server_reads_whole_and_exactly() {
let _one = alone();
let mut s = Server::spawn("/usr/bin/awk", &["{ for (i = 0; i < $1; i++) x += i; print x + 0; fflush() }"]);
let meter = s.meter();
let pid = s.pid();
let mut before = quiet(&[pid])[0];
for (rep, n) in [0u32, 1, 20_000, 100_000, 300_000, 0].into_iter().enumerate() {
let (reply, r) = meter.measure(|| s.ask(&n.to_string()));
let r = r.expect("a reading of the round trip");
let after = quiet(&[pid])[0];
let oracle = energy_nj(&after) - energy_nj(&before);
log(&format!("single process, loop {n}"), rep, &r, oracle);
assert!(!reply.is_empty());
let ledgers = |ru: &[u64; 64]| (billed_nj(ru), billed_ticks(ru), serviced_nj(ru), serviced_ticks(ru));
assert_eq!(ledgers(&after), ledgers(&before), "setup: pure CPU work billed or serviced nothing");
assert!(r.is_whole(), "{:?}", r.gaps);
assert_eq!((r.processes, r.pids.as_slice()), (1, &[pid as u32][..]));
assert!(r.cpu_nj > 0, "{r:?}");
assert_eq!(r.cpu_nj, oracle, "the meter's reading against the oracle's advance");
assert_eq!(r.cpu_uj, r.cpu_nj.div_ceil(1_000), "the µJ figure is the nJ figure rounded up");
before = after;
}
}
/// A server that runs each request in a worker it forks and reaps (a shell
/// loop running awk): the worker's energy is in no counter once it is
/// reaped, so each reading carries exactly one gap, `ReapedDescendant` for
/// the shell, with the worker's CPU time to the nanosecond the oracle's
/// reaped-descendant time gives, and is not whole. The shell's own advance
/// is still read, exactly.
#[test]
fn a_server_that_reaps_its_worker_is_not_whole() {
let _one = alone();
let mut s = Server::spawn(
"/bin/sh",
&["-c", r#"while read l; do /usr/bin/awk "BEGIN { for (i = 0; i < $l; i++) x += i }"; echo ok; done"#],
);
let meter = s.meter();
let pid = s.pid();
let (numer, denom) = timebase();
let mut before = quiet(&[pid])[0];
for rep in 0..3 {
let (reply, r) = meter.measure(|| s.ask("200000"));
let r = r.expect("a reading of the shell's own work");
let after = quiet(&[pid])[0];
log("shell reaping a worker", rep, &r, energy_nj(&after) - energy_nj(&before));
assert_eq!(reply, "ok");
let ticks = reaped_cpu_ticks(&after) - reaped_cpu_ticks(&before);
assert!(ticks > 0, "setup: the worker ran");
let cpu = Duration::from_nanos(ticks * numer / denom);
assert_eq!(r.gaps, [TreeGap::ReapedDescendant { pid: pid as u32, cpu }]);
assert!(!r.is_whole());
assert_eq!(r.cpu_nj, energy_nj(&after) - energy_nj(&before), "the shell's own advance");
before = after;
}
}
/// A server that answers each request with a worker it forks, which writes
/// the reply ITSELF, keeps computing for a few milliseconds and exits; the
/// server reaps it after the reply has gone (perl, with 60 idle children
/// beside it, so a walk of its tree takes long enough for the reap to land
/// inside one). Born in the span, the worker spent energy in it, and by the
/// time the stop is read it is in one of four states. Three are gaps: still
/// running (`Unsettled`), exited and unreaped (`Exited`), or reaped
/// (`ReapedDescendant` on the server). The fourth is not: alive and on no
/// core, blocked (a process can be, at any point of its work or of its
/// exit), which reads as settled. A walk that reads it there counts its
/// record as it stands, and the reading is whole and names it as a member.
/// Seen on a loaded host: one round trip of 21, in a run where a round trip
/// took up to 1.4 s, read the worker so, at 7.08 mJ, where the round trips
/// beside it that read the same work as `Exited` read 3.76 to 3.88 mJ (an
/// exited record lacks the process's last time on a core).
///
/// So what holds of every reading is the test's name: the worker is never
/// missed. A reading that does not hold the worker as a member is not whole
/// — in particular not one taken by a walk that read the server before the
/// reap and looked for the worker after it, which finds neither the worker
/// nor its reaped time.
#[test]
fn a_worker_reaped_while_its_tree_is_walked_is_never_missed() {
let _one = alone();
let mut s = Server::spawn(
"/usr/bin/perl",
&[
"-e",
r#"$| = 1;
for (1 .. 60) { if (fork() == 0) { sleep 1000 while 1 } }
while (my $l = <STDIN>) {
my $p = fork();
if ($p == 0) { print "$$\n"; my $x = 0; $x += $_ for 1 .. 150000; exit 0 }
waitpid($p, 0);
}"#,
],
);
let root = s.pid() as u32;
// The first round trip runs once all 60 are forked. It does not leave the
// tree quiet, though — the worker it forks prints its pid before doing its
// work, so the reply arrives while that worker is still running — hence
// the wait below rather than a bare walk.
let _ = s.ask("warm up");
// Not `ps_tree` + `assert_eq!` here: the warm-up round trip returns while
// its worker is still the server's child, so the first walk sees 62.
let tree = settled_tree(s.pid(), 61);
quiet(&tree);
let meter = s.meter();
let (mut exited, mut reaped, mut unsettled, mut live) = (0, 0, 0, 0);
for rep in 0..40 {
let (reply, r) = meter.measure(|| s.ask("go"));
let worker: u32 = reply.parse().expect("the worker's pid, as its reply");
let r = r.expect("a reading of the server's tree");
eprintln!(
"tree meter: worker reaped by its server rep {rep}: {} nJ over {} process(es), worker {worker} {}, gaps {:?}",
r.cpu_nj,
r.processes,
if r.pids.contains(&worker) { "read" } else { "not read" },
r.gaps
);
if r.is_whole() {
// Whole only with the worker read as a member: alive, settled,
// its record counted from its birth.
assert!(r.pids.contains(&worker), "a reading without the worker born in the span read whole: {r:?}");
assert_eq!(r.processes, 62, "the server, its 60 idle children and the worker: {r:?}");
live += 1;
continue;
}
if r.gaps.iter().any(|g| matches!(g, TreeGap::Unsettled { .. } | TreeGap::Changing { .. })) {
unsettled += 1;
continue;
}
if r.gaps.contains(&TreeGap::Exited { pid: worker }) {
exited += 1;
assert!(r.pids.contains(&worker), "an exited worker's record is counted: {r:?}");
} else if r.gaps.iter().any(|g| matches!(g, TreeGap::ReapedDescendant { pid, .. } if *pid == root)) {
reaped += 1;
} else {
panic!("the worker is in no gap of a settled reading: {r:?}");
}
}
eprintln!(
"tree meter: worker reaped by its server: {exited} read exited, {reaped} reaped, {unsettled} unsettled at the bound, {live} read alive and settled"
);
}
/// A two-process tree that lives across the span (a shell waiting on its
/// awk worker, which serves the requests): both are read at both ends, and
/// the reading is whole and equal to the sum of the oracle's two advances.
#[test]
fn a_tree_of_two_long_lived_processes_reads_both() {
let _one = alone();
let mut s = Server::spawn(
"/bin/sh",
&["-c", r#"/usr/bin/awk "$1"; true"#, "sh", "{ for (i = 0; i < $1; i++) x += i; print x + 0; fflush() }"],
);
let meter = s.meter();
// The worker exists once it has answered.
assert_eq!(s.ask("1"), "0");
let tree = ps_tree(s.pid());
assert_eq!(tree.len(), 2, "setup: the shell and its awk: {tree:?}");
let sum = |rus: &[[u64; 64]]| rus.iter().map(energy_nj).sum::<u64>();
let mut before = sum(&quiet(&tree));
for rep in 0..4 {
let (_, r) = meter.measure(|| s.ask("150000"));
let r = r.expect("a reading");
let after = sum(&quiet(&tree));
log("shell and awk", rep, &r, after - before);
assert!(r.is_whole(), "{:?}", r.gaps);
let mut pids = r.pids.clone();
pids.sort_unstable();
let mut want: Vec<u32> = tree.iter().map(|&p| p as u32).collect();
want.sort_unstable();
assert_eq!(pids, want);
assert_eq!(r.cpu_nj, after - before);
before = after;
}
}
/// A request that makes the tree grow: the shell reads it and starts a
/// second shell, which starts awk; awk burns, replies, and waits on its
/// stdin. A child and a grandchild born inside the span, alive and quiet at
/// its end (the reply comes from the youngest, after both ancestors have
/// done their part). Both are counted whole (they did not exist before it),
/// and the reading is whole and equals the oracle's: the first shell's
/// advance plus the two newcomers' whole counts.
#[test]
fn processes_born_in_the_span_are_counted_whole() {
let _one = alone();
let mut s = Server::spawn(
"/bin/sh",
&[
"-c",
r#"read l; /bin/sh -c '/usr/bin/awk "$1"; true' sh2 "$1"; true"#,
"sh",
"BEGIN { for (i = 0; i < 300000; i++) x += i; print x + 0; fflush(); getline line }",
],
);
let meter = s.meter();
let root = s.pid();
let before = energy_nj(&quiet(&[root])[0]);
assert_eq!(ps_tree(root), [root], "setup: the shell alone before the request");
let (reply, r) = meter.measure(|| s.ask("go"));
let r = r.expect("a reading");
assert!(!reply.is_empty());
let tree = ps_tree(root);
let rus = quiet(&tree);
let oracle = energy_nj(&rus[0]) - before + rus[1..].iter().map(energy_nj).sum::<u64>();
log("shell starting a shell starting awk", 0, &r, oracle);
assert_eq!(tree.len(), 3, "setup: the shell, the second shell and awk: {tree:?}");
assert!(r.is_whole(), "{:?}", r.gaps);
let mut pids = r.pids.clone();
pids.sort_unstable();
let mut want: Vec<u32> = tree.iter().map(|&p| p as u32).collect();
want.sort_unstable();
assert_eq!(pids, want);
assert_eq!(r.cpu_nj, oracle);
}
/// A server that keeps computing after it replies (awk: burn, reply, flush,
/// then burn again for a few milliseconds before it reads the next line).
/// Read the moment the reply lands, its count lacks most of that tail — the
/// kernel folds a thread's energy only when it leaves the core — so the
/// meter must wait it out at both ends: the stop right after this round
/// trip's reply, and the start right after the previous round trip's, while
/// that tail still runs. A stale stop would read less than the oracle; a
/// stale start, more.
#[test]
fn a_reply_with_a_tail_of_work_is_read_once_the_tail_is_done() {
let _one = alone();
let mut s = Server::spawn(
"/usr/bin/awk",
&["{ for (i = 0; i < $1; i++) x += i; print x + 0; fflush(); for (j = 0; j < $2; j++) y += j }"],
);
let meter = s.meter();
let pid = s.pid();
for rep in 0..4 {
// The previous round trip's tail is running as the span opens.
let _ = s.ask("20000 150000");
let start = meter.start();
let before = energy_nj(&quiet(&[pid])[0]);
let reply = s.ask("20000 150000");
let r = meter.stop(start).expect("a reading");
let after = energy_nj(&quiet(&[pid])[0]);
log("awk with a tail after its reply", rep, &r, after - before);
assert!(!reply.is_empty());
assert!(r.is_whole(), "{:?}", r.gaps);
assert_eq!(r.cpu_nj, after - before, "the meter's reading against the oracle's advance");
}
}
/// A server whose reply has a tail that waits on the disk (perl: burn,
/// reply, then twenty times write a block, `fsync` it and burn a little,
/// before it reads the next line). While the kernel writes a block the
/// server's one thread is on no core and queued for none, and two reads of
/// its counts agree — but it is not waiting for a request: it is in a wait
/// no signal ends, and runs again when the disk is done. A stop that took
/// that for rest would read the round trip short by the rest of its tail,
/// with no gap. So a member with a thread in such a wait is not settled, and
/// the stop waits the tail out: whole, and equal to the oracle's advance.
///
/// The settle bound here is far past the others': a write waits on the disk
/// for as long as the host's other writers make it, and the meter returns
/// as soon as the tail is done whatever the bound.
#[test]
fn a_reply_whose_tail_waits_on_the_disk_is_read_once_the_tail_is_done() {
let _one = alone();
let file = std::env::temp_dir().join(format!("tree-energy-disk-tail-{}", std::process::id()));
let mut s = Server::spawn(
"/usr/bin/perl",
&[
"-e",
r#"use IO::Handle;
$| = 1;
open(my $f, '>', $ARGV[0]) or die;
while (my $l = <STDIN>) {
my $x = 0; $x += $_ for 1 .. 20000;
print "$x\n";
for my $k (1 .. 20) { print $f 'x' x 4096; $f->flush; $f->sync; $x += $_ for 1 .. 3000 }
}"#,
file.to_str().expect("a UTF-8 temporary path"),
],
);
let meter = s.meter().with_settle_bound(Duration::from_secs(20));
let pid = s.pid();
// perl has compiled the script and opened the file once it has answered.
let _ = s.ask("warm up");
let mut before = energy_nj(&quiet(&[pid])[0]);
for rep in 0..6 {
let (reply, r) = meter.measure(|| s.ask("go"));
let r = r.expect("a reading");
let after = energy_nj(&quiet(&[pid])[0]);
log("perl with a tail that waits on the disk", rep, &r, after - before);
assert!(!reply.is_empty());
assert!(r.is_whole(), "{:?}", r.gaps);
assert_eq!(r.cpu_nj, after - before, "the meter's reading against the oracle's advance");
before = after;
}
drop(s);
let _ = std::fs::remove_file(&file);
}
/// Set in the environment of this test binary when it is run again inside a
/// sandbox that denies listing processes (the test below).
const IN_LISTING_SANDBOX: &str = "TREE_ENERGY_IN_LISTING_SANDBOX";
/// A sandbox that allows everything but listing processes: each process's
/// usage stays readable, while every `proc_listchildpids` fails.
const DENY_LISTING: &str = "(version 1)(allow default)(deny process-info-listpids)";
/// A meter that cannot list a member's children cannot see any descendant
/// below it, so its reading is not whole: it carries `Unlisted` for that
/// member at each end. The listing is made to fail for real: this test
/// binary runs itself again, this test alone, inside a sandbox that denies
/// listing processes, and there meters a perl root relaying each request to
/// an awk worker that does the work — a tree of two, of which the meter can
/// see only the root.
#[test]
fn a_member_whose_children_cannot_be_listed_is_a_gap() {
if std::env::var_os(IN_LISTING_SANDBOX).is_some() {
let mut s = Server::spawn(
"/usr/bin/perl",
&[
"-e",
r#"use IPC::Open2; $| = 1;
open2(my $out, my $in, "/usr/bin/awk", '{ for (i = 0; i < $1; i++) x += i; print x + 0; fflush() }');
$in->autoflush(1);
while (my $l = <STDIN>) { print $in $l; my $r = <$out>; print $r }"#,
],
);
assert_eq!(s.ask("1"), "0", "setup: the worker answers");
let meter = s.meter();
let root = s.pid() as u32;
for rep in 0..2 {
let (_, r) = meter.measure(|| s.ask("150000"));
let r = r.expect("a reading of the root");
eprintln!(
"tree meter: listing denied rep {rep}: {} nJ over {} process(es) {:?}, gaps {:?}",
r.cpu_nj, r.processes, r.pids, r.gaps
);
assert_eq!(r.pids, [root], "the meter saw the root alone");
assert!(!r.is_whole(), "a tree with its descendants unseen read whole: {:?}", r.gaps);
for end in [SpanEnd::Start, SpanEnd::Stop] {
assert!(
r.gaps.contains(&TreeGap::Unlisted { pid: root, errno: libc::EPERM, end }),
"no Unlisted gap for the root at the {end}: {:?}",
r.gaps
);
}
}
return;
}
let _one = alone();
let exe = std::env::current_exe().expect("this test binary");
let out = Command::new("/usr/bin/sandbox-exec")
.args(["-p", DENY_LISTING])
.arg(&exe)
.args(["--exact", "a_member_whose_children_cannot_be_listed_is_a_gap", "--nocapture", "--test-threads=1"])
.env(IN_LISTING_SANDBOX, "1")
.output()
.expect("run this test binary under sandbox-exec");
let stdout = String::from_utf8_lossy(&out.stdout);
let stderr = String::from_utf8_lossy(&out.stderr);
eprint!("{stderr}");
assert!(out.status.success(), "the sandboxed run failed:\n{stdout}\n{stderr}");
assert!(stdout.contains("1 passed"), "the sandboxed run did not run this test:\n{stdout}");
}
/// A child that had already exited at the start (a zombie its parent has
/// not reaped) and is reaped inside the span moves the parent's
/// reaped-descendant time, but by exactly its own CPU time, all spent before
/// the span: no gap. The reading is whole, of the parent alone.
#[test]
fn a_child_that_exited_before_the_span_and_is_reaped_in_it_is_no_gap() {
let _one = alone();
let mut s = Server::spawn(
"/usr/bin/perl",
&[
"-e",
r#"$| = 1; my $p = fork(); if ($p == 0) { my $x = 0; $x += $_ for 1 .. 200000; exit 0 }
select(undef, undef, undef, 0.3); print "ready $p\n";
my $l = <STDIN>; waitpid($p, 0); print "ok\n"; $l = <STDIN>;"#,
],
);
let ready = s.read_line();
let zombie: i32 = ready.strip_prefix("ready ").and_then(|p| p.parse().ok()).expect("the child's pid");
// Wait for the child to have exited (a zombie) before the span opens.
let deadline = Instant::now() + Duration::from_secs(20);
while oracle_rusage(zombie).is_none_or(|ru| exit_abstime(&ru) == 0) {
assert!(Instant::now() < deadline, "setup: the child never exited");
std::thread::sleep(Duration::from_millis(10));
}
let meter = s.meter();
let pid = s.pid();
let before = quiet(&[pid])[0];
let (reply, r) = meter.measure(|| s.ask("go"));
let r = r.expect("a reading");
let after = quiet(&[pid])[0];
log("perl reaping a child that exited before the span", 0, &r, energy_nj(&after) - energy_nj(&before));
assert_eq!(reply, "ok");
assert!(reaped_cpu_ticks(&after) > reaped_cpu_ticks(&before), "setup: the reap moved perl's reaped-descendant time");
assert!(oracle_rusage(zombie).is_none(), "setup: the child was reaped");
assert!(r.is_whole(), "{:?}", r.gaps);
assert_eq!((r.processes, r.cpu_nj), (1, energy_nj(&after) - energy_nj(&before)));
}
/// A Metal compute server for the GPU tests, built from this source when
/// they first run: it prints `ready`, then answers each request line with
/// one line. `persistent` runs a compute kernel on the command queue the
/// server keeps; `transient` runs it on a queue made for the request and
/// released after it; `cpu` does no GPU work. The reply is the GPU time the
/// request's command buffer took by its own timestamps, in ns (0 for `cpu`):
/// an oracle that shares nothing with the meter's I/O Registry reader.
const METAL_SERVER: &str = r##"#import <Metal/Metal.h>
#include <stdio.h>
#include <string.h>
static NSString *src = @"#include <metal_stdlib>\nusing namespace metal;\n"
"kernel void burn(device float *o [[buffer(0)]], uint i [[thread_position_in_grid]]) {\n"
" float x = float(i); for (int k = 0; k < 20000; k++) { x = x * 1.0000001f + 0.5f; } o[i] = x; }\n";
int main(void) {
@autoreleasepool {
id<MTLDevice> d = MTLCreateSystemDefaultDevice();
if (!d) { printf("no Metal device\n"); fflush(stdout); return 1; }
NSError *err = nil;
id<MTLLibrary> lib = [d newLibraryWithSource:src options:nil error:&err];
if (!lib) { printf("no library: %s\n", err.localizedDescription.UTF8String); fflush(stdout); return 1; }
id<MTLComputePipelineState> ps = [d newComputePipelineStateWithFunction:[lib newFunctionWithName:@"burn"] error:&err];
id<MTLCommandQueue> kept = [d newCommandQueue];
id<MTLBuffer> buf = [d newBufferWithLength:sizeof(float) * 65536 options:MTLResourceStorageModeShared];
printf("ready\n"); fflush(stdout);
char line[64];
while (fgets(line, sizeof line, stdin)) {
if (strncmp(line, "cpu", 3) == 0) { printf("0\n"); fflush(stdout); continue; }
double ns = 0;
@autoreleasepool {
id<MTLCommandQueue> q = strncmp(line, "transient", 9) == 0 ? [d newCommandQueue] : kept;
id<MTLCommandBuffer> b = [q commandBuffer];
id<MTLComputeCommandEncoder> e = [b computeCommandEncoder];
[e setComputePipelineState:ps];
[e setBuffer:buf offset:0 atIndex:0];
[e dispatchThreads:MTLSizeMake(65536, 1, 1) threadsPerThreadgroup:MTLSizeMake(256, 1, 1)];
[e endEncoding];
[b commit];
[b waitUntilCompleted];
ns = (b.GPUEndTime - b.GPUStartTime) * 1e9;
}
printf("%.0f\n", ns); fflush(stdout);
}
}
return 0;
}
"##;
/// The Metal server, built once per test run into Cargo's scratch directory
/// for integration tests, with the C compiler every macOS Rust build links
/// with.
fn metal_server() -> &'static str {
static BUILT: std::sync::OnceLock<String> = std::sync::OnceLock::new();
BUILT.get_or_init(|| {
let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"));
let source = dir.join("tree_energy_metal_server.m");
let binary = dir.join(format!("tree_energy_metal_server-{}", std::process::id()));
std::fs::write(&source, METAL_SERVER).expect("write the Metal server's source");
let out = Command::new("/usr/bin/cc")
.args(["-fobjc-arc", "-O2", "-framework", "Foundation", "-framework", "Metal", "-o"])
.arg(&binary)
.arg(&source)
.output()
.expect("run the C compiler");
assert!(out.status.success(), "the Metal server did not build:\n{}", String::from_utf8_lossy(&out.stderr));
binary.to_string_lossy().into_owned()
})
}
/// A server that holds a GPU client never reads whole, whatever a request
/// did, because its GPU use is not known to be zero: work on its kept queue
/// is recorded and carried as `Gpu` time (the reader sees it), while work
/// on a queue released inside the span leaves its client's recorded time
/// unmoved, as does a request with no GPU work at all, and both are
/// `GpuUnknown` for the server. The reply's command-buffer time is the
/// oracle that the GPU did work. What is tested is the GPU half alone, so
/// the meter waits only 200 ms for the server to go quiet: a Metal process
/// keeps threads that wake on their own, and on this host it is often still
/// running at an end, which is a gap of its own and no part of this test.
#[test]
fn a_server_holding_a_gpu_client_never_reads_whole() {
let _one = alone();
let mut s = Server::spawn(metal_server(), &[]);
assert_eq!(s.read_line(), "ready", "setup: the Metal server started");
let meter = TreeMeter::for_process(s.pid() as u32)
.expect("a tree meter for the Metal server")
.with_settle_bound(Duration::from_millis(200));
let pid = s.pid() as u32;
for (rep, mode) in ["persistent", "transient", "cpu", "transient", "persistent", "cpu", "transient"].into_iter().enumerate() {
let (reply, r) = meter.measure(|| s.ask(mode));
let gpu_ns: u64 = reply.parse().unwrap_or_else(|_| panic!("the server's reply: {reply}"));
let r = r.expect("a reading of the Metal server");
eprintln!(
"tree meter: Metal server, {mode} rep {rep}: command buffer {gpu_ns} ns on the GPU; reading {} nJ, gaps {:?}",
r.cpu_nj, r.gaps
);
assert!(!r.is_whole(), "a server holding a GPU client read whole after a {mode} request: {:?}", r.gaps);
if mode == "cpu" {
assert_eq!(gpu_ns, 0);
} else {
assert!(gpu_ns > 0, "setup: the {mode} request ran on the GPU");
}
let recorded = r.gaps.iter().find_map(|g| match g {
TreeGap::Gpu { ns } => Some(*ns),
_ => None,
});
let unknown = r.gaps.contains(&TreeGap::GpuUnknown { pid: Some(pid) });
if mode == "persistent" {
assert!(recorded.is_some_and(|ns| ns > 0), "the kept queue's work was not seen: {:?}", r.gaps);
} else {
assert!(recorded.is_some() || unknown, "no GPU gap for the server after a {mode} request: {:?}", r.gaps);
}
}
}
/// A server for the delegated-work test, built from this source when it
/// first runs: it prints `ready`, then answers each request line with one
/// line. `cpu N` burns an N-iteration loop in this process and replies
/// `ok 0`. `delegate K` looks up K account names no account has
/// (`getpwnam`): a system directory service does each lookup, on a message
/// that carries this process's voucher, so the kernel bills the service's
/// CPU work for it to this process. It then waits — up to 5 s, in 1 ms
/// sleeps — for its own billed counts to move before it replies `ok <ms
/// waited>`: a servicing thread is credited when it next changes voucher,
/// which can be after the reply, and the test is of how the meter counts a
/// bill that landed in its span, not of when the kernel lands it. Plain C,
/// with the compiler every macOS Rust build links with.
const DELEGATING_SERVER: &str = r##"#include <libproc.h>
#include <pwd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/resource.h>
#include <unistd.h>
static void billed(unsigned long long *nj, unsigned long long *ticks) {
struct rusage_info_v6 r;
memset(&r, 0, sizeof r);
proc_pid_rusage(getpid(), RUSAGE_INFO_V6, (rusage_info_t *)&r);
*nj = r.ri_billed_energy;
*ticks = r.ri_billed_system_time;
}
int main(void) {
char line[256];
unsigned long seq = 0;
printf("ready\n"); fflush(stdout);
while (fgets(line, sizeof line, stdin)) {
if (strncmp(line, "cpu ", 4) == 0) {
long n = atol(line + 4);
volatile unsigned long long x = 1;
for (long i = 0; i < n; i++) x = x * 6364136223846793005ULL + 1;
printf("ok 0\n"); fflush(stdout);
} else if (strncmp(line, "delegate ", 9) == 0) {
long k = atol(line + 9);
unsigned long long nj0, t0, nj1, t1;
billed(&nj0, &t0);
for (long i = 0; i < k; i++) {
char name[96];
snprintf(name, sizeof name, "no-such-account-%d-%lu", (int)getpid(), seq++);
(void)getpwnam(name);
}
int waited = 0;
for (; waited < 5000; waited++) {
billed(&nj1, &t1);
if (nj1 > nj0 || t1 > t0) break;
usleep(1000);
}
printf("ok %d\n", waited); fflush(stdout);
} else {
printf("?\n"); fflush(stdout);
}
}
return 0;
}
"##;
/// The delegating server, built once per test run into Cargo's scratch
/// directory for integration tests.
fn delegating_server() -> &'static str {
static BUILT: std::sync::OnceLock<String> = std::sync::OnceLock::new();
BUILT.get_or_init(|| {
let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"));
let source = dir.join("tree_energy_delegating_server.c");
let binary = dir.join(format!("tree_energy_delegating_server-{}", std::process::id()));
std::fs::write(&source, DELEGATING_SERVER).expect("write the delegating server's source");
let out = Command::new("/usr/bin/cc").args(["-O2", "-o"]).arg(&binary).arg(&source).output().expect("run the C compiler");
assert!(out.status.success(), "the delegating server did not build:\n{}", String::from_utf8_lossy(&out.stderr));
binary.to_string_lossy().into_owned()
})
}
/// `pid`'s rusage once its billed counts have held for 300 ms: whatever was
/// billed for its start, which can land late, has landed. Panics after 20 s.
fn billed_quiet(pid: i32) -> [u64; 64] {
let deadline = Instant::now() + Duration::from_secs(20);
loop {
let a = oracle_rusage(pid).expect("oracle read");
std::thread::sleep(Duration::from_millis(300));
let b = oracle_rusage(pid).expect("oracle read");
if (billed_nj(&a), billed_ticks(&a)) == (billed_nj(&b), billed_ticks(&b)) {
return b;
}
assert!(Instant::now() < deadline, "the oracle never saw {pid}'s billed counts hold");
}
}
/// Work a system service does for a tree member is billed to it by the
/// kernel, and the meter counts it and says so. One process, the delegating
/// server above, metered over two kinds of round trip:
///
/// - `cpu`, first (once anything billed for its start has landed): pure CPU
/// work, the control. No `Delegated` gap — the reading is whole — and the
/// figure is the server's own advance, exactly, as the oracle reads it.
/// - `delegate`: the server has a system service look up account names. The
/// reading carries `Delegated` and nothing else, its billed energy and CPU
/// time at least what the oracle saw billed inside the span (the oracle
/// reads inside the meter's span, after its start and before its stop, so
/// a bill landing late is in the meter's and maybe not the oracle's), and
/// the figure is at least the oracle's own advance less its serviced plus
/// its billed — and, less the gap's billed energy, exactly the server's
/// own advance less what it serviced: the bill is added once, on top of
/// an exact own count.
#[test]
fn work_a_system_service_does_for_a_member_is_counted_and_flagged() {
let _one = alone();
let mut s = Server::spawn(delegating_server(), &[]);
assert_eq!(s.read_line(), "ready", "setup: the delegating server started");
let pid = s.pid();
let meter = s.meter();
let (numer, denom) = timebase();
let ticks = |t: u64| Duration::from_nanos(t * numer / denom);
billed_quiet(pid);
for rep in 0..3 {
let span = meter.start();
let before = quiet(&[pid])[0];
let reply = s.ask("cpu 2000000");
let after = quiet(&[pid])[0];
let stopping = Instant::now();
let r = meter.stop(span).expect("a reading of the control");
let stop_took = stopping.elapsed();
let own = energy_nj(&after) - energy_nj(&before);
eprintln!(
"tree meter: delegating server, cpu control rep {rep}: {} nJ, oracle own {own} nJ, billed {} nJ / {} ticks; stop took {stop_took:?}; gaps {:?}",
r.cpu_nj,
billed_nj(&after) - billed_nj(&before),
billed_ticks(&after) - billed_ticks(&before),
r.gaps
);
assert_eq!(reply, "ok 0");
assert_eq!((billed_nj(&after), billed_ticks(&after)), (billed_nj(&before), billed_ticks(&before)), "setup: the control billed nothing");
assert!(!r.gaps.iter().any(|g| matches!(g, TreeGap::Delegated { .. })), "pure CPU work read as delegated: {:?}", r.gaps);
assert!(r.is_whole(), "{:?}", r.gaps);
assert_eq!(r.cpu_nj, own, "the control's figure is the server's own advance");
}
for rep in 0..4 {
let span = meter.start();
let before = quiet(&[pid])[0];
let reply = s.ask("delegate 3");
let after = quiet(&[pid])[0];
let stopping = Instant::now();
let r = meter.stop(span).expect("a reading of the delegating round trip");
let stop_took = stopping.elapsed();
let own = i128::from(energy_nj(&after) - energy_nj(&before));
let serviced = i128::from(serviced_nj(&after) - serviced_nj(&before));
let billed = billed_nj(&after) - billed_nj(&before);
let billed_cpu = ticks(billed_ticks(&after) - billed_ticks(&before));
let gap = r.gaps.iter().find_map(|g| match g {
TreeGap::Delegated { billed_nj, billed_cpu } => Some((*billed_nj, *billed_cpu)),
_ => None,
});
eprintln!(
"tree meter: delegating server, delegate rep {rep} ({reply}): {} nJ; oracle own {own} nJ, serviced {serviced} nJ, billed {billed} nJ / {billed_cpu:?}; gap {gap:?}; stop took {stop_took:?}; gaps {:?}",
r.cpu_nj, r.gaps
);
assert!(reply.starts_with("ok "), "{reply}");
assert!(billed > 0 || billed_cpu > Duration::ZERO, "setup: the oracle saw no work billed to the server");
let (gap_nj, gap_cpu) = gap.unwrap_or_else(|| panic!("no Delegated gap for work billed to the server: {:?}", r.gaps));
assert_eq!(r.gaps.len(), 1, "only the delegation is missing: {:?}", r.gaps);
assert!(!r.is_whole());
assert!(gap_nj >= billed && gap_cpu >= billed_cpu, "the gap's bill ({gap_nj} nJ, {gap_cpu:?}) is under the oracle's");
let figure = i128::from(r.cpu_nj);
assert!(figure >= own - serviced + i128::from(billed), "the figure is under the oracle's own + billed − serviced");
assert_eq!(figure - i128::from(gap_nj), own - serviced, "the figure less its bill is the server's own advance less its service");
assert!(r.cpu_nj > own as u64, "the bill was added to the server's own energy");
}
}