From 71ffb5259a246c0a0e97c84152f176a2cb7ff4be Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Fri, 6 Nov 2026 00:48:04 +0000 Subject: [PATCH] [star] add cpu and allocation benchmarks --- AGENTS.md | 1 + Cargo.lock | 2 + star/Cargo.toml | 10 +++ star/benches/alloc.rs | 157 +++++++++++++++++++++++++++++++++++++ star/benches/common/mod.rs | 103 ++++++++++++++++++++++++ star/benches/star.rs | 52 ++++++++++++ 6 files changed, 325 insertions(+) create mode 100644 star/benches/alloc.rs create mode 100644 star/benches/common/mod.rs create mode 100644 star/benches/star.rs diff --git a/AGENTS.md b/AGENTS.md index 24f1cbb..69ef685 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -142,6 +142,7 @@ Hydrant uses multiple `fjall` keyspaces: - `cargo test --lib --no-default-features --features relay` - The lib tests in relay mode. `cargo test --lib` only runs the default indexer build, so run both after touching shared code like `config` or `db`. - `cargo test --test representation_audit` - Checks that every representation boundary (serde types and attributes, codec calls, db key codecs and constants, car and stream code) has a row in `tests/representation_inventory.tsv` naming a test that runs it. `cargo test --lib` doesn't run it, so run it after touching any of those. Every `missing:` or `stale:` line is a failure, including ones that were there before your change. - `nu tests/bench.nu (fetch | compare-ref [filter] | save | compare )` - Criterion benchmarks (`benches/repo.rs`) for CAR parsing with and without CID checks, the full MST walk, the sparse scan, the record-storage check, and a full backfill minus the database, on real repos that `fetch` downloads once into `target/bench-data`. `compare-ref main` is the regression gate: it builds `main` in a worktree and this tree, runs each benchmark `main`, this tree, this tree, `main` back to back, and exits 1 when the geometric mean of the two comparisons is slower by more than `--threshold` percent (default 10) at the low end of its 95% interval; the full set takes about 15 minutes. `save`/`compare` against a named baseline is only a rough check within one sitting, on the machine that saved it. Run after changing any of those paths, from a normal-priority shell on a quiet machine. +- `cargo bench -p dasl-star --bench star` / `--bench alloc [filter]` - Criterion timings and exact allocation counts for the STAR writer and reader. Save a baseline with `-- --save-baseline ` before a change and compare with `-- --baseline-lenient ` after. - `nu tests/api_crawler_sources.nu` - Tests `/crawler/sources` CRUD plus dynamic/configured source restart behavior. - `nu tests/api_firehose_sources.nu` - Tests `/firehose/sources` CRUD behavior. - `nu tests/api_pds_tiers.nu` - Tests PDS tier APIs, tier persistence, and custom rate tiers. diff --git a/Cargo.lock b/Cargo.lock index b2db2c7..0177fed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -828,11 +828,13 @@ dependencies = [ name = "dasl-star" version = "0.1.0" dependencies = [ + "criterion", "ipld-core", "serde", "serde_ipld_dagcbor 0.6.4", "serde_json", "thiserror", + "zstd", ] [[package]] diff --git a/star/Cargo.toml b/star/Cargo.toml index 744f539..edff989 100644 --- a/star/Cargo.toml +++ b/star/Cargo.toml @@ -13,5 +13,15 @@ serde_ipld_dagcbor = "0.6" thiserror = "2.0" [dev-dependencies] +criterion = { version = "0.8", default-features = false, features = ["cargo_bench_support"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +zstd = "0.13" + +[[bench]] +name = "star" +harness = false + +[[bench]] +name = "alloc" +harness = false diff --git a/star/benches/alloc.rs b/star/benches/alloc.rs new file mode 100644 index 0000000..c7d1b10 --- /dev/null +++ b/star/benches/alloc.rs @@ -0,0 +1,157 @@ +//! zstd allocates through c malloc, which this doesn't see. the zstd rows count only the rust side. +//! run with `cargo bench -p dasl-star --bench alloc [filter]` + +#[allow(dead_code)] +mod common; + +use std::alloc::{GlobalAlloc, Layout, System}; +use std::hint::black_box; +use std::io::{self, BufReader}; +use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}; + +struct CountingAlloc; + +#[global_allocator] +static GLOBAL: CountingAlloc = CountingAlloc; + +static MEASURING: AtomicBool = AtomicBool::new(false); +static ALLOCS: AtomicU64 = AtomicU64::new(0); +static REALLOCS: AtomicU64 = AtomicU64::new(0); +static TOTAL_BYTES: AtomicU64 = AtomicU64::new(0); +static LIVE_BYTES: AtomicI64 = AtomicI64::new(0); +static PEAK_BYTES: AtomicI64 = AtomicI64::new(0); + +fn grow(size: usize) { + TOTAL_BYTES.fetch_add(size as u64, Ordering::Relaxed); + let live = LIVE_BYTES.fetch_add(size as i64, Ordering::Relaxed) + size as i64; + PEAK_BYTES.fetch_max(live, Ordering::Relaxed); +} + +unsafe impl GlobalAlloc for CountingAlloc { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let ptr = unsafe { System.alloc(layout) }; + if !ptr.is_null() && MEASURING.load(Ordering::Relaxed) { + ALLOCS.fetch_add(1, Ordering::Relaxed); + grow(layout.size()); + } + ptr + } + + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + let ptr = unsafe { System.alloc_zeroed(layout) }; + if !ptr.is_null() && MEASURING.load(Ordering::Relaxed) { + ALLOCS.fetch_add(1, Ordering::Relaxed); + grow(layout.size()); + } + ptr + } + + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + if MEASURING.load(Ordering::Relaxed) { + LIVE_BYTES.fetch_sub(layout.size() as i64, Ordering::Relaxed); + } + unsafe { System.dealloc(ptr, layout) }; + } + + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { + let new_ptr = unsafe { System.realloc(ptr, layout, new_size) }; + if !new_ptr.is_null() && MEASURING.load(Ordering::Relaxed) { + REALLOCS.fetch_add(1, Ordering::Relaxed); + match new_size.checked_sub(layout.size()) { + Some(more) => grow(more), + None => { + LIVE_BYTES.fetch_sub((layout.size() - new_size) as i64, Ordering::Relaxed); + } + } + } + new_ptr + } +} + +type Run<'a> = Box Stats + 'a>; + +struct Stats { + allocs: u64, + reallocs: u64, + total_bytes: u64, + peak_bytes: u64, +} + +fn measure(f: impl FnOnce() -> T) -> Stats { + for counter in [&ALLOCS, &REALLOCS, &TOTAL_BYTES] { + counter.store(0, Ordering::SeqCst); + } + LIVE_BYTES.store(0, Ordering::SeqCst); + PEAK_BYTES.store(0, Ordering::SeqCst); + MEASURING.store(true, Ordering::SeqCst); + black_box(f()); + MEASURING.store(false, Ordering::SeqCst); + Stats { + allocs: ALLOCS.load(Ordering::SeqCst), + reallocs: REALLOCS.load(Ordering::SeqCst), + total_bytes: TOTAL_BYTES.load(Ordering::SeqCst), + peak_bytes: PEAK_BYTES.load(Ordering::SeqCst).max(0) as u64, + } +} + +fn main() { + let filter = std::env::args().skip(1).find(|arg| !arg.starts_with('-')); + println!( + "{:<28} {:>9} {:>9} {:>14} {:>12}", + "bench", "allocs", "reallocs", "total_bytes", "peak_bytes" + ); + for w in common::workloads() { + let plain = common::encode(&w); + let compressed = common::encode_zstd(&w); + let rows: [(&str, Run); 5] = [ + ( + "write/vec", + Box::new(|| { + let mut out = Vec::with_capacity(plain.len()); + measure(|| common::write(&mut out, &w)) + }), + ), + ( + "write/zstd", + Box::new(|| { + measure(|| { + let mut out = zstd::Encoder::new(io::sink(), common::ZSTD_LEVEL).unwrap(); + common::write(&mut out, &w); + out.finish().unwrap() + }) + }), + ), + ( + "read/slice", + Box::new(|| measure(|| common::read(&plain[..]))), + ), + ( + "read/owned", + Box::new(|| measure(|| common::read_owned(&plain[..]))), + ), + ( + "read/zstd", + Box::new(|| { + measure(|| { + let decoder = zstd::Decoder::new(&compressed[..]).unwrap(); + common::read(BufReader::new(decoder)) + }) + }), + ), + ]; + for (name, run) in rows { + let name = format!("{}/{name}", w.name); + if filter + .as_ref() + .is_some_and(|pat| !name.contains(pat.as_str())) + { + continue; + } + let s = run(); + println!( + "{name:<28} {:>9} {:>9} {:>14} {:>12}", + s.allocs, s.reallocs, s.total_bytes, s.peak_bytes + ); + } + } +} diff --git a/star/benches/common/mod.rs b/star/benches/common/mod.rs new file mode 100644 index 0000000..f5e4d84 --- /dev/null +++ b/star/benches/common/mod.rs @@ -0,0 +1,103 @@ +use std::collections::BTreeMap; + +pub const ZSTD_LEVEL: i32 = 3; + +pub struct Workload { + pub name: &'static str, + pub pairs: Vec<(Vec, Vec)>, +} + +impl Workload { + pub fn bytes(&self) -> u64 { + self.pairs + .iter() + .map(|(k, v)| (k.len() + v.len()) as u64) + .sum() + } +} + +/// shaped like stream chunks, plus extremes where the per-pair overhead or the copy dominates +pub fn workloads() -> Vec { + let mut rng = 0x9e37_79b9_7f4a_7c15_u64; + let mut next = move || { + rng ^= rng << 13; + rng ^= rng >> 7; + rng ^= rng << 17; + rng + }; + let mut value = |len: usize| -> Vec { + // small alphabet so zstd has something to compress + (0..len) + .map(|_| b"abcdefghij{}:\"0123"[next() as usize % 18]) + .collect() + }; + let id = |i: usize| (i as u64).to_be_bytes().to_vec(); + let events = (0..20_000) + .map(|i| { + let len = 200 + (i * 7919) % 1800; + (id(i), value(len)) + }) + .collect(); + let tiny = (0..100_000).map(|i| (id(i), value(16))).collect(); + let large = (0..32).map(|i| (id(i), value(1 << 20))).collect(); + vec![ + Workload { + name: "events", + pairs: events, + }, + Workload { + name: "tiny", + pairs: tiny, + }, + Workload { + name: "large", + pairs: large, + }, + ] +} + +pub fn meta() -> BTreeMap<&'static str, &'static str> { + BTreeMap::from([ + ("$type", "systems.gaze.hydrant.streamChunk"), + ("stream", "bench"), + ]) +} + +pub fn write(out: &mut impl std::io::Write, w: &Workload) { + dasl_star::write_header(out, &meta()).unwrap(); + for (key, value) in &w.pairs { + dasl_star::write_pair(out, key, value).unwrap(); + } + dasl_star::write_end(out).unwrap(); +} + +pub fn encode(w: &Workload) -> Vec { + let mut out = Vec::new(); + write(&mut out, w); + out +} + +pub fn encode_zstd(w: &Workload) -> Vec { + zstd::encode_all(encode(w).as_slice(), ZSTD_LEVEL).unwrap() +} + +/// sums every pair so the compiler can't optimize the reading away +pub fn read(r: impl std::io::BufRead) -> u64 { + let mut reader = dasl_star::Reader::new(r).unwrap(); + let mut sum = 0u64; + while let Some((key, value)) = reader.next_pair().unwrap() { + sum = sum.wrapping_add(key.len() as u64 + value.len() as u64); + sum = sum.wrapping_add(u64::from(*value.last().unwrap_or(&0))); + } + sum +} + +pub fn read_owned(r: impl std::io::BufRead) -> u64 { + let mut sum = 0u64; + for pair in dasl_star::Reader::new(r).unwrap() { + let (key, value) = pair.unwrap(); + sum = sum.wrapping_add(key.len() as u64 + value.len() as u64); + sum = sum.wrapping_add(u64::from(*value.last().unwrap_or(&0))); + } + sum +} diff --git a/star/benches/star.rs b/star/benches/star.rs new file mode 100644 index 0000000..7d16f43 --- /dev/null +++ b/star/benches/star.rs @@ -0,0 +1,52 @@ +mod common; + +use std::hint::black_box; +use std::io::{self, BufReader}; +use std::time::Duration; + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; + +fn bench(c: &mut Criterion) { + for w in common::workloads() { + let plain = common::encode(&w); + let compressed = common::encode_zstd(&w); + + let mut group = c.benchmark_group(format!("star/{}", w.name)); + group + .throughput(Throughput::Bytes(w.bytes())) + .sample_size(20) + .warm_up_time(Duration::from_secs(1)) + .measurement_time(Duration::from_secs(4)); + group.bench_function(BenchmarkId::new("write", "vec"), |b| { + let mut out = Vec::with_capacity(plain.len()); + b.iter(|| { + out.clear(); + common::write(&mut out, black_box(&w)); + black_box(out.len()) + }) + }); + group.bench_function(BenchmarkId::new("write", "zstd"), |b| { + b.iter(|| { + let mut out = zstd::Encoder::new(io::sink(), common::ZSTD_LEVEL).unwrap(); + common::write(&mut out, black_box(&w)); + out.finish().unwrap() + }) + }); + group.bench_function(BenchmarkId::new("read", "slice"), |b| { + b.iter(|| common::read(black_box(&plain[..]))) + }); + group.bench_function(BenchmarkId::new("read", "owned"), |b| { + b.iter(|| common::read_owned(black_box(&plain[..]))) + }); + group.bench_function(BenchmarkId::new("read", "zstd"), |b| { + b.iter(|| { + let decoder = zstd::Decoder::new(black_box(&compressed[..])).unwrap(); + common::read(BufReader::new(decoder)) + }) + }); + group.finish(); + } +} + +criterion_group!(benches, bench); +criterion_main!(benches); -- knotmirror 0.0.1