diff --git a/wai/CHANGELOG.md b/wai/CHANGELOG.md index 975e8d7b7..81ec0c096 100644 --- a/wai/CHANGELOG.md +++ b/wai/CHANGELOG.md @@ -6,6 +6,24 @@ wire format may change and such changes are recorded here. ## v0.4.0 — current +### wai-quantum-service: every job's energy, signed with its result + +- **What a result carries.** `energy` holds the CPU energy of the job's worker process, from + its start to its exit, inside the body the receipt seals. It is given as `joules_micro`, + with the acquisition class `OnChipCounter`, a declared uncertainty + (`relative_ppm`, `window_us`) and the counter's filtering state, in + [energy-measurement](extensions/energy-measurement.md) §2's names. It also states its + coverage and what it excludes. +- **How it is read.** On macOS, the service reads the worker's kernel count + (`proc_pid_rusage`, `ri_energy_nj`, plus energy system services billed to it) after the + worker has exited and before it is reaped, waiting with `waitid(WNOWAIT)`. An exited + process has left every core, so the count is complete. The declaration covers the + rounding to whole microjoules, and meets the §2.3 floor at every size. +- **What it does not claim.** A `statevector_single` job runs on the GPU, which the counter + does not read, so its figure is `partial`. A job the service cannot read is + `{"unmetered": why}`, never zero joules. +- **Agents.** The MCP summary states the figure and its coverage. + ### `wai.live.steering`: each client's pathway, chosen by the joules its hops signed - **The extension** ([energy-steering](extensions/energy-steering.md)) registers the record diff --git a/wai/wai-quantum-service/Cargo.lock b/wai/wai-quantum-service/Cargo.lock index 98a0ba144..6a19a8d79 100644 --- a/wai/wai-quantum-service/Cargo.lock +++ b/wai/wai-quantum-service/Cargo.lock @@ -874,6 +874,7 @@ dependencies = [ "axum", "blake3", "getrandom 0.4.3", + "libc", "serde", "serde_json", "tokio", diff --git a/wai/wai-quantum-service/Cargo.toml b/wai/wai-quantum-service/Cargo.toml index ee8fb90de..d09f06dfe 100644 --- a/wai/wai-quantum-service/Cargo.toml +++ b/wai/wai-quantum-service/Cargo.toml @@ -19,5 +19,9 @@ blake3 = "1" getrandom = "0.4" wai-quantum = { path = "../wai-quantum", default-features = false, features = ["quantum", "quantum_toolchain", "quantum_receipt", "quantum_sv", "quantum_tn", "quantum_tableau", "quantum_mpsim", "quantum_spd", "quantum_noisy", "quantum_sv_metal"] } +# proc_pid_rusage and waitid, for each job's energy (src/energy.rs). +[target.'cfg(target_os = "macos")'.dependencies] +libc = "0.2" + [profile.release] lto = "thin" diff --git a/wai/wai-quantum-service/README.md b/wai/wai-quantum-service/README.md index 85fd98ff2..725935a10 100644 --- a/wai/wai-quantum-service/README.md +++ b/wai/wai-quantum-service/README.md @@ -86,6 +86,35 @@ Every finished result carries a `receipt`: an Ed25519 signature, by the key `GET So a result verifies after any JSON reader has carried it, a browser's included. Whole numbers in a request must be within ±(2^53 − 1), so that every reader reads them alike. Receipts issued before the canonical form carry no `canon` and still verify (see `src/receipt.rs`). +## Energy + +Every finished result carries `energy`, inside the body the receipt seals: the CPU energy the job's worker process spent, from its start to its exit. The service reads it from the kernel's per-process count (`proc_pid_rusage`, `ri_energy_nj`, on macOS) after the worker exits and before it is reaped. An exited process has left every core, so its count is complete. + +```json +"energy": { + "joules_micro": 2540113, + "energy_provenance": "OnChipCounter", + "energy_uncertainty": { "relative_ppm": 1, "window_us": 2603412 }, + "energy_counter_filtering": "Unknown", + "coverage": "whole", + "covers": "the CPU energy of the job's worker process, from its start to its exit, with the CPU energy system services billed to it", + "excludes": ["memory", "the host's idle draw"], + "cpu_ms": 31680, + "source": "proc_pid_rusage ri_energy_nj + ri_billed_energy − ri_serviced_energy, read after the worker exited and before it was reaped" +} +``` + +The fields follow WAI's energy-measurement extension (§2): + +- `energy_provenance` is the acquisition class. `OnChipCounter` means the silicon's own accounting, with no calibration behind it. +- `energy_uncertainty.relative_ppm` declares the part of the uncertainty that can be evaluated: here, the rounding to whole microjoules. It is not the figure's total error, because the counter's own systematic error is not evaluated. +- `window_us` is the worker's lifetime. +- `energy_counter_filtering` is `Unknown`, because the platform does not document whether the count is filtered. + +A `statevector_single` job runs on the GPU, which this counter does not read, so its figure is `partial` and says so in `excludes`. So is a figure that includes CPU energy system services billed to the worker, since such credits land late. + +Where the service cannot read the count (off macOS), `energy` is `{"unmetered": ""}`. It is never a figure of zero. `wai-quantum-service run FILE` runs a job in its own process and reports no energy. + ## Running it ```sh diff --git a/wai/wai-quantum-service/smoke.sh b/wai/wai-quantum-service/smoke.sh index 757084cf1..20f111d6e 100644 --- a/wai/wai-quantum-service/smoke.sh +++ b/wai/wai-quantum-service/smoke.sh @@ -57,6 +57,30 @@ assert job["request"]==req, job.get("request") json.dump({"request": job["request"], "result": job["result"]}, open(sys.argv[3],"w"))' "$tmp/ghz.json" "$tmp/ghz.out" "$tmp/vj.json" post /v1/verify "$tmp/vj.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is True, d; print("receipt: verifies from the job alone")' +# The job's energy: a counter figure with its class and declared uncertainty, +# or why it is unmetered, never zero. SMOKE_EXPECT_ENERGY=1 (a macOS host) +# requires the figure. A changed figure breaks the receipt: it is sealed. +export SMOKE_EXPECT_ENERGY="${SMOKE_EXPECT_ENERGY:-0}" +py 'import json,os,sys +e=json.load(open(sys.argv[1]))["result"]["energy"] +if "joules_micro" in e: + uj=e["joules_micro"]; u=e["energy_uncertainty"] + assert uj > 0 and e["energy_provenance"] == "OnChipCounter", e + assert u["relative_ppm"] > 0 and u["window_us"] > 0 and uj * u["relative_ppm"] >= 288676, e + assert e["energy_counter_filtering"] in ("Unknown", "Off", "On") and e["coverage"] in ("whole", "partial"), e + print("energy: %.6f J of CPU, %s, %s" % (uj / 1e6, e["coverage"], e["energy_provenance"])) +else: + assert os.environ["SMOKE_EXPECT_ENERGY"] != "1", e + assert e["unmetered"], e + print("energy: unmetered,", e["unmetered"])' "$tmp/ghz.out" +py 'import json,sys +d=json.load(open(sys.argv[1])) +e=d["result"]["energy"] +if "joules_micro" in e: e["joules_micro"] += 1 +else: e["unmetered"] += "." +json.dump(d, open(sys.argv[2],"w"))' "$tmp/vj.json" "$tmp/ve.json" +post /v1/verify "$tmp/ve.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is False, d; print("energy: a changed figure does not verify")' + # As a browser sends it back: whole doubles without a fraction, keys reversed. py 'import json,sys def js(v): @@ -138,7 +162,10 @@ if py 'import json,sys; d=json.load(open(sys.argv[1])); sys.exit(0 if any(m["met g=json.load(open(sys.argv[1])); c=json.load(open(sys.argv[2])); assert g.get("status")=="done" and c.get("status")=="done", (g, c) gv=[e["value"] for e in g["result"]["expectations"]]; cv=[e["value"] for e in c["result"]["expectations"]] assert g["result"]["accuracy"]["class"]=="single" and all(abs(a-b)<1e-5 for a,b in zip(gv,cv)), (gv, cv) -print("gpu: ok,", g["result"]["accuracy"]["device"], gv)' "$tmp/gpu.out" "$tmp/cpu.out" +e=g["result"]["energy"] +# The counter reads the CPU only, so a GPU job never claims a whole figure. +assert "joules_micro" not in e or (e["coverage"]=="partial" and "GPU" in e["excludes"][0]), e +print("gpu: ok,", g["result"]["accuracy"]["device"], gv, "energy", e.get("coverage", "unmetered"))' "$tmp/gpu.out" "$tmp/cpu.out" else echo "gpu: this instance has none" fi diff --git a/wai/wai-quantum-service/src/discovery.rs b/wai/wai-quantum-service/src/discovery.rs index 1027004b2..3f0719fbf 100644 --- a/wai/wai-quantum-service/src/discovery.rs +++ b/wai/wai-quantum-service/src/discovery.rs @@ -119,7 +119,7 @@ pub fn openapi(server: &str) -> Value { "method": { "type": "string" }, "why": { "type": "string", "description": "Why this method: what it is, and what the circuit is." }, "request": { "$ref": "#/components/schemas/JobRequest", "description": "The request as submitted: the receipt is checked against it." }, - "result": { "type": "object", "description": "counts (highest classical bit first), expectations, amplitudes, accuracy, elapsed_ms and a receipt." }, + "result": { "type": "object", "description": "counts (highest classical bit first), expectations, amplitudes, accuracy, elapsed_ms, energy and a receipt. energy is the CPU energy of the job's process: joules_micro with its class (energy_provenance), declared uncertainty (energy_uncertainty: relative_ppm, window_us) and counter filtering state, its coverage and what it excludes; or unmetered, saying why. It is inside the signed body." }, "error": { "type": "string" } } }); diff --git a/wai/wai-quantum-service/src/energy.rs b/wai/wai-quantum-service/src/energy.rs new file mode 100644 index 000000000..c93bcc6b7 --- /dev/null +++ b/wai/wai-quantum-service/src/energy.rs @@ -0,0 +1,352 @@ +//! A job's energy: what the CPU cores spent running its worker process, as the +//! kernel counts it, read once the worker has exited and before it is reaped. +//! +//! On macOS, `proc_pid_rusage(pid, RUSAGE_INFO_V6)` gives `ri_energy_nj`, the +//! energy in nanojoules the kernel attributes to a process's own threads from +//! the cores' own energy accounting. A thread's energy is added to its process +//! when the thread leaves a core, so a running process's count trails its work. +//! An exited process has left every core, so its count is complete, and it +//! stays readable until its parent reaps it. The service therefore waits for +//! the worker with `waitid(WNOWAIT)`, reads it, and only then reaps it. +//! +//! What the figure covers: the CPU energy of the worker's own threads from its +//! start to its exit (reading the job, running the plan, writing the result), +//! and the CPU energy system services billed to it for work it asked of them +//! (`ri_billed_energy`, less any `ri_serviced_energy`). Not the GPU: a +//! `statevector_single` job runs there, so its figure is partial and says so. +//! Not the memory, not the host's idle draw, not the service's own process. A +//! billed credit lands late and carries no GPU energy, so a figure with one is +//! partial too. +//! +//! The class is `OnChipCounter` (WAI energy-measurement §2): the silicon's own +//! accounting, with no calibration behind it. Its declared uncertainty is the +//! part that can be evaluated (§2.3). The reading is complete, so that part is +//! the rounding of nanojoules to whole microjoules, `0.5/√3 µJ`, and the +//! declaration meets the floor `joules_micro × relative_ppm ≥ 288 676`. The +//! counter's own systematic error is not evaluated, and the class says so. +//! Apple does not document whether the counter is filtered, so the filtering +//! state is `Unknown` (§2.4). +//! +//! Elsewhere the service has no per-process energy counter, and a job says it +//! is unmetered. It never says zero. + +use serde_json::{Value, json}; + +/// The least `joules_micro × relative_ppm` a hardware class may declare +/// (energy-measurement §2.3): `⌈10⁶ × 0.5/√3⌉`. +pub const RESOLUTION_FLOOR_UJ_PPM: u64 = 288_676; + +/// One read of an exited worker. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Reading { + /// `ri_energy_nj`: the CPU energy of the worker's own threads. + pub energy_nj: u64, + /// `ri_billed_energy`: CPU energy system services spent for it. + pub billed_nj: u64, + /// `ri_serviced_energy`: CPU energy it spent for other processes. + pub serviced_nj: u64, + /// Its own CPU time, nanoseconds. + pub cpu_ns: u64, + /// From its start to its exit, nanoseconds. + pub window_ns: u64, +} + +/// The figure a result carries, inside its signed body: a reading, or why +/// there is none. `gpu` says the job's method ran on the GPU, which the +/// counter does not read. +pub fn figure(reading: Option<&Reading>, gpu: bool, unmetered_why: &str) -> Value { + let Some(r) = reading else { + return json!({ "unmetered": unmetered_why }); + }; + if r.energy_nj == 0 { + // A process that ran and reads no energy means the counter is not live + // here; that is no reading, not a reading of zero. + return json!({ "unmetered": "the worker's energy counter did not advance: it is not live on this host" }); + } + let total_nj = r.energy_nj.saturating_add(r.billed_nj).saturating_sub(r.serviced_nj); + let joules_micro = total_nj.saturating_add(500) / 1000; + if joules_micro == 0 { + return json!({ "unmetered": "the worker spent under half a microjoule, below what the figure resolves" }); + } + let relative_ppm = RESOLUTION_FLOOR_UJ_PPM.div_ceil(joules_micro).max(1); + let mut excludes = vec!["memory", "the host's idle draw"]; + if gpu { + excludes.insert(0, "the GPU, which this method runs on"); + } + let delegated = r.billed_nj > 0; + if delegated { + excludes.push("any part of the system services' work not yet billed to the worker, and their GPU work"); + } + json!({ + "joules_micro": joules_micro, + "energy_provenance": "OnChipCounter", + "energy_uncertainty": { + "relative_ppm": relative_ppm, + "window_us": r.window_ns.div_ceil(1000).max(1), + }, + "energy_counter_filtering": "Unknown", + "coverage": if gpu || delegated { "partial" } else { "whole" }, + "covers": "the CPU energy of the job's worker process, from its start to its exit, with the CPU energy system services billed to it", + "excludes": excludes, + "cpu_ms": r.cpu_ns / 1_000_000, + "source": "proc_pid_rusage ri_energy_nj + ri_billed_energy − ri_serviced_energy, read after the worker exited and before it was reaped", + }) +} + +/// Why a job on this host carries no reading. +pub const UNMETERED_HERE: &str = if cfg!(target_os = "macos") { + "the worker's energy could not be read" +} else { + "this host has no per-process energy counter the service reads" +}; + +/// Whether child `pid` has exited, without reaping it: `Some(true)` once it +/// has (it is then readable by [`read`] until reaped), `Some(false)` while it +/// runs, `None` where this cannot be told (off macOS, or the wait failed), in +/// which case the caller reaps as before and has no reading. +#[cfg(target_os = "macos")] +pub fn exited_unreaped(pid: u32) -> Option { + if pid == 0 { + return None; + } + loop { + // SAFETY: an all-zero `siginfo_t` is a valid out-parameter. + let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() }; + // SAFETY: `info` is writable for the call; `pid` is this process's + // child. WNOWAIT leaves it unreaped; WNOHANG returns at once, with + // `si_pid` 0 while it still runs. + let rc = unsafe { + libc::waitid(libc::P_PID, pid as libc::id_t, &mut info, libc::WEXITED | libc::WNOWAIT | libc::WNOHANG) + }; + if rc == 0 { + return Some(info.si_pid != 0); + } + if std::io::Error::last_os_error().raw_os_error() != Some(libc::EINTR) { + return None; + } + } +} + +#[cfg(not(target_os = "macos"))] +pub fn exited_unreaped(_pid: u32) -> Option { + None +} + +/// The exited, unreaped worker's energy, CPU time and lifetime. +#[cfg(target_os = "macos")] +pub fn read(pid: u32) -> Option { + let info = sys::rusage(pid as libc::pid_t)?; + if info.ri_proc_exit_abstime == 0 { + return None; + } + let ns = |ticks: u64| sys::ticks_to_ns(ticks); + Some(Reading { + energy_nj: info.ri_energy_nj, + billed_nj: info.ri_billed_energy, + serviced_nj: info.ri_serviced_energy, + cpu_ns: ns(info.ri_user_time.saturating_add(info.ri_system_time)), + window_ns: ns(info.ri_proc_exit_abstime.saturating_sub(info.ri_proc_start_abstime)), + }) +} + +#[cfg(not(target_os = "macos"))] +pub fn read(_pid: u32) -> Option { + None +} + +#[cfg(target_os = "macos")] +mod sys { + /// `RUSAGE_INFO_V6` in ``. + const RUSAGE_INFO_V6: libc::c_int = 6; + + /// `struct rusage_info_v6` from the macOS SDK's ``, field + /// for field (464 bytes; the offsets read are pinned by a test). Times are + /// in mach ticks. + #[repr(C)] + #[derive(Clone, Copy)] + pub(super) struct RusageInfoV6 { + pub ri_uuid: [u8; 16], + pub ri_user_time: u64, + pub ri_system_time: u64, + pub ri_pkg_idle_wkups: u64, + pub ri_interrupt_wkups: u64, + pub ri_pageins: u64, + pub ri_wired_size: u64, + pub ri_resident_size: u64, + pub ri_phys_footprint: u64, + pub ri_proc_start_abstime: u64, + pub ri_proc_exit_abstime: u64, + pub ri_child_user_time: u64, + pub ri_child_system_time: u64, + pub ri_child_pkg_idle_wkups: u64, + pub ri_child_interrupt_wkups: u64, + pub ri_child_pageins: u64, + pub ri_child_elapsed_abstime: u64, + pub ri_diskio_bytesread: u64, + pub ri_diskio_byteswritten: u64, + pub ri_cpu_time_qos_default: u64, + pub ri_cpu_time_qos_maintenance: u64, + pub ri_cpu_time_qos_background: u64, + pub ri_cpu_time_qos_utility: u64, + pub ri_cpu_time_qos_legacy: u64, + pub ri_cpu_time_qos_user_initiated: u64, + pub ri_cpu_time_qos_user_interactive: u64, + pub ri_billed_system_time: u64, + pub ri_serviced_system_time: u64, + pub ri_logical_writes: u64, + pub ri_lifetime_max_phys_footprint: u64, + pub ri_instructions: u64, + pub ri_cycles: u64, + pub ri_billed_energy: u64, + pub ri_serviced_energy: u64, + pub ri_interval_max_phys_footprint: u64, + pub ri_runnable_time: u64, + pub ri_flags: u64, + pub ri_user_ptime: u64, + pub ri_system_ptime: u64, + pub ri_pinstructions: u64, + pub ri_pcycles: u64, + pub ri_energy_nj: u64, + pub ri_penergy_nj: u64, + pub ri_secure_time_in_system: u64, + pub ri_secure_ptime_in_system: u64, + pub ri_neural_footprint: u64, + pub ri_lifetime_max_neural_footprint: u64, + pub ri_interval_max_neural_footprint: u64, + pub ri_conclave_footprint: u64, + pub ri_page_wait_time_mach: u64, + pub ri_page_cache_hits: u64, + pub ri_reserved: [u64; 6], + } + + pub(super) fn rusage(pid: libc::pid_t) -> Option { + // SAFETY: an all-zero struct of plain integers is a valid value. + let mut info: RusageInfoV6 = unsafe { std::mem::zeroed() }; + // SAFETY: `info` is a writable `rusage_info_v6`, the struct the V6 + // flavor fills, and outlives the call. + let rc = unsafe { libc::proc_pid_rusage(pid, RUSAGE_INFO_V6, &mut info as *mut RusageInfoV6 as *mut libc::rusage_info_t) }; + (rc == 0).then_some(info) + } + + #[repr(C)] + struct Timebase { + numer: u32, + denom: u32, + } + + unsafe extern "C" { + // `libc` marks it deprecated in favour of the `mach2` crate; bound here. + fn mach_timebase_info(info: *mut Timebase) -> libc::c_int; + } + + /// Mach ticks to nanoseconds. + pub(super) fn ticks_to_ns(ticks: u64) -> u64 { + static TB: std::sync::OnceLock<(u32, u32)> = std::sync::OnceLock::new(); + let (numer, denom) = *TB.get_or_init(|| { + let mut tb = Timebase { numer: 0, denom: 0 }; + // SAFETY: `tb` is writable for the call. + let rc = unsafe { mach_timebase_info(&mut tb) }; + if rc == 0 && tb.denom != 0 { (tb.numer, tb.denom) } else { (1, 1) } + }); + (ticks as u128 * numer as u128 / denom as u128) as u64 + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn reading(energy_nj: u64) -> Reading { + Reading { energy_nj, billed_nj: 0, serviced_nj: 0, cpu_ns: 250_000_000, window_ns: 300_000_000 } + } + + #[test] + fn a_figure_is_a_counter_reading_with_its_declared_uncertainty() { + let f = figure(Some(&reading(812_345_678)), false, UNMETERED_HERE); + assert_eq!(f["joules_micro"], 812_346); + assert_eq!(f["energy_provenance"], "OnChipCounter"); + assert_eq!(f["energy_counter_filtering"], "Unknown"); + assert_eq!(f["energy_uncertainty"]["window_us"], 300_000); + assert_eq!(f["coverage"], "whole"); + assert_eq!(f["cpu_ms"], 250); + let ppm = f["energy_uncertainty"]["relative_ppm"].as_u64().unwrap(); + assert_eq!(ppm, 1); + assert!(f.get("unmetered").is_none()); + } + + #[test] + fn the_declaration_meets_the_resolution_floor_at_every_size() { + for nj in [500, 999, 1_000, 1_499, 1_500, 7_777, 288_675, 288_676_000, 288_677_000, 10_000_000_000_000] { + let f = figure(Some(&reading(nj)), false, UNMETERED_HERE); + let uj = f["joules_micro"].as_u64().unwrap(); + let ppm = f["energy_uncertainty"]["relative_ppm"].as_u64().unwrap(); + assert!(uj * ppm >= RESOLUTION_FLOOR_UJ_PPM, "{nj} nJ: {uj} µJ × {ppm} ppm"); + // and declares no more than the rounding: one ppm less breaks the floor + assert!(ppm == 1 || uj * (ppm - 1) < RESOLUTION_FLOOR_UJ_PPM, "{nj} nJ: {ppm} ppm overstates"); + } + } + + #[test] + fn no_reading_is_never_zero_joules() { + for f in [ + figure(None, false, "why"), + figure(Some(&reading(0)), false, UNMETERED_HERE), + figure(Some(&reading(499)), false, UNMETERED_HERE), + ] { + assert!(f.get("joules_micro").is_none() && f.get("energy_provenance").is_none(), "{f}"); + assert!(f["unmetered"].as_str().is_some_and(|s| !s.is_empty())); + } + } + + #[test] + fn a_gpu_method_or_billed_work_makes_the_figure_partial() { + let f = figure(Some(&reading(5_000_000)), true, UNMETERED_HERE); + assert_eq!(f["coverage"], "partial"); + assert!(f["excludes"][0].as_str().unwrap().contains("GPU")); + let billed = Reading { billed_nj: 1_000_000, ..reading(5_000_000) }; + let f = figure(Some(&billed), false, UNMETERED_HERE); + assert_eq!(f["coverage"], "partial"); + assert_eq!(f["joules_micro"], 6_000); + let serviced = Reading { serviced_nj: 1_000_000, ..reading(5_000_000) }; + assert_eq!(figure(Some(&serviced), false, UNMETERED_HERE)["joules_micro"], 4_000); + } + + #[cfg(target_os = "macos")] + #[test] + fn the_layout_is_the_sdks() { + use std::mem::{offset_of, size_of}; + assert_eq!(size_of::(), 464); + assert_eq!(offset_of!(sys::RusageInfoV6, ri_user_time), 16); + assert_eq!(offset_of!(sys::RusageInfoV6, ri_system_time), 24); + assert_eq!(offset_of!(sys::RusageInfoV6, ri_proc_start_abstime), 80); + assert_eq!(offset_of!(sys::RusageInfoV6, ri_proc_exit_abstime), 88); + assert_eq!(offset_of!(sys::RusageInfoV6, ri_billed_energy), 264); + assert_eq!(offset_of!(sys::RusageInfoV6, ri_serviced_energy), 272); + assert_eq!(offset_of!(sys::RusageInfoV6, ri_energy_nj), 336); + } + + /// A child that burns CPU and exits reads a complete, non-zero count while + /// it is unreaped, over a window that covers its run; once reaped, it is gone. + #[cfg(target_os = "macos")] + #[test] + fn an_exited_worker_reads_its_whole_energy_before_it_is_reaped() { + let mut child = std::process::Command::new("/bin/sh") + .args(["-c", "i=0; while [ $i -lt 300000 ]; do i=$((i+1)); done"]) + .spawn() + .expect("sh runs"); + let pid = child.id(); + let started = std::time::Instant::now(); + while exited_unreaped(pid) != Some(true) { + assert!(started.elapsed().as_secs() < 120, "the child did not exit"); + std::thread::sleep(std::time::Duration::from_millis(20)); + } + let r = read(pid).expect("an exited, unreaped child is readable"); + assert!(r.energy_nj > 0, "{r:?}"); + assert!(r.cpu_ns > 10_000_000, "{r:?}"); + assert!(r.window_ns >= r.cpu_ns / 2, "{r:?}"); + let f = figure(Some(&r), false, UNMETERED_HERE); + assert!(f["joules_micro"].as_u64().unwrap() > 0, "{f}"); + assert!(child.wait().unwrap().success()); + assert!(read(pid).is_none(), "a reaped child is not readable"); + } +} diff --git a/wai/wai-quantum-service/src/jobs.rs b/wai/wai-quantum-service/src/jobs.rs index c076d9567..6611bcac5 100644 --- a/wai/wai-quantum-service/src/jobs.rs +++ b/wai/wai-quantum-service/src/jobs.rs @@ -15,6 +15,9 @@ //! mode), at a lower scheduling priority, polled for its resident memory. //! One that passes its admitted memory or its time limit is killed, and the //! job fails saying which. +//! - **Energy.** A finished worker is read before it is reaped, and its +//! result carries the CPU energy its process spent, with the counter's +//! class and declared uncertainty, inside the signed body ([`energy`]). //! - **Fairness.** A client holds at most [`Config::per_client_active`] queued //! or running jobs, and submits at most [`Config::per_client_burst`] in any //! [`Config::per_client_window`]. The queue holds at most @@ -23,6 +26,7 @@ //! Results are kept in memory for [`Config::retain`], then dropped. use crate::catalog::Plan; +use crate::energy; use serde_json::{Value, json}; use std::collections::{HashMap, VecDeque}; use std::process::Stdio; @@ -452,11 +456,22 @@ impl Service { let limit = (mib + 256) << 20; let deadline = Instant::now() + self.config.job_timeout; let mut peak = 0u64; + let mut reading = None; let why = loop { - match child.try_wait() { - Ok(Some(status)) => break status.success().then_some(()).ok_or(format!("the worker ended: {status}")), - Ok(None) => {} - Err(e) => break Err(format!("the worker could not be watched: {e}")), + // The worker's energy is readable only between its exit and its + // reaping, so it is reaped only once it has been seen to exit and + // read ([`energy`]); where that cannot be seen, it is reaped as + // before, and the job is unmetered. + let exited = energy::exited_unreaped(pid); + if exited == Some(true) { + reading = energy::read(pid); + } + if exited != Some(false) { + match child.try_wait() { + Ok(Some(status)) => break status.success().then_some(()).ok_or(format!("the worker ended: {status}")), + Ok(None) => {} + Err(e) => break Err(format!("the worker could not be watched: {e}")), + } } if let Some(rss) = rss_bytes(pid) { peak = peak.max(rss); @@ -481,9 +496,13 @@ impl Service { return (Err(if detail.is_empty() { e } else { format!("{e}: {}", detail.lines().last().unwrap_or(detail)) }), peak); } match serde_json::from_slice::(&out) { - Ok(v) => match v.get("error").and_then(Value::as_str) { + Ok(mut v) => match v.get("error").and_then(Value::as_str) { Some(e) => (Err(e.to_string()), peak), - None => (Ok(v), peak), + None => { + // Inside the result, so the receipt seals the figure and its class. + v["energy"] = energy::figure(reading.as_ref(), plan.method == "statevector_single", energy::UNMETERED_HERE); + (Ok(v), peak) + } }, Err(e) => (Err(format!("the worker's output did not parse: {e}")), peak), } diff --git a/wai/wai-quantum-service/src/main.rs b/wai/wai-quantum-service/src/main.rs index ca6129ab9..05d98f35c 100644 --- a/wai/wai-quantum-service/src/main.rs +++ b/wai/wai-quantum-service/src/main.rs @@ -15,6 +15,7 @@ mod catalog; mod discovery; +mod energy; mod http; mod jobs; mod mcp; diff --git a/wai/wai-quantum-service/src/mcp.rs b/wai/wai-quantum-service/src/mcp.rs index 5906d62dd..0cb4205e2 100644 --- a/wai/wai-quantum-service/src/mcp.rs +++ b/wai/wai-quantum-service/src/mcp.rs @@ -145,6 +145,18 @@ pub fn summary(job: &Value) -> String { let parts: Vec = am.iter().map(|a| format!("{}: {:.6}{:+.6}i", compact_bits(a["bits"].as_str().unwrap_or("?")), a["re"].as_f64().unwrap_or(f64::NAN), a["im"].as_f64().unwrap_or(f64::NAN))).collect(); s.push(format!("amplitudes: {}", parts.join("; "))); } + let e = &r["energy"]; + if let Some(uj) = e["joules_micro"].as_u64() { + let gaps = e["excludes"].as_array().map(|x| x.iter().filter_map(Value::as_str).collect::>().join(", ")).unwrap_or_default(); + s.push(format!( + "energy: {:.6} J, the CPU energy of the job's process ({} reading, {} coverage; not counted: {gaps})", + uj as f64 / 1e6, + e["energy_provenance"].as_str().unwrap_or("?"), + e["coverage"].as_str().unwrap_or("?"), + )); + } else if let Some(why) = e["unmetered"].as_str() { + s.push(format!("energy: unmetered ({why})")); + } if let Some(ms) = r["elapsed_ms"].as_u64() { s.push(format!("ran in {ms} ms; signed receipt attached (verify_quantum_receipt checks it)")); }