Repository navigation
feat(sort): per-phase --sort-threads / --merge-threads control for sort and runall - #465
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (7)
WalkthroughAdds separate sort and merge thread caps, threads them through sort execution, and bounds worker activation and spill decompression admission with new concurrency controls. ChangesPhase-specific sort concurrency
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Suggested labels
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## feat-runall #465 +/- ##
==============================================
Coverage ? 94.00%
==============================================
Files ? 110
Lines ? 49694
Branches ? 0
==============================================
Hits ? 46714
Misses ? 2980
Partials ? 0 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
888f5c0 to
2817cd2
Compare
nh13
changed the title
fgumi sort
2817cd2 to
03d5884
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/fgumi-pipeline-io/src/sort/spill_decompress.rs`:
- Around line 25-55: The permit acquisition API in try_acquire currently mutates
active but only returns a bool, which lets callers separate success from
ownership and risks leaking a decompression slot. Change try_acquire and its
callers to return the DecompressPermit on success so ownership is encoded in the
type, and keep the RAII release behavior in DecompressPermit::drop so
acquisition and release cannot diverge.
In `@crates/fgumi-sort/src/external.rs`:
- Around line 5280-5335: The new parameterized test logic in
phase_thread_split_is_byte_identical is manually looping over sort orders and
thread-split variants; convert these matrices to rstest cases so each
combination is its own test case and failures clearly identify the exact
order/split. Refactor the test into #[rstest]-driven parameters for SortOrder,
sort_threads, and merge_threads, following the repo’s Rust test convention and
applying the same pattern to the related tests in the mentioned range.
In `@crates/fgumi-sort/src/worker_pool.rs`:
- Around line 2273-2322: The test currently creates a new uncapped
SortWorkerPool for the “reactivates” check, so it never verifies that previously
capped workers wake up after set_active_workers changes. Update
active_worker_limit_caps_then_reactivates to use the same pool instance for both
phases: start with a lower active-worker limit, submit and drain one batch, then
raise the limit on that same SortWorkerPool, submit a second batch, and assert
that workers with indices >= 2 perform work only after reactivation. Keep the
existing per-thread step count checks on pipeline_stats to distinguish the
before/after behavior.
🪄 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: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 161661f2-649f-4110-b959-fc6059d02030
📒 Files selected for processing (7)
crates/fgumi-pipeline-io/src/sort/spill_decompress.rscrates/fgumi-pipeline-io/src/sort/spill_decompress/tests.rscrates/fgumi-sort-cli/src/chains.rscrates/fgumi-sort-cli/src/sort.rscrates/fgumi-sort/src/external.rscrates/fgumi-sort/src/worker_pool.rssrc/lib/pipeline/chains/builder.rs
03d5884 to
d3a4d1f
Compare
d3a4d1f to
abead83
Compare
|
@coderabbitai review |
✅ Action performedReview finished.
|
|
@coderabbitai review |
✅ Action performedReview finished.
|
|
@coderabbitai full review |
✅ Action performedFull review finished. |
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/fgumi-sort/src/external.rs (1)
1858-1860: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick winUse the Phase-1 cap for consolidation readers.
Consolidation runs from
drain_pending_spillbefore the Phase-2 switch, butmake_reader_semaphore(self.threads)can still admit the full base thread count when--sort-threadsis lower. Usephase1_threads()here to keep Phase 1 CPU bounded.Proposed fix
- let sem = make_reader_semaphore(self.threads); + let sem = make_reader_semaphore(self.phase1_threads());🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/fgumi-sort/src/external.rs` around lines 1858 - 1860, The consolidation reader semaphore in the drain_pending_spill path is using the base thread count instead of the Phase-1 limit. Update the reader setup near make_reader_semaphore in the consolidation flow to use phase1_threads() so the readers stay bounded before the Phase-2 switch, and keep the existing files_to_merge reader construction unchanged otherwise.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@crates/fgumi-sort/src/external.rs`:
- Around line 1858-1860: The consolidation reader semaphore in the
drain_pending_spill path is using the base thread count instead of the Phase-1
limit. Update the reader setup near make_reader_semaphore in the consolidation
flow to use phase1_threads() so the readers stay bounded before the Phase-2
switch, and keep the existing files_to_merge reader construction unchanged
otherwise.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: c5b1f36c-4d45-4955-a5f5-a71886d3c299
📒 Files selected for processing (7)
crates/fgumi-pipeline-io/src/sort/spill_decompress.rscrates/fgumi-pipeline-io/src/sort/spill_decompress/tests.rscrates/fgumi-sort-cli/src/chains.rscrates/fgumi-sort-cli/src/sort.rscrates/fgumi-sort/src/external.rscrates/fgumi-sort/src/worker_pool.rssrc/lib/pipeline/chains/builder.rs
abead83 to
9506f0c
Compare
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/fgumi-sort/src/external.rs`:
- Around line 1670-1677: The residual queryname Phase-1 work is still using the
fallback thread count instead of the effective sort thread count, so
`sort_threads` can be ignored. Update the Phase-1 gating/chunking logic in
`external.rs` to use `phase1_threads()` everywhere queryname residual sorting
decides whether to run serially or split work, rather than `self.threads` or
`self.sorter.threads`. Keep the existing `phase1_threads` helper as the single
source of truth and preserve/adjust any identity tests that cover chunk-count
behavior.
In `@src/lib/pipeline/chains/builder.rs`:
- Around line 3783-3807: The test sole_stage_sort_resolves_per_phase_threads is
only validating a local resolve helper that duplicates the production expression
instead of exercising add_sort, so it won’t catch regressions. Update this test
to call the real sort-thread resolution path in the builder logic (the add_sort
code path) and assert the resulting per-phase thread behavior there, or remove
this test if equivalent coverage already exists elsewhere.
🪄 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: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 1dcac63f-190a-4bf4-9446-491efa16033f
📒 Files selected for processing (7)
crates/fgumi-pipeline-io/src/sort/spill_decompress.rscrates/fgumi-pipeline-io/src/sort/spill_decompress/tests.rscrates/fgumi-sort-cli/src/chains.rscrates/fgumi-sort-cli/src/sort.rscrates/fgumi-sort/src/external.rscrates/fgumi-sort/src/worker_pool.rssrc/lib/pipeline/chains/builder.rs
9506f0c to
aa279ab
Compare
|
@coderabbitai review |
✅ Action performedReview finished.
|
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
♻️ Duplicate comments (2)
crates/fgumi-sort/src/worker_pool.rs (1)
2425-2474: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winTest reactivation on the same pool.
The “reactivates” half calls
run(6, 300), which constructs a fresh uncapped pool. Raiseset_active_workerson the sameSortWorkerPoolafter a capped batch, then assert workers>= 2only process the second batch.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/fgumi-sort/src/worker_pool.rs` around lines 2425 - 2474, The reactivation test is using a fresh SortWorkerPool for the second phase instead of reusing the same pool, so it does not verify that set_active_workers can re-enable workers after a capped run. Update active_worker_limit_caps_then_reactivates to submit an initial batch with a capped limit on one SortWorkerPool, then raise the limit on that same pool and submit a second batch, using pipeline_stats/per_thread_step_counts to assert that workers at or above the old cap only process the second batch.crates/fgumi-sort/src/external.rs (1)
5332-5337: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConvert the remaining split variants to
rstestcases.This is still a parameter matrix inside the
rstest; make each(tag, sort_t, merge_t)a#[case]so failures identify the exact split and match the repo test convention. As per coding guidelines, "Userstestfor parameterized tests in Rust test files."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/fgumi-sort/src/external.rs` around lines 5332 - 5337, The split-variant test matrix inside the `rstest` needs to be converted from the inline `(tag, sort_t, merge_t)` loop into individual `#[case]` entries so each split is reported separately. Update the parameterized test around the `tag`, `sort_t`, and `merge_t` combinations to follow the repo’s `rstest` convention, keeping the same `sort_low`, `merge_low`, `both`, and `sort_high` cases but expressed as separate cases for clearer failure reporting.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Duplicate comments:
In `@crates/fgumi-sort/src/external.rs`:
- Around line 5332-5337: The split-variant test matrix inside the `rstest` needs
to be converted from the inline `(tag, sort_t, merge_t)` loop into individual
`#[case]` entries so each split is reported separately. Update the parameterized
test around the `tag`, `sort_t`, and `merge_t` combinations to follow the repo’s
`rstest` convention, keeping the same `sort_low`, `merge_low`, `both`, and
`sort_high` cases but expressed as separate cases for clearer failure reporting.
In `@crates/fgumi-sort/src/worker_pool.rs`:
- Around line 2425-2474: The reactivation test is using a fresh SortWorkerPool
for the second phase instead of reusing the same pool, so it does not verify
that set_active_workers can re-enable workers after a capped run. Update
active_worker_limit_caps_then_reactivates to submit an initial batch with a
capped limit on one SortWorkerPool, then raise the limit on that same pool and
submit a second batch, using pipeline_stats/per_thread_step_counts to assert
that workers at or above the old cap only process the second batch.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: b0542840-87fe-441d-9ae7-da379448a127
📒 Files selected for processing (7)
crates/fgumi-pipeline-io/src/sort/spill_decompress.rscrates/fgumi-pipeline-io/src/sort/spill_decompress/tests.rscrates/fgumi-sort-cli/src/chains.rscrates/fgumi-sort-cli/src/sort.rscrates/fgumi-sort/src/external.rscrates/fgumi-sort/src/worker_pool.rssrc/lib/pipeline/chains/builder.rs
Add --sort-threads (Phase 1 = accumulate/sort/spill) and --merge-threads (Phase 2 = merge/write) worker control to the sort engine. SortWorkerPool gains an active-worker cap (active_worker_limit + set_active_workers); RawExternalSorter gains sort_threads/merge_threads builders and phase1_threads()/phase2_threads(). The monolithic sort() sizes the pool to max(phase1, phase2) and switches the active cap at the ingest->write boundary; the streaming into_*_stream constructors size their Phase-1-only pool to phase1_threads(). Capped workers still drain their per-worker held items (only new-work acquisition is blocked) so lowering the cap can never strand output. Output is byte-identical (a pure scheduling knob); memory stays tied to --threads.
…n counter In a multi-stage runall chain, Phase 2 (decompress/merge) runs as framework pipeline steps rather than on the sort worker pool, and the merge is serial, so the only parallel CPU consumer is SortSpillDecompress. Add a global atomic admission counter (with_max_concurrency) shared across all worker clones that caps concurrent decompress workers. Only the decompress branch is gated, so event forwarding can never stall the merge. Byte-identical: decompression order is deterministic regardless of concurrency.
…ortOptions Expose the per-phase thread knobs on SortOptions so standalone fgumi sort accepts --sort-threads/--merge-threads and runall forwards them as --sort::sort-threads / --sort::merge-threads via the #[multi_options] macro. Each defaults to --threads. Wire them through SortStepCaptures into the RawExternalSorter builders and log the split when set.
…g chains Resolve per-phase worker counts from the forwarded SortOptions (defaulting to --threads) in both runall sort paths. The sole-stage (monolithic) path passes them through SortStepCaptures; the streaming chain sizes the Phase-1 pool with sort_threads and caps Phase-2 decompress concurrency via SortSpillDecompress::with_max_concurrency (merge_threads).
aa279ab to
7d219fc
Compare
|
@coderabbitai review |
✅ Action performedReview finished.
|
|
@coderabbitai review |
✅ Action performedReview finished.
|
This branch was previously deployed
1 participant
Summary
Adds per-phase worker-thread control to the sort engine, exposed and respected in both
fgumi sortandfgumi runall:--sort-threads— workers for Phase 1 (accumulate / sort / spill)--merge-threads— workers for Phase 2 (merge / write)Each defaults to
--threads. Infgumi runallthey are forwarded per-stage as--sort::sort-threads/--sort::merge-threads(via the#[multi_options]macro onSortOptions).Motivation: in a producer→sort pipeline (e.g.
bwa mem | fgumi sort, or runall's align→sort chain), Phase 1 runs concurrently with the upstream producer and should cede cores, while Phase 2 typically runs after the producer finishes and can use more. These are CPU-only knobs — the memory budget stays tied to--threads, and output is byte-identical regardless of the split (a pure scheduling knob).What's covered
--sort-threads)--merge-threads)fgumi sort(standalone)runallsole-stage sortrunallmulti-stage (streaming)sort_threadsSortSpillDecompressadmission counter; the merge step is serialThe standalone and sole-stage paths use the monolithic
RawExternalSorter::sort()— one worker pool whose active cap is switched at the ingest→write boundary (capped workers still drain their per-worker held items, so lowering the cap never strands output). The streaming path decomposes sort into framework pipeline steps: Phase 1 sizes the sort/spill pool tosort_threads; Phase 2's only parallel CPU consumer (spill decompression) is bounded by a shared atomic admission counter, since the merge itself is serial.Validation
phase1 > phase2cap-lowering case — plus a unit test pinning identity across all sort orders.--sort-threads 1→ Phase-1 ≈1 core, Phase-2 full;--merge-threads 1→ Phase-1 full, Phase-2 ≈1).clippy -D warnings(pedantic) green acrossfgumi-sort,fgumi-sort-cli,fgumi-pipeline-io, andfgumi, including a multi-threaded admission-counter test.Commits (suggested reading order)
feat(sort)— per-phase worker-thread control inRawExternalSorter(fields/builders +SortWorkerPoolactive cap; monolithic + streaming Phase-1).feat(sort)— streaming Phase-2 decompress admission counter (fgumi-pipeline-io).feat(sort-cli)—--sort-threads/--merge-threadsonSortOptions(standalone flags + runall forwarding).feat(runall)— wire per-phase counts into both runall sort paths.Summary by CodeRabbit