bench-scenarios/src/dataplane.rs
//! The request-path arms — W12's yardstick.
//!
//! `cd7636d` (2026-05-31) extracted the archive-level loops into this crate and
//! left the `handle_http2_request` hammering of
//! `server/repository/znippy-rust/tests/bench_znippy_throughput.rs` behind, so
//! from that day nothing measured the serving path as a CLIENT sees it. These
//! arms restore that measurement, over the wire, against the two doors and
//! the two write shapes stinky-01 section 2 named:
//!
//! - `http_large`: ONE 1 GiB blob, `PUT` then `GET`, through the raw
//! repository. MB/s each way, plus the server-side resident-set growth while
//! the PUT is in flight — the number that says whether the door holds the
//! whole body before it dispatches.
//! - `http_small`: 4 KiB blobs, `PUT` then `GET`, N of each, serial. ops/s and
//! p50/p99/p999 per request: the per-request overhead of the door with the
//! bytes taken out of the picture.
//! - `grpc_stream`: `RepositoryService.StreamArtifact` — reused from the
//! `grpc_door` module, with its resident-set-per-artifact-byte column.
//! - `oci_chunked`: one blob pushed the way `podman push` pushes it — `POST`
//! to open, `PATCH` per chunk, `PUT ?digest=` to finalize — through the
//! file-backed OCI repository. MB/s for the whole push and, separately, the
//! wall time of the finalize alone, because that is where section 2.4 said
//! the blob was read back, hashed and written a second time.
//!
//! Every arm carries an ANSWER floor (the bytes came back identical) so a
//! broken door reads as a failure and never as a fast one.
//!
//! Dev-only, like the rest of this crate (plan ?8): nothing here is linked by a
//! shipped binary.
use std::io::Read;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use sha2::{Digest, Sha256};
use repository_file_oci::OciRepoFile;
use server_lib::raw::RawBackend;
use traits::RepositoryBackendTrait;
use crate::{bench_root, served, Scale, Throughput};
/// The large blob. `HOLGER_BENCH_LARGE_MB` overrides (default 1024 = 1 GiB).
fn large_bytes() -> usize {
std::env::var("HOLGER_BENCH_LARGE_MB")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.filter(|n| *n > 0)
.unwrap_or(1024)
* 1024
* 1024
}
const SMALL_BYTES: usize = 4 * 1024;
/// The OCI chunk: the size `podman` sends per PATCH by default is a stream, but
/// the registries it talks to see MiB-scale frames; 4 MiB keeps the PATCH count
/// honest without making the arm about HTTP overhead.
const OCI_CHUNK_BYTES: usize = 4 * 1024 * 1024;
fn small_ops(scale: &Scale) -> usize {
std::env::var("HOLGER_BENCH_SMALL_OPS")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.filter(|n| *n > 0)
.unwrap_or_else(|| scale.entries.clamp(2_000, 20_000))
}
fn fresh_dir(name: &str) -> Result<PathBuf> {
let dir = bench_root().join(format!("dataplane-{name}"));
if dir.exists() {
std::fs::remove_dir_all(&dir).ok();
}
std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
Ok(dir)
}
/// A deterministic byte stream of `len` bytes, produced 64 KiB at a time, so a
/// 1 GiB upload never exists whole on the CLIENT side either — the arm is
/// about the server's memory, and a client that holds a gigabyte would hide
/// the server's gigabyte inside the same process's resident set.
pub struct PatternReader {
remaining: usize,
x: u64,
hasher: Sha256,
}
impl PatternReader {
pub fn new(seed: u64, len: usize) -> Self {
Self {
remaining: len,
x: seed.wrapping_mul(0x9E37_79B9_7F4A_7C15).wrapping_add(1),
hasher: Sha256::new(),
}
}
/// The sha256 of everything read so far.
pub fn digest_hex(&self) -> String {
hex_lower(&self.hasher.clone().finalize())
}
}
impl Read for PatternReader {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let n = buf.len().min(self.remaining);
// Eight bytes per xorshift step: the generator must be faster than
// the door or the arm measures the generator.
let mut x = self.x;
for chunk in buf[..n].chunks_mut(8) {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
let le = x.to_le_bytes();
chunk.copy_from_slice(&le[..chunk.len()]);
}
self.x = x;
self.hasher.update(&buf[..n]);
self.remaining -= n;
Ok(n)
}
}
/// The bytes of a [`PatternReader`] with the same seed and length, materialised.
fn pattern_bytes(seed: u64, len: usize) -> Vec<u8> {
let mut v = vec![0u8; len];
PatternReader::new(seed, len)
.read_exact(&mut v)
.expect("pattern");
v
}
fn hex_lower(bytes: &[u8]) -> String {
let mut s = String::with_capacity(bytes.len() * 2);
for b in bytes {
s.push_str(&format!("{b:02x}"));
}
s
}
fn rss_bytes() -> u64 {
let Ok(s) = std::fs::read_to_string("/proc/self/statm") else {
return 0;
};
s.split_whitespace()
.nth(1)
.and_then(|v| v.parse::<u64>().ok())
.map(|pages| pages * 4096)
.unwrap_or(0)
}
/// Run `f` while a watcher thread samples this process's resident set every
/// 500 µs; returns `f`'s result and the peak GROWTH over the baseline taken
/// before `f` started. The server lives in this process, so the growth is the
/// server's.
pub fn with_rss_peak<T>(f: impl FnOnce() -> T) -> (T, u64) {
let base = rss_bytes();
let stop = Arc::new(AtomicBool::new(false));
let peak = Arc::new(AtomicU64::new(base));
let (s, p) = (stop.clone(), peak.clone());
let watcher = std::thread::spawn(move || {
while !s.load(Ordering::Relaxed) {
p.fetch_max(rss_bytes(), Ordering::Relaxed);
std::thread::sleep(Duration::from_micros(500));
}
});
let out = f();
stop.store(true, Ordering::Relaxed);
let _ = watcher.join();
(out, peak.load(Ordering::Relaxed).saturating_sub(base))
}
fn throughput(ops: usize, bytes: u64, elapsed: Duration) -> Throughput {
let s = elapsed.as_secs_f64().max(f64::MIN_POSITIVE);
Throughput {
ops: ops as u64,
elapsed,
bytes,
ops_sec: ops as f64 / s,
mb_per_sec: bytes as f64 / s / 1_048_576.0,
}
}
fn percentiles(lat: &mut [f32]) -> (f64, f64, f64, f64) {
if lat.is_empty() {
return (0.0, 0.0, 0.0, 0.0);
}
lat.sort_unstable_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let pick = |p: f64| -> f64 {
let idx = ((lat.len() as f64 - 1.0) * p).round() as usize;
lat[idx.min(lat.len() - 1)] as f64
};
let mean = lat.iter().map(|v| *v as f64).sum::<f64>() / lat.len() as f64;
(pick(0.50), pick(0.99), pick(0.999), mean)
}
fn client() -> Result<reqwest::blocking::Client> {
Ok(tls_roots_blocking(reqwest::blocking::Client::builder())
.tcp_keepalive(Duration::from_secs(30))
.timeout(Duration::from_secs(600))
.build()?)
}
struct Gateway {
rt: tokio::runtime::Runtime,
addr: SocketAddr,
dir: PathBuf,
}
impl Gateway {
fn start(label: &str, repos: Vec<(String, Arc<dyn RepositoryBackendTrait>)>) -> Result<Self> {
let dir = fresh_dir(label)?;
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.thread_name("dataplane-bench")
.build()
.context("build the server runtime")?;
let addr = rt.block_on(served::start_holger_gateway(repos))?;
Ok(Self { rt, addr, dir })
}
fn url(&self, path: &str) -> String {
format!("http://{}{}", self.addr, path)
}
fn shutdown(self) {
self.rt.shutdown_timeout(Duration::from_secs(2));
std::fs::remove_dir_all(&self.dir).ok();
}
}
// ── http_large ───────────────────────────────────────────────────────────────
#[derive(Debug, Clone)]
pub struct LargeOutcome {
pub bytes: usize,
pub put: Throughput,
pub get: Throughput,
/// Peak resident-set growth of the server process while the PUT was in
/// flight, and while the GET was.
pub rss_growth_put: u64,
pub rss_growth_get: u64,
/// `rss_growth_put / bytes`. ≈ 1.0 means the door held the whole body.
pub rss_per_byte_put: f64,
pub rss_per_byte_get: f64,
}
/// ONE large blob, PUT then GET, over the raw repository's HTTP door.
pub fn http_large(_scale: &Scale) -> Result<LargeOutcome> {
let len = large_bytes();
let store = fresh_dir("http-large-store")?;
let backend: Arc<dyn RepositoryBackendTrait> =
Arc::new(RawBackend::new("raw".into(), store.clone()));
let gw = Gateway::start("http-large", vec![("raw".to_string(), backend)])?;
let out = (|| -> Result<LargeOutcome> {
let c = client()?;
let url = gw.url("/raw/large/blob.bin");
// PUT: the body is generated as it is sent, digest checked at the door.
let expected = {
let mut r = PatternReader::new(1, len);
std::io::copy(&mut r, &mut std::io::sink())?;
r.digest_hex()
};
let (put_res, rss_put) = with_rss_peak(|| -> Result<Duration> {
let t0 = Instant::now();
let body = reqwest::blocking::Body::sized(PatternReader::new(1, len), len as u64);
let resp = c
.put(&url)
.header("X-Checksum-Sha256", expected.as_str())
.body(body)
.send()?;
let status = resp.status();
let _ = resp.bytes();
anyhow::ensure!(status.is_success(), "large PUT refused with {status}");
Ok(t0.elapsed())
});
let put_elapsed = put_res?;
// GET: streamed into a digest, never held whole on the client.
let (get_res, rss_get) = with_rss_peak(|| -> Result<(Duration, u64, String)> {
let t0 = Instant::now();
let mut resp = c.get(&url).send()?;
anyhow::ensure!(
resp.status().is_success(),
"large GET refused with {}",
resp.status()
);
let mut h = Sha256::new();
let mut buf = vec![0u8; 1 << 20];
let mut n = 0u64;
loop {
let k = resp.read(&mut buf)?;
if k == 0 {
break;
}
h.update(&buf[..k]);
n += k as u64;
}
Ok((t0.elapsed(), n, hex_lower(&h.finalize())))
});
let (get_elapsed, got, got_digest) = get_res?;
anyhow::ensure!(got == len as u64, "GET returned {got} of {len} bytes");
anyhow::ensure!(
got_digest == expected,
"GET returned different bytes than were PUT"
);
Ok(LargeOutcome {
bytes: len,
put: throughput(1, len as u64, put_elapsed),
get: throughput(1, len as u64, get_elapsed),
rss_growth_put: rss_put,
rss_growth_get: rss_get,
rss_per_byte_put: rss_put as f64 / len as f64,
rss_per_byte_get: rss_get as f64 / len as f64,
})
})();
gw.shutdown();
std::fs::remove_dir_all(&store).ok();
out
}
// ── http_small ───────────────────────────────────────────────────────────────
#[derive(Debug, Clone)]
pub struct SmallOutcome {
pub bytes_each: usize,
pub ops: usize,
pub put: Throughput,
pub get: Throughput,
pub put_p50_ms: f64,
pub put_p99_ms: f64,
pub put_p999_ms: f64,
pub get_p50_ms: f64,
pub get_p99_ms: f64,
pub get_p999_ms: f64,
}
/// N 4 KiB blobs, PUT then GET, serial, one connection.
pub fn http_small(scale: &Scale) -> Result<SmallOutcome> {
let n = small_ops(scale);
let store = fresh_dir("http-small-store")?;
let backend: Arc<dyn RepositoryBackendTrait> =
Arc::new(RawBackend::new("raw".into(), store.clone()));
let gw = Gateway::start("http-small", vec![("raw".to_string(), backend)])?;
let out = (|| -> Result<SmallOutcome> {
let c = client()?;
let bodies: Vec<Vec<u8>> = (0..n)
.map(|i| pattern_bytes(i as u64 + 100, SMALL_BYTES))
.collect();
let mut put_lat = Vec::with_capacity(n);
let t0 = Instant::now();
for (i, body) in bodies.iter().enumerate() {
let t = Instant::now();
let resp = c
.put(gw.url(&format!("/raw/small/{i:06}.bin")))
.body(body.clone())
.send()?;
let status = resp.status();
let _ = resp.bytes();
anyhow::ensure!(status.is_success(), "small PUT {i} refused with {status}");
put_lat.push(t.elapsed().as_secs_f32() * 1000.0);
}
let put_elapsed = t0.elapsed();
let mut get_lat = Vec::with_capacity(n);
let t0 = Instant::now();
let mut verified = 0usize;
for (i, body) in bodies.iter().enumerate() {
let t = Instant::now();
let resp = c.get(gw.url(&format!("/raw/small/{i:06}.bin"))).send()?;
anyhow::ensure!(
resp.status().is_success(),
"small GET {i} refused with {}",
resp.status()
);
let got = resp.bytes()?;
get_lat.push(t.elapsed().as_secs_f32() * 1000.0);
if got.as_ref() == body.as_slice() {
verified += 1;
}
}
let get_elapsed = t0.elapsed();
anyhow::ensure!(
verified == n,
"{} of {n} small GETs returned different bytes",
n - verified
);
let (pp50, pp99, pp999, _) = percentiles(&mut put_lat);
let (gp50, gp99, gp999, _) = percentiles(&mut get_lat);
Ok(SmallOutcome {
bytes_each: SMALL_BYTES,
ops: n,
put: throughput(n, (n * SMALL_BYTES) as u64, put_elapsed),
get: throughput(n, (n * SMALL_BYTES) as u64, get_elapsed),
put_p50_ms: pp50,
put_p99_ms: pp99,
put_p999_ms: pp999,
get_p50_ms: gp50,
get_p99_ms: gp99,
get_p999_ms: gp999,
})
})();
gw.shutdown();
std::fs::remove_dir_all(&store).ok();
out
}
// ── oci_chunked ──────────────────────────────────────────────────────────────
#[derive(Debug, Clone)]
pub struct OciOutcome {
pub bytes: usize,
pub chunks: usize,
/// The whole push: POST + every PATCH + the finalizing PUT.
pub push: Throughput,
/// The finalizing PUT alone: what the server does once the last byte is in.
pub finalize: Duration,
/// Peak resident-set growth of the server while the finalize ran. A
/// finalize that reads the staged blob back whole shows ≈ `bytes` here.
pub rss_growth_finalize: u64,
pub rss_per_byte_finalize: f64,
}
/// One blob pushed as `podman push` pushes it: open, N PATCHes, finalize.
/// `HOLGER_BENCH_OCI_MB` sets the blob size (default 256).
pub fn oci_chunked(_scale: &Scale) -> Result<OciOutcome> {
let len = std::env::var("HOLGER_BENCH_OCI_MB")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.filter(|n| *n > 0)
.unwrap_or(256)
* 1024
* 1024;
let store = fresh_dir("oci-store")?;
let backend: Arc<dyn RepositoryBackendTrait> =
Arc::new(OciRepoFile::new("oci".into(), store.clone())?.with_max_blob_bytes(8 << 30));
let gw = Gateway::start("oci-chunked", vec![("oci".to_string(), backend)])?;
let out = (|| -> Result<OciOutcome> {
let c = client()?;
let expected = {
let mut r = PatternReader::new(7, len);
std::io::copy(&mut r, &mut std::io::sink())?;
r.digest_hex()
};
let mut chunks = 0usize;
let t0 = Instant::now();
let resp = c
.post(gw.url("/v2/oci/bench/layer/blobs/uploads/"))
.send()?;
anyhow::ensure!(
resp.status().as_u16() == 202,
"upload open refused with {}",
resp.status()
);
let loc = resp
.headers()
.get("Location")
.and_then(|v| v.to_str().ok())
.context("upload open returned no Location")?
.to_string();
let mut r = PatternReader::new(7, len);
let mut sent = 0usize;
let mut buf = vec![0u8; OCI_CHUNK_BYTES];
while sent < len {
let k = r.read(&mut buf)?;
if k == 0 {
break;
}
let resp = c.patch(gw.url(&loc)).body(buf[..k].to_vec()).send()?;
let status = resp.status();
let _ = resp.bytes();
anyhow::ensure!(
status.as_u16() == 202,
"PATCH {chunks} refused with {status}"
);
sent += k;
chunks += 1;
}
let tf = Instant::now();
let (fin, rss_fin) = with_rss_peak(|| -> Result<()> {
let resp = c
.put(gw.url(&format!("{loc}?digest=sha256:{expected}")))
.send()?;
let status = resp.status();
let _ = resp.bytes();
anyhow::ensure!(status.as_u16() == 201, "finalize refused with {status}");
Ok(())
});
fin?;
let finalize = tf.elapsed();
let push_elapsed = t0.elapsed();
// ANSWER floor: the blob is served back identical.
let mut resp = c
.get(gw.url(&format!("/v2/oci/bench/layer/blobs/sha256:{expected}")))
.send()?;
anyhow::ensure!(
resp.status().is_success(),
"blob GET refused with {}",
resp.status()
);
let mut h = Sha256::new();
let mut n = 0u64;
loop {
let k = resp.read(&mut buf)?;
if k == 0 {
break;
}
h.update(&buf[..k]);
n += k as u64;
}
anyhow::ensure!(
n == len as u64 && hex_lower(&h.finalize()) == expected,
"the pushed blob did not come back identical"
);
Ok(OciOutcome {
bytes: len,
chunks,
push: throughput(1, len as u64, push_elapsed),
finalize,
rss_growth_finalize: rss_fin,
rss_per_byte_finalize: rss_fin as f64 / len as f64,
})
})();
gw.shutdown();
std::fs::remove_dir_all(&store).ok();
out
}
// ── digest ───────────────────────────────────────────────────────────────────
#[derive(Debug, Clone)]
pub struct DigestOutcome {
pub bytes: usize,
/// `blake3::hash` on the request thread — the serial reference.
pub serial_mbs: f64,
/// `server_lib::digest::blake3_hash` — the same digest on gatling.
pub gatling_mbs: f64,
pub speedup: f64,
}
/// The content digest alone, serial against the engine, over one buffer.
/// ANSWER floor: the two digests are identical.
pub fn digest_throughput(_scale: &Scale) -> Result<DigestOutcome> {
let len = std::env::var("HOLGER_BENCH_DIGEST_MB")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.filter(|n| *n > 0)
.unwrap_or(256)
* 1024
* 1024;
let data = pattern_bytes(11, len);
let rounds = 5;
// Warm both paths once (page-in, thread pool spin-up), then time.
let _ = blake3::hash(&data);
let _ = server_lib::digest::blake3_hash(&data);
let t0 = Instant::now();
let mut serial = None;
for _ in 0..rounds {
serial = Some(blake3::hash(&data));
}
let serial_secs = t0.elapsed().as_secs_f64() / rounds as f64;
let t0 = Instant::now();
let mut forked = None;
for _ in 0..rounds {
forked = Some(server_lib::digest::blake3_hash(&data));
}
let gatling_secs = t0.elapsed().as_secs_f64() / rounds as f64;
anyhow::ensure!(
serial == forked,
"the forked blake3 differs from the reference"
);
let mb = len as f64 / 1_048_576.0;
Ok(DigestOutcome {
bytes: len,
serial_mbs: mb / serial_secs,
gatling_mbs: mb / gatling_secs,
speedup: serial_secs / gatling_secs,
})
}
/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
#[allow(dead_code)]
fn tls_roots_blocking(b: reqwest::blocking::ClientBuilder) -> reqwest::blocking::ClientBuilder {
let _ = rustls::crypto::ring::default_provider().install_default();
b.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
}