ci: add memtrack walltime benchmarks to CI

.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:

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

crates/memtrack/src/ebpf/attach_worker.rs

@@ -121,7 +121,7 @@ impl Drop for AttachWorker {
 
 struct Worker {
     poller: RingBufferPoller,
-    rx: mpsc::Receiver<AttachRequest>,
+    rx: mpsc::Receiver<Vec<AttachRequest>>,
     bpf: Arc<Mutex<MemtrackBpf>>,
     shutdown: Arc<AtomicBool>,
     fatal: Arc<Mutex<Option<String>>>,
@@ -139,7 +139,7 @@ impl Worker {
                     self.record_fatal(e);
                     return;
                 }
-                let mut batch: Vec<AttachRequest> = self.rx.try_iter().collect();
+                let mut batch: Vec<AttachRequest> = 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<AttachRequest> =
-                std::iter::once(first).chain(self.rx.try_iter()).collect();
+            let mut batch: Vec<AttachRequest> = 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();

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<runner_shared::artifacts::MemtrackEvent>,
+        tx: std::sync::mpsc::Sender<Vec<runner_shared::artifacts::MemtrackEvent>>,
     ) -> Result<RingBufferPoller> {
         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<crate::ebpf::events::AttachRequest>,
+        tx: std::sync::mpsc::Sender<Vec<crate::ebpf::events::AttachRequest>>,
     ) -> Result<RingBufferPoller> {
         with_skel!(self, skel => RingBufferPoller::new(
             &skel.maps.attach_requests,

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<T>(batch: &Mutex<Vec<T>>, tx: &Sender<Vec<T>>) {
+    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<T>(
+    control: std::result::Result<Sender<()>, RecvTimeoutError>,
+    consume: impl FnOnce(),
+    poll: impl FnOnce(),
+    batch: &Mutex<Vec<T>>,
+    tx: &Sender<Vec<T>>,
+) -> 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<M, T, F>(rb_map: &M, parse: F, tx: Sender<T>, poll_interval_ms: u64) -> Result<Self>
+    pub fn new<M, T, F>(
+        rb_map: &M,
+        parse: F,
+        tx: Sender<Vec<T>>,
+        poll_interval_ms: u64,
+    ) -> Result<Self>
     where
         M: MapCore,
         T: Send + 'static,
         F: Fn(&[u8]) -> Option<T> + Send + 'static,
     {
+        // Shared with the poll loop, which flushes whatever the callback left
+        // behind. `Arc<Mutex<_>>` rather than `Rc<RefCell<_>>` 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::<Sender<()>>();
         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<u8> {
+        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);
+    }
+}

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")?;