perf(sort): implement N+2 worker-pool model for parallel sort I/O

.gitignore

@@ -34,3 +34,6 @@ Session.vim
 # Local notes (DEV.md, RD.md)
 /notes/
 /plans/
+
+# Benchmark data
+/tmp/

Cargo.toml

@@ -67,6 +67,7 @@ murmur3 = "0.5"
 statrs = "0.18"
 bytesize = "2.3"
 sysinfo = { version = "0.38", default-features = false, features = ["system"] }
+num_cpus = "1.16"
 rayon = "1.10"
 approx = "0.5.1"
 ahash = "0.8"

crates/fgumi-bgzf/src/lib.rs

@@ -14,4 +14,4 @@ pub use reader::{
     BGZF_EOF, BGZF_FOOTER_SIZE, BGZF_HEADER_SIZE, RawBgzfBlock, decompress_block,
     decompress_block_into, decompress_block_slice_into, read_raw_blocks,
 };
-pub use writer::{CompressedBlock, InlineBgzfCompressor};
+pub use writer::{BGZF_MAX_BLOCK_SIZE, CompressedBlock, InlineBgzfCompressor};

crates/fgumi-bgzf/src/writer.rs

@@ -14,7 +14,7 @@ use std::io;
 // ============================================================================
 
 /// Maximum uncompressed size for a BGZF block (64KB - header/footer overhead).
-const BGZF_MAX_BLOCK_SIZE: usize = bgzf::BGZF_BLOCK_SIZE;
+pub const BGZF_MAX_BLOCK_SIZE: usize = bgzf::BGZF_BLOCK_SIZE;
 
 // ============================================================================
 // Block types

src/commands/common.rs

@@ -567,11 +567,14 @@ impl QueueMemoryOptions {
 
     /// Warns if the requested memory exceeds reasonable system limits.
     fn validate_against_system_memory(requested_bytes: u64) {
-        use sysinfo::System;
-        let mut system = System::new();
+        let mut system = sysinfo::System::new();
         system.refresh_memory();
 
-        let total_memory_bytes = system.total_memory();
+        // Use cgroup-aware total so the warning is accurate inside containers.
+        let total_memory_bytes =
+            system.cgroup_limits().map(|c| c.total_memory).unwrap_or_else(|| system.total_memory());
+
+        // available_memory reflects free physical pages; no cgroup equivalent.
         let available_memory_bytes = system.available_memory();
 
         // Calculate 90% limit using integer arithmetic to avoid precision loss
@@ -633,6 +636,10 @@ pub(crate) fn parse_bool(s: &str) -> Result<bool, String> {
 // Re-export from the library crate for backward compatibility.
 pub use fgumi_lib::validation::parse_memory_size;
 
+// Re-export from the library crate so command code doesn't need to reach into
+// fgumi_lib::system directly.
+pub(crate) use fgumi_lib::system::detect_total_memory;
+
 /// Builds a [`BamPipelineConfig`] from the common CLI option structs.
 ///
 /// This consolidates the pipeline configuration boilerplate that is repeated
@@ -1095,4 +1102,30 @@ mod tests {
     fn test_extended_bool_values_in_cli_invalid(#[case] args: &[&str]) {
         assert!(TestBoolFlags::try_parse_from(args).is_err());
     }
+
+    // -------------------------------------------------------------------------
+    // cgroup-aware memory detection
+    // -------------------------------------------------------------------------
+
+    #[test]
+    fn test_detect_total_memory_returns_nonzero() {
+        let total = detect_total_memory();
+        assert!(total > 0, "detect_total_memory returned 0");
+        // Sanity: at least 64 MiB (even the smallest CI runner has more than this).
+        assert!(total >= 64 * 1024 * 1024, "detect_total_memory returned < 64 MiB: {total}");
+    }
+
+    #[test]
+    fn test_detect_total_memory_bounded_by_sysinfo() {
+        // cgroup_limits().total_memory is min(cgroup_max, physical_ram), so
+        // detect_total_memory() can never exceed what sysinfo reports.
+        let total = detect_total_memory();
+        let mut system = sysinfo::System::new();
+        system.refresh_memory();
+        let sysinfo_total = usize::try_from(system.total_memory()).unwrap_or(usize::MAX);
+        assert!(
+            total <= sysinfo_total,
+            "cgroup-limited total {total} exceeded sysinfo total {sysinfo_total}"
+        );
+    }
 }

src/commands/sort.rs

@@ -31,7 +31,7 @@ use log::info;
 use std::path::PathBuf;
 
 use crate::commands::command::Command;
-use crate::commands::common::{CompressionOptions, parse_bool};
+use crate::commands::common::{CompressionOptions, detect_total_memory, parse_bool};
 use fgumi_lib::validation::parse_memory_size;
 
 /// Sort order for BAM files.
@@ -408,10 +408,10 @@ fn resolve_reserve(reserve: MemoryReserve, total_memory: usize) -> usize {
 
 /// Resolves a [`MemoryLimit`] to a concrete byte count.
 ///
-/// For [`MemoryLimit::Auto`]: detects total system memory, subtracts the
-/// reserve (via [`MemoryReserve`]), and divides by thread count (when
-/// `memory_per_thread` is true). The result is clamped to a minimum of
-/// 256 MiB per thread.
+/// For [`MemoryLimit::Auto`]: detects total system memory (cgroup-aware via
+/// [`detect_total_memory`]), subtracts the reserve (via [`MemoryReserve`]),
+/// and divides by thread count (when `memory_per_thread` is true). The result
+/// is clamped to a minimum of 256 MiB per thread.
 ///
 /// For [`MemoryLimit::Fixed`]: applies the same `memory_per_thread`
 /// multiplication as before. The reserve is ignored.
@@ -436,9 +436,7 @@ fn resolve_memory_limit(
             }
         }
         MemoryLimit::Auto => {
-            let mut system = sysinfo::System::new();
-            system.refresh_memory();
-            let total = usize::try_from(system.total_memory()).unwrap_or(usize::MAX);
+            let total = detect_total_memory();
 
             let margin = resolve_reserve(reserve, total);
             let available = total.saturating_sub(margin);
@@ -481,10 +479,8 @@ fn resolve_memory_limit(
         }
     };
 
-    // Post-resolution system memory check: warn if any mode exceeds host RAM
-    let mut system = sysinfo::System::new();
-    system.refresh_memory();
-    let total = usize::try_from(system.total_memory()).unwrap_or(usize::MAX);
+    // Post-resolution sanity check: warn if budget exceeds the effective memory limit.
+    let total = detect_total_memory();
     if total_budget > total {
         log::warn!(
             "Memory budget {} exceeds total system memory {}; spill-to-disk is likely",
@@ -838,9 +834,7 @@ mod tests {
 
     #[test]
     fn test_resolve_memory_limit_auto() {
-        let mut system = sysinfo::System::new();
-        system.refresh_memory();
-        let total = system.total_memory() as usize;
+        let total = detect_total_memory();
 
         let resolved = resolve_memory_limit(MemoryLimit::Auto, MemoryReserve::Auto, 4, true)
             .expect("should succeed");
@@ -850,7 +844,7 @@ mod tests {
             resolved >= min_expected,
             "auto resolved to {resolved} bytes, expected at least {min_expected}"
         );
-        // And not more than total system memory
+        // And not more than effective total memory (cgroup-aware)
         assert!(resolved <= total);
     }
 

src/lib/bam_io.rs

@@ -748,6 +748,103 @@ pub fn create_raw_bam_reader<P: AsRef<Path>>(
     Ok((raw_reader, header))
 }
 
+/// Create a raw BAM reader using the pool's Phase 1 integrated reading.
+///
+/// Workers in the pool do `ReadInputBlocks` + `DecompressInput`. The main thread
+/// consumes decompressed bytes via `PooledInputStream`. No extra threads are spawned.
+///
+/// # Flow
+///
+/// 1. Parse header with noodles (first pass, single-threaded)
+/// 2. Open file again, set as pool's input file
+/// 3. Set pool to PHASE1 — workers start reading/decompressing
+/// 4. Main thread skips header bytes from decompressed stream
+/// 5. Returns `RawBamReader<PooledInputStream>` for direct record iteration
+///
+/// # Errors
+///
+/// Returns an error if the BAM file cannot be opened, the header cannot be parsed,
+/// or header bytes cannot be skipped from the decompressed stream.
+pub fn create_raw_bam_reader_pool_integrated<P: AsRef<Path>>(
+    path: P,
+    pool: &std::sync::Arc<crate::sort::worker_pool::SortWorkerPool>,
+) -> Result<(RawBamReader<crate::sort::read_ahead::PooledInputStream>, Header)> {
+    use crate::sort::read_ahead::PooledInputStream;
+    use crate::sort::worker_pool::phase;
+
+    use std::io::{Seek, SeekFrom};
+
+    let path_ref = path.as_ref();
+
+    // Open once and reuse the same handle for both header reading and pool input.
+    // Re-opening would break non-replayable inputs (FIFOs, process substitution).
+    let mut file = File::open(path_ref)
+        .with_context(|| format!("Failed to open input BAM: {}", path_ref.display()))?;
+
+    let header = {
+        let bgzf = BgzfReader::new(&mut file);
+        let mut noodles_reader = noodles::bam::io::Reader::from(bgzf);
+        noodles_reader
+            .read_header()
+            .with_context(|| format!("Failed to read header from: {}", path_ref.display()))?
+    };
+
+    // Rewind to start so pool workers read from byte 0
+    file.seek(SeekFrom::Start(0))
+        .with_context(|| format!("Failed to rewind input BAM: {}", path_ref.display()))?;
+
+    let buf_reader = io::BufReader::with_capacity(2 * 1024 * 1024, file);
+
+    // Set the input file for pool workers and activate Phase 1
+    pool.set_input_file(Box::new(buf_reader));
+    pool.set_phase(phase::PHASE1);
+
+    // Create the decompressed input stream for the main thread.
+    // Uses the shared ArrayQueue and EOF flag from the pool's pipeline state.
+    let mut pooled_input = PooledInputStream::new(
+        pool.decompressed_input_queue(),
+        pool.decompressed_input_done_flag(),
+        pool.input_read_error_flag(),
+    );
+
+    // Skip BAM header from the decompressed stream
+    skip_bam_header(&mut pooled_input)
+        .with_context(|| format!("Failed to skip header from: {}", path_ref.display()))?;
+
+    let raw_reader = RawBamReader::new(pooled_input);
+    Ok((raw_reader, header))
+}
+
+/// Skip the BAM header from a reader positioned at the start of a BAM stream.
+///
+/// Reads and discards: magic (4 bytes), header text length + text, `n_ref` + reference entries.
+fn skip_bam_header<R: Read>(reader: &mut R) -> Result<()> {
+    let mut buf4 = [0u8; 4];
+
+    // Magic: "BAM\1"
+    reader.read_exact(&mut buf4)?;
+    anyhow::ensure!(&buf4 == b"BAM\x01", "Not a BAM file (bad magic)");
+
+    // Header text length + text
+    reader.read_exact(&mut buf4)?;
+    let l_text = u32::from_le_bytes(buf4) as usize;
+    io::copy(&mut reader.take(l_text as u64), &mut io::sink())?;
+
+    // Number of reference sequences
+    reader.read_exact(&mut buf4)?;
+    let n_ref = u32::from_le_bytes(buf4) as usize;
+
+    // Each reference: l_name (4 bytes) + name (l_name bytes) + l_ref (4 bytes)
+    for _ in 0..n_ref {
+        reader.read_exact(&mut buf4)?;
+        let l_name = u32::from_le_bytes(buf4) as usize;
+        io::copy(&mut reader.take(l_name as u64), &mut io::sink())?;
+        reader.read_exact(&mut buf4)?; // l_ref (discard)
+    }
+
+    Ok(())
+}
+
 /// Create a BAM writer and write the header in one operation
 ///
 /// # Arguments

src/lib/mod.rs

@@ -149,6 +149,7 @@ pub use fgumi_metrics::rejection;
 pub mod reorder_buffer;
 pub mod sam;
 pub mod sort;
+pub mod system;
 pub mod tag_reversal;
 pub mod template;
 pub mod umi;