[bench] add allocation and stage cpu benchmarks
100 unmodified lines10110210310410510610710810911011111213 unmodified lines100 unmodified lines[[bench]]name = "repo"harness = false
[[bench]]name = "alloc"harness = false
[[example]]name = "statusphere"required-features = ["indexer_stream"]
13 unmodified lines123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576//! deterministic heap allocation and memory footprint benchmarks for repo ingestion://! car parsing, commit proof parsing, mst node decoding, full mst walks, sparse scans,//! and streaming filtered cars (in-order and reverse out-of-order).//!//! wraps the production allocator from `src/allocator.rs` in a counting `GlobalAlloc` so//! every benchmark reports exact `allocs`, `reallocs`, `total_bytes` (churn), `peak_bytes`//! (high-water mark above baseline), and `retained_bytes` (live bytes held in the result).
#[path = "../src/car.rs"]#[allow(dead_code)]mod car;#[cfg(feature = "indexer")]#[path = "../src/filter.rs"]#[allow(dead_code, unused_imports)]mod filter;#[cfg(feature = "indexer")]#[path = "../src/backfill/filtered_car.rs"]#[allow(dead_code, unused_imports)]mod filtered_car;#[path = "../src/mst.rs"]#[allow(dead_code)]mod mst;#[path = "../src/sparse_mst.rs"]#[allow(dead_code, unused_imports)]mod sparse_mst;
use core::alloc::{GlobalAlloc, Layout};use std::collections::HashSet;use std::hint::black_box;use std::path::PathBuf;#[cfg(feature = "indexer")]use std::sync::Arc;use std::sync::OnceLock;use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
use bytes::Bytes;use cid::Cid as IpldCid;use smol_str::SmolStr;
use car::{CarBlock, CarBlocks};use mst::FlatEntry;use sparse_mst::{SparseScanner, sparse_ranges};
#[cfg(feature = "alloc-mimalloc")]static INNER: mimalloc::MiMalloc = mimalloc::MiMalloc;
#[cfg(feature = "alloc-jemalloc")]static INNER: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
#[cfg(all(feature = "alloc-jemalloc", target_os = "linux"))]#[allow(non_upper_case_globals)]#[unsafe(export_name = "_rjem_malloc_conf")]pub static _rjem_malloc_conf: Option<&'static core::ffi::CStr> = Some(c"background_thread:true,percpu_arena:percpu,dirty_decay_ms:1000,muzzy_decay_ms:0");
#[cfg(feature = "alloc-snmalloc")]static INNER: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
#[cfg(not(any( feature = "alloc-mimalloc", feature = "alloc-jemalloc", feature = "alloc-snmalloc")))]static INNER: std::alloc::System = std::alloc::System;
struct CountingAlloc;
#[global_allocator]static GLOBAL: CountingAlloc = CountingAlloc;
static MEASURING: AtomicBool = AtomicBool::new(false);static ALLOCS: AtomicU64 = AtomicU64::new(0);static DEALLOCS: 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);
unsafe impl GlobalAlloc for CountingAlloc { unsafe fn alloc(&self, layout: Layout) -> *mut u8 { let ptr = unsafe { INNER.alloc(layout) }; if !ptr.is_null() && MEASURING.load(Ordering::Relaxed) { let size = layout.size() as u64; ALLOCS.fetch_add(1, Ordering::Relaxed); TOTAL_BYTES.fetch_add(size, Ordering::Relaxed); let live = LIVE_BYTES.fetch_add(size as i64, Ordering::Relaxed) + size as i64; PEAK_BYTES.fetch_max(live, Ordering::Relaxed); } ptr }
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { let ptr = unsafe { INNER.alloc_zeroed(layout) }; if !ptr.is_null() && MEASURING.load(Ordering::Relaxed) { let size = layout.size() as u64; ALLOCS.fetch_add(1, Ordering::Relaxed); TOTAL_BYTES.fetch_add(size, Ordering::Relaxed); let live = LIVE_BYTES.fetch_add(size as i64, Ordering::Relaxed) + size as i64; PEAK_BYTES.fetch_max(live, Ordering::Relaxed); } ptr }
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { if MEASURING.load(Ordering::Relaxed) { DEALLOCS.fetch_add(1, Ordering::Relaxed); LIVE_BYTES.fetch_sub(layout.size() as i64, Ordering::Relaxed); } unsafe { INNER.dealloc(ptr, layout) }; }
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { let new_ptr = unsafe { INNER.realloc(ptr, layout, new_size) }; if !new_ptr.is_null() && MEASURING.load(Ordering::Relaxed) { REALLOCS.fetch_add(1, Ordering::Relaxed); let old_size = layout.size(); if new_size > old_size { let grow = (new_size - old_size) as u64; TOTAL_BYTES.fetch_add(grow, Ordering::Relaxed); let live = LIVE_BYTES.fetch_add(grow as i64, Ordering::Relaxed) + grow as i64; PEAK_BYTES.fetch_max(live, Ordering::Relaxed); } else { let shrink = (old_size - new_size) as i64; LIVE_BYTES.fetch_sub(shrink, Ordering::Relaxed); } } new_ptr }}
#[derive(Clone, Copy)]struct Stats { allocs: u64, deallocs: u64, reallocs: u64, total_bytes: u64, peak_bytes: u64, retained_bytes: u64,}
fn measure<T>(f: impl FnOnce() -> T) -> Stats { ALLOCS.store(0, Ordering::SeqCst); DEALLOCS.store(0, Ordering::SeqCst); REALLOCS.store(0, Ordering::SeqCst); TOTAL_BYTES.store(0, Ordering::SeqCst); LIVE_BYTES.store(0, Ordering::SeqCst); PEAK_BYTES.store(0, Ordering::SeqCst); MEASURING.store(true, Ordering::SeqCst);
let out = black_box(f());
MEASURING.store(false, Ordering::SeqCst); let retained_bytes = LIVE_BYTES.load(Ordering::SeqCst).max(0) as u64; let peak_bytes = PEAK_BYTES.load(Ordering::SeqCst).max(0) as u64; let stats = Stats { allocs: ALLOCS.load(Ordering::SeqCst), deallocs: DEALLOCS.load(Ordering::SeqCst), reallocs: REALLOCS.load(Ordering::SeqCst), total_bytes: TOTAL_BYTES.load(Ordering::SeqCst), peak_bytes, retained_bytes, }; drop(out); stats}
const REPOS: [&str; 3] = ["atproto.com", "jay.bsky.team", "pfrazee.com"];const PROOFS: usize = 2000;const DENSE_COLLECTION: &str = "app.bsky.feed.post";const SELECTIVE_COLLECTION: &str = "sh.tangled.*";const CHECKS: [(&str, bool); 2] = [("verify", true), ("trust", false)];const BODY_CHUNK: usize = 16 * 1024;
struct Repo { name: &'static str, car: Bytes, #[cfg(feature = "indexer")] reverse_car: OnceLock<Bytes>, mst_root: IpldCid, mst_nodes: Vec<(IpldCid, Bytes)>, leaves: Vec<(SmolStr, IpldCid)>,}
impl Repo { fn parse(&self, verify_cids: bool) -> car::ParsedCar<CarBlocks> { car::parse_car(self.car.clone(), verify_cids).expect("bench repo parses") }
/// the same car with record blocks first, mst nodes in bottom-up post-order (children before /// parents), and the commit block last, exercising out-of-order streaming and node parking. #[cfg(feature = "indexer")] #[cfg(feature = "indexer")] fn reverse_car(&self) -> &Bytes { self.reverse_car.get_or_init(|| { let parsed = self.parse(false); let blocks = parsed.blocks.map(); let mst_cids: HashSet<IpldCid> = self.mst_nodes.iter().map(|(cid, _)| *cid).collect(); let mut records: Vec<(IpldCid, &Bytes)> = blocks .iter() .filter(|(cid, _)| **cid != parsed.root && !mst_cids.contains(cid)) .map(|(cid, bytes)| (*cid, bytes)) .collect(); records.sort_unstable_by_key(|(cid, _)| *cid); let sections = records .into_iter() .chain(self.mst_nodes.iter().rev().map(|(cid, bytes)| (*cid, bytes))) .chain(std::iter::once((parsed.root, &blocks[&parsed.root]))); write_car(parsed.root, sections) }) }}
fn data_dir() -> PathBuf { std::env::var_os("HYDRANT_BENCH_DATA") .map(PathBuf::from) .unwrap_or_else(|| PathBuf::from("target/bench-data"))}
fn repos() -> &'static [Repo] { static LOADED: OnceLock<Vec<Repo>> = OnceLock::new(); LOADED.get_or_init(|| REPOS.into_iter().map(load).collect())}
fn load(name: &'static str) -> Repo { let path = data_dir().join(format!("{name}.car")); let car = std::fs::read(&path).unwrap_or_else(|e| { panic!( "{}: {e}; fetch the bench repos with `nu tests/bench.nu fetch`", path.display() ) }); let car = Bytes::from(car); let parsed = car::parse_car(car.clone(), true).expect("bench repo is a valid car"); let mst_root = commit_data(&parsed); let mst_nodes = collect_mst_nodes(parsed.blocks.map(), mst_root); let leaves = mst::leaves(parsed.blocks.map(), mst_root).expect("bench repo mst walks"); Repo { name, car, #[cfg(feature = "indexer")] reverse_car: OnceLock::new(), mst_root, mst_nodes, leaves, }}
fn collect_mst_nodes(blocks: &CarBlocks, root: IpldCid) -> Vec<(IpldCid, Bytes)> { let mut out = Vec::new(); let mut visited = HashSet::new(); let mut stack = vec![root]; while let Some(cid) = stack.pop() { if !visited.insert(cid) { continue; } let bytes = blocks.get(&cid).expect("mst node in car").clone(); let entries = mst::decode_node(&bytes).expect("bench repo mst node decodes"); let start = stack.len(); for entry in entries { if let FlatEntry::Tree { cid } = entry { stack.push(cid); } } stack[start..].reverse(); out.push((cid, bytes)); } out}
fn commit_data(parsed: &car::ParsedCar<CarBlocks>) -> IpldCid { let commit = parsed.blocks.get(&parsed.root).expect("car has its commit"); jacquard_repo::commit::Commit::<jacquard_common::DefaultStr>::from_cbor(commit) .expect("commit decodes") .data}
fn proofs() -> &'static [Bytes] { static CUT: OnceLock<Vec<Bytes>> = OnceLock::new(); CUT.get_or_init(|| { let repo = repos().last().expect("bench repos are listed"); let parsed = repo.parse(false); let blocks = parsed.blocks.map(); let step = (repo.leaves.len() / PROOFS).max(1); repo.leaves .iter() .step_by(step) .take(PROOFS) .map(|(key, record)| { let cids = std::iter::once(parsed.root) .chain(mst_path(blocks, repo.mst_root, key)) .chain([*record]); let proof = write_car(parsed.root, cids.map(|cid| (cid, &blocks[&cid]))); car::parse_commit_car(proof.clone(), true).expect("proof is a valid car"); proof }) .collect() })}
fn mst_path(blocks: &CarBlocks, root: IpldCid, key: &str) -> Vec<IpldCid> { let mut path = vec![root]; loop { let node = path.last().expect("path starts at the root"); let entries = mst::decode_node(&blocks[node]).expect("bench repo nodes decode"); let mut subtree = None; for entry in &entries { match entry { FlatEntry::Leaf { key: leaf, .. } if leaf == key => return path, FlatEntry::Leaf { key: leaf, .. } if leaf.as_str() > key => break, FlatEntry::Leaf { .. } => subtree = None, FlatEntry::Tree { cid } => subtree = Some(*cid), } } path.push(subtree.expect("key is in the tree")); }}
fn write_car<'a>(root: IpldCid, sections: impl Iterator<Item = (IpldCid, &'a Bytes)>) -> Bytes { let header = iroh_car::CarHeader::new_v1(vec![root]) .encode() .expect("header encodes"); let mut out = Vec::new(); put_uvarint(&mut out, header.len()); out.extend_from_slice(&header); for (cid, bytes) in sections { let cid = cid.to_bytes(); put_uvarint(&mut out, cid.len() + bytes.len()); out.extend_from_slice(&cid); out.extend_from_slice(bytes); } out.into()}
fn put_uvarint(out: &mut Vec<u8>, mut value: usize) { while value >= 0x80 { out.push(value as u8 | 0x80); value >>= 7; } out.push(value as u8);}
#[cfg(feature = "indexer")]fn filter_for(pattern: &str) -> Arc<filter::FilterConfig> { let mut cfg = filter::FilterConfig::new(filter::FilterMode::Filter); cfg.collections.push(SmolStr::new(pattern)); Arc::new(cfg)}
#[cfg(feature = "indexer")]fn run_filtered_car( car_bytes: &Bytes, mst_root: IpldCid, filter: &Arc<filter::FilterConfig>,) -> ( car::Blocks<car::CommitBlocks>, sparse_mst::SparseScanOutput,) { let mut filtered = filtered_car::FilteredCar::new(filter.clone(), false).unwrap(); for chunk in car_bytes.chunks(BODY_CHUNK) { filtered.push(chunk).unwrap(); } let repo = filtered.finish().unwrap(); let mut scanner = SparseScanner::new(repo.ranges, repo.blocks); scanner.rule_out(repo.ruled_out); let scan = scanner.scan(mst_root).unwrap().unwrap(); (scanner.take_blocks(), scan)}
fn record( rows: &mut Vec<(String, Stats)>, filter: Option<&str>, name: impl Into<String>, run: impl FnOnce() -> Stats,) { let name = name.into(); if filter.is_some_and(|pat| !name.contains(pat)) { return; } rows.push((name, run()));}
fn main() { let mut json = false; let mut filter: Option<String> = None; for arg in std::env::args().skip(1) { match arg.as_str() { "--bench" => {} "--json" => json = true, other if !other.starts_with('-') => filter = Some(other.to_string()), _ => {} } } let filter = filter.as_deref(); let mut rows = Vec::new();
for (check, verify_cids) in CHECKS { for repo in repos() { let car = repo.car.clone(); record( &mut rows, filter, format!("parse_car/{check}/{}", repo.name), || measure(|| car::parse_car(car, verify_cids).unwrap()), ); } }
let proofs = proofs(); for (check, verify_cids) in CHECKS { record(&mut rows, filter, format!("parse_commit_car/{check}"), || { measure(|| { for proof in proofs { black_box(car::parse_commit_car(proof.clone(), verify_cids).unwrap()); } }) }); }
for repo in repos() { record(&mut rows, filter, format!("mst_decode/{}", repo.name), || { measure(|| { for (_, bytes) in &repo.mst_nodes { black_box(mst::decode_node(bytes).unwrap()); } }) }); }
for repo in repos() { let parsed = repo.parse(false); record(&mut rows, filter, format!("mst_walk/{}", repo.name), || { measure(|| { let mut count = 0usize; for leaf in mst::walk(parsed.blocks.map(), repo.mst_root) { black_box(leaf.unwrap()); count += 1; } count }) }); }
for repo in repos() { let parsed = repo.parse(false); record(&mut rows, filter, format!("mst_leaves/{}", repo.name), || { measure(|| mst::leaves(parsed.blocks.map(), repo.mst_root).unwrap()) }); }
for (label, pattern) in [ ("dense", DENSE_COLLECTION), ("selective", SELECTIVE_COLLECTION), ] { let ranges = sparse_ranges(&[SmolStr::new(pattern)]); for repo in repos() { let blocks = car::parse_car_blocks(repo.car.clone(), false).unwrap(); let ranges = ranges.clone(); record( &mut rows, filter, format!("sparse_scan/{label}/{}", repo.name), move || { measure(move || { let mut scanner = SparseScanner::new(ranges, blocks); let scan = scanner.scan(repo.mst_root).unwrap().unwrap(); (scanner, scan) }) }, ); } }
#[cfg(feature = "indexer")] { let selective = filter_for(SELECTIVE_COLLECTION); let dense = filter_for(DENSE_COLLECTION); for (label, cfg) in [("selective", &selective), ("dense", &dense)] { for repo in repos() { record( &mut rows, filter, format!("filtered_car/{label}/{}", repo.name), || measure(|| run_filtered_car(&repo.car, repo.mst_root, cfg)), ); } } for repo in repos() { let reversed = repo.reverse_car(); record( &mut rows, filter, format!("filtered_car/reverse/{}", repo.name), || measure(|| run_filtered_car(reversed, repo.mst_root, &selective)), ); } }
for repo in repos() { for (check, verify_cids) in CHECKS { let parsed = repo.parse(verify_cids); let records: Vec<CarBlock> = repo .leaves .iter() .map(|(_, cid)| parsed.blocks.block(cid).expect("record is in the car")) .collect(); record( &mut rows, filter, format!("record_check/{check}/{}", repo.name), || { measure(|| { for block in &records { black_box(block.verify().unwrap()); } }) }, ); } }
for (check, verify_cids) in CHECKS { for repo in repos() { record( &mut rows, filter, format!("backfill_cpu/{check}/{}", repo.name), || { measure(|| { let mut whole = car::WholeCar::new(verify_cids); whole.reserve(repo.car.len()); for chunk in repo.car.chunks(BODY_CHUNK) { whole.push(chunk).unwrap(); } let parsed = whole.finish().unwrap(); let leaves = mst::leaves(parsed.blocks.map(), commit_data(&parsed)).unwrap(); for (_, cid) in &leaves { black_box(parsed.blocks.block(cid).unwrap().verify().unwrap()); } (parsed, leaves) }) }, ); } }
if json { let items: Vec<serde_json::Value> = rows .into_iter() .map(|(bench, s)| { serde_json::json!({ "bench": bench, "allocs": s.allocs, "deallocs": s.deallocs, "reallocs": s.reallocs, "total_bytes": s.total_bytes, "peak_bytes": s.peak_bytes, "retained_bytes": s.retained_bytes, }) }) .collect(); println!("{}", serde_json::to_string_pretty(&items).unwrap()); } else { println!( "{:<38} {:>10} {:>10} {:>14} {:>14} {:>14}", "bench", "allocs", "reallocs", "total_bytes", "peak_bytes", "retained_bytes" ); println!("{}", "-".repeat(104)); for (bench, s) in rows { println!( "{:<38} {:>10} {:>10} {:>14} {:>14} {:>14}", bench, s.allocs, s.reallocs, s.total_bytes, s.peak_bytes, s.retained_bytes ); } }}12342345678910111213141516171819202122231 unmodified line2526272829303132333435363712 unmodified lines505152534254555657585960616263646566676869707172731 unmodified line7576777879808182838485868788899091929394959697989910010110210310410518 unmodified lines124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167126 unmodified linesExpand all2942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293302313313323333343354 unmodified lines34034134234324434434534634734817 unmodified lines36636736836937037137237337437537637737837938038138238338438538638738838939039139239339439516 unmodified lines41241341441529429541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547626 unmodified lines503504505506332333507508509510//! criterion benchmarks for the cpu side of repo ingestion: car parsing with and without cid//! checks, the full-repo mst walk, the sparse scan, and the check before a record is stored.//! they run on real repos that `nu tests/bench.nu fetch` downloads once; that script also//! saves baselines and fails on regressions against them.//! checks, mst node decoding, the full-repo mst walk (streaming and collected), sparse scans//! (dense and selective), streaming filtered cars (in-order and reverse out-of-order), and the//! check before a record is stored. they run on real repos that `nu tests/bench.nu fetch`//! downloads once; that script also saves baselines and fails on regressions against them.
#[path = "../src/allocator.rs"]mod allocator;#[path = "../src/car.rs"]#[allow(dead_code)]mod car;#[cfg(feature = "indexer")]#[path = "../src/filter.rs"]#[allow(dead_code, unused_imports)]mod filter;#[cfg(feature = "indexer")]#[path = "../src/backfill/filtered_car.rs"]#[allow(dead_code, unused_imports)]mod filtered_car;#[path = "../src/mst.rs"]#[allow(dead_code)]mod mst;// cargo builds benches with `--cfg test` but no test harness, so test modules compile with1 unmodified line#[path = "../src/sparse_mst.rs"]#[allow(dead_code, unused_imports)]mod sparse_mst;
use std::collections::HashSet;use std::hint::black_box;use std::path::PathBuf;#[cfg(feature = "indexer")]use std::sync::Arc;use std::sync::OnceLock;use std::time::Duration;
use bytes::Bytes;12 unmodified lines/// one repo that fits in any cache, one that fits a big l3, and one that fits none.const REPOS: [&str; 3] = ["atproto.com", "jay.bsky.team", "pfrazee.com"];/// `getRecord`-shaped cars cut from the largest repo, standing in for firehose commits.const PROOFS: usize = 2000;/// the collection the sparse scan looks for./// the dense collection the sparse scan looks for.const SPARSE_COLLECTION: &str = "app.bsky.feed.post";/// the selective wildcard pattern for subtree-pruning and filtered-car benchmarks.const SELECTIVE_COLLECTION: &str = "sh.tangled.*";/// whether the parse checks cids (`verify_cids` on) or not.const CHECKS: [(&str, bool); 2] = [("verify", true), ("trust", false)];
struct Repo { name: &'static str, car: Bytes, #[cfg(feature = "indexer")] reverse_car: OnceLock<Bytes>, blocks: usize, mst_root: IpldCid, /// every reachable mst node in pre-order mst_nodes: Vec<(IpldCid, Bytes)>, /// every record in the repo, in key order leaves: Vec<(SmolStr, IpldCid)>,}
1 unmodified line /// a parse for setting a benchmark up, outside what it measures fn parse(&self, verify_cids: bool) -> car::ParsedCar<CarBlocks> { car::parse_car(self.car.clone(), verify_cids).expect("bench repo parses") }
/// the same car with record blocks first, mst nodes in bottom-up post-order (children before /// parents), and the commit block last, exercising out-of-order streaming and node parking. #[cfg(feature = "indexer")] #[cfg(feature = "indexer")] fn reverse_car(&self) -> &Bytes { self.reverse_car.get_or_init(|| { let parsed = self.parse(false); let blocks = parsed.blocks.map(); let mst_cids: HashSet<IpldCid> = self.mst_nodes.iter().map(|(cid, _)| *cid).collect(); let mut records: Vec<(IpldCid, &Bytes)> = blocks .iter() .filter(|(cid, _)| **cid != parsed.root && !mst_cids.contains(cid)) .map(|(cid, bytes)| (*cid, bytes)) .collect(); records.sort_unstable_by_key(|(cid, _)| *cid); let sections = records .into_iter() .chain(self.mst_nodes.iter().rev().map(|(cid, bytes)| (*cid, bytes))) .chain(std::iter::once((parsed.root, &blocks[&parsed.root]))); write_car(parsed.root, sections) }) }}
/// relative, because cargo runs benches from the package root: an absolute path baked in here/// would make builds of the same code from two checkouts lay their code out differently18 unmodified lines }); let car = Bytes::from(car); let parsed = car::parse_car(car.clone(), true).expect("bench repo is a valid car"); let mst_root = commit_data(&parsed); let mst_nodes = collect_mst_nodes(parsed.blocks.map(), mst_root); let leaves = mst::leaves(parsed.blocks.map(), mst_root).expect("bench repo mst walks"); Repo { name, blocks: parsed.blocks.map().len(), car, #[cfg(feature = "indexer")] reverse_car: OnceLock::new(), mst_root, mst_nodes, leaves, }}
fn collect_mst_nodes(blocks: &CarBlocks, root: IpldCid) -> Vec<(IpldCid, Bytes)> { let mut out = Vec::new(); let mut visited = HashSet::new(); let mut stack = vec![root]; while let Some(cid) = stack.pop() { if !visited.insert(cid) { continue; } let bytes = blocks.get(&cid).expect("mst node in car").clone(); let entries = mst::decode_node(&bytes).expect("bench repo mst node decodes"); let start = stack.len(); for entry in entries { if let FlatEntry::Tree { cid } = entry { stack.push(cid); } } stack[start..].reverse(); out.push((cid, bytes)); } out}
fn commit_data(parsed: &car::ParsedCar<CarBlocks>) -> IpldCid { let commit = parsed.blocks.get(&parsed.root).expect("car has its commit"); jacquard_repo::commit::Commit::<jacquard_common::DefaultStr>::from_cbor(commit) .expect("commit decodes")126 unmodified linesExpand all } group.finish();}
/// decoding every mst node in the repo without walking the tree.fn mst_decode(c: &mut Criterion) { let mut group = group(c, "mst_decode"); for repo in repos() { group.throughput(Throughput::Elements(repo.mst_nodes.len() as u64)); group.bench_function(repo.name, |b| { b.iter(|| { for (_, bytes) in &repo.mst_nodes { black_box(mst::decode_node(bytes).unwrap()); } }) }); } group.finish();}
/// streaming walk of every mst node to every leaf without collecting a Vec.fn mst_walk(c: &mut Criterion) { let mut group = group(c, "mst_walk"); for repo in repos() { let parsed = repo.parse(false); group.throughput(Throughput::Elements(repo.leaves.len() as u64)); group.bench_function(repo.name, |b| { b.iter(|| { for leaf in mst::walk(parsed.blocks.map(), repo.mst_root) { black_box(leaf.unwrap()); } }) }); } group.finish();}
/// full backfill's walk of every mst node to every record./// full backfill's walk of every mst node to every record, collected into a Vec.fn mst_leaves(c: &mut Criterion) { let mut group = group(c, "mst_leaves"); for repo in repos() { let parsed = repo.parse(false);4 unmodified lines } group.finish();}
/// sparse backfill's scan for one collection, with every block at hand so no round trips./// sparse backfill's scan for one dense collection, with every block at hand so no round trips.fn sparse_scan(c: &mut Criterion) { let ranges = sparse_ranges(&[SmolStr::new(SPARSE_COLLECTION)]); let mut group = group(c, "sparse_scan"); for repo in repos() {17 unmodified lines } group.finish();}
/// sparse backfill's scan for a selective wildcard collection (`sh.tangled.*`), where subtree/// pruning dominates.fn sparse_scan_selective(c: &mut Criterion) { let ranges = sparse_ranges(&[SmolStr::new(SELECTIVE_COLLECTION)]); let mut group = group(c, "sparse_scan_selective"); for repo in repos() { group.throughput(Throughput::Elements(repo.mst_nodes.len() as u64)); group.bench_function(repo.name, |b| { b.iter_batched( || car::parse_car_blocks(repo.car.clone(), false).unwrap(), |blocks| { let mut scanner = SparseScanner::new(ranges.clone(), blocks); let scan = scanner.scan(repo.mst_root).unwrap().unwrap(); (scanner, scan) }, BatchSize::PerIteration, ) }); } group.finish();}
/// the check before each record is stored: free after a verifying parse, a hash otherwise.fn record_check(c: &mut Criterion) { let mut group = group(c, "record_check"); for repo in repos() {16 unmodified lines } group.finish();}
/// how big the pieces of a car fed to [`car::WholeCar`] are, about what a getRepo body arrives/// in/// how big the pieces of a car fed to [`car::WholeCar`] and [`filtered_car::FilteredCar`] are,/// about what a getRepo body arrives inconst BODY_CHUNK: usize = 16 * 1024;
#[cfg(feature = "indexer")]fn filter_for(pattern: &str) -> Arc<filter::FilterConfig> { let mut cfg = filter::FilterConfig::new(filter::FilterMode::Filter); cfg.collections.push(SmolStr::new(pattern)); Arc::new(cfg)}
#[cfg(feature = "indexer")]fn run_filtered_car( car_bytes: &Bytes, mst_root: IpldCid, filter: &Arc<filter::FilterConfig>,) -> ( car::Blocks<car::CommitBlocks>, sparse_mst::SparseScanOutput,) { let mut filtered = filtered_car::FilteredCar::new(filter.clone(), false).unwrap(); for chunk in car_bytes.chunks(BODY_CHUNK) { filtered.push(chunk).unwrap(); } let repo = filtered.finish().unwrap(); let mut scanner = SparseScanner::new(repo.ranges, repo.blocks); scanner.rule_out(repo.ruled_out); let scan = scanner.scan(mst_root).unwrap().unwrap(); (scanner.take_blocks(), scan)}
/// streaming filtered backfill (`FilteredCar` + `SparseScanner`) across selective, dense, and/// reverse out-of-order cars.#[cfg(feature = "indexer")]fn filtered_car(c: &mut Criterion) { let selective = filter_for(SELECTIVE_COLLECTION); let dense = filter_for(SPARSE_COLLECTION); let mut group = group(c, "filtered_car"); for repo in repos() { group.throughput(Throughput::Elements(repo.blocks as u64)); group.bench_function(BenchmarkId::new("selective", repo.name), |b| { b.iter_with_large_drop(|| run_filtered_car(&repo.car, repo.mst_root, &selective)) }); group.bench_function(BenchmarkId::new("dense", repo.name), |b| { b.iter_with_large_drop(|| run_filtered_car(&repo.car, repo.mst_root, &dense)) }); let reversed = repo.reverse_car(); group.bench_function(BenchmarkId::new("reverse", repo.name), |b| { b.iter_with_large_drop(|| run_filtered_car(reversed, repo.mst_root, &selective)) }); } group.finish();}
#[cfg(not(feature = "indexer"))]fn filtered_car(_c: &mut Criterion) {}
/// a full backfill without the database: stream the car in, walk, and check every record for/// storage. verify against trust here is the whole cpu cost of `verify_cids` on a backfill.fn backfill_cpu(c: &mut Criterion) { let mut group = group(c, "backfill_cpu");26 unmodified lines .sample_size(20) .warm_up_time(Duration::from_secs(1)) .measurement_time(Duration::from_secs(4)) .without_plots(); targets = compute_cid, parse_car, parse_commit_car, mst_leaves, sparse_scan, record_check, backfill_cpu targets = compute_cid, parse_car, parse_commit_car, mst_decode, mst_walk, mst_leaves, sparse_scan, sparse_scan_selective, filtered_car, record_check, backfill_cpu}criterion_main!(benches);3 unmodified lines4567891011121375 unmodified lines89909192919293949596959798999810010110010210310410510694 unmodified lines20120220320420320520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128292 unmodified lines3 unmodified lines## usage:# nu tests/bench.nu fetch # download the bench repos (once, then kept)# nu tests/bench.nu compare-ref <git-ref> [filter] # this tree against <git-ref>: the gate# nu tests/bench.nu alloc [filter] # exact heap allocations & memory footprint# nu tests/bench.nu alloc-compare <git-ref> [filter]# compare heap allocations & memory vs <git-ref># nu tests/bench.nu save <name> [criterion args] # run the benches, save baseline <name># nu tests/bench.nu compare <name> [criterion args] # run again and compare against <name>## a benchmark looks slower when even the low end of criterion's 95% confidence interval for its75 unmodified lines with-env (bench-env) { ^cargo bench --bench repo -- ...$args }}
# builds the bench binary of the checkout at `dir` into `target` and returns its pathdef bench-binary [dir: string, target: string] { print $"building the benches in ($dir)..."def bench-binary [dir: string, target: string, bench: string = "repo"] { print $"building ($bench) bench in ($dir)..." cd $dir let built = (with-env (bench-env | merge {CARGO_TARGET_DIR: $target}) { ^cargo bench --bench repo --no-run | complete ^cargo bench --bench $bench --no-run | complete }) if $built.exit_code != 0 { error make {msg: $"building the benches in ($dir) failed:\n($built.stderr)"} error make {msg: $"building ($bench) bench in ($dir) failed:\n($built.stderr)"} } let exe = ($built.stderr | parse --regex 'Executable benches/repo\.rs \((?<path>[^)]+)\)') let exe = ($built.stderr | parse --regex ('Executable benches/' + $bench + '\.rs \((?<path>[^)]+)\)')) $dir | path join $exe.0.path | path expand}
def bench-ids [binary: string, filter?: string] {94 unmodified lines read-change ($change_dir | path join "estimates.json")}
def main [] { print "usage: nu tests/bench.nu (fetch | compare-ref <git-ref> [filter] | save <name> | compare <name>)" print "usage: nu tests/bench.nu (fetch | compare-ref <git-ref> [filter] | alloc [filter] | alloc-compare <git-ref> [filter] | save <name> | compare <name>)"}
def delta-str [before: int, after: int] { if $before == $after { return ($"($after) " + "(0%)") } if $before == 0 { return $"($before) -> ($after)" } let pct = (($after - $before) / $before * 100 | math round --precision 1) let sign = if $pct > 0 { "+" } else { "" } $"($before) -> ($after) ("(")($sign)($pct)%(")")"}
def run-alloc-json [binary: string, filter?: string] { let args = if $filter == null { ["--json"] } else { ["--json", $filter] } let ran = (with-env (bench-env) { ^$binary ...$args | complete }) if $ran.exit_code != 0 { error make {msg: $"($binary) failed:\n($ran.stdout)($ran.stderr)"} } $ran.stdout | from json}
# runs the deterministic allocation and memory footprint benchmark suitedef "main alloc" [filter?: string] { inputs | ignore let bin = (bench-binary (repo-root) (target-dir) "alloc") let rows = (run-alloc-json $bin $filter | each {|r| { bench: $r.bench allocs: $r.allocs reallocs: $r.reallocs total_bytes: $r.total_bytes peak_bytes: $r.peak_bytes retained_bytes: $r.retained_bytes } }) print ($rows | table --index false --width 160)}
# compares exact heap allocations and memory footprint against `ref`def "main alloc-compare" [ref: string, filter?: string] { inputs | ignore let scratch = (mktemp -d -t hydrant_alloc_bench.XXXXXX) let tree = ($scratch | path join "tree") ^git -c core.hooksPath=/dev/null worktree add --detach $tree $ref let built = try { { binary: (bench-binary $tree (target-dir | path join "bench-ref") "alloc") } } catch {|err| { failure: $err.msg } } ^git worktree remove --force $tree rm --recursive --force $scratch if "failure" in $built { error make {msg: $"building alloc bench at ($ref) failed: ($built.failure)"} } let ref_rows = (run-alloc-json $built.binary $filter) let now_bin = (bench-binary (repo-root) (target-dir) "alloc") let now_rows = (run-alloc-json $now_bin $filter) let table = ($now_rows | each {|now| let prev = ($ref_rows | where bench == $now.bench | first) { bench: $now.bench allocs: (delta-str $prev.allocs $now.allocs) reallocs: (delta-str $prev.reallocs $now.reallocs) total_bytes: (delta-str $prev.total_bytes $now.total_bytes) peak_bytes: (delta-str $prev.peak_bytes $now.peak_bytes) retained_bytes: (delta-str $prev.retained_bytes $now.retained_bytes) } }) print ($table | table --index false --width 160)}
# downloads the bench repos. existing files are kept: a repo grows as its owner posts, so a# fresh download is a different benchmark and every baseline saved on the old one goes stale.def "main fetch" [] { let dir = (data-dir)92 unmodified lines