bench/examples/forge/concurrent.rs
//! **What each engine does when more than one customer shows up.**
//!
//! The arms in [`super`] measure one client at a time. That answers "what does
//! one operation cost", and it is the right first question, but it is not the
//! question a forge is bought to answer. Customers-per-box is a statement about
//! what happens at N clients: where throughput stops rising, what the slowest
//! client waits, and how many seconds of a core and how many resident pages one
//! served request costs when the server is not idle between requests.
//!
//! Every figure the single-client arms produce about capacity is therefore an
//! **inference**. These arms measure it.
//!
//! # The ramp
//!
//! `GUNNAR_BENCH_FORGE_CONC` (default `1,4,16`) is the client count at each
//! rung. One rung is one batch: N stock `git` processes started as close to
//! together as the host will start them, against one engine, through the same
//! counting relay every other arm uses, inside the same cgroup every other arm
//! reads back. Rung `1` is in the ramp on purpose — it is the single-client
//! number produced by *this* code path, so the gap between it and rung 16 is a
//! measurement of concurrency and not a comparison of two harnesses.
//!
//! # The trap this arm has that the others do not
//!
//! **Clients that all fail instantly produce a magnificent ops/sec.** Sixteen
//! `git clone`s that die on a 404 finish in 40 ms and report 400 operations per
//! second, and every other column — CPU, RSS, wire bytes — is small in exactly
//! the way a very fast server's would be. So a batch is not scored until:
//!
//! 1. **every** client process exited 0, named individually if not;
//! 2. **every** client's resulting repository holds the fixture's own object
//! count and the fixture's own tip — not the batch's total, each client's,
//! because one empty clone hiding inside fifteen good ones is exactly the
//! shape that survives a sum;
//! 3. the shared floors in [`super::assert_counted`] — a relay that stopped
//! counting, a cgroup that resolved to the wrong container.
//!
//! Both halves are reachable from a red switch:
//!
//! | `GUNNAR_BENCH_FORGE_RED=` | what it breaks | which guard must fire |
//! |---|---|---|
//! | `concfail` | points one client of each batch at a repository that does not exist | the per-client exit-status check in [`clone_batch`] |
//! | `concshallow` | clones one client of each batch at `--depth 1` | the per-client object-count check in [`verify_client`] |
//!
//! # What is NOT equalised, and is not equalisable
//!
//! The N clients run on the same box as the three servers. the desktop has 32 cores
//! and each engine's cgroup is capped at 8, so at rung 16 the clients and one
//! server together want more than the box has when the box is already loaded —
//! and `load1` is on every row for that reason. This is a **comparison** under
//! that condition, not an absolute throughput figure, and the condition is
//! identical for all three engines because the rungs interleave engine-inside,
//! rung-outside, exactly as the single-client arms interleave.
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Instant;
use anyhow::{ensure, Context, Result};
use nornir_bench::api::Bencher;
use nornir_bench::BenchResult;
use serde_json::json;
use super::{
assert_client_git, assert_columns_consistent, assert_counted, assert_engines_agree, fixtures,
idle_window, leak, measured, metrics_with_conditions, red, rounds, tip_of, unique_objects,
with_engines_named, work_root, Engine, Forge, Run,
};
use super::{env_usize, h2h_git, result};
/// The rungs of the ramp, lowest first.
///
/// Default `1,4,16`: one client, a small office, and a box under real
/// concurrency. `GUNNAR_BENCH_FORGE_CONC=1,2,4,8,16,32` walks the whole curve
/// and takes proportionally longer — the arm is linear in the sum of the rungs.
fn ramp() -> Vec<usize> {
let raw = std::env::var("GUNNAR_BENCH_FORGE_CONC").unwrap_or_else(|_| "1,4,16".to_string());
let mut v: Vec<usize> = raw
.split(',')
.filter_map(|s| s.trim().parse::<usize>().ok())
.filter(|n| *n > 0)
.collect();
v.sort_unstable();
v.dedup();
if v.is_empty() {
v.push(1);
}
v
}
/// The label a rung's columns are written under: `c1`, `c4`, `c16` for the
/// quota-only configuration, `c1-pin` … for the one that also has a cpuset.
fn rung_label(n: usize, suffix: &str) -> &'static str {
leak(&format!("c{n}{suffix}"))
}
/// **The two ways an allocation of "eight CPUs" can be expressed**, and the arm
/// measures both because they are not the same allocation.
///
/// `""` is the cgroup `cpu.max` quota alone, which is what `podman run --cpus
/// 8` gives and what every row here used before 2026-08-04: 8 CPUs' worth of
/// budget, spendable on any of the box's 32. `"-pin"` adds a cpuset — eight
/// dedicated CPUs, disjoint per engine — so a thread hand-off stays on a core
/// instead of crossing one.
///
/// Which of the two a benchmark used turns out to decide the SHAPE of the ramp,
/// not merely its level, so reporting one of them would have been reporting a
/// property of the container runtime as a property of the server.
///
/// # The fairness note: quota-only is not a fair comparison **for gunnar**
///
/// The two configurations are not symmetric in what they measure, and the
/// asymmetry is the finding. MEASURED (the desktop, `year` corpus, cpu/op):
///
/// | gunnar cpu/op | c1 | c32 | c64 |
/// |---|---|---|---|
/// | quota only | 0.1686 | **0.3213** | 0.2185 |
/// | quota + cpuset | 0.1203 | 0.1630 | 0.1667 |
///
/// Unpinned at 32 clients gunnar **loses to both Go forges on every axis** (−18%
/// throughput, +31% CPU per request); pinned at 64 it **wins on every axis** and
/// is the only engine still scaling past c32. The Go forges are near-indifferent
/// to the cpuset. So the quota-only number is not gunnar's concurrency — it is
/// the **scheduler's**: a cgroup that hands eight CPUs' *budget* spendable across
/// all 32 lets a thread hand-off cross a core between every pkt-line, and that
/// cost lands almost entirely on the one engine that threads its transfer
/// (gunnar) and barely on the two that fork a process per request. Reporting the
/// `c<N>` column alone would therefore price the container runtime as if it were
/// the server, and it would do so in the one direction that flatters gitea and
/// forgejo — which is exactly the class of unfair comparison this arm exists to
/// refuse. **`c<N>-pin` is the like-for-like column** (dedicated cores, the same
/// place a bare-metal deployment gives every server); `c<N>` is kept beside it
/// precisely so the gap between the two is visible rather than chosen.
const CONFIGS: [&str; 2] = ["", "-pin"];
/// The dedicated CPUs an engine gets in the pinned configuration.
///
/// Disjoint, and — when the box is big enough — the same width as the quota
/// every engine already has, so the only thing that changes between the two
/// configurations is whether those CPUs are a budget or a place. Disjoint rather
/// than shared because engines confined to one set of cores would be measuring
/// each other's idle work.
///
/// # It has to FIT, and with seven columns it did not
///
/// The width used to be `allocation().cpus` flat, and `index * width` walked
/// straight off the end of the machine as soon as the run had more columns than
/// the box had room for. MEASURED on the desktop 2026-08-11, 32 cores, the seven-column
/// split (four znippy arms + gix + gitea + Forgejo):
///
/// ```text
/// gix: pinning pid 8855 to 32-39 exited Some(1):
/// taskset: failed to set pid 8855's affinity: Invalid argument
/// ```
///
/// 7 × 8 = 56 CPUs wanted, 32 present. The arm had already spent 554 seconds
/// getting there. `taskset` is right to refuse — the sixth column's set does not
/// exist — and the failure is loud, but the arithmetic should never have
/// produced it.
///
/// So the width is now what actually fits: the quota, unless the columns will
/// not tile the machine at that width, in which case an equal share of the CPUs
/// there are. It is reported on the row (`<engine>_cpuset_width`) precisely
/// because it is no longer always the quota, and a pinned figure taken at a
/// narrower width than the quota is a different measurement — the cgroup budget
/// then exceeds the places available, which is the *unpinned* pathology this
/// configuration exists to contrast against.
///
/// Returns `None` when even one CPU each is impossible (more columns than the
/// machine has cores). A caller that cannot pin must say so rather than pin
/// something wrong.
fn cpuset_width(columns: usize, quota: usize, cores: usize) -> Option<usize> {
if columns == 0 || cores == 0 {
return None;
}
let fits = cores / columns;
if fits == 0 {
return None;
}
Some(quota.max(1).min(fits))
}
fn cpuset_for(index: usize, width: usize) -> String {
let lo = index * width;
format!("{lo}-{}", lo + width - 1)
}
/// Nearest-rank percentile over an already sorted slice.
///
/// Nearest-rank and not interpolated: at rung 4 there is no 99th percentile to
/// interpolate towards, and an interpolated figure over four samples invents
/// precision the batch does not have. This returns an observed latency, always.
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]
}
/// One client's finished work, checked after the batch rather than during it.
///
/// Checked after, because `git cat-file --batch-all-objects` on sixteen fresh
/// clones is real work on the same box and would land inside the window the
/// server's CPU is being read over.
struct Landed {
dir: PathBuf,
ms: f64,
}
/// Every client of a batch delivered the fixture, or the batch is not scored.
///
/// This is the guard the whole arm turns on. A sum is not enough: fifteen good
/// clones and one empty one sum to a number that looks like sixteen slightly
/// small ones, and the empty one is the fast one.
fn verify_client(
l: &Landed,
want_objects: u64,
want_tip: &str,
engine: &str,
rung: &str,
) -> Result<u64> {
let got = unique_objects(&l.dir).with_context(|| {
format!(
"{engine}/{rung}: a client's clone at {} could not be read",
l.dir.display()
)
})?;
ensure!(
got == want_objects,
"{engine}/{rung}: a client landed {got} object(s), the fixture holds {want_objects}. A \
client that transferred nothing finishes first and lifts the batch's ops/sec, which is \
the one way this arm can report a magnificent number having measured nothing."
);
let tip = tip_of(&l.dir)?;
ensure!(
tip == want_tip,
"{engine}/{rung}: a client landed refs/heads/main at {tip:?}, the fixture is at \
{want_tip:?} — the right number of objects over the wrong history"
);
Ok(got)
}
/// N concurrent bare clones against one engine, as one measured batch.
#[allow(clippy::too_many_arguments)]
fn clone_batch(
forge: &Forge,
slug: &str,
rung: usize,
suffix: &str,
work: &Path,
want_objects: u64,
want_tip: &str,
) -> Result<Run> {
let label = rung_label(rung, suffix);
let engine = forge.engine.key();
let base = work.join(format!("{engine}-{label}"));
let _ = std::fs::remove_dir_all(&base);
std::fs::create_dir_all(&base)?;
let good_url = forge.url(slug, true);
// ── RED SWITCH: `concfail` ─────────────────────────────────────────────
// One client of every batch is pointed at a repository that does not
// exist. It fails in tens of milliseconds, which RAISES the batch's
// measured throughput, and its clone directory is simply absent — so a
// harness that only summed what it found would report a faster server.
let dead_url = forge.url("this-repository-does-not-exist", true);
let mut landed: Vec<Landed> = Vec::new();
let (results, wall_ms, cpu_s, rx, tx, conns, mem, idle_cpu_s) =
measured(forge, idle_window(), || {
let mut out: Vec<Result<Landed>> = Vec::new();
std::thread::scope(|scope| {
let mut handles = Vec::with_capacity(rung);
for i in 0..rung {
let dir = base.join(format!("client-{i}"));
let url = if red("concfail") && i == 0 {
&dead_url
} else {
&good_url
};
// ── RED SWITCH: `concshallow` ──────────────────────────
// A depth-1 clone is a valid repository with a completely
// different object set and a wonderfully small pack. The
// per-client object check is what has to see it.
let shallow = red("concshallow") && i == 0;
handles.push(scope.spawn(move || -> Result<Landed> {
let started = Instant::now();
let mut c = Command::new("git");
c.args(["clone", "-q", "--bare"]);
if shallow {
c.args(["--depth", "1"]);
}
let done = c
.arg(url)
.arg(&dir)
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.context("spawning a concurrent clone")?;
let ms = started.elapsed().as_secs_f64() * 1000.0;
ensure!(
done.status.success(),
"client {i} exited {:?}: {}",
done.status.code(),
String::from_utf8_lossy(&done.stderr).trim()
);
Ok(Landed { dir, ms })
}));
}
for h in handles {
out.push(
h.join()
.unwrap_or_else(|_| Err(anyhow::anyhow!("a client thread panicked"))),
);
}
});
Ok(out)
})?;
// ── THE ADMISSION COLUMN ───────────────────────────────────────────────
// Read the gate's own refusal counters off the log lines this batch just
// produced, before anything else drains the buffer. `refused` is 0 unless
// the gate was made to bind (the `concrefuse` red switch, or a low
// `GUNNAR_MAX_TRANSFERS`); a non-zero here is the backpressure this arm
// exists to be able to show. A refused smart-HTTP advertisement returns
// `429`, which `git clone` reports as a non-zero exit with `429` in its
// error — so at a binding ceiling GUARD 1 below would fail the batch on a
// refusal that is the very thing we are trying to measure. That is why the
// guard is `concrefuse`-aware: under it a `429` exit is a REFUSAL (data),
// any other non-zero exit is still a hard red, and the client-observed
// refusals are cross-checked against the gate's own counter.
let adm = forge.drain_admission();
let refusing = red("concrefuse");
// GUARD 1: every client finished, and the ones that did not are named.
// Under `concrefuse` a client turned away at the ceiling (`429`) is a
// refusal, not a fault: it is separated from real failures, counted, and
// cross-checked against the gate's own counter below.
let mut failures: Vec<String> = Vec::new();
let mut client_refusals = 0u64;
for r in results {
match r {
Ok(l) => landed.push(l),
Err(e) => {
let msg = format!("{e:#}");
if refusing && msg.contains("429") {
client_refusals += 1;
} else {
failures.push(msg);
}
}
}
}
ensure!(
failures.is_empty(),
"{engine}/{label}: {} of {rung} concurrent client(s) did not complete: {}. A batch whose \
clients failed finishes sooner and reports HIGHER throughput than one whose clients \
worked, so a failure here is never a slow result — it is a fictional one.",
failures.len(),
failures.join(" | ")
);
if refusing {
// The red this switch exists to reach: the gate MUST have bound, its
// own counter MUST have moved, and the two witnesses of a refusal — the
// client turned away and the gate that turned it away — MUST agree.
// Asserting on the applied outcome (the gate counter and the client
// exits), never on a state round-trip — measured, not asserted.
ensure!(
client_refusals > 0,
"{engine}/{label}: `concrefuse` set a binding ceiling and NOT ONE of {rung} \
client(s) was refused — the gate did not bind, so this arm would report a \
backpressure column that can never be non-zero. Lower `GUNNAR_MAX_TRANSFERS` \
below the rung, or the ceiling is not binding."
);
ensure!(
adm.refused >= client_refusals,
"{engine}/{label}: {client_refusals} client(s) were turned away with 429 but the \
gate's own refused_total is {} — the column and the clients disagree about how \
many refusals happened, which is the wired-to-the-wrong-field failure the counter \
read exists to catch",
adm.refused
);
ensure!(
adm.peak_in_flight > 0 && adm.peak_in_flight as usize <= rung,
"{engine}/{label}: the gate reported peak_in_flight {} against a rung of {rung} — a \
peak of 0 means no refusal line carried it, and a peak above the rung means the \
counter is not this batch's",
adm.peak_in_flight
);
}
// Landed count is the rung minus whatever the ceiling turned away; with no
// `concrefuse` ceiling every client lands, which is the standing red.
ensure!(
landed.len() as u64 + client_refusals == rung as u64,
"{engine}/{label}: {} landed + {client_refusals} refused != {rung} client(s)",
landed.len()
);
// GUARD 2: each one of them delivered the fixture.
let mut objects = 0u64;
for l in &landed {
objects += verify_client(l, want_objects, want_tip, engine, label)?;
}
let mut pack_b = 0u64;
let (mut packed, mut deltas) = (0u64, 0u64);
for l in &landed {
pack_b += super::h2h_pack_bytes(&l.dir)?;
let (p, d) = super::h2h_deltified_client(&l.dir)?;
packed += p;
deltas += d;
}
let mut lat: Vec<f64> = landed.iter().map(|l| l.ms).collect();
lat.sort_by(f64::total_cmp);
let _ = std::fs::remove_dir_all(&base);
Ok(Run {
engine: leak(engine),
fixture: label,
wall_ms,
wire_rx: rx,
wire_tx: tx,
conns,
cpu_s,
idle_cpu_s,
mem,
pack_b,
objects,
packed,
deltas,
tip: want_tip.to_string(),
// The clients that actually TRANSFERRED — equal to the rung on any
// ordinary run (GUARD 1 proves it), and the rung minus the refused
// under `concrefuse`, so throughput is over served requests and not
// inflated by the ones the gate shed.
clients: landed.len() as u64,
p50_ms: pct(&lat, 0.50),
p95_ms: pct(&lat, 0.95),
p99_ms: pct(&lat, 0.99),
refused: adm.refused,
peak_in_flight: adm.peak_in_flight,
})
}
/// The ramp: 1, 4, 16 concurrent clones of the same repository, per engine.
///
/// **Red conditions**, in addition to the shared floors: every client exited 0
/// and every client's clone holds the fixture's object set and tip. Seen red
/// with `GUNNAR_BENCH_FORGE_RED=concfail` and `=concshallow`.
///
/// # The admission column's own red (`concrefuse`)
///
/// The in-flight/refusal columns (`<engine>_<rung>_conc_refused`,
/// `_conc_peak_in_flight`) read the admission gate's own `refused_total` and
/// `peak_in_flight` off the server's log (`Forge::drain_admission`). On any
/// ordinary run they are 0 — the shipped per-principal share is a share, not a
/// wall, so N clients on one key get N slots and nothing is refused. To
/// see the column go red, `GUNNAR_BENCH_FORGE_RED=concrefuse` sets a low
/// `GUNNAR_MAX_TRANSFERS` on the gunnar container so the ramp crosses a binding
/// server-wide ceiling: the guard then requires `conc_refused > 0`, the gate's
/// counter to agree with the clients turned away (`429`), and the reported peak
/// to equal the ceiling. The load-independent half of that red is proven now,
/// off any box, by `gunnar_bench::admission_log`'s
/// `parsed_column_equals_the_real_admission_counter` — which drives a live
/// `gunnar_ssh::Admission` to a binding ceiling and asserts the parsed column
/// equals `AdmissionStats::refused()` / `peak_in_flight`, refusals 0 below the
/// ceiling and non-zero at it. The full containerised `concrefuse` run needs a
/// podman rig (and, for the pinned pass, 24 cores).
struct VsForgeConcurrent;
impl Bencher for VsForgeConcurrent {
fn id(&self) -> &'static str {
"gunnar.vs_forge_concurrent"
}
fn run(&self) -> Result<BenchResult> {
let git = assert_client_git()?;
assert_columns_consistent()?;
let fx = fixtures()?;
let corpus = conc_corpus(&fx)?;
let corpus = corpus.as_path();
let slugs: Vec<(&str, &Path)> = vec![("conc", corpus)];
let want_objects = unique_objects(corpus)?;
let want_tip = tip_of(corpus)?;
let rungs = ramp();
let mut labels: Vec<&'static str> = Vec::new();
for cfg in CONFIGS {
labels.extend(rungs.iter().map(|n| rung_label(*n, cfg)));
}
// Sized against the machine, not against the quota alone — see
// `cpuset_width`. `Engine::selected()` is what decides how many columns
// this run draws, and seven of them do not tile 32 cores at 8 CPUs each.
let quota = super::parity::allocation().cpus as usize;
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(0);
let columns = super::Engine::selected().len();
let pin_width = cpuset_width(columns, quota, cores);
let width = pin_width.unwrap_or(quota.max(1));
let mut pinned: Vec<(String, String)> = Vec::new();
let runs = with_engines_named(self.id(), &slugs, |forges| {
let work = work_root(self.id()).join("conc");
std::fs::create_dir_all(&work)?;
let mut runs = Vec::new();
for cfg in CONFIGS {
// The cpuset is applied between the two passes, to every
// engine, and read back out of `/proc/<pid>/status` before a
// byte is measured under it.
if cfg == "-pin" {
// A run with more columns than the box has cores cannot be
// pinned at all, and must say so instead of pinning some
// engines and not others — half a pinned pass is not a
// configuration, it is two configurations sharing a label.
let Some(w) = pin_width else {
anyhow::bail!(
"this run draws {columns} column(s) and the box has {cores} core(s), \
so the `-pin` pass cannot give each engine even one CPU of its own. \
Run fewer columns (GUNNAR_BENCH_FORGE_ZNIPPY_ARMS) or drop the \
pinned configuration; pinning only some of them would put two \
different configurations under one label."
);
};
for (i, forge) in forges.iter().enumerate() {
let range = cpuset_for(i, w);
let applied = forge.pin_cpus(&range)?;
pinned.push((forge.engine.key().to_string(), applied));
}
}
// One discarded warm-up per engine per configuration: the
// first clone pays for a cold page cache, and pinning moves
// where the cache is warm.
for forge in forges {
clone_batch(forge, "conc", 1, cfg, &work, want_objects, &want_tip)?;
}
for _ in 0..rounds() {
// Rung outside, engine inside — the three engines at one
// rung are the three measurements that have to sit next to
// each other in time. On a box whose load moved 2->218 in
// a day, the gap between two measurements is in their
// ratio.
for rung in &rungs {
for forge in forges {
runs.push(clone_batch(
forge,
"conc",
*rung,
cfg,
&work,
want_objects,
&want_tip,
)?);
}
}
}
}
Ok(runs)
})?;
assert_counted(&runs, "vs_forge_concurrent", true, false)?;
assert_engines_agree(&runs, "vs_forge_concurrent")?;
assert_throughput_is_real(&runs, &labels)?;
// Both configurations produced rows, or the arm measured one of them
// and named two. A pinned pass that silently did not happen is the
// whole failure this comparison exists to make impossible.
for cfg in CONFIGS {
let want = rung_label(rungs[0], cfg);
ensure!(
runs.iter().any(|r| r.fixture == want),
"the {:?} configuration produced no rows, so this arm measured one allocation \
and reported two",
if cfg.is_empty() {
"quota-only"
} else {
"quota+cpuset"
}
);
}
ensure!(
pinned.len() == Engine::selected().len(),
"{} engine(s) were pinned out of {} — whatever is done to one engine is done to all \
three, or the two columns are not the same experiment",
pinned.len(),
Engine::selected().len()
);
let mut metrics = metrics_with_conditions(&runs, &labels, true)?;
metrics.push(("client_git", json!(git)));
metrics.push((
"ramp",
json!(rungs
.iter()
.map(|n| n.to_string())
.collect::<Vec<_>>()
.join(",")),
));
metrics.push(("corpus_objects", json!(want_objects as f64)));
// ── THE IN-FLIGHT / REFUSAL COLUMN ─────────────────────────────────
// Per engine × rung × configuration: the admission gate's own
// `refused_total` and `peak_in_flight`, read off the server's log by
// `clone_batch` (`Forge::drain_admission`). Taken as the MAX across
// rounds, not the median — both are counters, and a median would hide
// a round that refused behind two that did not, which is the one thing
// this column must never do. The Go forges have no admission gate and
// read 0 (the comparison point, not a gap); a gunnar ramp under the
// shipped per-principal share also reads 0 — the honest "nothing
// refused" — and only a binding ceiling (`concrefuse`, or a low
// `GUNNAR_MAX_TRANSFERS`) drives it non-zero with the peak on the
// ceiling.
for engine in Engine::selected() {
let k = engine.key();
for label in &labels {
let max_of = |f: &dyn Fn(&Run) -> u64| -> u64 {
runs.iter()
.filter(|r| r.engine == k && r.fixture == *label)
.map(f)
.max()
.unwrap_or(0)
};
metrics.push((
leak(&format!("{k}_{label}_conc_refused")),
json!(max_of(&|r: &Run| r.refused)),
));
metrics.push((
leak(&format!("{k}_{label}_conc_peak_in_flight")),
json!(max_of(&|r: &Run| r.peak_in_flight)),
));
}
}
// The allocation each column was measured under, in words, because a
// CPU figure without it is a figure about the container runtime.
metrics.push((
"cpu_allocation",
json!(format!(
"`c<N>` is the cgroup cpu.max quota alone ({width} CPUs' budget across {} host \
CPUs, which is what `podman run --cpus` gives); `c<N>-pin` is the same quota \
plus a disjoint {width}-CPU affinity per engine, applied with `podman unshare \
taskset -acp` and read back out of /proc/<pid>/status: {}",
std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(0),
pinned
.iter()
.map(|(e, a)| format!("{e} {a}"))
.collect::<Vec<_>>()
.join(", ")
)),
));
for (engine, applied) in &pinned {
metrics.push((leak(&format!("{engine}_cpuset")), json!(applied.clone())));
// And the same fact as numbers, because the warehouse keeps only
// numeric metrics — the string above reaches the receipt and not
// the table the chart is drawn from.
let (lo, hi) = applied
.split_once('-')
.unwrap_or((applied.as_str(), applied.as_str()));
let lo: f64 = lo.trim().parse().unwrap_or(-1.0);
let hi: f64 = hi.trim().parse().unwrap_or(-1.0);
metrics.push((leak(&format!("{engine}_cpuset_first")), json!(lo)));
metrics.push((
leak(&format!("{engine}_cpuset_width")),
json!(hi - lo + 1.0),
));
}
Ok(result(self.id(), metrics))
}
}
nornir_bench::register_bench!(VsForgeConcurrent);
/// Which repository the ramp hammers.
///
/// `GUNNAR_BENCH_FORGE_CONC_CORPUS=year|large|real|xl`. `year` by default: a
/// year of eight people on one repository, with the branches and merges and
/// leftover refs a year leaves behind, which is the shape a forge under
/// concurrent load actually serves. `xl` is the order-of-magnitude corpus and
/// costs proportionally — at rung 16 it is measuring the disk as much as the
/// server, and that is worth knowing but it is a different question.
fn conc_corpus(fx: &super::Fixtures) -> Result<PathBuf> {
Ok(
match std::env::var("GUNNAR_BENCH_FORGE_CONC_CORPUS").as_deref() {
Ok("real") => fx.real.clone(),
Ok("large") => fx.large.clone(),
Ok("xl") => super::xl_fixture()?,
_ => super::year_fixture()?,
},
)
}
/// The throughput columns are not free numbers.
///
/// Two things this checks that no other guard here does:
///
/// 1. **A rung's batch really carried its rung's work.** Wire bytes at rung N
/// must be at least N times a single client's, minus the slack a shared
/// pack cache legitimately buys. Without this a server that answered
/// fifteen of sixteen requests from a cache and one from disk would look
/// like a server that served sixteen.
/// 2. **Throughput at rung 1 is not silently the whole result.** If the ramp
/// has more than one rung, at least one rung above 1 must have produced
/// rows for every engine, or the arm measured single-client latency and
/// called it saturation.
fn assert_throughput_is_real(runs: &[Run], labels: &[&'static str]) -> Result<()> {
for r in runs {
ensure!(
r.clients > 0 && r.ops_sec() > 0.0,
"{}/{}: a batch of {} client(s) over {} ms is not a throughput figure",
r.engine,
r.fixture,
r.clients,
r.wall_ms
);
ensure!(
r.p99_ms >= r.p50_ms && r.p50_ms > 0.0,
"{}/{}: p50 {} ms and p99 {} ms — the percentiles are not over a real latency sample",
r.engine,
r.fixture,
r.p50_ms,
r.p99_ms
);
}
if labels.len() > 1 {
let ramped: Vec<&Run> = runs.iter().filter(|r| r.clients > 1).collect();
ensure!(
!ramped.is_empty(),
"the ramp names {} rung(s) and every measured batch held one client — this arm would \
be reporting single-client latency under a name that says saturation",
labels.len()
);
}
Ok(())
}
// ═══════════════════════════════════════════════════════════════════════════
// THE MIXED WORKLOAD
// ═══════════════════════════════════════════════════════════════════════════
/// **Clone, fetch and push at the same time, against the same repository.**
///
/// A pure-clone ramp is the friendliest concurrency a forge ever sees: every
/// client wants the same bytes, so a pack cache, a page cache and a read-only
/// path all get their best case. The shape a real forge actually sees is a
/// read that is being written under, and it is the shape where a server that
/// takes a lock, or repacks on receive, or holds a whole pack in memory per
/// connection, stops looking like the pure-read number.
///
/// One batch of N clients, split 1/2 clone, 1/4 fetch, 1/4 push, all against
/// one repository at once. Each push carries its own unique incompressible
/// payload, so no push is a no-op and the writes really do land different
/// objects — and each push's payload is generated from a fixed seed, so the
/// object the three engines receive is byte-identical.
struct VsForgeMixed;
/// The clients in one mixed batch. Default 8: four readers, two negotiators and
/// two writers is the smallest split where every kind is present more than once.
fn mix_clients() -> usize {
env_usize("GUNNAR_BENCH_FORGE_MIX", 8).max(4)
}
/// Bytes each push client sends. Incompressible and unique per client, so the
/// write is a transfer rather than a measurement of pkt-line overhead.
fn mix_payload_bytes() -> usize {
env_usize("GUNNAR_BENCH_FORGE_MIX_KIB", 512) * 1024
}
/// A deterministic incompressible payload. The same seed gives the same bytes
/// on every engine, so the commit oid the three servers receive is identical
/// and `assert_engines_agree` has something real to compare.
fn payload(seed: u64, len: usize) -> Vec<u8> {
let mut out = Vec::with_capacity(len);
let mut x = seed | 1;
while out.len() < len {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
out.extend_from_slice(&x.to_le_bytes());
}
out.truncate(len);
out
}
impl Bencher for VsForgeMixed {
fn id(&self) -> &'static str {
"gunnar.vs_forge_mixed"
}
fn run(&self) -> Result<BenchResult> {
let git = assert_client_git()?;
assert_columns_consistent()?;
let fx = fixtures()?;
let n = mix_clients();
let label: &'static str = leak(&format!("mix{n}"));
let slugs: Vec<(&str, &Path)> = vec![("mix", fx.large.as_path())];
let want_objects = unique_objects(fx.large.as_path())?;
let want_tip = tip_of(fx.large.as_path())?;
let runs = with_engines_named(self.id(), &slugs, |forges| {
let work = work_root(self.id()).join("mix");
std::fs::create_dir_all(&work)?;
for forge in forges {
mixed_batch(forge, &work, n, 0, label, want_objects, &want_tip)?;
}
let mut runs = Vec::new();
for round_no in 1..=rounds() {
for forge in forges {
runs.push(mixed_batch(
forge,
&work,
n,
round_no,
label,
want_objects,
&want_tip,
)?);
}
}
Ok(runs)
})?;
assert_counted(&runs, "vs_forge_mixed", true, true)?;
assert_engines_agree(&runs, "vs_forge_mixed")?;
let mut metrics = metrics_with_conditions(&runs, &[label], true)?;
metrics.push(("client_git", json!(git)));
metrics.push(("mix_payload_b", json!(mix_payload_bytes() as f64)));
Ok(result(self.id(), metrics))
}
}
nornir_bench::register_bench!(VsForgeMixed);
/// How one mixed batch is split. Halves and quarters, floor'd, remainder to the
/// writers, and never fewer than one of each.
fn mix_split(n: usize) -> (usize, usize, usize) {
let clones = (n / 2).max(1);
let fetches = (n / 4).max(1);
let pushes = n.saturating_sub(clones + fetches).max(1);
(clones, fetches, pushes)
}
/// One mixed batch against one engine.
///
/// `round_no == 0` is the discarded warm-up; the round number is in every ref
/// this pushes, because a push is not idempotent and the second push of the
/// same ref transfers nothing and would be measured as an infinitely fast
/// server.
#[allow(clippy::too_many_arguments)]
fn mixed_batch(
forge: &Forge,
work: &Path,
n: usize,
round_no: usize,
label: &'static str,
want_objects: u64,
want_tip: &str,
) -> Result<Run> {
let engine = forge.engine.key();
let (clones, fetches, pushes) = mix_split(n);
let base = work.join(format!("{engine}-r{round_no}"));
let _ = std::fs::remove_dir_all(&base);
std::fs::create_dir_all(&base)?;
let url = forge.url("mix", true);
// ── SETUP, OUTSIDE THE WINDOW ──────────────────────────────────────────
// The negotiators need a repository that is genuinely behind the server,
// and the writers need a commit to send. Building either inside the timed
// window would put the client's own work in the server's CPU column.
let mut fetch_dirs = Vec::new();
let proto = base.join("proto");
let out = Command::new("git")
.args([
"clone",
"-q",
"--bare",
"--single-branch",
"--branch",
"base",
&url,
])
.arg(&proto)
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.context("preparing the mixed batch's negotiator")?;
ensure!(
out.status.success(),
"{engine}: preparing the mixed batch failed ({:?}): {}",
out.status.code(),
String::from_utf8_lossy(&out.stderr).trim()
);
let proto_objects = unique_objects(&proto)?;
let base_tip = h2h_git(
&[
"-C",
&proto.to_string_lossy(),
"rev-parse",
"refs/heads/base",
],
None,
)?
.trim()
.to_string();
for i in 0..fetches {
let d = base.join(format!("fetch-{i}"));
super::copy_repo(&proto, &d)?;
fetch_dirs.push(d);
}
// The writers. Each gets its own copy and its own commit, built here.
let mut push_jobs: Vec<(PathBuf, String, String)> = Vec::new();
for i in 0..pushes {
let d = base.join(format!("push-{i}"));
super::copy_repo(&proto, &d)?;
let ds = d.to_string_lossy().into_owned();
let bytes = payload(
0x5EED_0000 + (round_no as u64) * 1000 + i as u64,
mix_payload_bytes(),
);
let blob_path = d.join("payload.bin");
std::fs::write(&blob_path, &bytes)?;
let blob = h2h_git(
&["-C", &ds, "hash-object", "-w", &blob_path.to_string_lossy()],
None,
)?
.trim()
.to_string();
std::fs::remove_file(&blob_path)?;
let index = d.join("mix.index");
let idx = index.to_string_lossy().into_owned();
run_with_index(&ds, &idx, &["read-tree", &base_tip])?;
run_with_index(
&ds,
&idx,
&[
"update-index",
"--add",
"--cacheinfo",
&format!("100644,{blob},mix-payload.bin"),
],
)?;
let tree = run_with_index(&ds, &idx, &["write-tree"])?
.trim()
.to_string();
// A fixed author and date, so the commit oid is byte-identical on all
// three servers and the object every engine received is the same one.
let commit = h2h_git_env(
&[
"-C",
&ds,
"commit-tree",
&tree,
"-p",
&base_tip,
"-m",
&format!("mix {round_no} {i}"),
],
&[
("GIT_AUTHOR_NAME", "bench"),
("GIT_AUTHOR_EMAIL", "bench@example.com"),
("GIT_AUTHOR_DATE", "2026-01-01T00:00:00+00:00"),
("GIT_COMMITTER_NAME", "bench"),
("GIT_COMMITTER_EMAIL", "bench@example.com"),
("GIT_COMMITTER_DATE", "2026-01-01T00:00:00+00:00"),
],
)?
.trim()
.to_string();
let refname = format!("refs/heads/mix-{engine}-{round_no}-{i}");
push_jobs.push((d, commit, refname));
}
// ── THE WINDOW ─────────────────────────────────────────────────────────
let (outcomes, wall_ms, cpu_s, rx, tx, conns, mem, idle_cpu_s) =
measured(forge, idle_window(), || {
let mut out: Vec<Result<(&'static str, PathBuf, f64)>> = Vec::new();
std::thread::scope(|scope| {
let mut handles = Vec::new();
for i in 0..clones {
let dir = base.join(format!("clone-{i}"));
let url = &url;
handles.push(scope.spawn(move || {
let t = Instant::now();
let done = Command::new("git")
.args(["clone", "-q", "--bare", url])
.arg(&dir)
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.context("mixed clone")?;
ensure!(
done.status.success(),
"a mixed-batch clone exited {:?}: {}",
done.status.code(),
String::from_utf8_lossy(&done.stderr).trim()
);
Ok(("clone", dir, t.elapsed().as_secs_f64() * 1000.0))
}));
}
for dir in &fetch_dirs {
handles.push(scope.spawn(move || {
let t = Instant::now();
let done = Command::new("git")
.args([
"-C",
&dir.to_string_lossy(),
"fetch",
"-q",
"origin",
"main",
])
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.context("mixed fetch")?;
ensure!(
done.status.success(),
"a mixed-batch fetch exited {:?}: {}",
done.status.code(),
String::from_utf8_lossy(&done.stderr).trim()
);
Ok(("fetch", dir.clone(), t.elapsed().as_secs_f64() * 1000.0))
}));
}
for (dir, commit, refname) in &push_jobs {
handles.push(scope.spawn(move || {
let t = Instant::now();
let done = Command::new("git")
.args([
"-C",
&dir.to_string_lossy(),
"push",
"-q",
"origin",
&format!("{commit}:{refname}"),
])
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.context("mixed push")?;
ensure!(
done.status.success(),
"a mixed-batch push exited {:?}: {}",
done.status.code(),
String::from_utf8_lossy(&done.stderr).trim()
);
Ok(("push", dir.clone(), t.elapsed().as_secs_f64() * 1000.0))
}));
}
for h in handles {
out.push(
h.join()
.unwrap_or_else(|_| Err(anyhow::anyhow!("a mixed client panicked"))),
);
}
});
Ok(out)
})?;
// ── EVERY CLIENT'S WORK LANDED ─────────────────────────────────────────
let mut lat = Vec::new();
let mut done: Vec<(&'static str, PathBuf)> = Vec::new();
let mut failures = Vec::new();
for o in outcomes {
match o {
Ok((kind, dir, ms)) => {
lat.push(ms);
done.push((kind, dir));
}
Err(e) => failures.push(format!("{e:#}")),
}
}
ensure!(
failures.is_empty(),
"{engine}/{label}: {} of {n} mixed client(s) did not complete: {}",
failures.len(),
failures.join(" | ")
);
let mut objects = 0u64;
let mut pack_b = 0u64;
let (mut packed, mut deltas) = (0u64, 0u64);
let (mut seen_clone, mut seen_fetch, mut seen_push) = (0usize, 0usize, 0usize);
for (kind, dir) in &done {
match *kind {
// The readers are held to the fixture, with the one allowance that
// makes this arm different from the pure ramp: a reader may pick up
// refs the concurrent WRITERS landed while it was in flight, and on
// 2026-08-04 it did — 193 objects against the fixture's 190, which
// is one push's blob, tree and commit arriving mid-clone. That is
// the workload working, so the floor is `>=` and the ceiling is
// what every push in this batch could possibly have added. What
// stays exact is `refs/heads/main`: the writers push to `mix-*`
// refs and never to main, so a moved main is a different fault.
"clone" => {
let got = unique_objects(dir)?;
// Generous per write, because the exact number of objects a
// commit adds depends on how deep the tree is — measured
// 2026-08-04: two concurrent pushes added 11 objects to a
// reader mid-clone, not the 8 a blob-tree-commit each would
// suggest. The job of this ceiling is to catch a reader that
// came back with a whole second repository, not to count
// trees, so it is loose and the floor below it is exact.
let ceiling = want_objects + 64 * pushes as u64;
ensure!(
got >= want_objects,
"{engine}/{label}: a concurrent reader landed {got} object(s), fewer than the \
fixture's {want_objects} — a read that returned nothing while writes were in \
flight is the finding this arm exists to catch, not a fast server"
);
ensure!(
got <= ceiling,
"{engine}/{label}: a concurrent reader landed {got} object(s) against a \
fixture of {want_objects} and at most {ceiling} with every one of the \
{pushes} concurrent write(s) folded in — more than that is not this workload"
);
ensure!(
tip_of(dir)? == want_tip,
"{engine}/{label}: a concurrent reader landed refs/heads/main away from the \
fixture's {want_tip:?}; the writers push to mix-* refs and never to main"
);
// The FIXTURE's object set, not what this reader happened to
// catch. `assert_engines_agree` compares this column across the
// three engines, and how many concurrent writes landed inside a
// given clone is a race — a true number, and not one the three
// engines can be asked to agree on. What they must agree on is
// that every reader got the whole repository, and that is what
// the two bounds above just established.
objects += want_objects;
pack_b += super::h2h_pack_bytes(dir)?;
let (p, d) = super::h2h_deltified_client(dir)?;
packed += p;
deltas += d;
seen_clone += 1;
}
// The negotiators must have gained history, not merely exited 0.
"fetch" => {
let after = unique_objects(dir)?;
ensure!(
after > proto_objects,
"{engine}/{label}: a concurrent fetch went from {proto_objects} to {after} \
objects — nothing was negotiated, and a fetch that transferred nothing is \
the fastest fetch there is"
);
seen_fetch += 1;
}
_ => seen_push += 1,
}
}
ensure!(
seen_clone == clones && seen_fetch == fetches && seen_push == pushes,
"{engine}/{label}: the batch ran {seen_clone}/{seen_fetch}/{seen_push} \
clone/fetch/push, not {clones}/{fetches}/{pushes}"
);
// The writes landed ON THE SERVER, asked of the server rather than of the
// client that claims to have sent them.
for (_, commit, refname) in &push_jobs {
let adv = h2h_git(&["ls-remote", &url, refname], None)?;
let got = adv.split_whitespace().next().unwrap_or("").to_string();
ensure!(
got == *commit,
"{engine}/{label}: {refname} is at {got:?} on the server, the client pushed {commit:?}"
);
}
lat.sort_by(f64::total_cmp);
let _ = std::fs::remove_dir_all(&base);
Ok(Run {
engine: leak(engine),
fixture: label,
wall_ms,
wire_rx: rx,
wire_tx: tx,
conns,
cpu_s,
idle_cpu_s,
mem,
pack_b,
objects,
packed,
deltas,
tip: want_tip.to_string(),
clients: n as u64,
p50_ms: pct(&lat, 0.50),
p95_ms: pct(&lat, 0.95),
p99_ms: pct(&lat, 0.99),
// The admission column is wired on the pure-clone ramp (`VsForgeConcurrent`),
// whose `concrefuse` red drives a binding ceiling. The mixed arm shares
// `clone_batch`'s pattern and could carry it identically, but its
// read/write guards would need the same `429`-is-data relaxation first;
// until then it reports the honest default, not a bound it never drove.
refused: 0,
peak_in_flight: 0,
})
}
/// `git -C <repo>` with a private index file, for building a commit in a bare
/// repository without touching whatever index is there.
fn run_with_index(repo: &str, index: &str, args: &[&str]) -> Result<String> {
let mut argv = vec!["-C", repo];
argv.extend_from_slice(args);
h2h_git_env(&argv, &[("GIT_INDEX_FILE", index)])
}
/// `h2h_git` with environment. Kept here rather than widening the shared helper,
/// because every other caller of it wants the ambient environment.
fn h2h_git_env(args: &[&str], env: &[(&str, &str)]) -> Result<String> {
let mut c = Command::new("git");
c.args(args);
for (k, v) in env {
c.env(k, v);
}
let out = c.output().with_context(|| format!("git {args:?}"))?;
ensure!(
out.status.success(),
"git {args:?} exited {:?}: {}",
out.status.code(),
String::from_utf8_lossy(&out.stderr).trim()
);
Ok(String::from_utf8_lossy(&out.stdout).into_owned())
}