[cpp] Multi-threaded ParallelReader by facontidavide · Pull Request #1703 · foxglove/mcap

facontidavide · 2026-06-09T07:46:59Z

Summary

Adds a standalone, multithreaded mcap::ParallelReader for log-time-ordered reads.
It decompresses chunks on a thread pool ahead of a k-way merge frontier and emits
messages in exactly the same order as the serial IndexedMessageReader (verified
by parity tests). Purely additive: adds mcap::ParallelReader (+ a concurrent
file reader) without modifying McapReader; the only touch to an existing type is
IReadable::supportsConcurrentRead() (defaults false). Define MCAP_NO_PARALLEL
to omit it.

What's included

parallel_reader.hpp — concurrent chunk decompression + ordered
single-consumer k-way merge.
concurrent_file_reader.hpp — the open(path) source: positioned reads
(pread on POSIX, ReadFile+OVERLAPPED offset on Windows). The offset is
passed per call, so many workers read concurrently with no shared cursor, through
the page cache.
thread_pool.hpp, byte_semaphore.hpp, parallel_budget.hpp — worker pool
- memory back-pressure.
message_byte_store.hpp — deferred per-message byte access.
cpp/test/parallel_reader_test.cpp — parity + behavior tests.

I/O: positioned reads through the page cache

The source uses positioned reads (pread / ReadFile with an OVERLAPPED offset).
The offset is passed per call, so workers read concurrently with no shared cursor,
and reads stay resident in the page cache: a full read does ~0 major page faults,
keeps RSS low with no extra RSS-bounding machinery, and is faster than the serial
reader even single-threaded on a ~17 GB file.

Memory back-pressure + thread default

ParallelReadOptions::memoryCap: ByteBudget (default, precise resident-byte
ceiling) or ChunkCount (opt-in, cap live chunks). Default worker threads is 4
(capped at 8) instead of hardware_concurrency() — benchmarks showed the latter
was consistently slower (reads are usually consumer-bound and decompression
saturates memory bandwidth before all cores).

Reviewer note: the cap currently also clamps an explicit threadCount; happy to
honor explicit values and cap only the automatic (0) default if preferred.

Per-message read-path optimizations (profiled with `perf`)

Resolve channel/schema by const-ref from one-time snapshots instead of
McapReader::channel()/schema() returning shared_ptr by value (removes
per-message atomic refcount churn).
Order each chunk's entries by k-way merging the already-sorted per-channel
MessageIndex runs instead of a full std::sort (fallback to std::sort for
reverse / non-monotonic / single-run). Emit order unchanged (parity-verified).

clalancette

Thanks for the improvements, I've taken a look. Overall, I like the idea. Besides the things I've added inline, I have a few overarching things to think about:

All of the other parts of the C++ library use the .hpp/.inl split to split between declarations and implementation. This new implementation does not. I think we should follow the existing convention.
From what I can tell, the per-chunk work is handed out to the pool, while the per-message count is still done on the caller's thread. That suggests to me that this will have the most benefit for large messages with lots of chunks, rather than lots of small messages. Have you measured the performance of this vs. the default reader with lots of small messages?
I'm somewhat reluctant to commit to maintaining another reader implementation. What do you think it would take to consider making this the one and only reader implementation?

clalancette · 2026-06-18T14:30:42Z

Is this file needed in this PR? I don't see it being used anywhere in the implementation or the tests. It also does things like log directly to stderr, and references things like "PJ4", the "ObjectStore", etc. which aren't in the mcap library. Can you explain more about this file?

clalancette · 2026-06-18T14:40:48Z

+  Adapt,                 // raise the effective budget up to the floor (never deadlocks;
+                         // may exceed the user cap — flagged in BudgetDecision)
+  Strict,                // do not exceed the user cap; report infeasible (caller decides)
+  EvictAndReDecompress,  // honor a sub-floor cap by evicting + re-decompressing chunks


I might be wrong, but it doesn't seem like the EvictAndReDecompress mode is implemented. resolveBudget computes for it, but nothing ever looks at requiresEviction or estReDecompressionFactor. Can you explain more about it?

clalancette · 2026-06-18T14:56:21Z

+// Back-pressure policy for the parallel reader's resident decompressed memory.
+enum class MemoryCapMode {
+  ByteBudget,  // default: cap resident decompressed BYTES (precise memory ceiling)
+  ChunkCount,  // opt-in: cap the NUMBER of concurrently-live chunks (coarser bound)


It's not really clear to me whether we should bother implementing the ChunkCount right now. It's not a huge additional amount of code, but the benefits are unclear. Can you make an argument for whether we should keep it?

clalancette · 2026-06-18T15:11:55Z

+  // Blocks until `n` credits can be granted. If `n > capacity`, blocks until the
+  // semaphore is fully replenished (available_ == capacity_), then grants it,
+  // driving available_ negative so all other acquirers wait.
+  void acquire(uint64_t n) {
+    std::unique_lock<std::mutex> lk(m_);
+    const int64_t need = int64_t(n);
+    const int64_t threshold = std::min<int64_t>(need, int64_t(capacity_));
+    cv_.wait(lk, [&] {
+      return available_ >= threshold;
+    });
+    available_ -= need;
+  }


Other than in tests, is acquire actually used anywhere?

it will be in a follow up PR

clalancette · 2026-06-18T15:20:41Z

+        if (!rr.status().ok()) {
+          rc->status = rr.status();
+          break;
+        }


I think this condition is unnecessary; in reader.inl, if a record is failed to be read, the loop would have exited. Checking the status again here isn't bad, but probably unnecessary.

clalancette · 2026-06-18T15:23:19Z

+      if (rc->status.ok() && !gotChunk) {
+        rc->status = Status{StatusCode::InvalidChunkOffset, "no chunk record at planned offset"};
+      }
+      if (rc->status.ok()) {


Combined with the last comment, I think this can lead to some data loss. If we successfully decompress a chunk, in in the next iteration we get a ReadRecord failure (rr->next returns nullopt), then we ignore that error. I think we need an additional check on rr->status() here to make sure that doesn't happen.

A standalone, log-time-ordered parallel MCAP reader. It decompresses chunks on a thread pool ahead of a k-way merge frontier and emits messages in exactly the same order as the serial IndexedMessageReader (verified by parity tests). Purely additive: adds mcap::ParallelReader (+ a concurrent file reader) without changing McapReader; the only touch to an existing type is IReadable::supportsConcurrentRead() (defaults false). Define MCAP_NO_PARALLEL to omit it. Components: - parallel_reader.hpp concurrent chunk decompression + ordered single-consumer k-way merge. - concurrent_file_reader.hpp the open(path) source: positioned reads -- pread (POSIX) / ReadFile with OVERLAPPED offset (Windows). The offset is passed per call, so many workers read concurrently with no shared cursor, through the page cache. - thread_pool.hpp, byte_semaphore.hpp, parallel_budget.hpp worker pool + memory back-pressure. - message_byte_store.hpp deferred per-message byte access (a retained message view pins only its message, not the whole chunk). - test/parallel_reader_test.cpp parity + behavior tests. Memory back-pressure is selectable via ParallelReadOptions::memoryCap: ByteBudget (default, precise resident-byte ceiling) or ChunkCount (opt-in, cap live chunks). Default worker threads is 4 (capped at 8) rather than hardware_concurrency(), which benchmarks showed was consistently slower (reads are usually consumer-bound and decompression saturates memory bandwidth before all cores). I/O uses positioned reads (pread/ReadFile) through the page cache: the offset is passed per call, so workers read concurrently with no shared cursor, reads stay cache-resident (a full read does ~0 major page faults), RSS stays low without any extra RSS-bounding machinery, and a full read is faster than the serial reader even single-threaded on a ~17 GB file. Read-path micro-optimizations (profiled with perf): resolve channel/schema by const-ref from one-time snapshots instead of McapReader::channel()/schema() returning shared_ptr by value (removes per-message atomic refcount churn), and order each chunk's entries by k-way merging the already-sorted per-channel MessageIndex runs (fallback to std::sort for reverse/non-monotonic/single-run). Emit order is unchanged (parity-verified). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

- parallel_reader: surface a RecordReader failure that ends the per-chunk loop after a successful Chunk decompress. Previously a nullopt return from rr.next() with a non-ok status was masked by gotChunk=true, dropping the read error. Also removes the now-redundant inner status check that could never fire (the loop's rec.has_value() condition already gates it). - message_byte_store.hpp: removed. It was an experimental lazy-byte layer for a downstream consumer (PJ4 / ObjectStore), not used by any mcap library or test and inappropriate for this PR. - parallel_budget: removed the EvictAndReDecompress policy. Its branch computed requiresEviction / estReDecompressionFactor but the reader never read those fields, so the mode was dead. Dropped the matching BudgetDecision fields and unit test. - byte_semaphore: removed the blocking acquire() (and its orphaned condition_variable). The reader uses only tryAcquire + forceAcquire; acquire() was test-only. Migrated the two tests to tryAcquire / forceAcquire so the same invariants are covered. Verified end-to-end on a real dexory MCAP (84 MB, 1.28 M msgs, 162 topics): identical message counts and payload bytes across before/after; latency distributions overlap (parallel-8 /tf: median 0.048 s before vs 0.051 s after over 30 iterations, identical min, IQRs overlap). The 9 Catch2 test cases (1.66 M assertions) pass against the updated headers. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Address two of clalancette's review points on the ParallelReader: - Split parallel_reader.hpp into declarations + parallel_reader.inl (function bodies), included under MCAP_IMPLEMENTATION like the rest of the C++ library (reader.hpp/reader.inl, writer.hpp/writer.inl). No behavior change; matches the established convention. - Remove the ChunkCount memory-cap mode. ByteBudget already provides a precise, portable hard memory ceiling (resident decompressed bytes); ChunkCount only bounded the live-chunk count, a coarser ~cap*max-chunk proxy. Both modes performed the identical single mutex-guarded integer acquire, so the "faster on chunk-dense layouts" claim did not hold. Dropped MemoryCapMode, ParallelReadOptions::memoryCap, and maxLiveChunks. Verified: all build variants compile clean with -Werror (default, MCAP_COMPRESSION_NO_*, MCAP_NO_PARALLEL); parallel-reader-test passes 1,660,027 assertions across 9 cases (zstd+lz4) and 71,547 (no-compression). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Rebasing the ParallelReader branch onto main (which migrated C++ from Conan 1 to Conan 2, foxglove#1706) merged textually but left the three branch-added test targets linking the now-empty ${CONAN_LIBS}, so the mcap::mcap include dirs were dropped and every test failed with "'mcap/mcap.hpp' file not found" (conformance-cpp, cpp-windows). - test/CMakeLists.txt: link parallel-reader-test{,-nocompress} and unit-tests-noparallel against ${MCAP_TEST_LIBS} (Catch2 + nlohmann_json + mcap::mcap), matching the migrated targets; keep Threads::Threads. - parallel_reader_test.cpp: clang-format the TEST_CASE line that the prior commit left wrapped (fixes the ci-format-check step). Verified: clang-format clean across cpp/; parallel-reader-test passes 1,660,027 assertions / 9 cases. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

facontidavide · 2026-06-22T10:55:41Z

I think the two readers don't really overlap, since some platform may not support what the parallel one requires.

But the final decision is up to you.
I addressed the other comments

facontidavide requested review from gasmith, james-rms and jtbandes as code owners June 9, 2026 07:47

facontidavide changed the title ~~feat(cpp): multithreaded ParallelReader~~ Multi-threaded ParallelReader Jun 9, 2026

facontidavide marked this pull request as draft June 9, 2026 15:50

facontidavide force-pushed the feature/parallel-reader branch from 3076831 to 28eec56 Compare June 9, 2026 16:03

facontidavide changed the title ~~Multi-threaded ParallelReader~~ [cpp] Multi-threaded ParallelReader Jun 9, 2026

facontidavide force-pushed the feature/parallel-reader branch 2 times, most recently from 1351bf8 to 4aad7cf Compare June 9, 2026 17:42

facontidavide marked this pull request as ready for review June 9, 2026 17:53

facontidavide force-pushed the feature/parallel-reader branch from 4aad7cf to 9afa1ca Compare June 9, 2026 17:56

amacneil requested a review from clalancette June 12, 2026 05:15

clalancette reviewed Jun 18, 2026

View reviewed changes

facontidavide and others added 4 commits June 22, 2026 12:34

facontidavide force-pushed the feature/parallel-reader branch from 5802660 to 01b5bb3 Compare June 22, 2026 10:37

Navigation Menu

Uh oh!

[cpp] Multi-threaded ParallelReader - #1703

[cpp] Multi-threaded ParallelReader#1703
facontidavide wants to merge 4 commits into
foxglove:mainfrom
facontidavide:feature/parallel-reader

facontidavide commented Jun 9, 2026 •

edited

Loading

Uh oh!

clalancette left a comment

Uh oh!

clalancette Jun 18, 2026

Uh oh!

clalancette Jun 18, 2026

Uh oh!

facontidavide Jun 22, 2026

Uh oh!

clalancette Jun 18, 2026

Uh oh!

facontidavide Jun 22, 2026

Uh oh!

clalancette Jun 18, 2026

Uh oh!

facontidavide Jun 22, 2026

Uh oh!

clalancette Jun 18, 2026

Uh oh!

clalancette Jun 18, 2026

Uh oh!

facontidavide Jun 22, 2026

Uh oh!

facontidavide commented Jun 22, 2026

Uh oh!

Reviewers

Assignees

Labels

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

facontidavide commented Jun 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

What's included

I/O: positioned reads through the page cache

Memory back-pressure + thread default

Per-message read-path optimizations (profiled with perf)

Uh oh!

clalancette left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

facontidavide commented Jun 22, 2026

Uh oh!

Reviewers

Assignees

Labels

Milestone

Development

Uh oh!

2 participants

facontidavide commented Jun 9, 2026 •

edited

Loading

Per-message read-path optimizations (profiled with `perf`)