From e4af80d55e4fa20044f5fa176a90ce15ba0e3fbc Mon Sep 17 00:00:00 2001 From: not-matthias Date: Thu, 10 Sep 2026 09:51:34 -0400 Subject: [PATCH 1/3] ci: add memtrack walltime benchmarks to CI --- .github/workflows/ci.yml | 44 ++++++++++++++++++++++++++++++++++++ crates/memtrack/codspeed.yml | 43 +++++++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+) create mode 100644 crates/memtrack/codspeed.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f8fe75ea..cdc7aa55 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -153,6 +153,49 @@ jobs: mode: ${{ matrix.mode }} run: cargo codspeed run -p runner-shared + memtrack-benchmarks: + runs-on: ${{ matrix.mode == 'walltime' && 'codspeed-macro' || 'ubuntu-latest' }} + strategy: + fail-fast: false + matrix: + mode: [walltime, memory] + env: + CODSPEED_REV: ${{ github.event.pull_request.head.sha || github.sha }} + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + submodules: true + + - uses: ./.github/actions/install-rust + - uses: ./.github/actions/install-bpf-deps + + - name: Install memtrack + run: cargo install --git https://github.com/CodSpeedHQ/codspeed --rev "$CODSPEED_REV" --locked memtrack + + - name: Grant memtrack file capabilities + run: cargo r -- setup --mode memory + + - name: Verify memtrack binary + run: codspeed-memtrack --version + + - name: Prepare memtrack output directory + run: mkdir -p /tmp/codspeed-memtrack-bench + + # The write benchmarks overwrite these archives every round; creating + # them inside a measured round would time one-off disk allocation. + - name: Pre-create benchmark archives + run: | + tar -cf /tmp/memtrack-bench.tar /usr/bin + dd if=/dev/zero of=/tmp/memtrack-bench-dd.bin bs=1M count=64 2> /dev/null + + - name: Run memtrack ${{ matrix.mode }} benchmarks + uses: CodSpeedHQ/action@3194d9a39c4d46684cb44bf7207fc56626aad8fd # v4 + with: + config: crates/memtrack/codspeed.yml + mode: ${{ matrix.mode }} + runner-version: rev:${{ env.CODSPEED_REV }} + skip-hash-check-warning: true + check: runs-on: ubuntu-latest if: always() @@ -164,6 +207,7 @@ jobs: - macos-basic-run-test - bpf-tests - benchmarks + - memtrack-benchmarks steps: - uses: re-actors/alls-green@release/v1 with: diff --git a/crates/memtrack/codspeed.yml b/crates/memtrack/codspeed.yml new file mode 100644 index 00000000..16891e63 --- /dev/null +++ b/crates/memtrack/codspeed.yml @@ -0,0 +1,43 @@ +$schema: https://raw.githubusercontent.com/CodSpeedHQ/codspeed/refs/heads/main/schemas/codspeed.schema.json + +# These benchmarks measure codspeed-memtrack's own overhead across a few +# representative workloads, not the memory usage of the tracked command. They +# run in both walltime mode (execution time) and memory mode (memtrack's own +# peak RSS/allocations while tracking each workload). Each workload's +# RSS-only and with-physical variants run back to back so physical +# tracking's overhead is directly comparable. +# +# The warmup/max times are generous because a single tracked run already pays a +# fixed BPF load + uprobe attach cost, which is far above the defaults tuned +# for near-instant commands. +# +# The write workloads overwrite archives pre-created by CI before measurement: +# a first-time multi-GB allocation on the runner disk would dominate one +# variant's warmup round purely from I/O ordering, not tracking overhead. +options: + warmup-time: "5s" + max-time: "60s" + +benchmarks: + # Read-only, low-allocation baseline. The tracked command string is run + # through `bash -c`, so output can be redirected away: otherwise every round + # dumps the whole listing into the runner log. + - name: "memtrack track ls" + exec: env CODSPEED_MEMTRACK_TRACK_PHYSICAL=0 codspeed-memtrack track "ls -la /usr/bin > /dev/null" --output /tmp/codspeed-memtrack-bench + + - name: "memtrack track ls (with physical)" + exec: env CODSPEED_MEMTRACK_TRACK_PHYSICAL=1 codspeed-memtrack track "ls -la /usr/bin > /dev/null" --output /tmp/codspeed-memtrack-bench + + # I/O-heavy with minimal allocation. + - name: "memtrack track dd" + exec: env CODSPEED_MEMTRACK_TRACK_PHYSICAL=0 codspeed-memtrack track "dd if=/dev/zero of=/tmp/memtrack-bench-dd.bin bs=1M count=64 2> /dev/null" --output /tmp/codspeed-memtrack-bench + + - name: "memtrack track dd (with physical)" + exec: env CODSPEED_MEMTRACK_TRACK_PHYSICAL=1 codspeed-memtrack track "dd if=/dev/zero of=/tmp/memtrack-bench-dd.bin bs=1M count=64 2> /dev/null" --output /tmp/codspeed-memtrack-bench + + # Allocation- and I/O-heavy: many small file reads. + - name: "memtrack track tar" + exec: env CODSPEED_MEMTRACK_TRACK_PHYSICAL=0 codspeed-memtrack track "tar -cf /tmp/memtrack-bench.tar /usr/bin" --output /tmp/codspeed-memtrack-bench + + - name: "memtrack track tar (with physical)" + exec: env CODSPEED_MEMTRACK_TRACK_PHYSICAL=1 codspeed-memtrack track "tar -cf /tmp/memtrack-bench.tar /usr/bin" --output /tmp/codspeed-memtrack-bench From dc896705fd634d526318f198774b9d97676aaae9 Mon Sep 17 00:00:00 2001 From: not-matthias Date: Wed, 16 Sep 2026 15:42:07 +0200 Subject: [PATCH 2/3] perf(memtrack): batch ring-buffer events over the channel The poll thread sent one item per channel message, and std's mpsc allocates a 31-slot block per 31 messages, so a run that captured 704k events also allocated 22.7k blocks purely to hand them over. The callback now fills a shared 1024-item buffer and sends it whole. Partial batches are flushed after every poll, after the drain-path consume before its ack, and on the shutdown consume, so `drain()` keeps promising that all pending entries sit in the channel once it returns. Consumers take `Vec` and flatten: the encode pipeline keeps its `IntoIterator` contract unchanged. --- crates/memtrack/src/ebpf/attach_worker.rs | 14 +- crates/memtrack/src/ebpf/memtrack/mod.rs | 4 +- crates/memtrack/src/ebpf/poller.rs | 172 +++++++++++++++++++--- crates/memtrack/src/main.rs | 3 +- crates/memtrack/src/session.rs | 6 +- crates/memtrack/tests/shared.rs | 2 +- 6 files changed, 169 insertions(+), 32 deletions(-) diff --git a/crates/memtrack/src/ebpf/attach_worker.rs b/crates/memtrack/src/ebpf/attach_worker.rs index 61a49c4e..892fe621 100644 --- a/crates/memtrack/src/ebpf/attach_worker.rs +++ b/crates/memtrack/src/ebpf/attach_worker.rs @@ -121,7 +121,7 @@ impl Drop for AttachWorker { struct Worker { poller: RingBufferPoller, - rx: mpsc::Receiver, + rx: mpsc::Receiver>, bpf: Arc>, shutdown: Arc, fatal: Arc>>, @@ -139,7 +139,7 @@ impl Worker { self.record_fatal(e); return; } - let mut batch: Vec = self.rx.try_iter().collect(); + let mut batch: Vec = self.rx.try_iter().flatten().collect(); if batch.is_empty() { break; } @@ -151,12 +151,14 @@ impl Worker { } let first = match self.rx.recv_timeout(RECV_TIMEOUT) { - Ok(req) => req, + Ok(reqs) => reqs, Err(RecvTimeoutError::Timeout) => continue, Err(RecvTimeoutError::Disconnected) => return, }; - let mut batch: Vec = - std::iter::once(first).chain(self.rx.try_iter()).collect(); + let mut batch: Vec = first + .into_iter() + .chain(self.rx.try_iter().flatten()) + .collect(); if let Err(e) = self.process_batch(&mut batch, &mut known) { self.record_fatal(e); @@ -195,7 +197,7 @@ impl Worker { // Every producer is stopped, so a synchronous drain is complete. self.poller.drain()?; - batch.extend(self.rx.try_iter()); + batch.extend(self.rx.try_iter().flatten()); } let mut seen: HashSet<(u64, u64)> = HashSet::new(); diff --git a/crates/memtrack/src/ebpf/memtrack/mod.rs b/crates/memtrack/src/ebpf/memtrack/mod.rs index 2ec1f04f..1c6508a9 100644 --- a/crates/memtrack/src/ebpf/memtrack/mod.rs +++ b/crates/memtrack/src/ebpf/memtrack/mod.rs @@ -218,7 +218,7 @@ impl MemtrackBpf { pub fn poll_events_with_channel( &self, poll_interval_ms: u64, - tx: std::sync::mpsc::Sender, + tx: std::sync::mpsc::Sender>, ) -> Result { with_skel!(self, skel => RingBufferPoller::new( &skel.maps.events, @@ -233,7 +233,7 @@ impl MemtrackBpf { pub(crate) fn poll_attach_with_channel( &self, poll_interval_ms: u64, - tx: std::sync::mpsc::Sender, + tx: std::sync::mpsc::Sender>, ) -> Result { with_skel!(self, skel => RingBufferPoller::new( &skel.maps.attach_requests, diff --git a/crates/memtrack/src/ebpf/poller.rs b/crates/memtrack/src/ebpf/poller.rs index d80a1554..38dfd8de 100644 --- a/crates/memtrack/src/ebpf/poller.rs +++ b/crates/memtrack/src/ebpf/poller.rs @@ -1,11 +1,59 @@ use anyhow::{Context, Result}; use libbpf_rs::{MapCore, RingBufferBuilder}; +use parking_lot::Mutex; +use std::sync::Arc; use std::sync::mpsc::{self, RecvTimeoutError, Sender}; use std::thread::JoinHandle; use std::time::Duration; +/// 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. +const BATCH_ITEMS: usize = 1024; + +/// Hand a non-empty partial batch to the channel, leaving the buffer empty but +/// still reserved. The lock is released before the send so a slow consumer +/// never blocks the ring-buffer callback. +fn flush_batch(batch: &Mutex>, tx: &Sender>) { + let mut buf = batch.lock(); + if buf.is_empty() { + return; + } + let items = std::mem::replace(&mut *buf, Vec::with_capacity(BATCH_ITEMS)); + drop(buf); + let _ = tx.send(items); +} + +fn poll_iteration( + control: std::result::Result, RecvTimeoutError>, + consume: impl FnOnce(), + poll: impl FnOnce(), + batch: &Mutex>, + tx: &Sender>, +) -> bool { + match control { + Ok(ack) => { + consume(); + // `drain` promises pending entries are in the channel before returning. + flush_batch(batch, tx); + let _ = ack.send(()); + true + } + Err(RecvTimeoutError::Timeout) => { + poll(); + flush_batch(batch, tx); + true + } + Err(RecvTimeoutError::Disconnected) => { + consume(); + flush_batch(batch, tx); + false + } + } +} + /// Polls a BPF ring buffer in a background thread, parsing raw entries with a -/// user-supplied closure and forwarding them to an mpsc channel. +/// user-supplied closure and forwarding them to an mpsc channel in batches. /// /// The poll thread runs until the poller is dropped, doing a final full /// `consume()` on shutdown so no buffered entries are lost. @@ -15,17 +63,38 @@ pub struct RingBufferPoller { } impl RingBufferPoller { - pub fn new(rb_map: &M, parse: F, tx: Sender, poll_interval_ms: u64) -> Result + pub fn new( + rb_map: &M, + parse: F, + tx: Sender>, + poll_interval_ms: u64, + ) -> Result where M: MapCore, T: Send + 'static, F: Fn(&[u8]) -> Option + Send + 'static, { + // Shared with the poll loop, which flushes whatever the callback left + // behind. `Arc>` rather than `Rc>` because the + // built `RingBuffer` is moved into the poll thread, so the callback + // must be `Send`. + let batch = Arc::new(Mutex::new(Vec::with_capacity(BATCH_ITEMS))); + let cb_batch = Arc::clone(&batch); + let cb_tx = tx.clone(); + let mut builder = RingBufferBuilder::new(); builder.add(rb_map, move |data| { - if let Some(item) = parse(data) { - let _ = tx.send(item); + let Some(item) = parse(data) else { + return 0; + }; + let mut buf = cb_batch.lock(); + buf.push(item); + if buf.len() < BATCH_ITEMS { + return 0; } + let items = std::mem::replace(&mut *buf, Vec::with_capacity(BATCH_ITEMS)); + drop(buf); + let _ = cb_tx.send(items); 0 })?; let ringbuf = builder.build()?; @@ -35,21 +104,17 @@ impl RingBufferPoller { // poll tick, and disconnection is the shutdown signal. let (ctl, ctl_rx) = mpsc::channel::>(); let poll_thread = std::thread::spawn(move || { - loop { - match ctl_rx.recv_timeout(Duration::from_millis(poll_interval_ms)) { - Ok(ack) => { - let _ = ringbuf.consume(); - let _ = ack.send(()); - } - Err(RecvTimeoutError::Timeout) => { - let _ = ringbuf.poll(Duration::ZERO); - } - Err(RecvTimeoutError::Disconnected) => { - let _ = ringbuf.consume(); - break; - } - } - } + while poll_iteration( + ctl_rx.recv_timeout(Duration::from_millis(poll_interval_ms)), + || { + let _ = ringbuf.consume(); + }, + || { + let _ = ringbuf.poll(Duration::ZERO); + }, + &batch, + &tx, + ) {} }); Ok(Self { @@ -77,3 +142,72 @@ impl Drop for RingBufferPoller { } } } + +#[cfg(test)] +mod tests { + use super::*; + use std::cell::Cell; + + fn partial_batch() -> Vec { + vec![42; BATCH_ITEMS - 1] + } + + #[test] + fn timeout_flushes_partial_batch() { + let expected = partial_batch(); + let batch = Mutex::new(expected.clone()); + let (tx, rx) = mpsc::channel(); + let polled = Cell::new(false); + + assert!(poll_iteration( + Err(RecvTimeoutError::Timeout), + || unreachable!(), + || polled.set(true), + &batch, + &tx, + )); + + assert!(polled.get()); + assert_eq!(rx.recv().unwrap(), expected); + } + + #[test] + fn drain_flushes_partial_batch() { + let expected = partial_batch(); + let batch = Mutex::new(expected.clone()); + let (tx, rx) = mpsc::channel(); + let (ack_tx, ack_rx) = mpsc::channel(); + let consumed = Cell::new(false); + + assert!(poll_iteration( + Ok(ack_tx), + || consumed.set(true), + || unreachable!(), + &batch, + &tx, + )); + + assert!(consumed.get()); + assert_eq!(rx.recv().unwrap(), expected); + ack_rx.recv().unwrap(); + } + + #[test] + fn shutdown_flushes_partial_batch() { + let expected = partial_batch(); + let batch = Mutex::new(expected.clone()); + let (tx, rx) = mpsc::channel(); + let consumed = Cell::new(false); + + assert!(!poll_iteration( + Err(RecvTimeoutError::Disconnected), + || consumed.set(true), + || unreachable!(), + &batch, + &tx, + )); + + assert!(consumed.get()); + assert_eq!(rx.recv().unwrap(), expected); + } +} diff --git a/crates/memtrack/src/main.rs b/crates/memtrack/src/main.rs index 283cff19..ecec5d95 100644 --- a/crates/memtrack/src/main.rs +++ b/crates/memtrack/src/main.rs @@ -130,7 +130,8 @@ 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, out_file, n_workers)); + let pipeline_thread = + thread::spawn(move || encode_events(event_rx.into_iter().flatten(), out_file, n_workers)); // Wait for the command to complete let status = session.wait().context("Failed to wait for command")?; diff --git a/crates/memtrack/src/session.rs b/crates/memtrack/src/session.rs index 7aec33fe..8adc7d76 100644 --- a/crates/memtrack/src/session.rs +++ b/crates/memtrack/src/session.rs @@ -8,14 +8,14 @@ use std::sync::mpsc::Receiver; /// stays alive as long as the session does; dropping it stops event delivery. pub struct Session { child: Child, - events: Option>, + events: Option>>, _poller: RingBufferPoller, } impl Session { pub(crate) fn new( child: Child, - events: Receiver, + events: Receiver>, poller: RingBufferPoller, ) -> Self { Self { @@ -30,7 +30,7 @@ impl Session { } /// Take ownership of the event receiver. Can only be taken once. - pub fn take_events(&mut self) -> Result> { + pub fn take_events(&mut self) -> Result>> { self.events.take().context("events already taken") } diff --git a/crates/memtrack/tests/shared.rs b/crates/memtrack/tests/shared.rs index c3bb37bc..a4e2317a 100644 --- a/crates/memtrack/tests/shared.rs +++ b/crates/memtrack/tests/shared.rs @@ -361,7 +361,7 @@ fn run_tracked( // Dropping the session does a final ring buffer drain and closes the // channel, so collecting terminates without a silence timeout. drop(session); - let events: Vec = rx.iter().collect(); + let events: Vec = rx.into_iter().flatten().collect(); tracker.finish()?; From 829da2f9fba1b9f7b2d095dfaeb0778490069490 Mon Sep 17 00:00:00 2001 From: not-matthias Date: Wed, 16 Sep 2026 15:42:27 +0200 Subject: [PATCH 3/3] test(memtrack): bench the channel and worker-scaling paths The encoder benches fed `encode_events` from a Vec, while production feeds it an mpsc receiver, so the channel the events actually arrive through was absent from the benchmarks. Adds a per-event and a batched channel bench that differ only in batching, plus a worker sweep bounded by the core count, since asking for more workers than pinned cores measures oversubscription rather than scaling. --- .../runner-shared/benches/memtrack_writer.rs | 76 +++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/crates/runner-shared/benches/memtrack_writer.rs b/crates/runner-shared/benches/memtrack_writer.rs index 700f4db8..fb402ea8 100644 --- a/crates/runner-shared/benches/memtrack_writer.rs +++ b/crates/runner-shared/benches/memtrack_writer.rs @@ -2,6 +2,8 @@ use divan::Bencher; use rand::rngs::StdRng; use rand::{Rng, SeedableRng}; use runner_shared::artifacts::{MemtrackEvent, MemtrackEventKind, MemtrackWriter, encode_events}; +use std::sync::mpsc; +use std::thread; fn main() { divan::main(); @@ -139,3 +141,77 @@ fn encode_events_realistic(bencher: Bencher, n_workers: usize) { output }); } + +/// Matches the batch size the ring-buffer poller flushes with, so the batched +/// channel bench reproduces the production send pattern. +const CHANNEL_BATCH: usize = 1024; + +/// Fixed for both channel benches so they stay a valid paired comparison. +const CHANNEL_WORKERS: usize = 8; + +/// Production feeds `encode_events` from an mpsc channel filled by the +/// ring-buffer poll thread, so the channel itself is part of the hot path. +#[divan::bench(max_time = 10.0)] +fn encode_events_via_channel(bencher: Bencher) { + let events = generate_realistic_events(REALISTIC_EVENTS); + + bencher.bench_local(|| { + let producer_events = events.as_slice(); + let (tx, rx) = mpsc::channel(); + let mut output = Vec::new(); + + thread::scope(|scope| { + scope.spawn(move || { + for event in producer_events { + if tx.send(*event).is_err() { + return; + } + } + }); + + encode_events(rx, &mut output, CHANNEL_WORKERS).unwrap(); + }); + + output + }); +} + +/// A/B counterpart of `encode_events_via_channel`: identical except that the +/// producer sends batches, amortizing the per-message channel block allocation. +#[divan::bench(max_time = 10.0)] +fn encode_events_via_batched_channel(bencher: Bencher) { + let events = generate_realistic_events(REALISTIC_EVENTS); + + bencher.bench_local(|| { + let producer_events = events.as_slice(); + let (tx, rx) = mpsc::channel(); + let mut output = Vec::new(); + + thread::scope(|scope| { + scope.spawn(move || { + for batch in producer_events.chunks(CHANNEL_BATCH) { + if tx.send(batch.to_vec()).is_err() { + return; + } + } + }); + + encode_events(rx.into_iter().flatten(), &mut output, CHANNEL_WORKERS).unwrap(); + }); + + output + }); +} + +// Worker counts stay at or below the physical core count: above it the pinned +// benchmark process oversubscribes and the timings measure scheduling, not scaling. +#[divan::bench(args = [1, 2, 4, 8], max_time = 10.0)] +fn encode_events_worker_scaling(bencher: Bencher, n_workers: usize) { + let events = generate_realistic_events(REALISTIC_EVENTS); + + bencher.bench_local(|| { + let mut output = Vec::new(); + encode_events(events.iter().copied(), &mut output, n_workers).unwrap(); + output + }); +}