🧪 test(bench): decide the measurement contracts in peryx-bench-core

crates/peryx-bench-core/src/compare.rs

@@ -87,18 +87,24 @@ fn party(table: &Table) -> Option<usize> {
     table.parties.iter().position(|entry| entry.name == PARTY)
 }
 
+/// One row of the printed table. The change is a percentage oriented so above zero reads worse,
+/// and an ungated row carries the reason it does not count.
+fn describe(change: &Change) -> String {
+    let delta = (change.worse - 1.0) * 100.0;
+    let flag = if change.gated { "" } else { change.reason };
+    format!(
+        "{:<18} {:<34} {:>12.3} {:>12.3} {:>+8.1}%  {flag}",
+        change.table, change.row, change.base, change.head, delta
+    )
+}
+
 fn verdict(changes: &[Change]) -> bool {
     println!(
         "\n{:<18} {:<34} {:>12} {:>12} {:>9}  flag",
         "table", "metric", "base", "head", "change"
     );
     for change in changes {
-        let delta = (change.worse - 1.0) * 100.0;
-        let flag = if change.gated { "" } else { change.reason };
-        println!(
-            "{:<18} {:<34} {:>12.3} {:>12.3} {:>+8.1}%  {flag}",
-            change.table, change.row, change.base, change.head, delta
-        );
+        println!("{}", describe(change));
     }
     let kept: Vec<f64> = changes
         .iter()

crates/peryx-bench-core/src/machine.rs

@@ -239,10 +239,14 @@ fn volumes(scratch: &Path) -> Vec<Volume> {
 /// answer would be the read-only system volume instead of the disk the bytes land on.
 fn mount_for<'a>(disks: &'a Disks, path: &Path) -> Option<&'a sysinfo::Disk> {
     reported_mount(path)
-        .and_then(|mount| disks.list().iter().find(|disk| disk.mount_point() == Path::new(&mount)))
+        .and_then(|mount| exact_mount(disks, Path::new(&mount)))
         .or_else(|| longest_prefix(disks, path))
 }
 
+fn exact_mount<'a>(disks: &'a Disks, mount: &Path) -> Option<&'a sysinfo::Disk> {
+    disks.list().iter().find(|disk| disk.mount_point() == mount)
+}
+
 fn longest_prefix<'a>(disks: &'a Disks, path: &Path) -> Option<&'a sysinfo::Disk> {
     disks
         .list()
@@ -356,8 +360,15 @@ fn summarize(mut samples: Vec<f64>) -> String {
     format!("{} ±{spread:.0}%", rate(median))
 }
 
-fn memory_copy(workers: usize, bytes: usize) -> f64 {
+/// What one worker moves, and the total that reaches the reported rate. The total rounds `bytes`
+/// down to a multiple of `workers` so every stream carries the same load.
+const fn shares(bytes: usize, workers: usize) -> (usize, usize) {
     let each = bytes / workers;
+    (each, each * workers)
+}
+
+fn memory_copy(workers: usize, bytes: usize) -> f64 {
+    let (each, moved) = shares(bytes, workers);
     let mut buffers: Vec<(Vec<u8>, Vec<u8>)> = (0..workers).map(|_| (vec![7u8; each], vec![0u8; each])).collect();
     // Fault every destination page in before the clock starts. A freshly allocated `Vec` is untouched
     // zero pages, so the first write to each takes a page fault, and timing that measures the virtual
@@ -373,12 +384,12 @@ fn memory_copy(workers: usize, bytes: usize) -> f64 {
     });
     let elapsed = start.elapsed().as_secs_f64();
     std::hint::black_box(&buffers);
-    throughput(each * workers, elapsed)
+    throughput(moved, elapsed)
 }
 
 fn disk_write(scratch: &Path, workers: usize, bytes: usize, chunk_bytes: usize) -> anyhow::Result<f64> {
     let directory = tempfile::tempdir_in(scratch)?;
-    let each = bytes / workers;
+    let (each, moved) = shares(bytes, workers);
     let chunk = vec![7u8; chunk_bytes];
     let mut outcomes = Vec::with_capacity(workers);
     let start = Instant::now();
@@ -397,21 +408,24 @@ fn disk_write(scratch: &Path, workers: usize, bytes: usize, chunk_bytes: usize)
     for outcome in outcomes {
         outcome?;
     }
-    Ok(throughput(each * workers, elapsed))
+    Ok(throughput(moved, elapsed))
 }
 
 fn write_one(path: &Path, chunk: &[u8], bytes: usize) -> anyhow::Result<()> {
     let mut file = File::create(path).with_context(|| format!("cannot create {}", path.display()))?;
-    let mut written = 0;
-    while written < bytes {
-        let span = chunk.len().min(bytes - written);
+    for span in spans(bytes, chunk.len()) {
         file.write_all(&chunk[..span])?;
-        written += span;
     }
     file.sync_all().context("the write did not reach the device")?;
     Ok(())
 }
 
+/// The writes that cover `bytes`, each at most `chunk` long and the last one short. Driving the
+/// walk off the offsets rather than a running total keeps every write bounded by the buffer.
+fn spans(bytes: usize, chunk: usize) -> impl Iterator<Item = usize> {
+    (0..bytes).step_by(chunk).map(move |written| chunk.min(bytes - written))
+}
+
 fn page_cache_read(scratch: &Path, workers: usize, bytes: usize, chunk_bytes: usize) -> anyhow::Result<f64> {
     let directory = tempfile::tempdir_in(scratch)?;
     let path = directory.path().join("read");
@@ -474,13 +488,14 @@ async fn loopback_http(clients: usize, payload_bytes: usize) -> anyhow::Result<f
         let (http, url) = (http.clone(), url.clone());
         streams.push(tokio::spawn(async move { drain(&http, &url, payload_bytes).await }));
     }
+    let mut moved = Vec::with_capacity(clients);
     for stream in streams {
-        stream.await??;
+        moved.push(stream.await??);
     }
     let elapsed = start.elapsed().as_secs_f64();
     let _ = stop.send(());
     serving.join().expect("the loopback server thread does not panic")?;
-    Ok(throughput(payload_bytes * clients, elapsed))
+    Ok(throughput(moved.iter().sum(), elapsed))
 }
 
 /// Serve the payload on a private runtime until told to stop, reporting the bound address back.
@@ -511,7 +526,9 @@ async fn serve_loopback(
     }
 }
 
-async fn drain(http: &reqwest::Client, url: &str, payload_bytes: usize) -> anyhow::Result<()> {
+/// Reports the bytes it read, so the rate is computed from what arrived rather than from what the
+/// caller asked for.
+async fn drain(http: &reqwest::Client, url: &str, payload_bytes: usize) -> anyhow::Result<usize> {
     let mut response = http
         .get(url)
         .send()
@@ -525,7 +542,7 @@ async fn drain(http: &reqwest::Client, url: &str, payload_bytes: usize) -> anyho
         total == payload_bytes,
         "loopback served {total} bytes, expected {payload_bytes}"
     );
-    Ok(())
+    Ok(total)
 }
 
 #[expect(clippy::cast_precision_loss, reason = "byte counts here fit f64 exactly")]

crates/peryx-bench-core/src/servers.rs

@@ -5,12 +5,16 @@
 use std::os::unix::process::CommandExt as _;
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use anyhow::{Context as _, bail};
 
 use crate::context::BenchmarkContext;
 
+#[cfg(test)]
+#[path = "../tests/unit/servers.rs"]
+mod tests;
+
 /// How long a server gets to answer its first request (uvx may resolve an environment first).
 const START_TIMEOUT: Duration = Duration::from_mins(3);
 
@@ -75,7 +79,7 @@ impl Drop for Active {
             // gunicorn forks workers, and a `uvx` shim execs its payload: killing the direct child
             // orphans the rest, which then linger holding CPU and skewing every later measurement.
             // The child leads its own process group (see `start`), so signal the whole group.
-            kill_process_group(&process);
+            let _ = kill_process_group(&process);
             let _ = process.kill();
             let _ = process.wait();
         }
@@ -87,14 +91,15 @@ impl Drop for Active {
 
 // Shelling out to `kill -KILL -<pgid>` took the whole GitHub-hosted runner down with the group after
 // every cold build, three runs out of three; the syscall reaches the group and nothing else.
-fn kill_process_group(process: &Child) {
+fn kill_process_group(process: &Child) -> std::io::Result<()> {
     #[cfg(unix)]
     {
         let group = rustix::process::Pid::from_child(process);
-        let _ = rustix::process::kill_process_group(group, rustix::process::Signal::KILL);
+        rustix::process::kill_process_group(group, rustix::process::Signal::KILL)?;
     }
     #[cfg(not(unix))]
     let _ = process;
+    Ok(())
 }
 
 impl Server {
@@ -170,15 +175,20 @@ impl Server {
 impl Active {
     async fn wait_ready(&mut self, client: &reqwest::Client, policy: StartupPolicy) -> anyhow::Result<()> {
         let probe = self.probe_url.clone();
-        let deadline = Instant::now() + policy.timeout;
-        while Instant::now() < deadline {
-            self.ensure_running()?;
-            // Any HTTP status means the server is up and routing; only transport errors retry.
-            if client.get(&probe).timeout(policy.request_timeout).send().await.is_ok() {
-                return Ok(());
+        let polling = async {
+            loop {
+                self.ensure_running()?;
+                // Any HTTP status means the server is up and routing; only transport errors retry.
+                if client.get(&probe).timeout(policy.request_timeout).send().await.is_ok() {
+                    return anyhow::Ok(());
+                }
+                self.ensure_running()?;
+                tokio::time::sleep(policy.poll_interval).await;
             }
-            self.ensure_running()?;
-            tokio::time::sleep(policy.poll_interval).await;
+        };
+        let outcome = tokio::time::timeout(policy.timeout, polling).await;
+        if let Ok(ready) = outcome {
+            return ready;
         }
         self.ensure_running()?;
         bail!("server never answered at {probe}")

crates/peryx-bench-core/src/usage.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, SyncSender, channel, sync_channel};
@@ -153,35 +154,40 @@ fn process_tree_sample(system: &System, root: Pid) -> anyhow::Result<(u64, u64)>
         .filter_map(|pid| system.process(*pid))
         .map(|process| f64::from(process.cpu_usage()))
         .sum();
-    #[expect(
-        clippy::cast_possible_truncation,
-        clippy::cast_sign_loss,
-        reason = "usage percent over a 200ms tick is small and non-negative"
-    )]
-    Ok((rss, (usage / 100.0 * SAMPLE_INTERVAL.as_secs_f64() * 1000.0) as u64))
+    Ok((rss, cpu_millis(usage)))
+}
+
+/// The milliseconds of CPU a percentage observed over one sample tick stands for.
+#[expect(
+    clippy::cast_possible_truncation,
+    clippy::cast_sign_loss,
+    reason = "usage percent over a 200ms tick is small and non-negative"
+)]
+fn cpu_millis(percent: f64) -> u64 {
+    (percent / 100.0 * SAMPLE_INTERVAL.as_secs_f64() * 1000.0) as u64
 }
 
 fn record(sample: (u64, u64), peak_rss: &AtomicU64, cpu_millis: &AtomicU64) {
     peak_rss.fetch_max(sample.0, Ordering::Relaxed);
     cpu_millis.fetch_add(sample.1, Ordering::Relaxed);
 }
 
+/// Walking down from the root and taking each parent's children once means a cycle in the process
+/// table cannot spin, and each process is visited once rather than once per descendant.
 fn tree_of(system: &System, root: Pid) -> Vec<Pid> {
-    system
-        .processes()
-        .keys()
-        .filter(|&&pid| {
-            let mut cursor = pid;
-            loop {
-                if cursor == root {
-                    return true;
-                }
-                match system.process(cursor).and_then(sysinfo::Process::parent) {
-                    Some(parent) if parent != cursor => cursor = parent,
-                    _ => return false,
-                }
-            }
-        })
-        .copied()
-        .collect()
+    let mut children: HashMap<Pid, Vec<Pid>> = HashMap::new();
+    for (pid, process) in system.processes() {
+        if let Some(parent) = process.parent() {
+            children.entry(parent).or_default().push(*pid);
+        }
+    }
+    let mut tree = vec![root];
+    let mut pending = vec![root];
+    while let Some(pid) = pending.pop() {
+        for child in children.remove(&pid).into_iter().flatten() {
+            tree.push(child);
+            pending.push(child);
+        }
+    }
+    tree
 }

crates/peryx-bench-core/tests/servers.rs

@@ -22,16 +22,23 @@ fn test_server(_: &BenchmarkContext, port: u16, state: &Path) -> Command {
     let requests = if state.join("setup").exists() { 1 } else { 2 };
     let body = std::fs::read(state.join("index.html")).expect("fixture body exists");
     let listener = std::net::TcpListener::bind(("127.0.0.1", port)).expect("fixture server binds");
+    let address = listener.local_addr().expect("fixture server reports its address");
     let handle = std::thread::spawn(move || serve_fixture(&listener, &body, requests));
-    FIXTURE_THREADS
-        .lock()
-        .expect("fixture thread registry locks")
-        .insert(state.to_path_buf(), FixtureThread { handle });
+    FIXTURE_THREADS.lock().expect("fixture thread registry locks").insert(
+        state.to_path_buf(),
+        FixtureThread {
+            handle,
+            address,
+            requests,
+        },
+    );
     long_running_command()
 }
 
 struct FixtureThread {
     handle: JoinHandle<()>,
+    address: std::net::SocketAddr,
+    requests: usize,
 }
 
 static FIXTURE_THREADS: LazyLock<Mutex<HashMap<PathBuf, FixtureThread>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
@@ -51,6 +58,11 @@ impl Drop for FixtureThreadGuard {
             .expect("fixture thread registry locks")
             .remove(&self.0)
             .expect("fixture thread was registered");
+        // Release every accept the test never reached, so a server that failed to become ready ends
+        // the fixture thread instead of leaving this join to hang.
+        for _ in 0..thread.requests {
+            let _ = std::net::TcpStream::connect(thread.address);
+        }
         thread.handle.join().expect("fixture thread joins");
     }
 }
@@ -166,8 +178,9 @@ async fn server_waits_until_http_is_ready() {
         .text()
         .await
         .unwrap();
+    let pid = active.pid();
     drop(fixture_thread);
-    assert_eq!(body, "ready");
+    assert_eq!((body.as_str(), pid.is_some()), ("ready", true));
 }
 
 #[tokio::test]
@@ -183,7 +196,9 @@ async fn server_reports_early_exit_and_log() {
         .await
         .err()
         .expect("early exit fails");
-    assert!(error.to_string().contains("peryx-invalid-argument"), "{error:#}");
+    let reported = format!("{error:#}");
+    assert!(reported.contains("server exited early with"));
+    assert!(reported.contains("peryx-invalid-argument"));
 }
 
 #[tokio::test]
@@ -253,13 +268,14 @@ fn serve_fixture(listener: &std::net::TcpListener, body: &[u8], requests: usize)
     for stream in listener.incoming().take(requests) {
         let mut stream = stream.expect("fixture accepts a request");
         let mut request = [0; 1024];
-        let _ = stream.read(&mut request).expect("fixture reads a request");
-        write!(
+        // A caller that only wanted to release this accept closes without reading, so neither half
+        // of the exchange is an assertion; the tests assert on what the client received.
+        let _ = stream.read(&mut request);
+        let _ = write!(
             stream,
             "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
             body.len()
-        )
-        .expect("fixture writes headers");
-        stream.write_all(body).expect("fixture writes the body");
+        );
+        let _ = stream.write_all(body);
     }
 }