znippy-plugin-git 0.2.4

//! **What one OPEN STORE costs in resident memory — the slope, not the guess.**
//!
//! # The question
//!
//! gunnar's server budget prices a cached open repository at
//! `gunnar_server::budget::BYTES_PER_OPEN_REPO = 8 MiB` and derives the whole
//! handle ceiling from it:
//!
//! ```text
//! open_repos = min((soft − 256) / 3, MemAvailable × 20 % / 8 MiB)
//! ```
//!
//! An appliance with 25 repositories idles at 10.0 G of 16 G with load 0.00 and
//! nothing in flight (`gunnar/.nornir/stinky.md` S-079). 25 × 8 MiB is 200 MiB,
//! so either that divisor is wrong or the memory is somewhere else. S-079 wrote
//! down an arithmetic *hypothesis* — two 64 MiB redb page caches plus a 4 × 64 MiB
//! exploded batch cache, ~384 MiB a store — and said in the entry that it was
//! unmeasured. **This is the measurement.**
//!
//! # What it measures, and what it deliberately does not
//!
//! One process. N identical stores, opened one at a time, each read from, all
//! held open — which is exactly the state a server's handle cache is in, because
//! its bound (256) never fires over 25 repositories and nothing is ever evicted.
//! After each open it reports **anonymous** RSS out of `/proc/self/smaps_rollup`
//! and glibc's own `mallinfo2` books.
//!
//! `Anonymous:` and not `Rss:`, for the same reason
//! `tests/rss_retention_probe.rs` gives: `Rss` counts file-backed pages — the
//! mapped pack, the page cache a `pread` pulls in — which the kernel reclaims on
//! demand and which no budget change would touch. The dashboard figure S-079 is
//! about is `MemTotal − MemAvailable`, and `MemAvailable` already discounts
//! reclaimable page cache, so anonymous is the comparable quantity.
//!
//! It measures the **store**, not the server: `gunnar-server` adds one
//! `control.redb` and its own baseline, both of which are per-process and not
//! per-store, so they cancel out of a slope.
//!
//! # Counts, not clocks
//!
//! Every number here is a byte count. That is deliberate: the rig this runs on
//! is shared, and a byte count does not care what else is on the box, where a
//! latency would. The load average is printed on every row anyway so a reader
//! can see what it was taken under.
//!
//! # Run
//!
//! ```bash
//! cargo run --release -p znippy-plugin-git --example rss_per_open_store -- \
//!   <pack file> <scratch dir> <N> [reads per store]
//! ```
//!
//! The pack is pushed into store 0 once and the finished directory is **copied**
//! to the other N−1, so every store is byte-identical and the run costs one
//! index rather than N. Set `ZNIPPY_GIT_EXPLODE` / `ZNIPPY_GIT_REDB_CACHE_BYTES`
//! to measure a different arrangement; both are echoed on the header line.
//!
//! # `synthetic:<objects>` in place of a pack path
//!
//! The cost per store turns out to be **per object**, not per store, so the
//! interesting axis is object count and the packs lying around on a developer
//! box all sit within a factor of two of each other. `synthetic:400000` builds a
//! pack of that many distinct 64-byte blobs in memory — real pack framing, real
//! zlib streams, real sha1 oids, `git index-pack` would accept it — so the
//! object term can be measured over two decades instead of over 1.8×.
//!
//! Blobs and no trees or commits, deliberately: it isolates the term that scales
//! with **objects in the index** from the one that scales with *commits*, and
//! the two are folded by different code.

use std::path::{Path, PathBuf};

use anyhow::{Context, Result};
use git_storage_trait::GitOps as _;
use znippy_plugin_git::git_ops::GitStore;

/// Anonymous RSS of this process, in KiB, from `/proc/self/smaps_rollup`.
fn anon_rss_kb() -> u64 {
    let s = std::fs::read_to_string("/proc/self/smaps_rollup")
        .expect("/proc/self/smaps_rollup (Linux 4.14+)");
    for line in s.lines() {
        if let Some(rest) = line.strip_prefix("Anonymous:") {
            let kb = rest.trim().trim_end_matches("kB").trim();
            return kb.parse().expect("Anonymous: is a kB count");
        }
    }
    panic!("no `Anonymous:` line in smaps_rollup");
}

/// Total RSS in KiB — anonymous plus file-backed. Printed beside the anonymous
/// figure so a reader can see how much of the process is mapped pack.
fn total_rss_kb() -> u64 {
    let s = std::fs::read_to_string("/proc/self/smaps_rollup").unwrap_or_default();
    for line in s.lines() {
        if let Some(rest) = line.strip_prefix("Rss:") {
            let kb = rest.trim().trim_end_matches("kB").trim();
            return kb.parse().unwrap_or(0);
        }
    }
    0
}

/// `(in_use, from_kernel)` in KiB, via `mallinfo2`. `uordblks + hblkhd` is what
/// the program has asked for and not given back — live, reachable bytes — which
/// is the reading that says whether resident growth is a structure somebody owns
/// or an allocator that has not handed pages back.
fn mallinfo_kb() -> Option<(u64, u64)> {
    #[cfg(all(target_os = "linux", target_env = "gnu"))]
    {
        // SAFETY: `mallinfo2` takes no arguments and returns a plain POD struct.
        let mi = unsafe { libc::mallinfo2() };
        return Some((
            (mi.uordblks as u64 + mi.hblkhd as u64) / 1024,
            (mi.arena as u64 + mi.hblkhd as u64) / 1024,
        ));
    }
    #[cfg(not(all(target_os = "linux", target_env = "gnu")))]
    None
}

fn malloc_trim() {
    #[cfg(all(target_os = "linux", target_env = "gnu"))]
    unsafe {
        libc::malloc_trim(0)
    };
}

fn mb(kb: u64) -> f64 {
    kb as f64 / 1024.0
}

fn loadavg() -> String {
    std::fs::read_to_string("/proc/loadavg")
        .unwrap_or_default()
        .split_whitespace()
        .take(3)
        .collect::<Vec<_>>()
        .join(" ")
}

/// Copy a directory tree. Two `File`s and `std::io::copy`; no shelling (LAW).
fn copy_tree(from: &Path, to: &Path) -> Result<()> {
    std::fs::create_dir_all(to)?;
    for e in std::fs::read_dir(from)? {
        let e = e?;
        let src = e.path();
        let dst = to.join(e.file_name());
        if e.file_type()?.is_dir() {
            copy_tree(&src, &dst)?;
        } else {
            std::fs::copy(&src, &dst)
                .with_context(|| format!("copying {} to {}", src.display(), dst.display()))?;
        }
    }
    Ok(())
}

/// **A v2 packfile of `n` distinct blobs**, framed by hand.
///
/// Written here rather than taken from `src/pack_fixture.rs` because that module
/// is `pub(crate)` — an example cannot see it — and because what is wanted is
/// the opposite of a fixture: not a fixed corpus with a named delta shape, but
/// one knob, the object count, over two decades. It is deliberately delta-free,
/// so the number that comes out is the cost of *an oid in the index* and not the
/// cost of a chain.
fn synthetic_pack(n: usize) -> Vec<u8> {
    use std::io::Write as _;

    use sha1::Digest as _;

    let mut out = Vec::with_capacity(n * 80 + 32);
    out.extend_from_slice(b"PACK");
    out.extend_from_slice(&2u32.to_be_bytes());
    out.extend_from_slice(&(n as u32).to_be_bytes());
    for i in 0..n {
        // 64 bytes, distinct, and incompressible enough that zlib does not turn
        // the whole corpus into one repeated stream: the pack's size is not the
        // subject here, but a corpus that deflates to nothing would put the
        // whole thing in one page and stop resembling a repository.
        let mut body = [0u8; 64];
        let mut s = (i as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1;
        for c in body.chunks_mut(8) {
            s ^= s << 13;
            s ^= s >> 7;
            s ^= s << 17;
            c.copy_from_slice(&s.to_le_bytes());
        }
        // Entry header: 3 = OBJ_BLOB, then the uncompressed size, 4 bits in the
        // first byte and 7 per byte after it.
        let mut size = body.len() as u64;
        let mut b0 = ((3u8) << 4) | (size & 0x0F) as u8;
        size >>= 4;
        if size > 0 {
            b0 |= 0x80;
        }
        out.push(b0);
        while size > 0 {
            let mut b = (size & 0x7F) as u8;
            size >>= 7;
            if size > 0 {
                b |= 0x80;
            }
            out.push(b);
        }
        let mut z = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
        z.write_all(&body).expect("deflate a blob");
        out.extend_from_slice(&z.finish().expect("finish a blob"));
    }
    let mut h = sha1::Sha1::new();
    h.update(&out);
    out.extend_from_slice(&h.finalize());
    out
}

/// Every `n`th oid of `all`, at most `want` of them.
fn stride(all: &[Vec<u8>], want: usize) -> Vec<Vec<u8>> {
    if all.is_empty() || want == 0 {
        return Vec::new();
    }
    let step = (all.len() / want).max(1);
    all.iter().step_by(step).take(want).cloned().collect()
}

fn main() -> Result<()> {
    let mut args = std::env::args().skip(1);
    let spec = args
        .next()
        .context("usage: <pack | synthetic:OBJECTS> <scratch dir> <N> [reads]")?;
    let root = PathBuf::from(args.next().context("usage: <pack> <scratch dir> <N> [reads]")?);
    let n: usize = args.next().unwrap_or_else(|| "25".into()).parse().unwrap_or(25);
    let reads: usize = args.next().unwrap_or_else(|| "64".into()).parse().unwrap_or(64);

    if root.exists() {
        std::fs::remove_dir_all(&root).with_context(|| format!("clearing {}", root.display()))?;
    }
    std::fs::create_dir_all(&root)?;

    // Either arm hands the push the same thing: a `&[u8]` of pack bytes. The
    // real one is mapped rather than read so an 8 GiB pack is not 8 GiB of anon
    // this measurement did not want to be about; the synthetic one is built.
    let (map, built, label): (Option<memmap2::Mmap>, Vec<u8>, String) =
        match spec.strip_prefix("synthetic:") {
            Some(count) => {
                let objects: usize = count.parse().context("synthetic:<objects>")?;
                (None, synthetic_pack(objects), format!("synthetic, {objects} blobs"))
            }
            None => {
                let pack_path = PathBuf::from(&spec);
                let f = std::fs::File::open(&pack_path)
                    .with_context(|| format!("opening {}", pack_path.display()))?;
                let m = unsafe { memmap2::Mmap::map(&f)? };
                let label = pack_path.display().to_string();
                (Some(m), Vec::new(), label)
            }
        };
    let pack: &[u8] = match &map {
        Some(m) => &m[..],
        None => &built[..],
    };

    println!("=== RSS PER OPEN STORE ===");
    println!("pack        {:>14} B  {label}", pack.len());
    println!("scratch     {}", root.display());
    println!("stores      {n}   reads per store {reads}");
    println!(
        "explode     {}   redb cache {}",
        std::env::var("ZNIPPY_GIT_EXPLODE").unwrap_or_else(|_| "(default)".into()),
        std::env::var("ZNIPPY_GIT_REDB_CACHE_BYTES").unwrap_or_else(|_| "(default 64 MiB)".into()),
    );
    println!("loadavg     {}", loadavg());

    // ── PHASE A — build ONE store, then copy it ─────────────────────────────
    //
    // Copied and not pushed N times, so every store is byte-identical: a slope
    // over stores that differ is a slope over the difference, not over the
    // handle.
    let seed = root.join("store-0");
    let (objects, picks) = {
        let store = GitStore::open(&seed, "alice").context("opening the seed store")?;
        let tx = store.put_pack(pack).context("the push")?;
        store.wait_indexed();
        let objects = store.object_count();
        let extent = tx.extent.context("the push reported no extent")?;
        let oids = store.oids_in_extent(extent).context("listing the pack's oids")?;
        let picks = stride(&oids, reads);
        (objects, picks)
    };
    drop(map);
    drop(built);
    println!("seeded      {objects} objects, {} oids picked for reads", picks.len());
    for i in 1..n {
        copy_tree(&seed, &root.join(format!("store-{i}")))?;
    }

    // The seed store's own open, its push and its index all ran in this
    // process. Trim first so the baseline is the process AFTER that work, not
    // during it — otherwise every store's delta is measured against a high
    // baseline and the slope comes out flatter than it is.
    malloc_trim();
    let base = anon_rss_kb();
    let base_total = total_rss_kb();
    let (base_use, _) = mallinfo_kb().unwrap_or((0, 0));
    println!(
        "\nbaseline    anon {:8.1} MB   rss {:8.1} MB   malloc in-use {:8.1} MB\n",
        mb(base),
        mb(base_total),
        mb(base_use)
    );

    // ── PHASE B — open them one at a time, holding every handle ─────────────
    println!(
        "{:>5}  {:>12}  {:>12}  {:>12}  {:>12}  {:>12}",
        "open", "anon MB", "Δ anon MB", "per store MB", "malloc MB", "rss MB"
    );
    let mut held = Vec::with_capacity(n);
    let mut rows: Vec<(usize, u64)> = Vec::with_capacity(n);
    let mut live: Vec<(usize, u64)> = Vec::with_capacity(n);
    for i in 0..n {
        let dir = root.join(format!("store-{i}"));
        let store = GitStore::open(&dir, "alice")
            .with_context(|| format!("opening store {i}"))?;
        // READ. Without it the exploded batch cache holds nothing and the redb
        // page cache holds only what the open itself touched — the exact way a
        // memory probe under-reports a cache.
        for oid in &picks {
            store
                .content(oid)?
                .with_context(|| format!("store {i}: an object of its own pack is absent"))?;
        }
        // QUIESCE. A reopen adopts the journal and may re-queue index work, and
        // a read admits write-backs into the lazy exploded pool; both run on
        // their own threads. Sampling without waiting reads a random point
        // inside that work and is what made the per-open increments scatter
        // 11.5–43.5 MB on the first synthetic run.
        store.wait_indexed();
        held.push(store);

        // TRIM BEFORE SAMPLING, and it is not cosmetic. Without it the first
        // synthetic run read 23.7 / 36.4 / 42.6 / 37.8 / 39.9 MB for five
        // identical stores — a 44 % spread, which `locked_on_target.md` §6 calls
        // noise rather than a result. The variance is the arena: an open
        // allocates and frees a transient O(objects) scan, and whether those
        // pages are on a free list at the instant of the reading depends on
        // nothing the measurement is about. Trimming first makes every row the
        // same question — what is still HELD — and `malloc in-use`, which is
        // reachable bytes and is not moved by a trim at all, is printed beside
        // it as the control: if the two disagree, the anonymous figure is
        // fragmentation and not a structure.
        malloc_trim();
        let now = anon_rss_kb();
        let (in_use, _) = mallinfo_kb().unwrap_or((0, 0));
        rows.push((i + 1, now));
        live.push((i + 1, in_use));
        println!(
            "{:>5}  {:>12.1}  {:>12.1}  {:>12.1}  {:>12.1}  {:>12.1}",
            i + 1,
            mb(now),
            mb(now.saturating_sub(base)),
            mb(now.saturating_sub(base)) / (i + 1) as f64,
            mb(in_use),
            mb(total_rss_kb()),
        );
    }

    // ── THE SLOPE ───────────────────────────────────────────────────────────
    //
    // The marginal cost of the LAST store rather than the average, because the
    // first open also pays this process's one-off allocations and the question
    // is what the 26th repository costs, not what the 1st did.
    let first = rows.first().copied().unwrap_or((1, base));
    let last = rows.last().copied().unwrap_or((1, base));
    let marginal = if last.0 > first.0 {
        (last.1.saturating_sub(first.1)) as f64 / (last.0 - first.0) as f64
    } else {
        (last.1.saturating_sub(base)) as f64
    };
    // The same slope over `mallinfo`'s in-use figure. Printed beside the
    // resident one because it answers a different question: resident says what
    // the box is paying, in-use says whether the program still OWNS it. A slope
    // that is real in both is a structure; one that is real only in the resident
    // figure is the allocator.
    let live_first = live.first().copied().unwrap_or((1, 0));
    let live_last = live.last().copied().unwrap_or((1, 0));
    let live_marginal = if live_last.0 > live_first.0 {
        (live_last.1.saturating_sub(live_first.1)) as f64 / (live_last.0 - live_first.0) as f64
    } else {
        live_last.1 as f64
    };
    // The spread over the per-store increments, so a reader can see whether the
    // number above is a line or an average of scatter. Over 25 % is noise.
    let mut steps: Vec<f64> = Vec::new();
    for w in rows.windows(2) {
        steps.push(w[1].1.saturating_sub(w[0].1) as f64);
    }
    let (lo, hi) = steps.iter().fold((f64::MAX, 0.0f64), |(l, h), s| (l.min(*s), h.max(*s)));
    println!(
        "\nSLOPE       {:.1} MB per additional open store  (stores {}→{}, \
         {:.1} MB total over baseline)",
        marginal / 1024.0,
        first.0,
        last.0,
        mb(last.1.saturating_sub(base))
    );
    println!(
        "            {:.1} MB by malloc in-use (reachable bytes; agreement with the line \
         above means a structure, not the arena)",
        live_marginal / 1024.0
    );
    if !steps.is_empty() {
        println!(
            "            per-open increments {:.1}–{:.1} MB, spread {:.1} % of the mean",
            lo / 1024.0,
            hi / 1024.0,
            100.0 * (hi - lo) / (steps.iter().sum::<f64>() / steps.len() as f64).max(1.0)
        );
    }
    println!(
        "            {:.1} bytes per object per open store ({objects} objects)",
        marginal * 1024.0 / objects.max(1) as f64
    );
    println!(
        "            gunnar prices this at BYTES_PER_OPEN_REPO = 8.0 MB → measured/budgeted \
         = {:.1}x",
        marginal / 1024.0 / 8.0
    );

    // ── PHASE C — does releasing a handle give the bytes back? ──────────────
    drop(held);
    let dropped = anon_rss_kb();
    malloc_trim();
    let trimmed = anon_rss_kb();
    println!(
        "\nafter drop  {:8.1} MB anon ({:.1}% of the growth still resident)\nafter trim  \
         {:8.1} MB anon ({:.1}% still resident)",
        mb(dropped),
        100.0 * dropped.saturating_sub(base) as f64 / last.1.saturating_sub(base).max(1) as f64,
        mb(trimmed),
        100.0 * trimmed.saturating_sub(base) as f64 / last.1.saturating_sub(base).max(1) as f64,
    );
    println!("loadavg     {}", loadavg());
    Ok(())
}