Repository navigation
refactor(pipeline): PerThreadAccumulator for all per-batch metric collection - #290
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #290 +/- ##
=======================================
Coverage 90.03% 90.04%
=======================================
Files 122 123 +1
Lines 61811 61894 +83
=======================================
+ Hits 55652 55732 +80
- Misses 6159 6162 +3 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
Rejects are now written to the rejects BAM directly from serialize_fn via a shared BGZF writer guarded by a parking_lot mutex. The mutex only serializes the byte append; BGZF compression runs on the writer's own thread pool. This removes the last unbounded Vec<Vec<u8>> growth path from simplex, complementing the metric-memory fix in #285/#290. Peak reject memory drops from proportional to the reject count to a small fixed working set of the BGZF block buffer.
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
|
Warning Rate limit exceeded
Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 18 minutes and 21 seconds. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (5)
📝 WalkthroughWalkthroughAdded a new public module 🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
src/lib/per_thread_accumulator.rs (2)
28-31: Process-wide TLS index is intentional — worth a sharper doc.
SLOT_COUNTERandTHREAD_SLOTare shared across everyPerThreadAccumulatorin the process. The module docs call this out, but the practical consequence is: for a long-lived thread pool that services multiple accumulators with mismatchednum_slots, theidx % slot_countmapping can cluster several threads onto the same slot deterministically (not random). Correctness is fine (mutex handles it), throughput isn't — just lock contention.Not blocking; a one-liner noting "distribution is only uniform when
num_slots≥ the count of distinct global indices ever assigned" would save a future reader the analysis.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/per_thread_accumulator.rs` around lines 28 - 31, Update the module-level documentation for PerThreadAccumulator to explicitly state that SLOT_COUNTER and THREAD_SLOT are process-global and that thread-to-slot mapping uses a global index (SLOT_COUNTER -> THREAD_SLOT) so that idx % num_slots can lead to deterministic clustering when num_slots is smaller than the total number of distinct global indices assigned; add a one-line note near the PerThreadAccumulator docs mentioning "distribution is only uniform when num_slots ≥ the count of distinct global indices ever assigned" and reference SLOT_COUNTER and THREAD_SLOT to clarify the behaviour and performance implication.
55-70:with_slotbody is tight — nit on readability.The
c.get().unwrap_or_else(...)+ trailing% slot_countis correct but reads as two unrelated operations glued together. Extracting the index acquisition into a private helper (e.g.,fn global_thread_index() -> usize) would makewith_slota two-liner and make the global-state dependency explicit at the call site. Optional.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/per_thread_accumulator.rs` around lines 55 - 70, Extract the thread index acquisition logic from with_slot into a small private helper fn global_thread_index() -> usize that encapsulates THREAD_SLOT and SLOT_COUNTER access (using c.get().unwrap_or_else(...) and SLOT_COUNTER.fetch_add(...)), then simplify with_slot to compute slot_count, call global_thread_index() % slot_count to pick the slot, lock self.slots[slot], and call f(&mut guard); keep all existing ordering/unwrap behavior and names (with_slot, THREAD_SLOT, SLOT_COUNTER) unchanged so behavior is identical but readability improves.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/lib/per_thread_accumulator.rs`:
- Around line 87-92: The fallback branch in into_slots can silently drop
concurrent writes because mem::take replaces slot values with A::default(), so
either make the behavior explicit in the docs and mark it lossy or enforce
unique ownership; specifically, update the doc comment on into_slots to state
the fallback is lossy under concurrent use, and add a
debug_assert!(Arc::strong_count(&self) == 1) (or similar unique-ownership
assert) before attempting Arc::try_unwrap in into_slots (referencing the slots
field and with_slot callers) so during development you catch misuse rather than
silently losing data.
---
Nitpick comments:
In `@src/lib/per_thread_accumulator.rs`:
- Around line 28-31: Update the module-level documentation for
PerThreadAccumulator to explicitly state that SLOT_COUNTER and THREAD_SLOT are
process-global and that thread-to-slot mapping uses a global index (SLOT_COUNTER
-> THREAD_SLOT) so that idx % num_slots can lead to deterministic clustering
when num_slots is smaller than the total number of distinct global indices
assigned; add a one-line note near the PerThreadAccumulator docs mentioning
"distribution is only uniform when num_slots ≥ the count of distinct global
indices ever assigned" and reference SLOT_COUNTER and THREAD_SLOT to clarify the
behaviour and performance implication.
- Around line 55-70: Extract the thread index acquisition logic from with_slot
into a small private helper fn global_thread_index() -> usize that encapsulates
THREAD_SLOT and SLOT_COUNTER access (using c.get().unwrap_or_else(...) and
SLOT_COUNTER.fetch_add(...)), then simplify with_slot to compute slot_count,
call global_thread_index() % slot_count to pick the slot, lock self.slots[slot],
and call f(&mut guard); keep all existing ordering/unwrap behavior and names
(with_slot, THREAD_SLOT, SLOT_COUNTER) unchanged so behavior is identical but
readability improves.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: db69432a-f124-4929-bf4e-7a96a2b734da
📒 Files selected for processing (10)
CHANGELOG.mdsrc/lib/commands/clip.rssrc/lib/commands/codec.rssrc/lib/commands/correct.rssrc/lib/commands/duplex.rssrc/lib/commands/filter.rssrc/lib/commands/group.rssrc/lib/commands/simplex.rssrc/lib/mod.rssrc/lib/per_thread_accumulator.rs
7b092d8 to
68c17b7
Compare
3072739 to
e7ed89c
Compare
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/lib/per_thread_accumulator.rs`:
- Line 1: Add the crate-level attribute to deny unsafe code in this module by
inserting #![deny(unsafe_code)] at the very top of
src/lib/per_thread_accumulator.rs (place it before the existing module doc
comment `//! Sharded per-thread accumulator...`). Ensure the attribute is the
first line so the compiler enforces no unsafe code for this file (no other code
changes required).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 01001777-07d1-4778-abc5-9f01162206c6
📒 Files selected for processing (10)
CHANGELOG.mdsrc/lib/commands/clip.rssrc/lib/commands/codec.rssrc/lib/commands/correct.rssrc/lib/commands/duplex.rssrc/lib/commands/filter.rssrc/lib/commands/group.rssrc/lib/commands/simplex.rssrc/lib/mod.rssrc/lib/per_thread_accumulator.rs
✅ Files skipped from review due to trivial changes (2)
- src/lib/mod.rs
- src/lib/commands/simplex.rs
🚧 Files skipped from review as they are similar to previous changes (4)
- CHANGELOG.md
- src/lib/commands/duplex.rs
- src/lib/commands/correct.rs
- src/lib/commands/clip.rs
…group Introduces `crate::per_thread_accumulator::PerThreadAccumulator<A>`, a sharded accumulator backed by a Vec of mutex-protected slots. Workers claim a slot lazily via TLS-cached indices from a process-wide atomic counter, then merge per-group results into the slot immediately. After the pipeline completes, callers fold the slots into the final aggregate. This generalizes the inline accumulator introduced for `fgumi group` (#285, #287) so the same pattern can replace `SegQueue<CollectedXxxMetrics>` in the other pipeline commands without re-implementing the slot-assignment machinery each time. Migrates `fgumi group` to the shared helper.
Replaces `Arc<SegQueue<CollectedFilterMetrics>>` with a `PerThreadAccumulator` sized to the pipeline's worker count. Each worker merges per-batch counts into its slot directly, eliminating the unbounded queue that previously retained one metrics struct per batch for the life of the run (see issue #285).
Replaces `Arc<SegQueue<CollectedClipMetrics>>` with a `PerThreadAccumulator`. See issue #285 for the shared pattern.
e7ed89c to
cadc96c
Compare
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/lib/commands/correct.rs (1)
975-985: Minor: avoid key clone and duplicated merge logic.
entry(umi.clone()).or_insert_with(|| UmiCorrectionMetrics::new(umi))clones the key even on hit. Swap toor_insert_with_keyto clone only on insert. The same merge shape is repeated verbatim at lines 1031–1039; a tiny helper would deduplicate.♻️ Suggested tweak
- for (umi, counts) in umi_matches { - let entry = m - .umi_matches - .entry(umi.clone()) - .or_insert_with(|| UmiCorrectionMetrics::new(umi)); - entry.total_matches += counts.total_matches; - entry.perfect_matches += counts.perfect_matches; - entry.one_mismatch_matches += counts.one_mismatch_matches; - entry.two_mismatch_matches += counts.two_mismatch_matches; - entry.other_matches += counts.other_matches; - } + for (umi, counts) in umi_matches { + merge_umi_counts(&mut m.umi_matches, umi, &counts); + }With a free function:
fn merge_umi_counts( dst: &mut AHashMap<String, UmiCorrectionMetrics>, umi: String, counts: &UmiCorrectionMetrics, ) { let entry = dst.entry(umi).or_insert_with_key(|k| UmiCorrectionMetrics::new(k.clone())); entry.total_matches += counts.total_matches; entry.perfect_matches += counts.perfect_matches; entry.one_mismatch_matches += counts.one_mismatch_matches; entry.two_mismatch_matches += counts.two_mismatch_matches; entry.other_matches += counts.other_matches; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/commands/correct.rs` around lines 975 - 985, The loop clones keys even on entry hits and duplicates merge logic; replace the current entry(umi.clone()).or_insert_with(...) pattern by using entry(umi).or_insert_with_key(|k| UmiCorrectionMetrics::new(k.clone())) (or call a tiny helper) so the key is only cloned on insert, and factor the repeated accumulation into a small function such as merge_umi_counts(dst: &mut AHashMap<String, UmiCorrectionMetrics>, umi: String, counts: &UmiCorrectionMetrics) that performs the entry lookup with or_insert_with_key and increments total_matches, perfect_matches, one_mismatch_matches, two_mismatch_matches, and other_matches to deduplicate the logic used at both sites.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@src/lib/commands/correct.rs`:
- Around line 975-985: The loop clones keys even on entry hits and duplicates
merge logic; replace the current entry(umi.clone()).or_insert_with(...) pattern
by using entry(umi).or_insert_with_key(|k| UmiCorrectionMetrics::new(k.clone()))
(or call a tiny helper) so the key is only cloned on insert, and factor the
repeated accumulation into a small function such as merge_umi_counts(dst: &mut
AHashMap<String, UmiCorrectionMetrics>, umi: String, counts:
&UmiCorrectionMetrics) that performs the entry lookup with or_insert_with_key
and increments total_matches, perfect_matches, one_mismatch_matches,
two_mismatch_matches, and other_matches to deduplicate the logic used at both
sites.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: b6c0ec55-b33e-4133-b9da-df1092b217ae
📒 Files selected for processing (10)
CHANGELOG.mdsrc/lib/commands/clip.rssrc/lib/commands/codec.rssrc/lib/commands/correct.rssrc/lib/commands/duplex.rssrc/lib/commands/filter.rssrc/lib/commands/group.rssrc/lib/commands/simplex.rssrc/lib/mod.rssrc/lib/per_thread_accumulator.rs
✅ Files skipped from review due to trivial changes (2)
- src/lib/mod.rs
- src/lib/per_thread_accumulator.rs
🚧 Files skipped from review as they are similar to previous changes (4)
- CHANGELOG.md
- src/lib/commands/simplex.rs
- src/lib/commands/clip.rs
- src/lib/commands/filter.rs
Replaces `Arc<SegQueue<CollectedCorrectMetrics>>` with a `PerThreadAccumulator`. Each worker merges per-batch template counts, per-UMI match counts, and rejected records into its own slot. Rejected records are drained by-move from the slot at pipeline end, preserving the existing write-all-rejects-after-pipeline semantics. See issue #285.
Replaces `Arc<SegQueue<CollectedSimplexMetrics>>` with a `PerThreadAccumulator`. Each worker merges per-batch consensus stats, overlapping-correction stats, group counts, and raw reject bytes into its own slot. Rejects buffering semantics are preserved; streaming rejects to disk during the pipeline is tracked as a separate follow-up. See #285.
Mirrors the simplex change: replaces `Arc<SegQueue<CollectedDuplexMetrics>>` with a `PerThreadAccumulator`. Rejects buffering semantics preserved. See #285.
Mirrors the simplex/duplex changes: replaces `Arc<SegQueue<CollectedCodecMetrics>>` with a `PerThreadAccumulator`. Rejects buffering semantics preserved. See #285.
cadc96c to
ee7ac26
Compare
Rejects are now written to the rejects BAM directly from serialize_fn via a shared BGZF writer guarded by a parking_lot mutex. The mutex only serializes the byte append; BGZF compression runs on the writer's own thread pool. This removes the last unbounded Vec<Vec<u8>> growth path from simplex, complementing the metric-memory fix in #285/#290. Peak reject memory drops from proportional to the reject count to a small fixed working set of the BGZF block buffer.
Rejects are now written to the rejects BAM directly from serialize_fn via a shared BGZF writer guarded by a parking_lot mutex. The mutex only serializes the byte append; BGZF compression runs on the writer's own thread pool. This removes the last unbounded Vec<Vec<u8>> growth path from simplex, complementing the metric-memory fix in #285/#290. Peak reject memory drops from proportional to the reject count to a small fixed working set of the BGZF block buffer.
Rejects are now written to the rejects BAM directly from serialize_fn via a shared BGZF writer guarded by a parking_lot mutex. The mutex only serializes the byte append; BGZF compression runs on the writer's own thread pool. This removes the last unbounded Vec<Vec<u8>> growth path from simplex, complementing the metric-memory fix in #285/#290. Peak reject memory drops from proportional to the reject count to a small fixed working set of the BGZF block buffer.
…#293) * fix(simplex): stream rejects to disk instead of buffering in memory Rejects are now written to the rejects BAM directly from serialize_fn via a shared BGZF writer guarded by a parking_lot mutex. The mutex only serializes the byte append; BGZF compression runs on the writer's own thread pool. This removes the last unbounded Vec<Vec<u8>> growth path from simplex, complementing the metric-memory fix in #285/#290. Peak reject memory drops from proportional to the reject count to a small fixed working set of the BGZF block buffer. * fix(duplex): stream rejects to disk instead of buffering in memory Same pattern as the simplex change: a shared BGZF writer guarded by a parking_lot mutex receives reject bytes directly from serialize_fn, so rejected raw BAM records no longer accumulate in per-thread Vecs. * fix(codec): stream rejects to disk instead of buffering in memory Same pattern as the simplex/duplex changes: shared BGZF writer guarded by a parking_lot mutex receives reject bytes directly from serialize_fn. * fix(correct): stream rejects to disk instead of buffering in memory Rejected raw records are now written to the rejects BAM directly from serialize_fn via a shared BGZF writer guarded by a parking_lot mutex, completing the rejects-streaming work started in simplex/duplex/codec. Removes the last per-thread Vec<Vec<u8>> accumulator in the pipeline commands.
This branch was previously deployed
1 participant
Summary
Stacked on #287. Generalizes the group-specific accumulator into a reusable
`PerThreadAccumulator` helper and rolls it out across the remaining
pipeline commands that previously used `Arc<SegQueue>`
to buffer one metrics struct per batch for the whole run.
Commits:
Why
Issue #285 showed that the `SegQueue` pattern retained one metrics struct per batch for the life of the run, scaling memory with input size. Group was the acute case (~10 GB at 180 M records via eager `AHashMap` allocations) and is the sole known OOM so far, but the other six commands share the same unbounded-growth shape and should be audited together — `filter` and `clip` leak tens-of-MB per billion records, `correct` can retain a per-UMI hashmap per batch, and the consensus commands retain reject bytes that scale with both batches and rejection rate.
This PR does not fix reject-buffering in `simplex`/`duplex`/`codec` — those still collect rejected raw BAM bytes into per-thread `Vec<Vec>`s and write them after the pipeline. Streaming rejects to the rejects BAM writer during the pipeline is a separate follow-up (PR #3, to be stacked on this branch).
What's next
Test plan