//! Small-file throughput of the slot pool: fixed-slot mode vs the coalescing
//! mode added by Phase 1 of `.nornir/chunk_revolver-unsorted.md`.
//!
//! Both arms run the SAME 1-reader / N-worker pipeline over the SAME corpus at
//! the SAME memory reservation, and both must produce the same digest — the
//! only thing that changes is how the reader gets bytes to the workers:
//!
//! - `fixed-slot` — `try_get_chunk` / `return_chunk`: one file = one whole
//! slot, so the reservation buys `budget / max_file_size` files in flight and
//! every file costs a claim plus a return hop back through the reader.
//! - `coalescing` — `claim_slot` / `commit_slice` / `publish` +
//! `Returner::release_one`: many files packed into one slot, so the same
//! reservation buys thousands of files in flight and a claim is amortised
//! over a whole slot's worth of them.
//! - `coalescing+reuse` — the same, publishing through `publish_into` into one
//! reused buffer instead of `publish`'s fresh `Vec` per slot.
//!
//! Corpus: real files off disk (default: this repo's own tree; override with
//! `ZZ_BENCH_CORPUS=<dir>`), loaded into RAM up front so the bench measures the
//! pool and not the disk.
//!
//! - `magazine-fork` — a bench-local replica of znippy's `Magazine`
//! (`znippy-common/src/slotpool.rs`), the fork this mode exists to retire.
//! Faithful to its algorithm (bounded free channel, `AtomicUsize` per slot,
//! a fresh `Vec<Round>` per claim); the one deviation is `std`'s
//! `sync_channel` in place of `crossbeam_channel::bounded`, because this leaf
//! crate has no crossbeam dependency. It is here for ONE question: does
//! flipping znippy onto the canonical pool cost anything?
//!
//! Run: `cargo bench --bench revolver_coalescing`.
use std::alloc::{GlobalAlloc, Layout, System};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::mpsc::sync_channel;
use std::time::Instant;
use znippy_zoomies::chunk_revolver::{ChunkRevolver, PackedSlice, get_chunk_slice};
// ── allocation counter ──────────────────────────────────────────────────────
// The coalescing mode's claim is "steady state allocates nothing". That is a
// measurable claim, so it gets measured rather than asserted.
static ALLOCS: AtomicUsize = AtomicUsize::new(0);
struct Counting;
unsafe impl GlobalAlloc for Counting {
unsafe fn alloc(&self, l: Layout) -> *mut u8 {
ALLOCS.fetch_add(1, Ordering::Relaxed);
unsafe { System.alloc(l) }
}
unsafe fn dealloc(&self, p: *mut u8, l: Layout) {
unsafe { System.dealloc(p, l) }
}
unsafe fn realloc(&self, p: *mut u8, l: Layout, n: usize) -> *mut u8 {
ALLOCS.fetch_add(1, Ordering::Relaxed);
unsafe { System.realloc(p, l, n) }
}
}
#[global_allocator]
static ALLOC: Counting = Counting;
// ── workload ────────────────────────────────────────────────────────────────
const MAX_FILE: usize = 64 * 1024;
const BUDGET: usize = 64 * 1024 * 1024;
const SLOT: usize = 8 * 1024 * 1024;
const REPEATS: usize = 9;
#[inline]
fn digest(b: &[u8]) -> u64 {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for &x in b {
h ^= x as u64;
h = h.wrapping_mul(0x0000_0100_0000_01b3);
}
h
}
fn collect(dir: &Path, out: &mut Vec<PathBuf>, cap: usize) {
if out.len() >= cap {
return;
}
let Ok(rd) = std::fs::read_dir(dir) else {
return;
};
for e in rd.flatten() {
if out.len() >= cap {
return;
}
let p = e.path();
let Ok(m) = e.metadata() else { continue };
if m.is_dir() {
if p.file_name().is_some_and(|n| n == "target" || n == ".git") {
continue;
}
collect(&p, out, cap);
} else if m.is_file() && m.len() > 0 && (m.len() as usize) <= MAX_FILE {
out.push(p);
}
}
}
fn load_corpus() -> Vec<Vec<u8>> {
let root = std::env::var("ZZ_BENCH_CORPUS").unwrap_or_else(|_| {
Path::new(env!("CARGO_MANIFEST_DIR"))
.to_string_lossy()
.into_owned()
});
let mut paths = Vec::new();
collect(Path::new(&root), &mut paths, 200_000);
paths.sort();
paths
.into_iter()
.filter_map(|p| std::fs::read(p).ok())
.filter(|b| !b.is_empty())
.collect()
}
// ── arm A: fixed-slot (`try_get_chunk` / `return_chunk`) ────────────────────
fn fixed_slot(corpus: &[Vec<u8>], workers: usize) -> (u64, f64) {
let slots = BUDGET / MAX_FILE;
let mut rev = ChunkRevolver::new(MAX_FILE, slots, workers);
let rings = rev.num_rings();
let bases = rev.base_ptrs();
let total = Arc::new(AtomicU64::new(0));
let (ret_tx, ret_rx) = sync_channel::<(u8, u64)>(slots);
let mut job_tx = Vec::with_capacity(rings);
let mut handles = Vec::with_capacity(rings);
for r in 0..rings {
let (tx, rx) = sync_channel::<(u64, usize)>(64);
job_tx.push(tx);
let base = bases[r];
let ret = ret_tx.clone();
let total = Arc::clone(&total);
handles.push(std::thread::spawn(move || {
let mut acc: u64 = 0;
while let Ok((index, used)) = rx.recv() {
// SAFETY: the slot is in flight — the reader will not re-lease
// it until this thread posts the return below.
let bytes = unsafe { get_chunk_slice(base.as_ptr(), MAX_FILE, index as u32, used) };
acc = acc.wrapping_add(digest(bytes));
ret.send((r as u8, index)).expect("return channel");
}
total.fetch_add(acc, Ordering::Relaxed);
}));
}
drop(ret_tx);
let t0 = Instant::now();
for f in corpus {
let (ring, index) = loop {
if let Some(c) = rev.try_get_chunk() {
c.data[..f.len()].copy_from_slice(f);
break (c.ring_nr, c.index);
}
let (rn, ix) = ret_rx.recv().expect("a worker must return a slot");
rev.return_chunk(rn, ix);
};
job_tx[ring as usize]
.send((index, f.len()))
.expect("job channel");
}
drop(job_tx);
for h in handles {
h.join().expect("worker");
}
let secs = t0.elapsed().as_secs_f64();
(total.load(Ordering::Relaxed), secs)
}
// ── arms B/C: coalescing (`claim_slot` / `publish[_into]` / `release_one`) ──
fn coalescing(corpus: &[Vec<u8>], workers: usize, reuse: bool) -> (u64, f64) {
let slots = BUDGET / SLOT;
let rev = ChunkRevolver::with_coalescing(slots, SLOT, MAX_FILE);
let returner = rev.returner();
let total = Arc::new(AtomicU64::new(0));
let mut job_tx = Vec::with_capacity(workers);
let mut handles = Vec::with_capacity(workers);
for _ in 0..workers {
let (tx, rx) = sync_channel::<PackedSlice>(256);
job_tx.push(tx);
let ret = returner.clone();
let total = Arc::clone(&total);
handles.push(std::thread::spawn(move || {
let mut acc: u64 = 0;
while let Ok(s) = rx.recv() {
// SAFETY: the slot is published and this slice's release has
// not been posted yet, so the bytes are live and immutable.
let bytes = unsafe { s.as_slice() };
acc = acc.wrapping_add(digest(bytes));
ret.release_one(s.slot_id);
}
total.fetch_add(acc, Ordering::Relaxed);
}));
}
let mut published: Vec<PackedSlice> = Vec::with_capacity(SLOT / MAX_FILE + 1);
let mut next = 0usize;
let t0 = Instant::now();
let mut filler = rev.claim_slot().expect("first slot");
for (i, f) in corpus.iter().enumerate() {
if filler.remaining() < f.len() {
if reuse {
filler.publish_into(&mut published);
} else {
published = filler.publish();
}
for s in published.drain(..) {
job_tx[next].send(s).expect("job channel");
next = (next + 1) % workers;
}
filler = rev.claim_slot().expect("a worker must free a slot");
}
filler.writable(f.len()).copy_from_slice(f);
filler.commit_slice(f.len(), false, i as u64, 0, 0);
}
if reuse {
filler.publish_into(&mut published);
} else {
published = filler.publish();
}
for s in published.drain(..) {
job_tx[next].send(s).expect("job channel");
next = (next + 1) % workers;
}
drop(job_tx);
for h in handles {
h.join().expect("worker");
}
let secs = t0.elapsed().as_secs_f64();
(total.load(Ordering::Relaxed), secs)
}
// ── arm D: the Magazine fork, replicated ────────────────────────────────────
// Kept deliberately close to `znippy-common/src/slotpool.rs` so the comparison
// is against the fork's real mechanics, not a strawman.
mod fork {
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, SyncSender, sync_channel};
#[derive(Copy, Clone)]
struct SlotPtr(*mut u8);
unsafe impl Send for SlotPtr {}
unsafe impl Sync for SlotPtr {}
pub struct Round {
pub slot_id: u32,
ptr: *const u8,
pub len: usize,
}
unsafe impl Send for Round {}
impl Round {
/// # Safety
/// The slot must be unreleased.
pub unsafe fn as_slice<'a>(&self) -> &'a [u8] {
unsafe { std::slice::from_raw_parts(self.ptr, self.len) }
}
}
struct EjectorInner {
free_tx: SyncSender<u32>,
outstanding: Vec<AtomicUsize>,
}
#[derive(Clone)]
pub struct Ejector {
inner: Arc<EjectorInner>,
}
impl Ejector {
pub fn release_one(&self, slot_id: u32) {
let prev = self.inner.outstanding[slot_id as usize].fetch_sub(1, Ordering::AcqRel);
if prev == 1 {
self.inner.free_tx.send(slot_id).ok();
}
}
}
pub struct Magazine {
slot_size: usize,
base: Vec<SlotPtr>,
_mem: Vec<Box<[u8]>>,
free_rx: Receiver<u32>,
ret: Ejector,
}
impl Magazine {
pub fn new(num_slots: usize, slot_size: usize) -> Self {
let mut mem: Vec<Box<[u8]>> = (0..num_slots)
.map(|_| vec![0u8; slot_size].into_boxed_slice())
.collect();
let base: Vec<SlotPtr> = mem.iter_mut().map(|b| SlotPtr(b.as_mut_ptr())).collect();
let (free_tx, free_rx) = sync_channel(num_slots);
for id in 0..num_slots as u32 {
free_tx.send(id).expect("init");
}
let outstanding = (0..num_slots).map(|_| AtomicUsize::new(0)).collect();
Magazine {
slot_size,
base,
_mem: mem,
free_rx,
ret: Ejector {
inner: Arc::new(EjectorInner {
free_tx,
outstanding,
}),
},
}
}
pub fn returner(&self) -> Ejector {
self.ret.clone()
}
pub fn claim(&self) -> Option<Clip<'_>> {
let slot_id = self.free_rx.recv().ok()?;
Some(Clip {
pool: self,
slot_id,
cursor: 0,
slices: Vec::new(),
})
}
}
pub struct Clip<'a> {
pool: &'a Magazine,
slot_id: u32,
cursor: usize,
slices: Vec<Round>,
}
impl<'a> Clip<'a> {
pub fn remaining(&self) -> usize {
self.pool.slot_size - self.cursor
}
pub fn writable(&mut self, max: usize) -> &mut [u8] {
let n = max.min(self.remaining());
let base = self.pool.base[self.slot_id as usize].0;
unsafe { std::slice::from_raw_parts_mut(base.add(self.cursor), n) }
}
pub fn commit_slice(&mut self, len: usize) {
let base = self.pool.base[self.slot_id as usize].0 as *const u8;
let ptr = unsafe { base.add(self.cursor) };
self.slices.push(Round {
slot_id: self.slot_id,
ptr,
len,
});
self.cursor += len;
}
#[must_use]
pub fn publish(self) -> Vec<Round> {
let n = self.slices.len();
if n == 0 {
self.pool.ret.inner.free_tx.send(self.slot_id).ok();
return Vec::new();
}
self.pool.ret.inner.outstanding[self.slot_id as usize].store(n, Ordering::Release);
self.slices
}
}
}
fn magazine_fork(corpus: &[Vec<u8>], workers: usize) -> (u64, f64) {
let slots = BUDGET / SLOT;
let pool = fork::Magazine::new(slots, SLOT);
let returner = pool.returner();
let total = Arc::new(AtomicU64::new(0));
let mut job_tx = Vec::with_capacity(workers);
let mut handles = Vec::with_capacity(workers);
for _ in 0..workers {
let (tx, rx) = sync_channel::<fork::Round>(256);
job_tx.push(tx);
let ret = returner.clone();
let total = Arc::clone(&total);
handles.push(std::thread::spawn(move || {
let mut acc: u64 = 0;
while let Ok(s) = rx.recv() {
// SAFETY: published and not yet released.
let bytes = unsafe { s.as_slice() };
acc = acc.wrapping_add(digest(bytes));
ret.release_one(s.slot_id);
}
total.fetch_add(acc, Ordering::Relaxed);
}));
}
let mut next = 0usize;
let t0 = Instant::now();
let mut clip = pool.claim().expect("first slot");
for f in corpus.iter() {
if clip.remaining() < f.len() {
for s in clip.publish() {
job_tx[next].send(s).expect("job channel");
next = (next + 1) % workers;
}
clip = pool.claim().expect("a worker must free a slot");
}
clip.writable(f.len()).copy_from_slice(f);
clip.commit_slice(f.len());
}
for s in clip.publish() {
job_tx[next].send(s).expect("job channel");
next = (next + 1) % workers;
}
drop(job_tx);
for h in handles {
h.join().expect("worker");
}
(total.load(Ordering::Relaxed), t0.elapsed().as_secs_f64())
}
fn main() {
let corpus = load_corpus();
assert!(
corpus.len() > 100,
"corpus too small ({} files) — set ZZ_BENCH_CORPUS",
corpus.len()
);
let bytes: usize = corpus.iter().map(|f| f.len()).sum();
let workers = std::thread::available_parallelism().map_or(4, |n| n.get());
println!(
"corpus: {} files, {:.1} MiB, mean {} B/file · workers {workers} · budget {} MiB",
corpus.len(),
bytes as f64 / (1024.0 * 1024.0),
bytes / corpus.len(),
BUDGET / (1024 * 1024),
);
println!(
"{:<20} {:>10} {:>12} {:>14} {:>12}",
"arm", "best s", "MiB/s", "files/s", "allocs"
);
// Interleave the arms: one round runs every arm once, and the best of N
// rounds is reported per arm. Running all N reps of one arm back-to-back
// hands whichever arm ran during a quiet minute a win that is the machine's,
// not the code's — this box shares its cores with other work.
type Arm<'c> = (&'static str, Box<dyn Fn() -> (u64, f64) + 'c>);
let arms: Vec<Arm> = vec![
("fixed-slot", Box::new(|| fixed_slot(&corpus, workers))),
(
"magazine-fork",
Box::new(|| magazine_fork(&corpus, workers)),
),
(
"coalescing",
Box::new(|| coalescing(&corpus, workers, false)),
),
(
"coalescing+reuse",
Box::new(|| coalescing(&corpus, workers, true)),
),
];
let mut best = vec![(f64::INFINITY, 0usize); arms.len()];
let mut reference: Option<u64> = None;
for _ in 0..REPEATS {
for (i, (name, f)) in arms.iter().enumerate() {
ALLOCS.store(0, Ordering::Relaxed);
let (d, s) = f();
let a = ALLOCS.load(Ordering::Relaxed);
match reference {
None => reference = Some(d),
Some(r) => assert_eq!(r, d, "{name} produced a different digest"),
}
if s < best[i].0 {
best[i] = (s, a);
}
}
}
for (i, (name, _)) in arms.iter().enumerate() {
let (s, a) = best[i];
println!(
"{name:<20} {s:>10.3} {:>12.1} {:>14.0} {a:>12}",
bytes as f64 / (1024.0 * 1024.0) / s,
corpus.len() as f64 / s,
);
}
}