open-standards/jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/tests/process_cpu.rs

//! This process's own CPU-energy counter on real hardware, asserted in a process
//! of its own.
//!
//! The counter reads every thread of this process, so a test that bounds it from
//! above cannot share its process with other tests running at the same time:
//! their threads would be in the figure. This file is one test binary with one
//! test, which runs its checks in sequence on one thread. Nothing in it returns
//! early: off macOS it is not compiled, and on macOS a counter that is not live
//! fails it.

#![cfg(target_os = "macos")]

use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};

use jouleclaw_energy::meter::{detect_in_process_meter, detect_meter, EnergyMeter};
use jouleclaw_energy::process::{process_cpu_liveness, ProcessCpuCounter};
use jouleclaw_energy::{CounterLiveness, EnergyCounter, EnergyDomain, EnergyError, Provenance};

/// This thread's CPU time.
fn thread_cpu() -> Duration {
    let mut ts = libc::timespec { tv_sec: 0, tv_nsec: 0 };
    // SAFETY: `ts` is a valid, writable timespec for the duration of the call.
    let rc = unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &mut ts) };
    assert_eq!(rc, 0, "the thread CPU clock");
    Duration::new(ts.tv_sec as u64, ts.tv_nsec as u32)
}

/// Burn `cpu` of this thread's own CPU time (not wall time, so a loaded host
/// that deschedules the thread still gets the whole burn).
fn burn(cpu: Duration) {
    let start = thread_cpu();
    let mut x = 1u64;
    while thread_cpu() - start < cpu {
        for _ in 0..10_000 {
            x = std::hint::black_box(x.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1));
        }
    }
}

/// Another process's cumulative CPU energy, in nJ, read independently of the
/// crate: `proc_pid_rusage(pid, RUSAGE_INFO_V6)` into a 464-byte buffer, whose
/// `ri_energy_nj` is the u64 at byte offset 336 (the SDK's layout).
fn energy_nj_of(pid: libc::pid_t) -> u64 {
    let mut buf = [0u64; 58];
    // SAFETY: `buf` is 464 writable bytes, the size of `rusage_info_v6`.
    let rc = unsafe { libc::proc_pid_rusage(pid, 6, buf.as_mut_ptr() as *mut libc::rusage_info_t) };
    assert_eq!(rc, 0, "proc_pid_rusage({pid})");
    buf[336 / 8]
}

/// How long [`Running`] waits for a process it has killed to be reaped. `SIGKILL` ends a
/// process at once unless it is stuck in the kernel; a drop that waited on such a one without
/// a bound would hang the test binary instead of failing it.
const REAPED_WITHIN: Duration = Duration::from_secs(10);

/// A process the test started, killed and reaped when it is dropped: at the end of the test, or
/// while a failed assertion unwinds through it. A [`Child`] is not killed when it is dropped, so
/// an assertion that failed while the busy child ran left it spinning a core after the test
/// binary had ended, one more for each failed run. It reads as the child it holds.
struct Running(Child);

impl Running {
    /// Kills the process and reaps it, waiting up to [`REAPED_WITHIN`]: whether it was reaped.
    /// A process already reaped is not signalled again (`Child::kill` does nothing then, as its
    /// pid may be another process's by now).
    fn stop(&mut self) -> bool {
        let _ = self.0.kill();
        let t = Instant::now();
        loop {
            match self.0.try_wait() {
                Ok(Some(_)) => return true,
                Ok(None) if t.elapsed() < REAPED_WITHIN => std::thread::sleep(Duration::from_millis(10)),
                _ => return false,
            }
        }
    }
}

impl std::ops::Deref for Running {
    type Target = Child;

    fn deref(&self) -> &Child {
        &self.0
    }
}

impl std::ops::DerefMut for Running {
    fn deref_mut(&mut self) -> &mut Child {
        &mut self.0
    }
}

impl Drop for Running {
    fn drop(&mut self) {
        if !self.stop() {
            eprintln!("process {} was killed and not reaped within {} s", self.0.id(), REAPED_WITHIN.as_secs());
        }
    }
}

#[test]
fn the_process_counter_meters_this_process_and_nothing_else() {
    // 1. Live, probed once per process however many threads ask.
    let askers: Vec<_> = (0..4).map(|_| std::thread::spawn(|| ProcessCpuCounter::new().liveness())).collect();
    let seen: Vec<CounterLiveness> = askers.into_iter().map(|t| t.join().expect("asker")).collect();
    let counter = ProcessCpuCounter::new();
    let _ = detect_in_process_meter();
    let _ = detect_in_process_meter();
    // Counted before anything here asks for the liveness directly: the counters
    // and meters above must have run the probe themselves, once.
    let runs = jouleclaw_energy::process_cpu_probe_runs();
    println!("process-cpu: liveness={:?} seen by 4 threads={seen:?} probe runs={runs}", counter.liveness());
    assert_eq!(runs, 1, "the counters did not run the liveness probe exactly once");
    assert_eq!(counter.liveness(), CounterLiveness::Live, "this process's CPU-energy counter is not live here");
    assert_eq!(process_cpu_liveness(), CounterLiveness::Live);
    assert!(seen.iter().all(|l| *l == CounterLiveness::Live), "{seen:?}");
    assert_eq!(jouleclaw_energy::process_cpu_probe_runs(), 1, "the liveness probe ran more than once");
    assert!(counter.is_available());
    let r = counter.read().expect("a live counter reads");
    assert_eq!((r.domain, r.provenance), (EnergyDomain::ProcessCpu, Provenance::HwShunt));
    assert_eq!((counter.domain(), counter.provenance(), counter.name()), (EnergyDomain::ProcessCpu, Provenance::HwShunt, "process-cpu"));

    // 2. 200 ms of one thread's CPU time is a reading, and the reading is this
    //    process's own energy over the burn: no more than an oracle's window
    //    around it, and most of it. No joule figure is asserted — what 200 ms
    //    of CPU costs depends on the core class that ran it (0.099 J on an
    //    efficiency core at a load average of 114; 0.18-1.24 J otherwise on
    //    an M5 Max), so a fixed floor fails on the host's load, not on the
    //    meter. The oracle sleeps before each read: a thread that leaves its
    //    core has its energy folded into the process's count, so the oracle's
    //    window holds the whole burn. The counter's own reads may each miss
    //    up to a scheduling quantum of the reading thread, hence "most of".
    // SAFETY: `getpid` has no preconditions.
    let me = unsafe { libc::getpid() };
    std::thread::sleep(Duration::from_millis(2));
    let oracle_before = energy_nj_of(me);
    let t = Instant::now();
    let ((), burned) = counter.measure(|| burn(Duration::from_millis(200)));
    let wall = t.elapsed();
    std::thread::sleep(Duration::from_millis(2));
    let oracle_nj = energy_nj_of(me) - oracle_before;
    let uj = burned.expect("a 200 ms burn is a reading");
    println!(
        "process-cpu: 200 ms one-thread CPU burn = {:.3} J over {:.0} ms wall ({:.2} J per CPU-second); oracle window {:.3} J",
        uj as f64 / 1e6,
        wall.as_secs_f64() * 1e3,
        uj as f64 / 1e6 / 0.2,
        oracle_nj as f64 / 1e9
    );
    assert!(uj > 0, "a burn read as zero");
    assert!(uj * 1_000 <= oracle_nj + 1_000, "the burn ({uj} µJ) exceeds the oracle's window around it ({oracle_nj} nJ)");
    assert!(uj * 1_000 * 4 >= oracle_nj * 3, "the burn ({uj} µJ) is under three quarters of the oracle's window ({oracle_nj} nJ)");

    // 3. A 200 ms sleep reads under a tenth of the burn — or, if nothing of
    //    this process ran, is not a reading at all; never a zero. A counter
    //    that reads wall time at a fixed power reads the sleep at 200 ms over
    //    the burn's wall time of the burn's figure — a sixth of it at the
    //    heaviest load seen here (1.26 s of wall for the burn) — and fails.
    let ((), slept) = counter.measure(|| std::thread::sleep(Duration::from_millis(200)));
    println!("process-cpu: 200 ms sleep = {slept:?} µJ");
    match slept {
        Ok(slept_uj) => assert!(slept_uj * 10 < uj, "{slept_uj} µJ for a sleep, against {uj} µJ for the burn"),
        Err(EnergyError::NotAdvanced) => {}
        Err(e) => panic!("a sleep window failed: {e}"),
    }

    // 4. `detect_meter` does not include it; `detect_in_process_meter` selects it.
    let names = detect_meter().names();
    println!("process-cpu: detect_meter holds {names:?}, in-process meter is {}", detect_in_process_meter().active_name());
    assert!(!names.contains(&"process-cpu"), "{names:?}");
    assert_ne!(detect_meter().active_name(), "process-cpu");
    assert_eq!(detect_in_process_meter().active_name(), "process-cpu");

    // 5. A child process's work is not attributed: a child burns CPU while this
    //    process sleeps, and is killed and reaped inside the window. The child's
    //    own counter, read independently, shows the work happened. The child
    //    runs until its own counter has read 0.2 J (at least 300 ms, at most
    //    5 s), not for a fixed 300 ms: on a heavily loaded host a fixed 300 ms
    //    got the child too little CPU (0.076 J at a load average of 170), and
    //    the test failed on the host's load, not on the meter. The child is
    //    held as it starts, so a read that fails while it runs kills it too.
    let mut child = Running(
        Command::new("/bin/sh")
            .args(["-c", "while :; do :; done"])
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("spawn a busy child"),
    );
    let pid = child.id() as libc::pid_t;
    let child_before = energy_nj_of(pid);
    let started = Instant::now();
    let (child_burned_nj, parent) = counter.measure(|| {
        let mut burned = 0;
        while started.elapsed() < Duration::from_secs(5) {
            std::thread::sleep(Duration::from_millis(50));
            burned = energy_nj_of(pid) - child_before;
            if burned >= 200_000_000 && started.elapsed() >= Duration::from_millis(300) {
                break;
            }
        }
        child.kill().expect("kill the child");
        child.wait().expect("reap the child");
        burned
    });
    println!(
        "process-cpu: child burned {:.3} J in {:?}; this process's window = {parent:?} µJ",
        child_burned_nj as f64 / 1e9,
        started.elapsed()
    );
    assert!(child_burned_nj >= 200_000_000, "the child did not burn: {child_burned_nj} nJ");
    match parent {
        // Under a tenth of what the child burned: this process only polled.
        Ok(uj) => assert!(uj * 1_000 * 10 < child_burned_nj, "{uj} µJ in this process while only the child worked ({child_burned_nj} nJ)"),
        Err(EnergyError::NotAdvanced) => {}
        Err(e) => panic!("the parent window failed: {e}"),
    }
}