//! 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");
    }
}