From a571db35c2316c25d972a5d5251f6edc71063f78 Mon Sep 17 00:00:00 2001 From: not-matthias Date: Fri, 25 Sep 2026 14:41:49 +0200 Subject: [PATCH 1/4] perf(runner-shared): stream memtrack encoder frames instead of windows encode_events collected a whole window of 16 x 64k events, encoded it, then wrote it before reading more. Reading stopped for the whole encode, so events piled up in the unbounded channel feeding it, and the window kept about 1M events alive at once and freed them in one go. On a large memory benchmark suite with stack capture the encoder waited for input ~90% of the time, yet held ~4 GB per window and drove memtrack's RSS to ~6.4 GiB. Each 64k-event frame now goes to the worker pool as soon as it fills, and finished frames are written in input order. At most 2 frames per worker are in flight; at that cap the reader waits for the oldest one. Output order and the empty-stream frame are unchanged. Closes COD-3658 Co-Authored-By: Claude --- .../src/artifacts/memtrack/pipeline.rs | 97 +++++++++++++------ 1 file changed, 65 insertions(+), 32 deletions(-) diff --git a/crates/runner-shared/src/artifacts/memtrack/pipeline.rs b/crates/runner-shared/src/artifacts/memtrack/pipeline.rs index c47b3aed9..2c7cb9fe3 100644 --- a/crates/runner-shared/src/artifacts/memtrack/pipeline.rs +++ b/crates/runner-shared/src/artifacts/memtrack/pipeline.rs @@ -1,6 +1,6 @@ +use std::collections::VecDeque; use std::io::{BufWriter, Write}; - -use rayon::prelude::*; +use std::sync::mpsc::{self, Receiver, TryRecvError}; use super::MemtrackEvent; use super::writer::MemtrackWriter; @@ -8,19 +8,20 @@ use super::writer::MemtrackWriter; /// Events per self-contained zstd frame. Larger frames compress better; smaller /// frames cap the work (and memory) a single worker holds while encoding. const FRAME_EVENTS: usize = 64 * 1024; -/// Frames compressed in parallel per window. A window is encoded across the -/// worker pool and then written before the next one starts, so this bounds peak -/// memory to roughly `FRAME_EVENTS * WINDOW_FRAMES` events regardless of how long -/// the source runs. -const WINDOW_FRAMES: usize = 16; +/// Frames allowed in flight per worker. At the cap the reader waits for the +/// oldest frame, bounding memory to about `cap + 1` frames. +const MAX_IN_FLIGHT_PER_WORKER: usize = 2; + +type FrameResult = Receiver>>; /// 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 -/// starts, so the output matches the input order and peak memory stays bounded. +/// zstd frame. Full frames are submitted as soon as they fill, and completed +/// frames are written in input order. At most +/// `MAX_IN_FLIGHT_PER_WORKER * n_workers` frames are in flight; at the cap the +/// reader waits for the oldest frame before pulling more events. /// /// Blocks the calling thread until `events` is exhausted. Returns the total /// number of events written. @@ -32,32 +33,58 @@ where let pool = rayon::ThreadPoolBuilder::new() .num_threads(n_workers.max(1)) .build()?; + let max_in_flight = MAX_IN_FLIGHT_PER_WORKER * n_workers.max(1); let mut out = BufWriter::new(out); let mut total = 0u64; let mut wrote_any = false; - - let cap = FRAME_EVENTS * WINDOW_FRAMES; - let mut events = events.into_iter(); - let mut window: Vec = Vec::with_capacity(cap); - loop { - window.clear(); - window.extend(events.by_ref().take(cap)); - if window.is_empty() { - break; + let mut in_flight: VecDeque = VecDeque::with_capacity(max_in_flight); + + let submit = |frame: Vec, in_flight: &mut VecDeque| { + let (tx, rx) = mpsc::sync_channel(1); + pool.spawn(move || { + let _ = tx.send(encode_frame(&frame)); + }); + in_flight.push_back(rx); + }; + + let mut frame: Vec = Vec::with_capacity(FRAME_EVENTS); + for event in events { + frame.push(event); + if frame.len() < FRAME_EVENTS { + continue; + } + total += frame.len() as u64; + let full = std::mem::replace(&mut frame, Vec::with_capacity(FRAME_EVENTS)); + submit(full, &mut in_flight); + + while let Some(rx) = in_flight.front() { + match rx.try_recv() { + Ok(encoded) => { + out.write_all(&encoded?)?; + wrote_any = true; + in_flight.pop_front(); + } + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Disconnected) => { + anyhow::bail!("frame encoder worker exited without a result") + } + } } - total += window.len() as u64; - - let frames: Vec> = pool.install(|| { - window - .par_chunks(FRAME_EVENTS) - .map(encode_frame) - .collect::>() - })?; - for frame in frames { - out.write_all(&frame)?; + if in_flight.len() >= max_in_flight { + let rx = in_flight.pop_front().expect("in-flight queue is non-empty"); + out.write_all(&recv_frame(&rx)?)?; + wrote_any = true; } + } + + if !frame.is_empty() { + total += frame.len() as u64; + submit(frame, &mut in_flight); + } + for rx in in_flight.drain(..) { + out.write_all(&recv_frame(&rx)?)?; wrote_any = true; } @@ -71,6 +98,12 @@ where Ok(total) } +/// Block until the frame behind `rx` is encoded. +fn recv_frame(rx: &FrameResult) -> anyhow::Result> { + rx.recv() + .map_err(|_| anyhow::anyhow!("frame encoder worker exited without a result"))? +} + /// Encode one batch as a single self-contained zstd frame. fn encode_frame(batch: &[MemtrackEvent]) -> anyhow::Result> { let mut writer = MemtrackWriter::new(Vec::new())?; @@ -116,11 +149,11 @@ mod tests { } #[test] - fn preserves_order_across_window_boundary() -> anyhow::Result<()> { - let events = malloc_events(0..(FRAME_EVENTS * WINDOW_FRAMES + 1) as u64); + fn preserves_order_beyond_in_flight_cap() -> anyhow::Result<()> { + let events = malloc_events(0..(FRAME_EVENTS as u64 * 5 + 3)); let mut out = Vec::new(); - let total = encode_events(events.clone(), &mut out, 4)?; + let total = encode_events(events.clone(), &mut out, 1)?; assert_eq!(total, events.len() as u64); let decoded: Vec<_> = MemtrackArtifact::decode_streamed(Cursor::new(out))?.collect(); From 51ef46fc31878db430289f107fc902b160e4e1b7 Mon Sep 17 00:00:00 2001 From: CodSpeed Bot Date: Fri, 25 Sep 2026 13:49:00 +0000 Subject: [PATCH 2/4] perf(runner-shared): serialize memtrack frames into a buffer and compress in one shot --- .../src/artifacts/memtrack/pipeline.rs | 145 +++++++++++++----- .../src/artifacts/memtrack/writer.rs | 7 +- 2 files changed, 113 insertions(+), 39 deletions(-) diff --git a/crates/runner-shared/src/artifacts/memtrack/pipeline.rs b/crates/runner-shared/src/artifacts/memtrack/pipeline.rs index 2c7cb9fe3..1155555b9 100644 --- a/crates/runner-shared/src/artifacts/memtrack/pipeline.rs +++ b/crates/runner-shared/src/artifacts/memtrack/pipeline.rs @@ -1,9 +1,12 @@ +use std::cell::RefCell; use std::collections::VecDeque; use std::io::{BufWriter, Write}; use std::sync::mpsc::{self, Receiver, TryRecvError}; +use serde::Serialize; + use super::MemtrackEvent; -use super::writer::MemtrackWriter; +use super::writer::COMPRESSION_LEVEL; /// Events per self-contained zstd frame. Larger frames compress better; smaller /// frames cap the work (and memory) a single worker holds while encoding. @@ -12,7 +15,9 @@ const FRAME_EVENTS: usize = 64 * 1024; /// oldest frame, bounding memory to about `cap + 1` frames. const MAX_IN_FLIGHT_PER_WORKER: usize = 2; -type FrameResult = Receiver>>; +/// An encoded frame, plus the event buffer it was built from so the reader can +/// reuse it for the next frame instead of allocating a new one. +type EncodedFrame = (anyhow::Result>, Vec); /// Encode a stream of events into a single compressed artifact stream, /// compressing frames in parallel across a Rayon pool of `n_workers` threads. @@ -38,14 +43,24 @@ where let mut out = BufWriter::new(out); let mut total = 0u64; let mut wrote_any = false; - let mut in_flight: VecDeque = VecDeque::with_capacity(max_in_flight); + let mut in_flight: VecDeque> = VecDeque::with_capacity(max_in_flight); + // Event buffers handed back by workers, reused for the next frames. + let mut spare: Vec> = Vec::new(); - let submit = |frame: Vec, in_flight: &mut VecDeque| { + let submit = |frame: Vec| { let (tx, rx) = mpsc::sync_channel(1); pool.spawn(move || { - let _ = tx.send(encode_frame(&frame)); + let encoded = encode_frame(&frame); + let _ = tx.send((encoded, frame)); }); - in_flight.push_back(rx); + rx + }; + let mut collect = |(encoded, mut frame): EncodedFrame, spare: &mut Vec>| { + out.write_all(&encoded?)?; + wrote_any = true; + frame.clear(); + spare.push(frame); + anyhow::Ok(()) }; let mut frame: Vec = Vec::with_capacity(FRAME_EVENTS); @@ -55,37 +70,32 @@ where continue; } total += frame.len() as u64; - let full = std::mem::replace(&mut frame, Vec::with_capacity(FRAME_EVENTS)); - submit(full, &mut in_flight); + let next = spare + .pop() + .unwrap_or_else(|| Vec::with_capacity(FRAME_EVENTS)); + in_flight.push_back(submit(std::mem::replace(&mut frame, next))); + // Write finished frames in input order. Wait for the oldest frame only + // while the queue is at the cap. while let Some(rx) = in_flight.front() { - match rx.try_recv() { - Ok(encoded) => { - out.write_all(&encoded?)?; - wrote_any = true; - in_flight.pop_front(); - } - Err(TryRecvError::Empty) => break, - Err(TryRecvError::Disconnected) => { - anyhow::bail!("frame encoder worker exited without a result") - } - } - } - - if in_flight.len() >= max_in_flight { - let rx = in_flight.pop_front().expect("in-flight queue is non-empty"); - out.write_all(&recv_frame(&rx)?)?; - wrote_any = true; + let at_cap = in_flight.len() >= max_in_flight; + let result = match rx.try_recv() { + Ok(result) => result, + Err(TryRecvError::Empty) if !at_cap => break, + // Blocks at the cap; fails at once if the worker is gone. + Err(_) => recv_frame(rx)?, + }; + in_flight.pop_front(); + collect(result, &mut spare)?; } } if !frame.is_empty() { total += frame.len() as u64; - submit(frame, &mut in_flight); + in_flight.push_back(submit(frame)); } for rx in in_flight.drain(..) { - out.write_all(&recv_frame(&rx)?)?; - wrote_any = true; + collect(recv_frame(&rx)?, &mut spare)?; } // Always emit at least one (possibly empty) frame so the artifact stream is @@ -99,25 +109,69 @@ where } /// Block until the frame behind `rx` is encoded. -fn recv_frame(rx: &FrameResult) -> anyhow::Result> { +fn recv_frame(rx: &Receiver) -> anyhow::Result { rx.recv() - .map_err(|_| anyhow::anyhow!("frame encoder worker exited without a result"))? + .map_err(|_| anyhow::anyhow!("frame encoder worker exited without a result")) +} + +/// Upper estimate of the msgpack size of one event, used to size the frame +/// buffer up front. Growing it by doubling instead would leave each worker +/// pinning about twice a frame's msgpack size. +const MSGPACK_BYTES_PER_EVENT: usize = 80; + +/// Per-thread scratch state reused across frames: the msgpack buffer and the +/// zstd compression context. +/// +/// Each worker keeps its buffer (about one frame's msgpack, ~5 MiB) until the +/// pool exits, even when idle. Peak usage needs it anyway, since every busy +/// worker holds one, and allocating it per frame measured about 9% slower with +/// a single worker. +struct FrameEncoder { + msgpack: Vec, + compressor: zstd::bulk::Compressor<'static>, +} + +thread_local! { + static FRAME_ENCODER: RefCell> = const { RefCell::new(None) }; } /// Encode one batch as a single self-contained zstd frame. +/// +/// The batch is serialized with `rmp_serde` into a reused buffer, then +/// compressed in one shot with a reused zstd context. The decoded bytes are the +/// same msgpack stream `MemtrackWriter` produces. fn encode_frame(batch: &[MemtrackEvent]) -> anyhow::Result> { - let mut writer = MemtrackWriter::new(Vec::new())?; - for event in batch { - writer.write_event(event)?; - } - writer.finish() + FRAME_ENCODER.with(|cell| { + let mut slot = cell.borrow_mut(); + let enc = match slot.as_mut() { + Some(enc) => enc, + None => slot.insert(FrameEncoder { + msgpack: Vec::new(), + compressor: zstd::bulk::Compressor::new(COMPRESSION_LEVEL)?, + }), + }; + + enc.msgpack.clear(); + enc.msgpack + .reserve_exact(batch.len() * MSGPACK_BYTES_PER_EVENT); + let mut serializer = rmp_serde::Serializer::new(&mut enc.msgpack); + for event in batch { + event.serialize(&mut serializer)?; + } + + let mut compressed = enc.compressor.compress(&enc.msgpack)?; + // Trim the worst-case compression bound so in-flight frames only hold + // their actual size. + compressed.shrink_to_fit(); + Ok(compressed) + }) } #[cfg(test)] mod tests { use std::io::Cursor; - use super::super::{MemtrackArtifact, MemtrackEventKind}; + use super::super::{MemtrackArtifact, MemtrackEventKind, MemtrackWriter}; use super::*; fn malloc_events(range: std::ops::Range) -> Vec { @@ -162,6 +216,25 @@ mod tests { Ok(()) } + #[test] + fn frame_payload_matches_memtrack_writer() -> anyhow::Result<()> { + let events = malloc_events(0..10_000); + + let mut reference = MemtrackWriter::new(Vec::new())?; + for event in &events { + reference.write_event(event)?; + } + let reference = zstd::decode_all(Cursor::new(reference.finish()?))?; + + // Encode twice to also exercise the reused per-thread buffers. + for _ in 0..2 { + let frame = zstd::decode_all(Cursor::new(encode_frame(&events)?))?; + assert_eq!(frame, reference); + } + + Ok(()) + } + #[test] fn empty_source_writes_a_valid_stream() -> anyhow::Result<()> { let events: Vec = Vec::new(); diff --git a/crates/runner-shared/src/artifacts/memtrack/writer.rs b/crates/runner-shared/src/artifacts/memtrack/writer.rs index 2f665766b..976cd5f65 100644 --- a/crates/runner-shared/src/artifacts/memtrack/writer.rs +++ b/crates/runner-shared/src/artifacts/memtrack/writer.rs @@ -3,6 +3,10 @@ use std::io::{BufWriter, Write}; use super::MemtrackEvent; +/// zstd level used for memtrack artifacts. We're dealing with a lot of events, so +/// we want to compress as much as possible while not taking too much time. +pub(crate) const COMPRESSION_LEVEL: i32 = -5; + /// Streaming writer for memtrack events, serializing into a zstd-compressed sink. pub struct MemtrackWriter { serializer: rmp_serde::Serializer, @@ -10,9 +14,6 @@ pub struct MemtrackWriter { impl MemtrackWriter>> { pub fn new(writer: W) -> anyhow::Result { - // We're dealing with a lot of events, so we want to compress as much as possible - // while not taking too much time to compress. - const COMPRESSION_LEVEL: i32 = -5; const BUFFER_SIZE: usize = 256 * 1024 /* 256 KB */; let encoder = zstd::Encoder::new(writer, COMPRESSION_LEVEL)?; From b432df01830f22f7d0859354c3e5264e2a294d9f Mon Sep 17 00:00:00 2001 From: CodSpeed Bot Date: Fri, 25 Sep 2026 14:28:43 +0000 Subject: [PATCH 3/4] ci: run the runner-shared benchmarks in memory mode Track the peak memory of the memtrack encoder benchmarks alongside simulation and walltime. --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 926b6978a..54d8b8dc2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -133,7 +133,7 @@ jobs: strategy: fail-fast: false matrix: - mode: [simulation, walltime] + mode: [simulation, walltime, memory] steps: - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 with: From 530c88f685331d8982a451143f58a41ffdb7b73b Mon Sep 17 00:00:00 2001 From: CodSpeed Bot Date: Fri, 25 Sep 2026 14:28:43 +0000 Subject: [PATCH 4/4] ci: run the clang-format hook serially The hook pip-installs clang-format on first use. On a cold prek cache its parallel batches race on that install and fail with a PermissionError. --- .pre-commit-config.yaml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index c39411993..84ff4f297 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -27,6 +27,9 @@ repos: - id: clang-format files: ^crates/memtrack/src/ebpf/c/.*\.(c|h|bpf\.c)$ args: [--style=file, -i, --version=22.1.2] + # The hook pip-installs clang-format on first use; parallel batches + # race on that install and fail with a PermissionError on a cold cache. + require_serial: true - repo: local hooks: - id: init-worktree