jcp-gateway: meter each call on the upstream server's own process tree #198
17 changed files with 4408 additions and 115 deletions
19
.github/workflows/standards-build.yml
vendored
19
.github/workflows/standards-build.yml
vendored
|
|
@ -400,3 +400,22 @@ jobs:
|
|||
log=$(mktemp)
|
||||
cargo test -p sandbox-meter -- --nocapture 2>&1 | tee "$log"
|
||||
grep -q 'ioreport liveness:' "$log"
|
||||
# The tree meter's real-process tests (tests/tree_energy.rs) are
|
||||
# gated to Apple Silicon macOS too; each prints this line.
|
||||
grep -q 'tree meter:' "$log"
|
||||
|
||||
# jcp-gateway meters each gated call on the upstream server's own
|
||||
# process tree with that meter. The jcp-rs matrix entry runs on Linux,
|
||||
# where there is no tree meter and every call is metered machine-wide,
|
||||
# so its real-process test (tests/upstream_tree.rs, Apple Silicon macOS
|
||||
# only) runs nowhere else. Each call it meters on a real server's tree
|
||||
# prints this line (the machine-wide fallback test prints another), so
|
||||
# the grep fails if a cfg or a filter left none of them to run.
|
||||
- name: jcp-gateway (upstream process tree)
|
||||
shell: bash
|
||||
working-directory: jcp/jcp-rs
|
||||
run: |
|
||||
set -o pipefail
|
||||
log=$(mktemp)
|
||||
cargo test -p jcp-gateway -- --nocapture 2>&1 | tee "$log"
|
||||
grep -q 'upstream tree:' "$log"
|
||||
|
|
|
|||
|
|
@ -46,7 +46,7 @@ jcp/
|
|||
- **`jcp-intercept`** — **attested interceptor chains**: a composable, ordered pipeline of transforms *around* each model and tool call — the `before_model` / `after_model` / `before_tool` seams an agent's middleware stack runs, composed onion-order (forward in, reverse out) and stopping at the first refusal or human-approval hold. An interceptor may inspect or **rewrite** the request (redact a secret before it reaches the model, trim stale context), **refuse** it, or **hold** it for a human. The difference from the incumbent's arbitrary in-process hooks: every step seals an **`intercept/1`** JCR-1 receipt carrying the request digest **before and after** the step (so a redaction or a trim is provable — the content itself never sealed, only its hash), the seam, the outcome, and the **metered joules** at their honest provenance, hash-chained — the exact transform pipeline that produced a model request becomes reconstructable and verifiable by anyone, and a step cannot run un-sealed. Three reference interceptors — `Redact` (a content transform), `CallLimit` (a guardrail), `Approve` (a human-in-the-loop hold) — each reuse a primitive the family already holds rather than reinventing it.
|
||||
- **`jcp-efficiency`** — the **energy-efficiency spec**, surfaced as a sealed receipt. The dark-factory economics turn on four figures — **joules per accepted token, tokens per second, value per token, and the tokens/joules/dollars spent per COMPLETED, VERIFIED unit of work** — and the incumbents publish, at best, a *rate* (cost per token) that can't tell you whether the spend bought a result. An **`efficiency/1`** JCR-1 receipt records all four at honest provenance, hash-chained, with the one thing the rates omit: the **outcome anchor** (`completed` + a verdict). A measurement that didn't complete is spend that bought nothing, and the receipt says so. `aggregate` then splits value-delivered from total spend across a chain — charging *every* attempt (including the failed ones) against the results actually delivered — so `tokens_per_completed_unit` / `uj_per_completed_unit` / `value_per_joule` become the honest efficiency a "cost per token" number is designed to hide. Integers and strings only (JCR-1 forbids bools/floats); the rate figures recompute from the sealed inputs on read.
|
||||
- **`jcp-energy`** — the metering ladder (SPEC §5.1): RAPL package counters (Linux powercap, wraparound-corrected, `HwShunt`, only while every package counter is live — `rapl_status()`), NVML GPU energy behind the `nvml` feature (`HwShunt` and `Partial`, since it does not read the CPU's share; only while live — `nvml_status()`), the macOS kernel's per-process CPU energy count (`proc_pid_rusage`'s `ri_energy_nj`; `HwShunt`, whole for in-process CPU work, only while live — `process_energy_status()`; it covers this process's threads, not children, the GPU or DRAM, so a span that reaps a child process or ends with one is `Partial`; it is live on macOS 27.0.1 where IOReport's CPU rails are not), Apple Silicon IOReport (`HwShunt`, only while every CPU energy rail, each die's on a multi-die part, is live — on macOS 27.0.1 that rail is republished in lumps, so there the rung is declined and `ioreport_status()` says why; `Meter::detect_explained()` names the rung selected and why each one above it was declined), a time×calibrated-power model (`ModelBased`, always available), and a constant per-op estimator (`Estimator`). A vendor counter's reading is real but vendor-opaque and relayed by the platform, and it counts as a reading only while the counter is live (SPEC §5). Measured when the platform exposes it, estimated when it doesn't — and the provenance always says which; a hardware backend that fails mid-reading degrades honestly to the model.
|
||||
- **`jcp-gateway`** — the **energy-native MCP gateway**: a JCP enforcement point that sits in front of an MCP server so an existing client/server pair keeps its transport while every `tools/call` becomes capability-scoped, joule-budgeted, and sealed. The novel mechanic is two-phase metered enforcement: the call is *authorized* before it runs (no debit, because the cost isn't yet known), the upstream call is **metered** by the sibling Sandbox standard's energy probe (Apple Silicon IOReport, RAPL, NVML — vendor energy counters, whose readings are real but vendor-opaque and relayed by the platform, and which count as readings only while the counter is live for the work, SPEC §5; otherwise the figure is an estimate and is graded as one), and the metered joules are debited from the grant — so the budget that stops a runaway agent is denominated in joules, not a token estimate. Every decision, allow and deny, seals into a **hash-chained receipt log**: excise, reorder, or splice a receipt and the chain breaks at that point. A denied call never reaches the server (the live 2025–26 MCP exploits land on tool *output*, which the gateway frames as untrusted `data` that can never exercise a capability), and the `jcp-gateway` binary runs it over stdio in front of any MCP-server subprocess. This is the chokepoint the AI-gateway field occupies, on the two axes nobody there holds: the unit of account is the joule, and the audit is cryptographic and tamper-evident rather than a dollar figure and a span.
|
||||
- **`jcp-gateway`** — the **energy-native MCP gateway**: a JCP enforcement point that sits in front of an MCP server so an existing client/server pair keeps its transport while every `tools/call` becomes capability-scoped, joule-budgeted, and sealed. The novel mechanic is two-phase metered enforcement: the call is *authorized* before it runs (no debit, because the cost isn't yet known), the upstream call is **metered** by the sibling Sandbox standard's energy probe (Apple Silicon IOReport, RAPL, NVML — vendor energy counters, whose readings are real but vendor-opaque and relayed by the platform, and which count as readings only while the counter is live for the work, SPEC §5; otherwise the figure is an estimate and is graded as one), and the metered joules are debited from the grant — so the budget that stops a runaway agent is denominated in joules, not a token estimate. Where the upstream server is a local process whose CPU energy the host's kernel counts live (macOS `proc_pid_rusage`), each call is read on that server's own process tree, the work's own process rather than the machine (SPEC §5.1, rung 3): sealed `HwShunt`, and `Whole` only when nothing was seen missing from the tree (no descendant reaped or exited in the span, no GPU client held, no in-process Neural Engine memory, every member read and quiet at both ends) and the operator has declared, with `--upstream-self-contained`, that the server hands no work to a process outside its tree; otherwise `Partial`, with the response's `_meta` naming why. Every decision, allow and deny, seals into a **hash-chained receipt log**: excise, reorder, or splice a receipt and the chain breaks at that point. A denied call never reaches the server (the live 2025–26 MCP exploits land on tool *output*, which the gateway frames as untrusted `data` that can never exercise a capability), and the `jcp-gateway` binary runs it over stdio in front of any MCP-server subprocess. This is the chokepoint the AI-gateway field occupies, on the two axes nobody there holds: the unit of account is the joule, and the audit is cryptographic and tamper-evident rather than a dollar figure and a span.
|
||||
- **`jcp-wire`** — the `jcp/1` wire binding (SPEC §11): 4-byte length prefix + canonical det-CBOR over any byte stream, with TCP and Unix-socket endpoints and a client. The decoder rejects non-canonical bodies, so a message has exactly one wire form. Requests carry the full delegation chain and are **caller-signed** — the endpoint authenticates from the request itself and denies an impersonation `bad-signature`, sealed like any other decision. Responses carry the deterministic decision and the sealed receipt.
|
||||
- **`jcp-threshold`** — threshold actors (SPEC §2.2): a `did:key` whose private key is split k-of-n with FROST (RFC 9591); the group key is an ordinary Ed25519 key, so a 2-of-3 team issues, holds, and signs wire requests with nothing else in JCP changing. Shares come from a trusted dealer or **distributed key generation** — with DKG the group's private key never exists anywhere, not even momentarily. The two-agent rule for irreversible actions, refused by arithmetic rather than policy.
|
||||
- **`jcp-channel`** — receipt-settled budget channels (SPEC §4.5, the OPTIONAL profile): N enforcement points spend optimistically against one grant and settle by exchanging signed statements of cumulative spend — forged, stale, and regressing statements refused; between settlements the exposure is bounded by the window, and after settlement any overdraft is provable from the signed statements. Where bounded overdraft is unacceptable, `shard()` remains the baseline.
|
||||
|
|
|
|||
73
jcp/SPEC.md
73
jcp/SPEC.md
|
|
@ -625,6 +625,79 @@ Recommended ladder, best first:
|
|||
SoC-wide counter because it attributes the work's own CPU energy rather
|
||||
than the machine's. A span over which it did not advance is metered by a
|
||||
lower rung.
|
||||
|
||||
The same count, read for every process of ANOTHER process tree the
|
||||
metering process can read — the process the work runs in, such as the
|
||||
server a gateway relays a call to, and its descendants — meters that
|
||||
work → `HwShunt`. Its domain is the CPU energy of the tree's processes
|
||||
over the span (as above, not DRAM): the advance of every process present
|
||||
at both ends, plus the whole count of every process born inside the
|
||||
span. A process is identified by its pid and its start time, so a reused
|
||||
pid is another process. The count is folded only when a thread leaves a
|
||||
core, so at each end of the span a member's count MUST be read with none
|
||||
of its threads on a core or queued for one, or the member is a gap
|
||||
(below). A walk of the tree reads a parent before it lists and reads
|
||||
the parent's children, so a child that exits and is reaped in between
|
||||
is in that walk neither as a member nor in its parent's reaped time;
|
||||
the walk at each end MUST therefore be confirmed by another, one after
|
||||
the other, that agrees with it on the tree's processes, each one's
|
||||
parent, reaped-descendant CPU time and exit time, with no listed child
|
||||
gone by its read, or that end is a gap (below). The span SHOULD end
|
||||
after the work's reply. The figure is all the tree spent inside the
|
||||
span, whatever for: work the process defers past an earlier reply, or
|
||||
does on a timer of its own, is read in whichever span is open when it
|
||||
runs, and work after this span's stop is in no figure of this span. So
|
||||
`Whole` says that nothing was seen missing from the tree's energy over
|
||||
the span, not that all of it was spent on the span's own work. The
|
||||
figure is `Partial` (§5.2) when the meter sees any of these over the
|
||||
span; each is energy of the tree's over the span that the figure does
|
||||
not hold, but for the one noted:
|
||||
- a descendant reaped inside the span (the reaper's reaped-descendant
|
||||
CPU time moved; a reaped process's energy is in no counter), unless
|
||||
it had already exited at the start, its energy then all spent before
|
||||
the span;
|
||||
- a member that exited inside the span: its record is final and is
|
||||
counted, but it lacks the energy of the member's last time on a core
|
||||
(an exiting thread's last slice is never folded into it: up to all of
|
||||
a short-lived process's energy), and its GPU time is lost at exit, so
|
||||
its GPU use is unknown;
|
||||
- a member present at the start and gone at the stop (reaped, or moved
|
||||
out of the tree): what it spent in the span before it went is not in
|
||||
the figure;
|
||||
- a member present at the stop that started before the span and was not
|
||||
read at its start: its energy from before the span cannot be
|
||||
separated, so none of it is counted;
|
||||
- a member that could not be read (a process of another user), or
|
||||
whose children could not be listed;
|
||||
- a member still running at an end: at the stop its reading lacks up to
|
||||
a scheduler quantum of energy per running thread; at the START its
|
||||
reading is short by as much, so the figure may hold up to that much
|
||||
energy spent just BEFORE the span. This one can make the figure
|
||||
exceed the tree's energy over the span, so a `Partial` tree figure
|
||||
with it is not a lower bound, and an implementation that names the
|
||||
gaps beside the figure SHOULD name the end (the reference gateway:
|
||||
`unsettled-at-start`, `unsettled-at-stop`);
|
||||
- an end whose walk no agreeing walk confirmed before the meter stopped
|
||||
waiting: the tree kept changing, and a process that came and went may
|
||||
be missing;
|
||||
- a member holding a GPU client at either end: GPU time its client
|
||||
recorded, or GPU use the reader cannot rule out, because a GPU
|
||||
client's recorded time is per live command queue and a queue
|
||||
created and released inside the span takes its time with it; or GPU
|
||||
time that could not be read;
|
||||
- a member that has held Neural Engine model memory in-process.
|
||||
|
||||
An implementation MUST NOT grade a tree's figure `Whole` on an empty
|
||||
list alone. `Whole` further requires the operator's declaration that the
|
||||
process hands no work to a process outside its tree — a system service,
|
||||
on-device model inference run by a system daemon, a database server, a
|
||||
remote API — because a reader of the tree cannot see such delegation:
|
||||
that work's energy is charged to the process that does it, which may be
|
||||
another user's and unreadable. Without the declaration the figure is
|
||||
`Partial`. A span whose root is gone or unreadable at either end, or
|
||||
whose figure did not advance, is metered by a lower rung. The reference
|
||||
gateway (`jcp-gateway`) meters each gated call this way where the
|
||||
upstream server is a local process it can read.
|
||||
4. **Apple IOReport** (the Energy Model group) — the SoC energy counters
|
||||
→ `HwShunt`, selected only while every CPU energy rail (one per die on a
|
||||
multi-die part) is live by the rule above. A span over which no CPU rail
|
||||
|
|
|
|||
1
jcp/jcp-rs/Cargo.lock
generated
1
jcp/jcp-rs/Cargo.lock
generated
|
|
@ -629,6 +629,7 @@ dependencies = [
|
|||
"jcp-receipt",
|
||||
"jcp-runtime",
|
||||
"jcp-schema",
|
||||
"libc",
|
||||
"sandbox-meter",
|
||||
"sandbox-schema",
|
||||
"serde_json",
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
[package]
|
||||
name = "jcp-gateway"
|
||||
description = "The energy-native MCP gateway: a JCP enforcement point in front of an MCP server. Every tools/call is gated by a capability grant, the upstream call is metered (a hardware reading where the host exposes a live energy counter, an estimate where it does not, each sealed as what it is), the metered joules are debited from the grant's budget, and the decision seals into a hash-chained receipt log. The chokepoint position the AI-gateway field occupies — denominated in joules, with a tamper-evident audit, not dollars and a span."
|
||||
description = "The energy-native MCP gateway: a JCP enforcement point in front of an MCP server. Every tools/call is gated by a capability grant, the upstream call is metered (on the upstream server's own process tree where the host's kernel counts per-process CPU energy, and otherwise machine-wide: a hardware reading where the host exposes a live energy counter, an estimate where it does not, each sealed as what it is), the metered joules are debited from the grant's budget, and the decision seals into a hash-chained receipt log. The chokepoint position the AI-gateway field occupies — denominated in joules, with a tamper-evident audit, not dollars and a span."
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
|
@ -21,6 +21,11 @@ sandbox-schema = { path = "../../../../sandbox/sandbox-rs/crates/sandbox-schema"
|
|||
ed25519-dalek = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
# tests/upstream_tree.rs's oracle reads the upstream server's energy count
|
||||
# with proc_pid_rusage into a raw buffer, sharing no code with the meter.
|
||||
libc = "0.2"
|
||||
|
||||
[[bin]]
|
||||
name = "jcp-gateway"
|
||||
path = "src/bin/jcp_gateway.rs"
|
||||
|
|
|
|||
|
|
@ -2,8 +2,8 @@
|
|||
//! MCP server subprocess.
|
||||
//!
|
||||
//! ```bash
|
||||
//! jcp-gateway --server <name> --grant <tool>=<budget_uj> [--grant …] -- \
|
||||
//! <mcp-server-cmd> [args…]
|
||||
//! jcp-gateway --server <name> --grant <tool>=<budget_uj> [--grant …] \
|
||||
//! [--upstream-self-contained] -- <mcp-server-cmd> [args…]
|
||||
//! ```
|
||||
//!
|
||||
//! The gateway speaks newline-delimited JSON-RPC with its own client on
|
||||
|
|
@ -11,11 +11,29 @@
|
|||
//! `tools/call` is gated, metered, debited, and sealed; receipts stream to
|
||||
//! `JCP_GATEWAY_RECEIPTS` (default `./jcp-gateway-receipts.cose`, length-
|
||||
//! prefixed envelopes) and a one-line summary per call goes to stderr.
|
||||
//!
|
||||
//! What the server writes that is not a response — a notification, or a
|
||||
//! request of its own in the middle of a call — goes straight to the
|
||||
//! client, and the client's answer to such a request straight back to the
|
||||
//! server, while the call stays open. A call ends at the response carrying
|
||||
//! its own id, and its span with it. A client notification is sent on
|
||||
//! without waiting for anything back.
|
||||
//!
|
||||
//! Each call is metered on the server's own process tree where this host's
|
||||
//! kernel counts per-process CPU energy live (macOS), and machine-wide
|
||||
//! otherwise; the startup line says which. `--upstream-self-contained` is
|
||||
//! the operator's declaration that the server hands no work to a process
|
||||
//! outside its tree (no system service, no model inference in a system
|
||||
//! daemon, no database server, no remote API): the meter cannot see that,
|
||||
//! and without the declaration every reading of the tree is sealed
|
||||
//! `Partial`.
|
||||
|
||||
use ed25519_dalek::SigningKey;
|
||||
use jcp_gateway::{Gateway, Relayed, Upstream};
|
||||
use jcp_gateway::{is_response, message_id, Gateway, Relayed, Upstream};
|
||||
use std::io::{BufRead, BufReader, Write};
|
||||
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
|
||||
use std::process::{Child, ChildStdin, Command, Stdio};
|
||||
use std::sync::mpsc::{self, Receiver};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
/// The energy grade a verified receipt seals: its `energy_provenance`, plus
|
||||
/// `energy_coverage` when that is `Partial`. `None` if it does not verify.
|
||||
|
|
@ -29,21 +47,85 @@ fn sealed_grade(envelope: &[u8], steward: &str, pubkey: &[u8; 32]) -> Option<Str
|
|||
})
|
||||
}
|
||||
|
||||
/// One line to the client, whole and flushed: the relay loop and the
|
||||
/// server's reader both write to it.
|
||||
fn to_client(line: &str) -> std::io::Result<()> {
|
||||
let mut out = std::io::stdout().lock();
|
||||
writeln!(out, "{line}")?;
|
||||
out.flush()
|
||||
}
|
||||
|
||||
/// One line to the server, whole and flushed: the relay loop and the
|
||||
/// client's reader both write to it.
|
||||
fn to_server(stdin: &Mutex<ChildStdin>, line: &str) -> std::io::Result<()> {
|
||||
let mut w = stdin.lock().unwrap_or_else(|e| e.into_inner());
|
||||
w.write_all(line.as_bytes())?;
|
||||
w.write_all(b"\n")?;
|
||||
w.flush()
|
||||
}
|
||||
|
||||
/// The upstream MCP server as a child process speaking JSON-RPC over stdio.
|
||||
///
|
||||
/// A server writes more than responses on its stdout: in the middle of a
|
||||
/// call it may send notifications (`notifications/progress`,
|
||||
/// `notifications/message`) and requests of its own (`sampling/createMessage`)
|
||||
/// before the call's response. So a thread reads the server's stdout and
|
||||
/// routes each line: a response (an `id` and no `method`) to the call
|
||||
/// waiting for it, everything else straight to the client. A call returns
|
||||
/// the response carrying its own request's id — the line its metered span
|
||||
/// ends at — and hands any other response to the client as written.
|
||||
struct ChildServer {
|
||||
_child: Child,
|
||||
stdin: ChildStdin,
|
||||
stdout: BufReader<ChildStdout>,
|
||||
/// Held, and never waited on while the gateway serves: an unreaped child
|
||||
/// keeps its pid, so the pid the meter reads stays this server's.
|
||||
child: Child,
|
||||
stdin: Arc<Mutex<ChildStdin>>,
|
||||
/// The server's responses, in the order it wrote them; closed when its
|
||||
/// stdout closes.
|
||||
responses: Receiver<String>,
|
||||
}
|
||||
|
||||
impl ChildServer {
|
||||
fn new(mut child: Child) -> ChildServer {
|
||||
let stdin = Arc::new(Mutex::new(child.stdin.take().expect("child stdin")));
|
||||
let stdout = BufReader::new(child.stdout.take().expect("child stdout"));
|
||||
let (tx, responses) = mpsc::channel();
|
||||
std::thread::spawn(move || {
|
||||
for line in stdout.lines() {
|
||||
let Ok(line) = line else { break };
|
||||
if is_response(&line) {
|
||||
if tx.send(line).is_err() {
|
||||
break;
|
||||
}
|
||||
} else if to_client(&line).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
ChildServer { child, stdin, responses }
|
||||
}
|
||||
}
|
||||
|
||||
impl Upstream for ChildServer {
|
||||
fn call(&mut self, request_line: &str) -> std::io::Result<String> {
|
||||
self.stdin.write_all(request_line.as_bytes())?;
|
||||
self.stdin.write_all(b"\n")?;
|
||||
self.stdin.flush()?;
|
||||
let mut line = String::new();
|
||||
self.stdout.read_line(&mut line)?;
|
||||
Ok(line.trim_end().to_string())
|
||||
to_server(&self.stdin, request_line)?;
|
||||
let id = message_id(request_line);
|
||||
loop {
|
||||
// The server closed its stdout: no response is coming.
|
||||
let Ok(line) = self.responses.recv() else { return Ok(String::new()) };
|
||||
match &id {
|
||||
Some(id) if message_id(&line).as_ref() != Some(id) => to_client(&line)?,
|
||||
_ => return Ok(line),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Written and flushed, and nothing read: the server sends no reply.
|
||||
fn notify(&mut self, line: &str) -> std::io::Result<()> {
|
||||
to_server(&self.stdin, line)
|
||||
}
|
||||
|
||||
fn pid(&self) -> Option<u32> {
|
||||
Some(self.child.id())
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -52,6 +134,7 @@ fn main() -> std::io::Result<()> {
|
|||
let mut server = "mcp".to_string();
|
||||
let mut grants: Vec<(String, u64)> = Vec::new();
|
||||
let mut cmd: Vec<String> = Vec::new();
|
||||
let mut self_contained = false;
|
||||
let mut i = 1;
|
||||
while i < args.len() {
|
||||
match args[i].as_str() {
|
||||
|
|
@ -59,6 +142,10 @@ fn main() -> std::io::Result<()> {
|
|||
server = args.get(i + 1).cloned().unwrap_or_default();
|
||||
i += 2;
|
||||
}
|
||||
"--upstream-self-contained" => {
|
||||
self_contained = true;
|
||||
i += 1;
|
||||
}
|
||||
"--grant" => {
|
||||
if let Some((t, b)) = args.get(i + 1).and_then(|s| s.split_once('=')) {
|
||||
if let Ok(budget) = b.parse() {
|
||||
|
|
@ -75,43 +162,67 @@ fn main() -> std::io::Result<()> {
|
|||
}
|
||||
}
|
||||
if cmd.is_empty() {
|
||||
eprintln!("usage: jcp-gateway --server <name> --grant <tool>=<uj> … -- <server-cmd> …");
|
||||
eprintln!(
|
||||
"usage: jcp-gateway --server <name> --grant <tool>=<uj> … [--upstream-self-contained] -- <server-cmd> …"
|
||||
);
|
||||
std::process::exit(2);
|
||||
}
|
||||
|
||||
let mut child = Command::new(&cmd[0])
|
||||
let child = Command::new(&cmd[0])
|
||||
.args(&cmd[1..])
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.spawn()?;
|
||||
let mut up = ChildServer {
|
||||
stdin: child.stdin.take().expect("child stdin"),
|
||||
stdout: BufReader::new(child.stdout.take().expect("child stdout")),
|
||||
_child: child,
|
||||
};
|
||||
let mut up = ChildServer::new(child);
|
||||
|
||||
let grant_refs: Vec<(&str, u64)> = grants.iter().map(|(t, b)| (t.as_str(), *b)).collect();
|
||||
let steward_sk = SigningKey::from_bytes(&[0x5e; 32]);
|
||||
let steward_pk = steward_sk.verifying_key().to_bytes();
|
||||
let steward = format!("did:jcp:gateway:{server}");
|
||||
let mut gw = Gateway::new(&server, &grant_refs, steward_sk);
|
||||
if self_contained {
|
||||
gw.declare_upstream_self_contained();
|
||||
}
|
||||
// Which meter reads each call, decided now: building the tree meter pays
|
||||
// the per-process counter's liveness probe once, here rather than on the
|
||||
// first call. Each call's own grade is on its line below, read back from
|
||||
// its sealed receipt.
|
||||
let call_meter = gw.call_meter(&up);
|
||||
eprintln!(
|
||||
"jcp-gateway: server={server} tools={:?} metering={}",
|
||||
"jcp-gateway: server={server} tools={:?} metering={call_meter} upstream-self-contained={}",
|
||||
gw.tools(),
|
||||
if gw.metering_is_hardware() { "hardware" } else { "estimator" }
|
||||
if gw.upstream_self_contained() { "declared" } else { "not declared" },
|
||||
);
|
||||
|
||||
let receipts_path =
|
||||
std::env::var("JCP_GATEWAY_RECEIPTS").unwrap_or_else(|_| "./jcp-gateway-receipts.cose".into());
|
||||
|
||||
let stdin = std::io::stdin();
|
||||
let mut stdout = std::io::stdout();
|
||||
let mut prev_receipts = 0usize;
|
||||
for line in stdin.lock().lines() {
|
||||
let line = line?;
|
||||
if line.trim().is_empty() {
|
||||
continue;
|
||||
// The client's lines are read on a thread of their own. A response of
|
||||
// the client's (an `id` and no `method`), answering a request the server
|
||||
// made in the middle of a call, goes straight to the server: the relay
|
||||
// loop below is then waiting on that very call, and the server will not
|
||||
// answer the call until it has its answer. Every other line is relayed,
|
||||
// in order, by the loop.
|
||||
let (tx, client_lines) = mpsc::channel::<String>();
|
||||
let server_stdin = Arc::clone(&up.stdin);
|
||||
std::thread::spawn(move || {
|
||||
for line in std::io::stdin().lock().lines() {
|
||||
let Ok(line) = line else { break };
|
||||
if line.trim().is_empty() {
|
||||
continue;
|
||||
}
|
||||
if is_response(&line) {
|
||||
if to_server(&server_stdin, &line).is_err() {
|
||||
break;
|
||||
}
|
||||
} else if tx.send(line).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let mut prev_receipts = 0usize;
|
||||
for line in client_lines {
|
||||
match gw.relay(&line, &mut up)? {
|
||||
Relayed::Allowed { response, cost_uj, budget_remaining_uj } => {
|
||||
// How the joules were obtained, read back from the receipt the
|
||||
|
|
@ -123,19 +234,19 @@ fn main() -> std::io::Result<()> {
|
|||
.unwrap_or_else(|| "unverified".into());
|
||||
eprintln!("jcp-gateway: allowed, {cost_uj} µJ [{grade}], {budget_remaining_uj} µJ left");
|
||||
if !response.is_empty() {
|
||||
writeln!(stdout, "{response}")?;
|
||||
stdout.flush()?;
|
||||
to_client(&response)?;
|
||||
}
|
||||
}
|
||||
Relayed::Denied { response, reason } => {
|
||||
eprintln!("jcp-gateway: DENIED ({reason})");
|
||||
writeln!(stdout, "{response}")?;
|
||||
stdout.flush()?;
|
||||
// Empty for a call sent as a notification: it gets no reply.
|
||||
if !response.is_empty() {
|
||||
to_client(&response)?;
|
||||
}
|
||||
}
|
||||
Relayed::PassedThrough(resp) => {
|
||||
if !resp.is_empty() {
|
||||
writeln!(stdout, "{resp}")?;
|
||||
stdout.flush()?;
|
||||
to_client(&resp)?;
|
||||
}
|
||||
}
|
||||
Relayed::Notified => {}
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
192
jcp/jcp-rs/crates/jcp-gateway/tests/stdio_binary.rs
Normal file
192
jcp/jcp-rs/crates/jcp-gateway/tests/stdio_binary.rs
Normal file
|
|
@ -0,0 +1,192 @@
|
|||
//! The `jcp-gateway` binary itself, over stdio, in front of a real server
|
||||
//! process: the way an MCP client runs it. Each test drives the binary with
|
||||
//! newline-delimited JSON-RPC on its stdin and reads its stdout on a thread
|
||||
//! of its own, so a gateway that stops answering fails the test at a
|
||||
//! deadline instead of hanging it.
|
||||
//!
|
||||
//! The server is awk, answering only the lines that carry an `id`, as an MCP
|
||||
//! server answers requests and never notifications. It runs on every host
|
||||
//! this crate builds on.
|
||||
|
||||
use serde_json::{json, Value};
|
||||
use std::io::{BufRead, BufReader, Read, Write};
|
||||
use std::process::{Child, ChildStdin, Command, Stdio};
|
||||
use std::sync::mpsc::{self, Receiver};
|
||||
use std::time::Duration;
|
||||
|
||||
/// How long a test waits for one line from the gateway: far past anything a
|
||||
/// call takes, on a host at a load of 14 to 260.
|
||||
const LINE_DEADLINE: Duration = Duration::from_secs(120);
|
||||
|
||||
/// An MCP-ish server in awk: a request (a line with an `id`) gets one
|
||||
/// response carrying that id; a line without one gets nothing.
|
||||
const ANSWERS_REQUESTS_ONLY: &str = r#"/"id"/ {
|
||||
id = "null"
|
||||
if (match($0, /"id":[0-9]+/)) id = substr($0, RSTART + 5, RLENGTH - 5)
|
||||
if ($0 ~ /"method":"initialize"/) printf "{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":{\"protocolVersion\":\"2025-06-18\",\"capabilities\":{}}}\n", id
|
||||
else printf "{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"done\"}]}}\n", id
|
||||
fflush()
|
||||
}"#;
|
||||
|
||||
/// The gateway binary, running, with its server.
|
||||
struct Gateway {
|
||||
child: Child,
|
||||
stdin: ChildStdin,
|
||||
lines: Receiver<String>,
|
||||
stderr: Receiver<String>,
|
||||
}
|
||||
|
||||
impl Gateway {
|
||||
fn spawn(name: &str, server: &[&str], extra: &[&str]) -> Gateway {
|
||||
let receipts = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"))
|
||||
.join(format!("jcp-gateway-{name}-{}.cose", std::process::id()));
|
||||
let _ = std::fs::remove_file(&receipts);
|
||||
let mut child = Command::new(env!("CARGO_BIN_EXE_jcp-gateway"))
|
||||
.args(["--server", name, "--grant", "echo=1000000000"])
|
||||
.args(extra)
|
||||
.arg("--")
|
||||
.args(server)
|
||||
.env("JCP_GATEWAY_RECEIPTS", &receipts)
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
.spawn()
|
||||
.expect("spawn the jcp-gateway binary");
|
||||
let stdin = child.stdin.take().unwrap();
|
||||
let (tx, lines) = mpsc::channel();
|
||||
let out = child.stdout.take().unwrap();
|
||||
std::thread::spawn(move || {
|
||||
for line in BufReader::new(out).lines() {
|
||||
let Ok(line) = line else { break };
|
||||
if tx.send(line).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
let (etx, stderr) = mpsc::channel();
|
||||
let mut err = child.stderr.take().unwrap();
|
||||
std::thread::spawn(move || {
|
||||
let mut all = String::new();
|
||||
let _ = err.read_to_string(&mut all);
|
||||
let _ = etx.send(all);
|
||||
});
|
||||
Gateway { child, stdin, lines, stderr }
|
||||
}
|
||||
|
||||
fn send(&mut self, msg: &Value) {
|
||||
writeln!(self.stdin, "{msg}").expect("write to the gateway");
|
||||
self.stdin.flush().expect("flush to the gateway");
|
||||
}
|
||||
|
||||
/// The next line the gateway writes, parsed; fails the test, with the
|
||||
/// gateway's stderr, if none comes by the deadline.
|
||||
fn next(&mut self, waiting_for: &str) -> Value {
|
||||
match self.lines.recv_timeout(LINE_DEADLINE) {
|
||||
Ok(line) => serde_json::from_str(&line).unwrap_or_else(|e| panic!("not JSON ({e}): {line}")),
|
||||
Err(_) => {
|
||||
let _ = self.child.kill();
|
||||
let _ = self.child.wait();
|
||||
let stderr = self.stderr.recv_timeout(Duration::from_secs(5)).unwrap_or_default();
|
||||
panic!("no line from the gateway while waiting for {waiting_for}; its stderr:\n{stderr}")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Gateway {
|
||||
fn drop(&mut self) {
|
||||
// Its server reads the gateway's pipe, so it sees EOF and exits too.
|
||||
let _ = self.child.kill();
|
||||
let _ = self.child.wait();
|
||||
}
|
||||
}
|
||||
|
||||
fn call(id: i64) -> Value {
|
||||
json!({"jsonrpc":"2.0","id":id,"method":"tools/call","params":{"name":"echo","arguments":{}}})
|
||||
}
|
||||
|
||||
/// The handshake every MCP client opens with — `initialize`, then the
|
||||
/// `notifications/initialized` notification, which the server answers with
|
||||
/// nothing — and then calls. The gateway must send the notification on
|
||||
/// without waiting for a reply, or it never reads another line.
|
||||
#[test]
|
||||
fn the_binary_serves_calls_after_the_initialized_notification() {
|
||||
let mut g = Gateway::spawn("smoke", &["/usr/bin/awk", ANSWERS_REQUESTS_ONLY], &[]);
|
||||
g.send(&json!({"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{}}}));
|
||||
let init = g.next("the initialize response");
|
||||
assert_eq!(init["id"], json!(1), "{init}");
|
||||
assert!(init["result"]["capabilities"]["extensions"]["science.transaction/jcp"].is_object(), "{init}");
|
||||
g.send(&json!({"jsonrpc":"2.0","method":"notifications/initialized"}));
|
||||
for id in 2..5 {
|
||||
g.send(&call(id));
|
||||
let r = g.next(&format!("the response to call {id}"));
|
||||
assert_eq!(r["id"], json!(id), "{r}");
|
||||
assert_eq!(r["result"]["content"][0]["text"], json!("done"), "{r}");
|
||||
let meta = &r["result"]["_meta"]["science.transaction/jcp"];
|
||||
assert!(meta["receipt_b64"].is_string(), "each call carries its sealed receipt: {r}");
|
||||
eprintln!("stdio binary: call {id}: cost_uj {} [{}] domain {}", meta["cost_uj"], meta["energy_provenance"], meta["energy_domain"]);
|
||||
// A notification between calls is sent on, unanswered, too.
|
||||
g.send(&json!({"jsonrpc":"2.0","method":"notifications/cancelled","params":{"requestId":99}}));
|
||||
}
|
||||
}
|
||||
|
||||
/// An MCP-ish server in awk that, inside each `tools/call`, first sends the
|
||||
/// client a `notifications/progress` and a request of its own
|
||||
/// (`sampling/createMessage`, id `s<call id>`), waits for the client's
|
||||
/// answer, and only then sends the call's response, saying whether it got
|
||||
/// the answer. Every other request gets a plain response; notifications get
|
||||
/// nothing.
|
||||
const TALKS_MID_CALL: &str = r#"/"method":"tools\/call"/ {
|
||||
id = "null"
|
||||
if (match($0, /"id":[0-9]+/)) id = substr($0, RSTART + 5, RLENGTH - 5)
|
||||
printf "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\",\"params\":{\"progressToken\":%s,\"progress\":1}}\n", id
|
||||
printf "{\"jsonrpc\":\"2.0\",\"id\":\"s%s\",\"method\":\"sampling/createMessage\",\"params\":{\"maxTokens\":1}}\n", id
|
||||
fflush()
|
||||
answer = ""
|
||||
getline answer
|
||||
got = (answer ~ ("\"id\":\"s" id "\"") && answer ~ /"result"/) ? "sampled" : "unanswered"
|
||||
printf "{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"%s\"}]}}\n", id, got
|
||||
fflush()
|
||||
next
|
||||
}
|
||||
/"id"/ {
|
||||
id = "null"
|
||||
if (match($0, /"id":[0-9]+/)) id = substr($0, RSTART + 5, RLENGTH - 5)
|
||||
printf "{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":{\"protocolVersion\":\"2025-06-18\",\"capabilities\":{}}}\n", id
|
||||
fflush()
|
||||
}"#;
|
||||
|
||||
/// A server that talks in the middle of a call: the gateway hands its
|
||||
/// notification and its request to the client as they come, carries the
|
||||
/// client's answer back to the server while the call is still open, and
|
||||
/// returns, as the call's response, the line with the call's own id — so
|
||||
/// each call gets its own response, sealed, and the next call is not handed
|
||||
/// the previous one's. Under `--upstream-self-contained` the awk server is
|
||||
/// one process; whatever its figure is sealed as, it is the response with
|
||||
/// the call's id that ends its span.
|
||||
#[test]
|
||||
fn the_binary_relays_what_a_server_sends_in_the_middle_of_a_call() {
|
||||
let mut g = Gateway::spawn("chatty", &["/usr/bin/awk", TALKS_MID_CALL], &["--upstream-self-contained"]);
|
||||
g.send(&json!({"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{}}}));
|
||||
assert_eq!(g.next("the initialize response")["id"], json!(1));
|
||||
g.send(&json!({"jsonrpc":"2.0","method":"notifications/initialized"}));
|
||||
for id in 2..5 {
|
||||
g.send(&call(id));
|
||||
let progress = g.next(&format!("call {id}'s progress notification"));
|
||||
assert_eq!(progress["method"], json!("notifications/progress"), "{progress}");
|
||||
assert_eq!(progress["params"]["progressToken"], json!(id), "{progress}");
|
||||
let ask = g.next(&format!("call {id}'s sampling request"));
|
||||
assert_eq!(ask["method"], json!("sampling/createMessage"), "{ask}");
|
||||
let ask_id = ask["id"].clone();
|
||||
assert_eq!(ask_id, json!(format!("s{id}")), "{ask}");
|
||||
g.send(&json!({"jsonrpc":"2.0","id":ask_id,"result":{"role":"assistant","content":{"type":"text","text":"ok"}}}));
|
||||
let r = g.next(&format!("the response to call {id}"));
|
||||
assert_eq!(r["id"], json!(id), "each call gets its own response: {r}");
|
||||
assert_eq!(r["result"]["content"][0]["text"], json!("sampled"), "the server got the client's answer mid-call: {r}");
|
||||
let meta = &r["result"]["_meta"]["science.transaction/jcp"];
|
||||
assert!(meta["receipt_b64"].is_string(), "{r}");
|
||||
let because = meta["energy_partial_because"].as_array().cloned().unwrap_or_default();
|
||||
assert!(!because.contains(&json!("reply-not-the-response")), "the span ended at the call's own response: {meta}");
|
||||
eprintln!("stdio binary: chatty call {id}: cost_uj {} [{}] domain {} because {because:?}", meta["cost_uj"], meta["energy_provenance"], meta["energy_domain"]);
|
||||
}
|
||||
}
|
||||
498
jcp/jcp-rs/crates/jcp-gateway/tests/upstream_tree.rs
Normal file
498
jcp/jcp-rs/crates/jcp-gateway/tests/upstream_tree.rs
Normal file
|
|
@ -0,0 +1,498 @@
|
|||
//! The gateway metering real calls on the upstream server's own process
|
||||
//! tree. Each test spawns an MCP-ish server of its own (a process group of
|
||||
//! its own, killed at the end) that answers one JSON-RPC `tools/call` per
|
||||
//! line, burning the loop count its arguments name before it replies, and
|
||||
//! relays `tools/call`s to it through [`Gateway::relay`].
|
||||
//!
|
||||
//! The oracle shares no code with the meter or the gateway: the server's
|
||||
//! `proc_pid_rusage(pid, RUSAGE_INFO_V6)` read into a raw buffer and taken
|
||||
//! at the SDK's byte offsets (energy at 336, reaped-descendant CPU time at
|
||||
//! 96 and 104), read before and after each relay once the server is quiet
|
||||
//! on the oracle's own terms: no thread on a core or queued
|
||||
//! (`proc_pidinfo(PROC_PIDTASKINFO)` into a raw buffer, `pti_numrunning` at
|
||||
//! byte 88) at two reads 20 ms apart whose energy agrees. A server blocked
|
||||
//! waiting for its next request does not move its count, so between the
|
||||
//! meter's reads and the oracle's nothing is spent: the call's figure must
|
||||
//! be the oracle's advance across the relay, in nanojoules rounded UP to
|
||||
//! the receipt's microjoules (an enforcement point never debits less than
|
||||
//! was read). That is the tolerance — exact to the µJ: the figure is at
|
||||
//! least the oracle's advance and less than 1 µJ over it.
|
||||
//!
|
||||
//! What a call is sealed as is read back from its receipt, verified, and
|
||||
//! compared with the plaintext beside the response. The tests log what they
|
||||
//! saw (`upstream tree:`), 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 ed25519_dalek::SigningKey;
|
||||
use jcp_gateway::{CallMeter, Gateway, Relayed, Upstream, JCP_META_NS};
|
||||
use serde_json::{json, Value};
|
||||
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 gateway's tree meter may wait for the server to go quiet in
|
||||
/// these tests: far past the default 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 server
|
||||
/// still running after it would be a gap and the call `Partial`, failing a
|
||||
/// test that expects it whole, which is the point: a whole reading is never
|
||||
/// a stale one.
|
||||
const TEST_SETTLE: Duration = Duration::from_secs(2);
|
||||
|
||||
/// The JSON-RPC result for the request line in `l`: its id, and its
|
||||
/// `arguments.n` as a loop count burned before replying.
|
||||
const REPLY: &str = r#"id = "null"; n = 0
|
||||
if (match(l, /"id":[0-9]+/)) id = substr(l, RSTART + 5, RLENGTH - 5)
|
||||
if (match(l, /"n":[0-9]+/)) n = substr(l, RSTART + 4, RLENGTH - 4) + 0
|
||||
for (i = 0; i < n; i++) x += i
|
||||
printf "{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"burned %d\"}]}}\n", id, n"#;
|
||||
|
||||
const STEWARD: &str = "did:jcp:gateway:burner";
|
||||
|
||||
fn steward_sk() -> SigningKey {
|
||||
SigningKey::from_bytes(&[0x2c; 32])
|
||||
}
|
||||
|
||||
/// `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 reaped_cpu_ticks(ru: &[u64; 64]) -> u64 {
|
||||
ru[96 / 8] + ru[104 / 8]
|
||||
}
|
||||
|
||||
/// `pti_numrunning` of `pid`: `proc_pidinfo(pid, PROC_PIDTASKINFO = 4)` into
|
||||
/// a raw 96-byte buffer, the int32 at byte 88.
|
||||
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()))
|
||||
}
|
||||
|
||||
/// `pid`'s rusage once it is quiet by the oracle's rule: not running or
|
||||
/// queued at two reads 20 ms apart, with the same energy at both. Panics
|
||||
/// after 20 s.
|
||||
fn quiet(pid: i32) -> [u64; 64] {
|
||||
let deadline = Instant::now() + Duration::from_secs(20);
|
||||
loop {
|
||||
let first = oracle_rusage(pid).expect("oracle read");
|
||||
let before = oracle_running(pid) == Some(0);
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
let second = oracle_rusage(pid).expect("oracle read");
|
||||
let after = oracle_running(pid) == Some(0);
|
||||
if before && after && energy_nj(&first) == energy_nj(&second) {
|
||||
return second;
|
||||
}
|
||||
assert!(Instant::now() < deadline, "the oracle never saw {pid} quiet");
|
||||
}
|
||||
}
|
||||
|
||||
/// An MCP-ish server: a child in a process group of its own, one JSON-RPC
|
||||
/// request per line on its stdin and one response per line on its stdout,
|
||||
/// killed with its whole group on drop. It names its pid to the gateway
|
||||
/// (`named`, normally its own), and is never waited on while it serves, so
|
||||
/// its pid stays its own.
|
||||
struct Server {
|
||||
child: Child,
|
||||
stdin: ChildStdin,
|
||||
stdout: BufReader<ChildStdout>,
|
||||
named: Option<u32>,
|
||||
}
|
||||
|
||||
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());
|
||||
let named = Some(child.id());
|
||||
Server { child, stdin, stdout, named }
|
||||
}
|
||||
|
||||
/// awk answering every request itself: a single-process server.
|
||||
fn single_process() -> Server {
|
||||
Server::spawn("/usr/bin/awk", &[&format!("{{ l = $0; {REPLY}; fflush() }}")])
|
||||
}
|
||||
|
||||
/// A shell answering each request with an awk worker it forks, waits on
|
||||
/// and reaps before it replies: every call reaps a descendant.
|
||||
fn reaping_a_worker_per_call() -> Server {
|
||||
Server::spawn(
|
||||
"/bin/sh",
|
||||
&[
|
||||
"-c",
|
||||
r#"while IFS= read -r l; do r=$(L="$l" /usr/bin/awk "$1"); printf '%s\n' "$r"; done"#,
|
||||
"sh",
|
||||
&format!(r#"BEGIN {{ l = ENVIRON["L"]; {REPLY} }}"#),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
fn pid(&self) -> i32 {
|
||||
self.child.id() as i32
|
||||
}
|
||||
}
|
||||
|
||||
impl Upstream for Server {
|
||||
fn call(&mut self, line: &str) -> std::io::Result<String> {
|
||||
writeln!(self.stdin, "{line}")?;
|
||||
self.stdin.flush()?;
|
||||
let mut reply = String::new();
|
||||
self.stdout.read_line(&mut reply)?;
|
||||
Ok(reply.trim_end().to_string())
|
||||
}
|
||||
|
||||
fn pid(&self) -> Option<u32> {
|
||||
self.named
|
||||
}
|
||||
}
|
||||
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
/// A gateway granting `burn` a budget no test reaches.
|
||||
fn gateway(declared: bool) -> Gateway {
|
||||
let mut g = Gateway::new("burner", &[("burn", 50_000_000_000)], steward_sk());
|
||||
g.set_upstream_settle_bound(TEST_SETTLE);
|
||||
if declared {
|
||||
g.declare_upstream_self_contained();
|
||||
}
|
||||
g
|
||||
}
|
||||
|
||||
fn burn(id: i64, n: u64) -> String {
|
||||
json!({"jsonrpc":"2.0","id":id,"method":"tools/call","params":{"name":"burn","arguments":{"n":n}}}).to_string()
|
||||
}
|
||||
|
||||
/// What one allowed call came back as.
|
||||
struct Call {
|
||||
cost_uj: u64,
|
||||
/// The call's sealed receipt, verified and decoded.
|
||||
sealed: Value,
|
||||
/// The plaintext beside the response, without the receipt's bytes.
|
||||
meta: Value,
|
||||
}
|
||||
|
||||
fn relay(g: &mut Gateway, up: &mut Server, id: i64, n: u64) -> Call {
|
||||
let Relayed::Allowed { response, cost_uj, .. } = g.relay(&burn(id, n), up).expect("relay") else {
|
||||
panic!("the call was not allowed")
|
||||
};
|
||||
let pubkey = steward_sk().verifying_key().to_bytes();
|
||||
let core = jcp_receipt::jcr1::verify_cose_sign1(g.receipts().last().unwrap(), STEWARD, &pubkey).expect("the receipt verifies");
|
||||
let sealed: Value = jcp_schema::from_canonical_cbor(&core).expect("the receipt decodes");
|
||||
let response: Value = serde_json::from_str(&response).expect("a JSON response");
|
||||
let mut meta = response.pointer(&format!("/result/_meta/{}", JCP_META_NS.replace('/', "~1"))).expect("the jcp _meta").clone();
|
||||
meta.as_object_mut().unwrap().remove("receipt_b64");
|
||||
assert_eq!(response["id"].as_i64(), Some(id), "the server's own response, relayed: {response}");
|
||||
assert_eq!(response["result"]["content"][0]["text"].as_str(), Some(format!("burned {n}").as_str()), "{response}");
|
||||
assert_eq!(sealed["cost_uj"].as_u64(), Some(cost_uj), "{sealed}");
|
||||
assert_eq!(meta["cost_uj"].as_u64(), Some(cost_uj), "{meta}");
|
||||
// The invariant: the plaintext grade is the sealed grade.
|
||||
assert_eq!(meta.get("energy_provenance"), sealed.get("energy_provenance"), "{meta} vs {sealed}");
|
||||
assert_eq!(meta.get("energy_coverage"), sealed.get("energy_coverage"), "{meta} vs {sealed}");
|
||||
Call { cost_uj, sealed, meta }
|
||||
}
|
||||
|
||||
fn grade(c: &Call) -> String {
|
||||
match c.sealed.get("energy_coverage").and_then(Value::as_str) {
|
||||
Some(coverage) => format!("{}, {coverage}", c.sealed["energy_provenance"].as_str().unwrap_or("?")),
|
||||
None => c.sealed["energy_provenance"].as_str().unwrap_or("?").to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
fn the_log_verifies(g: &Gateway) {
|
||||
let pubkey = steward_sk().verifying_key().to_bytes();
|
||||
assert_eq!(jcp_receipt::verify_log(g.receipts(), STEWARD, &pubkey), Ok(g.receipts().len()));
|
||||
}
|
||||
|
||||
/// A single-process server whose operator declared it self-contained: each
|
||||
/// call, from a bare round trip to a few milliseconds of CPU, is sealed
|
||||
/// `HwShunt` and whole, read on the server's own process tree, and its
|
||||
/// figure is the oracle's advance across the relay, exact to the µJ.
|
||||
#[test]
|
||||
fn a_declared_self_contained_server_is_read_whole_on_its_own_process() {
|
||||
let _one = alone();
|
||||
let mut s = Server::single_process();
|
||||
let pid = s.pid();
|
||||
let mut g = gateway(true);
|
||||
assert_eq!(
|
||||
g.call_meter(&s),
|
||||
CallMeter::UpstreamTree { pid: pid as u32 },
|
||||
"a child of this process, on a host whose per-process counter is live (every host this runs on)"
|
||||
);
|
||||
for (rep, n) in [0u64, 20_000, 150_000, 300_000, 0].into_iter().enumerate() {
|
||||
let before = energy_nj(&quiet(pid));
|
||||
let c = relay(&mut g, &mut s, rep as i64, n);
|
||||
let oracle = energy_nj(&quiet(pid)) - before;
|
||||
eprintln!(
|
||||
"upstream tree: declared single process, loop {n} rep {rep}: {} µJ [{}], oracle {oracle} nJ, _meta {}",
|
||||
c.cost_uj,
|
||||
grade(&c),
|
||||
c.meta
|
||||
);
|
||||
assert!(oracle > 0, "the server's count advanced over the call");
|
||||
assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
|
||||
assert!(c.sealed.get("energy_coverage").is_none(), "whole: {}", c.sealed);
|
||||
assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta);
|
||||
assert!(c.meta.get("energy_partial_because").is_none(), "{}", c.meta);
|
||||
assert_eq!(c.cost_uj, oracle.div_ceil(1_000), "the figure against the oracle's advance, to the µJ");
|
||||
}
|
||||
the_log_verifies(&g);
|
||||
}
|
||||
|
||||
/// The same server with no declaration: the same reading, to the µJ, but
|
||||
/// sealed `Partial`, because nothing in the tree can show that the server
|
||||
/// hands no work to a process outside it; the plaintext says that is why.
|
||||
#[test]
|
||||
fn an_undeclared_server_is_read_partial_for_want_of_the_declaration() {
|
||||
let _one = alone();
|
||||
let mut s = Server::single_process();
|
||||
let pid = s.pid();
|
||||
let mut g = gateway(false);
|
||||
assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 });
|
||||
for (rep, n) in [0u64, 150_000, 300_000].into_iter().enumerate() {
|
||||
let before = energy_nj(&quiet(pid));
|
||||
let c = relay(&mut g, &mut s, rep as i64, n);
|
||||
let oracle = energy_nj(&quiet(pid)) - before;
|
||||
eprintln!(
|
||||
"upstream tree: undeclared single process, loop {n} rep {rep}: {} µJ [{}], oracle {oracle} nJ, _meta {}",
|
||||
c.cost_uj,
|
||||
grade(&c),
|
||||
c.meta
|
||||
);
|
||||
assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
|
||||
assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "{}", c.sealed);
|
||||
assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta);
|
||||
assert_eq!(c.meta["energy_partial_because"], json!(["not-declared-self-contained"]), "{}", c.meta);
|
||||
assert_eq!(c.cost_uj, oracle.div_ceil(1_000), "the figure against the oracle's advance, to the µJ");
|
||||
}
|
||||
the_log_verifies(&g);
|
||||
}
|
||||
|
||||
/// A server that answers each call with a worker it forks and reaps (a
|
||||
/// shell loop running awk) is `Partial` even when declared: the worker's
|
||||
/// energy is in no counter once it is reaped. The figure is the shell's own
|
||||
/// advance, exactly, and the plaintext names the reaped descendant as why.
|
||||
#[test]
|
||||
fn a_server_that_reaps_a_worker_per_call_is_partial_even_when_declared() {
|
||||
let _one = alone();
|
||||
let mut s = Server::reaping_a_worker_per_call();
|
||||
let pid = s.pid();
|
||||
let mut g = gateway(true);
|
||||
assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 });
|
||||
for (rep, n) in [20_000u64, 200_000, 200_000].into_iter().enumerate() {
|
||||
let before = quiet(pid);
|
||||
let c = relay(&mut g, &mut s, rep as i64, n);
|
||||
let after = quiet(pid);
|
||||
let oracle = energy_nj(&after) - energy_nj(&before);
|
||||
let reaped = reaped_cpu_ticks(&after) - reaped_cpu_ticks(&before);
|
||||
eprintln!(
|
||||
"upstream tree: declared shell reaping a worker, loop {n} rep {rep}: {} µJ [{}], oracle (the shell's own) {oracle} nJ, reaped {reaped} ticks, _meta {}",
|
||||
c.cost_uj,
|
||||
grade(&c),
|
||||
c.meta
|
||||
);
|
||||
assert!(reaped > 0, "setup: the worker ran and was reaped in the call");
|
||||
assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
|
||||
assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "{}", c.sealed);
|
||||
assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta);
|
||||
assert_eq!(c.meta["energy_partial_because"], json!(["reaped-descendant"]), "{}", c.meta);
|
||||
assert_eq!(c.cost_uj, oracle.div_ceil(1_000), "the shell's own advance, to the µJ");
|
||||
}
|
||||
the_log_verifies(&g);
|
||||
}
|
||||
|
||||
/// A declared server whose worker writes the reply ITSELF, keeps computing
|
||||
/// for a few milliseconds and exits, and is reaped by the server 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). The worker was
|
||||
/// born in the call and spent energy in it, and is in no counter once
|
||||
/// reaped, so no call is sealed whole — not even one whose stop walk read
|
||||
/// the server before the reap and looked for the worker after it.
|
||||
#[test]
|
||||
fn a_worker_reaped_after_writing_the_reply_never_leaves_a_call_whole() {
|
||||
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 ($id) = $l =~ /"id":(\d+)/; my ($n) = $l =~ /"n":(\d+)/; $n //= 0;
|
||||
my $p = fork();
|
||||
if ($p == 0) {
|
||||
print qq({"jsonrpc":"2.0","id":$id,"result":{"content":[{"type":"text","text":"burned $n"}]}}\n);
|
||||
my $x = 0; $x += $_ for 1 .. $n; exit 0
|
||||
}
|
||||
waitpid($p, 0);
|
||||
}"#,
|
||||
],
|
||||
);
|
||||
let pid = s.pid();
|
||||
let mut g = gateway(true);
|
||||
assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 });
|
||||
let _ = relay(&mut g, &mut s, 0, 0);
|
||||
let mut reasons: Vec<String> = Vec::new();
|
||||
for rep in 1..31 {
|
||||
let c = relay(&mut g, &mut s, rep, 150_000);
|
||||
eprintln!("upstream tree: declared server, worker reaped after its reply, rep {rep}: {} µJ [{}], _meta {}", c.cost_uj, grade(&c), c.meta);
|
||||
assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
|
||||
assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "a call whose worker is in no reading sealed whole: {}", c.meta);
|
||||
let because = c.meta["energy_partial_because"].as_array().expect("why it is partial");
|
||||
assert!(
|
||||
because.iter().any(|k| matches!(k.as_str(), Some("reaped-descendant" | "exited" | "unsettled-at-start" | "unsettled-at-stop" | "tree-changing"))),
|
||||
"{}",
|
||||
c.meta
|
||||
);
|
||||
reasons.extend(because.iter().filter_map(Value::as_str).map(str::to_string));
|
||||
}
|
||||
reasons.sort();
|
||||
reasons.dedup();
|
||||
eprintln!("upstream tree: declared server, worker reaped after its reply: reasons seen {reasons:?}");
|
||||
the_log_verifies(&g);
|
||||
}
|
||||
|
||||
/// A Metal compute server speaking the tests' JSON-RPC, built from this
|
||||
/// source when the test first runs: each request's `arguments.n` picks what
|
||||
/// it does before it replies — 0 nothing on the GPU, 1 a compute kernel on
|
||||
/// the command queue it keeps, 2 the same kernel on a queue made for the
|
||||
/// request and released after it — and it replies `burned <n>`.
|
||||
const METAL_SERVER: &str = r##"#import <Metal/Metal.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.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";
|
||||
static long field(const char *line, const char *key) {
|
||||
const char *p = strstr(line, key);
|
||||
return p ? strtol(p + strlen(key), NULL, 10) : 0;
|
||||
}
|
||||
int main(void) {
|
||||
@autoreleasepool {
|
||||
id<MTLDevice> d = MTLCreateSystemDefaultDevice();
|
||||
if (!d) return 1;
|
||||
NSError *err = nil;
|
||||
id<MTLLibrary> lib = [d newLibraryWithSource:src options:nil error:&err];
|
||||
if (!lib) 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];
|
||||
char line[4096];
|
||||
while (fgets(line, sizeof line, stdin)) {
|
||||
long rid = field(line, "\"id\":"), n = field(line, "\"n\":");
|
||||
if (n == 1 || n == 2) {
|
||||
@autoreleasepool {
|
||||
id<MTLCommandQueue> q = n == 2 ? [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];
|
||||
}
|
||||
}
|
||||
printf("{\"jsonrpc\":\"2.0\",\"id\":%ld,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"burned %ld\"}]}}\n", rid, n);
|
||||
fflush(stdout);
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
"##;
|
||||
|
||||
/// A declared server that holds a GPU client is never sealed whole, whatever
|
||||
/// a call did: its kept queue's work is recorded GPU time ("gpu"), and a
|
||||
/// call whose work ran on a queue released inside it, or that did no GPU
|
||||
/// work, leaves the server's recorded time unmoved, which does not show it
|
||||
/// used none ("gpu-unknown"). The meter waits only 200 ms for the server to
|
||||
/// go quiet: a Metal process keeps threads that wake on their own, and a
|
||||
/// server still running at an end is a reason of its own.
|
||||
#[test]
|
||||
fn a_declared_server_holding_a_gpu_client_is_never_sealed_whole() {
|
||||
let _one = alone();
|
||||
let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"));
|
||||
let source = dir.join("upstream_tree_metal_server.m");
|
||||
let binary = dir.join(format!("upstream_tree_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));
|
||||
let mut s = Server::spawn(binary.to_str().unwrap(), &[]);
|
||||
let pid = s.pid();
|
||||
let mut g = gateway(true);
|
||||
g.set_upstream_settle_bound(Duration::from_millis(200));
|
||||
assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 });
|
||||
for (rep, n) in [1u64, 2, 0, 2, 1, 0].into_iter().enumerate() {
|
||||
let c = relay(&mut g, &mut s, rep as i64, n);
|
||||
eprintln!("upstream tree: declared Metal server, mode {n} rep {rep}: {} µJ [{}], _meta {}", c.cost_uj, grade(&c), c.meta);
|
||||
assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
|
||||
assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "a GPU-holding server sealed whole: {}", c.meta);
|
||||
let because: Vec<&str> = c.meta["energy_partial_because"].as_array().expect("why").iter().filter_map(Value::as_str).collect();
|
||||
let want = if n == 1 { "gpu" } else { "gpu-unknown" };
|
||||
assert!(because.contains(&want) || because.contains(&"gpu"), "{want} not among {because:?}");
|
||||
}
|
||||
the_log_verifies(&g);
|
||||
}
|
||||
|
||||
/// An upstream naming a process this one may not read (launchd, another
|
||||
/// user's) gets no tree meter, and its calls are metered machine-wide,
|
||||
/// sealed as that meter grades them.
|
||||
#[test]
|
||||
fn an_upstream_naming_a_process_this_one_cannot_read_is_metered_machine_wide() {
|
||||
// SAFETY: no preconditions.
|
||||
if unsafe { libc::geteuid() } == 0 {
|
||||
eprintln!("upstream fallback: running as root, launchd is readable; nothing to check");
|
||||
return;
|
||||
}
|
||||
let _one = alone();
|
||||
let mut s = Server::single_process();
|
||||
s.named = Some(1);
|
||||
let mut g = gateway(true);
|
||||
let machine = if g.metering_is_hardware() { CallMeter::MachineHardware } else { CallMeter::Estimator };
|
||||
assert_eq!(g.call_meter(&s), machine);
|
||||
let c = relay(&mut g, &mut s, 1, 20_000);
|
||||
eprintln!("upstream fallback: pid 1 named: {} µJ [{}], _meta {}", c.cost_uj, grade(&c), c.meta);
|
||||
assert_eq!(c.meta["energy_domain"].as_str(), Some("machine"), "{}", c.meta);
|
||||
assert!(c.meta.get("energy_partial_because").is_none(), "{}", c.meta);
|
||||
if !g.metering_is_hardware() {
|
||||
assert_eq!(c.sealed["energy_provenance"].as_str(), Some("Estimator"), "{}", c.sealed);
|
||||
}
|
||||
the_log_verifies(&g);
|
||||
}
|
||||
|
|
@ -6,7 +6,7 @@ Sandbox is the **isolated-execution substrate** of the Transaction Science open
|
|||
|
||||
Every other sandbox in the field isolates a **CPU process** — a container, a microVM, a Linux process pinned to a core. The accelerator, where present, is an opaque passthrough that nobody meters and nobody attests. But agentic and AI work does not run on a CPU. It runs across a **fabric** — CPU, GPU, NPU, TPU, photonic, quantum. So in this standard the unit of isolation is not a process; it is a **heterogeneous unit of work**, and the accelerator is a first-class citizen of the isolation boundary: a capability you grant, a resource you meter in joules, and a line item you attest.
|
||||
|
||||
A Sandbox run carries exactly four guarantees: **isolation** (the unit runs at the lowest tier of an eight-rung ladder, S0–S7, that satisfies its needs, and cannot reach past it), **capability** (every resource the unit touches — each device class included — was explicitly granted; ungranted access is denied), **metering** (the run emits a signed energy receipt with per-device-class joule attribution and an honest provenance tag on every reading — and the reference meter reads vendor energy counters: Apple Silicon's energy model via IOReport, Intel/AMD RAPL via powercap, NVIDIA's total-energy counter via NVML. A reading is tagged `HwShunt` only while its counter is demonstrably live: each vendor counter is used only if every rail of it that the meter sums into one figure keeps advancing while the meter's probe burns CPU — in at least 5 sample intervals, at least 4 of every 5, over at least 300 ms — every IOReport CPU energy rail (the rail itself, one per die on a multi-die part), every RAPL package and every NVML GPU; and a span across which the CPU counter held is never a reading. A counter that is absent, or present but not live — frozen, or republished in lumps, as the IOReport CPU rails are on macOS 27 — yields no hardware-graded figure: the span degrades to the labeled `Estimator` (wall-clock × TDP), never a fabricated measurement, and `ioreport_status()`, `rapl_status()` and `nvml_status()` say why a host is estimated. For work on the meter's own process there is one more source, kept out of the default meter: on macOS, the CPU energy the kernel attributes to the process's own threads (`proc_pid_rusage`), read through `PowerMeter::detect_in_process()` under the same liveness rule — never for a subprocess, which it does not cover; a single-process child's own figure is read from the exited child instead. Because a live counter reports the whole SoC, the meter can **attribute** it to a single process tree by CPU-time apportionment — tagged `ModelBased`, since the reading is a measurement but the split is a model), and **attestation** (a signed statement of *what image ran, at what tier, under what grants, on what hardware, for what cost*, with a deterministic-replay journal hash where one was produced).
|
||||
A Sandbox run carries exactly four guarantees: **isolation** (the unit runs at the lowest tier of an eight-rung ladder, S0–S7, that satisfies its needs, and cannot reach past it), **capability** (every resource the unit touches — each device class included — was explicitly granted; ungranted access is denied), **metering** (the run emits a signed energy receipt with per-device-class joule attribution and an honest provenance tag on every reading — and the reference meter reads vendor energy counters: Apple Silicon's energy model via IOReport, Intel/AMD RAPL via powercap, NVIDIA's total-energy counter via NVML. A reading is tagged `HwShunt` only while its counter is demonstrably live: each vendor counter is used only if every rail of it that the meter sums into one figure keeps advancing while the meter's probe burns CPU — in at least 5 sample intervals, at least 4 of every 5, over at least 300 ms — every IOReport CPU energy rail (the rail itself, one per die on a multi-die part), every RAPL package and every NVML GPU; and a span across which the CPU counter held is never a reading. A counter that is absent, or present but not live — frozen, or republished in lumps, as the IOReport CPU rails are on macOS 27 — yields no hardware-graded figure: the span degrades to the labeled `Estimator` (wall-clock × TDP), never a fabricated measurement, and `ioreport_status()`, `rapl_status()` and `nvml_status()` say why a host is estimated. For work on the meter's own process there is one more source, kept out of the default meter: on macOS, the CPU energy the kernel attributes to the process's own threads (`proc_pid_rusage`), read through `PowerMeter::detect_in_process()` under the same liveness rule — never for a subprocess, which it does not cover; a single-process child's own figure is read from the exited child instead. For work in another live process of this user — a server, and its descendants — `TreeMeter` reads that same count for every process of its tree over a span, each end read once no thread of the tree is on a core and confirmed by a second walk that finds the tree's processes unchanged, and lists beside the figure every way it was seen to differ from the tree's energy over the span — short of it (a descendant reaped or exited in the span, a GPU client held, in-process Neural Engine memory, a member unreadable or still running at the stop), or, for a member still running at the start, holding some from just before the span; whether the tree hands work to a process outside it is not visible to any reader, so that is for the caller's operator to declare, never the reading's to claim. Because a live counter reports the whole SoC, the meter can **attribute** it to a single process tree by CPU-time apportionment — tagged `ModelBased`, since the reading is a measurement but the split is a model), and **attestation** (a signed statement of *what image ran, at what tier, under what grants, on what hardware, for what cost*, with a deterministic-replay journal hash where one was produced).
|
||||
|
||||
The substrate is intentionally **ownerless**: no protocol-level token, permissive licensing on spec text and reference code, and conformance vectors so any implementer can verify canonicity without a trust relationship with the authors. A conforming host runs a unit, enforces its grants, meters its energy, and seals its attestation **without contacting any single steward's servers as a condition of operation**. Transaction Science is one steward: it publishes the spec, ships a reference implementation, and operates commercial services around the substrate — not its proprietor.
|
||||
|
||||
|
|
|
|||
|
|
@ -5,6 +5,14 @@
|
|||
//! (`measured + estimated`, integer microjoules), every reading carrying a
|
||||
//! [`Provenance`] tag. It seals as a [`jcr1`] envelope — the family's shared
|
||||
//! signed-receipt format.
|
||||
//!
|
||||
//! The readings come from [`probe`]: a [`PowerMeter`] over the host's
|
||||
//! machine-wide energy counters (or its estimator where none is live), the
|
||||
//! same meter over this process's own CPU energy count
|
||||
//! ([`PowerMeter::detect_in_process`]), and a [`TreeMeter`] over that count
|
||||
//! read for every process of another live process tree — a server this
|
||||
//! process spawned, and its descendants — with every way the tree's figure
|
||||
//! was seen to fall short listed beside it ([`TreeGap`]).
|
||||
|
||||
use ed25519_dalek::{SigningKey, VerifyingKey};
|
||||
use sandbox_schema::{ContentId, DeviceClass, Tier};
|
||||
|
|
@ -17,7 +25,8 @@ pub mod probe;
|
|||
pub use contention::{StorageActivity, StorageMeterError};
|
||||
pub use probe::{
|
||||
ioreport_status, nvml_status, process_energy_status, rapl_status, run_attributed, Attribution,
|
||||
CounterStatus, IoReportStatus, Measurement, PowerMeter, CPU_RAIL_LIVENESS_BUDGET,
|
||||
CounterStatus, IoReportStatus, Measurement, PowerMeter, SpanEnd, TreeError, TreeGap, TreeMeter,
|
||||
TreeReading, TreeStart, CPU_RAIL_LIVENESS_BUDGET,
|
||||
};
|
||||
|
||||
/// The receipt's JCR-1 profile tag (SPEC §6.4).
|
||||
|
|
|
|||
282
sandbox/sandbox-rs/crates/sandbox-meter/src/probe/gpu_clients.rs
Normal file
282
sandbox/sandbox-rs/crates/sandbox-meter/src/probe/gpu_clients.rs
Normal file
|
|
@ -0,0 +1,282 @@
|
|||
//! Every GPU client's accumulated GPU time, from the I/O Registry (Apple
|
||||
//! Silicon macOS): the evidence a [`TreeMeter`](super::TreeMeter) reads to
|
||||
//! say whether a process tree used the GPU over a span. Ported from
|
||||
//! jouleclaw-energy's `apple_gpu`, with one change: a client with no
|
||||
//! `AppUsage` is told apart from one whose time is zero, because a client
|
||||
//! LOSES its `AppUsage` when its process exits.
|
||||
//!
|
||||
//! Each `IOAccelerator` service (on Apple Silicon one, `AGXAccelerator…`) has
|
||||
//! one user client per connection a process opened to the GPU, as a child in
|
||||
//! the `IOService` plane (`AGXDeviceUserClient`). Each carries
|
||||
//! `IOUserClientCreator` (`"pid N, name"`) and `AppUsage`, an array with one
|
||||
//! dictionary per live command queue (its `CommandQueueCount` matched the
|
||||
//! entries), whose `accumulatedGPUTime` is the GPU time that queue has
|
||||
//! spent, in ns. This is public IOKit API, read
|
||||
//! unprivileged; it is what `ioreg -r -c IOAccelerator -l` prints. A process
|
||||
//! can hold several clients. A CPU-only process has none.
|
||||
//!
|
||||
//! **What a reading can and cannot say** (measured on an M5 Max, macOS
|
||||
//! 27.0.1, unprivileged): a Metal child's time is live, with no read lag (a
|
||||
//! read right after its work and one 200 ms later agreed in 11 of 11 runs).
|
||||
//! But a client's time is NOT the process's GPU use: a command queue's entry
|
||||
//! leaves `AppUsage`, with its time, when the queue is released, so a
|
||||
//! request run on a queue created and released between two reads (5.2 ms of
|
||||
//! GPU work by its command buffer's own timestamps) moved its client's time
|
||||
//! by 0, as did one on a device opened and closed between them. A client
|
||||
//! whose time did not move may have done GPU work; only a process with no
|
||||
//! client is known to have done none. And GPU time is LOST AT EXIT:
|
||||
//! about 2 ms after its process exits a client has no `AppUsage` (its time
|
||||
//! reads as none), and by about 9 ms the client is gone from the registry —
|
||||
//! while the process itself may live on for as long as nobody reaps it. So a
|
||||
//! client whose time went back, to zero or to none, or that vanished, is a
|
||||
//! client whose time is unknown, never a decrease; the caller's rule
|
||||
//! ([`tree`](super::tree)) treats it so. One walk costs about 1 ms (63 ms at
|
||||
//! worst seen, on a host at load ~200).
|
||||
//!
|
||||
//! The CoreFoundation calls `ioreport` already binds are its declarations,
|
||||
//! not new ones; this module adds the type-ID and number calls and IOKit.
|
||||
|
||||
#![allow(non_snake_case)]
|
||||
|
||||
use super::ioreport::{
|
||||
CFArrayGetCount, CFArrayGetValueAtIndex, CFArrayRef, CFDictionaryGetValue, CFDictionaryRef, CFIndex,
|
||||
CFMutableDictionaryRef, CFRelease, CFStringCreateWithCString, CFStringGetCString, CFStringRef, CFTypeRef,
|
||||
KCF_STRING_ENCODING_UTF8,
|
||||
};
|
||||
use super::tree::GpuClient;
|
||||
use std::ffi::{c_char, c_void, CStr};
|
||||
|
||||
type CFTypeID = usize;
|
||||
type IoObject = libc::mach_port_t;
|
||||
|
||||
/// `kCFNumberSInt64Type`.
|
||||
const K_CFNUMBER_SINT64: CFIndex = 4;
|
||||
/// `kIOMainPortDefault` is `MACH_PORT_NULL`.
|
||||
const MAIN_PORT_DEFAULT: libc::mach_port_t = 0;
|
||||
const KERN_SUCCESS: libc::kern_return_t = 0;
|
||||
|
||||
#[link(name = "CoreFoundation", kind = "framework")]
|
||||
extern "C" {
|
||||
fn CFGetTypeID(cf: CFTypeRef) -> CFTypeID;
|
||||
fn CFStringGetTypeID() -> CFTypeID;
|
||||
fn CFArrayGetTypeID() -> CFTypeID;
|
||||
fn CFDictionaryGetTypeID() -> CFTypeID;
|
||||
fn CFNumberGetTypeID() -> CFTypeID;
|
||||
fn CFNumberGetValue(number: CFTypeRef, the_type: CFIndex, value_ptr: *mut c_void) -> bool;
|
||||
}
|
||||
|
||||
#[link(name = "IOKit", kind = "framework")]
|
||||
extern "C" {
|
||||
fn IOServiceMatching(name: *const c_char) -> CFMutableDictionaryRef;
|
||||
fn IOServiceGetMatchingServices(
|
||||
main_port: libc::mach_port_t,
|
||||
matching: CFDictionaryRef,
|
||||
existing: *mut IoObject,
|
||||
) -> libc::kern_return_t;
|
||||
fn IOIteratorNext(iterator: IoObject) -> IoObject;
|
||||
fn IOObjectRelease(object: IoObject) -> libc::kern_return_t;
|
||||
fn IORegistryEntryGetChildIterator(entry: IoObject, plane: *const c_char, iterator: *mut IoObject)
|
||||
-> libc::kern_return_t;
|
||||
fn IORegistryEntryGetRegistryEntryID(entry: IoObject, id: *mut u64) -> libc::kern_return_t;
|
||||
fn IORegistryEntryCreateCFProperty(
|
||||
entry: IoObject,
|
||||
key: CFStringRef,
|
||||
allocator: *const c_void,
|
||||
options: u32,
|
||||
) -> CFTypeRef;
|
||||
}
|
||||
|
||||
/// An owned CF object, released on drop.
|
||||
struct Cf(CFTypeRef);
|
||||
|
||||
impl Drop for Cf {
|
||||
fn drop(&mut self) {
|
||||
if !self.0.is_null() {
|
||||
// SAFETY: `self.0` is a CF object this process owns (a Create/Copy
|
||||
// result), released once.
|
||||
unsafe { CFRelease(self.0) }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// An owned I/O Kit object, released on drop.
|
||||
struct Io(IoObject);
|
||||
|
||||
impl Drop for Io {
|
||||
fn drop(&mut self) {
|
||||
if self.0 != 0 {
|
||||
// SAFETY: `self.0` is an object reference this process owns,
|
||||
// released once.
|
||||
unsafe { IOObjectRelease(self.0) };
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Io {
|
||||
/// The next object of an iterator, or `None` at its end.
|
||||
fn next_of(iterator: &Io) -> Option<Io> {
|
||||
// SAFETY: `iterator` is a live iterator this process owns.
|
||||
let o = unsafe { IOIteratorNext(iterator.0) };
|
||||
(o != 0).then_some(Io(o))
|
||||
}
|
||||
}
|
||||
|
||||
fn cfstr(s: &CStr) -> Option<Cf> {
|
||||
// SAFETY: `s` is a NUL-terminated C string for the duration of the call.
|
||||
let r = unsafe { CFStringCreateWithCString(std::ptr::null(), s.as_ptr() as *const u8, KCF_STRING_ENCODING_UTF8) };
|
||||
(!r.is_null()).then_some(Cf(r))
|
||||
}
|
||||
|
||||
/// `entry`'s property `key`, owned, if it is set and of the CF type `type_id`.
|
||||
fn property(entry: &Io, key: &Cf, type_id: CFTypeID) -> Option<Cf> {
|
||||
// SAFETY: `entry` is a live registry entry and `key` a live CFString; the
|
||||
// result is a new reference this process owns (wrapped at once).
|
||||
let v = Cf(unsafe { IORegistryEntryCreateCFProperty(entry.0, key.0, std::ptr::null(), 0) });
|
||||
// SAFETY: `v.0` is a live CF object when non-null.
|
||||
(!v.0.is_null() && unsafe { CFGetTypeID(v.0) } == type_id).then_some(v)
|
||||
}
|
||||
|
||||
/// A CFString's text, copied out of it.
|
||||
fn text(s: &Cf) -> Option<String> {
|
||||
let mut buf = [0u8; 256];
|
||||
// SAFETY: `s.0` is a live CFString and `buf` a writable buffer of the size
|
||||
// passed.
|
||||
let ok = unsafe { CFStringGetCString(s.0, buf.as_mut_ptr(), buf.len() as CFIndex, KCF_STRING_ENCODING_UTF8) };
|
||||
if !ok {
|
||||
return None;
|
||||
}
|
||||
let end = buf.iter().position(|&b| b == 0).unwrap_or(buf.len());
|
||||
Some(String::from_utf8_lossy(&buf[..end]).into_owned())
|
||||
}
|
||||
|
||||
/// The pid in an `IOUserClientCreator` value, `"pid 172, WindowServer"`.
|
||||
pub(super) fn creator_pid(creator: &str) -> Option<u32> {
|
||||
let rest = creator.strip_prefix("pid ")?;
|
||||
let digits = rest.split(',').next()?.trim();
|
||||
digits.parse().ok()
|
||||
}
|
||||
|
||||
/// The `accumulatedGPUTime` of every `AppUsage` entry, summed, in ns; `None`
|
||||
/// if an entry is not a dictionary or its time is not a 64-bit integer.
|
||||
fn app_usage_ns(usage: &Cf, time_key: &Cf) -> Option<u64> {
|
||||
let mut sum: u64 = 0;
|
||||
// SAFETY: `usage` is a live CFArray (its type was checked); its values and
|
||||
// their values are borrowed from it for the duration of this call.
|
||||
unsafe {
|
||||
for i in 0..CFArrayGetCount(usage.0 as CFArrayRef) {
|
||||
let entry = CFArrayGetValueAtIndex(usage.0 as CFArrayRef, i);
|
||||
if entry.is_null() || CFGetTypeID(entry) != CFDictionaryGetTypeID() {
|
||||
return None;
|
||||
}
|
||||
let t = CFDictionaryGetValue(entry as CFDictionaryRef, time_key.0);
|
||||
if t.is_null() {
|
||||
// An entry that has not recorded GPU time yet.
|
||||
continue;
|
||||
}
|
||||
if CFGetTypeID(t) != CFNumberGetTypeID() {
|
||||
return None;
|
||||
}
|
||||
let mut ns: i64 = 0;
|
||||
if !CFNumberGetValue(t, K_CFNUMBER_SINT64, &mut ns as *mut i64 as *mut c_void) || ns < 0 {
|
||||
return None;
|
||||
}
|
||||
sum = sum.checked_add(ns as u64)?;
|
||||
}
|
||||
}
|
||||
Some(sum)
|
||||
}
|
||||
|
||||
/// Every GPU client in the registry now: each `IOAccelerator`'s children in
|
||||
/// the `IOService` plane that carry an `IOUserClientCreator`, with their
|
||||
/// accumulated GPU time — `None` for a client with no `AppUsage`, which has
|
||||
/// spent none yet or (its process having exited) has lost it. `None` overall
|
||||
/// if the registry could not be walked, or a client's creator, usage or ID
|
||||
/// could not be read: a reading that silently left a client out would call
|
||||
/// its GPU time zero.
|
||||
pub(super) fn gpu_clients() -> Option<Vec<GpuClient>> {
|
||||
let creator_key = cfstr(c"IOUserClientCreator")?;
|
||||
let usage_key = cfstr(c"AppUsage")?;
|
||||
let time_key = cfstr(c"accumulatedGPUTime")?;
|
||||
// SAFETY: plain type-ID getters.
|
||||
let (string_t, array_t) = unsafe { (CFStringGetTypeID(), CFArrayGetTypeID()) };
|
||||
|
||||
// SAFETY: `IOServiceMatching` returns a new dictionary, which
|
||||
// `IOServiceGetMatchingServices` consumes whatever it returns.
|
||||
let services = unsafe {
|
||||
let matching = IOServiceMatching(c"IOAccelerator".as_ptr());
|
||||
if matching.is_null() {
|
||||
return None;
|
||||
}
|
||||
let mut it: IoObject = 0;
|
||||
if IOServiceGetMatchingServices(MAIN_PORT_DEFAULT, matching as CFDictionaryRef, &mut it) != KERN_SUCCESS {
|
||||
return None;
|
||||
}
|
||||
Io(it)
|
||||
};
|
||||
let mut out = Vec::new();
|
||||
while let Some(accelerator) = Io::next_of(&services) {
|
||||
let mut it: IoObject = 0;
|
||||
// SAFETY: `accelerator` is a live registry entry; the iterator is ours.
|
||||
if unsafe { IORegistryEntryGetChildIterator(accelerator.0, c"IOService".as_ptr(), &mut it) } != KERN_SUCCESS {
|
||||
return None;
|
||||
}
|
||||
let children = Io(it);
|
||||
while let Some(child) = Io::next_of(&children) {
|
||||
let Some(creator) = property(&child, &creator_key, string_t) else {
|
||||
// Not a user client.
|
||||
continue;
|
||||
};
|
||||
let pid = creator_pid(&text(&creator)?)?;
|
||||
let gpu_ns = match property(&child, &usage_key, array_t) {
|
||||
Some(usage) => Some(app_usage_ns(&usage, &time_key)?),
|
||||
None => None,
|
||||
};
|
||||
let mut id: u64 = 0;
|
||||
// SAFETY: `child` is a live registry entry; `id` is writable.
|
||||
if unsafe { IORegistryEntryGetRegistryEntryID(child.0, &mut id) } != KERN_SUCCESS {
|
||||
return None;
|
||||
}
|
||||
out.push(GpuClient { id, pid, gpu_ns });
|
||||
}
|
||||
}
|
||||
Some(out)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn the_creator_names_its_pid() {
|
||||
assert_eq!(creator_pid("pid 172, WindowServer"), Some(172));
|
||||
assert_eq!(creator_pid("pid 99917, Google Chrome He"), Some(99_917));
|
||||
assert_eq!(creator_pid("pid 5"), Some(5));
|
||||
assert_eq!(creator_pid("172, WindowServer"), None);
|
||||
assert_eq!(creator_pid("pid x, y"), None);
|
||||
assert_eq!(creator_pid("pid -3, y"), None);
|
||||
}
|
||||
|
||||
/// On this host the registry walks, every client is a distinct entry,
|
||||
/// some client has recorded GPU time, and this test process — which
|
||||
/// never opened the GPU — holds none.
|
||||
#[test]
|
||||
fn the_registry_lists_gpu_clients_once_each() {
|
||||
let started = std::time::Instant::now();
|
||||
let clients = gpu_clients().expect("the I/O Registry's GPU clients");
|
||||
let took = started.elapsed();
|
||||
let mut ids: Vec<u64> = clients.iter().map(|c| c.id).collect();
|
||||
ids.sort_unstable();
|
||||
ids.dedup();
|
||||
let with_usage = clients.iter().filter(|c| c.gpu_ns.is_some()).count();
|
||||
eprintln!("gpu clients: {} (distinct ids {}, {with_usage} with AppUsage) in {took:?}", clients.len(), ids.len());
|
||||
assert!(!clients.is_empty(), "no GPU client at all: WindowServer holds some on any Mac with a display");
|
||||
// The time key read: a misspelt key reads every client as having
|
||||
// spent nothing, and a reader that sees no GPU time anywhere is blind.
|
||||
assert!(
|
||||
clients.iter().any(|c| c.gpu_ns.is_some_and(|ns| ns > 0)),
|
||||
"no client has recorded any GPU time, not even the display server's"
|
||||
);
|
||||
assert_eq!(ids.len(), clients.len());
|
||||
assert!(!clients.iter().any(|c| c.pid == std::process::id()), "a CPU-only test process holds a GPU client");
|
||||
}
|
||||
}
|
||||
|
|
@ -51,29 +51,32 @@ use std::ptr;
|
|||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
type CFTypeRef = *const c_void;
|
||||
type CFStringRef = *const c_void;
|
||||
type CFDictionaryRef = *const c_void;
|
||||
type CFMutableDictionaryRef = *mut c_void;
|
||||
type CFArrayRef = *const c_void;
|
||||
// The CoreFoundation types and calls are shared with `gpu_clients`, the I/O
|
||||
// Registry reader built under the same configuration, so one declaration of
|
||||
// each CF symbol serves both.
|
||||
pub(super) type CFTypeRef = *const c_void;
|
||||
pub(super) type CFStringRef = *const c_void;
|
||||
pub(super) type CFDictionaryRef = *const c_void;
|
||||
pub(super) type CFMutableDictionaryRef = *mut c_void;
|
||||
pub(super) type CFArrayRef = *const c_void;
|
||||
type CFAllocatorRef = *const c_void;
|
||||
type CFIndex = isize;
|
||||
pub(super) type CFIndex = isize;
|
||||
type IOReportSubscriptionRef = *mut c_void;
|
||||
|
||||
const KCF_STRING_ENCODING_UTF8: u32 = 0x0800_0100;
|
||||
pub(super) const KCF_STRING_ENCODING_UTF8: u32 = 0x0800_0100;
|
||||
|
||||
#[link(name = "CoreFoundation", kind = "framework")]
|
||||
extern "C" {
|
||||
fn CFStringCreateWithCString(
|
||||
pub(super) fn CFStringCreateWithCString(
|
||||
alloc: CFAllocatorRef,
|
||||
c_str: *const u8,
|
||||
encoding: u32,
|
||||
) -> CFStringRef;
|
||||
fn CFStringGetCString(s: CFStringRef, buf: *mut u8, len: CFIndex, encoding: u32) -> bool;
|
||||
fn CFDictionaryGetValue(d: CFDictionaryRef, key: CFTypeRef) -> CFTypeRef;
|
||||
fn CFArrayGetCount(a: CFArrayRef) -> CFIndex;
|
||||
fn CFArrayGetValueAtIndex(a: CFArrayRef, idx: CFIndex) -> CFTypeRef;
|
||||
fn CFRelease(t: CFTypeRef);
|
||||
pub(super) fn CFStringGetCString(s: CFStringRef, buf: *mut u8, len: CFIndex, encoding: u32) -> bool;
|
||||
pub(super) fn CFDictionaryGetValue(d: CFDictionaryRef, key: CFTypeRef) -> CFTypeRef;
|
||||
pub(super) fn CFArrayGetCount(a: CFArrayRef) -> CFIndex;
|
||||
pub(super) fn CFArrayGetValueAtIndex(a: CFArrayRef, idx: CFIndex) -> CFTypeRef;
|
||||
pub(super) fn CFRelease(t: CFTypeRef);
|
||||
}
|
||||
|
||||
#[link(name = "IOReport", kind = "dylib")]
|
||||
|
|
|
|||
|
|
@ -32,6 +32,18 @@
|
|||
//! ([`process_energy_status`]), through [`PowerMeter::detect_in_process`]. It
|
||||
//! does not cover child processes, so it is never the meter for a subprocess.
|
||||
//!
|
||||
//! The same per-process count, read for every member of ANOTHER live process
|
||||
//! tree (a server this process spawned, and its descendants), is a
|
||||
//! [`TreeMeter`]: that tree's CPU energy over a span, with every way the
|
||||
//! figure was seen to fall short of the tree's whole energy listed beside it
|
||||
//! (a descendant reaped in the span, GPU or Neural Engine use, a member that
|
||||
//! would not go quiet). macOS only, and only where the count is live. A
|
||||
//! reading with nothing listed is whole for that tree only: work the tree
|
||||
//! hands to a process outside it is charged to that process, and no reader
|
||||
//! of the tree sees it, so whether there is any is for the caller to have
|
||||
//! declared (jcp-gateway takes it as the operator's
|
||||
//! `--upstream-self-contained`), never for the reading to claim.
|
||||
//!
|
||||
//! A probe can degrade but it can never lie: if the hardware path fails
|
||||
//! mid-span, the reading comes back `Estimator`, not a zero dressed as a
|
||||
//! measurement. The same holds when the CPU energy counter did not advance
|
||||
|
|
@ -54,6 +66,15 @@ mod nvml;
|
|||
// `detect_in_process()`.
|
||||
#[cfg(target_os = "macos")]
|
||||
mod process;
|
||||
// The CPU energy of another live process tree, read the same way for every
|
||||
// member. The rule and its types are built and tested everywhere; only macOS
|
||||
// walks a tree. Public for its module docs: what a reading covers.
|
||||
pub mod tree;
|
||||
pub use tree::{SpanEnd, TreeError, TreeGap, TreeMeter, TreeReading, TreeStart, SETTLE_BOUND, SETTLE_POLL};
|
||||
// Every GPU client's GPU time from the I/O Registry: the GPU evidence of a
|
||||
// tree reading. Shares `ioreport`'s CoreFoundation bindings.
|
||||
#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
|
||||
mod gpu_clients;
|
||||
// Built everywhere: its zone listing, reads and liveness run take the
|
||||
// powercap root as a path, so a caller's [`RaplPackages::live_in`] and the
|
||||
// tests run them on fake powercap trees on any platform. Only Linux opens the
|
||||
|
|
|
|||
|
|
@ -24,9 +24,13 @@
|
|||
//!
|
||||
//! The same read, on a child that has exited but not yet been reaped, gives
|
||||
//! that child's own energy: [`child_reading`] is the rule
|
||||
//! [`run_attributed`](super::run_attributed) applies to it.
|
||||
//! [`run_attributed`](super::run_attributed) applies to it. And the same read
|
||||
//! of any same-uid process, live or exited-unreaped, is what the
|
||||
//! [`TreeMeter`](super::TreeMeter) walks a process tree with
|
||||
//! ([`read_usage`]).
|
||||
|
||||
use super::liveness::{counter_admits, counter_run, counter_status_of, counter_tick, gated, RailWatch, Sample, PROBE_TICK};
|
||||
use super::tree::Usage;
|
||||
use super::{CounterStatus, Probe, ProbeError, CPU_RAIL_LIVENESS_BUDGET};
|
||||
use sandbox_schema::DeviceClass;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
|
@ -37,8 +41,9 @@ const RUSAGE_INFO_V6: libc::c_int = 6;
|
|||
|
||||
/// `struct rusage_info_v6` from the macOS 27 SDK's `<sys/resource.h>`, field
|
||||
/// for field. Apple appends to this struct in new versions and never
|
||||
/// reorders a published one; the layout is pinned by a test (464 bytes,
|
||||
/// `ri_energy_nj` at byte 336, as the SDK's own `offsetof` gives them).
|
||||
/// reorders a published one; the layout is pinned by a test (464 bytes, and
|
||||
/// the byte offset of every field [`read_usage`] reads, as the SDK's own
|
||||
/// `offsetof` gives them).
|
||||
#[repr(C)]
|
||||
#[derive(Clone, Copy)]
|
||||
struct RusageInfoV6 {
|
||||
|
|
@ -96,25 +101,19 @@ struct RusageInfoV6 {
|
|||
ri_reserved: [u64; 6],
|
||||
}
|
||||
|
||||
/// What this module reads of one process's `rusage_info_v6`.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub(super) struct Usage {
|
||||
/// `ri_energy_nj`: the CPU energy the kernel attributes to the process's
|
||||
/// own threads, cumulative, in nanojoules.
|
||||
pub(super) energy_nj: u64,
|
||||
/// `ri_user_time + ri_system_time`: the process's own CPU time (Mach
|
||||
/// absolute-time units; only compared with zero here).
|
||||
pub(super) own_cpu: u64,
|
||||
/// `ri_child_user_time + ri_child_system_time`: the CPU time of the
|
||||
/// descendants the process reaped (same units). Their energy is in no
|
||||
/// field at all.
|
||||
pub(super) reaped_descendants_cpu: u64,
|
||||
}
|
||||
|
||||
/// One `proc_pid_rusage(pid, RUSAGE_INFO_V6)` read: of a live process, or of
|
||||
/// an exited child that has not yet been reaped. `None` if the call fails
|
||||
/// an exited process that has not yet been reaped. `None` if the call fails
|
||||
/// (no such process, or not one this process may read).
|
||||
pub(super) fn usage_of(pid: libc::pid_t) -> Option<Usage> {
|
||||
read_usage(pid).ok()
|
||||
}
|
||||
|
||||
/// [`usage_of`], with the call's `errno` when it fails: `EPERM` for a process
|
||||
/// of another user (listing it needs no privilege; reading it does), `ESRCH`
|
||||
/// for one that does not exist or was reaped. A [`TreeMeter`](super::TreeMeter)
|
||||
/// tells the two apart: the first is a member it cannot read, the second a
|
||||
/// member that is gone.
|
||||
pub(super) fn read_usage(pid: libc::pid_t) -> Result<Usage, i32> {
|
||||
// SAFETY: an all-zero `RusageInfoV6` is a valid value of a struct of
|
||||
// plain integers.
|
||||
let mut info: RusageInfoV6 = unsafe { std::mem::zeroed() };
|
||||
|
|
@ -123,10 +122,16 @@ pub(super) fn usage_of(pid: libc::pid_t) -> Option<Usage> {
|
|||
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(|| Usage {
|
||||
if rc != 0 {
|
||||
return Err(std::io::Error::last_os_error().raw_os_error().unwrap_or(0));
|
||||
}
|
||||
Ok(Usage {
|
||||
energy_nj: info.ri_energy_nj,
|
||||
own_cpu: info.ri_user_time.saturating_add(info.ri_system_time),
|
||||
reaped_descendants_cpu: info.ri_child_user_time.saturating_add(info.ri_child_system_time),
|
||||
start_abstime: info.ri_proc_start_abstime,
|
||||
exit_abstime: info.ri_proc_exit_abstime,
|
||||
neural_lifetime_max: info.ri_lifetime_max_neural_footprint,
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -227,26 +232,46 @@ mod tests {
|
|||
|
||||
/// The struct is the SDK's: the sizes and offsets the macOS 27 SDK's own
|
||||
/// `sizeof`/`offsetof` give (measured with a C program against
|
||||
/// `<sys/resource.h>`).
|
||||
/// `<sys/resource.h>`), for every field [`read_usage`] reads.
|
||||
#[test]
|
||||
fn the_rusage_struct_is_the_sdks_layout() {
|
||||
use std::mem::offset_of;
|
||||
assert_eq!(std::mem::size_of::<RusageInfoV6>(), 464);
|
||||
assert_eq!(std::mem::offset_of!(RusageInfoV6, ri_energy_nj), 336);
|
||||
assert_eq!(std::mem::offset_of!(RusageInfoV6, ri_penergy_nj), 344);
|
||||
assert_eq!(std::mem::offset_of!(RusageInfoV6, ri_child_user_time), 96);
|
||||
assert_eq!(offset_of!(RusageInfoV6, ri_user_time), 16);
|
||||
assert_eq!(offset_of!(RusageInfoV6, ri_system_time), 24);
|
||||
assert_eq!(offset_of!(RusageInfoV6, ri_proc_start_abstime), 80);
|
||||
assert_eq!(offset_of!(RusageInfoV6, ri_proc_exit_abstime), 88);
|
||||
assert_eq!(offset_of!(RusageInfoV6, ri_child_user_time), 96);
|
||||
assert_eq!(offset_of!(RusageInfoV6, ri_child_system_time), 104);
|
||||
assert_eq!(offset_of!(RusageInfoV6, ri_energy_nj), 336);
|
||||
assert_eq!(offset_of!(RusageInfoV6, ri_penergy_nj), 344);
|
||||
assert_eq!(offset_of!(RusageInfoV6, ri_lifetime_max_neural_footprint), 376);
|
||||
}
|
||||
|
||||
/// This process can read itself, and a pid that does not exist reads
|
||||
/// nothing.
|
||||
/// nothing. A live process has a start time and no exit time.
|
||||
#[test]
|
||||
fn this_process_reads_itself_and_a_missing_pid_reads_nothing() {
|
||||
let me = usage_of(unsafe { libc::getpid() }).expect("this process reads itself");
|
||||
assert!(me.own_cpu > 0, "{me:?}");
|
||||
assert!(me.start_abstime > 0 && me.exit_abstime == 0, "{me:?}");
|
||||
assert_eq!(usage_of(-7), None);
|
||||
}
|
||||
|
||||
/// The errno tells a process of another user from one that is gone:
|
||||
/// `launchd` (root) is listed to anyone and read by no one else; a pid
|
||||
/// past the kernel's range names no process.
|
||||
#[test]
|
||||
fn a_read_names_why_it_failed() {
|
||||
// SAFETY: no preconditions.
|
||||
if unsafe { libc::geteuid() } != 0 {
|
||||
assert_eq!(read_usage(1), Err(libc::EPERM), "launchd read by a non-root user");
|
||||
}
|
||||
assert_eq!(read_usage(99_999_999), Err(libc::ESRCH));
|
||||
}
|
||||
|
||||
fn child(energy_nj: u64, reaped_descendants_cpu: u64) -> Usage {
|
||||
Usage { energy_nj, own_cpu: 1_000_000, reaped_descendants_cpu }
|
||||
Usage { energy_nj, own_cpu: 1_000_000, reaped_descendants_cpu, ..Usage::default() }
|
||||
}
|
||||
|
||||
/// A single-process child's own energy is a reading only while the
|
||||
|
|
|
|||
1543
sandbox/sandbox-rs/crates/sandbox-meter/src/probe/tree.rs
Normal file
1543
sandbox/sandbox-rs/crates/sandbox-meter/src/probe/tree.rs
Normal file
File diff suppressed because it is too large
Load diff
637
sandbox/sandbox-rs/crates/sandbox-meter/tests/tree_energy.rs
Normal file
637
sandbox/sandbox-rs/crates/sandbox-meter/tests/tree_energy.rs
Normal file
|
|
@ -0,0 +1,637 @@
|
|||
//! 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), 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) 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]
|
||||
}
|
||||
|
||||
/// `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()))
|
||||
}
|
||||
|
||||
/// 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
|
||||
}
|
||||
|
||||
/// Each of `pids`' rusage once every one of them is quiet by the oracle's
|
||||
/// rule: not running or queued (or 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) || 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.
|
||||
#[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 = energy_nj(&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 = energy_nj(&quiet(&[pid])[0]);
|
||||
let oracle = after - before;
|
||||
log(&format!("single process, loop {n}"), rep, &r, oracle);
|
||||
assert!(!reply.is_empty());
|
||||
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 still running (a gap), exited and unreaped
|
||||
/// (`Exited`), or reaped (`ReapedDescendant` on the server). So no reading
|
||||
/// is 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; the tree is quiet
|
||||
// before the first span.
|
||||
let _ = s.ask("warm up");
|
||||
let tree = ps_tree(s.pid());
|
||||
assert_eq!(tree.len(), 61, "setup: the server and its 60 idle children: {tree:?}");
|
||||
quiet(&tree);
|
||||
let meter = s.meter();
|
||||
let (mut exited, mut reaped, mut unsettled) = (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
|
||||
);
|
||||
assert!(!r.is_whole(), "a reading without the worker born in the span read whole: {r:?}");
|
||||
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");
|
||||
}
|
||||
|
||||
/// 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");
|
||||
}
|
||||
}
|
||||
|
||||
/// 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue