Something went wrong. Try again.
very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
rust fjall at-protocol atproto indexer
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575//! 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, MstKey};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<(MstKey, 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"); mst::AtpCommit::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 = 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 mut repo = filtered.finish().unwrap(); let scan = repo.scanner.scan(mst_root).unwrap().unwrap(); (repo.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 ); } }}