//! criterion benchmarks for the cpu side of repo ingestion: car parsing with and without cid
//! 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 with
// their `#[test]`s stripped and their imports unused
#[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;
use cid::Cid as IpldCid;
use criterion::measurement::WallTime;
use criterion::{
BatchSize, BenchmarkGroup, BenchmarkId, Criterion, SamplingMode, Throughput, criterion_group,
criterion_main,
};
use smol_str::SmolStr;
use car::{CarBlock, CarBlocks};
use mst::{FlatEntry, MstKey};
use sparse_mst::{SparseScanner, sparse_ranges};
/// 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 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<(MstKey, IpldCid)>,
}
impl Repo {
/// 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 differently
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,
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");
mst::AtpCommit::from_cbor(commit)
.expect("commit decodes")
.data
}
/// cars shaped like a `getRecord` proof, for records spread evenly over the largest repo: the
/// commit, every mst node from the root down to the record, and the record. firehose commits
/// carry the same blocks for the records they touch.
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()
})
}
/// the nodes from `root` down to the one holding `key`.
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");
// the subtree left of the first key after `key`, unless `key` is in this node
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);
}
/// flat sampling: the large repo takes tens of milliseconds per iteration, too long for
/// criterion's default of running more iterations in each successive sample
fn group<'a>(c: &'a mut Criterion, name: &str) -> BenchmarkGroup<'a, WallTime> {
let mut group = c.benchmark_group(name);
group.sampling_mode(SamplingMode::Flat);
group
}
/// sha256 plus cid framing on blocks that stay in cache: the floor every cid check sits on,
/// and what drops if the hardware sha256 path is lost.
fn compute_cid(c: &mut Criterion) {
let repo = &repos()[0];
let parsed = repo.parse(false);
let blocks: Vec<&Bytes> = parsed.blocks.map().values().collect();
let mut group = group(c, "compute_cid");
group.throughput(Throughput::Elements(blocks.len() as u64));
group.bench_function(repo.name, |b| {
b.iter(|| {
for bytes in &blocks {
black_box(car::compute_cid(bytes));
}
})
});
group.finish();
}
/// full backfill's `getRepo` parse into the hashed block map.
fn parse_car(c: &mut Criterion) {
let mut group = group(c, "parse_car");
for repo in repos() {
group.throughput(Throughput::Elements(repo.blocks as u64));
for (check, verify_cids) in CHECKS {
group.bench_function(BenchmarkId::new(check, repo.name), |b| {
b.iter_with_large_drop(|| car::parse_car(repo.car.clone(), verify_cids).unwrap())
});
}
}
group.finish();
}
/// firehose `#commit`/`#sync` validation, `getRecord` proofs and the resync buffer replay.
fn parse_commit_car(c: &mut Criterion) {
let proofs = proofs();
let mut group = group(c, "parse_commit_car");
group.throughput(Throughput::Elements(proofs.len() as u64));
for (check, verify_cids) in CHECKS {
group.bench_function(check, |b| {
b.iter(|| {
for proof in proofs {
black_box(car::parse_commit_car(proof.clone(), verify_cids).unwrap());
}
})
});
}
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, 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);
group.throughput(Throughput::Elements(repo.leaves.len() as u64));
group.bench_function(repo.name, |b| {
b.iter_with_large_drop(|| mst::leaves(parsed.blocks.map(), repo.mst_root).unwrap())
});
}
group.finish();
}
/// 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() {
let wanted = repo
.leaves
.iter()
.filter(|(key, _)| ranges.iter().any(|range| range.contains(key)))
.count();
group.throughput(Throughput::Elements(wanted 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();
}
/// 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() {
group.throughput(Throughput::Elements(repo.leaves.len() as u64));
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();
group.bench_function(BenchmarkId::new(check, repo.name), |b| {
b.iter(|| {
for block in &records {
black_box(block.verify().unwrap());
}
})
});
}
}
group.finish();
}
/// how big the pieces of a car fed to [`car::WholeCar`] and [`filtered_car::FilteredCar`] are,
/// about what a getRepo body arrives in
const 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 mut repo = filtered.finish().unwrap();
let scan = repo.scanner.scan(mst_root).unwrap().unwrap();
(repo.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");
for repo in repos() {
group.throughput(Throughput::Elements(repo.blocks as u64));
for (check, verify_cids) in CHECKS {
group.bench_function(BenchmarkId::new(check, repo.name), |b| {
b.iter_with_large_drop(|| {
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)
})
});
}
}
group.finish();
}
criterion_group! {
name = benches;
config = Criterion::default()
.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_decode, mst_walk, mst_leaves,
sparse_scan, sparse_scan_selective, filtered_car, record_check, backfill_cpu
}
criterion_main!(benches);