znippy-plugin-git 0.1.1

//! What the fall-through costs, and what a rebuild costs — the two numbers the
//! [`RebuildTriggers`](znippy_plugin_git::read_stack::RebuildTriggers) defaults
//! are set from.
//!
//! ```text
//! CARGO_TARGET_DIR=/home/rickard/scratch/cargo/znippy-plugin-git-readstack \
//!   cargo run --release --no-default-features --example read_stack_bench
//! ```
//!
//! Arguments (`key=value`): `sizes=100000,1000000`, `queries=100000`, `runs=3`,
//! `oid=20`, `tail_pct=10`, `force`.
//!
//! ## What is measured, and why in this shape
//!
//! * **Batch, not serial.** `have` negotiation sends up to 1000 oids at a time;
//!   a single lookup is the rare case. Batch 1000 throughout.
//! * **Three arms per cell.** `OneTableFourColumns` alone is the floor — the
//!   stack cannot be faster than its own projection. The **absorbed** stack is
//!   the same rows with the redb tail present but never consulted for a hit, so
//!   the gap between those two is what the stack costs when it is up to date.
//!   The **split** stack has `tail_pct`% of its rows outside the projection, so
//!   the gap between absorbed and split is what a fall-through costs.
//! * **Hit/miss mixes, and the counters printed with them.** A `have`
//!   negotiation is mostly misses, and a miss costs differently from a hit — in
//!   this stack it costs a redb transaction that finds nothing. The
//!   projection-hit / tail-hit / absent counts are read back off the stack and
//!   printed, so the timing line and the routing it describes are the same run.
//! * **`/proc/loadavg` with every figure.** oden is shared.
//!
//! Single-threaded on purpose. (No rayon here or anywhere — LAW 3.)

use std::time::Instant;

use znippy_plugin_git::index_layout::{
    IndexEntry, ObjectIndex, OneTableFourColumns, Rng, synthetic_entries,
};
use znippy_plugin_git::read_stack::{ObjectReadStack, RebuildTriggers, StackStats};

type Stack = ObjectReadStack<OneTableFourColumns>;

const BATCH: usize = 1000;

fn loadavg() -> (f64, String) {
    let s = std::fs::read_to_string("/proc/loadavg").unwrap_or_default();
    let one = s
        .split_whitespace()
        .next()
        .and_then(|x| x.parse::<f64>().ok())
        .unwrap_or(f64::NAN);
    (one, s.trim().to_string())
}

/// Median / min / max. The spread is the noise band and nothing smaller than it
/// is claimed.
#[derive(Clone, Copy)]
struct Stat {
    med: f64,
    lo: f64,
    hi: f64,
}

impl Stat {
    fn of(mut v: Vec<f64>) -> Self {
        v.sort_by(|a, b| a.partial_cmp(b).unwrap());
        Stat {
            med: v[v.len() / 2],
            lo: v[0],
            hi: v[v.len() - 1],
        }
    }
    fn band(&self) -> f64 {
        if self.med == 0.0 {
            0.0
        } else {
            (self.hi - self.lo) / self.med
        }
    }
}

/// `total` queries, `hit_pct`% of them present, shuffled so hits and misses do
/// not arrive in runs.
fn queries(
    present: &[IndexEntry],
    absent: &[IndexEntry],
    total: usize,
    hit_pct: usize,
    seed: u64,
) -> (Vec<Vec<u8>>, usize) {
    let mut rng = Rng(seed);
    let mut owned = Vec::with_capacity(total);
    let mut hits = 0usize;
    for _ in 0..total {
        if (rng.next_u64() % 100) < hit_pct as u64 {
            owned.push(
                present[(rng.next_u64() as usize) % present.len()]
                    .oid
                    .clone(),
            );
            hits += 1;
        } else {
            owned.push(absent[(rng.next_u64() as usize) % absent.len()].oid.clone());
        }
    }
    (owned, hits)
}

/// ns per lookup through the batch path, plus a checksum the optimiser cannot
/// delete and the observed hit count.
fn time_batch(idx: &dyn ObjectIndex, refs: &[&[u8]], runs: usize) -> (Stat, u64, usize) {
    let mut samples = Vec::with_capacity(runs);
    let mut checksum = 0u64;
    let mut hits = 0usize;
    for _ in 0..runs {
        let t = Instant::now();
        let mut acc = 0u64;
        let mut h = 0usize;
        for chunk in refs.chunks(BATCH) {
            for r in idx.lookup_batch(chunk).iter().flatten() {
                h += 1;
                acc = acc
                    .wrapping_add(r.offset)
                    .wrapping_add(r.len)
                    .wrapping_add(r.uncompressed_size);
            }
        }
        samples.push(t.elapsed().as_nanos() as f64 / refs.len() as f64);
        checksum = acc;
        hits = h;
    }
    (Stat::of(samples), checksum, hits)
}

fn delta(a: StackStats, b: StackStats) -> (u64, u64) {
    (b.tail_hits - a.tail_hits, b.absent - a.absent)
}

fn main() {
    // S-070: agent sandboxes set PR_SET_THP_DISABLE and children inherit it, so
    // every MADV_HUGEPAGE this bench's trees issue is a silent no-op and every
    // number here was taken THP-less without saying so. Clearing the flag is a
    // per-binary decision (the library must never flip process state); this
    // binary wants real numbers, and prints which kind it got.
    let thp = znippy_zoomies::stree::thp_enable_for_process();
    eprintln!("thp_enabled={}", thp);
    let mut sizes: Vec<usize> = vec![100_000, 1_000_000];
    let mut total_queries = 100_000usize;
    let mut runs = 3usize;
    let mut oid_len = 20usize;
    let mut tail_pct = 10usize;
    let mut force = false;
    for a in std::env::args().skip(1) {
        let (k, v) = a.split_once('=').unwrap_or((a.as_str(), ""));
        match k {
            "sizes" => sizes = v.split(',').map(|x| x.parse().unwrap()).collect(),
            "queries" => total_queries = v.parse().unwrap(),
            "runs" => runs = v.parse().unwrap(),
            "oid" => oid_len = v.parse().unwrap(),
            "tail_pct" => tail_pct = v.parse().unwrap(),
            "force" => force = true,
            other => panic!("unknown argument `{other}`"),
        }
    }

    let (load1, load) = loadavg();
    println!("# ObjectReadStack — what the redb fall-through costs, and what a rebuild costs");
    println!();
    println!("host loadavg at start: `{load}`  (1-min {load1:.2})");
    println!(
        "oid {oid_len} B · batch {BATCH} · {total_queries} lookups per cell · {runs} runs per \
         cell · split arm keeps {tail_pct}% of rows out of the projection"
    );
    if load1 > 4.0 && !force {
        println!();
        println!("**REFUSING TO MEASURE**: 1-minute load is {load1:.2}. oden is shared.");
        std::process::exit(2);
    }

    let mut bands: Vec<f64> = Vec::new();

    for &n in &sizes {
        let present = synthetic_entries(n, oid_len, 0x5EED_0001 ^ n as u64);
        let absent = synthetic_entries(n.clamp(1000, 200_000), oid_len, 0xDEAD_0002 ^ n as u64);
        let split_at = n - n * tail_pct / 100;

        println!();
        println!("## {n} objects");
        println!();

        // ── build costs ──
        let t = Instant::now();
        let plain = OneTableFourColumns::build(&present).expect("arrow arm");
        let arrow_build_ms = t.elapsed().as_secs_f64() * 1e3;

        let stack = Stack::in_memory(RebuildTriggers::manual()).expect("stack");
        let t = Instant::now();
        stack.append(&present).expect("ingest");
        let ingest_ms = t.elapsed().as_secs_f64() * 1e3;
        let t = Instant::now();
        stack.rebuild().expect("rebuild");
        let rebuild_ms = t.elapsed().as_secs_f64() * 1e3;
        assert_eq!(stack.projection_len(), n);
        assert_eq!(stack.len(), n);

        // The split arm: same rows, but the projection only covers `split_at`.
        let split = Stack::in_memory(RebuildTriggers::manual()).expect("split stack");
        split.append(&present[..split_at]).expect("seal");
        split.rebuild().expect("rebuild");
        split.append(&present[split_at..]).expect("tail");
        assert_eq!(split.projection_len(), split_at);
        assert_eq!(split.len(), n);

        let (_, here) = loadavg();
        println!("loadavg: `{here}`");
        println!();
        println!("| build step | ms |");
        println!("|---|---:|");
        println!("| `OneTableFourColumns::build` (the projection alone) | {arrow_build_ms:.1} |");
        println!("| redb ingest of {n} rows (one write transaction) | {ingest_ms:.1} |");
        println!("| **rebuild** = ordered tail scan + projection build | **{rebuild_ms:.1}** |");

        println!();
        println!(
            "| mix | arrow arm ns | absorbed stack ns | split stack ns | proj hits | tail hits | absent | band |"
        );
        println!("|---|---:|---:|---:|---:|---:|---:|---:|");

        for (hit_pct, label) in [
            (100usize, "100% hit (want)"),
            (50, "50% hit"),
            (10, "10% hit (have)"),
        ] {
            let (owned, expected) = queries(
                &present,
                &absent,
                total_queries,
                hit_pct,
                1 + hit_pct as u64,
            );
            let refs: Vec<&[u8]> = owned.iter().map(|o| o.as_slice()).collect();

            let (s_plain, c_plain, h_plain) = time_batch(&plain, &refs, runs);
            assert_eq!(h_plain, expected, "arrow arm hit count");

            let before = stack.stats();
            let (s_abs, c_abs, h_abs) = time_batch(&stack, &refs, runs);
            let (abs_tail, abs_absent) = delta(before, stack.stats());
            assert_eq!(h_abs, expected, "absorbed stack hit count");
            assert_eq!(c_abs, c_plain, "the absorbed stack is not the same index");
            assert_eq!(
                abs_tail, 0,
                "an absorbed stack served {abs_tail} rows from the tail"
            );

            let before = split.stats();
            let (s_split, c_split, h_split) = time_batch(&split, &refs, runs);
            let (sp_tail, sp_absent) = delta(before, split.stats());
            assert_eq!(h_split, expected, "split stack hit count");
            assert_eq!(c_split, c_plain, "the split stack is not the same index");
            assert!(
                sp_tail > 0,
                "the split arm never fell through — nothing was measured"
            );

            // Applied output, per run: rows the projection answered vs rows redb
            // answered vs rows nothing answered.
            let txns = split.stats().tail_txns;
            let per_run = |x: u64| x / runs as u64;
            let proj_hits = expected as u64 - per_run(sp_tail);
            let band = s_plain.band().max(s_abs.band()).max(s_split.band());
            bands.push(band);
            println!(
                "| {label} | {:.0} [{:.0}–{:.0}] | {:.0} [{:.0}–{:.0}] | {:.0} [{:.0}–{:.0}] | {proj_hits} | {} | {} | {:.1}% |",
                s_plain.med,
                s_plain.lo,
                s_plain.hi,
                s_abs.med,
                s_abs.lo,
                s_abs.hi,
                s_split.med,
                s_split.lo,
                s_split.hi,
                per_run(sp_tail),
                per_run(sp_absent),
                band * 100.0
            );
            let _ = (abs_absent, txns);
        }

        // The number the miss threshold is set from: the cost of one lookup the
        // tail has to answer, isolated by querying ONLY tail rows.
        let tail_only: Vec<&[u8]> = present[split_at..]
            .iter()
            .map(|e| e.oid.as_slice())
            .collect();
        let proj_only: Vec<&[u8]> = present[..split_at.min(tail_only.len())]
            .iter()
            .map(|e| e.oid.as_slice())
            .collect();
        let (s_tail, _, _) = time_batch(&split, &tail_only, runs);
        let (s_proj, _, _) = time_batch(&split, &proj_only, runs);
        println!();
        println!(
            "Isolated: a lookup the **projection** answers costs {:.0} ns; one the **tail** \
             answers costs {:.0} ns; the fall-through therefore costs **{:.0} ns** per row it \
             has to carry.",
            s_proj.med,
            s_tail.med,
            s_tail.med - s_proj.med
        );
        // The break-even the default miss trigger is derived from: rebuilding
        // is worth it once the fall-through has cost as much as the rebuild.
        let per_row_rebuild_ns = rebuild_ms * 1e6 / n as f64;
        let fallthrough_ns = s_tail.med - s_proj.med;
        println!(
            "A rebuild costs {per_row_rebuild_ns:.0} ns per row in the repository. Break-even is \
             therefore **{:.2} tail-served lookups per row** — the default \
             `tail_hits_per_row` is {:.1} (floor {} misses), which fires here at {} misses.",
            per_row_rebuild_ns / fallthrough_ns,
            RebuildTriggers::DEFAULT_TAIL_HITS_PER_ROW,
            RebuildTriggers::DEFAULT_MIN_TAIL_HITS,
            RebuildTriggers::default()
                .miss_threshold(split_at as u64)
                .unwrap()
        );
    }

    bands.sort_by(|a, b| a.partial_cmp(b).unwrap());
    let (_, end) = loadavg();
    println!();
    println!("## Noise band");
    println!();
    println!(
        "Run-to-run spread `(max-min)/median` over {} timed rows: median **{:.1}%**, worst \
         **{:.1}%**. No difference smaller than a row's own band is a result.",
        bands.len(),
        bands[bands.len() / 2] * 100.0,
        bands.last().unwrap() * 100.0
    );
    println!();
    println!("loadavg at end: `{end}`");
}