bench/src/readside.rs
//! One fetch, fully counted. See [`probe`].
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;
use anyhow::{ensure, Context, Result};
use gix_hash::ObjectId;
use gunnar_store::git_storage_trait::{
Caps, Extent, GcReport, GitOps, GitServe, ObjType, Oid, PackStats, ReachSet, RefCas, RefRow,
RefUpdate, Stored, TxId,
};
use gunnar_wire::upload_pack::UploadPack;
#[derive(Default, Debug)]
pub(crate) struct Tally {
reads: AtomicU64,
read_bytes: AtomicU64,
headers: AtomicU64,
contains: AtomicU64,
/// Per-kind read counts and inflated bytes, indexed by `Kind as usize`.
per_kind: Mutex<BTreeMap<&'static str, (u64, u64)>>,
/// Distinct oids `read` was called on, so re-reads are visible.
distinct: Mutex<std::collections::HashSet<ObjectId>>,
rereads: AtomicU64,
}
impl Tally {
fn note(&self, id: ObjectId, kind: ObjType, len: u64) {
self.reads.fetch_add(1, Ordering::Relaxed);
self.read_bytes.fetch_add(len, Ordering::Relaxed);
// `OfsDelta`/`RefDelta` are pack-encoding types, not object types.
// `GitServe::read` is the INFLATED read and contracts to return a real
// object, so they cannot arrive here; they are named rather than folded
// into a catch-all so that an engine that broke that promise shows up as
// a visible bucket instead of being silently counted as a blob.
let name = match kind {
ObjType::Commit => "commit",
ObjType::Tree => "tree",
ObjType::Blob => "blob",
ObjType::Tag => "tag",
ObjType::OfsDelta => "BUG-ofs-delta",
ObjType::RefDelta => "BUG-ref-delta",
};
let mut m = self.per_kind.lock().unwrap();
let e = m.entry(name).or_insert((0, 0));
e.0 += 1;
e.1 += len;
drop(m);
if !self.distinct.lock().unwrap().insert(id) {
self.rereads.fetch_add(1, Ordering::Relaxed);
}
}
/// Read-side counters, for a caller in another module that reuses this
/// decorator (the advertisement probe). Named accessors rather than public
/// fields so the atomics stay encapsulated.
pub(crate) fn reads(&self) -> u64 {
self.reads.load(Ordering::Relaxed)
}
pub(crate) fn read_bytes(&self) -> u64 {
self.read_bytes.load(Ordering::Relaxed)
}
pub(crate) fn headers(&self) -> u64 {
self.headers.load(Ordering::Relaxed)
}
pub(crate) fn rereads(&self) -> u64 {
self.rereads.load(Ordering::Relaxed)
}
pub(crate) fn kinds(&self) -> BTreeMap<&'static str, (u64, u64)> {
self.per_kind.lock().unwrap().clone()
}
}
/// Every store touch one fetch makes, charged to a [`Tally`].
///
/// # It wraps a `GitServe`, and that changed which methods exist to count
///
/// This used to decorate `gunnar_store::ObjectStore` — five methods, of which
/// `contains` / `contains_without_rescan` were the two existence probes and
/// `read(&oid, &mut buf) -> Data<'_>` was the inflating read. That trait died in
/// the 2026-08-10 demolition. The contract is now `GitOps` (the twelve) plus
/// `GitServe` (the reading half), and the mapping onto the three counters this
/// arm publishes is:
///
/// * `store_contains` <- [`GitOps::has`]. The single existence probe; the
/// `contains` / `contains_without_rescan` split is gone, because the *reason*
/// for the split was a filesystem re-scan on a miss and the contract no longer
/// promises anyone a filesystem. Both used to count into this same counter
/// anyway, so the published number means what it always meant.
/// * `store_reads` / `store_read_inflated_bytes` / `by_kind` <- [`GitServe::read`],
/// which now returns an owned `(ObjType, Vec<u8>)` rather than borrowing a
/// caller's buffer. The byte figure is still the INFLATED length.
/// * `store_headers` <- [`GitServe::header`] **and** [`GitServe::sizes`], the
/// latter charged once per id asked about rather than once per call. `sizes`
/// is the bulk spelling of `header` and the negotiation path calls it a
/// thousand at a time; charging it one would let a change that moved work from
/// `header` into `sizes` read as a change that removed the work. This mirrors
/// `gunnar-wire/tests/common/mod.rs`'s `Counting`, deliberately, so the two
/// agree on what a "touch" is.
///
/// [`GitOps::get`] is *not* counted as a read. It is the stored-bytes accessor,
/// not the inflating one, and folding it in would double-count an engine that
/// serves `read` through it.
pub(crate) struct Counting<'a> {
pub(crate) inner: &'a dyn GitServe,
pub(crate) tally: &'a Tally,
}
/// The twelve, forwarded; only [`GitOps::has`] is counted. The rest are writes
/// and bulk index questions that this arm publishes no counter for, and a
/// counter nothing reads is a number that drifts unnoticed.
impl GitOps for Counting<'_> {
fn put(&self, pack: &[u8], refs: &[RefUpdate]) -> anyhow::Result<TxId> {
self.inner.put(pack, refs)
}
fn put_pack(&self, bytes: &[u8]) -> anyhow::Result<TxId> {
self.inner.put_pack(bytes)
}
fn put_refs(&self, updates: &[RefUpdate]) -> anyhow::Result<TxId> {
self.inner.put_refs(updates)
}
fn get(&self, oid: Oid<'_>) -> anyhow::Result<Option<Stored>> {
self.inner.get(oid)
}
fn has(&self, oid: Oid<'_>) -> anyhow::Result<bool> {
self.tally.contains.fetch_add(1, Ordering::Relaxed);
self.inner.has(oid)
}
fn size(&self, oid: Oid<'_>) -> anyhow::Result<Option<u64>> {
self.inner.size(oid)
}
fn extents(&self, oids: &[Oid<'_>]) -> anyhow::Result<Vec<Option<Extent>>> {
self.inner.extents(oids)
}
fn refs(&self) -> anyhow::Result<Vec<RefRow>> {
self.inner.refs()
}
fn update_ref(
&self,
name: &str,
old: Option<Oid<'_>>,
new: Option<Oid<'_>>,
) -> anyhow::Result<TxId> {
self.inner.update_ref(name, old, new)
}
fn put_refs_cas(&self, edits: &[RefCas<'_>]) -> anyhow::Result<TxId> {
self.inner.put_refs_cas(edits)
}
fn reachable(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> anyhow::Result<Vec<Vec<u8>>> {
self.inner.reachable(want, have)
}
fn gc(&self) -> anyhow::Result<GcReport> {
self.inner.gc()
}
}
impl GitServe for Counting<'_> {
fn read(&self, oid: Oid<'_>) -> anyhow::Result<Option<(ObjType, Vec<u8>)>> {
let owned = ObjectId::try_from(oid).ok();
let out = self.inner.read(oid)?;
if let (Some(id), Some((kind, bytes))) = (owned, out.as_ref()) {
self.tally.note(id, *kind, bytes.len() as u64);
}
Ok(out)
}
fn header(&self, oid: Oid<'_>) -> anyhow::Result<Option<(ObjType, u64)>> {
self.tally.headers.fetch_add(1, Ordering::Relaxed);
self.inner.header(oid)
}
/// Counted once per id asked about; see the type header.
fn sizes(&self, oids: &[Oid<'_>]) -> anyhow::Result<Vec<Option<u64>>> {
self.tally
.headers
.fetch_add(oids.len() as u64, Ordering::Relaxed);
self.inner.sizes(oids)
}
fn head(&self) -> anyhow::Result<Option<RefRow>> {
self.inner.head()
}
fn set_head(&self, target: &str) -> anyhow::Result<TxId> {
self.inner.set_head(target)
}
fn emit_pack(
&self,
objects: &[Oid<'_>],
have: &[Oid<'_>],
caps: &Caps,
out: &mut dyn std::io::Write,
) -> anyhow::Result<PackStats> {
self.inner.emit_pack(objects, have, caps, out)
}
fn select(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> anyhow::Result<Option<ReachSet>> {
self.inner.select(want, have)
}
}
/// `read_bytes` and `rchar` out of `/proc/self/io`, the same pair
/// `gunnar.page_cache` reads out of the daemon.
fn proc_io() -> (u64, u64) {
let text = std::fs::read_to_string("/proc/self/io").unwrap_or_default();
let get = |key: &str| -> u64 {
text.lines()
.find_map(|l| l.strip_prefix(key))
.and_then(|v| v.trim().parse().ok())
.unwrap_or(0)
};
(get("read_bytes:"), get("rchar:"))
}
/// Drop every pack out of the page cache, so `read_bytes` measures real device
/// traffic rather than a second warm run.
///
/// `posix_fadvise(DONTNEED)` is advisory; the caller must check that residency
/// actually fell, which is what `evicted_resident_pages` reports.
/// Drop this repository's store out of the page cache, **whatever engine wrote
/// it**.
///
/// `objects/pack` is the gix engine's store and does not exist on the znippy
/// arm, where the bytes live in `gunnar-git.znippy/`. Evicting a directory that
/// is not there is not a smaller eviction: `read_dir` fails, the probe exits
/// non-zero, and `gunnar.fetch_readside` reported it as *"the full readside
/// child exited exit status: 1"* with an empty stderr. The shape is detected
/// through `crate::store_bytes`, which refuses a layout it does not recognise
/// rather than returning an empty list.
pub(crate) fn evict(repo: &std::path::Path) -> Result<u64> {
use std::os::unix::io::AsRawFd;
let dir = crate::store_bytes::pack_dir(repo)
.with_context(|| format!("{repo:?} holds no object store to evict"))?;
let mut still = 0u64;
for entry in std::fs::read_dir(&dir)? {
let path = entry?.path();
let f = std::fs::File::open(&path)?;
let len = f.metadata()?.len();
unsafe {
libc::posix_fadvise(
f.as_raw_fd(),
0,
len as libc::off_t,
libc::POSIX_FADV_DONTNEED,
);
}
still += resident_pages(&f, len)?;
}
Ok(still)
}
fn resident_pages(f: &std::fs::File, len: u64) -> Result<u64> {
use std::os::unix::io::AsRawFd;
if len == 0 {
return Ok(0);
}
let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) } as u64;
let pages = len.div_ceil(page) as usize;
unsafe {
let map = libc::mmap(
std::ptr::null_mut(),
len as usize,
libc::PROT_READ,
libc::MAP_SHARED,
f.as_raw_fd(),
0,
);
ensure!(map != libc::MAP_FAILED, "mmap failed");
let mut vec = vec![0u8; pages];
let rc = libc::mincore(map, len as usize, vec.as_mut_ptr());
libc::munmap(map, len as usize);
ensure!(rc == 0, "mincore failed");
Ok(vec.iter().filter(|b| **b & 1 == 1).count() as u64)
}
}
/// Resident bytes of the store, **whichever directory this engine keeps it in**.
/// See `store_bytes::pack_dir`; a hardcoded `objects/pack` here was half of the
/// reason this probe died on every znippy column.
fn pack_resident_bytes(repo: &std::path::Path) -> Result<u64> {
let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) } as u64;
let mut out = 0;
let dir = crate::store_bytes::pack_dir(repo)
.with_context(|| format!("{repo:?} holds no object store to measure"))?;
for entry in std::fs::read_dir(&dir)? {
let path = entry?.path();
let f = std::fs::File::open(&path)?;
let len = f.metadata()?.len();
out += resident_pages(&f, len)? * page;
}
Ok(out)
}
fn pkt(payload: &str) -> Vec<u8> {
let mut out = format!("{:04x}", payload.len() + 4).into_bytes();
out.extend_from_slice(payload.as_bytes());
out
}
/// Serve exactly one incremental fetch and report what it cost.
///
/// Shared by `examples/readside-probe.rs` (the diagnostic) and by
/// `gunnar.fetch_readside` (the persisted arm), which is the whole reason this
/// crate grew a library: two targets needed the same driver, so it is shared
/// rather than kept as two copies that have to be nursed into agreement.
pub fn probe(repo: &std::path::Path, from: &str, to: &str) -> Result<serde_json::Value> {
let repo = repo.to_path_buf();
let hash = gix_hash::Kind::Sha1;
// Evict first, then open: opening the store is itself a read the arm must
// be able to see, and it is charged to `open_disk_mib` below.
// `PROBE_NO_EVICT=1` drops the eviction *and* the two `mincore` sweeps over
// a 377 MiB pack, which are ~96k page probes each and would otherwise sit
// as a large constant inside any `perf stat` of this binary. The disk and
// residency figures are then meaningless and are reported as `null` rather
// than as a confident zero.
let measure_io = std::env::var_os("PROBE_NO_EVICT").is_none();
let still_resident_pages = if measure_io { evict(&repo)? } else { 0 };
let (io0, rchar0) = proc_io();
let store =
crate::engine::open_repo(&repo, hash).with_context(|| format!("opening {repo:?}"))?;
// One handle now, not two. `RepoStore::objects()`/`refs()` were the split
// that `ObjectStore`/`RefStore` forced; `RepoStore::git()` hands back a
// single `Arc<dyn GitServe>` that owns both halves, so the counting
// decorator sees the ref reads too rather than only the object reads.
let git = store.git();
let (io_open, _) = proc_io();
// Resolved out of the ref ROW STREAM rather than through a by-name lookup,
// because the contract has no by-name read: `GitOps::refs()` returns every
// row and the caller picks. That is not a workaround — it is settled that
// `refs()` is the enumeration and `HEAD` is deliberately not in it — so a
// probe naming `HEAD` here would find nothing and must say so rather than
// resolve it behind the caller's back.
let rows = git.refs().context("listing refs")?;
let resolve = |name: &str| -> Result<ObjectId> {
let row = rows
.iter()
.find(|r| r.name == name)
.with_context(|| format!("{name} not found"))?;
match (&row.oid, &row.symref_target) {
(Some(raw), _) => ObjectId::try_from(raw.as_slice())
.map_err(|e| anyhow::anyhow!("{name} has an unreadable oid: {e}")),
(None, Some(s)) => anyhow::bail!("{name} is symbolic to {s}"),
(None, None) => anyhow::bail!("{name} has neither an oid nor a symref target"),
}
};
let want = resolve(to)?;
let have = resolve(from)?;
let tally = Tally::default();
let counting = Counting {
inner: git.as_ref(),
tally: &tally,
};
// A v2 `command=fetch` with one want, one have, and `done`, which is the
// shape stock git sends for an incremental fetch it has already negotiated.
let mut req = Vec::new();
req.extend(pkt("command=fetch\n"));
req.extend(pkt("object-format=sha1\n"));
req.extend_from_slice(b"0001");
req.extend(pkt("thin-pack\n"));
req.extend(pkt("no-progress\n"));
req.extend(pkt("ofs-delta\n"));
req.extend(pkt(&format!("want {want}\n")));
req.extend(pkt(&format!("have {have}\n")));
req.extend(pkt("done\n"));
req.extend_from_slice(b"0000");
// `UploadPack::new(git, hash_kind)`. It took `(&objects, &refs)` when refs
// were a second trait; the hash kind is now passed rather than asked of the
// store, because `GitServe` does not carry one — `RepoStore::hash_kind()`
// does, and that is the handle the caller already has.
let upload = UploadPack::new(&counting, store.hash_kind());
let mut input = std::io::BufReader::new(std::io::Cursor::new(req));
let mut out: Vec<u8> = Vec::new();
let t0 = std::time::Instant::now();
let summary = upload.serve_v2(&mut input, &mut out)?;
let ms = t0.elapsed().as_secs_f64() * 1000.0;
let (io1, rchar1) = proc_io();
let resident = if measure_io {
pack_resident_bytes(&repo)?
} else {
0
};
ensure!(
summary.objects_sent > 0,
"the probe's fetch sent ZERO objects, so every tally below describes an empty request"
);
// `PROBE_OUT_PACK=<path>` writes the packfile stock git would have received,
// demultiplexed out of the v2 `packfile` section, so `git index-pack` and
// `git fsck` can be the judge of whether an exclusion rule that is allowed
// to send a *superset* ever sent a subset instead.
if let Some(path) = std::env::var_os("PROBE_OUT_PACK") {
let mut pack = Vec::new();
let mut cursor = &out[..];
let mut in_packfile = false;
while cursor.len() >= 4 {
let len = usize::from_str_radix(std::str::from_utf8(&cursor[..4])?, 16)?;
if len < 4 {
// A delimiter (0000/0001/0002): no payload.
cursor = &cursor[4..];
continue;
}
ensure!(
len <= cursor.len(),
"a pkt-line ran past the end of the response"
);
let payload = &cursor[4..len];
if in_packfile {
ensure!(!payload.is_empty(), "an empty band in the packfile section");
// Band 1 is data; 2 is progress, 3 is a fatal error.
match payload[0] {
1 => pack.extend_from_slice(&payload[1..]),
3 => anyhow::bail!(
"server sent band 3: {}",
String::from_utf8_lossy(&payload[1..])
),
_ => {}
}
} else if payload == b"packfile\n" {
in_packfile = true;
}
cursor = &cursor[len..];
}
ensure!(
pack.starts_with(b"PACK"),
"the demultiplexed bytes do not start with the pack signature, so \
PROBE_OUT_PACK would hand `git index-pack` something that is not a pack"
);
std::fs::write(std::path::PathBuf::from(path), &pack)?;
}
let per_kind = tally.per_kind.lock().unwrap().clone();
Ok(serde_json::json!({
"repo": repo.to_string_lossy(),
"want": want.to_string(),
"have": have.to_string(),
"objects_sent": summary.objects_sent,
"wire_bytes": out.len(),
"io_measured": measure_io,
"evict_left_resident_pages": if measure_io { serde_json::json!(still_resident_pages) } else { serde_json::Value::Null },
"open_disk_bytes": if measure_io { serde_json::json!(io_open - io0) } else { serde_json::Value::Null },
"open_disk_mib": if measure_io { serde_json::json!(((io_open - io0) as f64 / 1048576.0 * 100.0).round() / 100.0) } else { serde_json::Value::Null },
"serve_disk_bytes": if measure_io { serde_json::json!(io1 - io_open) } else { serde_json::Value::Null },
"serve_disk_mib": if measure_io { serde_json::json!(((io1 - io_open) as f64 / 1048576.0 * 100.0).round() / 100.0) } else { serde_json::Value::Null },
"serve_syscall_read_mib": if measure_io { serde_json::json!(((rchar1 - rchar0) as f64 / 1048576.0 * 100.0).round() / 100.0) } else { serde_json::Value::Null },
"pack_resident_after_mib": if measure_io { serde_json::json!((resident as f64 / 1048576.0 * 100.0).round() / 100.0) } else { serde_json::Value::Null },
"serve_ms": (ms * 100.0).round() / 100.0,
"store_reads": tally.reads.load(Ordering::Relaxed),
"store_read_inflated_bytes": tally.read_bytes.load(Ordering::Relaxed),
"store_read_inflated_mib": (tally.read_bytes.load(Ordering::Relaxed) as f64 / 1048576.0 * 100.0).round() / 100.0,
"store_rereads": tally.rereads.load(Ordering::Relaxed),
"store_headers": tally.headers.load(Ordering::Relaxed),
"store_contains": tally.contains.load(Ordering::Relaxed),
"by_kind": per_kind.iter().map(|(k, (n, b))| (k.to_string(), serde_json::json!({"reads": n, "bytes": b}))).collect::<serde_json::Map<_,_>>(),
}))
}