[star] add cpu and allocation benchmarks
140 unmodified linesExpand all141142143144145146147148149160 unmodified linesExpand all140 unmodified linesExpand all- `nu tests/feature_matrix.nu` - Checks every supported cargo feature combination compiles error- and warning-free (`--only "<name> ..."` for a subset). Run this after touching any `#[cfg(feature = ...)]` gate or adding/removing items in shared modules. The supported matrix lives in the script: `indexer` and `relay` are mutually exclusive modes; `indexer_stream`/`backlinks` require `indexer`; `jetstream` requires `indexer_stream` or `relay`; the bare no-features core must also build.- `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 <git-ref> [filter] | save <name> | compare <name>)` - 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 <name>` before a change and compare with `-- --baseline-lenient <name>` 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.- `nu tests/api_pds_banned.nu` - Tests PDS ban APIs and ban persistence.160 unmodified linesExpand all826 unmodified linesExpand all8278288298308318328338348358368378388398408413575 unmodified linesExpand all826 unmodified linesExpand all[[package]]name = "dasl-star"version = "0.1.0"dependencies = [ "criterion", "ipld-core", "serde", "serde_ipld_dagcbor 0.6.4", "serde_json", "thiserror", "zstd",]
[[package]]name = "data-encoding"3575 unmodified linesExpand all11 unmodified lines1213141516171819202122232425262711 unmodified linesserde_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 = false123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157//! 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<dyn Fn() -> Stats + 'a>;
struct Stats { allocs: u64, reallocs: u64, total_bytes: u64, peak_bytes: u64,}
fn measure<T>(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 ); } }}12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152mod 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);123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103use std::collections::BTreeMap;
pub const ZSTD_LEVEL: i32 = 3;
pub struct Workload { pub name: &'static str, pub pairs: Vec<(Vec<u8>, Vec<u8>)>,}
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 dominatespub fn workloads() -> Vec<Workload> { 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<u8> { // 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<u8> { let mut out = Vec::new(); write(&mut out, w); out}
pub fn encode_zstd(w: &Workload) -> Vec<u8> { zstd::encode_all(encode(w).as_slice(), ZSTD_LEVEL).unwrap()}
/// sums every pair so the compiler can't optimize the reading awaypub 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}