feat(sort): add --async-reader for prefetch I/O on input BAM

[nh13]

Summary

• Wire PrefetchReader (from perf(io): async prefetch reader with POSIX_FADV_WILLNEED for BAM/FASTQ inputs #258) into the sort command's pool-integrated input path
• When --async-reader is enabled, a dedicated I/O thread reads raw bytes into a bounded channel ahead of the pool's BGZF decompression workers, hiding disk latency during Phase 1
• Add hidden --async-reader flag to Sort CLI, async_reader field + builder to RawExternalSorter, and conditional PrefetchReader wrapping in create_raw_bam_reader_pool_integrated
• Add 4 integration tests (coordinate 1T/4T, queryname 4T, template-coordinate 4T) comparing output with/without the flag
• Update performance-tuning docs to mention sort support

Benchmark (c7g.4xlarge, 6.5 GB BAM, template-coordinate, 8 threads, cold cache, gp3 @ 500 MB/s)

Phase	Baseline	--async-reader	Delta
Read + decompress	24.3s (44%)	21.9s (42%)	-10%
In-memory sort	2.1s (4%)	2.1s (4%)	0%
Spill write	12.1s (23%)	12.1s (24%)	0%
K-way merge	15.6s (29%)	15.6s (30%)	0%
Total wall	54.1s	51.6s	-4.6%

Async reader also reduces variance (baseline 52.9–55.9s vs async 51.0–52.9s).

Stacked on

• perf(io): async prefetch reader with POSIX_FADV_WILLNEED for BAM/FASTQ inputs #258 (nh/async-prefetch-reader)

Test plan

• cargo ci-test (2426 passed)
• cargo ci-fmt clean
• cargo ci-lint clean
• 4 sort-specific --ignored integration tests pass (require samtools)
• Benchmarked on EC2 c7g.4xlarge with cold cache

[nh13]

@coderabbitai review

[coderabbitai]

✅ Actions performed

Review triggered.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

📝 Walkthrough

Walkthrough

A hidden --async-reader boolean flag was added to the sort command and exposed as async_reader on Sort and RawExternalSorter. The sorter now passes this flag into create_raw_bam_reader_pool_integrated, which conditionally uses PrefetchReader::from_file with advise_sequential when enabled, instead of the buffered io::BufReader path; stdin/fifo behavior is unchanged. Documentation was updated to state --async-reader applies to all commands that read BAM/FASTQ input, and integration tests were added to compare results with and without the flag.

🚥 Pre-merge checks | ✅ 3

✅ Passed checks (3 passed)

Check name	Status	Explanation
Title check	✅ Passed	Title clearly and specifically describes the main change: adding --async-reader flag for prefetch I/O optimization in the sort command.
Description check	✅ Passed	Description is detailed and directly related to the changeset, explaining the feature, implementation approach, test coverage, and performance impact.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch nh/sort-async-reader

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 3

🧹 Nitpick comments (3)

src/lib/commands/common.rs (1)

122-124: Prefer constructing via Default to avoid drift.

Line 123 hardcodes async_reader: false; using struct update from Default keeps constructor behavior aligned if defaults evolve.

♻️ Suggested refactor

 pub fn new(input: impl Into<PathBuf>, output: impl Into<PathBuf>) -> Self {
-    Self { input: input.into(), output: output.into(), async_reader: false }
+    Self { input: input.into(), output: output.into(), ..Self::default() }
 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/lib/commands/common.rs` around lines 122 - 124, The constructor new
currently hardcodes async_reader: false which can drift from the type's Default;
change the initializer in pub fn new(input: impl Into<PathBuf>, output: impl
Into<PathBuf>) -> Self to build from the type's Default using struct update
syntax (e.g. set input: input.into(), output: output.into(),
..Default::default()) so async_reader and any future defaults stay synchronized
with Default::default() for this type.

src/lib/os_hints.rs (1)

42-65: Keep the raw-fd helpers crate-private.

These APIs expose a lifetime-sensitive raw descriptor contract, but the docs describe them as PrefetchReader internals. Making them pub(crate) keeps that footgun out of the crate’s public surface.

Proposed visibility change

-pub fn advise_willneed_raw(fd: i32, offset: i64, len: i64) {
+pub(crate) fn advise_willneed_raw(fd: i32, offset: i64, len: i64) {
@@
-pub fn hint_fd(file: &File) -> Option<i32> {
+pub(crate) fn hint_fd(file: &File) -> Option<i32> {
@@
-pub fn hint_fd(_file: &File) -> Option<i32> {
+pub(crate) fn hint_fd(_file: &File) -> Option<i32> {

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/lib/os_hints.rs` around lines 42 - 65, Change the visibility of the
raw-fd helper APIs so they are crate-private: make advise_willneed_raw and both
hint_fd definitions pub(crate) instead of pub; update the declaration for
advise_willneed_raw and the two cfg-gated hint_fd functions (the #[cfg(target_os
= "linux")] hint_fd and the #[cfg(not(target_os = "linux"))] hint_fd) to use
pub(crate) to keep the raw descriptor contract internal to the crate.

tests/integration/test_async_reader.rs (1)

96-127: Compare normalized headers too.

This helper only checks records, so it would miss regressions in @HD/@SQ/@RG/sort metadata. Stripping just the @PG command-line delta and asserting the rest of the header matches would make these tests cover the full output contract.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/integration/test_async_reader.rs` around lines 96 - 127, Extend
assert_bam_records_equal to also read and compare the BAM headers: call the
existing header-reading helper (or obtain header via the same reader used in
read_bam_records) for both path1 and path2, remove/skip only the `@PG` header
records from each header, normalize ordering if necessary (e.g., sort `@SQ/`@RG
entries or compare canonical representations) and assert the remaining header
lines/fields (`@HD`, `@SQ`, `@RG`, sort, etc.) are equal with a descriptive message;
keep the existing record comparisons intact and fail with a clear message if
headers differ.

🤖 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/bam_io.rs`:
- Around line 1225-1230: The async prefetch path can hang on stdin because
PrefetchReader::drop() waits on join() while the stdin producer may block in
read(); change the condition that creates PrefetchReader so that when
opts.async_reader is true you also check the input is not "-" (stdin) before
constructing PrefetchReader, and otherwise fall back to returning the chained
reader; update the branch around PrefetchReader::new(chained) and the
Ok((Box::new(prefetch), header)) return to gate --async-reader off for "-" (or
implement an explicit cancellation mechanism on PrefetchReader before enabling
it).

In `@src/lib/commands/extract.rs`:
- Around line 510-514: The docstring for the async_reader flag is inaccurate:
update the comment above the async_reader field (the attribute with #[arg(long =
"async-reader", default_value_t = false, hide = true)] and the public field
async_reader) to remove the claim that it "has no effect on gzip/plain inputs"
and instead state that the flag wraps FASTQ inputs (including plain, gzip, and
BGZF) in a userspace async prefetch reader—reference open_fastq_reader() as the
implementation that now applies the wrapper to all FASTQ input types; keep the
notes about dedicating one OS thread per input and that the flag is
experimental/hidden.

In `@src/lib/unified_pipeline/fastq.rs`:
- Around line 3699-3723: The code currently enables PrefetchReader whenever
config.async_reader is true, which can violate the BGZF-only contract if
inputs_are_bgzf is false (or decompressed_readers == None is desynced); update
the logic in the mapping that constructs StreamReader::Bgzf so that
PrefetchReader is used only when both config.async_reader and inputs_are_bgzf
are true (i.e., change the branch to if config.async_reader && inputs_are_bgzf {
... } else { ... }), or alternatively add an early assertion like
assert!(!config.async_reader || inputs_are_bgzf, "async_reader requires BGZF
inputs") to enforce the invariants before creating the readers; reference the
config.async_reader flag, inputs_are_bgzf (or decompressed_readers presence) and
the StreamReader::Bgzf / PrefetchReader construction to locate where to change
the code.

---

Nitpick comments:
In `@src/lib/commands/common.rs`:
- Around line 122-124: The constructor new currently hardcodes async_reader:
false which can drift from the type's Default; change the initializer in pub fn
new(input: impl Into<PathBuf>, output: impl Into<PathBuf>) -> Self to build from
the type's Default using struct update syntax (e.g. set input: input.into(),
output: output.into(), ..Default::default()) so async_reader and any future
defaults stay synchronized with Default::default() for this type.

In `@src/lib/os_hints.rs`:
- Around line 42-65: Change the visibility of the raw-fd helper APIs so they are
crate-private: make advise_willneed_raw and both hint_fd definitions pub(crate)
instead of pub; update the declaration for advise_willneed_raw and the two
cfg-gated hint_fd functions (the #[cfg(target_os = "linux")] hint_fd and the
#[cfg(not(target_os = "linux"))] hint_fd) to use pub(crate) to keep the raw
descriptor contract internal to the crate.

In `@tests/integration/test_async_reader.rs`:
- Around line 96-127: Extend assert_bam_records_equal to also read and compare
the BAM headers: call the existing header-reading helper (or obtain header via
the same reader used in read_bam_records) for both path1 and path2, remove/skip
only the `@PG` header records from each header, normalize ordering if necessary
(e.g., sort `@SQ/`@RG entries or compare canonical representations) and assert the
remaining header lines/fields (`@HD`, `@SQ`, `@RG`, sort, etc.) are equal with a
descriptive message; keep the existing record comparisons intact and fail with a
clear message if headers differ.

🪄 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: 34e28527-e6b5-4a6f-811f-38bbaa6f99f2

📥 Commits

Reviewing files that changed from the base of the PR and between 52a1d54 and 226eb0e.

⛔ Files ignored due to path filters (1)

• Cargo.lock is excluded by !**/*.lock

📒 Files selected for processing (22)

• Cargo.toml
• docs/src/guide/performance-tuning.md
• src/lib/bam_io.rs
• src/lib/commands/clip.rs
• src/lib/commands/codec.rs
• src/lib/commands/common.rs
• src/lib/commands/correct.rs
• src/lib/commands/dedup.rs
• src/lib/commands/downsample.rs
• src/lib/commands/duplex.rs
• src/lib/commands/extract.rs
• src/lib/commands/filter.rs
• src/lib/commands/group.rs
• src/lib/commands/simplex.rs
• src/lib/commands/sort.rs
• src/lib/mod.rs
• src/lib/os_hints.rs
• src/lib/prefetch_reader.rs
• src/lib/sort/raw.rs
• src/lib/unified_pipeline/fastq.rs
• tests/integration/main.rs
• tests/integration/test_async_reader.rs

[codecov]

Codecov Report

❌ Patch coverage is 76.19048% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 90.03%. Comparing base (f893802) to head (f1366dc).
⚠️ Report is 4 commits behind head on main.

Files with missing lines	Patch %	Lines
src/lib/bam_io.rs	50.00%	4 Missing ⚠️
src/lib/sort/raw.rs	90.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #261      +/-   ##
==========================================
+ Coverage   89.87%   90.03%   +0.16%     
==========================================
  Files         120      122       +2     
  Lines       61416    61771     +355     
==========================================
+ Hits        55197    55615     +418     
+ Misses       6219     6156      -63     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[nh13]

@coderabbitai review

[coderabbitai]

✅ Actions performed

Review triggered.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

🧹 Nitpick comments (1)

src/lib/bam_io.rs (1)

828-833: Update function docs to reflect async thread behavior.

The docs above still say no extra threads are spawned, but async_reader=true now introduces a prefetch thread. Please align the comment with current behavior.

🤖 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 828 - 833, Update the function documentation
for create_raw_bam_reader_pool_integrated to state that when async_reader is
true a background prefetch thread is spawned (using the pooled read-ahead input
stream) and that this changes threading/ownership considerations; mention that
async_reader=false keeps the previous single-thread/no-extra-thread behavior,
and reference the PooledInputStream/read_ahead mechanism and the async_reader
parameter so callers know the concurrency impact.

🤖 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/bam_io.rs`:
- Around line 828-833: Update the function documentation for
create_raw_bam_reader_pool_integrated to state that when async_reader is true a
background prefetch thread is spawned (using the pooled read-ahead input stream)
and that this changes threading/ownership considerations; mention that
async_reader=false keeps the previous single-thread/no-extra-thread behavior,
and reference the PooledInputStream/read_ahead mechanism and the async_reader
parameter so callers know the concurrency impact.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 47e7959e-210d-440c-b02f-59a63abad1b9

📥 Commits

Reviewing files that changed from the base of the PR and between 226eb0e and c98a3d7.

📒 Files selected for processing (6)

• docs/src/guide/performance-tuning.md
• src/lib/bam_io.rs
• src/lib/commands/extract.rs
• src/lib/commands/sort.rs
• src/lib/sort/raw.rs
• tests/integration/test_async_reader.rs

✅ Files skipped from review due to trivial changes (1)

• src/lib/commands/extract.rs

🚧 Files skipped from review as they are similar to previous changes (3)

• src/lib/commands/sort.rs
• docs/src/guide/performance-tuning.md
• src/lib/sort/raw.rs

[nh13]

@coderabbitai review

[coderabbitai]

✅ Actions performed

Review triggered.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

🧹 Nitpick comments (1)

src/lib/bam_io.rs (1)

880-890: Minor inconsistency: advise_sequential only in async branch.

Other similar functions (create_bam_reader_with_opts at line 713, create_bam_reader_for_pipeline_with_opts at line 1246) call advise_sequential unconditionally before the async check. Here it's inside the if async_reader block.

The kernel read-ahead hint benefits sequential reads regardless of PrefetchReader. Consider moving it before the conditional for consistency.

♻️ Proposed fix

+        crate::os_hints::advise_sequential(&file);
+
         let reader: Box<dyn io::Read + Send> = if async_reader {
-            crate::os_hints::advise_sequential(&file);
             log::info!(
                 "async sort reader enabled: spawning fgumi-prefetch thread for {}",
                 path_ref.display()

🤖 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 880 - 890, Move the call to
crate::os_hints::advise_sequential(&file) out of the async branch so the kernel
read-ahead hint is applied for both sync and async readers; specifically, in the
reader-construction block that returns (header, reader) (the same area that
constructs Box<dyn io::Read + Send>), invoke advise_sequential(&file) once
before the if async_reader { ... } else { ... } so both the
PrefetchReader::from_file(file) path and the io::BufReader::with_capacity(...)
path benefit from the hint (mirror the approach used in
create_bam_reader_with_opts and create_bam_reader_for_pipeline_with_opts).

🤖 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/bam_io.rs`:
- Around line 880-890: Move the call to
crate::os_hints::advise_sequential(&file) out of the async branch so the kernel
read-ahead hint is applied for both sync and async readers; specifically, in the
reader-construction block that returns (header, reader) (the same area that
constructs Box<dyn io::Read + Send>), invoke advise_sequential(&file) once
before the if async_reader { ... } else { ... } so both the
PrefetchReader::from_file(file) path and the io::BufReader::with_capacity(...)
path benefit from the hint (mirror the approach used in
create_bam_reader_with_opts and create_bam_reader_for_pipeline_with_opts).

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 2d93f08f-9ece-415b-8dd2-a9585eef491b

📥 Commits

Reviewing files that changed from the base of the PR and between c98a3d7 and f1366dc.

📒 Files selected for processing (6)

• docs/src/guide/performance-tuning.md
• src/lib/bam_io.rs
• src/lib/commands/extract.rs
• src/lib/commands/sort.rs
• src/lib/sort/raw.rs
• tests/integration/test_async_reader.rs

✅ Files skipped from review due to trivial changes (1)

• src/lib/commands/extract.rs

🚧 Files skipped from review as they are similar to previous changes (4)

• docs/src/guide/performance-tuning.md
• src/lib/commands/sort.rs
• src/lib/sort/raw.rs
• tests/integration/test_async_reader.rs