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 44.86%

⚠️ 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.

⚡ 4 improved benchmarks
✅ 3 untouched benchmarks
🆕 3 new benchmarks

Performance Changes

	Mode	Benchmark	BASE	HEAD	Efficiency
⚡	Simulation	write_events[1000000]	2.1 s	1.4 s	+47.8%
⚡	Simulation	write_events[100000]	207.3 ms	141.3 ms	+46.74%
⚡	Simulation	write_events[10000]	20 ms	13.9 ms	+43.69%
⚡	Simulation	write_events[500000]	1,036.6 ms	733.6 ms	+41.31%
🆕	Simulation	encode_events_realistic[16]	N/A	1.2 s	N/A
🆕	Simulation	encode_events_realistic[8]	N/A	1.2 s	N/A
🆕	Simulation	encode_events_realistic[4]	N/A	1.2 s	N/A

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 (99b9b68) with main (58d994a)}

[greptile-apps]

Greptile Summary

This PR replaces the single-threaded memtrack writer with a windowed parallel encoder: events are collected into fixed-size batches (FRAME_EVENTS = 64K), frames within each window (WINDOW_FRAMES = 16) are compressed in parallel via a dedicated Rayon pool, and then written to the output in input order. The channel backend is also upgraded from std::sync::mpsc to crossbeam-channel, and RingBufferPoller gains a proper Drop impl that calls shutdown(), fixing a previous thread-leak when the poller was dropped without explicitly calling shutdown.

• Parallel frame encoding (pipeline.rs): A bounded window loop collects up to 1M events, compresses them in parallel with par_chunks, and writes frames in order — preserving event ordering while bounding peak memory and enabling throughput scaling with CPU count.
• MemtrackWriter refactor (writer.rs): finish() now returns the inner Vec<u8> instead of (), enabling callers to collect the encoded frame bytes; the BufWriter is now placed above the zstd encoder (batching msgpack writes before feeding zstd) rather than below it.
• Poller lifecycle cleanup (poller.rs, tracker.rs): RingBufferPoller now implements Drop calling shutdown(), so Tracker::stop_polling correctly signals and joins the poll thread by simply dropping the Option<RingBufferPoller>.

Confidence Score: 5/5

Safe to merge; the parallel encoding pipeline is correct, event ordering is preserved, and the poller lifecycle is properly handled through the new Drop impl.

The core changes are well-tested: concatenated-frame decoding, ordering across window boundaries, empty-source handling, and byte-identical serialization are all covered by new tests. The RingBufferPoller Drop impl correctly signals and joins the poll thread before tx is dropped, so stop_polling() reliably closes the channel. No data correctness or liveness issues were found in the production path.

No files require special attention in the production path. The benchmark file inflates per-iteration timings by creating the Rayon pool inside the timed loop, but this does not affect production correctness.

Important Files Changed

Filename	Overview
crates/runner-shared/src/artifacts/memtrack/pipeline.rs	New parallel encoding pipeline; windowed par_chunks + ordered write is correct; the Rayon pool is created per-call (fine for production, inflates benchmark timings).
crates/runner-shared/src/artifacts/memtrack/writer.rs	MemtrackWriter refactored to return inner Vec from finish(); BufWriter now sits above zstd encoder — minor layer-order change, no correctness issue.
crates/memtrack/src/ebpf/poller.rs	Adds Drop impl calling shutdown() so dropping a RingBufferPoller always signals and joins the poll thread; consolidates two constructors into with_channel; correct.
crates/memtrack/src/ebpf/tracker.rs	Tracker now owns the RingBufferPoller directly; stop_polling() drops it via take(), triggering the Drop/shutdown() sequence — correct and clean.
crates/memtrack/src/main.rs	track_command simplified to spawn one pipeline_thread; correctly calls stop_polling() before joining to close the channel; no issues.
crates/runner-shared/src/artifacts/memtrack/mod.rs	Reorganized into submodules; concatenated-frames decoding test confirms the zstd streaming decoder handles multiple frames correctly; test suite expanded appropriately.
crates/runner-shared/benches/memtrack_writer.rs	Adds encode_events_realistic bench; Rayon thread pool is created inside bench_local on every iteration, adding thread-spawn overhead to each measurement.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant BPF as BPF Ring Buffer
    participant Poller as RingBufferPoller (thread)
    participant Chan as crossbeam channel
    participant Pipeline as encode_events (pipeline thread)
    participant Workers as Rayon Worker Pool
    participant File as Output File

    BPF->>Poller: raw events (ring buffer)
    Poller->>Chan: tx.send(event)

    loop collect window (up to 1M events)
        Chan->>Pipeline: events.by_ref().take(cap)
    end

    Pipeline->>Workers: par_chunks(64K) encode_frame()
    Workers->>Workers: zstd compress each chunk in parallel
    Workers-->>Pipeline: Vec frames (ordered)

    Pipeline->>File: write frames in order
    Pipeline->>Pipeline: repeat until channel closed

    Note over Poller,Chan: stop_polling() drops RingBufferPoller
    Note over Poller,Chan: Drop::drop() calls shutdown()
    Note over Poller,Chan: signals thread + joins + tx dropped
    Chan-->>Pipeline: channel disconnected (iterator ends)
    Pipeline->>File: flush + return total count

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 BPF as BPF Ring Buffer
    participant Poller as RingBufferPoller (thread)
    participant Chan as crossbeam channel
    participant Pipeline as encode_events (pipeline thread)
    participant Workers as Rayon Worker Pool
    participant File as Output File

    BPF->>Poller: raw events (ring buffer)
    Poller->>Chan: tx.send(event)

    loop collect window (up to 1M events)
        Chan->>Pipeline: events.by_ref().take(cap)
    end

    Pipeline->>Workers: par_chunks(64K) encode_frame()
    Workers->>Workers: zstd compress each chunk in parallel
    Workers-->>Pipeline: Vec frames (ordered)

    Pipeline->>File: write frames in order
    Pipeline->>Pipeline: repeat until channel closed

    Note over Poller,Chan: stop_polling() drops RingBufferPoller
    Note over Poller,Chan: Drop::drop() calls shutdown()
    Note over Poller,Chan: signals thread + joins + tx dropped
    Chan-->>Pipeline: channel disconnected (iterator ends)
    Pipeline->>File: flush + return total count

Loading

_{Reviews (9): Last reviewed commit: "perf: pre-size and reuse the memtrack en..." | Re-trigger Greptile}