bench/examples/multiuser/mod.rs
//! **`gunnar.multiuser_scaling` — do authenticated operations scale with the
//! number of distinct users?**
//!
//! Every other concurrency arm in this suite drives N clients through ONE
//! credential (`vs_forge_concurrent`) or none (`vs_git_clone_concurrent`).
//! This one drives the population a real server holds after bootstrap — the
//! 102 principals of `crates/gunnar-cli/tests/control_cli.rs`'s scale test:
//! `alice`, `bob`, `user1`…`user100` — each with its OWN access token, so
//! every request runs the full authenticated smart-HTTP path including
//! [`AuthTable::touch_token`], the one-write-per-request CAS that
//! `gunnar-http/src/service.rs` documents as the serving tier's measured
//! ceiling (573 → 1416 req/s at 16-way ls-refs when skipped, the desktop 2026-08-11).
//! The arm measures WITH the touch in place, because that is what ships, and it
//! counts the ceiling's own signature out of the server's log: CAS retries and
//! silently dropped `last_used` updates per rung.
//!
//! # The corpus — real repositories, pinned
//!
//! Four real GitHub Rust projects, small and medium, each pinned to a released
//! tag's commit so every run serves the identical object set:
//!
//! | member | pin (tag) | objects |
//! |---|---|---|
//! | anyhow | 1.0.104 | 3 726 |
//! | thiserror | 2.0.20 | 4 002 |
//! | serde | v1.0.229 | 30 832 |
//! | clap | clap_complete_nushell-v4.6.2 | 58 139 |
//!
//! The mirrors live under [`corpus_root`] and are NOT cloned by the arm: a
//! benchmark that reaches the network mid-run flakes with the network, so a
//! missing mirror is a refusal naming the exact `git clone --bare` to run. The
//! seeded state is pinned by the COMMIT, not by the mirror's tips: seeding
//! pushes `<pin>:refs/heads/main` and nothing else, so upstream drift in the
//! mirror can never leak into the measured object set.
//!
//! # The op mix — identical per user, documented here
//!
//! Every participating user performs the same ten operations, in a per-user
//! rotation of one fixed multiset (so op kinds interleave across users rather
//! than phase-locking):
//!
//! ```text
//! 5 × ls-remote (ref advertisement — a forge's dominant request shape,
//! and the no-packfile path the touch_token ceiling was
//! measured on)
//! 2 × fetch (up-to-date negotiation against a local clone)
//! 2 × clone (full bare clone, verified object-for-object)
//! 1 × push (one small unique commit to the user's own scratch repo)
//! ```
//!
//! The mix is identical at every rung, so a rung-to-rung ratio compares the
//! same workload at different concurrency and nothing else. Users are assigned
//! to corpus members round-robin (`user index % 4`).
//!
//! # The ramp
//!
//! `GUNNAR_BENCH_MULTIUSER_CONC` (default `4,8,16,32,64,102`) names the
//! rungs. Rung K runs the first K users of the roster concurrently, one OS
//! thread per user, each thread performing its ten operations sequentially
//! with stock `git`. Throughput is `K × 10 / wall`; the scaling verdict is
//! `ops_sec(K) / ops_sec(first rung)` published per rung as `scale_x`, with
//! per-user efficiency `scale_x / K` beside it.
//!
//! **The default ladder starts at 4, and that is a correctness choice.** A
//! user's repository is `user index % 4`, so below four users the corpus mix
//! is a SLICE — rungs 1 and 2 see only the two small members, and the first
//! rung with clap and serde in it then shows a "p99 stall" that is nothing
//! but a warm clap clone (~0.9 s solo, measured 2026-08-12; the 2026-08-12
//! run log's "rung-4 p99 stall" was exactly this misreading). From rung 4
//! up every rung carries the same mix and rung-to-rung ratios compare one
//! workload. Rungs 1 and 2 can still be asked for by env, with that caveat.
//! The per-member clone latencies below are what keep the composition visible
//! in the receipt.
//!
//! # What is measured, and what is not
//!
//! The server is ONE `gunnar serve` process on the host (engine: whatever
//! `GUNNAR_ENGINE` says, read back from the banner's own log line), with CPU
//! read from `/proc/<pid>/stat` and peak RSS sampled from `/proc/<pid>/status`
//! across each rung's window. There is no container and no cgroup cap: this
//! arm asks how ONE binary's throughput moves with user count on shared
//! hardware, a self-relative question, and `load1`/PSI at each rung's start is
//! on the row so a loaded box is visible in the receipt. Cross-engine
//! comparisons belong to the `vs_forge_*` arms and their pinned containers.
//!
//! # Red conditions — each guard has been seen to fire
//!
//! A batch is scored only when every operation exited 0 AND produced the
//! pinned output: an ls-remote must advertise the pin, a clone must land the
//! member's exact object count and tip, a fetch must exit clean, and every
//! push is read back from the server by ref. Two red switches prove the guards
//! fire:
//!
//! | `GUNNAR_BENCH_MULTIUSER_RED=` | what it breaks | which guard must fire |
//! |---|---|---|
//! | `badtoken` | user 0 authenticates with a mangled token | the per-op exit-status guard |
//! | `shallowclone` | user 0 clones at `--depth 1` | the per-clone object-count guard |
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;
use anyhow::{bail, ensure, Context, Result};
use nornir_bench::api::Bencher;
use nornir_bench::BenchResult;
use serde_json::json;
use super::{env_usize, leak, preview, result};
// ───────────────────────────────────────────────────────────────────────────
// The pinned corpus
// ───────────────────────────────────────────────────────────────────────────
struct Member {
name: &'static str,
upstream: &'static str,
/// The commit of a released tag. Tags can be re-pointed upstream; this
/// SHA cannot, and it is the only thing seeding pushes.
pin: &'static str,
/// `git rev-list --objects <pin> | wc -l`, counted 2026-08-12. Asserted
/// against every clone, and against the mirror before the run starts.
objects: u64,
}
const CORPUS: &[Member] = &[
Member {
name: "anyhow",
upstream: "https://github.com/dtolnay/anyhow",
pin: "1dbe1862aae650423e3361fbd20b7d17c5109cc3", // tag 1.0.104
objects: 3726,
},
Member {
name: "thiserror",
upstream: "https://github.com/dtolnay/thiserror",
pin: "b1d5db5e039275d95bf7536a2b2192aeb4dc28bf", // tag 2.0.20
objects: 4002,
},
Member {
name: "serde",
upstream: "https://github.com/serde-rs/serde",
pin: "7fc3b4c30c94f73a96ebd1553f2b090d928fc3a8", // tag v1.0.229
objects: 30832,
},
Member {
name: "clap",
upstream: "https://github.com/clap-rs/clap",
pin: "8e50ec891e30cd2af4f9a62a405a5b7ee31aac4f", // tag clap_complete_nushell-v4.6.2
objects: 58139,
},
];
fn corpus_root() -> PathBuf {
PathBuf::from(
std::env::var("GUNNAR_BENCH_MULTIUSER_CORPUS").unwrap_or_else(|_| {
gunnar_bench::expand_home("~/scratch/cargo/gunnar-multiuser/corpus")
.to_string_lossy()
.into_owned()
}),
)
}
/// Working state: the server's data root, the local clones, the push chains.
/// md1, never `/tmp` — `/tmp` is RAM on the desktop.
fn work_root() -> PathBuf {
PathBuf::from(
std::env::var("GUNNAR_BENCH_MULTIUSER_WORK").unwrap_or_else(|_| {
gunnar_bench::expand_home("~/scratch/cargo/gunnar-multiuser/work")
.to_string_lossy()
.into_owned()
}),
)
}
// ───────────────────────────────────────────────────────────────────────────
// The roster and the mix
// ───────────────────────────────────────────────────────────────────────────
/// The 102 users of assignment 1, in bootstrap order: the two gunnar-admins
/// first, then the hundred. Only the first `max rung` are provisioned.
fn roster(n: usize) -> Vec<String> {
let mut v = vec!["alice".to_string(), "bob".to_string()];
v.extend((1..=100).map(|i| format!("user{i}")));
v.truncate(n.max(1));
v
}
#[derive(Clone, Copy, PartialEq, Debug)]
enum Op {
Ls,
Fetch,
Clone,
Push,
}
/// The documented multiset. Its LENGTH is the per-user op count and its
/// composition is the mix; both are in the module doc and neither is tunable,
/// because a run with a different mix is a different benchmark.
const MIX: [Op; 10] = [
Op::Ls,
Op::Ls,
Op::Ls,
Op::Ls,
Op::Ls,
Op::Fetch,
Op::Fetch,
Op::Clone,
Op::Clone,
Op::Push,
];
/// User `p`'s op order: [`MIX`] rotated by `p`, so at any instant the in-flight
/// ops are a blend of kinds rather than every user cloning at once.
fn schedule(p: usize) -> Vec<Op> {
let r = p % MIX.len();
MIX[r..].iter().chain(MIX[..r].iter()).copied().collect()
}
fn ramp() -> Vec<usize> {
let default = if preview() { "4,8" } else { "4,8,16,32,64,102" };
let raw = std::env::var("GUNNAR_BENCH_MULTIUSER_CONC").unwrap_or_else(|_| default.to_string());
let mut v: Vec<usize> = raw
.split(',')
.filter_map(|s| s.trim().parse().ok())
.filter(|n| (1..=102).contains(n))
.collect();
v.sort_unstable();
v.dedup();
if v.is_empty() {
v.push(1);
}
v
}
fn red(which: &str) -> bool {
std::env::var("GUNNAR_BENCH_MULTIUSER_RED")
.map(|v| v.split(',').any(|s| s.trim() == which))
.unwrap_or(false)
}
// ───────────────────────────────────────────────────────────────────────────
// Process plumbing
// ───────────────────────────────────────────────────────────────────────────
/// Run a command, insisting on exit 0, returning stdout.
fn run(cmd: &mut Command, what: &str) -> Result<String> {
let out = cmd.output().with_context(|| what.to_string())?;
ensure!(
out.status.success(),
"{what} exited {:?}: {}",
out.status.code(),
String::from_utf8_lossy(&out.stderr).trim()
);
Ok(String::from_utf8_lossy(&out.stdout).into_owned())
}
/// `git` with the environment every client here runs under.
fn git(args: &[&str], dir: Option<&Path>) -> Result<String> {
let mut c = Command::new("git");
if let Some(d) = dir {
c.current_dir(d);
}
c.args(args).env("GIT_TERMINAL_PROMPT", "0");
run(&mut c, &format!("git {args:?}"))
}
/// The offline provisioning verbs, against a stopped server's data root.
fn admin(binary: &Path, data: &Path, verbs: &[&str]) -> Result<String> {
let mut c = Command::new(binary);
c.arg("--data-root").arg(data).args(verbs);
run(&mut c, &format!("gunnar {}", verbs.join(" ")))
}
/// The server's CPU seconds so far, from `/proc/<pid>/stat` (utime + stime).
fn proc_cpu_s(pid: u32) -> f64 {
let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).unwrap_or_default();
// comm may contain spaces; fields are positional after the closing paren,
// where utime and stime are the 12th and 13th (0-indexed 11 and 12).
let after = stat.rsplit_once(')').map(|(_, a)| a).unwrap_or("");
let f: Vec<&str> = after.split_whitespace().collect();
let utime: f64 = f.get(11).and_then(|v| v.parse().ok()).unwrap_or(0.0);
let stime: f64 = f.get(12).and_then(|v| v.parse().ok()).unwrap_or(0.0);
(utime + stime) / 100.0
}
fn proc_rss_kb(pid: u32) -> u64 {
let status = std::fs::read_to_string(format!("/proc/{pid}/status")).unwrap_or_default();
status
.lines()
.find(|l| l.starts_with("VmRSS:"))
.and_then(|l| l.split_whitespace().nth(1))
.and_then(|v| v.parse().ok())
.unwrap_or(0)
}
fn load1() -> f64 {
std::fs::read_to_string("/proc/loadavg")
.ok()
.and_then(|s| s.split_whitespace().next().map(str::to_owned))
.and_then(|v| v.parse().ok())
.unwrap_or(-1.0)
}
fn psi_cpu_avg10() -> f64 {
std::fs::read_to_string("/proc/pressure/cpu")
.ok()
.and_then(|s| {
s.lines()
.find(|l| l.starts_with("some"))
.and_then(|l| l.split("avg10=").nth(1))
.and_then(|v| v.split_whitespace().next().map(str::to_owned))
})
.and_then(|v| v.parse().ok())
.unwrap_or(-1.0)
}
// ───────────────────────────────────────────────────────────────────────────
// The server under test
// ───────────────────────────────────────────────────────────────────────────
struct Server {
child: std::process::Child,
http: String,
/// `http` or `https` — which listener [`Server::start_with`] asked for.
scheme: &'static str,
log_path: PathBuf,
/// The engine the server itself logged, read back rather than assumed.
engine: String,
}
impl Server {
fn start(binary: &Path, data: &Path, log_path: PathBuf) -> Result<Server> {
Self::start_with(binary, data, log_path, false)
}
/// `tls` starts the listener with `--http-tls-self`: gunnar terminates TLS
/// itself with a certificate minted from its own issuer, ALPN h2 first —
/// the single-executable deployment shape gunnar is documented to run in, and the
/// one the torture arm runs so every operation pays a real handshake.
fn start_with(binary: &Path, data: &Path, log_path: PathBuf, tls: bool) -> Result<Server> {
let log = std::fs::File::create(&log_path)?;
let mut args = vec![
"serve",
"--ssh-bind",
"127.0.0.1:0",
"--grpc-bind",
"127.0.0.1:0",
"--http-bind",
"127.0.0.1:0",
];
if tls {
args.push("--http-tls-self");
} else {
args.push("--http-allow-cleartext-credentials");
}
let mut child = Command::new(binary)
.arg("--data-root")
.arg(data)
.args(&args)
// `warn` so the doc_txn contention lines (warn from the second
// retry, error on a dropped update) land in the log this arm
// counts, without paying info-per-request formatting — plus the
// one info line the read-back below needs, the store's own
// `store.engine_selected`.
.env("RUST_LOG", "warn,gunnar::store=info")
.stdout(Stdio::piped())
.stderr(Stdio::from(log))
.spawn()
.context("spawning gunnar serve")?;
use std::io::BufRead as _;
let stdout = child.stdout.take().context("piped stdout")?;
let mut reader = std::io::BufReader::new(stdout);
let mut http = None;
let deadline = Instant::now() + std::time::Duration::from_secs(60);
while Instant::now() < deadline {
let mut line = String::new();
match reader.read_line(&mut line) {
Ok(0) | Err(_) => break,
Ok(_) => {}
}
if let Some(rest) = line.strip_prefix("http ") {
http = Some(rest.split_whitespace().next().unwrap_or("").to_string());
}
if let Some(rest) = line.strip_prefix("https ") {
http = Some(rest.split_whitespace().next().unwrap_or("").to_string());
}
// `log` is the banner's last line; once it arrives, everything the
// banner will ever say has been said.
if line.starts_with("log ") {
break;
}
}
let http = http.context(
"gunnar serve never announced an http listener; the binary must be built with \
`--features http` (see forge::gunnar_binary) and the banner must carry the `http` \
line this arm added to serve.rs",
)?;
// Keep draining stdout for the server's life: Rust panics on a failed
// println once the pipe's read end closes (the control_cli suite met
// this as a 1-in-8 flake).
std::thread::spawn(move || {
let mut sink = String::new();
let _ = std::io::Read::read_to_string(&mut reader, &mut sink);
});
// The engine, from the server's own store line in the log — read back,
// not inherited from the environment that asked for it.
let mut engine = String::from("unknown");
for _ in 0..50 {
let text = std::fs::read_to_string(&log_path).unwrap_or_default();
if let Some(pos) = text.find("engine=\"") {
let rest = &text[pos + 8..];
engine = rest.split('"').next().unwrap_or("unknown").to_string();
break;
}
std::thread::sleep(std::time::Duration::from_millis(100));
}
Ok(Server {
child,
http,
scheme: if tls { "https" } else { "http" },
log_path,
engine,
})
}
fn url(&self, user: &str, token: &str, repo: &str) -> String {
format!("{}://{user}:{token}@{}/{repo}.git", self.scheme, self.http)
}
fn log_len(&self) -> u64 {
std::fs::metadata(&self.log_path)
.map(|m| m.len())
.unwrap_or(0)
}
/// The doc_txn contention counters in the log slice written since `from`:
/// `(retries logged at warn, updates dropped after MAX_ATTEMPTS)`.
fn contention_since(&self, from: u64) -> (u64, u64) {
use std::io::{Read as _, Seek as _};
let Ok(mut f) = std::fs::File::open(&self.log_path) else {
return (0, 0);
};
let _ = f.seek(std::io::SeekFrom::Start(from));
let mut text = String::new();
let _ = f.read_to_string(&mut text);
let retries = text
.lines()
.filter(|l| l.contains("changed under us AGAIN"))
.count() as u64;
let dropped = text
.lines()
.filter(|l| l.contains("lost every compare-and-swap"))
.count() as u64;
(retries, dropped)
}
}
impl Drop for Server {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
}
}
// ───────────────────────────────────────────────────────────────────────────
// One user's work
// ───────────────────────────────────────────────────────────────────────────
/// Everything one participant needs before the clock starts.
struct Participant {
name: String,
token: String,
member: &'static Member,
/// A local bare clone of the member, for the up-to-date fetches.
fetch_dir: PathBuf,
/// The local repo holding this user's pre-built push chain.
push_dir: PathBuf,
/// The chain, oldest first; rung r pushes `chain[r]`.
chain: Vec<String>,
/// Their own scratch repository on the server.
scratch: String,
}
/// One finished op: kind, wall ms, which corpus member it hit, and what it
/// printed (for the ls guard).
struct Done {
op: Op,
ms: f64,
member: &'static str,
out: String,
}
/// The ten ops of user `p` at chain index `push_idx`, sequentially.
fn drive(
server: &Server,
part: &Participant,
p: usize,
push_idx: usize,
clone_base: &Path,
) -> Result<Vec<Done>> {
let bad = red("badtoken") && p == 0;
let token = if bad {
format!("{}WRONG", part.token)
} else {
part.token.clone()
};
let url = server.url(&part.name, &token, &format!("corpus/{}", part.member.name));
let scratch_url = server.url(&part.name, &token, &part.scratch);
let mut done = Vec::with_capacity(MIX.len());
let mut clone_no = 0usize;
for op in schedule(p) {
let t = Instant::now();
let out = match op {
Op::Ls => git(&["ls-remote", &url, "refs/heads/main"], None)?,
Op::Fetch => git(&["fetch", "-q", "origin", "main"], Some(&part.fetch_dir))?,
Op::Clone => {
clone_no += 1;
let dir = clone_base.join(format!("{}-{clone_no}", part.name));
let _ = std::fs::remove_dir_all(&dir);
let mut args = vec!["clone", "-q", "--bare"];
// RED SWITCH `shallowclone`: a depth-1 clone is a valid
// repository with a wonderfully small object set. The count
// guard below is what has to see it.
if red("shallowclone") && p == 0 {
args.extend(["--depth", "1"]);
}
let dirs = dir.to_string_lossy().into_owned();
args.push(&url);
args.push(&dirs);
git(&args, None)?;
// Verified per clone, not per batch: one empty clone hiding
// inside a good batch is exactly what survives a sum. The
// check is client-side and ~2 ms; it runs inside the window
// and is priced by being identical at every rung.
let counted = git(&["count-objects", "-v"], Some(&dir))?;
let in_pack: u64 = counted
.lines()
.find_map(|l| l.strip_prefix("in-pack: "))
.and_then(|v| v.trim().parse().ok())
.unwrap_or(0);
let tip = git(&["rev-parse", "refs/heads/main"], Some(&dir))?;
let _ = std::fs::remove_dir_all(&dir);
ensure!(
in_pack == part.member.objects,
"{}: a clone of corpus/{} landed {in_pack} object(s), the pin holds {} — a \
clone that transferred less finishes faster and lifts the rung's ops/sec, \
which is the one way this arm reports a magnificent number having measured \
nothing",
part.name,
part.member.name,
part.member.objects
);
ensure!(
tip.trim() == part.member.pin,
"{}: a clone landed main at {} against the pin {}",
part.name,
tip.trim(),
part.member.pin
);
counted
}
Op::Push => {
let commit = &part.chain[push_idx];
git(
&[
"push",
"-q",
&scratch_url,
&format!("{commit}:refs/heads/main"),
],
Some(&part.push_dir),
)?
}
};
done.push(Done {
op,
ms: t.elapsed().as_secs_f64() * 1000.0,
member: part.member.name,
out,
});
}
Ok(done)
}
/// Nearest-rank percentile over a sorted slice (the same choice
/// `forge::concurrent` documents: an observed latency, never an interpolation).
fn pct(sorted: &[f64], p: f64) -> f64 {
if sorted.is_empty() {
return 0.0;
}
let rank = (p * sorted.len() as f64).ceil().max(1.0) as usize;
sorted[rank.min(sorted.len()) - 1]
}
// ───────────────────────────────────────────────────────────────────────────
// The arm
// ───────────────────────────────────────────────────────────────────────────
struct MultiuserScaling;
impl Bencher for MultiuserScaling {
fn id(&self) -> &'static str {
"gunnar.multiuser_scaling"
}
fn run(&self) -> Result<BenchResult> {
let binary = super::forge::gunnar_binary()?;
let rungs = ramp();
let max = *rungs.last().expect("ramp is never empty");
let users = roster(max);
// ── the corpus is present and holds every pin, or this is a refusal ─
let corpus = corpus_root();
for m in CORPUS {
let mirror = corpus.join(format!("{}.git", m.name));
ensure!(
mirror.join("HEAD").is_file(),
"corpus member {} is not mirrored at {}. Mirror it once:\n git clone --bare {} {}\n\
(the arm never clones from the network itself: a benchmark that reaches the \
network mid-run flakes with the network)",
m.name,
mirror.display(),
m.upstream,
mirror.display()
);
git(
&["cat-file", "-e", &format!("{}^{{commit}}", m.pin)],
Some(&mirror),
)
.with_context(|| {
format!(
"corpus member {} does not hold the pinned commit {}; `git fetch` the mirror",
m.name, m.pin
)
})?;
}
// ── a fresh server: provision offline, then boot ────────────────────
let work = work_root();
let _ = std::fs::remove_dir_all(&work);
std::fs::create_dir_all(&work)?;
let data = work.join("data");
std::fs::create_dir_all(&data)?;
admin(&binary, &data, &["namespace", "create", "corpus"])?;
admin(&binary, &data, &["namespace", "create", "scratch"])?;
for m in CORPUS {
admin(
&binary,
&data,
&[
"repo",
"create",
&format!("corpus/{}", m.name),
"--hash",
"sha1",
],
)?;
}
let mut tokens: Vec<String> = Vec::with_capacity(users.len());
for (p, u) in users.iter().enumerate() {
admin(&binary, &data, &["principal", "add", u])?;
// The first two are the gunnar-admins of assignment 1's roster;
// alice's server-wide admin is also what seeds the corpus.
if p < 2 {
admin(&binary, &data, &["grant", u, "admin", "--server"])?;
}
let minted = admin(&binary, &data, &["token", "add", u])?;
let token = minted
.lines()
.flat_map(|l| l.split_whitespace())
.find(|w| w.starts_with("gunnar_pat_"))
.context("`gunnar token add` printed no gunnar_pat_ token")?
.to_string();
tokens.push(token);
admin(
&binary,
&data,
&["grant", u, "read", "--namespace", "corpus"],
)?;
admin(
&binary,
&data,
&["repo", "create", &format!("scratch/{u}"), "--hash", "sha1"],
)?;
admin(
&binary,
&data,
&["grant", u, "write", "--repo", &format!("scratch/{u}")],
)?;
}
let server = Server::start(&binary, &data, work.join("server.log"))?;
let pid = server.child.id();
// ── seed: exactly the pin, from the mirror, as the first admin ──────
for (i, m) in CORPUS.iter().enumerate() {
let url = server.url(&users[0], &tokens[0], &format!("corpus/{}", m.name));
let _ = i;
git(
&["push", "-q", &url, &format!("{}:refs/heads/main", m.pin)],
Some(&corpus.join(format!("{}.git", m.name))),
)
.with_context(|| format!("seeding corpus/{}", m.name))?;
}
// ── per-participant setup, all outside every measured window ────────
// Proto clones (one per member, from the SERVER, so a fetch negotiates
// against exactly the seeded state), then a cheap local clone per user.
let protos = work.join("proto");
std::fs::create_dir_all(&protos)?;
for m in CORPUS {
let url = server.url(&users[0], &tokens[0], &format!("corpus/{}", m.name));
git(&["clone", "-q", "--bare", &url], Some(&protos))
.with_context(|| format!("proto clone of {}", m.name))?;
std::fs::rename(protos.join(format!("{}.git", m.name)), protos.join(m.name))?;
}
let chain_len = rungs.len() + 1; // one per rung, plus the warm-up
let mut parts: Vec<Participant> = Vec::with_capacity(users.len());
for (p, u) in users.iter().enumerate() {
let member = &CORPUS[p % CORPUS.len()];
let fetch_dir = work.join("fetch").join(u);
std::fs::create_dir_all(fetch_dir.parent().context("fetch parent")?)?;
// A plain local-path clone hardlinks the object store: cheap and
// correct here, since these repos are only ever read.
git(
&[
"clone",
"-q",
"--bare",
&protos.join(member.name).to_string_lossy(),
&fetch_dir.to_string_lossy(),
],
None,
)?;
// `origin` in a local clone points at the proto; the fetches must
// negotiate with the SERVER.
git(
&[
"remote",
"set-url",
"origin",
&server.url(u, &tokens[p], &format!("corpus/{}", member.name)),
],
Some(&fetch_dir),
)?;
let (push_dir, chain) = build_chain(&work, u, chain_len)?;
parts.push(Participant {
name: u.clone(),
token: tokens[p].clone(),
member,
fetch_dir,
push_dir,
chain,
scratch: format!("scratch/{u}"),
});
}
// ── the ramp ────────────────────────────────────────────────────────
let clone_base = work.join("clones");
std::fs::create_dir_all(&clone_base)?;
// One discarded warm-up at FULL breadth, so every user's first-touch
// costs — token first use, per-repo handle caches, allocator growth —
// are paid before any measured window. Seen in the preview run before
// this was full-breadth: rung 4's p99 was 1059 ms against rung 1's 65,
// and it was users 2–4 arriving cold, not concurrency.
run_rung(&server, &parts[..], 0, &clone_base)?;
let mut metrics: Vec<(&'static str, serde_json::Value)> = Vec::new();
let mut first_ops_sec = 0.0f64;
for (ri, k) in rungs.iter().enumerate() {
let k = (*k).min(parts.len());
let label = leak(format!("c{k}"));
let log_from = server.log_len();
let cpu0 = proc_cpu_s(pid);
let rss_peak = Arc::new(AtomicU64::new(0));
let stop = Arc::new(AtomicBool::new(false));
let sampler = {
let rss_peak = Arc::clone(&rss_peak);
let stop = Arc::clone(&stop);
std::thread::spawn(move || {
while !stop.load(Ordering::Relaxed) {
rss_peak.fetch_max(proc_rss_kb(pid), Ordering::Relaxed);
std::thread::sleep(std::time::Duration::from_millis(50));
}
})
};
let load_before = load1();
let psi_before = psi_cpu_avg10();
let started = Instant::now();
let outcome = run_rung(&server, &parts[..k], ri + 1, &clone_base);
let wall_s = started.elapsed().as_secs_f64();
stop.store(true, Ordering::Relaxed);
let _ = sampler.join();
let cpu_s = proc_cpu_s(pid) - cpu0;
let (retries, dropped) = server.contention_since(log_from);
let all = outcome?;
// ── the guards that make the number a number ────────────────────
let ops = all.len() as f64;
ensure!(
all.len() == k * MIX.len(),
"rung {k}: {} op(s) completed out of {} — a batch with missing ops finishes \
sooner and reports HIGHER throughput, so a shortfall is a fictional result, \
never a slow one",
all.len(),
k * MIX.len()
);
for d in &all {
if d.op == Op::Ls {
// The advertisement must carry the pin. An empty 200 or a
// stale tip would exit 0 and read exactly like a serve.
ensure!(
CORPUS.iter().any(|m| d.out.contains(m.pin)),
"rung {k}: an ls-remote advertised none of the four pins: {:?}",
d.out.trim()
);
}
}
// Every push landed ON THE SERVER, asked of the server rather
// than of the client that claims to have sent it.
for part in &parts[..k] {
let url = server.url(&part.name, &part.token, &part.scratch);
let adv = git(&["ls-remote", &url, "refs/heads/main"], None)?;
let got = adv.split_whitespace().next().unwrap_or("");
ensure!(
got == part.chain[ri + 1],
"rung {k}: {}'s scratch main is at {got:?}, the rung pushed {:?}",
part.name,
part.chain[ri + 1]
);
}
let mut lat: Vec<f64> = all.iter().map(|d| d.ms).collect();
lat.sort_by(f64::total_cmp);
let ops_sec = ops / wall_s;
if ri == 0 {
first_ops_sec = ops_sec;
}
let per_kind = |kind: Op| -> f64 {
let mut v: Vec<f64> = all.iter().filter(|d| d.op == kind).map(|d| d.ms).collect();
v.sort_by(f64::total_cmp);
pct(&v, 0.50)
};
let m = |suffix: &str| leak(format!("{label}_{suffix}"));
metrics.push((m("users"), json!(k as f64)));
metrics.push((m("ops"), json!(ops)));
metrics.push((m("wall_ms"), json!(round(wall_s * 1000.0, 1))));
metrics.push((m("ops_sec"), json!(round(ops_sec, 3))));
metrics.push((
m("scale_x"),
json!(round(ops_sec / first_ops_sec.max(1e-9), 3)),
));
metrics.push((
m("efficiency"),
json!(round(
ops_sec / first_ops_sec.max(1e-9) / (k as f64 / rungs[0] as f64),
3
)),
));
metrics.push((m("server_cpu_s"), json!(round(cpu_s, 3))));
metrics.push((m("server_cores_busy"), json!(round(cpu_s / wall_s, 3))));
metrics.push((
m("server_rss_peak_mb"),
json!(round(rss_peak.load(Ordering::Relaxed) as f64 / 1024.0, 1)),
));
metrics.push((m("p50_ms"), json!(round(pct(&lat, 0.50), 2))));
metrics.push((m("p99_ms"), json!(round(pct(&lat, 0.99), 2))));
metrics.push((m("ls_p50_ms"), json!(round(per_kind(Op::Ls), 2))));
metrics.push((m("fetch_p50_ms"), json!(round(per_kind(Op::Fetch), 2))));
metrics.push((m("clone_p50_ms"), json!(round(per_kind(Op::Clone), 2))));
metrics.push((m("push_p50_ms"), json!(round(per_kind(Op::Push), 2))));
// Clone latency SPLIT BY MEMBER, so a repo-mix change between two
// rungs can never again read as a server stall: a clap p50 beside
// an anyhow p50 says which part of a tail is corpus and which is
// concurrency.
for member in CORPUS {
let mut v: Vec<f64> = all
.iter()
.filter(|d| d.op == Op::Clone && d.member == member.name)
.map(|d| d.ms)
.collect();
v.sort_by(f64::total_cmp);
if !v.is_empty() {
metrics.push((
leak(format!("{label}_clone_{}_p50_ms", member.name)),
json!(round(pct(&v, 0.50), 2)),
));
}
}
metrics.push((m("cas_retries"), json!(retries as f64)));
metrics.push((m("last_used_dropped"), json!(dropped as f64)));
metrics.push((m("load1_before"), json!(load_before)));
metrics.push((m("psi_cpu_avg10_before"), json!(psi_before)));
}
metrics.push(("users_total", json!(users.len() as f64)));
metrics.push(("ops_per_user", json!(MIX.len() as f64)));
metrics.push((
"mix",
json!(
"per user: 5 ls-remote, 2 up-to-date fetch, 2 verified bare clone, 1 push of \
one unique commit; multiset rotated by user index"
),
));
metrics.push((
"corpus",
json!(CORPUS
.iter()
.map(|m| format!("{}@{} ({} objects)", m.name, &m.pin[..12], m.objects))
.collect::<Vec<_>>()
.join(", ")),
));
metrics.push(("engine", json!(server.engine.clone())));
metrics.push((
"harness",
json!(
"one gunnar serve process on the host, no cgroup cap; every request \
token-authenticated smart HTTP (touch_token IN PLACE); CPU from \
/proc/<pid>/stat, RSS sampled from /proc/<pid>/status; self-relative \
scaling, not a cross-engine comparison"
),
));
Ok(result(self.id(), metrics))
}
}
nornir_bench::register_bench!(MultiuserScaling);
/// One rung: the first `parts.len()` users, each running their ten ops on an
/// OS thread of their own. `push_idx` names which chain commit this rung
/// pushes (0 is the warm-up's).
fn run_rung(
server: &Server,
parts: &[Participant],
push_idx: usize,
clone_base: &Path,
) -> Result<Vec<Done>> {
let mut out: Vec<Done> = Vec::new();
let mut failures: Vec<String> = Vec::new();
std::thread::scope(|scope| {
let mut handles = Vec::with_capacity(parts.len());
for (p, part) in parts.iter().enumerate() {
handles.push(scope.spawn(move || drive(server, part, p, push_idx, clone_base)));
}
for h in handles {
match h.join() {
Ok(Ok(done)) => out.extend(done),
Ok(Err(e)) => failures.push(format!("{e:#}")),
Err(_) => failures.push("a client thread panicked".to_string()),
}
}
});
ensure!(
failures.is_empty(),
"{} of {} user(s) did not complete their ops: {}",
failures.len(),
parts.len(),
failures.join(" | ")
);
Ok(out)
}
/// Build user `u`'s push chain: `len` commits, each adding one tiny unique
/// blob, via one `git fast-import` stream — one child process per user rather
/// than four per commit. Returns the repo and the commit ids, oldest first.
fn build_chain(work: &Path, u: &str, len: usize) -> Result<(PathBuf, Vec<String>)> {
let dir = work.join("push").join(u);
std::fs::create_dir_all(&dir)?;
git(&["init", "-q", "--bare", "."], Some(&dir))?;
let mut stream = String::new();
for i in 0..len {
stream.push_str(&format!(
"commit refs/heads/chain\n\
committer {u} <{u}@bench.invalid> {} +0000\n\
data <<EOT\nmultiuser {u} {i}\nEOT\n",
1_700_000_000 + i
));
if i > 0 {
// fast-import continues from the branch head automatically.
}
stream.push_str(&format!(
"M 100644 inline payload-{i}.txt\ndata <<EOT\n{u} payload {i}\nEOT\n\n"
));
}
let mut c = Command::new("git");
c.current_dir(&dir)
.args(["fast-import", "--quiet"])
.stdin(Stdio::piped())
.stdout(Stdio::null())
.stderr(Stdio::piped());
let mut child = c.spawn().context("git fast-import")?;
{
use std::io::Write as _;
child
.stdin
.as_mut()
.context("fast-import stdin")?
.write_all(stream.as_bytes())?;
}
let status = child.wait()?;
ensure!(status.success(), "git fast-import for {u} failed");
let list = git(&["rev-list", "--reverse", "refs/heads/chain"], Some(&dir))?;
let chain: Vec<String> = list.split_whitespace().map(str::to_owned).collect();
ensure!(
chain.len() == len,
"{u}'s chain holds {} commit(s), wanted {len}",
chain.len()
);
Ok((dir, chain))
}
fn round(v: f64, places: i32) -> f64 {
let f = 10f64.powi(places);
(v * f).round() / f
}
// ═══════════════════════════════════════════════════════════════════════════
// THE POPULATION AXIS — 1 → 100 000 users
// ═══════════════════════════════════════════════════════════════════════════
//
// `gunnar.multiuser_scaling` above scales CONCURRENCY at a fixed population of
// 102. These two arms scale the POPULATION — how many principals, tokens and
// grants the server carries — at a fixed, box-sized concurrency, which is the
// half of "1 to 100 000 users" no concurrency ramp can answer: does a request
// get slower because 99 968 *other* users exist?
//
// Provisioning goes through `AuthTable::bulk_add_users` in-process — one
// document CAS for the whole population — because per-user CLI spawns at 100k
// would be ~500 000 processes and the bench would be measuring provisioning.
// The bulk import's wall time is itself a published metric (`provision_s`):
// "how long until a fresh appliance holds 100k users" is part of the story.
//
// What each population rung measures, all on one host `gunnar serve`:
// * `first_ls_ms` — the first authenticated request, which pays the auth
// document's first load at that population;
// * `ls_req_s` etc. — 32 concurrent workers, each a DISTINCT user sampled
// evenly across the population (so token lookups spread
// over the identity shards), driving ls-remote;
// * a small clone sample, for the transfer path;
// * server RSS, CAS retries, dropped `last_used` — the shared-document
// counters that found the touch_token wall.
//
// The write arm creates repositories only for its ≤32 active writers: the
// population axis is the auth table, not 100 000 store roots.
/// The populations, smallest first. `GUNNAR_BENCH_POP` overrides.
fn populations() -> Vec<usize> {
let default = if preview() {
"100,2000"
} else {
"1000,10000,50000,100000"
};
let raw = std::env::var("GUNNAR_BENCH_POP").unwrap_or_else(|_| default.to_string());
let mut v: Vec<usize> = raw
.split(',')
.filter_map(|s| s.trim().parse().ok())
.filter(|n| (1..=1_000_000).contains(n))
.collect();
v.sort_unstable();
v.dedup();
if v.is_empty() {
v.push(100);
}
v
}
/// Deterministic bench token secret for population user `i`. Scratch-root
/// credentials for a throwaway server; minted per run would be 100k RNG calls
/// for no property this arm measures.
fn pop_secret(i: usize) -> String {
format!("gunnar_pat_population-{i:06}")
}
fn pop_user(i: usize) -> String {
format!("user{i}")
}
/// Provision `pop` users — principals, tokens, corpus read grants, and
/// server-admin for user0 (the seeder) — through the bulk door. Returns the
/// wall seconds the import took.
fn provision_population(data: &Path, pop: usize) -> Result<f64> {
use gunnar_core::{token_hash, BulkUser, Grant, Level, Principal, PrincipalKind, Scope};
let stores = gunnar_server::ServerStores::open(data)
.map_err(|e| anyhow::anyhow!("open stores for bulk provisioning: {e}"))?;
let users: Vec<BulkUser> = (0..pop)
.map(|i| {
let id = pop_user(i);
let mut grants = vec![Grant {
principal: id.clone(),
scope: Scope::Namespace("corpus".into()),
level: Level::Read,
}];
if i == 0 {
grants.push(Grant {
principal: id.clone(),
scope: Scope::Server,
level: Level::Admin,
});
}
Ok(BulkUser {
principal: Principal::new(id.clone(), id.clone(), PrincipalKind::User)
.map_err(|e| anyhow::anyhow!("principal id: {e}"))?,
token_hash: Some(token_hash(&pop_secret(i))),
token_label: "population".into(),
openssh_public_key: None,
grants,
})
})
.collect::<Result<_>>()?;
let t = Instant::now();
let n = stores
.auth()
.bulk_add_users(&users, 1_700_000_000_000)
.map_err(|e| anyhow::anyhow!("bulk import of {pop} users: {e}"))?;
ensure!(n == pop, "bulk import reported {n} of {pop}");
Ok(t.elapsed().as_secs_f64())
}
/// A fresh data root with the corpus repos created and namespaces in place,
/// ready for [`provision_population`]. CLI verbs, server down.
fn pop_data_root(binary: &Path, work: &Path) -> Result<PathBuf> {
let data = work.join("data");
let _ = std::fs::remove_dir_all(&data);
std::fs::create_dir_all(&data)?;
admin(binary, &data, &["namespace", "create", "corpus"])?;
admin(binary, &data, &["namespace", "create", "scratch"])?;
for m in CORPUS {
admin(
binary,
&data,
&[
"repo",
"create",
&format!("corpus/{}", m.name),
"--hash",
"sha1",
],
)?;
}
Ok(data)
}
/// Seed the four corpus members from their mirrors, as user0. `ca` pins the
/// issuer when the server terminates its own TLS.
fn pop_seed(server: &Server, corpus: &Path, ca: Option<&Path>) -> Result<()> {
for m in CORPUS {
let url = server.url(&pop_user(0), &pop_secret(0), &format!("corpus/{}", m.name));
let mut c = Command::new("git");
if let Some(ca) = ca {
c.env("GIT_SSL_CAINFO", ca);
}
c.current_dir(corpus.join(format!("{}.git", m.name)))
.env("GIT_TERMINAL_PROMPT", "0")
.args(["push", "-q", &url, &format!("{}:refs/heads/main", m.pin)]);
run(&mut c, &format!("seeding corpus/{}", m.name))?;
}
Ok(())
}
/// Evenly-spread population sample: `count` distinct user indices.
fn pop_sample(pop: usize, count: usize) -> Vec<usize> {
let count = count.min(pop);
(0..count).map(|w| w * pop / count).collect()
}
/// **`gunnar.population_read`** — the read path against population size.
struct PopulationRead;
impl Bencher for PopulationRead {
fn id(&self) -> &'static str {
"gunnar.population_read"
}
fn run(&self) -> Result<BenchResult> {
let binary = super::forge::gunnar_binary()?;
let corpus = corpus_root();
for m in CORPUS {
ensure!(
corpus
.join(format!("{}.git", m.name))
.join("HEAD")
.is_file(),
"corpus member {} is not mirrored under {} (see gunnar.multiuser_scaling)",
m.name,
corpus.display()
);
}
let work = work_root().join("population-read");
std::fs::create_dir_all(&work)?;
let workers = env_usize("GUNNAR_BENCH_POP_WORKERS", 32);
let ls_each = env_usize("GUNNAR_BENCH_POP_LS", 40);
let mut metrics: Vec<(&'static str, serde_json::Value)> = Vec::new();
for pop in populations() {
let label = leak(format!("pop{pop}"));
let m = |suffix: &str| leak(format!("{label}_{suffix}"));
let data = pop_data_root(&binary, &work)?;
let provision_s = provision_population(&data, pop)?;
let t_start = Instant::now();
let server = Server::start(&binary, &data, work.join(format!("server-{pop}.log")))?;
let start_ms = t_start.elapsed().as_secs_f64() * 1000.0;
let pid = server.child.id();
pop_seed(&server, &corpus, None)?;
// The first authenticated request from a COLD user pays whatever
// the auth path pays at this population. user0 seeded, so sample
// from the far end.
let probe = pop_user(pop - 1);
let probe_url = server.url(&probe, &pop_secret(pop - 1), "corpus/anyhow");
let t = Instant::now();
let out = git(&["ls-remote", &probe_url, "refs/heads/main"], None)?;
let first_ls_ms = t.elapsed().as_secs_f64() * 1000.0;
ensure!(
out.contains(CORPUS[0].pin),
"the first ls-remote at pop {pop} did not advertise the pin: {out:?}"
);
// ── the measured window: 32 distinct users × ls-remote ────────
let log_from = server.log_len();
let cpu0 = proc_cpu_s(pid);
let sample = pop_sample(pop, workers);
let t = Instant::now();
let mut lat: Vec<f64> = Vec::new();
let mut failures: Vec<String> = Vec::new();
std::thread::scope(|scope| {
let mut handles = Vec::new();
for (w, idx) in sample.iter().enumerate() {
let member = &CORPUS[w % CORPUS.len()];
let url = server.url(
&pop_user(*idx),
&pop_secret(*idx),
&format!("corpus/{}", member.name),
);
handles.push(scope.spawn(move || -> Result<Vec<f64>> {
let mut v = Vec::with_capacity(ls_each);
for _ in 0..ls_each {
let t = Instant::now();
let out = git(&["ls-remote", &url, "refs/heads/main"], None)?;
v.push(t.elapsed().as_secs_f64() * 1000.0);
ensure!(out.contains(member.pin), "stale advertisement: {out:?}");
}
Ok(v)
}));
}
for h in handles {
match h.join() {
Ok(Ok(v)) => lat.extend(v),
Ok(Err(e)) => failures.push(format!("{e:#}")),
Err(_) => failures.push("a worker panicked".into()),
}
}
});
let wall = t.elapsed().as_secs_f64();
ensure!(
failures.is_empty(),
"pop {pop}: {} worker(s) failed: {}",
failures.len(),
failures.join(" | ")
);
let expect = sample.len() * ls_each;
ensure!(
lat.len() == expect,
"pop {pop}: {} ls-remotes completed of {expect}",
lat.len()
);
let cpu = proc_cpu_s(pid) - cpu0;
let (retries, dropped) = server.contention_since(log_from);
lat.sort_by(f64::total_cmp);
// ── a small clone sample, for the transfer path ───────────────
let clone_ms = {
let idx = pop / 2;
let url = server.url(&pop_user(idx), &pop_secret(idx), "corpus/clap");
let dir = work.join(format!("clone-{pop}"));
let _ = std::fs::remove_dir_all(&dir);
let t = Instant::now();
git(
&["clone", "-q", "--bare", &url, &dir.to_string_lossy()],
None,
)?;
let ms = t.elapsed().as_secs_f64() * 1000.0;
let counted = git(&["count-objects", "-v"], Some(&dir))?;
let in_pack: u64 = counted
.lines()
.find_map(|l| l.strip_prefix("in-pack: "))
.and_then(|v| v.trim().parse().ok())
.unwrap_or(0);
let _ = std::fs::remove_dir_all(&dir);
ensure!(
in_pack == CORPUS[3].objects,
"pop {pop}: the clone sample landed {in_pack} objects, clap's pin holds {}",
CORPUS[3].objects
);
ms
};
let rss = proc_rss_kb(pid) as f64 / 1024.0;
metrics.push((m("users"), json!(pop as f64)));
metrics.push((m("provision_s"), json!(round(provision_s, 3))));
metrics.push((m("server_start_ms"), json!(round(start_ms, 1))));
metrics.push((m("first_ls_ms"), json!(round(first_ls_ms, 2))));
metrics.push((m("ls_req_s"), json!(round(lat.len() as f64 / wall, 1))));
metrics.push((m("ls_p50_ms"), json!(round(pct(&lat, 0.50), 2))));
metrics.push((m("ls_p99_ms"), json!(round(pct(&lat, 0.99), 2))));
metrics.push((m("server_cpu_s"), json!(round(cpu, 3))));
metrics.push((m("clone_clap_ms"), json!(round(clone_ms, 1))));
metrics.push((m("server_rss_mb"), json!(round(rss, 1))));
metrics.push((m("cas_retries"), json!(retries as f64)));
metrics.push((m("last_used_dropped"), json!(dropped as f64)));
metrics.push((m("load1_before"), json!(load1())));
}
metrics.push((
"workers",
json!(env_usize("GUNNAR_BENCH_POP_WORKERS", 32) as f64),
));
metrics.push((
"harness",
json!(
"population axis: N principals+tokens+grants bulk-imported in one CAS; 32 \
distinct sampled users drive authenticated ls-remote against one host gunnar \
serve; the question is whether a request slows because the OTHER users exist"
),
));
Ok(result(self.id(), metrics))
}
}
nornir_bench::register_bench!(PopulationRead);
/// **`gunnar.population_write`** — the write path against population size.
///
/// Writers get one repository each (created for the ≤32 actives only), each
/// pushes a chain of unique commits sequentially; concurrent writers touch
/// disjoint store roots, which is the "1000 users = 1000 znippy files" axis.
/// The population's job is to be ballast: pushes/s should not move when
/// 100 000 dormant users exist.
struct PopulationWrite;
impl Bencher for PopulationWrite {
fn id(&self) -> &'static str {
"gunnar.population_write"
}
fn run(&self) -> Result<BenchResult> {
let binary = super::forge::gunnar_binary()?;
let work = work_root().join("population-write");
std::fs::create_dir_all(&work)?;
let writer_rungs: Vec<usize> = {
let raw = std::env::var("GUNNAR_BENCH_POP_WRITERS")
.unwrap_or_else(|_| if preview() { "1,8" } else { "1,8,32" }.to_string());
raw.split(',')
.filter_map(|s| s.trim().parse().ok())
.collect()
};
let pushes_each = env_usize("GUNNAR_BENCH_POP_PUSHES", 5);
let max_writers = writer_rungs.iter().copied().max().unwrap_or(1);
let mut metrics: Vec<(&'static str, serde_json::Value)> = Vec::new();
for pop in populations() {
let label = leak(format!("pop{pop}"));
let data = pop_data_root(&binary, &work)?;
let provision_s = provision_population(&data, pop)?;
// The active writers' repositories, and write grants for them —
// offline, before boot, like all provisioning.
let writers = pop_sample(pop, max_writers);
for idx in &writers {
let u = pop_user(*idx);
admin(
&binary,
&data,
&["repo", "create", &format!("scratch/{u}"), "--hash", "sha1"],
)?;
admin(
&binary,
&data,
&["grant", &u, "write", "--repo", &format!("scratch/{u}")],
)?;
}
let server = Server::start(&binary, &data, work.join(format!("server-{pop}.log")))?;
let pid = server.child.id();
// One pre-built chain per writer, long enough for every rung.
let chain_len = writer_rungs.len() * pushes_each + 1;
let mut chains: Vec<(usize, PathBuf, Vec<String>)> = Vec::new();
for idx in &writers {
let (dir, chain) =
build_chain(&work, &format!("p{pop}-{}", pop_user(*idx)), chain_len)?;
chains.push((*idx, dir, chain));
}
let mut consumed = 0usize; // chain commits pushed so far, per writer
for w in &writer_rungs {
let w = (*w).min(chains.len());
let m = |suffix: &str| leak(format!("{label}_w{w}_{suffix}"));
let log_from = server.log_len();
let cpu0 = proc_cpu_s(pid);
let t = Instant::now();
let mut lat: Vec<f64> = Vec::new();
let mut failures: Vec<String> = Vec::new();
let base = consumed;
std::thread::scope(|scope| {
let mut handles = Vec::new();
for (idx, dir, chain) in chains.iter().take(w) {
let url = server.url(
&pop_user(*idx),
&pop_secret(*idx),
&format!("scratch/{}", pop_user(*idx)),
);
handles.push(scope.spawn(move || -> Result<Vec<f64>> {
let mut v = Vec::with_capacity(pushes_each);
for t_i in 0..pushes_each {
let commit = &chain[base + t_i];
let t = Instant::now();
git(
&["push", "-q", &url, &format!("{commit}:refs/heads/main")],
Some(dir),
)?;
v.push(t.elapsed().as_secs_f64() * 1000.0);
}
Ok(v)
}));
}
for h in handles {
match h.join() {
Ok(Ok(v)) => lat.extend(v),
Ok(Err(e)) => failures.push(format!("{e:#}")),
Err(_) => failures.push("a writer panicked".into()),
}
}
});
let wall = t.elapsed().as_secs_f64();
consumed += pushes_each;
ensure!(
failures.is_empty(),
"pop {pop} w{w}: {} writer(s) failed: {}",
failures.len(),
failures.join(" | ")
);
ensure!(
lat.len() == w * pushes_each,
"pop {pop} w{w}: {} pushes completed of {}",
lat.len(),
w * pushes_each
);
// Every writer's ref landed on the server at its final chain
// commit for this rung.
for (idx, _, chain) in chains.iter().take(w) {
let url = server.url(
&pop_user(*idx),
&pop_secret(*idx),
&format!("scratch/{}", pop_user(*idx)),
);
let adv = git(&["ls-remote", &url, "refs/heads/main"], None)?;
let got = adv.split_whitespace().next().unwrap_or("");
ensure!(
got == chain[consumed - 1],
"pop {pop} w{w}: {}'s main is {got:?}, the rung pushed {:?}",
pop_user(*idx),
chain[consumed - 1]
);
}
let cpu = proc_cpu_s(pid) - cpu0;
let (retries, dropped) = server.contention_since(log_from);
lat.sort_by(f64::total_cmp);
metrics.push((m("push_s"), json!(round(lat.len() as f64 / wall, 2))));
metrics.push((m("push_p50_ms"), json!(round(pct(&lat, 0.50), 2))));
metrics.push((m("push_p99_ms"), json!(round(pct(&lat, 0.99), 2))));
metrics.push((m("server_cpu_s"), json!(round(cpu, 3))));
metrics.push((m("cas_retries"), json!(retries as f64)));
metrics.push((m("last_used_dropped"), json!(dropped as f64)));
}
let m = |suffix: &str| leak(format!("{label}_{suffix}"));
metrics.push((m("users"), json!(pop as f64)));
metrics.push((m("provision_s"), json!(round(provision_s, 3))));
metrics.push((
m("server_rss_mb"),
json!(round(proc_rss_kb(pid) as f64 / 1024.0, 1)),
));
}
metrics.push((
"harness",
json!(
"population axis for the write path: N dormant users bulk-imported; 1/8/32 \
active writers push unique pre-built commits to their own repositories \
(disjoint store roots — the one-user-one-znippy-file axis); pushes/s must \
not move with the dormant population"
),
));
Ok(result(self.id(), metrics))
}
}
nornir_bench::register_bench!(PopulationWrite);
// ═══════════════════════════════════════════════════════════════════════════
// THE SAME POPULATION AXIS, ON GITEA
// ═══════════════════════════════════════════════════════════════════════════
/// **`gunnar.population_read_gitea`** — the read-path population axis against
/// gitea, for comparison with [`PopulationRead`].
///
/// Same shape where it can be: N users provisioned, 32 distinct sampled users
/// drive authenticated ls-remote against one seeded corpus repository, then a
/// clone sample. Differences, stated rather than hidden:
///
/// * **provisioning is gitea's own admin REST API, one call per user** — a
/// forge without a bulk import provisions at its API's pace, and that pace
/// is published (`provision_s`, `provision_per_user_ms`). The default
/// populations stop at 10 000; `pop100000_provision_projected_s`
/// extrapolates from the measured per-user cost so the 100k row exists
/// honestly without a multi-hour run. `GUNNAR_BENCH_POP_GITEA=…,100000`
/// runs it for real.
/// * **auth is HTTP Basic** (username + password), gunnar's is a token — both
/// are one authenticated credential check per request at the same seam.
/// * the corpus repo is **public** and readers hold no per-repo grant; gunnar's
/// readers hold a namespace grant. Both requests authenticate; the ACL
/// question differs, and the row says so.
/// * the container is a plain pinned `podman run`, not the forge harness's
/// cgroup-parity rig: this arm compares POPULATION sensitivity, not CPU
/// ratios, and its numbers must not be quoted against `vs_forge_*` columns.
struct PopulationReadGitea;
/// The same pinned image the forge harness measures.
const POP_GITEA_IMAGE: &str = "docker.io/gitea/gitea:1.24.3";
const POP_GITEA_NAME: &str = "gunnar-population-gitea";
const POP_GITEA_ADMIN: &str = "benchadmin";
const POP_GITEA_PASSWORD: &str = "population-pass-1";
fn gitea_populations() -> Vec<usize> {
let default = if preview() { "100" } else { "1000,10000" };
let raw = std::env::var("GUNNAR_BENCH_POP_GITEA").unwrap_or_else(|_| default.to_string());
raw.split(',')
.filter_map(|s| s.trim().parse().ok())
.collect()
}
/// `curl` against the gitea API, returning the HTTP status code.
fn gitea_api(port: u16, method: &str, path: &str, body: Option<&str>) -> Result<u16> {
let url = format!("http://127.0.0.1:{port}{path}");
let mut c = Command::new("curl");
c.args(["-s", "-o", "/dev/null", "-w", "%{http_code}", "-X", method])
.args(["-u", &format!("{POP_GITEA_ADMIN}:{POP_GITEA_PASSWORD}")])
.args(["-H", "Content-Type: application/json"]);
if let Some(b) = body {
c.args(["-d", b]);
}
c.arg(&url);
let out = run(&mut c, &format!("curl {method} {path}"))?;
out.trim()
.parse()
.with_context(|| format!("curl printed {out:?}, not a status code"))
}
/// A fresh gitea container on an ephemeral loopback port, admin created,
/// ready for API provisioning. Returns the host port.
fn gitea_up() -> Result<u16> {
let _ = Command::new("podman")
.args(["rm", "-f", POP_GITEA_NAME])
.output();
let ok = Command::new("podman")
.args(["image", "exists", POP_GITEA_IMAGE])
.status()?;
ensure!(
ok.success(),
"the pinned gitea image {POP_GITEA_IMAGE} is not present; `podman pull {POP_GITEA_IMAGE}`"
);
run(
Command::new("podman").args([
"run",
"-d",
"--rm",
"--name",
POP_GITEA_NAME,
"-p",
"127.0.0.1::3000",
"-e",
"GITEA__security__INSTALL_LOCK=true",
"-e",
"GITEA__server__DOMAIN=127.0.0.1",
POP_GITEA_IMAGE,
]),
"podman run gitea",
)?;
let ports = run(
Command::new("podman").args(["port", POP_GITEA_NAME, "3000"]),
"podman port",
)?;
let port: u16 = ports
.trim()
.rsplit(':')
.next()
.and_then(|p| p.parse().ok())
.context("podman printed no host port")?;
// Wait for HTTP, then create the admin. `su git`: gitea refuses root.
let deadline = Instant::now() + std::time::Duration::from_secs(120);
loop {
let probe = Command::new("curl")
.args(["-s", "-o", "/dev/null", "-w", "%{http_code}"])
.arg(format!("http://127.0.0.1:{port}/api/healthz"))
.output()?;
if String::from_utf8_lossy(&probe.stdout).trim() == "200" {
break;
}
ensure!(
Instant::now() < deadline,
"gitea never answered /api/healthz on port {port}"
);
std::thread::sleep(std::time::Duration::from_millis(250));
}
run(
Command::new("podman").args([
"exec",
POP_GITEA_NAME,
"/bin/sh",
"-c",
&format!(
"su git -c 'gitea admin user create --username {POP_GITEA_ADMIN} \
--password {POP_GITEA_PASSWORD} --email {POP_GITEA_ADMIN}@bench.invalid \
--admin --must-change-password=false'"
),
]),
"gitea admin user create",
)?;
Ok(port)
}
impl Bencher for PopulationReadGitea {
fn id(&self) -> &'static str {
"gunnar.population_read_gitea"
}
fn run(&self) -> Result<BenchResult> {
let corpus = corpus_root();
let anyhow_mirror = corpus.join("anyhow.git");
ensure!(
anyhow_mirror.join("HEAD").is_file(),
"corpus member anyhow is not mirrored under {}",
corpus.display()
);
let workers = env_usize("GUNNAR_BENCH_POP_WORKERS", 32);
let ls_each = env_usize("GUNNAR_BENCH_POP_LS", 40);
let port = gitea_up()?;
let out = (|| -> Result<Vec<(&'static str, serde_json::Value)>> {
let mut metrics: Vec<(&'static str, serde_json::Value)> = Vec::new();
// The corpus repo, public, seeded once with the same pin gunnar's
// arm serves.
let status = gitea_api(
port,
"POST",
"/api/v1/user/repos",
Some(r#"{"name":"anyhow","private":false,"default_branch":"main"}"#),
)?;
ensure!(status == 201, "creating the corpus repo answered {status}");
let seed_url = format!(
"http://{POP_GITEA_ADMIN}:{POP_GITEA_PASSWORD}@127.0.0.1:{port}/{POP_GITEA_ADMIN}/anyhow.git"
);
git(
&[
"push",
"-q",
&seed_url,
&format!("{}:refs/heads/main", CORPUS[0].pin),
],
Some(&anyhow_mirror),
)?;
let mut provisioned = 0usize;
for pop in gitea_populations() {
let label = leak(format!("pop{pop}"));
let m = |suffix: &str| leak(format!("{label}_{suffix}"));
// Populations are cumulative on one container — users are only
// ever added — so each rung provisions the difference. The
// published cost is normalised per user.
let t = Instant::now();
for i in provisioned..pop {
let status = gitea_api(
port,
"POST",
"/api/v1/admin/users",
Some(&format!(
r#"{{"username":"{}","email":"{}@bench.invalid","password":"{POP_GITEA_PASSWORD}","must_change_password":false}}"#,
pop_user(i),
pop_user(i)
)),
)?;
ensure!(status == 201, "creating {} answered {status}", pop_user(i));
}
let provision_s = t.elapsed().as_secs_f64();
let added = pop - provisioned;
provisioned = pop;
let sample = pop_sample(pop, workers);
let t = Instant::now();
let mut lat: Vec<f64> = Vec::new();
let mut failures: Vec<String> = Vec::new();
std::thread::scope(|scope| {
let mut handles = Vec::new();
for idx in &sample {
let url = format!(
"http://{}:{POP_GITEA_PASSWORD}@127.0.0.1:{port}/{POP_GITEA_ADMIN}/anyhow.git",
pop_user(*idx)
);
handles.push(scope.spawn(move || -> Result<Vec<f64>> {
let mut v = Vec::with_capacity(ls_each);
for _ in 0..ls_each {
let t = Instant::now();
let out = git(&["ls-remote", &url, "refs/heads/main"], None)?;
v.push(t.elapsed().as_secs_f64() * 1000.0);
ensure!(out.contains(CORPUS[0].pin), "stale: {out:?}");
}
Ok(v)
}));
}
for h in handles {
match h.join() {
Ok(Ok(v)) => lat.extend(v),
Ok(Err(e)) => failures.push(format!("{e:#}")),
Err(_) => failures.push("a worker panicked".into()),
}
}
});
let wall = t.elapsed().as_secs_f64();
ensure!(
failures.is_empty(),
"gitea pop {pop}: {} worker(s) failed: {}",
failures.len(),
failures.join(" | ")
);
ensure!(lat.len() == sample.len() * ls_each, "short batch");
lat.sort_by(f64::total_cmp);
let clone_ms = {
let dir = work_root().join(format!("gitea-clone-{pop}"));
let _ = std::fs::remove_dir_all(&dir);
let idx = pop / 2;
let url = format!(
"http://{}:{POP_GITEA_PASSWORD}@127.0.0.1:{port}/{POP_GITEA_ADMIN}/anyhow.git",
pop_user(idx)
);
let t = Instant::now();
git(
&["clone", "-q", "--bare", &url, &dir.to_string_lossy()],
None,
)?;
let ms = t.elapsed().as_secs_f64() * 1000.0;
let counted = git(&["count-objects", "-v"], Some(&dir))?;
let in_pack: u64 = counted
.lines()
.find_map(|l| l.strip_prefix("in-pack: "))
.and_then(|v| v.trim().parse().ok())
.unwrap_or(0);
let _ = std::fs::remove_dir_all(&dir);
ensure!(
in_pack == CORPUS[0].objects,
"gitea clone landed {in_pack} of {}",
CORPUS[0].objects
);
ms
};
metrics.push((m("users"), json!(pop as f64)));
metrics.push((m("provision_s"), json!(round(provision_s, 2))));
metrics.push((
m("provision_per_user_ms"),
json!(round(provision_s * 1000.0 / added.max(1) as f64, 2)),
));
metrics.push((m("ls_req_s"), json!(round(lat.len() as f64 / wall, 1))));
metrics.push((m("ls_p50_ms"), json!(round(pct(&lat, 0.50), 2))));
metrics.push((m("ls_p99_ms"), json!(round(pct(&lat, 0.99), 2))));
metrics.push((m("clone_anyhow_ms"), json!(round(clone_ms, 1))));
metrics.push((m("load1_before"), json!(load1())));
}
// The 100k row, projected from the measured per-user API cost when
// it was not run for real — a forge that provisions at N calls per
// population deserves its projection printed beside gunnar's
// measured one-CAS import.
if !gitea_populations().contains(&100_000) && provisioned > 0 {
let per_user = metrics
.iter()
.rev()
.find(|(k, _)| k.ends_with("provision_per_user_ms"))
.and_then(|(_, v)| v.as_f64())
.unwrap_or(0.0);
metrics.push((
"pop100000_provision_projected_s",
json!(round(per_user * 100_000.0 / 1000.0, 1)),
));
}
metrics.push(("gitea_image", json!(POP_GITEA_IMAGE)));
Ok(metrics)
})();
let _ = Command::new("podman")
.args(["rm", "-f", POP_GITEA_NAME])
.output();
Ok(result(self.id(), out?))
}
}
nornir_bench::register_bench!(PopulationReadGitea);
// ═══════════════════════════════════════════════════════════════════════════
// THE DAY — open-loop torture over TLS
// ═══════════════════════════════════════════════════════════════════════════
/// **`gunnar.population_day`** — a population's day, compressed and served
/// over gunnar's own TLS.
///
/// The closed-loop arms above answer "how fast can K workers go"; a day is the
/// opposite shape — work ARRIVES at a rate that does not care how the server
/// is doing. This arm holds a target arrival rate for a duration against one
/// `gunnar serve --http-tls-self` (the single-executable shape: rustls
/// terminated in-process, ALPN h2, certificate minted from gunnar's own
/// issuer), with the whole population provisioned at full identity weight —
/// **every user carries a token AND its own ed25519 key** — in one bulk CAS.
///
/// Every operation is a fresh stock-git or curl process pinning the issuer, so
/// every operation pays a real TLS handshake: deliberately the worst case, and
/// therefore the honest one for "can 100k work against gunnar-server".
///
/// # The daily mix, per 20 arrivals
///
/// ```text
/// 13 × ls-remote 2 × up-to-date fetch 2 × verified clone
/// 2 × push 1 × LFS (PUT a unique 64 KiB object, GET it back,
/// bytes compared)
/// ```
///
/// Arrivals are scheduled on a fixed clock (`rate` per second). An arrival
/// that would exceed the in-flight cap is counted as **shed**, not queued —
/// the arm refuses to let its own client pile-up masquerade as server latency.
/// The verdict columns: achieved vs target rate, shed, per-kind p50/p99,
/// server cores and RSS, and the CAS counters that found every previous wall.
struct PopulationDay;
/// One finished arrival.
struct DayDone {
kind: &'static str,
ms: f64,
ok: bool,
detail: String,
}
fn day_mix() -> Vec<&'static str> {
let mut v = vec!["ls"; 13];
v.extend(["fetch", "fetch", "clone", "clone", "push", "push", "lfs"]);
v
}
impl Bencher for PopulationDay {
fn id(&self) -> &'static str {
"gunnar.population_day"
}
fn run(&self) -> Result<BenchResult> {
let binary = super::forge::gunnar_binary()?;
let corpus = corpus_root();
for m in CORPUS {
ensure!(
corpus
.join(format!("{}.git", m.name))
.join("HEAD")
.is_file(),
"corpus member {} is not mirrored under {}",
m.name,
corpus.display()
);
}
let pop = env_usize(
"GUNNAR_BENCH_DAY_POP",
if preview() { 2_000 } else { 100_000 },
);
let rate = env_usize("GUNNAR_BENCH_DAY_RATE", if preview() { 10 } else { 50 });
let secs = env_usize("GUNNAR_BENCH_DAY_SECS", if preview() { 20 } else { 300 });
let actives = env_usize("GUNNAR_BENCH_DAY_ACTIVES", 256).min(pop);
let cap = env_usize("GUNNAR_BENCH_DAY_INFLIGHT", 128);
let work = work_root().join("population-day");
let _ = std::fs::remove_dir_all(&work);
std::fs::create_dir_all(&work)?;
let data = pop_data_root(&binary, &work)?;
// ── the population, at FULL identity weight: token + own key each ──
let t = Instant::now();
let users: Vec<gunnar_core::BulkUser> = (0..pop)
.map(|i| {
use gunnar_core::{
token_hash, BulkUser, Grant, Level, Principal, PrincipalKind, Scope,
};
let id = pop_user(i);
let private = gunnar_core::mint_ed25519()
.map_err(|e| anyhow::anyhow!("minting a key for {id}: {e}"))?;
let parsed = gunnar_core::ssh_key::PrivateKey::from_openssh(&private)
.map_err(|e| anyhow::anyhow!("parsing {id}'s minted key: {e}"))?;
let public = parsed
.public_key()
.to_openssh()
.map_err(|e| anyhow::anyhow!("rendering {id}'s public key: {e}"))?;
let mut grants = vec![Grant {
principal: id.clone(),
scope: Scope::Namespace("corpus".into()),
level: Level::Read,
}];
if i == 0 {
grants.push(Grant {
principal: id.clone(),
scope: Scope::Server,
level: Level::Admin,
});
}
Ok(BulkUser {
principal: Principal::new(id.clone(), id.clone(), PrincipalKind::User)
.map_err(|e| anyhow::anyhow!("principal id: {e}"))?,
token_hash: Some(token_hash(&pop_secret(i))),
token_label: "day".into(),
openssh_public_key: Some(public),
grants,
})
})
.collect::<Result<_>>()?;
let keygen_s = t.elapsed().as_secs_f64();
let stores = gunnar_server::ServerStores::open(&data)
.map_err(|e| anyhow::anyhow!("open stores: {e}"))?;
let t = Instant::now();
stores
.auth()
.bulk_add_users(&users, 1_700_000_000_000)
.map_err(|e| anyhow::anyhow!("bulk import: {e}"))?;
let provision_s = t.elapsed().as_secs_f64();
drop(stores);
drop(users);
// The active pool's scratch repositories, offline like all provisioning.
let pool = pop_sample(pop, actives);
for idx in &pool {
let u = pop_user(*idx);
admin(
&binary,
&data,
&["repo", "create", &format!("scratch/{u}"), "--hash", "sha1"],
)?;
admin(
&binary,
&data,
&["grant", &u, "write", "--repo", &format!("scratch/{u}")],
)?;
}
// ── the server, terminating its own TLS ────────────────────────────
let server = Server::start_with(&binary, &data, work.join("server.log"), true)?;
let pid = server.child.id();
// The issuer, straight out of the handshake: the offline verb cannot
// run while the server holds the store lock, and the chain the
// listener serves carries the issuer as its second certificate.
let ca = work.join("issuer.pem");
{
let out = Command::new("sh")
.arg("-c")
.arg(format!(
"echo | openssl s_client -connect {} -showcerts 2>/dev/null | \
awk '/BEGIN CERT/{{n++}} n==2,/END CERT/{{print}}'",
server.http
))
.output()?;
ensure!(
!out.stdout.is_empty(),
"no issuer certificate in the TLS chain"
);
std::fs::write(&ca, &out.stdout)?;
}
pop_seed(&server, &corpus, Some(&ca))?;
// ── per-active setup outside the window ────────────────────────────
let protos = work.join("proto");
std::fs::create_dir_all(&protos)?;
for m in CORPUS {
let url = server.url(&pop_user(0), &pop_secret(0), &format!("corpus/{}", m.name));
let mut c = Command::new("git");
c.env("GIT_SSL_CAINFO", &ca)
.args(["clone", "-q", "--bare", &url])
.current_dir(&protos);
run(&mut c, "proto clone")?;
std::fs::rename(protos.join(format!("{}.git", m.name)), protos.join(m.name))?;
}
let mut fetch_dirs: Vec<PathBuf> = Vec::with_capacity(pool.len());
let mut chains: Vec<(PathBuf, Vec<String>)> = Vec::with_capacity(pool.len());
let pushes_expected = rate * secs / 10 / pool.len().max(1) + 8;
for idx in &pool {
let u = pop_user(*idx);
let member = &CORPUS[*idx % CORPUS.len()];
let dir = work.join("fetch").join(&u);
std::fs::create_dir_all(dir.parent().context("parent")?)?;
git(
&[
"clone",
"-q",
"--bare",
&protos.join(member.name).to_string_lossy(),
&dir.to_string_lossy(),
],
None,
)?;
git(
&[
"remote",
"set-url",
"origin",
&server.url(&u, &pop_secret(*idx), &format!("corpus/{}", member.name)),
],
Some(&dir),
)?;
fetch_dirs.push(dir);
let (cdir, chain) = build_chain(&work, &format!("day-{u}"), pushes_expected)?;
chains.push((cdir, chain));
}
// ── THE DAY ────────────────────────────────────────────────────────
use std::sync::atomic::AtomicUsize;
let inflight = Arc::new(AtomicUsize::new(0));
let shed = Arc::new(AtomicUsize::new(0));
let done: Arc<std::sync::Mutex<Vec<DayDone>>> = Arc::new(std::sync::Mutex::new(Vec::new()));
let push_cursor: Vec<AtomicUsize> = pool.iter().map(|_| AtomicUsize::new(0)).collect();
let log_from = server.log_len();
let cpu0 = proc_cpu_s(pid);
let rss_peak = Arc::new(AtomicU64::new(0));
{
let rss_peak = Arc::clone(&rss_peak);
let stop_at = Instant::now() + std::time::Duration::from_secs(secs as u64 + 30);
std::thread::spawn(move || {
while Instant::now() < stop_at {
rss_peak.fetch_max(proc_rss_kb(pid), Ordering::Relaxed);
std::thread::sleep(std::time::Duration::from_millis(200));
}
});
}
let mix = day_mix();
let started = Instant::now();
let total_ticks = rate * secs;
std::thread::scope(|scope| -> Result<()> {
for tick in 0..total_ticks {
// Fixed-clock arrival: sleep until this tick's scheduled time.
let due = started
+ std::time::Duration::from_micros((tick as u64) * 1_000_000 / rate as u64);
let now = Instant::now();
if due > now {
std::thread::sleep(due - now);
}
if inflight.load(Ordering::Relaxed) >= cap {
shed.fetch_add(1, Ordering::Relaxed);
continue;
}
let kind = mix[tick % mix.len()];
let who = tick % pool.len();
let idx = pool[who];
let user = pop_user(idx);
let secret = pop_secret(idx);
let member = &CORPUS[idx % CORPUS.len()];
let corpus_url = server.url(&user, &secret, &format!("corpus/{}", member.name));
let scratch_url = server.url(&user, &secret, &format!("scratch/{user}"));
let fetch_dir = fetch_dirs[who].clone();
let (chain_dir, chain) = (&chains[who].0, &chains[who].1);
let push_i = if kind == "push" {
push_cursor[who].fetch_add(1, Ordering::Relaxed)
} else {
0
};
let commit = chain
.get(push_i.min(chain.len() - 1))
.cloned()
.unwrap_or_default();
let chain_dir = chain_dir.clone();
let ca = ca.clone();
let work = work.clone();
let inflight = Arc::clone(&inflight);
let done = Arc::clone(&done);
inflight.fetch_add(1, Ordering::Relaxed);
scope.spawn(move || {
let t = Instant::now();
let outcome: Result<String> = (|| {
let g = |args: &[&str], dir: Option<&Path>| -> Result<String> {
let mut c = Command::new("git");
if let Some(d) = dir {
c.current_dir(d);
}
c.env("GIT_SSL_CAINFO", &ca)
.env("GIT_TERMINAL_PROMPT", "0")
.args(args);
run(&mut c, &format!("git {args:?}"))
};
match kind {
"ls" => {
let out = g(&["ls-remote", &corpus_url, "refs/heads/main"], None)?;
ensure!(out.contains(member.pin), "stale advertisement");
Ok(out)
}
"fetch" => g(&["fetch", "-q", "origin", "main"], Some(&fetch_dir)),
"clone" => {
let dir = work.join("clones").join(format!("{user}-{tick}"));
let _ = std::fs::remove_dir_all(&dir);
g(
&["clone", "-q", "--bare", &corpus_url, &dir.to_string_lossy()],
None,
)?;
let counted = g(&["count-objects", "-v"], Some(&dir))?;
let in_pack: u64 = counted
.lines()
.find_map(|l| l.strip_prefix("in-pack: "))
.and_then(|v| v.trim().parse().ok())
.unwrap_or(0);
let _ = std::fs::remove_dir_all(&dir);
ensure!(
in_pack == member.objects,
"clone landed {in_pack} of {}",
member.objects
);
Ok(counted)
}
"push" => g(
&[
"push",
"-q",
"-f",
&scratch_url,
&format!("{commit}:refs/heads/day"),
],
Some(&chain_dir),
),
"lfs" => {
// A unique 64 KiB object: PUT, then GET, bytes compared.
use std::io::Write as _;
let mut payload = Vec::with_capacity(64 * 1024);
let mut x: u64 = 0x9E3779B97F4A7C15 ^ (tick as u64);
while payload.len() < 64 * 1024 {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
payload.write_all(&x.to_le_bytes())?;
}
let tmp = work.join("lfs").join(format!("{user}-{tick}"));
std::fs::create_dir_all(tmp.parent().context("parent")?)?;
std::fs::write(&tmp, &payload)?;
// sha256 via the coreutils binary: the payload
// is already on disk for curl, and the bench
// links no hash crate of its own.
let sum = run(Command::new("sha256sum").arg(&tmp), "sha256sum")?;
let oid = sum.split_whitespace().next().unwrap_or("").to_string();
ensure!(oid.len() == 64, "sha256sum printed {sum:?}");
let base = scratch_url.trim_end_matches(".git").to_string();
let obj = format!("{base}.git/info/lfs/objects/{oid}");
let put = run(
Command::new("curl").args([
"-s",
"-o",
"/dev/null",
"-w",
"%{http_code}",
"--cacert",
&ca.to_string_lossy(),
"-X",
"PUT",
"--data-binary",
&format!("@{}", tmp.display()),
&obj,
]),
"lfs put",
)?;
ensure!(
put.trim() == "200" || put.trim() == "201",
"LFS PUT answered {put}"
);
let back = Command::new("curl")
.args(["-s", "--cacert", &ca.to_string_lossy(), &obj])
.output()?;
let _ = std::fs::remove_file(&tmp);
ensure!(
back.stdout == payload,
"LFS GET returned {} bytes against {} put",
back.stdout.len(),
payload.len()
);
Ok(oid)
}
other => bail!("unknown op kind {other}"),
}
})();
let ms = t.elapsed().as_secs_f64() * 1000.0;
inflight.fetch_sub(1, Ordering::Relaxed);
let (ok, detail) = match outcome {
Ok(_) => (true, String::new()),
Err(e) => (false, format!("{e:#}")),
};
done.lock().expect("day results").push(DayDone {
kind,
ms,
ok,
detail,
});
});
}
Ok(())
})?;
let wall = started.elapsed().as_secs_f64();
let cpu = proc_cpu_s(pid) - cpu0;
let (retries, dropped) = server.contention_since(log_from);
let done = Arc::try_unwrap(done)
.map_err(|_| anyhow::anyhow!("day results still shared"))?
.into_inner()
.expect("day results");
let shed = shed.load(Ordering::Relaxed);
// ── the verdict ────────────────────────────────────────────────────
let failures: Vec<&DayDone> = done.iter().filter(|d| !d.ok).collect();
ensure!(
failures.is_empty(),
"{} of {} arrivals FAILED; first three: {}",
failures.len(),
done.len(),
failures
.iter()
.take(3)
.map(|d| format!("[{}] {}", d.kind, d.detail))
.collect::<Vec<_>>()
.join(" | ")
);
ensure!(
done.len() + shed == total_ticks,
"{} completed + {} shed != {} scheduled",
done.len(),
shed,
total_ticks
);
let mut metrics: Vec<(&'static str, serde_json::Value)> = Vec::new();
metrics.push(("users", json!(pop as f64)));
metrics.push(("keygen_s", json!(round(keygen_s, 2))));
metrics.push(("provision_s", json!(round(provision_s, 2))));
metrics.push(("target_rate", json!(rate as f64)));
metrics.push(("duration_s", json!(round(wall, 1))));
metrics.push(("arrivals", json!(total_ticks as f64)));
metrics.push(("completed", json!(done.len() as f64)));
metrics.push(("achieved_rate", json!(round(done.len() as f64 / wall, 2))));
metrics.push(("shed", json!(shed as f64)));
metrics.push(("server_cpu_s", json!(round(cpu, 2))));
metrics.push(("server_cores_busy", json!(round(cpu / wall, 3))));
metrics.push((
"server_rss_peak_mb",
json!(round(rss_peak.load(Ordering::Relaxed) as f64 / 1024.0, 1)),
));
metrics.push(("cas_retries", json!(retries as f64)));
metrics.push(("last_used_dropped", json!(dropped as f64)));
metrics.push(("load1_end", json!(load1())));
for kind in ["ls", "fetch", "clone", "push", "lfs"] {
let mut v: Vec<f64> = done
.iter()
.filter(|d| d.kind == kind)
.map(|d| d.ms)
.collect();
if v.is_empty() {
continue;
}
v.sort_by(f64::total_cmp);
metrics.push((leak(format!("{kind}_n")), json!(v.len() as f64)));
metrics.push((
leak(format!("{kind}_p50_ms")),
json!(round(pct(&v, 0.50), 2)),
));
metrics.push((
leak(format!("{kind}_p99_ms")),
json!(round(pct(&v, 0.99), 2)),
));
}
metrics.push(("mix", json!("per 20 arrivals: 13 ls, 2 fetch, 2 clone, 2 push, 1 lfs(64KiB PUT+GET+compare); fixed-clock arrivals, in-flight cap counted as shed")));
metrics.push(("harness", json!("open loop over gunnar's OWN TLS (--http-tls-self, ALPN h2, issuer pinned); every op a fresh process = a real handshake per op; population at full identity weight (token + ed25519 key each)")));
Ok(result(self.id(), metrics))
}
}
nornir_bench::register_bench!(PopulationDay);