holperf/src/harness.rs
//! HOL-PERF-FORMATS scratch harness (lane-local, never committed).
//!
//! Three tiers, the same for every format:
//! * in-process: the backend's own serving call (the door's streamed read when the
//! backend has `streaming_reads + path_mapping`, then its `stream_get`, else
//! `handle_http2_request`),
//! single thread, with a counting allocator → latency p50/p99 + allocs/bytes per req;
//! * cold: a FRESH backend over the same archive (empty in-process caches) with the
//! archive's pages evicted (posix_fadvise DONTNEED), first request timed;
//! * served: the real native HTTP door (`start_holger_gateway`), a reqwest pool →
//! p50/p99 at c=1, req/s at c=32, MB/s for small and large artifacts.
use std::alloc::{GlobalAlloc, Layout, System};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use traits::RepositoryBackendTrait;
// ───────────────────────── counting allocator ─────────────────────────
pub struct Counting;
static ON: AtomicBool = AtomicBool::new(false);
static N_ALLOC: AtomicU64 = AtomicU64::new(0);
static B_ALLOC: AtomicU64 = AtomicU64::new(0);
static LIVE: AtomicU64 = AtomicU64::new(0);
static PEAK: AtomicU64 = AtomicU64::new(0);
/// Every live heap byte of the process, always counted (the counters above count one call).
static ALL_LIVE: AtomicU64 = AtomicU64::new(0);
static ALL_PEAK: AtomicU64 = AtomicU64::new(0);
/// The process's live heap now, and its peak since the last [`reset_heap_peak`].
pub fn heap_live_peak() -> (u64, u64) {
(
ALL_LIVE.load(Ordering::Relaxed),
ALL_PEAK.load(Ordering::Relaxed),
)
}
/// Start a new heap peak from the live heap now.
pub fn reset_heap_peak() {
ALL_PEAK.store(ALL_LIVE.load(Ordering::Relaxed), Ordering::Relaxed);
}
unsafe impl GlobalAlloc for Counting {
unsafe fn alloc(&self, l: Layout) -> *mut u8 {
let p = System.alloc(l);
if !p.is_null() {
let a = ALL_LIVE.fetch_add(l.size() as u64, Ordering::Relaxed) + l.size() as u64;
ALL_PEAK.fetch_max(a, Ordering::Relaxed);
}
if ON.load(Ordering::Relaxed) && !p.is_null() {
N_ALLOC.fetch_add(1, Ordering::Relaxed);
B_ALLOC.fetch_add(l.size() as u64, Ordering::Relaxed);
let live = LIVE.fetch_add(l.size() as u64, Ordering::Relaxed) + l.size() as u64;
PEAK.fetch_max(live, Ordering::Relaxed);
}
p
}
unsafe fn dealloc(&self, p: *mut u8, l: Layout) {
ALL_LIVE.fetch_sub(l.size() as u64, Ordering::Relaxed);
if ON.load(Ordering::Relaxed) {
LIVE.fetch_sub(l.size() as u64, Ordering::Relaxed);
}
System.dealloc(p, l)
}
unsafe fn realloc(&self, p: *mut u8, l: Layout, n: usize) -> *mut u8 {
let q = System.realloc(p, l, n);
if !q.is_null() {
if n >= l.size() {
let a = ALL_LIVE.fetch_add((n - l.size()) as u64, Ordering::Relaxed)
+ (n - l.size()) as u64;
ALL_PEAK.fetch_max(a, Ordering::Relaxed);
} else {
ALL_LIVE.fetch_sub((l.size() - n) as u64, Ordering::Relaxed);
}
}
if ON.load(Ordering::Relaxed) && !q.is_null() {
N_ALLOC.fetch_add(1, Ordering::Relaxed);
B_ALLOC.fetch_add(n as u64, Ordering::Relaxed);
let live = if n >= l.size() {
LIVE.fetch_add((n - l.size()) as u64, Ordering::Relaxed) + (n - l.size()) as u64
} else {
LIVE.fetch_sub((l.size() - n) as u64, Ordering::Relaxed) - (l.size() - n) as u64
};
PEAK.fetch_max(live, Ordering::Relaxed);
}
q
}
}
/// Allocations, bytes allocated and peak live bytes over one call.
pub fn counted<T>(f: impl FnOnce() -> T) -> (T, u64, u64, u64) {
N_ALLOC.store(0, Ordering::SeqCst);
B_ALLOC.store(0, Ordering::SeqCst);
LIVE.store(0, Ordering::SeqCst);
PEAK.store(0, Ordering::SeqCst);
ON.store(true, Ordering::SeqCst);
let t = f();
ON.store(false, Ordering::SeqCst);
(
t,
N_ALLOC.load(Ordering::SeqCst),
B_ALLOC.load(Ordering::SeqCst),
PEAK.load(Ordering::SeqCst),
)
}
// ───────────────────────── page-cache eviction ─────────────────────────
/// Evict every file under `dir` from the page cache (no root needed for own files).
pub fn evict(dir: &Path) {
fn walk(p: &Path) {
if let Ok(rd) = std::fs::read_dir(p) {
for e in rd.flatten() {
let path = e.path();
if path.is_dir() {
walk(&path);
} else if let Ok(f) = std::fs::File::open(&path) {
use std::os::unix::io::AsRawFd;
let _ = f.sync_all();
unsafe {
libc::posix_fadvise(f.as_raw_fd(), 0, 0, libc::POSIX_FADV_DONTNEED);
}
}
}
}
}
walk(dir);
}
// ───────────────────────── one request, in process ─────────────────────────
/// Serve `suburl` the way the HTTP door does for a plain (non-intake) repository:
/// the streamed read door first when the backend can, else the whole-body door.
/// Returns (status, body bytes).
pub fn serve(b: &dyn RepositoryBackendTrait, suburl: &str) -> (u16, u64) {
let caps = b.caps();
if caps.streaming_reads && caps.path_mapping {
if let Some(id) = b.coordinate(suburl) {
if let Ok(Some(chunks)) = b.fetch_chunks(&id, 256 * 1024) {
let mut n = 0u64;
for c in chunks {
match c {
Ok(piece) => n += piece.len() as u64,
Err(_) => return (500, n),
}
}
return (200, n);
}
}
}
// the whole-body door's first question: a download the backend reads in pieces
if let Ok(Some(st)) = b.stream_get(suburl, 1024 * 1024) {
let mut n = 0u64;
for c in st.body {
match c {
Ok(piece) => n += piece.len() as u64,
Err(_) => return (500, n),
}
}
return (200, n);
}
match b.handle_http2_request("GET", suburl, &[]) {
Ok((s, _, body)) => (s, body.len() as u64),
Err(_) => (500, 0),
}
}
pub fn pct(v: &mut [f64], p: f64) -> f64 {
if v.is_empty() {
return 0.0;
}
v.sort_by(|a, b| a.partial_cmp(b).unwrap());
v[((v.len() as f64 * p).floor() as usize).min(v.len() - 1)]
}
#[derive(Debug, Default, Clone)]
pub struct InProc {
pub status: u16,
pub bytes: u64,
pub p50_us: f64,
pub p99_us: f64,
pub allocs: u64,
pub alloc_bytes: u64,
pub peak_bytes: u64,
pub first_us: f64,
}
/// Warm in-process: one untimed call (fills caches), then `iters` timed calls
/// cycling `urls`; alloc counters over one more call of urls[0].
pub fn inproc(b: &dyn RepositoryBackendTrait, urls: &[String], iters: usize) -> InProc {
let t0 = Instant::now();
let (status, bytes) = serve(b, &urls[0]);
let first_us = t0.elapsed().as_secs_f64() * 1e6;
for u in urls {
serve(b, u);
}
let mut lat = Vec::with_capacity(iters);
let mut i = 0usize;
let deadline = Instant::now() + Duration::from_secs(8);
while lat.len() < iters && Instant::now() < deadline {
let u = &urls[i % urls.len()];
let t = Instant::now();
std::hint::black_box(serve(b, u));
lat.push(t.elapsed().as_secs_f64() * 1e6);
i += 1;
}
let (_, allocs, alloc_bytes, peak_bytes) = counted(|| serve(b, &urls[0]));
InProc {
status,
bytes,
p50_us: pct(&mut lat.clone(), 0.5),
p99_us: pct(&mut lat, 0.99),
allocs,
alloc_bytes,
peak_bytes,
first_us,
}
}
// ───────────────────────── served over HTTP ─────────────────────────
#[derive(Debug, Default, Clone)]
pub struct Served {
pub ok: u64,
pub fail: u64,
pub rps: f64,
pub mbs: f64,
pub p50_ms: f64,
pub p99_ms: f64,
}
/// Read `r` to its end through `buf`, keeping nothing: the bytes read.
pub fn drain(r: &mut impl std::io::Read, buf: &mut [u8]) -> std::io::Result<u64> {
let mut n = 0u64;
loop {
match r.read(buf) {
Ok(0) => return Ok(n),
Ok(k) => n += k as u64,
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
Err(e) => return Err(e),
}
}
}
pub fn served(urls: &[String], conc: usize, dur: Duration) -> Served {
let start = Instant::now();
let res: Vec<(u64, u64, u64, Vec<f64>)> = std::thread::scope(|s| {
let hs: Vec<_> = (0..conc)
.map(|tid| {
s.spawn(move || {
let c = reqwest::blocking::Client::builder()
.pool_max_idle_per_host(4)
.timeout(Duration::from_secs(120))
.build()
.unwrap();
let (mut ok, mut fail, mut bytes) = (0u64, 0u64, 0u64);
let mut lat = Vec::new();
let mut buf = vec![0u8; 64 * 1024];
let mut i = tid * 7919;
while start.elapsed() < dur {
let u = &urls[i % urls.len()];
i += 1;
let t = Instant::now();
match c.get(u).send() {
Ok(mut r) => {
let good = r.status().is_success();
// read through one buffer and drop the bytes: the
// client keeps no body, so the process's memory is
// the server's (lane HOL-RSS)
match drain(&mut r, &mut buf) {
Ok(n) if good => {
ok += 1;
bytes += n;
lat.push(t.elapsed().as_secs_f64() * 1e3);
}
_ => fail += 1,
}
}
Err(_) => fail += 1,
}
}
(ok, fail, bytes, lat)
})
})
.collect();
hs.into_iter().map(|h| h.join().unwrap()).collect()
});
let el = start.elapsed().as_secs_f64();
let mut out = Served::default();
let mut lat = Vec::new();
let mut bytes = 0u64;
for (o, f, b, mut l) in res {
out.ok += o;
out.fail += f;
bytes += b;
lat.append(&mut l);
}
out.rps = out.ok as f64 / el;
out.mbs = bytes as f64 / 1_048_576.0 / el;
out.p50_ms = pct(&mut lat.clone(), 0.5);
out.p99_ms = pct(&mut lat, 0.99);
out
}
/// A format under test.
pub struct Scenario {
pub format: &'static str,
pub route: String,
/// Builds a FRESH backend over the seeded store (for the cold tier).
pub open: Box<dyn Fn() -> Arc<dyn RepositoryBackendTrait> + Send + Sync>,
pub dir: PathBuf,
/// Index / metadata suburls (`/<route>/...`), cycled.
pub index: Vec<String>,
/// Second-level metadata (e.g. Release / repomd / packument abbreviated), optional.
pub meta2: Vec<String>,
pub small: Vec<String>,
pub large: Vec<String>,
pub small_size: u64,
pub large_size: u64,
pub packages: usize,
}
/// The real HTTP door with the given routes and (optionally) the intake chain:
/// the same settings `bench::served::start_holger_gateway` uses, plus `intake`.
pub async fn start_gateway(
repos: Vec<(String, Arc<dyn RepositoryBackendTrait>)>,
intake: Option<Arc<server_lib::promote::intake::IntakeChain>>,
) -> anyhow::Result<std::net::SocketAddr> {
use server_lib::exposed::fast_routes::FastRoutes;
use server_lib::exposed::http::{start_http_gateway, GatewaySettings};
let l = std::net::TcpListener::bind("127.0.0.1:0")?;
let addr = l.local_addr()?;
drop(l);
let settings = Arc::new(GatewaySettings {
routes: FastRoutes::new(repos),
auth_config: Arc::new(server_lib::auth::AuthConfig::default()),
read_gate: Arc::new(server_lib::auth::ReadGate::GLOBAL_ONLY),
tls: None,
max_body_bytes: 1024 * 1024 * 1024,
audit: Arc::new(server_lib::audit::NoopAuditLog),
operation_mode: Default::default(),
twin: Default::default(),
grow: Default::default(),
security_profile: Default::default(),
admission: server_lib::admission::DoorAdmission::pass_through(),
scan_probe: None,
ready: None,
sboms: None,
attest: None,
evidence_key: None,
repo_key: None,
reserved_properties: Default::default(),
promotion_required_property: None,
properties: None,
intake,
imports: None,
admit_upload: None,
trusted_proxies: None,
door: "bench".to_string(),
visitors: Arc::new(server_lib::visitors::Visitors::empty()),
console: Default::default(),
peers: Default::default(),
board: None,
flow: None,
passkeys: None,
});
start_http_gateway(addr, settings)?;
let deadline = Instant::now() + Duration::from_secs(10);
while Instant::now() < deadline {
if std::net::TcpStream::connect(addr).is_ok() {
return Ok(addr);
}
std::thread::sleep(Duration::from_millis(20));
}
anyhow::bail!("gateway not up")
}
// ───────────────────────── memory by phase ─────────────────────────
/// The process's peak RSS and peak live heap phase by phase (lane HOL-RSS): each
/// [`mark`](Self::mark) records the phase that just ended and starts the next from the
/// RSS and heap of now, so a row shows WHICH tier held the memory, and [`hwm_mib`](Self::hwm_mib)
/// is the peak over every phase.
#[derive(Default)]
pub struct Phases {
pub rows: serde_json::Map<String, serde_json::Value>,
hwm_mib: f64,
heap_peak_mib: f64,
}
/// One key of `/proc/self/status`, KiB.
pub fn status_kib(key: &str) -> f64 {
std::fs::read_to_string("/proc/self/status")
.unwrap_or_default()
.lines()
.find(|l| l.starts_with(key))
.and_then(|l| l.split_whitespace().nth(1))
.and_then(|v| v.parse().ok())
.unwrap_or(0.0)
}
const MIB: f64 = 1024.0 * 1024.0;
impl Phases {
/// Start: the RSS peak and the heap peak count from now.
pub fn start() -> Self {
let _ = std::fs::write("/proc/self/clear_refs", "5");
reset_heap_peak();
Self::default()
}
/// The phase `name` ended now: keep its peaks, start the next phase's.
pub fn mark(&mut self, name: &str) {
let hwm = status_kib("VmHWM:") / 1024.0;
let (live, peak) = heap_live_peak();
let (live, peak) = (live as f64 / MIB, peak as f64 / MIB);
self.hwm_mib = self.hwm_mib.max(hwm);
self.heap_peak_mib = self.heap_peak_mib.max(peak);
self.rows.insert(
name.into(),
serde_json::json!({ "hwm_mib": hwm, "heap_peak_mib": peak, "heap_live_mib": live,
"rss_mib": status_kib("VmRSS:") / 1024.0,
"rss_anon_mib": status_kib("RssAnon:") / 1024.0,
"rss_file_mib": status_kib("RssFile:") / 1024.0,
"threads": status_kib("Threads:") }),
);
let _ = std::fs::write("/proc/self/clear_refs", "5");
reset_heap_peak();
}
/// The peak RSS over every phase marked, MiB.
pub fn hwm_mib(&self) -> f64 {
self.hwm_mib
}
/// The peak live heap over every phase marked, MiB.
pub fn heap_peak_mib(&self) -> f64 {
self.heap_peak_mib
}
}