Repository navigation
fix(group): cap metric memory with per-thread accumulators (#285) - #287
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #287 +/- ##
==========================================
- Coverage 90.03% 90.03% -0.01%
==========================================
Files 122 122
Lines 61791 61811 +20
==========================================
+ Hits 55632 55650 +18
- Misses 6159 6161 +2 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
…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.
|
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 41 minutes and 2 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 (2)
📝 WalkthroughWalkthroughThe PR changes metrics aggregation in the UMI grouping command from a global lock-free queue of per-position-group hashmaps to per-thread accumulator slots protected by mutexes. Each worker serializes directly into a stable thread-assigned accumulator; final metrics are produced by merging all accumulator slots. A changelog entry documents the memory-usage fix. 🚥 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 review |
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/lib/commands/group.rs (1)
1414-1420: Add a threaded metrics regression test.This is the new correctness path, but the shown
--metricstests use the single-threaded fast path. Please add or parameterize aThreadingOptions::new(...)metrics test that asserts family sizes, grouping metrics, and position-group sizes match the single-threaded output.Also applies to: 1549-1573
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/commands/group.rs` around lines 1414 - 1420, Add a threaded regression test variant that uses ThreadingOptions::new(...) to exercise the multi-threaded metrics path (the code using group_thread_slot, accumulators_clone, and the record_group call that consumes processed.family_sizes and processed.filter_metrics) and assert that family sizes, grouping metrics, and position-group sizes equal the existing single-threaded (--metrics) output; implement this by parameterizing the existing metrics test to run both single-threaded and threaded configurations and compare their outputs, and apply the same parameterization/update for the other metrics test block referenced around the second region (the logic near lines 1549-1573) so both test cases validate parity between threading modes.
🤖 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/group.rs`:
- Around line 1414-1420: Add a threaded regression test variant that uses
ThreadingOptions::new(...) to exercise the multi-threaded metrics path (the code
using group_thread_slot, accumulators_clone, and the record_group call that
consumes processed.family_sizes and processed.filter_metrics) and assert that
family sizes, grouping metrics, and position-group sizes equal the existing
single-threaded (--metrics) output; implement this by parameterizing the
existing metrics test to run both single-threaded and threaded configurations
and compare their outputs, and apply the same parameterization/update for the
other metrics test block referenced around the second region (the logic near
lines 1549-1573) so both test cases validate parity between threading modes.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 4b39da25-ed1f-4873-875e-0190ff40244e
📒 Files selected for processing (2)
CHANGELOG.mdsrc/lib/commands/group.rs
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/lib/commands/group.rs (1)
2644-2770: Add one discarded template to this parity test.This test would still pass if
GroupMetricsAccumulator::record_group()stopped mergingfilter_metrics, because all discard counters are expected to be zero. Add a low-MAPQ or N-containing UMI case and assert the non-zero discard counter across threading modes.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/commands/group.rs` around lines 2644 - 2770, The test test_metrics_parity_across_threading_modes currently has all discard counters asserted as zero, so it would not catch regressions in GroupMetricsAccumulator::record_group merging of filter_metrics; modify the test to add one extra template that will be discarded (e.g., create an additional pair with either a low MAPQ or a UMI containing 'N') before create_test_bam(records) and then update the grouping assertions to expect that this template increments the appropriate discard counter (reference grouping[0].discarded_poor_alignment or grouping[0].discarded_ns_in_umi), adjust total_records and accepted_records accordingly while leaving total_families unchanged, so the parity check across threading modes includes a non-zero discard metric.
🤖 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/group.rs`:
- Around line 2644-2770: The test test_metrics_parity_across_threading_modes
currently has all discard counters asserted as zero, so it would not catch
regressions in GroupMetricsAccumulator::record_group merging of filter_metrics;
modify the test to add one extra template that will be discarded (e.g., create
an additional pair with either a low MAPQ or a UMI containing 'N') before
create_test_bam(records) and then update the grouping assertions to expect that
this template increments the appropriate discard counter (reference
grouping[0].discarded_poor_alignment or grouping[0].discarded_ns_in_umi), adjust
total_records and accepted_records accordingly while leaving total_families
unchanged, so the parity check across threading modes includes a non-zero
discard metric.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 57859605-633f-4354-944a-fe90b8fc9397
📒 Files selected for processing (2)
CHANGELOG.mdsrc/lib/commands/group.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- CHANGELOG.md
The multi-threaded `group` pipeline accumulated one `CollectedMetrics` struct per position group in a shared `SegQueue`, drained only after the pipeline completed. Each entry held an `AHashMap<usize, u64>` with `with_capacity(50)` (~1 KB even when empty), so a 500 M-pair run accumulated ~10 GB just for metrics before OOMing. Replace the `SegQueue` with an `Arc<Vec<Mutex<GroupMetricsAccumulator>>>` sized to `num_threads`. Each worker claims a slot via a TLS-cached index from a process-wide atomic counter and merges per-position-group counts into that slot immediately, so retained memory is now O(threads × distinct family/position-group sizes) rather than O(position groups). Closes #285.
…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.
…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.
…lection (#290) * refactor(pipeline): extract PerThreadAccumulator helper and adopt in 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. * fix(filter): cap metric memory with PerThreadAccumulator 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). * fix(clip): cap metric memory with PerThreadAccumulator Replaces `Arc<SegQueue<CollectedClipMetrics>>` with a `PerThreadAccumulator`. See issue #285 for the shared pattern. * fix(correct): cap metric memory with PerThreadAccumulator 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. * fix(simplex): cap metric memory with PerThreadAccumulator 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. * fix(duplex): cap metric memory with PerThreadAccumulator Mirrors the simplex change: replaces `Arc<SegQueue<CollectedDuplexMetrics>>` with a `PerThreadAccumulator`. Rejects buffering semantics preserved. See #285. * fix(codec): cap metric memory with PerThreadAccumulator Mirrors the simplex/duplex changes: replaces `Arc<SegQueue<CollectedCodecMetrics>>` with a `PerThreadAccumulator`. Rejects buffering semantics preserved. See #285. * docs(changelog): note PerThreadAccumulator rollout across pipeline commands
This branch was previously deployed
1 participant
Summary
Arc<SegQueue<CollectedMetrics>>infgumi groupwithArc<Vec<Mutex<GroupMetricsAccumulator>>>, sized tonum_threads.AHashMap::with_capacity(50)allocation that was the dominant consumer at ~10 GB on a 500 M-pair input (issuegroupOOMs at ~30 GB RSS on ~500 M-pair real data, regardless of--queue-memory#285).Why
On a 500 M-pair CODEC BAM (~38 GB) on a 30 GiB box,
fgumi group --strategy adjacencyOOMed at ~30 GB RSS after ~170–180 M records regardless of--queue-memoryor--threadssettings. Static analysis pinned the dominant consumer to theSegQueueofCollectedMetrics, which held oneAHashMap<usize, u64>(~1 KB backing table even when empty) per position group and was only drained at pipeline end. Downstreamwrite_all_metricsonly needs the aggregated counters — the per-group objects were pure scratch.The single-threaded path already merges incrementally; this change brings the pipeline path in line with that.
What's next
SegQueue<CollectedXxxMetrics>) exists insimplex,duplex,codec,filter,clip,correct. Stacked follow-up PR will cover the tractable cases; the consensus commands additionally buffer rejected records as pipeline output, which needs a careful migration.Test plan
cargo ci-fmtcargo ci-lintcargo nextest run— all 225 group-touching tests passCollectedMetricsno longer dominatesGroupReadsByUmibyte-identity check on a shared-scale input