crates/gunnar-stage/src/lib.rs
//! **Stage attribution.** What did *which part* of serving this request cost.
//!
//! # The question this exists to answer
//!
//! Every benchmark gunnar had measured a whole operation — a clone, a fetch, a
//! push — and reported one number. When a number moved, nothing said which
//! stage moved. Two results on 2026-08-04 stalled on exactly that:
//!
//! * server CPU per request rises **2.54x** between 1 and 32 concurrent clients
//! (0.418 s → 1.063 s) where gitea's *falls*, and "nothing here localises it";
//! * a negotiated fetch costs **2.2–2.6x** on the znippy backend where a warm
//! clone costs 1.06–1.41x, narrowed to "object reads, not graph traversal"
//! and then stopped for want of a read path broken into pieces.
//!
//! A symbol histogram (`gunnar.cpu_phase`) answers *which code* — inflate,
//! deflate, hash. It cannot answer *which stage*, because `inflate` runs in the
//! commit walk, in the tree walk and in the pack writer, and the three call for
//! completely different fixes. This answers the second question.
//!
//! # Shape
//!
//! ```text
//! let _s = gunnar_stage::enter("negotiate"); // ← one line, at a boundary
//! ```
//!
//! Nested stages are charged **exclusively**: entering a child suspends the
//! parent, leaving it resumes. So the `excl_*` columns of a report sum to the
//! whole, which is what makes a *fraction* meaningful — and a fraction, not a
//! raw number, is the deliverable. A stage that is 2% of a fetch is not worth
//! anyone's next week however bad its absolute figure looks.
//!
//! [`Tally::incl_wall_ns`] is kept beside it because "select_objects is 71% of
//! the fetch" and "the tree walk inside it is 64% of the fetch" are both true
//! and both wanted.
//!
//! # Everything is per **thread**
//!
//! Not per process. This repository wrote a counting `#[global_allocator]`
//! before and it was **blind**: it counted per process while `cargo test` runs
//! cases on threads, so every byte landed on whichever case happened to be
//! running. The concurrency question above is a question about 32 threads, so
//! an instrument that cannot separate them cannot answer it at all.
//!
//! Each thread accumulates into its own [`thread_local!`] table and calls
//! [`flush_thread`] to fold that table into the process-wide one.
//! `tests/attribution.rs::allocation_is_per_thread` is what holds that: two
//! threads sit inside the same `walk` stage behind barriers and allocate
//! 1 000 000 and 3 bytes, and it requires the stage to report **1 000 003**.
//! A process-wide counter reports 1 000 003 twice — 2 000 006 — because each
//! thread's interval then spans the other's work. Verified red 2026-08-04 by
//! swapping the per-thread cell for an `AtomicU64`.
//!
//! # What it will not measure for you
//!
//! The kernel facts — per-thread user/sys CPU, resident set size — and the
//! allocator arrive as **installed hooks**, because a crate the appliance links
//! may not grow a `libc` dependency for a benchmark (§2). See
//! [`set_cpu_clock`], [`set_rss_probe`] and [`note_alloc`]. Nothing is
//! installed by default and nothing is installed by this crate; a server that
//! links it and installs nothing gets zeros in those columns, which is the
//! honest answer to "what did you measure" rather than a plausible number.
//!
//! # Cost
//!
//! Disabled (the default) a stage boundary is one relaxed atomic load. Enabled
//! it is two `Instant::now()`, two hook calls and a handful of adds, all
//! without allocating once a run is warm ([`reset`] pre-reserves). It is
//! therefore only safe at **coarse** boundaries; putting one inside a per-object
//! loop would measure itself. Every boundary this repository has is per
//! *request*, and `gunnar.stage_overhead` reports what they cost against the
//! same build with the probes compiled out.
#![deny(missing_docs)]
#![forbid(unsafe_code)]
use std::cell::{Cell, RefCell};
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Mutex;
use std::time::Instant;
// ── the switch ──────────────────────────────────────────────────────────────
static ENABLED: AtomicBool = AtomicBool::new(false);
/// Start recording. Off until something says otherwise.
pub fn enable() {
ENABLED.store(true, Ordering::Relaxed);
}
/// Stop recording. Stages already on the stack finish without being charged.
pub fn disable() {
ENABLED.store(false, Ordering::Relaxed);
}
/// Whether recording is on.
#[inline(always)]
pub fn enabled() -> bool {
ENABLED.load(Ordering::Relaxed)
}
// ── the hooks ───────────────────────────────────────────────────────────────
/// A per-thread CPU clock: `(user_ns, sys_ns)` consumed by **this thread**.
///
/// On Linux the only thing that answers it is `getrusage(RUSAGE_THREAD)`, which
/// needs `libc`; `/proc/self/task/<tid>/stat` answers in 10 ms clock ticks,
/// which is coarser than most of the stages here are long and would report a
/// tidy zero for all of them.
pub type CpuClock = fn() -> (u64, u64);
/// A resident-set probe, in bytes. Process-wide by nature — see
/// [`Tally::rss_growth_bytes`].
pub type RssProbe = fn() -> u64;
static CPU_CLOCK: Mutex<Option<CpuClock>> = Mutex::new(None);
static RSS_PROBE: Mutex<Option<RssProbe>> = Mutex::new(None);
thread_local! {
static CPU_CLOCK_T: Cell<Option<CpuClock>> = const { Cell::new(None) };
static RSS_PROBE_T: Cell<Option<RssProbe>> = const { Cell::new(None) };
static HOOKS_READ: Cell<bool> = const { Cell::new(false) };
}
/// Install the per-thread CPU clock. Call before [`enable`].
///
/// Threads that started earlier pick it up on their first [`reset`].
pub fn set_cpu_clock(f: CpuClock) {
*CPU_CLOCK.lock().expect("stage hook mutex") = Some(f);
HOOKS_READ.with(|h| h.set(false));
}
/// Install the resident-set probe. Call before [`enable`].
pub fn set_rss_probe(f: RssProbe) {
*RSS_PROBE.lock().expect("stage hook mutex") = Some(f);
HOOKS_READ.with(|h| h.set(false));
}
fn hooks() -> (Option<CpuClock>, Option<RssProbe>) {
if !HOOKS_READ.with(|h| h.get()) {
let cpu = *CPU_CLOCK.lock().expect("stage hook mutex");
let rss = *RSS_PROBE.lock().expect("stage hook mutex");
CPU_CLOCK_T.with(|c| c.set(cpu));
RSS_PROBE_T.with(|c| c.set(rss));
HOOKS_READ.with(|h| h.set(true));
}
(CPU_CLOCK_T.with(|c| c.get()), RSS_PROBE_T.with(|c| c.get()))
}
// ── the allocator's side ────────────────────────────────────────────────────
thread_local! {
static ALLOC_N: Cell<u64> = const { Cell::new(0) };
static ALLOC_B: Cell<u64> = const { Cell::new(0) };
static LIVE: Cell<i64> = const { Cell::new(0) };
static PEAK: Cell<i64> = const { Cell::new(0) };
}
/// Tell the accounting an allocation of `bytes` happened on this thread.
///
/// Meant to be called from a wrapping `#[global_allocator]`, which is why it
/// touches nothing but four `Cell`s: no `RefCell`, no map, no allocation of its
/// own, so it cannot re-enter the allocator that is calling it. That
/// re-entrancy is not hypothetical — a `RefCell<BTreeMap>` here deadlocks or
/// panics the first time a stage's own bookkeeping allocates.
#[inline]
pub fn note_alloc(bytes: usize) {
if !enabled() {
return;
}
ALLOC_N.with(|c| c.set(c.get().wrapping_add(1)));
ALLOC_B.with(|c| c.set(c.get().wrapping_add(bytes as u64)));
let live = LIVE.with(|c| {
let v = c.get() + bytes as i64;
c.set(v);
v
});
PEAK.with(|c| {
if live > c.get() {
c.set(live)
}
});
}
/// Tell the accounting a deallocation of `bytes` happened on this thread.
///
/// A block allocated on one thread and freed on another makes `live` drift
/// negative there and high here. That is real and is not smoothed over: it is
/// why [`Tally::retained_bytes`] is signed, and why the headline memory column
/// is [`Tally::peak_growth_bytes`], which only ever counts growth this thread
/// caused.
#[inline]
pub fn note_dealloc(bytes: usize) {
if !enabled() {
return;
}
LIVE.with(|c| c.set(c.get() - bytes as i64));
}
// ── the tally ───────────────────────────────────────────────────────────────
/// What one named stage cost.
///
/// `excl_*` is this stage with its children subtracted, so the `excl_*` columns
/// of a whole report sum to the whole request; `incl_*` is this stage with its
/// children included. Both are wanted: the first is what a fraction can be
/// taken of, the second is what a name means.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Tally {
/// How many times the stage was entered.
pub calls: u64,
/// Wall time with children subtracted.
pub excl_wall_ns: u64,
/// Wall time with children included.
pub incl_wall_ns: u64,
/// User CPU with children subtracted. Zero unless a [`CpuClock`] is
/// installed.
pub excl_user_ns: u64,
/// System CPU with children subtracted. Zero unless a [`CpuClock`] is
/// installed.
pub excl_sys_ns: u64,
/// Allocations with children subtracted. Zero unless the binary's allocator
/// calls [`note_alloc`].
pub excl_alloc_count: u64,
/// Allocated bytes with children subtracted — *churn*, not residency: a
/// stage that allocates and frees a gigabyte a page at a time reports a
/// gigabyte here and nothing in [`peak_growth_bytes`](Self::peak_growth_bytes).
pub excl_alloc_bytes: u64,
/// How much this stage, **children subtracted**, raised the thread's
/// high-water mark of live allocated bytes.
///
/// The memory headline, and exclusive for the same reason the wall column
/// is. Charged inclusively it double-counts through every nesting level —
/// MEASURED 2026-08-04 on a 12 181-object clone, where `select` held 14.48%
/// of a "peak" that was its own children's, and the fractions added to
/// 114%. It is monotone by construction: a later stage that allocates
/// heavily but never passes an earlier stage's peak reports zero, which is
/// the true statement about who owns the peak.
pub peak_growth_bytes: u64,
/// Live allocated bytes still held at exit, children subtracted. Signed:
/// a stage that frees more than it takes is a real and useful answer, and
/// the column sums to what the thread still holds.
pub retained_bytes: i64,
/// How much the **process** resident set grew across this stage, children
/// subtracted. Zero unless an [`RssProbe`] is installed.
///
/// Process-wide, so under concurrency it is every thread's growth attributed
/// to whichever one happened to be in this stage. Quotable single-threaded,
/// not quotable at 32 clients, and the arms that report it say which they
/// were.
pub rss_growth_bytes: u64,
}
impl Tally {
fn merge(&mut self, other: &Tally) {
self.calls += other.calls;
self.excl_wall_ns += other.excl_wall_ns;
self.incl_wall_ns += other.incl_wall_ns;
self.excl_user_ns += other.excl_user_ns;
self.excl_sys_ns += other.excl_sys_ns;
self.excl_alloc_count += other.excl_alloc_count;
self.excl_alloc_bytes += other.excl_alloc_bytes;
self.peak_growth_bytes += other.peak_growth_bytes;
self.retained_bytes += other.retained_bytes;
self.rss_growth_bytes += other.rss_growth_bytes;
}
}
/// A whole attribution: every stage that ran, by name.
pub type Report = BTreeMap<&'static str, Tally>;
// ── the per-thread machine ──────────────────────────────────────────────────
struct Frame {
slot: usize,
/// Start of the *current slice*: reset every time a child returns.
slice_wall: Instant,
slice_user: u64,
slice_sys: u64,
slice_alloc_n: u64,
slice_alloc_b: u64,
slice_peak: i64,
slice_live: i64,
slice_rss: u64,
/// Start of the whole call, for the one inclusive column there is.
incl_wall: Instant,
}
#[derive(Default)]
struct State {
stack: Vec<Frame>,
names: Vec<&'static str>,
tallies: Vec<Tally>,
}
impl State {
fn slot(&mut self, name: &'static str) -> usize {
// Pointer-first, because these are `&'static str` literals and the
// common case is the same literal every time.
if let Some(i) = self
.names
.iter()
.position(|n| std::ptr::eq(n.as_ptr(), name.as_ptr()) || *n == name)
{
return i;
}
self.names.push(name);
self.tallies.push(Tally::default());
self.names.len() - 1
}
}
thread_local! {
static STATE: RefCell<State> = RefCell::new(State::default());
}
static GLOBAL: Mutex<Option<Report>> = Mutex::new(None);
/// Clear this thread's table and pre-reserve it, so a warm run allocates
/// nothing inside a stage boundary and the allocation columns describe the
/// *work*, not the instrument.
pub fn reset() {
HOOKS_READ.with(|h| h.set(false));
ALLOC_N.with(|c| c.set(0));
ALLOC_B.with(|c| c.set(0));
LIVE.with(|c| c.set(0));
PEAK.with(|c| c.set(0));
STATE.with(|s| {
let mut s = s.borrow_mut();
s.stack.clear();
s.names.clear();
s.tallies.clear();
s.stack.reserve(32);
s.names.reserve(64);
s.tallies.reserve(64);
});
}
/// Clear the process-wide table too.
pub fn reset_global() {
*GLOBAL.lock().expect("stage global mutex") = None;
}
/// This thread's table so far.
pub fn thread_report() -> Report {
STATE.with(|s| {
let s = s.borrow();
s.names
.iter()
.copied()
.zip(s.tallies.iter().copied())
.collect()
})
}
/// Fold this thread's table into the process-wide one and clear it.
///
/// Called by each worker before it finishes. Nothing merges implicitly: a
/// thread that is still running holds numbers that are not yet a measurement of
/// anything.
pub fn flush_thread() {
let mine = thread_report();
let mut global = GLOBAL.lock().expect("stage global mutex");
let table = global.get_or_insert_with(Report::new);
for (name, tally) in mine {
table.entry(name).or_default().merge(&tally);
}
STATE.with(|s| {
let mut s = s.borrow_mut();
s.names.clear();
s.tallies.clear();
});
}
/// Everything every flushed thread recorded.
pub fn global_report() -> Report {
GLOBAL
.lock()
.expect("stage global mutex")
.clone()
.unwrap_or_default()
}
/// The innermost stage running on this thread, if any.
///
/// What an instrumented [`ObjectStore`](https://docs.rs/) wrapper asks so a
/// read can be charged to the stage that issued it — the fact that separates
/// "the fetch read 12 000 objects" from "the *tree walk* read 11 400 of them
/// and the pack writer re-read 600".
pub fn current() -> Option<&'static str> {
STATE.with(|s| {
let s = s.try_borrow().ok()?;
s.stack.last().map(|f| s.names[f.slot])
})
}
/// Enter a stage. Hold the returned guard for the stage's extent.
///
/// ```
/// let _s = gunnar_stage::enter("negotiate");
/// ```
///
/// **Bind it to a name.** `let _ = enter("x")` drops the guard immediately and
/// records a stage that took no time, which looks exactly like a stage that is
/// free.
#[inline]
#[must_use = "the stage ends when the guard is dropped; `let _ = enter(..)` measures nothing"]
pub fn enter(name: &'static str) -> Guard {
if !enabled() {
return Guard { active: false };
}
let (cpu, rss) = hooks();
let now = Instant::now();
let (user, sys) = cpu.map_or((0, 0), |f| f());
let alloc_n = ALLOC_N.with(|c| c.get());
let alloc_b = ALLOC_B.with(|c| c.get());
let live = LIVE.with(|c| c.get());
let peak = PEAK.with(|c| c.get());
let rss_now = rss.map_or(0, |f| f());
STATE.with(|s| {
let Ok(mut s) = s.try_borrow_mut() else {
return;
};
// Suspend the parent: close its current slice.
if let Some(top) = s.stack.last() {
let slot = top.slot;
let (w, u, y, an, ab) = (
now.saturating_duration_since(top.slice_wall).as_nanos() as u64,
user.saturating_sub(top.slice_user),
sys.saturating_sub(top.slice_sys),
alloc_n.wrapping_sub(top.slice_alloc_n),
alloc_b.wrapping_sub(top.slice_alloc_b),
);
let (pk, lv, rs) = (
(peak - top.slice_peak).max(0) as u64,
live - top.slice_live,
rss_now.saturating_sub(top.slice_rss),
);
let t = &mut s.tallies[slot];
t.excl_wall_ns += w;
t.excl_user_ns += u;
t.excl_sys_ns += y;
t.excl_alloc_count += an;
t.excl_alloc_bytes += ab;
t.peak_growth_bytes += pk;
t.retained_bytes += lv;
t.rss_growth_bytes += rs;
}
let slot = s.slot(name);
s.tallies[slot].calls += 1;
s.stack.push(Frame {
slot,
slice_wall: now,
slice_user: user,
slice_sys: sys,
slice_alloc_n: alloc_n,
slice_alloc_b: alloc_b,
slice_peak: peak,
slice_live: live,
slice_rss: rss_now,
incl_wall: now,
});
});
Guard { active: true }
}
/// Ends a stage when dropped. See [`enter`].
pub struct Guard {
active: bool,
}
impl Drop for Guard {
fn drop(&mut self) {
if !self.active {
return;
}
let (cpu, rss) = hooks();
let now = Instant::now();
let (user, sys) = cpu.map_or((0, 0), |f| f());
let alloc_n = ALLOC_N.with(|c| c.get());
let alloc_b = ALLOC_B.with(|c| c.get());
let live = LIVE.with(|c| c.get());
let peak = PEAK.with(|c| c.get());
let rss_now = rss.map_or(0, |f| f());
STATE.with(|s| {
let Ok(mut s) = s.try_borrow_mut() else {
return;
};
let Some(frame) = s.stack.pop() else { return };
let t = &mut s.tallies[frame.slot];
t.excl_wall_ns += now.saturating_duration_since(frame.slice_wall).as_nanos() as u64;
t.excl_user_ns += user.saturating_sub(frame.slice_user);
t.excl_sys_ns += sys.saturating_sub(frame.slice_sys);
t.excl_alloc_count += alloc_n.wrapping_sub(frame.slice_alloc_n);
t.excl_alloc_bytes += alloc_b.wrapping_sub(frame.slice_alloc_b);
t.peak_growth_bytes += (peak - frame.slice_peak).max(0) as u64;
t.retained_bytes += live - frame.slice_live;
t.rss_growth_bytes += rss_now.saturating_sub(frame.slice_rss);
t.incl_wall_ns += now.saturating_duration_since(frame.incl_wall).as_nanos() as u64;
// Resume the parent.
if let Some(parent) = s.stack.last_mut() {
parent.slice_wall = now;
parent.slice_user = user;
parent.slice_sys = sys;
parent.slice_alloc_n = alloc_n;
parent.slice_alloc_b = alloc_b;
parent.slice_peak = peak;
parent.slice_live = live;
parent.slice_rss = rss_now;
}
});
}
}
/// Exclusive wall time of every stage in a report, in nanoseconds.
///
/// The denominator every fraction in a report is taken over, and the one number
/// that makes the fractions add to 100% by construction rather than by hoping.
pub fn total_excl_wall_ns(report: &Report) -> u64 {
report.values().map(|t| t.excl_wall_ns).sum()
}
/// Exclusive CPU (user + sys) of every stage in a report, in nanoseconds.
pub fn total_excl_cpu_ns(report: &Report) -> u64 {
report
.values()
.map(|t| t.excl_user_ns + t.excl_sys_ns)
.sum()
}