perf(memtrack): reduce serialization bottleneck

[not-matthias]

Summary

• replace derived flattened MemtrackEvent serialization with a byte-identical manual serializer
• make MemtrackWriter::finish return encoded frame bytes for batching
• encode contiguous memtrack event batches on worker threads and write sequence-numbered zstd frames in order

Verification

• cargo test -p runner-shared artifacts::memtrack
• cargo test -p runner-shared
• cargo check -p runner-shared
• cargo test --manifest-path crates/memtrack/Cargo.toml via Nix shell with libclang/build deps; eBPF integration cases compiled but were ignored because GITHUB_ACTIONS was unset
• cargo bench -p runner-shared --bench memtrack_writer

Notes

• memtrack_writer bench measures the Phase A single-writer encoder path only, not the full parallel memtrack pipeline.
• End-to-end memory-mode speedup was not measured in this environment.

[codspeed-hq]

Merging this PR will improve performance by 15.77%

⚠️ Unknown Walltime execution environment detected

Using the Walltime instrument on standard Hosted Runners will lead to inconsistent data.

For the most accurate results, we recommend using CodSpeed Macro Runners: bare-metal machines fine-tuned for performance measurement consistency.

⚠️ Different runtime environments detected

Some benchmarks with significant performance changes were compared across different runtime environments,
which may affect the accuracy of the results.

Open the report in CodSpeed to investigate

⚡ 4 improved benchmarks
✅ 3 untouched benchmarks

Performance Changes

	Mode	Benchmark	BASE	HEAD	Efficiency
⚡	Simulation	write_events[1000000]	2.1 s	1.8 s	+16.72%
⚡	Simulation	write_events[100000]	207.3 ms	179 ms	+15.78%
⚡	Simulation	write_events[10000]	20 ms	17.3 ms	+15.33%
⚡	Simulation	write_events[500000]	1,036.6 ms	899.5 ms	+15.25%

Tip

Curious why this is faster? Comment @codspeedbot explain why this is faster on this PR, or directly use the CodSpeed MCP with your agent.

---

_{Comparing cod-3071-fix-serialization-bottleneck (1b48fb0) with main (7ce2c98)}

[greptile-apps]

Greptile Summary

This PR targets the memtrack serialization bottleneck with three coordinated changes: a manual Serialize impl for MemtrackEvent (tested byte-identical to the derived+flatten form), a corrected BufWriter-outside-encoder wrapping that coalesces the ~13 tiny msgpack writes per event before they reach zstd, and a parallel encode pipeline where batches are sequence-numbered, compressed concurrently by N worker threads, and written to disk in order by a single reorder-buffered writer.

• runner-shared/src/artifacts/memtrack.rs: manual Serialize, BufWriter order swap, finish() returns W so workers can recover the encoded Vec<u8> as a complete zstd frame.
• crates/memtrack/src/main.rs: dispatcher → bounded work channel → N encoder workers → unbounded frame channel → BTreeMap-ordered writer; stop_polling() gates the shutdown sequence.
• crates/memtrack/tests/shared.rs: updated to extend(batch) but does not call stop_polling() before the collection loop, causing the final partial batch (all events for processes under 64 K total) to be silently lost.

Confidence Score: 4/5

Safe to merge for the production binary; the integration test helper will silently collect zero events for any test process that generates fewer than 64 K allocation events.

The core serialization and parallel pipeline changes are correct and well-tested (byte-identity test, concatenated-frame decode test). The test helper in tests/shared.rs has a structural mismatch with the new batching contract: it never calls stop_polling() before the 10-second timeout loop, so the final partial batch is flushed after the receiver has been dropped and is permanently lost. The eBPF integration tests were not run as part of this PR's verification, so the regression is undetected.

crates/memtrack/tests/shared.rs — the track_command helper needs stop_polling() called (and the child waited on) before draining the batch channel.

Important Files Changed

Filename	Overview
crates/memtrack/tests/shared.rs	Collection loop relies on a 10-second timeout but stop_polling() is never called before the loop; tail batch for any process generating < 64 K events is permanently lost.
crates/runner-shared/src/artifacts/memtrack.rs	Manual Serialize impl (tested byte-identical), BufWriter wrapping order corrected to coalesce tiny writes before zstd, finish() now returns W for batching; all changes are correct and well-tested.
crates/memtrack/src/main.rs	New parallel encode pipeline (dispatcher → N workers → ordered writer with BTreeMap reorder buffer) is logically correct; previously-flagged join-order concern remains but is pre-existing.
crates/memtrack/src/ebpf/poller.rs	BatchAccumulator correctly batches events on the poll thread; shutdown drain (consume → sleep → consume → flush) properly captures stragglers.
crates/memtrack/src/ebpf/tracker.rs	Tracker now owns the RingBufferPoller and exposes stop_polling(); clean redesign, no issues.
crates/memtrack/src/ebpf/memtrack.rs	API rename from start_polling_with_channel to start_polling_with_batches; straightforward delegation to the new poller constructor.
crates/memtrack/Cargo.toml	Adds crossbeam-channel 0.5.15 as a direct (non-workspace) dependency; previously flagged for workspace alignment.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant PollThread as Poll Thread
    participant BatchAccum as BatchAccumulator
    participant Dispatcher as Dispatcher Thread
    participant Workers as Worker Threads
    participant Writer as Ordered Writer Thread
    participant Disk as Output File

    PollThread->>BatchAccum: push(event)
    BatchAccum->>Dispatcher: send((seq, batch)) [unbounded]
    Dispatcher->>Workers: send((seq, batch)) [bounded]
    Workers->>Workers: encode batch to zstd frame
    Workers->>Writer: send((seq, frame)) [unbounded]
    Writer->>Writer: BTreeMap reorder buffer
    Writer->>Disk: write_all(frame) in order

    Note over PollThread,Disk: Shutdown
    PollThread->>BatchAccum: flush() final batch
    BatchAccum->>Dispatcher: channel closes
    Dispatcher->>Workers: work_tx drops
    Workers->>Writer: frame_tx drops
    Writer->>Disk: flush()

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant PollThread as Poll Thread
    participant BatchAccum as BatchAccumulator
    participant Dispatcher as Dispatcher Thread
    participant Workers as Worker Threads
    participant Writer as Ordered Writer Thread
    participant Disk as Output File

    PollThread->>BatchAccum: push(event)
    BatchAccum->>Dispatcher: send((seq, batch)) [unbounded]
    Dispatcher->>Workers: send((seq, batch)) [bounded]
    Workers->>Workers: encode batch to zstd frame
    Workers->>Writer: send((seq, frame)) [unbounded]
    Writer->>Writer: BTreeMap reorder buffer
    Writer->>Disk: write_all(frame) in order

    Note over PollThread,Disk: Shutdown
    PollThread->>BatchAccum: flush() final batch
    BatchAccum->>Dispatcher: channel closes
    Dispatcher->>Workers: work_tx drops
    Workers->>Writer: frame_tx drops
    Writer->>Disk: flush()

Loading

Comments Outside Diff (1)

1. crates/memtrack/tests/shared.rs, line 149-157 (link)

   Final partial batch is silently dropped for every real integration test

   The batch channel only delivers a batch when the accumulator is full (64 K events) or when flush() is called at poller shutdown. Because tracker.stop_polling() is never called before the collection loop, the tail batch stays in the BatchAccumulator for any process that generates fewer than 64 K events — which is every typical integration test. After the 10-second timeout fires, the loop exits. The tracker is then dropped in a background thread, which triggers shutdown() → flush(), but by that point rx has gone out of scope and the channel's receiver has been dropped. The let _ = self.tx.send(full) in BatchAccumulator::flush silently discards the batch.

   Under the old one-event-at-a-time channel, events arrived as they were generated and the timeout only served as a sentinel for "no more activity." With the new batch channel the timeout fires first (nothing was ever enqueued), so the test consistently collects zero events for any process below the 64 K batch threshold.

   A fix is to wait for the child to exit, call tracker.stop_polling() to drain the ring-buffer and close the channel, and then drain with rx.iter() (no timeout needed since the channel closes automatically):

   child.wait().context("Failed to wait for command")?;
   tracker.stop_polling(); // flushes final partial batch, closes channel
   
   let mut events = Vec::new();
   for batch in rx.iter() {
       events.extend(batch);
   }

   Prompt To Fix With AI

   This is a comment left during a code review.
   Path: crates/memtrack/tests/shared.rs
   Line: 149-157
   
   Comment:
   **Final partial batch is silently dropped for every real integration test**
   
   The batch channel only delivers a batch when the accumulator is full (64 K events) or when `flush()` is called at poller shutdown. Because `tracker.stop_polling()` is never called before the collection loop, the tail batch stays in the `BatchAccumulator` for any process that generates fewer than 64 K events — which is every typical integration test. After the 10-second timeout fires, the loop exits. The tracker is then dropped in a background thread, which triggers `shutdown()` → `flush()`, but by that point `rx` has gone out of scope and the channel's receiver has been dropped. The `let _ = self.tx.send(full)` in `BatchAccumulator::flush` silently discards the batch.
   
   Under the old one-event-at-a-time channel, events arrived as they were generated and the timeout only served as a sentinel for "no more activity." With the new batch channel the timeout fires first (nothing was ever enqueued), so the test consistently collects zero events for any process below the 64 K batch threshold.
   
   A fix is to wait for the child to exit, call `tracker.stop_polling()` to drain the ring-buffer and close the channel, and then drain with `rx.iter()` (no timeout needed since the channel closes automatically):
   
   ```rust
   child.wait().context("Failed to wait for command")?;
   tracker.stop_polling(); // flushes final partial batch, closes channel
   
   let mut events = Vec::new();
   for batch in rx.iter() {
       events.extend(batch);
   }
   ```
   
   How can I resolve this? If you propose a fix, please make it concise.

   

Prompt To Fix All With AI

Fix the following 1 code review issue. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 1
crates/memtrack/tests/shared.rs:149-157
**Final partial batch is silently dropped for every real integration test**

The batch channel only delivers a batch when the accumulator is full (64 K events) or when `flush()` is called at poller shutdown. Because `tracker.stop_polling()` is never called before the collection loop, the tail batch stays in the `BatchAccumulator` for any process that generates fewer than 64 K events — which is every typical integration test. After the 10-second timeout fires, the loop exits. The tracker is then dropped in a background thread, which triggers `shutdown()` → `flush()`, but by that point `rx` has gone out of scope and the channel's receiver has been dropped. The `let _ = self.tx.send(full)` in `BatchAccumulator::flush` silently discards the batch.

Under the old one-event-at-a-time channel, events arrived as they were generated and the timeout only served as a sentinel for "no more activity." With the new batch channel the timeout fires first (nothing was ever enqueued), so the test consistently collects zero events for any process below the 64 K batch threshold.

A fix is to wait for the child to exit, call `tracker.stop_polling()` to drain the ring-buffer and close the channel, and then drain with `rx.iter()` (no timeout needed since the channel closes automatically):

```rust
child.wait().context("Failed to wait for command")?;
tracker.stop_polling(); // flushes final partial batch, closes channel

let mut events = Vec::new();
for batch in rx.iter() {
    events.extend(batch);
}
```

_{Reviews (2): Last reviewed commit: "perf(memtrack): batch ring-buffer drain ..." | Re-trigger Greptile}

[greptile-apps]

IO errors from the writer thread are masked by cascaded channel errors

When ordered_writer_thread fails (e.g. a disk write error inside writer.write_all), it exits, dropping frame_rx. Workers then get SendError on frame_tx.send, exit with that error, and their work_rx clones are dropped. The dispatcher then gets SendError on work_tx.send and also exits. Because we join the dispatcher before the ordered writer thread, the error surfaced to the caller is the dispatcher's misleading "sending on a closed channel", not the actual IO failure. The ordered_writer_thread is never joined in the error path and its Err is silently dropped.

Moving the ordered_writer_thread.join() before the dispatcher join (or at minimum before early returns) would surface the real root-cause error.

Prompt To Fix With AI

This is a comment left during a code review.
Path: crates/memtrack/src/main.rs
Line: 258-268

Comment:
**IO errors from the writer thread are masked by cascaded channel errors**

When `ordered_writer_thread` fails (e.g. a disk write error inside `writer.write_all`), it exits, dropping `frame_rx`. Workers then get `SendError` on `frame_tx.send`, exit with that error, and their `work_rx` clones are dropped. The dispatcher then gets `SendError` on `work_tx.send` and also exits. Because we join the dispatcher *before* the ordered writer thread, the error surfaced to the caller is the dispatcher's misleading "sending on a closed channel", not the actual IO failure. The `ordered_writer_thread` is never joined in the error path and its `Err` is silently dropped.

Moving the `ordered_writer_thread.join()` before the dispatcher join (or at minimum before early returns) would surface the real root-cause error.

How can I resolve this? If you propose a fix, please make it concise.