diff --git a/crates/memtrack/AGENTS.md b/crates/memtrack/AGENTS.md index 6c7b09d37..78bf8d0b3 100644 --- a/crates/memtrack/AGENTS.md +++ b/crates/memtrack/AGENTS.md @@ -84,7 +84,7 @@ sudo -E cargo test --test c_tests -- --test-threads 1 - **Build toolchain:** `clang` + BTF/vmlinux headers, `libbpf-dev`, `zlib1g-dev`, `pkgconf`, `build-essential`; vendored libbpf also needs `autopoint`/`bison`/`flex`. - `vmlinux.h` is pinned to a specific git rev; `libbpf-rs` uses the `vendored` feature (dist links `libbpf-rs/static`). -Env vars actually wired: `CODSPEED_MEMTRACK_BINARIES` (extra static-allocator binaries), `CODSPEED_MEMTRACK_TRACK_ALLOCATORS` (0/false disables), `CODSPEED_MEMTRACK_TRACK_PHYSICAL` (1 enables), `CODSPEED_MEMTRACK_CAPTURE_STACKS` (1 enables), `CODSPEED_MEMTRACK_STACK_BUDGET` (stack copy size in bytes, default 8192), `CODSPEED_LOG` (log filter, default `info`), `SUDO_UID`/`SUDO_GID` (privilege drop), `GITHUB_ACTIONS` (build rebuild trigger + test gate). +Env vars actually wired: `CODSPEED_MEMTRACK_BINARIES` (extra static-allocator binaries), `CODSPEED_MEMTRACK_TRACK_ALLOCATORS` (0/false disables), `CODSPEED_MEMTRACK_TRACK_PHYSICAL` (1 enables), `CODSPEED_MEMTRACK_CAPTURE_STACKS` (1 enables), `CODSPEED_MEMTRACK_STACK_BUDGET` (stack copy size in bytes, default 8192), `CODSPEED_MEMTRACK_STATS` (absolute path; writes per-tick ring positions, pressure episodes, pipeline backlog + RSS, stack-resolver batches and encoder windows as JSONL, plotted by `scripts/plot_stats.py`), `CODSPEED_LOG` (log filter, default `info`), `SUDO_UID`/`SUDO_GID` (privilege drop), `GITHUB_ACTIONS` (build rebuild trigger + test gate). ### Minimum kernel version diff --git a/crates/memtrack/scripts/plot_stats.py b/crates/memtrack/scripts/plot_stats.py new file mode 100644 index 000000000..cce3d482c --- /dev/null +++ b/crates/memtrack/scripts/plot_stats.py @@ -0,0 +1,199 @@ +# /// script +# requires-python = ">=3.11" +# dependencies = ["polars", "matplotlib"] +# /// +"""Plot memtrack pipeline stats written via CODSPEED_MEMTRACK_STATS.""" + +import argparse +import json +import sys +from pathlib import Path + +import matplotlib + +matplotlib.use("Agg") +import matplotlib.pyplot as plt +import polars as pl + +MB = 1e6 +MIB = 1024 * 1024 +BIN_NS = 100_000_000 +TIME_COLS = ("t", "t0", "t1", "stopped_at") + + +def load(path: Path) -> dict[str, pl.DataFrame]: + rows = [] + for n, line in enumerate(path.read_text().splitlines(), 1): + try: + rows.append(json.loads(line)) + except json.JSONDecodeError: + # A killed run can leave a cut-off last line. + print(f"{path}:{n}: skipping unparsable line", file=sys.stderr) + if not any(r["k"] == "ring" for r in rows): + sys.exit(f"{path}: no ring records") + t_min = min(r.get("t", r.get("t0", 0)) for r in rows) + frames = {} + for kind in ("ring_open", "ring", "backlog", "resolve", "encode"): + kind_rows = [r for r in rows if r["k"] == kind] + frames[kind] = pl.DataFrame(kind_rows).drop("k") if kind_rows else pl.DataFrame() + pressure = [ + {"ring": r["ring"], "t": r["t"], "pid": pid, "stopped_at": at} + for r in rows + if r["k"] == "pressure" + for pid, at in r["pids"] + ] + schema = {"ring": pl.Utf8, "t": pl.Int64, "pid": pl.Int64, "stopped_at": pl.Int64} + frames["pressure"] = pl.DataFrame(pressure, schema=schema) + for kind, df in frames.items(): + if kind != "ring_open" and len(df): + frames[kind] = df.with_columns(pl.col(c) - t_min for c in TIME_COLS if c in df.columns) + return frames + + +def episodes(pressure: pl.DataFrame) -> pl.DataFrame: + """One row per released pid; episode bounds are shared by all pids released together.""" + return pressure.with_columns(start=pl.col("stopped_at").min().over("ring", "t"), end=pl.col("t")) + + +def write_rate(r: pl.DataFrame) -> pl.DataFrame: + # Positions are cumulative bytes; spread each delta over the real gap since + # the previous sample, since samples are sparse when the ring is idle. + r = r.sort("t1") + dt = pl.col("t1").diff() + return r.select("t1", mbps=pl.col("prod1").diff() / MB / (dt / 1e9)).filter(dt > 0) + + +def drain_rate(r: pl.DataFrame) -> pl.DataFrame: + # Aggregated per bin: ticks of a few us give meaningless per-tick ratios. + return ( + r.select(t1=pl.col("t1") // BIN_NS * BIN_NS, bytes=pl.col("cons1") - pl.col("cons0"), busy=pl.col("t1") - pl.col("t0")) + .group_by("t1").agg(pl.col("bytes", "busy").sum()).sort("t1") + .filter(pl.col("busy") > 0) + .select("t1", mbps=pl.col("bytes") / MB / (pl.col("busy") / 1e9)) + ) + + +def busy_pct(intervals: pl.DataFrame) -> pl.DataFrame: + """Share of each bin a thread spent inside short (t0, t1) work intervals.""" + return ( + intervals.select(t=pl.col("t1") // BIN_NS * BIN_NS, busy=pl.col("t1") - pl.col("t0")) + .group_by("t").agg(pl.col("busy").sum()).sort("t") + .select("t", pct=100 * pl.col("busy") / BIN_NS) + ) + + +def encoder_window_ns() -> pl.Expr: + # Windows run back to back, so wait + encode + write is the wall time each one covers. + return pl.col("wait_ns") + pl.col("encode_ns") + pl.col("write_ns") + + +def plot(f: dict[str, pl.DataFrame], eps: pl.DataFrame, out: Path) -> None: + sizes = dict(zip(f["ring_open"]["ring"], f["ring_open"]["size"])) if len(f["ring_open"]) else {} + rings, enc, backlog, resolve = f["ring"], f["encode"], f["backlog"], f["resolve"] + n = 4 + bool(len(eps)) + fig, axes = plt.subplots(n, 1, sharex=True, figsize=(14, 3.2 * n)) + ax_fill, ax_tp, ax_busy, ax_backlog = axes[:4] + + # One color per ring across every panel. + for i, name in enumerate(sorted(rings["ring"].unique())): + r, color = rings.filter(pl.col("ring") == name).sort("t0"), f"C{i}" + if size := sizes.get(name): + xs = [v for a, b in zip(r["t0"], r["t1"]) for v in (a, b)] + ys = [v for a, b in zip(r["prod0"] - r["cons0"], r["prod1"] - r["cons1"]) for v in (a, b)] + ax_fill.plot([x / 1e9 for x in xs], [100 * y / size for y in ys], color=color, lw=0.8, label=name) + w, d = write_rate(r), drain_rate(r) + ax_tp.step(w["t1"] / 1e9, w["mbps"], where="pre", color=color, lw=0.8, label=f"{name} ring write") + ax_tp.plot(d["t1"] / 1e9, d["mbps"], color=color, lw=0.8, ls=":", label=f"{name} drain (while busy)") + b = busy_pct(r) + ax_busy.step(b["t"] / 1e9, b["pct"], where="post", color=color, lw=0.8, label=f"{name} poller") + ax_fill.axhline(75, ls="--", color="gray") + for e in eps.unique(["ring", "t"]).iter_rows(named=True): + for ax in axes: + ax.axvspan(e["start"] / 1e9, e["end"] / 1e9, color="red", alpha=0.08, lw=0) + ax_fill.set_ylabel("ring fill %") + + if len(enc): + # A window can span seconds, so draw each one across the time it covers. + e = enc.with_columns(window=encoder_window_ns()) + start, end = (e["t"] - e["window"]) / 1e9, e["t"] / 1e9 + for col, label, color in (("msgpack_bytes", "encoder in (msgpack)", "C6"), ("zstd_bytes", "encoder out (zstd, disk)", "C7")): + ax_tp.hlines(e[col] / MB / (e["window"] / 1e9), start, end, color=color, lw=2, label=label) + busy = 100 * (e["encode_ns"] + e["write_ns"]) / e["window"] + ax_busy.hlines(busy, start, end, color="C6", lw=2, label="encoder (encode + write)") + if len(resolve): + b = busy_pct(resolve) + ax_busy.step(b["t"] / 1e9, b["pct"], where="post", color="C5", lw=0.8, label="stack resolver") + ax_tp.set_yscale("log") + ax_tp.set_ylabel("MB/s (log)") + ax_busy.set_ylabel("stage busy %") + ax_busy.set_ylim(0, 105) + + if len(backlog): + depth = (backlog["sent"] - backlog["received"]) / 1e6 + ax_backlog.plot(backlog["t"] / 1e9, depth, color="C2", lw=1, label="events in flight") + ax_backlog.set_ylabel("M events in flight", color="C2") + ax_rss = ax_backlog.twinx() + ax_rss.plot(backlog["t"] / 1e9, backlog["rss"] / MIB, color="gray", lw=1, ls=":") + ax_rss.set_ylabel("memtrack RSS (MiB)", color="gray") + + if len(eps): + ax = axes[4] + pids = sorted(eps["pid"].unique()) + for e in eps.iter_rows(named=True): + ax.barh(pids.index(e["pid"]), (e["end"] - e["stopped_at"]) / 1e9, left=e["stopped_at"] / 1e9, color="C4") + ax.set_yticks(range(len(pids)), [str(p) for p in pids]) + ax.set_ylabel("paused pid") + for ax in axes: + if ax.get_legend_handles_labels()[0]: + ax.legend(loc="upper right", fontsize="small") + axes[-1].set_xlabel("seconds") + fig.tight_layout() + fig.savefig(out, dpi=120) + + +def summary(f: dict[str, pl.DataFrame], eps: pl.DataFrame) -> None: + sizes = dict(zip(f["ring_open"]["ring"], f["ring_open"]["size"])) if len(f["ring_open"]) else {} + print(f"{'ring':<16} {'MB':>9} {'wr avg':>8} {'wr peak':>8} {'dr avg':>8} {'dr peak':>8}" + f" {'fill%':>6} {'busy%':>6} {'eps':>4} {'paused ms':>10} {'max ms':>8}") + for name, r in f["ring"].group_by("ring", maintain_order=True): + name = name[0] + span = max(r["t1"].max() - r["t0"].min(), 1) + written = r["prod1"].max() - r["prod0"].min() + w, d = write_rate(r), drain_rate(r) + size = sizes.get(name) + fill = 100 * max((r["prod0"] - r["cons0"]).max(), (r["prod1"] - r["cons1"]).max()) / size if size else float("nan") + busy = 100 * (r["t1"] - r["t0"]).sum() / span + e = eps.filter(pl.col("ring") == name) + paused = (e["end"] - e["stopped_at"]) / 1e6 + print(f"{name:<16} {written / MB:>9.1f} {written / MB / (span / 1e9):>8.1f} {w['mbps'].max() or 0:>8.1f}" + f" {d['mbps'].mean() or 0:>8.1f} {d['mbps'].max() or 0:>8.1f} {fill:>6.1f} {busy:>6.1f}" + f" {e.unique('t').height:>4} {paused.sum():>10.1f} {paused.max() or 0:>8.1f}") + + if len(enc := f["encode"]): + wall = enc.select(encoder_window_ns().sum()).item() + msgpack, zstd = enc["msgpack_bytes"].sum(), enc["zstd_bytes"].sum() + print(f"encoder: {len(enc)} windows, {enc['events'].sum()} events, {msgpack / MB:.1f} MB msgpack ->" + f" {zstd / MB:.1f} MB zstd ({msgpack / max(zstd, 1):.1f}x), {msgpack / MB / (wall / 1e9):.1f} MB/s in," + f" busy {100 * (enc['encode_ns'].sum() + enc['write_ns'].sum()) / wall:.1f}%") + if len(res := f["resolve"]): + span = max(res["t1"].max() - res["t0"].min(), 1) + print(f"resolver: {len(res)} batches, {res['n'].sum()} stacks, busy {100 * (res['t1'] - res['t0']).sum() / span:.1f}%") + if len(bl := f["backlog"]): + depth = bl["sent"] - bl["received"] + print(f"backlog: peak {depth.max()} events in flight, peak RSS {bl['rss'].max() / MIB:.1f} MiB") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("stats", type=Path) + parser.add_argument("-o", "--output", type=Path, default=Path("stats.png")) + args = parser.parse_args() + frames = load(args.stats) + eps = episodes(frames["pressure"]) + plot(frames, eps, args.output) + summary(frames, eps) + print(f"wrote {args.output}") + + +if __name__ == "__main__": + main() diff --git a/crates/memtrack/src/ebpf/c/utils/pressure.bpf.h b/crates/memtrack/src/ebpf/c/utils/pressure.bpf.h index e0ae4a5ae..07706ce35 100644 --- a/crates/memtrack/src/ebpf/c/utils/pressure.bpf.h +++ b/crates/memtrack/src/ebpf/c/utils/pressure.bpf.h @@ -31,8 +31,8 @@ static __always_inline void memtrack_check_ring_pressure(void* ring, __u32 curre return; } - __u8 marker = 1; - if (bpf_map_update_elem(&pressure_stopped, ¤t_tgid, &marker, BPF_ANY) != 0) { + __u64 stopped_at = bpf_ktime_get_ns(); + if (bpf_map_update_elem(&pressure_stopped, ¤t_tgid, &stopped_at, BPF_ANY) != 0) { return; } bpf_send_signal(MEMTRACK_SIGSTOP); diff --git a/crates/memtrack/src/ebpf/c/utils/stopped.h b/crates/memtrack/src/ebpf/c/utils/stopped.h index 1efe0ad58..885fbee50 100644 --- a/crates/memtrack/src/ebpf/c/utils/stopped.h +++ b/crates/memtrack/src/ebpf/c/utils/stopped.h @@ -5,11 +5,12 @@ #define MEMTRACK_SIGSTOP 19 -/* tgid -> 1 for every process BPF stopped, one map per reason. A process is - * only stopped once it is recorded, and userspace resumes only recorded - * processes, so a process stopped for both reasons resumes once neither map - * holds it. Sized like tracked_pids. */ -BPF_HASH_MAP(pressure_stopped, __u32, __u8, 10000); +/* tgid -> stop record for every process BPF stopped, one map per reason: + * pressure_stopped holds the stop ktime (ns), attach_stopped a 1 marker. A + * process is only stopped once it is recorded, and userspace resumes only + * recorded processes, so a process stopped for both reasons resumes once + * neither map holds it. Sized like tracked_pids. */ +BPF_HASH_MAP(pressure_stopped, __u32, __u64, 10000); BPF_HASH_MAP(attach_stopped, __u32, __u8, 10000); #endif /* __STOPPED_H__ */ diff --git a/crates/memtrack/src/ebpf/memtrack/maps.rs b/crates/memtrack/src/ebpf/memtrack/maps.rs index 6424eee25..74f7e7541 100644 --- a/crates/memtrack/src/ebpf/memtrack/maps.rs +++ b/crates/memtrack/src/ebpf/memtrack/maps.rs @@ -91,10 +91,10 @@ impl MemtrackBpf { } /// Callback that resumes every pressure-stopped process. - pub(super) fn on_ring_drained(&self) -> Box { + pub(super) fn on_ring_drained(&self) -> crate::ebpf::poller::OnDrained { let stopped = self.stopped.clone(); - Box::new(move || { - if let Err(error) = stopped.release_pressure() { + Box::new(move |ring| { + if let Err(error) = stopped.release_pressure(ring) { error!("failed to release pressure-stopped producers: {error:#}"); } }) diff --git a/crates/memtrack/src/ebpf/memtrack/mod.rs b/crates/memtrack/src/ebpf/memtrack/mod.rs index dbe4680ad..b2ea068ed 100644 --- a/crates/memtrack/src/ebpf/memtrack/mod.rs +++ b/crates/memtrack/src/ebpf/memtrack/mod.rs @@ -243,9 +243,14 @@ impl MemtrackBpf { poll_interval_ms: u64, tx: std::sync::mpsc::Sender>, ) -> Result { + let parse = |data: &[u8]| { + let event = crate::ebpf::events::parse_event(data)?; + crate::ebpf::stats::add_sent(1); + Some(event) + }; with_skel!(self, skel => RingBufferPoller::new( &skel.maps.events, - crate::ebpf::events::parse_event, + parse, tx, poll_interval_ms, None, @@ -276,9 +281,15 @@ impl MemtrackBpf { event }; + let parse = |data: &[u8]| { + let stack = events::parse_stack(data)?; + crate::ebpf::stats::add_sent(1); + Some(stack) + }; + with_skel!(self, skel => ThreadedRingBufferPoller::new( &skel.maps.stacks, - events::parse_stack, + parse, resolve, tx, poll_interval_ms, diff --git a/crates/memtrack/src/ebpf/mod.rs b/crates/memtrack/src/ebpf/mod.rs index 3aba823ac..76c726222 100644 --- a/crates/memtrack/src/ebpf/mod.rs +++ b/crates/memtrack/src/ebpf/mod.rs @@ -6,6 +6,7 @@ pub(crate) mod poller; mod proc_fs; mod spawn; mod stacks; +pub mod stats; mod tracker; pub use memtrack::{ diff --git a/crates/memtrack/src/ebpf/pause.rs b/crates/memtrack/src/ebpf/pause.rs index a3c7a17f7..1adbd121f 100644 --- a/crates/memtrack/src/ebpf/pause.rs +++ b/crates/memtrack/src/ebpf/pause.rs @@ -1,3 +1,4 @@ +use crate::ebpf::stats; use crate::prelude::*; use libbpf_rs::{MapCore, MapFlags, MapHandle}; use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; @@ -34,13 +35,15 @@ impl StoppedProcesses { } /// Resume every pressure-stopped producer; call once a ring is flushed. - pub(crate) fn release_pressure(&self) -> Result<()> { + pub(crate) fn release_pressure(&self, ring: &str) -> Result<()> { // Deleting while iterating restarts hash iteration, so snapshot the keys first. let keys: Vec> = self.pressure_stopped.keys().collect(); if keys.is_empty() { return Ok(()); } self.pressure_stops.fetch_add(keys.len() as u64, Relaxed); + let record_stats = stats::enabled(); + let mut stopped_at = Vec::new(); for key in keys { let pid = u32::from_le_bytes( key.as_slice() @@ -48,8 +51,23 @@ impl StoppedProcesses { .context("Invalid pressure_stopped key size")?, ); debug!("Releasing pressure stop of pid {pid}"); + // Read the stop time before release deletes the entry; a missing + // entry means the pid already exited. + if record_stats + && let Some(value) = self.pressure_stopped.lookup(&key, MapFlags::ANY)? + && let Ok(bytes) = <[u8; 8]>::try_from(value.as_slice()) + { + stopped_at.push((pid, u64::from_le_bytes(bytes))); + } Self::release(pid, &self.pressure_stopped, &self.attach_stopped)?; } + if record_stats { + stats::emit(&stats::Record::Pressure { + ring, + t: stats::now_ns(), + pids: &stopped_at, + }); + } Ok(()) } diff --git a/crates/memtrack/src/ebpf/poller.rs b/crates/memtrack/src/ebpf/poller.rs index 637ed27e7..8439ffa97 100644 --- a/crates/memtrack/src/ebpf/poller.rs +++ b/crates/memtrack/src/ebpf/poller.rs @@ -1,3 +1,4 @@ +use crate::ebpf::stats::{self, RingSampler}; use anyhow::{Context, Result}; use libbpf_rs::{AsRawLibbpf, MapCore, RingBuffer, RingBufferBuilder, libbpf_sys}; use parking_lot::Mutex; @@ -9,6 +10,9 @@ use std::time::Duration; /// Ring-buffer poll interval shared by every poller. pub(crate) const POLL_INTERVAL_MS: u64 = 1; +/// Called with the ring's map name each time the poller finds the ring empty. +pub(crate) type OnDrained = Box; + /// Items buffered before a channel send. `std::sync::mpsc` allocates a block /// every 31 messages, so sending one item at a time makes that allocation /// dominate the pipeline; batching amortizes it over a whole batch. @@ -69,6 +73,11 @@ fn poll_iteration( } } +fn ring_of(ringbuf: &RingBuffer) -> *mut libbpf_sys::ring { + // SAFETY: a built `RingBuffer` holds exactly the one ring added in `new`. + unsafe { libbpf_sys::ring_buffer__ring(ringbuf.as_libbpf_object().as_ptr(), 0) } +} + /// Polls a BPF ring buffer in a background thread, parsing raw entries with a /// user-supplied closure and forwarding them to an mpsc channel in batches. /// @@ -85,7 +94,7 @@ impl RingBufferPoller { parse: F, tx: Sender>, poll_interval_ms: u64, - on_drained: Option>, + on_drained: Option, ) -> Result where M: MapCore, @@ -114,32 +123,40 @@ impl RingBufferPoller { 0 })?; let ringbuf = builder.build()?; + let name = rb_map.name().to_string_lossy().into_owned(); // The control channel doubles as the poll pacing: a received message is // a drain request (acked after a full consume), a timeout is a regular // poll tick, and disconnection is the shutdown signal. let (ctl, ctl_rx) = mpsc::channel::>(); let poll_thread = std::thread::spawn(move || { - // SAFETY: the built `RingBuffer` holds exactly the one ring added above. - let ring = - unsafe { libbpf_sys::ring_buffer__ring(ringbuf.as_libbpf_object().as_ptr(), 0) }; - while poll_iteration( - ctl_rx.recv_timeout(Duration::from_millis(poll_interval_ms)), - || consume_all(&ringbuf, ring), - || { - let _ = ringbuf.poll(Duration::ZERO); - }, - &batch, - &tx, - ) { + let mut sampler = RingSampler::new(name.clone(), ring_of(&ringbuf)); + loop { + let control = ctl_rx.recv_timeout(Duration::from_millis(poll_interval_ms)); + let tick = sampler.as_ref().map(RingSampler::begin); + let running = poll_iteration( + control, + || consume_all(&ringbuf, ring_of(&ringbuf)), + || { + let _ = ringbuf.poll(Duration::ZERO); + }, + &batch, + &tx, + ); + if let (Some(sampler), Some(tick)) = (&mut sampler, tick) { + sampler.end(tick); + } + if !running { + break; + } if let Some(on_drained) = &on_drained - && unsafe { libbpf_sys::ring__avail_data_size(ring) } == 0 + && unsafe { libbpf_sys::ring__avail_data_size(ring_of(&ringbuf)) } == 0 { - on_drained(); + on_drained(&name); } } if let Some(on_drained) = &on_drained { - on_drained(); + on_drained(&name); } }); @@ -192,7 +209,7 @@ impl ThreadedRingBufferPoller { resolve: R, tx: Sender>, poll_interval_ms: u64, - on_drained: Option>, + on_drained: Option, ) -> Result where M: MapCore, @@ -204,8 +221,18 @@ impl ThreadedRingBufferPoller { let (parsed_tx, parsed_rx) = mpsc::channel::>(); let ring = RingBufferPoller::new(rb_map, parse, parsed_tx, poll_interval_ms, on_drained)?; let resolver = std::thread::spawn(move || { + let record_stats = stats::enabled(); for batch in parsed_rx { + let t0 = record_stats.then(stats::now_ns); + let n = batch.len(); let resolved = batch.into_iter().map(&resolve).collect(); + if let Some(t0) = t0 { + stats::emit(&stats::Record::Resolve { + t0, + t1: stats::now_ns(), + n, + }); + } let _ = tx.send(resolved); } }); diff --git a/crates/memtrack/src/ebpf/stats.rs b/crates/memtrack/src/ebpf/stats.rs new file mode 100644 index 000000000..d5186a1a9 --- /dev/null +++ b/crates/memtrack/src/ebpf/stats.rs @@ -0,0 +1,264 @@ +//! Pipeline samples as JSON lines, enabled with `CODSPEED_MEMTRACK_STATS=`. +//! +//! Only raw counters are recorded; rates and fill levels are derived offline +//! by `scripts/plot_stats.py`. Every `t*` is CLOCK_MONOTONIC ns, the clock of +//! `bpf_ktime_get_ns()` and of the artifact's event timestamps. + +use crate::prelude::*; +use libbpf_rs::libbpf_sys; +use parking_lot::Mutex; +use runner_shared::artifacts::WindowStats; +use serde::Serialize; +use std::fs::File; +use std::io::{BufWriter, Write}; +use std::path::PathBuf; +use std::sync::OnceLock; +use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; +use std::sync::mpsc::{self, RecvTimeoutError, Sender}; +use std::thread::JoinHandle; +use std::time::Duration; + +/// Emptied on the first write error, so a full disk stops sampling instead of +/// logging on every tick. +static SINK: OnceLock>>> = OnceLock::new(); + +/// Events parsed from the rings vs. taken by the encoder. The difference is +/// everything in flight between them: poll batches, the resolver queue and +/// the unbounded encoder channel. +static SENT: AtomicU64 = AtomicU64::new(0); +static RECEIVED: AtomicU64 = AtomicU64::new(0); + +/// Poll-independent so the backlog stays visible under slow poll intervals. +const BACKLOG_INTERVAL: Duration = Duration::from_millis(10); +static BACKLOG_THREAD: Mutex, JoinHandle<()>)>> = Mutex::new(None); + +#[derive(Serialize)] +#[serde(tag = "k", rename_all = "snake_case")] +pub(crate) enum Record<'a> { + RingOpen { + t: u64, + ring: &'a str, + size: u64, + }, + Ring { + ring: &'a str, + t0: u64, + prod0: u64, + cons0: u64, + t1: u64, + prod1: u64, + cons1: u64, + }, + Pressure { + ring: &'a str, + t: u64, + pids: &'a [(u32, u64)], + }, + /// `rss` is memtrack's own resident set in bytes. + Backlog { + t: u64, + sent: u64, + received: u64, + rss: u64, + }, + /// One stack-resolver batch of `n` records. + Resolve { + t0: u64, + t1: u64, + n: usize, + }, + /// One encoder window, emitted after its frames are written at `t`. + Encode { + t: u64, + events: usize, + wait_ns: u64, + encode_ns: u64, + write_ns: u64, + msgpack_bytes: u64, + zstd_bytes: u64, + }, +} + +pub fn init_from_env() -> Result<()> { + let Some(path) = std::env::var_os("CODSPEED_MEMTRACK_STATS").map(PathBuf::from) else { + return Ok(()); + }; + let file = File::create(&path) + .with_context(|| format!("Failed to create memtrack stats file {}", path.display()))?; + SINK.set(Mutex::new(Some(BufWriter::new(file)))) + .map_err(|_| anyhow!("memtrack stats already initialized"))?; + info!("Writing memtrack stats to {}", path.display()); + + let (stop, stopped) = mpsc::channel::<()>(); + let thread = std::thread::spawn(move || { + while stopped.recv_timeout(BACKLOG_INTERVAL) == Err(RecvTimeoutError::Timeout) { + emit_backlog(); + } + emit_backlog(); + }); + *BACKLOG_THREAD.lock() = Some((stop, thread)); + Ok(()) +} + +pub fn finish() -> Result<()> { + if let Some((stop, thread)) = BACKLOG_THREAD.lock().take() { + drop(stop); + let _ = thread.join(); + } + let Some(mut out) = SINK.get().and_then(|sink| sink.lock().take()) else { + return Ok(()); + }; + out.flush().context("Failed to flush memtrack stats") +} + +pub(crate) fn add_sent(events: usize) { + if enabled() { + SENT.fetch_add(events as u64, Relaxed); + } +} + +pub fn add_received(events: usize) { + if enabled() { + RECEIVED.fetch_add(events as u64, Relaxed); + } +} + +pub fn encoder_window(window: &WindowStats) { + emit(&Record::Encode { + t: now_ns(), + events: window.events, + wait_ns: window.wait.as_nanos() as u64, + encode_ns: window.encode.as_nanos() as u64, + write_ns: window.write.as_nanos() as u64, + msgpack_bytes: window.msgpack_bytes, + zstd_bytes: window.zstd_bytes, + }); +} + +fn emit_backlog() { + emit(&Record::Backlog { + t: now_ns(), + sent: SENT.load(Relaxed), + received: RECEIVED.load(Relaxed), + rss: self_rss_bytes(), + }); +} + +/// 0 if `/proc/self/statm` is unreadable; the sample is diagnostic only. +fn self_rss_bytes() -> u64 { + let Ok(statm) = std::fs::read_to_string("/proc/self/statm") else { + return 0; + }; + let pages: u64 = statm + .split_whitespace() + .nth(1) + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + // SAFETY: sysconf has no preconditions. + pages * unsafe { libc::sysconf(libc::_SC_PAGESIZE) } as u64 +} + +pub(crate) fn enabled() -> bool { + SINK.get().is_some() +} + +pub(crate) fn emit(record: &Record) { + let Some(sink) = SINK.get() else { + return; + }; + let mut sink = sink.lock(); + let Some(out) = sink.as_mut() else { + return; + }; + let written = serde_json::to_writer(&mut *out, record) + .map_err(std::io::Error::from) + .and_then(|()| out.write_all(b"\n")); + if let Err(error) = written { + error!("Stopping memtrack stats after a write error: {error}"); + *sink = None; + } +} + +pub(crate) fn now_ns() -> u64 { + let mut ts = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + // SAFETY: `ts` is a valid out-pointer. + unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) }; + ts.tv_sec as u64 * 1_000_000_000 + ts.tv_nsec as u64 +} + +/// Positions of one ring around each poll tick, from libbpf's mmapped +/// producer/consumer pages. The producer position counts every reserved byte, +/// including 8-byte record headers and records BPF later discarded, so it is +/// ring pressure rather than artifact bytes. +pub(crate) struct RingSampler { + name: String, + ring: *const libbpf_sys::ring, + last: (u64, u64), +} + +pub(crate) struct Tick { + t: u64, + prod: u64, + cons: u64, +} + +impl RingSampler { + /// `None` unless stats are enabled. `ring` must outlive the sampler. + pub(crate) fn new(name: String, ring: *const libbpf_sys::ring) -> Option { + if !enabled() { + return None; + } + // SAFETY: `ring` is valid per this function's contract. + let size = unsafe { libbpf_sys::ring__size(ring) } as u64; + emit(&Record::RingOpen { + t: now_ns(), + ring: &name, + size, + }); + let mut sampler = Self { + name, + ring, + last: (0, 0), + }; + sampler.last = sampler.positions(); + Some(sampler) + } + + pub(crate) fn begin(&self) -> Tick { + let t = now_ns(); + let (prod, cons) = self.positions(); + Tick { t, prod, cons } + } + + /// Positions only grow, so equal end positions mean nothing was written or + /// read since the last emitted tick. + pub(crate) fn end(&mut self, tick: Tick) { + let (prod1, cons1) = self.positions(); + if (prod1, cons1) == self.last { + return; + } + self.last = (prod1, cons1); + emit(&Record::Ring { + ring: &self.name, + t0: tick.t, + prod0: tick.prod, + cons0: tick.cons, + t1: now_ns(), + prod1, + cons1, + }); + } + + fn positions(&self) -> (u64, u64) { + // SAFETY: `ring` is valid per `new`'s contract. + unsafe { + ( + libbpf_sys::ring__producer_pos(self.ring), + libbpf_sys::ring__consumer_pos(self.ring), + ) + } + } +} diff --git a/crates/memtrack/src/main.rs b/crates/memtrack/src/main.rs index 5ff58cb41..1806f73e5 100644 --- a/crates/memtrack/src/main.rs +++ b/crates/memtrack/src/main.rs @@ -4,7 +4,7 @@ static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc; use clap::Parser; use ipc_channel::ipc; use memtrack::prelude::*; -use memtrack::{MemtrackIpcMessage, Tracker, handle_ipc_message}; +use memtrack::{MemtrackIpcMessage, Tracker, handle_ipc_message, stats}; use runner_shared::artifacts::{ArtifactExt, MemtrackArtifact, encode_events}; use std::path::{Path, PathBuf}; use std::process::Command; @@ -87,6 +87,7 @@ fn track_command( None }; + stats::init_from_env()?; let tracker = Arc::new(Tracker::new()?); // Spawn IPC handler thread with the now-available tracker @@ -133,8 +134,13 @@ fn track_command( .map(|n| n.get().saturating_sub(2).max(1)) .unwrap_or(4); - let pipeline_thread = - thread::spawn(move || encode_events(event_rx.into_iter().flatten(), out_file, n_workers)); + let pipeline_thread = thread::spawn(move || { + let events = event_rx + .into_iter() + .inspect(|batch| stats::add_received(batch.len())) + .flatten(); + encode_events(events, out_file, n_workers, stats::encoder_window) + }); // A worker failure must not skip disabling tracking, draining, joining the // encoder, or detaching probes. Keep the wait result until teardown is done. @@ -164,6 +170,9 @@ fn track_command( if let Ok(total) = &total { info!("Wrote {total} memtrack events to disk"); } + if let Err(error) = stats::finish() { + warn!("{error:#}"); + } // Stop background workers after the ring pipeline has drained. Fatal // worker errors mean the capture is incomplete. diff --git a/crates/memtrack/src/perf_mappings.rs b/crates/memtrack/src/perf_mappings.rs index 33c2122a9..6dd6cd3ab 100644 --- a/crates/memtrack/src/perf_mappings.rs +++ b/crates/memtrack/src/perf_mappings.rs @@ -66,6 +66,7 @@ impl PerfMappingPoller { } mappings.sort_unstable_by_key(|event| (event.pid, event.timestamp)); if !mappings.is_empty() { + crate::ebpf::stats::add_sent(mappings.len()); let _ = tx.send(mappings); } }); diff --git a/crates/runner-shared/benches/memtrack_writer.rs b/crates/runner-shared/benches/memtrack_writer.rs index 10c329379..a0fc434c2 100644 --- a/crates/runner-shared/benches/memtrack_writer.rs +++ b/crates/runner-shared/benches/memtrack_writer.rs @@ -200,7 +200,7 @@ fn write_stack_events(bencher: Bencher) { fn encode(events: &[MemtrackEvent], n_workers: usize) -> Vec { let mut output = Vec::new(); - encode_events(events.iter().cloned(), &mut output, n_workers).unwrap(); + encode_events(events.iter().cloned(), &mut output, n_workers, |_| {}).unwrap(); output } diff --git a/crates/runner-shared/src/artifacts/memtrack/pipeline.rs b/crates/runner-shared/src/artifacts/memtrack/pipeline.rs index 5cc9a10e2..e998d9997 100644 --- a/crates/runner-shared/src/artifacts/memtrack/pipeline.rs +++ b/crates/runner-shared/src/artifacts/memtrack/pipeline.rs @@ -1,4 +1,5 @@ use std::io::{BufWriter, Write}; +use std::time::{Duration, Instant}; use rayon::prelude::*; @@ -18,17 +19,34 @@ const WINDOW_FRAMES: usize = 16; /// in-flight frame; undershooting only costs the doublings it fails to avoid. const FRAME_BYTES_PER_EVENT: usize = 16; +/// Timing and size of one encoded window, reported through `encode_events`' +/// `on_window` callback. +pub struct WindowStats { + pub events: usize, + /// Blocked on the input iterator, i.e. waiting for events. + pub wait: Duration, + pub encode: Duration, + pub write: Duration, + pub msgpack_bytes: u64, + pub zstd_bytes: u64, +} + /// Encode a stream of events into a single compressed artifact stream, /// compressing frames in parallel across a Rayon pool of `n_workers` threads. /// /// Events are grouped into fixed-size frames; each frame is one self-contained /// zstd frame. Frames are encoded a window at a time: a window is compressed in -/// parallel, then its frames are written in input order before the next window +/// parallel, then its frames are written in input order before the next one /// starts, so the output matches the input order and peak memory stays bounded. /// /// Blocks the calling thread until `events` is exhausted. Returns the total /// number of events written. -pub fn encode_events(events: S, out: W, n_workers: usize) -> anyhow::Result +pub fn encode_events( + events: S, + out: W, + n_workers: usize, + mut on_window: impl FnMut(&WindowStats), +) -> anyhow::Result where S: IntoIterator, W: Write, @@ -45,6 +63,7 @@ where let mut events = events.into_iter(); let mut window: Vec = Vec::with_capacity(cap); loop { + let waiting = Instant::now(); window.clear(); window.extend(events.by_ref().take(cap)); if window.is_empty() { @@ -52,36 +71,48 @@ where } total += window.len() as u64; - let frames: Vec> = pool.install(|| { + let encoding = Instant::now(); + let frames: Vec<(Vec, u64)> = pool.install(|| { window .par_chunks(FRAME_EVENTS) .map(encode_frame) .collect::>() })?; - for frame in frames { - out.write_all(&frame)?; + let writing = Instant::now(); + for (frame, _) in &frames { + out.write_all(frame)?; } + on_window(&WindowStats { + events: window.len(), + wait: encoding - waiting, + encode: writing - encoding, + write: writing.elapsed(), + msgpack_bytes: frames.iter().map(|(_, raw)| raw).sum(), + zstd_bytes: frames.iter().map(|(frame, _)| frame.len() as u64).sum(), + }); wrote_any = true; } // Always emit at least one (possibly empty) frame so the artifact stream is // valid and decodable even when no events were recorded. if !wrote_any { - out.write_all(&encode_frame(&[])?)?; + out.write_all(&encode_frame(&[])?.0)?; } out.flush()?; Ok(total) } -/// Encode one batch as a single self-contained zstd frame. -fn encode_frame(batch: &[MemtrackEvent]) -> anyhow::Result> { +/// Encode one batch as a single self-contained zstd frame, returned together +/// with its uncompressed msgpack size. +fn encode_frame(batch: &[MemtrackEvent]) -> anyhow::Result<(Vec, u64)> { let mut writer = MemtrackWriter::new(Vec::with_capacity(batch.len() * FRAME_BYTES_PER_EVENT))?; for event in batch { writer.write_event(event)?; } - writer.finish() + let raw = writer.uncompressed_bytes(); + Ok((writer.finish()?, raw)) } #[cfg(test)] @@ -113,7 +144,7 @@ mod tests { let events = malloc_events(0..(FRAME_EVENTS as u64 * 3 + 7)); let mut out = Vec::new(); - let total = encode_events(events.clone(), &mut out, 4)?; + let total = encode_events(events.clone(), &mut out, 4, |_| {})?; assert_eq!(total, events.len() as u64); let decoded: Vec<_> = MemtrackArtifact::decode_streamed(Cursor::new(out))?.collect(); @@ -127,7 +158,7 @@ mod tests { let events = malloc_events(0..(FRAME_EVENTS * WINDOW_FRAMES + 1) as u64); let mut out = Vec::new(); - let total = encode_events(events.clone(), &mut out, 4)?; + let total = encode_events(events.clone(), &mut out, 4, |_| {})?; assert_eq!(total, events.len() as u64); let decoded: Vec<_> = MemtrackArtifact::decode_streamed(Cursor::new(out))?.collect(); @@ -141,7 +172,7 @@ mod tests { let events: Vec = Vec::new(); let mut out = Vec::new(); - let total = encode_events(events, &mut out, 4)?; + let total = encode_events(events, &mut out, 4, |_| {})?; assert_eq!(total, 0); assert!(MemtrackArtifact::is_empty(Cursor::new(out))); diff --git a/crates/runner-shared/src/artifacts/memtrack/writer.rs b/crates/runner-shared/src/artifacts/memtrack/writer.rs index 2f665766b..17866441b 100644 --- a/crates/runner-shared/src/artifacts/memtrack/writer.rs +++ b/crates/runner-shared/src/artifacts/memtrack/writer.rs @@ -5,7 +5,25 @@ use super::MemtrackEvent; /// Streaming writer for memtrack events, serializing into a zstd-compressed sink. pub struct MemtrackWriter { - serializer: rmp_serde::Serializer, + serializer: rmp_serde::Serializer>, +} + +/// Counts the uncompressed msgpack bytes handed to the compressor. +pub struct ByteCounter { + inner: W, + bytes: u64, +} + +impl Write for ByteCounter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let n = self.inner.write(buf)?; + self.bytes += n as u64; + Ok(n) + } + + fn flush(&mut self) -> std::io::Result<()> { + self.inner.flush() + } } impl MemtrackWriter>> { @@ -18,13 +36,16 @@ impl MemtrackWriter>> { let encoder = zstd::Encoder::new(writer, COMPRESSION_LEVEL)?; let writer = BufWriter::with_capacity(BUFFER_SIZE, encoder); Ok(Self { - serializer: rmp_serde::Serializer::new(writer), + serializer: rmp_serde::Serializer::new(ByteCounter { + inner: writer, + bytes: 0, + }), }) } /// Finish writing, flush the compression stream, and return the sink pub fn finish(self) -> anyhow::Result { - let buffered = self.serializer.into_inner(); + let buffered = self.serializer.into_inner().inner; let encoder = buffered.into_inner().map_err(|e| e.into_error())?; let mut writer = encoder.finish()?; writer.flush()?; @@ -38,4 +59,9 @@ impl MemtrackWriter { event.serialize(&mut self.serializer)?; Ok(()) } + + /// Uncompressed bytes serialized so far. + pub fn uncompressed_bytes(&self) -> u64 { + self.serializer.get_ref().bytes + } }