Repository navigation
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #237 +/- ##
==========================================
+ Coverage 88.97% 88.99% +0.02%
==========================================
Files 114 119 +5
Lines 55481 57682 +2201
==========================================
+ Hits 49366 51336 +1970
- Misses 6115 6346 +231 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
|
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:
📝 WalkthroughWalkthroughThis PR adds a phase-aware, fixed-size SortWorkerPool and related modules to pipeline BGZF I/O and parallel compression/decompression. New items include PooledInputStream, RecordSource, StagingBuffer, io_writer_loop, PooledBamWriter, PooledChunkWriter, and worker-pool types (CompressJob/Result, BufferPool, Tagged blocks, phase constants). Raw sorter and read-ahead logic are updated to integrate the pool (new merge consumer, timing, and spill plumbing). Tests and integration scenarios for pooled I/O, writers, and end-to-end sorting were added. A 🚥 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: 10
🧹 Nitpick comments (1)
tests/integration/test_sort_correctness.rs (1)
61-80: Template-coordinate cases never populate the requestedCBtag.The fixture writes
RGandMI, but noCB:Z:tag, while the template-coordinate tests pass--cell-tag CB. That means these assertions only cover the missing-cell-tag fallback, not the actual cell-aware sort key. Add at least two differentCBvalues if you want this suite to exercise the--cell-tagpath.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/integration/test_sort_correctness.rs` around lines 61 - 80, The SAM fixture never includes the CB tag so tests that pass --cell-tag CB only hit the missing-cell-tag fallback; update the writeln! calls that currently add "RG:Z:rg1\tMI:Z:umi_{i}" (for paired records) and "RG:Z:rg1" (for unmapped records) to also emit a CB:Z:... tag (e.g., "CB:Z:cellA" or "CB:Z:cellB"), ensuring both R1 and R2 lines and the unmapped pairs include CB, and vary the CB value across records (at least two distinct CB values) so the test exercises the --cell-tag path and cell-aware sort key logic.
🤖 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/commands/sort.rs`:
- Around line 263-266: The x_pool_decompress flag uses clap's default SetTrue
action and default_value_t = true which prevents passing explicit false; change
its clap configuration for the x_pool_decompress field to use action =
clap::ArgAction::Set and value_parser = parse_bool (matching other toggles) so
the CLI can accept --x-pool-decompress=false while keeping the default true
behavior; update the attribute on the x_pool_decompress field to follow the
established pattern used elsewhere (use ArgAction::Set and parse_bool).
In `@src/lib/bam_io.rs`:
- Around line 777-795: Open the input once and reuse that same handle: replace
the two File::open calls with a single File::open(path_ref) into a mutable
variable, use that handle to construct the BgzfReader and call
noodles::bam::io::Reader::from(...) and read_header(), then seek the same handle
back to the start (e.g. using Seek::seek(SeekFrom::Start(0))) before wrapping it
in an io::BufReader and passing it to pool.set_input_file; keep
pool.set_phase(phase::PHASE1) as-is. Make sure to reference the existing symbols
File::open, BgzfReader::new, noodles::bam::io::Reader::read_header (or
read_header()), Seek::seek/SeekFrom::Start, pool.set_input_file, and
pool.set_phase in the change.
In `@src/lib/sort/bgzf_io.rs`:
- Around line 38-68: The staging API currently allows enqueuing oversized BGZF
jobs because buf() exposes direct appends and flush()/flush_if_full() forward
the entire Vec; fix by implementing a chunked-write path (similar to
InlineBgzfCompressor::write_all) so any single append larger than
BGZF_MAX_BLOCK_SIZE is split into BGZF_MAX_BLOCK_SIZE-sized CompressJob chunks
before submission, update buf()/or add a write_chunked(&mut self, data: &[u8])
helper that slices large inputs and pushes/submit multiple CompressJob entries,
ensure flush() is a no-op when the buffer is empty (i.e. do nothing if
self.buf.is_empty()) and keep buffer pool reuse and serial assignment semantics
(use self.next_serial for each chunk submitted and clone result_tx per
CompressJob).
- Around line 89-119: The current writer finalizes the BGZF stream even if there
are missing serials; after the recv() loop and the in-order draining loop, check
whether reorder_buf is empty and whether any expected serials remain missing
(i.e., next_expected did not advance to cover all produced serials); if
reorder_buf is non-empty or missing serials are detected, return an error (e.g.,
io::Error::new(io::ErrorKind::UnexpectedEof, "...")) instead of writing BGZF_EOF
and flushing; update the code paths around result_rx, reorder_buf,
next_expected, writer and BGZF_EOF to fail early on these conditions so
truncated streams are not silently emitted.
In `@src/lib/sort/pooled_chunk_writer.rs`:
- Around line 122-142: SpillWriteHandle currently allows silently detaching the
writer thread if dropped, so mark the struct with #[must_use] and implement Drop
for SpillWriteHandle to eagerly join the inner io_handle (the
JoinHandle<Result<()>>) if present: on drop call join() and handle errors by
logging any panic or write error (do not panic from Drop), e.g., convert
join/map_err/Result into a logged message; keep the existing public wait(self)
-> Result<()> method unchanged but ensure Drop uses the same join logic against
io_handle to avoid losing errors.
In `@src/lib/sort/raw.rs`:
- Around line 1398-1415: The pooled worker pool is being constructed with
self.temp_compression so PooledBamWriter jobs always use temp compression and
ignore output_compression(); change the design so compression for temporary BGZF
work and final BAM output are distinct: either add a separate field (e.g.,
temp_compression and output_compression) on the struct used to build
SortWorkerPool and use temp_compression for pool construction while passing
output_compression into PooledBamWriter, or include the desired compression
level in the job payload submitted to the pool (update the job type used by
SortWorkerPool/PooledBamWriter to carry compression), and ensure calls that
currently reference output_compression() (and the indexing writer path)
propagate that output compression into the writer creation path rather than
relying on self.temp_compression.
- Around line 895-928: poll_decompressed_blocks currently fully drains
self.decompressed_chunks into per_source.pending which removes the intended
backpressure; change the drain to respect a hard cap (e.g. num_workers * 8 or a
configurable MAX_PENDING_TOTAL) so we never transfer more than that many blocks
into per-source buffers at once. Implement this by tracking a running total of
buffered pending blocks while popping from self.decompressed_chunks (use the
existing decompressed_chunks.pop()), and when the cap is reached stop draining
and push the last-read block back onto decompressed_chunks (or stop before
popping it) instead of inserting into per_source.pending; keep the existing
checks on per_source.current_buf/next_serial and the EOF/unpark logic
(references: poll_decompressed_blocks, self.decompressed_chunks.pop,
per_source.pending.insert, per_source.current_buf/current_pos/next_serial,
all_chunks_eof, std::thread::park).
In `@src/lib/sort/worker_pool.rs`:
- Around line 1428-1446: The shutdown logic in fn shutdown(self) is only
executed when Arc::try_unwrap succeeds; to ensure workers are always signaled
and joined, implement Drop for the WorkerPool (or introduce a guard owned by the
pool) that performs the same teardown: call self.stats.log_summary() and
self.pipeline_stats.log_summary(), store phase::SHUTDOWN into self.shared.phase
(Ordering::Release), lock each sender in self.shared.chunk_file_senders and set
them to None (handling poisoned mutexes gracefully), and join all handles in
self.workers (or detach only after join) so no worker threads are left running
on early returns; reuse the same symbols (shutdown, shared.phase,
shared.chunk_file_senders, stats, pipeline_stats, workers) so behavior matches
the explicit shutdown path.
- Around line 1085-1090: The code currently treats any Err from
read_raw_blocks(...) as EOF by setting shared.input_eof and returning
StepResult::InputEmpty; instead capture and store the first read error in shared
state (e.g., a new shared.input_error / atomic/error-cell) rather than
collapsing it into EOF, and return a distinct error/failed StepResult so callers
(PooledInputStream / MainThreadChunkConsumer) can observe and surface the
underlying I/O error; apply the same change to the other read_raw_blocks() error
path around lines 1202-1208 so corrupt/truncated inputs are propagated rather
than silently treated as EOF.
- Around line 1154-1174: The terminal flag (shared.decompressed_input_done /
similar flags at the other sites) is being set while a thread may still be
holding the last block in worker.held_decompressed_input or while the downstream
queue still contains items; modify the EOF condition to also ensure
worker.held_decompressed_input is None (or empty) and
shared.decompressed_input.is_empty() (and raw_input_blocks.is_empty()) before
storing the done flag and unparking the main thread. Update the same pattern at
the other occurrences (around the input_blocks_decompressed / input_read_serial
checks and the all_chunks_eof logic) so we only mark decompressed/finished when
the held buffer and queues are drained, not just when counts/eof are reached.
---
Nitpick comments:
In `@tests/integration/test_sort_correctness.rs`:
- Around line 61-80: The SAM fixture never includes the CB tag so tests that
pass --cell-tag CB only hit the missing-cell-tag fallback; update the writeln!
calls that currently add "RG:Z:rg1\tMI:Z:umi_{i}" (for paired records) and
"RG:Z:rg1" (for unmapped records) to also emit a CB:Z:... tag (e.g.,
"CB:Z:cellA" or "CB:Z:cellB"), ensuring both R1 and R2 lines and the unmapped
pairs include CB, and vary the CB value across records (at least two distinct CB
values) so the test exercises the --cell-tag path and cell-aware sort key logic.
🪄 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: 4053558e-e81f-4bbc-8399-4ac35d3739ef
📒 Files selected for processing (14)
.gitignorecrates/fgumi-bgzf/src/lib.rscrates/fgumi-bgzf/src/writer.rssrc/commands/sort.rssrc/lib/bam_io.rssrc/lib/sort/bgzf_io.rssrc/lib/sort/mod.rssrc/lib/sort/pooled_bam_writer.rssrc/lib/sort/pooled_chunk_writer.rssrc/lib/sort/raw.rssrc/lib/sort/read_ahead.rssrc/lib/sort/worker_pool.rstests/integration/main.rstests/integration/test_sort_correctness.rs
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 20:34
c6c8a1a to
79461ac
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (4)
src/lib/sort/worker_pool.rs (3)
1150-1151: Panic on decompression failure is aggressive but documented.
expect()will crash the worker if input is corrupted. Consider propagating the error instead of panicking, especially since read errors are already being silently swallowed upstream.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/worker_pool.rs` around lines 1150 - 1151, The code currently uses expect on decompress_block (let data = decompress_block(&block, &mut worker.decompressor).expect(...)) which panics on corrupt input; change this to propagate the error instead: have the containing function (the worker loop function that calls decompress_block) return a Result, replace expect with the ? operator or a match that converts the decompression error into your function's error type (or a custom enum), and update callers to handle the Result rather than silently swallowing or allowing a panic; reference decompress_block, worker.decompressor and the worker loop function to locate and update the signature and call sites so errors are handled and surfaced appropriately.
1262-1263: Same panic-on-decompress issue for chunk blocks.Line 1263 uses
expect()for decompression. Consistent with line 1150-1151, but both should handle errors gracefully.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/worker_pool.rs` around lines 1262 - 1263, The BGZF decompression call using decompress_block(&tagged.block, &mut worker.decompressor).expect(...) must not panic; replace the expect with proper error handling: call decompress_block and match or use ? to propagate a Result from the surrounding function, or convert the error into the worker error-handling path (e.g., log via worker.logger/worker.error, mark the job/worker failed, and return an Err). Update the function that contains this code to return a Result if needed and ensure symbols referenced (decompress_block, tagged.block, worker.decompressor) propagate or handle decompression errors consistently with the earlier handling used around lines 1150-1151.
654-665: Jitter calculation uses low-entropy seed.
subsec_nanos()as a jitter seed provides weak randomness—multiple threads calling simultaneously get similar values. Acceptable for backoff, but consider thread ID mixing for better distribution.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/worker_pool.rs` around lines 654 - 665, The jitter seed currently uses SystemTime::now().duration_since(...).subsec_nanos() which yields low entropy across concurrent threads; modify the jitter_seed computation in this block (symbols: jitter_seed, SystemTime::now, subsec_nanos, jitter_offset, actual_us, MIN_BACKOFF_US) to mix in the current thread identifier before deriving the u64 seed (e.g., hash std::thread::current().id() together with subsec_nanos using a Hasher or use a per-thread RNG) so simultaneous threads produce more varied jitter values while keeping the subsequent jitter_offset, actual_us, and sleep logic unchanged.src/lib/sort/bgzf_io.rs (1)
50-60:flush()submits empty buffers wastefully.When
bufis empty, this still allocates a serial and submits a zero-byte compress job. Add an early return.Proposed fix
/// Flush the staging buffer: swap it with a recycled buffer and submit for compression. pub(crate) fn flush(&mut self) { + if self.buf.is_empty() { + return; + } let data = std::mem::replace(&mut self.buf, self.pool.buffer_pool.checkout());🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/bgzf_io.rs` around lines 50 - 60, The flush() method should early-return when the current buffer is empty to avoid creating a serial and submitting a zero-byte CompressJob: add a guard at the start of flush() that checks self.buf.is_empty() and returns immediately if true; otherwise keep the existing behavior (replace self.buf with pool.buffer_pool.checkout(), ensure capacity, increment self.next_serial, and call self.pool.submit_compress(CompressJob { data, serial, result_tx: self.result_tx.clone() })). Reference symbols: flush(), self.buf, self.next_serial, self.pool.submit_compress, CompressJob, result_tx.
🤖 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/sort/read_ahead.rs`:
- Around line 382-419: The EOF race happens because input_blocks_decompressed is
incremented before a worker successfully pushes into decompressed_input (blocks
can be placed into held_decompressed_input), so decompressed_input_done can
become true while held items remain; update the logic so decompressed_input_done
is only set when there are no held items across workers or ensure
held_decompressed_input is drained into decompressed_input before marking done.
Concretely: change the producer-side flow around input_blocks_decompressed,
held_decompressed_input and the code that sets decompressed_input_done so it
checks/flushes held_decompressed_input (or decrements the counter only after
successful push), and adjust is_eof/next_block to consider any
held_decompressed_input/worker-held counts before returning EOF; locate and
modify the functions/variables input_blocks_decompressed,
held_decompressed_input, decompressed_input_done, drain_queue, is_eof, and
next_block to implement this synchronized check/flush.
In `@src/lib/sort/worker_pool.rs`:
- Around line 1477-1482: Replace the silent discard of send errors for
job.result_tx.send in the worker loop: instead of using "let _ =
job.result_tx.send(CompressResult { serial: job.serial, compressed,
recycled_buf: recycled });", handle the Result and surface failures (e.g., if
let Err(e) = job.result_tx.send(...) { /* log including job.serial and error, or
increment a dropped-results metric, or otherwise track the dropped
CompressResult */ }). Ensure you reference the CompressResult payload and
job.serial in the log/metric so dropped compressed data can be diagnosed, and
avoid panicking the worker unless intended.
- Around line 1370-1399: The distribute_chunk_files function currently panics
when opening a chunk file; change its signature to return Result<(),
std::io::Error> (or a crate error type) and propagate failures instead of
calling unwrap_or_else/panic. Replace the explicit panic in the File::open
branch by using ? (or map_err to add context) so errors bubble up, ensure
callers of distribute_chunk_files are updated to handle the Result, and keep the
rest of the logic (building per_worker Vec<ChunkFileState> and sending via
shared.chunk_file_senders) unchanged; reference distribute_chunk_files,
ChunkFileState::new, and shared.chunk_file_senders when locating the change.
---
Nitpick comments:
In `@src/lib/sort/bgzf_io.rs`:
- Around line 50-60: The flush() method should early-return when the current
buffer is empty to avoid creating a serial and submitting a zero-byte
CompressJob: add a guard at the start of flush() that checks self.buf.is_empty()
and returns immediately if true; otherwise keep the existing behavior (replace
self.buf with pool.buffer_pool.checkout(), ensure capacity, increment
self.next_serial, and call self.pool.submit_compress(CompressJob { data, serial,
result_tx: self.result_tx.clone() })). Reference symbols: flush(), self.buf,
self.next_serial, self.pool.submit_compress, CompressJob, result_tx.
In `@src/lib/sort/worker_pool.rs`:
- Around line 1150-1151: The code currently uses expect on decompress_block (let
data = decompress_block(&block, &mut worker.decompressor).expect(...)) which
panics on corrupt input; change this to propagate the error instead: have the
containing function (the worker loop function that calls decompress_block)
return a Result, replace expect with the ? operator or a match that converts the
decompression error into your function's error type (or a custom enum), and
update callers to handle the Result rather than silently swallowing or allowing
a panic; reference decompress_block, worker.decompressor and the worker loop
function to locate and update the signature and call sites so errors are handled
and surfaced appropriately.
- Around line 1262-1263: The BGZF decompression call using
decompress_block(&tagged.block, &mut worker.decompressor).expect(...) must not
panic; replace the expect with proper error handling: call decompress_block and
match or use ? to propagate a Result from the surrounding function, or convert
the error into the worker error-handling path (e.g., log via
worker.logger/worker.error, mark the job/worker failed, and return an Err).
Update the function that contains this code to return a Result if needed and
ensure symbols referenced (decompress_block, tagged.block, worker.decompressor)
propagate or handle decompression errors consistently with the earlier handling
used around lines 1150-1151.
- Around line 654-665: The jitter seed currently uses
SystemTime::now().duration_since(...).subsec_nanos() which yields low entropy
across concurrent threads; modify the jitter_seed computation in this block
(symbols: jitter_seed, SystemTime::now, subsec_nanos, jitter_offset, actual_us,
MIN_BACKOFF_US) to mix in the current thread identifier before deriving the u64
seed (e.g., hash std::thread::current().id() together with subsec_nanos using a
Hasher or use a per-thread RNG) so simultaneous threads produce more varied
jitter values while keeping the subsequent jitter_offset, actual_us, and sleep
logic unchanged.
🪄 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: 8e758347-2c68-4dce-a3ee-a0881f68e406
📒 Files selected for processing (14)
.gitignorecrates/fgumi-bgzf/src/lib.rscrates/fgumi-bgzf/src/writer.rssrc/commands/sort.rssrc/lib/bam_io.rssrc/lib/sort/bgzf_io.rssrc/lib/sort/mod.rssrc/lib/sort/pooled_bam_writer.rssrc/lib/sort/pooled_chunk_writer.rssrc/lib/sort/raw.rssrc/lib/sort/read_ahead.rssrc/lib/sort/worker_pool.rstests/integration/main.rstests/integration/test_sort_correctness.rs
✅ Files skipped from review due to trivial changes (5)
- .gitignore
- crates/fgumi-bgzf/src/writer.rs
- crates/fgumi-bgzf/src/lib.rs
- src/commands/sort.rs
- src/lib/sort/pooled_bam_writer.rs
🚧 Files skipped from review as they are similar to previous changes (6)
- tests/integration/main.rs
- src/lib/sort/mod.rs
- src/lib/bam_io.rs
- tests/integration/test_sort_correctness.rs
- src/lib/sort/pooled_chunk_writer.rs
- src/lib/sort/raw.rs
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 20:54
79461ac to
850be0b
Compare
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 21:11
850be0b to
48d8c4b
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (2)
src/lib/sort/raw.rs (1)
1424-1425:⚠️ Potential issue | 🟠 Major
output_compressionis still bypassed on pooled BAM writes.The shared pool is initialized with
self.temp_compression, and every non-index final BAM path later writes throughPooledBamWriterusing that same pool. That means--output-compressiononly applies on the indexing path;--temp-compression=0will also make ordinary output BAMs uncompressed. Split temp/output compression state or carry the level in each compress job.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/raw.rs` around lines 1424 - 1425, The shared SortWorkerPool is being created with self.temp_compression so PooledBamWriter jobs always use the temp level and ignore --output-compression; change the design to carry compression level per compress job (or maintain two pools) so final BAM writes use self.output_compression while internal/temp operations use self.temp_compression. Concretely, update SortWorkerPool/CompressJob (or the job submit API used by PooledBamWriter) to accept a compression_level field and ensure PooledBamWriter submits jobs with compression_level = self.output_compression, while internal/temp submits use self.temp_compression; alternatively instantiate a separate pool for output writes named e.g. SortWorkerPool::new_output(self.output_compression) and use that for PooledBamWriter to preserve existing temp pool behavior.src/lib/bam_io.rs (1)
781-794:⚠️ Potential issue | 🟠 MajorKeep a streaming fallback for non-seekable inputs.
seek(SeekFrom::Start(0))makes pooled mode fail on FIFOs,/dev/fd/*, and process substitution. The legacy path can stream those inputs, and--x-pool-decompressnow defaults to true, so this becomes a user-visible regression. Fall back to the legacy reader or a tee/replay path when rewind is unsupported instead of hard-failing.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/bam_io.rs` around lines 781 - 794, The current code assumes the input File is seekable by calling file.seek(SeekFrom::Start(0)) after creating a BgzfReader/noodles Reader which breaks pooled mode for non-seekable inputs (FIFOs, /dev/fd/*, process substitution); change this to detect whether rewinding is supported (attempt a small seek and catch errors or use metadata/is_seekable helper) and if seek fails, preserve the legacy streaming path or switch to a tee/replay path instead of hard-failing: when a seek on the opened `file` returns an error indicating unsupported seek, skip the rewind and instantiate the legacy streaming reader (the same code path used prior to pooling) or create a temporary tee buffer to allow pool workers to read, ensuring the logic around `BgzfReader` and `noodles::bam::io::Reader` uses the streaming fallback in that case.
🧹 Nitpick comments (2)
src/lib/sort/worker_pool.rs (2)
1174-1175: Decompression failures panic the worker.Consider surfacing decompression errors via a shared error flag instead of panicking, similar to read errors.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/worker_pool.rs` around lines 1174 - 1175, The code currently panics on decompression failure by calling expect on decompress_block; instead handle the Result from decompress_block and surface errors via the same shared error mechanism used for read errors (e.g., the Atomic/Mutex error flag your pool uses), e.g. match decompress_block(&block, &mut worker.decompressor) { Ok(data) => { ... } Err(e) => { set_shared_error_flag_with(e); return/stop_this_worker_thread; } }, referencing decompress_block and worker.decompressor and ensuring the worker returns/terminates without panicking so the pool coordinator can observe and handle the error.
1229-1242: Chunk read errors surface indirectly.Unlike input read errors (which set a flag), chunk read errors mark EOF and rely on
io_writer_loopserial gap detection. This works but the error message will be less informative.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/worker_pool.rs` around lines 1229 - 1242, The chunk read error is currently hidden by only setting file_state.eof and logging; instead record and propagate the actual error so callers can surface it: when read_raw_blocks returns Err(e) set a persistent error field on the FileState (e.g., file_state.read_error or file_state.io_error) with e (or its string) in addition to setting file_state.eof and calling Self::maybe_mark_all_eof(shared); then update io_writer_loop (and any serial-gap checks) to inspect that FileState error field and return/handle the real I/O error rather than only treating it as EOF so the failure message is informative (refer to read_raw_blocks, file_state.eof, maybe_mark_all_eof, and io_writer_loop).
🤖 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/sort/pooled_bam_writer.rs`:
- Around line 90-100: write_raw_record currently pre-flushes only when the
current buffer would overflow but still creates a single oversized compression
job if 4 + record_bytes.len() > BGZF_MAX_BLOCK_SIZE; change write_raw_record to
handle records larger than BGZF_MAX_BLOCK_SIZE by chunking them the same way
headers are handled: if 4 + record_bytes.len() > BGZF_MAX_BLOCK_SIZE, iterate
over record_bytes and for each chunk compute chunk_len <= BGZF_MAX_BLOCK_SIZE -
4, pre-flush staging when needed, write chunk_len as little-endian u32 via
staging.buf().extend_from_slice(&chunk_len.to_le_bytes()), write the chunk bytes
via staging.buf().extend_from_slice(...), call staging.flush_if_full() after
each chunk, and repeat until the entire record is written (keeping existing
behavior for small records).
In `@src/lib/sort/pooled_chunk_writer.rs`:
- Around line 81-94: The pre-flush size check in write_record only accounts for
4 + record.len() and ignores the encoded key when K::EMBEDDED_IN_RECORD is
false, which can overflow BGZF_MAX_BLOCK_SIZE; update write_record to compute
the total bytes needed (include key's encoded size when !K::EMBEDDED_IN_RECORD)
and pre-flush based on that, or switch to streaming the entry using
write_chunked to write key, length prefix, and record without buffering the
whole entry; ensure staging.flush() is called before writing and preserve
staging.flush_if_full() behavior.
In `@src/lib/sort/read_ahead.rs`:
- Around line 404-423: In next_block(), after calling drain_queue() and before
returning None on is_eof(), perform a final drain and re-check the reorder
buffer for the needed serial: call drain_queue() again, then attempt
self.reorder.remove(&self.next_serial) (incrementing next_serial and returning
Some if found); only return None if that second check finds nothing. This avoids
the race where a block is pushed between the first drain_queue() and the
is_eof() check.
---
Duplicate comments:
In `@src/lib/bam_io.rs`:
- Around line 781-794: The current code assumes the input File is seekable by
calling file.seek(SeekFrom::Start(0)) after creating a BgzfReader/noodles Reader
which breaks pooled mode for non-seekable inputs (FIFOs, /dev/fd/*, process
substitution); change this to detect whether rewinding is supported (attempt a
small seek and catch errors or use metadata/is_seekable helper) and if seek
fails, preserve the legacy streaming path or switch to a tee/replay path instead
of hard-failing: when a seek on the opened `file` returns an error indicating
unsupported seek, skip the rewind and instantiate the legacy streaming reader
(the same code path used prior to pooling) or create a temporary tee buffer to
allow pool workers to read, ensuring the logic around `BgzfReader` and
`noodles::bam::io::Reader` uses the streaming fallback in that case.
In `@src/lib/sort/raw.rs`:
- Around line 1424-1425: The shared SortWorkerPool is being created with
self.temp_compression so PooledBamWriter jobs always use the temp level and
ignore --output-compression; change the design to carry compression level per
compress job (or maintain two pools) so final BAM writes use
self.output_compression while internal/temp operations use
self.temp_compression. Concretely, update SortWorkerPool/CompressJob (or the job
submit API used by PooledBamWriter) to accept a compression_level field and
ensure PooledBamWriter submits jobs with compression_level =
self.output_compression, while internal/temp submits use self.temp_compression;
alternatively instantiate a separate pool for output writes named e.g.
SortWorkerPool::new_output(self.output_compression) and use that for
PooledBamWriter to preserve existing temp pool behavior.
---
Nitpick comments:
In `@src/lib/sort/worker_pool.rs`:
- Around line 1174-1175: The code currently panics on decompression failure by
calling expect on decompress_block; instead handle the Result from
decompress_block and surface errors via the same shared error mechanism used for
read errors (e.g., the Atomic/Mutex error flag your pool uses), e.g. match
decompress_block(&block, &mut worker.decompressor) { Ok(data) => { ... } Err(e)
=> { set_shared_error_flag_with(e); return/stop_this_worker_thread; } },
referencing decompress_block and worker.decompressor and ensuring the worker
returns/terminates without panicking so the pool coordinator can observe and
handle the error.
- Around line 1229-1242: The chunk read error is currently hidden by only
setting file_state.eof and logging; instead record and propagate the actual
error so callers can surface it: when read_raw_blocks returns Err(e) set a
persistent error field on the FileState (e.g., file_state.read_error or
file_state.io_error) with e (or its string) in addition to setting
file_state.eof and calling Self::maybe_mark_all_eof(shared); then update
io_writer_loop (and any serial-gap checks) to inspect that FileState error field
and return/handle the real I/O error rather than only treating it as EOF so the
failure message is informative (refer to read_raw_blocks, file_state.eof,
maybe_mark_all_eof, and io_writer_loop).
🪄 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: 0ba1ec6a-3af2-48d9-a789-ce88c79e1e60
📒 Files selected for processing (14)
.gitignorecrates/fgumi-bgzf/src/lib.rscrates/fgumi-bgzf/src/writer.rssrc/commands/sort.rssrc/lib/bam_io.rssrc/lib/sort/bgzf_io.rssrc/lib/sort/mod.rssrc/lib/sort/pooled_bam_writer.rssrc/lib/sort/pooled_chunk_writer.rssrc/lib/sort/raw.rssrc/lib/sort/read_ahead.rssrc/lib/sort/worker_pool.rstests/integration/main.rstests/integration/test_sort_correctness.rs
✅ Files skipped from review due to trivial changes (3)
- .gitignore
- src/lib/sort/mod.rs
- crates/fgumi-bgzf/src/writer.rs
🚧 Files skipped from review as they are similar to previous changes (3)
- crates/fgumi-bgzf/src/lib.rs
- tests/integration/main.rs
- tests/integration/test_sort_correctness.rs
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 21:26
48d8c4b to
42a417d
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (4)
src/lib/sort/raw.rs (1)
1401-1402:⚠️ Potential issue | 🟠 Major
output_compression()is still bypassed on pooled BAM output.The shared pool is initialized with
self.temp_compression, and the worker compressors are fixed when that pool is built. Every non-indexPooledBamWriterpath therefore inherits temp compression unless the compression level is carried per job or the final BAM uses a separate pool.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/raw.rs` around lines 1401 - 1402, The pooled worker pool is being created with self.temp_compression, which fixes compressor settings for all PooledBamWriter jobs and bypasses output_compression(); change the pooling logic so final BAM writers use the output compression settings (call self.output_compression()) instead of temp_compression — either by creating a separate SortWorkerPool for final outputs via SortWorkerPool::new(self.threads.max(1), self.output_compression()) and passing that to PooledBamWriter, or by making SortWorkerPool able to select compression per job (propagate output_compression() into the job-level API used by PooledBamWriter) so that PooledBamWriter no longer inherits self.temp_compression.src/lib/sort/bgzf_io.rs (1)
112-145:⚠️ Potential issue | 🔴 CriticalMissing tail blocks still finalize a truncated BGZF stream.
This catches gaps only when a later serial is already buffered in
reorder_buf. If the last submitted serial never arrives,recv()ends withreorder_bufempty andnext_expectedstill short, and this loop still writesBGZF_EOF. Pass the expected block count intoio_writer_loop(or send an explicit finish sentinel) and require an exact match before finalizing.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/bgzf_io.rs` around lines 112 - 145, The io writer currently writes BGZF_EOF even if some trailing blocks never arrived because it only detects gaps when a later serial is buffered; modify the io_writer_loop to require an exact final block count (or accept an explicit finish sentinel) before writing BGZF_EOF: add an expected_total parameter (or a finish message on result_rx), track next_expected vs expected_total, and after the recv loop verify next_expected == expected_total+1 (or that the finish sentinel was received) before writing BGZF_EOF; if they differ return an error instead of finalizing. Ensure you update uses of result_rx, reorder_buf, buffer_pool.checkin, and writer.write_all to follow this contract so missing tail blocks are detected and the stream is not silently truncated.src/lib/sort/worker_pool.rs (2)
1335-1342:⚠️ Potential issue | 🔴 Critical
all_chunks_eofis raised before the chunk pipeline is drained.This flips when every reader has hit EOF, but
raw_chunk_blocks,decompressed_chunks, and other workers’ held chunk outputs can still contain tail data.MainThreadChunkConsumertreats that flag plus an empty queue as final EOF, which lines up with the record-loss failure intest_sort_many_chunks_with_semaphore.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/worker_pool.rs` around lines 1335 - 1342, maybe_mark_all_eof currently sets shared.all_chunks_eof as soon as all readers hit EOF, but that can occur while raw_chunk_blocks, decompressed_chunks, or other worker queues still hold data; change the logic so maybe_mark_all_eof only marks sources-as-done (increment sources_at_eof) and then calls a new check function (e.g., check_and_mark_all_eof) which verifies that shared.raw_chunk_blocks and shared.decompressed_chunks (and any other in-flight queues/buffers) are empty and no worker is producing before setting shared.all_chunks_eof and unpark-ing shared.main_thread_handle; update maybe_mark_all_eof to call that check instead of unconditionally storing true into all_chunks_eof so MainThreadChunkConsumer only sees EOF after the pipeline is drained.
1191-1200:⚠️ Potential issue | 🔴 CriticalPhase 1 can still publish EOF while another worker is holding the missing serial.
count >= totalonly proves every block was decompressed once. Another worker can still be sitting on the next block inheld_decompressed_input, which letsPooledInputStreamseedecompressed_input_donewith an empty queue and stop early. This needs a global drained check, not just this worker’spushedstate.
🤖 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/sort/read_ahead.rs`:
- Around line 489-497: The Direct branch in the iterator currently swallows
reader.read_record() errors by returning None (EOF); update the iterator/item
type used by record_source so Self::Direct(reader) yields a Result<RawRecord,
Error> (or propagates the existing error type) instead of mapping Err(e) to
None, modify the match in read_record to return Err(e) on error, and adjust
callers in src/lib/sort/raw.rs that iterate for record in record_source to
handle Result items (propagate or fail the sort) so BGZF/read/parse failures
bubble up rather than being treated as EOF.
In `@tests/integration/test_sort_correctness.rs`:
- Line 56: The test fixture uses zero-padded names (let name =
format!("read_{i:06}")) which are identical under lexicographic and natural
sorting; change the format to produce non-zero-padded names so natural vs
lexicographic order differs (e.g., replace "read_{i:06}" with "read_{i}" or
"read_{i}" plus a suffix/prefix like "read_{i}" or "read_{i}_v" in the test
where the variable name is created) so the queryname::natural cases actually
validate natural-sort behavior.
---
Duplicate comments:
In `@src/lib/sort/bgzf_io.rs`:
- Around line 112-145: The io writer currently writes BGZF_EOF even if some
trailing blocks never arrived because it only detects gaps when a later serial
is buffered; modify the io_writer_loop to require an exact final block count (or
accept an explicit finish sentinel) before writing BGZF_EOF: add an
expected_total parameter (or a finish message on result_rx), track next_expected
vs expected_total, and after the recv loop verify next_expected ==
expected_total+1 (or that the finish sentinel was received) before writing
BGZF_EOF; if they differ return an error instead of finalizing. Ensure you
update uses of result_rx, reorder_buf, buffer_pool.checkin, and writer.write_all
to follow this contract so missing tail blocks are detected and the stream is
not silently truncated.
In `@src/lib/sort/raw.rs`:
- Around line 1401-1402: The pooled worker pool is being created with
self.temp_compression, which fixes compressor settings for all PooledBamWriter
jobs and bypasses output_compression(); change the pooling logic so final BAM
writers use the output compression settings (call self.output_compression())
instead of temp_compression — either by creating a separate SortWorkerPool for
final outputs via SortWorkerPool::new(self.threads.max(1),
self.output_compression()) and passing that to PooledBamWriter, or by making
SortWorkerPool able to select compression per job (propagate
output_compression() into the job-level API used by PooledBamWriter) so that
PooledBamWriter no longer inherits self.temp_compression.
In `@src/lib/sort/worker_pool.rs`:
- Around line 1335-1342: maybe_mark_all_eof currently sets shared.all_chunks_eof
as soon as all readers hit EOF, but that can occur while raw_chunk_blocks,
decompressed_chunks, or other worker queues still hold data; change the logic so
maybe_mark_all_eof only marks sources-as-done (increment sources_at_eof) and
then calls a new check function (e.g., check_and_mark_all_eof) which verifies
that shared.raw_chunk_blocks and shared.decompressed_chunks (and any other
in-flight queues/buffers) are empty and no worker is producing before setting
shared.all_chunks_eof and unpark-ing shared.main_thread_handle; update
maybe_mark_all_eof to call that check instead of unconditionally storing true
into all_chunks_eof so MainThreadChunkConsumer only sees EOF after the pipeline
is drained.
🪄 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: f52d5711-3dfa-4e3d-8e6d-88944503d4b9
📒 Files selected for processing (13)
.gitignorecrates/fgumi-bgzf/src/lib.rscrates/fgumi-bgzf/src/writer.rssrc/lib/bam_io.rssrc/lib/sort/bgzf_io.rssrc/lib/sort/mod.rssrc/lib/sort/pooled_bam_writer.rssrc/lib/sort/pooled_chunk_writer.rssrc/lib/sort/raw.rssrc/lib/sort/read_ahead.rssrc/lib/sort/worker_pool.rstests/integration/main.rstests/integration/test_sort_correctness.rs
✅ Files skipped from review due to trivial changes (3)
- .gitignore
- tests/integration/main.rs
- crates/fgumi-bgzf/src/writer.rs
🚧 Files skipped from review as they are similar to previous changes (5)
- crates/fgumi-bgzf/src/lib.rs
- src/lib/sort/mod.rs
- src/lib/bam_io.rs
- src/lib/sort/pooled_bam_writer.rs
- src/lib/sort/pooled_chunk_writer.rs
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 21:43
42a417d to
0814bce
Compare
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 21:48
0814bce to
49b4d7a
Compare
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 22:11
49b4d7a to
fc7c1fd
Compare
Member
Author
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 22:16
fc7c1fd to
d0029cf
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (2)
src/lib/sort/raw.rs (1)
3230-3230:⚠️ Potential issue | 🟠 MajorThe new
Naturalcases still don't validate the natural comparator.These additions reuse zero-padded numeric suffixes, and the merge test's final check is plain
Stringorder. A regression back to lexicographic comparison would still pass. Use non-padded names and compare adjacent records with the same natural key/comparator used in production.Minimal fixture change
- .name(&format!("read{i:04}")) + .name(&format!("read{i}"))Also applies to: 3281-3281, 3333-3333, 3388-3388, 3652-3672
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/raw.rs` at line 3230, The test case labeled queryname_natural currently uses zero-padded numeric suffixes and validates ordering via plain String comparison, which won't catch regressions to lexicographic sorting; update the fixture records used by the SortOrder::Queryname(QuerynameComparator::Natural) cases to use non-padded numeric suffixes (e.g., "name1", "name2", "name10") and modify the assertions in the merge/ordering tests to compare adjacent records using the same natural comparator (QuerynameComparator::Natural) or the production comparison function rather than raw String ordering so the test verifies numeric-aware ordering for QuerynameComparator::Natural.src/lib/sort/bgzf_io.rs (1)
112-145:⚠️ Potential issue | 🟠 MajorA missing tail block still finalizes a truncated BGZF stream.
If the last serial never arrives,
reorder_bufis empty whenresult_rxcloses, so this still appendsBGZF_EOFand returnsOk(()). The current gap check only catches missing middle blocks. Emit an explicit terminal message with the final serial count, and refuse to write EOF untilnext_expectedmatches it.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/bgzf_io.rs` around lines 112 - 145, The consumer currently writes BGZF_EOF even if the final block never arrived because it has no knowledge of the expected final serial; fix by adding a terminal signal that communicates the final serial count and by refusing to write BGZF_EOF unless next_expected matches that final count. Change the producer/worker protocol so one final message is sent on result_rx (e.g. a special Result variant or a Result with a dedicated flag/terminal_serial) that the consumer reads and stores into a new final_serial variable; then, after the receive loop and after draining reorder_buf, verify final_serial is Some and that next_expected == final_serial + 1 (or the agreed end convention) and return an Err if not, otherwise proceed to write BGZF_EOF and flush; update references in the consumer loop that currently use result_rx, result.serial, reorder_buf, next_expected, and BGZF_EOF.
🧹 Nitpick comments (3)
src/lib/sort/worker_pool.rs (2)
1192-1193: Corrupt input will panic the worker thread.
expect()on decompression is common in BAM tools, but corrupted user input will crash the pool. Consider returning an error and setting aninput_decompress_errorflag if graceful handling is desired.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/worker_pool.rs` around lines 1192 - 1193, The call to decompress_block currently uses expect and will panic on corrupt input; replace the expect in the worker loop where decompress_block(&block, &mut worker.decompressor) is called with proper error handling: match the Result, on Err set a new or existing worker.input_decompress_error flag (or create one in the worker struct) and return/propagate an Err from the worker thread instead of panicking, or otherwise skip/abort processing gracefully; ensure the worker's API (thread join result or channel) surfaces the error so the caller can handle it rather than letting the thread panic.
1252-1265: Chunk file I/O errors silently truncate that source.Error is logged but the file is marked EOF. Downstream merge detects missing serials, but no explicit error flag is surfaced (unlike
input_read_errorfor Phase 1). Consider adding a similarchunk_read_errorflag for parity.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/worker_pool.rs` around lines 1252 - 1265, The read_raw_blocks error path currently logs and marks file_state.eof but doesn't set an explicit error flag analogous to Phase 1's input_read_error; add a new chunk_read_error field to the per-source state (the same struct that contains file_state.eof and input_read_error), set file_state.chunk_read_error = Some(e) (or true) inside the Err(e) branch in the read_raw_blocks match, and ensure any callers that check/propagate input_read_error (e.g., io_writer_loop / merge logic and Self::maybe_mark_all_eof/shared handling) are updated to also inspect/propagate chunk_read_error so the truncation is surfaced explicitly rather than only via missing-serial detection.src/lib/sort/read_ahead.rs (1)
518-535: Caller must checktake_error()after iteration to detect I/O failures.The Direct path stores errors in the slot but returns
None(EOF). Callers iterating withfor record in record_sourcewon't see errors unless they calltake_error()afterward. This is documented at line 506, but easy to miss.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/sort/read_ahead.rs` around lines 518 - 535, The iterator currently swallows I/O errors in Direct by storing them in error_slot and returning None, so callers using for record in ... never see failures unless they call take_error(); change the iterator to surface errors directly: make Iterator::Item a Result<RawRecord, io::Error> (or an appropriate error type), update next() (the Direct branch in fn next()) to return Some(Err(e)) instead of storing to error_slot and returning None, and update the ReadAhead path (Self::ReadAhead and its r.next()) and any callers to handle Result items; keep or remove take_error() accordingly and update docs to reflect the new contract.
🤖 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/sort/raw.rs`:
- Around line 2377-2385: build_chunk_sources() calls
pool.distribute_chunk_files(...) which merely enqueues files; before creating
PoolDisk sources and/or calling set_phase(PHASE2) you must wait for PHASE1
workers to acknowledge they've drained held state. Modify build_chunk_sources
(around the call to Pool::distribute_chunk_files and before pushing
ChunkSource::PoolDisk entries) to perform a quiescence/ack step: invoke a new or
existing Pool/WorkerPool method (e.g. pool.wait_for_phase_ack() or
worker_pool.quiesce_workers()) that blocks until all PHASE1 workers report
drained state, then proceed to set_phase(PHASE2) and create PoolDisk sources;
ensure the same change is applied at the other spot noted (lines ~2471-2480) so
PHASE2 is never entered while PHASE1 state remains.
- Around line 928-970: The drain loop currently stops when pending_total >=
max_pending_blocks which can deadlock if the target source has its next block in
decompressed_chunks; change the logic to keep popping decompressed_chunks until
either decompressed_chunks is empty or the target source becomes runnable:
compute the cap only for buffering blocks of other sources (i.e., when updating
pending_total and new_total, exclude blocks with source_id == target_source),
and always allow installing/consuming a block if block.source_id ==
target_source or if it is immediately consumable (block.serial ==
source.next_serial && source.remaining() == 0); ensure the break condition uses
the adjusted new_total that excludes the target source so we never park while
the target has an available decompressed block in decompressed_chunks.
In `@tests/integration/test_sort_correctness.rs`:
- Around line 120-126: The helper count_records should fail hard when samtools
output cannot be parsed as a number; replace the final .parse().unwrap_or(0)
with a fallible parse that panics with a clear message (e.g.,
.parse::<u64>().expect(...)) and include the raw stdout/stderr in the error
message so a broken samtools call isn't silently treated as zero; update the
parsing step in function count_records to use expect and surface the actual
output for debugging.
---
Duplicate comments:
In `@src/lib/sort/bgzf_io.rs`:
- Around line 112-145: The consumer currently writes BGZF_EOF even if the final
block never arrived because it has no knowledge of the expected final serial;
fix by adding a terminal signal that communicates the final serial count and by
refusing to write BGZF_EOF unless next_expected matches that final count. Change
the producer/worker protocol so one final message is sent on result_rx (e.g. a
special Result variant or a Result with a dedicated flag/terminal_serial) that
the consumer reads and stores into a new final_serial variable; then, after the
receive loop and after draining reorder_buf, verify final_serial is Some and
that next_expected == final_serial + 1 (or the agreed end convention) and return
an Err if not, otherwise proceed to write BGZF_EOF and flush; update references
in the consumer loop that currently use result_rx, result.serial, reorder_buf,
next_expected, and BGZF_EOF.
In `@src/lib/sort/raw.rs`:
- Line 3230: The test case labeled queryname_natural currently uses zero-padded
numeric suffixes and validates ordering via plain String comparison, which won't
catch regressions to lexicographic sorting; update the fixture records used by
the SortOrder::Queryname(QuerynameComparator::Natural) cases to use non-padded
numeric suffixes (e.g., "name1", "name2", "name10") and modify the assertions in
the merge/ordering tests to compare adjacent records using the same natural
comparator (QuerynameComparator::Natural) or the production comparison function
rather than raw String ordering so the test verifies numeric-aware ordering for
QuerynameComparator::Natural.
---
Nitpick comments:
In `@src/lib/sort/read_ahead.rs`:
- Around line 518-535: The iterator currently swallows I/O errors in Direct by
storing them in error_slot and returning None, so callers using for record in
... never see failures unless they call take_error(); change the iterator to
surface errors directly: make Iterator::Item a Result<RawRecord, io::Error> (or
an appropriate error type), update next() (the Direct branch in fn next()) to
return Some(Err(e)) instead of storing to error_slot and returning None, and
update the ReadAhead path (Self::ReadAhead and its r.next()) and any callers to
handle Result items; keep or remove take_error() accordingly and update docs to
reflect the new contract.
In `@src/lib/sort/worker_pool.rs`:
- Around line 1192-1193: The call to decompress_block currently uses expect and
will panic on corrupt input; replace the expect in the worker loop where
decompress_block(&block, &mut worker.decompressor) is called with proper error
handling: match the Result, on Err set a new or existing
worker.input_decompress_error flag (or create one in the worker struct) and
return/propagate an Err from the worker thread instead of panicking, or
otherwise skip/abort processing gracefully; ensure the worker's API (thread join
result or channel) surfaces the error so the caller can handle it rather than
letting the thread panic.
- Around line 1252-1265: The read_raw_blocks error path currently logs and marks
file_state.eof but doesn't set an explicit error flag analogous to Phase 1's
input_read_error; add a new chunk_read_error field to the per-source state (the
same struct that contains file_state.eof and input_read_error), set
file_state.chunk_read_error = Some(e) (or true) inside the Err(e) branch in the
read_raw_blocks match, and ensure any callers that check/propagate
input_read_error (e.g., io_writer_loop / merge logic and
Self::maybe_mark_all_eof/shared handling) are updated to also inspect/propagate
chunk_read_error so the truncation is surfaced explicitly rather than only via
missing-serial detection.
🪄 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: 870c89f4-a9f5-4180-bc3c-4b9d0bd8f136
📒 Files selected for processing (13)
.gitignorecrates/fgumi-bgzf/src/lib.rscrates/fgumi-bgzf/src/writer.rssrc/lib/bam_io.rssrc/lib/sort/bgzf_io.rssrc/lib/sort/mod.rssrc/lib/sort/pooled_bam_writer.rssrc/lib/sort/pooled_chunk_writer.rssrc/lib/sort/raw.rssrc/lib/sort/read_ahead.rssrc/lib/sort/worker_pool.rstests/integration/main.rstests/integration/test_sort_correctness.rs
✅ Files skipped from review due to trivial changes (5)
- .gitignore
- tests/integration/main.rs
- crates/fgumi-bgzf/src/lib.rs
- src/lib/sort/mod.rs
- crates/fgumi-bgzf/src/writer.rs
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 22:34
d0029cf to
a502428
Compare
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 22:50
a502428 to
45020cf
Compare
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 22:56
45020cf to
14b9b6c
Compare
Member
Author
|
@coderabbitai resume |
Replace ad-hoc thread spawning in fgumi sort with a structured N+2 model: N pool workers + 1 I/O writer thread + 1 main thread. Worker pool (src/lib/sort/worker_pool.rs): - ArrayQueue (lock-free bounded) replaces crossbeam channels for all inter-step queues; eliminates blocking push/pop in the worker loop - generic_worker_loop pattern: held-item advancement → backpressure priorities → exclusive step ownership → adaptive backoff (10µs min, yield_now + jitter, up to 1ms) - Phase 1: Worker 0 owns ReadInputBlocks exclusively; all workers share DecompressInput and Compress steps - Phase 2: each worker owns its own disjoint chunk files for ReadChunkBlocks (no exclusivity needed); workers share DecompressChunks and Compress steps - park()/unpark() on main thread: workers call unpark() after pushing to main-thread-facing queues, eliminating polling delays - Workers stay alive until SHUTDOWN (not phase completion) to prevent premature exit deadlocks during phase transitions Raw sorter (src/lib/sort/raw.rs): - pool_decompress field and use_sequential_sort() helper gate whether the worker pool handles read/decompress I/O; when enabled, rayon par_sort is replaced with single-threaded sort to preserve N+2 - Fix: merge_chunks_with_index passes pool_decompress=false to build_chunk_sources since the indexing path has no pool consumer set up - Remove dead max_memory_total field (superseded by MemoryLimit::Auto from #236); --max-memory=auto continues to work as intended Staging buffer (src/lib/sort/bgzf_io.rs): - Remove dead is_spill field from CompressJob and StagingBuffer; the worker pool does not need to distinguish spill vs output jobs Correctness: all 2318 tests pass; outputs verified against baseline
Member
Author
|
@coderabbitai review |
nh13
force-pushed
the
nh/sort-investigation
branch
from
April 6, 2026 23:50
14b9b6c to
45653c6
Compare
Member
Author
|
@coderabbitai review |
Member
Author
|
Closing to reopen under a more descriptive branch name. |
This branch was previously deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Replaces ad-hoc thread spawning in
fgumi sortwith a structured N+2 model: N pool workers + 1 I/O writer thread + 1 main thread. At--threads 4, this reduces the thread count from ~30 to exactly 6 and cuts wall time by 35–40% vs the previous implementation.Worker pool (
src/lib/sort/worker_pool.rs)ArrayQueueeverywhere: lock-free bounded queues replace crossbeam channels for all inter-step queues, eliminating blocking push/pop in the worker loopgeneric_worker_looppattern: held-item advancement → backpressure priorities → exclusive step ownership → adaptive backoff (10 µs min,yield_now+ jitter, up to 1 ms)ReadInputBlocksexclusively; all workers shareDecompressInputandCompressReadChunkBlocks(no exclusivity needed); workers shareDecompressChunksandCompresspark()/unpark(): workers callunpark()after pushing to main-thread-facing queues, eliminating polling delays between the main thread and the poolRaw sorter (
src/lib/sort/raw.rs)pool_decompressfield anduse_sequential_sort()helper gate worker-pool I/O; when enabled, rayonpar_sortis replaced with single-threaded sort to preserve the N+2 budgetmerge_chunks_with_index(coordinate sort with--write-index) was passingpool_decompress=truetobuild_chunk_sourcesbut then callingnext_record(…, None), which panics forPoolDisksources. Fixed to always usepool_decompress=falsein that path until the indexing path gets a pooled writer (see TODO comment)max_memory_totalfield (superseded byMemoryLimit::Autofrom feat(sort): add --max-memory=auto with system memory detection #236);--max-memory=autocontinues to work as intendedStaging buffer (
src/lib/sort/bgzf_io.rs)is_spillfield fromCompressJobandStagingBuffer; the worker pool does not distinguish spill vs output jobsPerformance (coordinate sort, 60 M records, MacBook Pro M-series)
fgumi is slower than samtools at t1 (pool overhead dominates; single-thread fast path is a follow-on). From t2 onward fgumi leads and the gap widens with thread count.
Test plan
cargo ci-test)cargo ci-fmt && cargo ci-lint)fgumi sort --verifypasses for all four sort orders at t1/t2/t4/t8--write-indexno longer panics (regression for themerge_chunks_with_indexfix)ps -M)Follow-on work
--threads 1, skip the pool entirely (matches samtools t1 behavior; tracked as a separate issue)io_writer_loopBTreeMapreorder buffer is unbounded and can hold many in-flight compressed blocks; mimalloc arena retention also inflates RSS. Will file a separate issue.merge_chunks_with_indexcurrently falls back to non-pooled I/O; replacing withPooledIndexingBamWriterwould extend the N+2 benefit to--write-indexpaths.