mux: pipelined PES highway — read+decrypt → demux → parse on 3 threads

Introduces the freemkv mux throughput highway: a three-stage thread
pipeline that replaces the inline single-thread read path for any
file-backed source (ISO and m2ts file URLs both route through it).

  Thread A: read + decrypt  (PrefetchedSectorSource / BytePrefetcher)
  Thread B: M2TS demux      (DemuxThread)
  Thread C: codec parse     (PipelinedPesStream, on caller thread)

Each handoff uses a bounded crossbeam channel with a recycled buffer
pool — no allocations or memcpys in the steady-state hot loop.

Component map:

* io/byte_prefetcher.rs (new) — std::io::Read producer thread with
  recycled Vec<u8> pool. Pairs with PrefetchedSectorSource (sector
  side) so demux_thread::spawn_zero_copy can wire either upstream.
* sector/prefetched.rs — recycled buffer pool added; into_channels()
  peels off the rx/recycle_tx/shell triple for zero-copy demux.
* mux/demux_thread.rs (new) — owns the TsDemuxer/PsDemuxer, runs
  feed() on its thread, ships Vec<PesPacket> batches.
* mux/pipelined_stream.rs (new) — the read-side Stream impl. Pulls
  packets from the demux thread and runs codec parse on the caller.
* mux/resolve.rs — build_iso_pipeline (public) / build_m2ts_pipeline
  (private) assemble the three stages; iso:// and m2ts:// both
  return PipelinedPesStream.
* mux/m2ts.rs — collapsed to a write-only sink (Mode::Read deleted;
  the read direction lives on the highway now).
* mux/codec/h264.rs — find_start_code uses memchr SIMD memmem::find.
* mux/codec/hevc.rs — tightened frame_data initial capacity.
* mux/ts.rs — boundary-packet handling avoids the per-batch 16 MiB
  remainder copy; PesAssembler starts at 16 KiB to dodge the 64-page
  first-touch fault tax that the previous 256 KiB pre-alloc paid on
  every PES boundary.
* mux/disc.rs — gains DiscStream::new_pipeline + read_pipeline as
  the legacy autorip ingress (drive + multipass paths still need
  on_event / skip_errors before they migrate to the highway).
* io/file_sector_source/* — per-OS prefetch() syscall hook
  (Linux readahead, macOS F_RDADVISE, Windows/other no-op).
* decrypt.rs — FREEMKV_DECRYPT_THREADS renamed to FREEMKV_THREADS;
  pool sized to all cores by default.

Measured on rip1 testbed (Civil War UHD, 62 GiB ISO → null://):

  60 → 322 MB/s warm cache (old new_pipeline path)
  60 → 660 MB/s warm cache (highway path, this commit)
  60 → 126 MB/s sustained disk-bound

The IsoSectorReader baseline reader was deleted in favour of
FileSectorSource so the freemkv CLI and autorip exercise the same
read path.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "libfreemkv"
-version = "0.23.2"
+version = "0.24.0"
 edition = "2024"
 rust-version = "1.86"
 license = "AGPL-3.0-only"
@@ -34,6 +34,17 @@ tracing = "0.1"
 # throughput at ~1 MB/s (see (internal)/memory/
 # feedback_send_with_halt_poll_throttle.md, 0.21.7).
 crossbeam-channel = "0.5"
+# Persistent work-stealing thread pool for parallel AACS unit
+# decryption. Per-call std::thread::scope spawned fresh OS threads
+# and that overhead dominated for typical batch sizes (60 units).
+# rayon's global pool initialises once on first use.
+rayon = "1"
+# SIMD-accelerated bytestring search. Drives the HEVC/H.264 start-code
+# scan in `mux::codec::h264::find_start_code` — naive byte-by-byte
+# walk is ~500 MB/s single-thread on x86_64; memchr's vectorised
+# `memmem::find` for the 3-byte `00 00 01` needle hits ~5 GB/s on
+# AVX2-capable hosts.
+memchr = "2"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 libc = "0.2"

src/decrypt.rs

@@ -3,9 +3,131 @@
 //! Decrypts sectors in-place using resolved keys from disc scanning.
 //! Handles AACS 1.0, AACS 2.0, and CSS transparently.
 //! The caller never sees encrypted data unless explicitly bypassed.
+//!
+//! ## Parallel AACS decrypt
+//!
+//! Each AACS aligned unit (6144 bytes) is decrypted INDEPENDENTLY of
+//! every other unit — per-unit key derivation from the unit_key plus
+//! the unit's own first-16-byte header. There is no cross-unit
+//! dependency, so a buffer of N units can be decrypted on N threads
+//! in parallel via a persistent rayon thread pool.
+//!
+//! Small buffers (< [`PARALLEL_MIN_UNITS`] units) fall through to the
+//! serial path to avoid pool dispatch overhead beating the per-unit
+//! AES work.
+//!
+//! ## Thread-count configuration — three layers
+//!
+//! Resolution order (highest wins):
+//! 1. The most recent [`set_decrypt_threads`] call with `n > 0`.
+//!    Calling this *replaces* the live thread pool — useful for a
+//!    settings-page slider in a long-running daemon.
+//! 2. `FREEMKV_THREADS` env var, if set and `> 0`. Single knob
+//!    covering decrypt today, intended to also drive any future
+//!    input-side / output-side worker pools.
+//! 3. Default: all available cores. Algorithm optimisation comes
+//!    first — we measure single-thread performance to find serial
+//!    bottlenecks before throwing parallelism at it — but once a
+//!    pool is engaged we use the whole box. Hard cap at
+//!    [`MAX_THREADS`] (rayon stack memory).
 
 use crate::aacs;
 use crate::css;
+use rayon::prelude::*;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, RwLock};
+
+/// Minimum units in a buffer before we pay the pool-dispatch cost of
+/// fanning out. Below this, serial is faster.
+const PARALLEL_MIN_UNITS: usize = 8;
+
+/// Hard upper bound on configurable thread count. Anything larger is
+/// almost certainly a misconfiguration; rayon would happily allocate
+/// thousands of worker stacks otherwise.
+pub const MAX_THREADS: usize = 64;
+
+/// Process-wide decrypt thread count override. `0` means "use env
+/// var, else default" — see [`decrypt_threads`] for the resolution
+/// order.
+static DECRYPT_THREADS: AtomicUsize = AtomicUsize::new(0);
+
+/// Current rayon pool. `RwLock<Option<Arc<...>>>` so that
+/// [`set_decrypt_threads`] can swap the pool out without leaking the
+/// old one and without blocking ongoing decrypt work (in-flight calls
+/// hold an `Arc` clone via [`decrypt_pool`] and finish on the old
+/// pool; new calls pick up the new pool).
+static DECRYPT_POOL: RwLock<Option<Arc<rayon::ThreadPool>>> = RwLock::new(None);
+
+/// Configure how many threads to use for AACS unit decryption. A value
+/// of `0` resets to the env / default resolution. `1` forces serial.
+/// `N > 1` builds a new rayon pool of size N and atomically replaces
+/// the live pool.
+///
+/// Thread-safe. Live decrypt calls keep their previously-acquired
+/// pool reference for the rest of the call — no mid-call pool
+/// switch. Subsequent calls see the new pool.
+///
+/// Pool construction is ~ms-scale; safe to call from a settings POST
+/// handler.
+pub fn set_decrypt_threads(n: usize) {
+    let clamped = n.min(MAX_THREADS);
+    DECRYPT_THREADS.store(clamped, Ordering::Relaxed);
+    // Drop the existing pool. Next decrypt_pool() call rebuilds with
+    // the new resolved thread count.
+    if let Ok(mut guard) = DECRYPT_POOL.write() {
+        *guard = None;
+    }
+}
+
+/// Get (or lazily build) the active rayon thread pool. Returns an
+/// `Arc` so in-flight work survives a concurrent
+/// [`set_decrypt_threads`] swap.
+fn decrypt_pool() -> Arc<rayon::ThreadPool> {
+    // Fast path: pool already built.
+    if let Ok(guard) = DECRYPT_POOL.read() {
+        if let Some(pool) = guard.as_ref() {
+            return Arc::clone(pool);
+        }
+    }
+    // Slow path: build a new one under the write lock. Double-check
+    // after acquiring in case another caller built it first.
+    let mut guard = DECRYPT_POOL.write().expect("DECRYPT_POOL RwLock poisoned");
+    if let Some(pool) = guard.as_ref() {
+        return Arc::clone(pool);
+    }
+    let n = decrypt_threads();
+    let pool = Arc::new(
+        rayon::ThreadPoolBuilder::new()
+            .num_threads(n)
+            .thread_name(|i| format!("freemkv-decrypt-{i}"))
+            .build()
+            .expect("rayon decrypt pool build failed"),
+    );
+    *guard = Some(Arc::clone(&pool));
+    pool
+}
+
+/// Current effective decrypt thread count. Resolution order:
+/// 1. Most recent [`set_decrypt_threads`] value (if > 0)
+/// 2. `FREEMKV_THREADS` env var (if set and > 0)
+/// 3. Default: all available cores, capped at [`MAX_THREADS`].
+pub fn decrypt_threads() -> usize {
+    let explicit = DECRYPT_THREADS.load(Ordering::Relaxed);
+    if explicit > 0 {
+        return explicit;
+    }
+    let env = std::env::var("FREEMKV_THREADS")
+        .ok()
+        .and_then(|v| v.parse::<usize>().ok())
+        .unwrap_or(0);
+    if env > 0 {
+        return env.min(MAX_THREADS);
+    }
+    let cores = std::thread::available_parallelism()
+        .map(|n| n.get())
+        .unwrap_or(2);
+    cores.clamp(1, MAX_THREADS)
+}
 
 /// Resolved decryption state from disc scanning.
 /// Passed to `decrypt_sectors()` — the caller doesn't need to know
@@ -57,29 +179,44 @@ pub fn decrypt_sectors(
                     return Err(crate::error::Error::DecryptFailed);
                 }
             };
-            let rdk = read_data_key.as_ref();
+            let rdk: Option<[u8; 16]> = *read_data_key;
             let unit_len = aacs::ALIGNED_UNIT_LEN;
+            let nthreads = decrypt_threads();
+            let chunks: Vec<&mut [u8]> = buf.chunks_mut(unit_len).collect();
+            let nunits = chunks.len();
 
-            for chunk in buf.chunks_mut(unit_len) {
+            // Per-unit decrypt closure. The is_unit_encrypted check is
+            // a byte-0 heuristic; on a misfire we snapshot+restore via
+            // the original bytes so non-m2ts (e.g. MPLS/CLPI nav files)
+            // survive. See test `nav_file_unit_survives_decrypt_attempt`.
+            let decrypt_one = |chunk: &mut [u8]| {
                 if chunk.len() == unit_len && aacs::is_unit_encrypted(chunk) {
-                    // `is_unit_encrypted` is a byte-0 heuristic: it fires on any
-                    // unit whose first byte has the top 2 bits set, which is
-                    // correct for m2ts source packets (where those bits are the
-                    // copy-control marker) but false-positives on any other binary
-                    // data with similarly-shaped first bytes — notably MPLS/CLPI
-                    // navigation files that begin with ASCII magic ('M', 'H'…)
-                    // and survive sweep mixed in with encrypted m2ts payloads.
-                    // `decrypt_unit_full` self-checks via TS-sync verification and
-                    // returns false on a misfire, but it has already mutated the
-                    // chunk by then. Snapshot and restore on verification failure
-                    // — same pattern `decrypt_unit_try_keys` uses for multi-key
-                    // discs. Real m2ts units verify and stay decrypted; nav-file
-                    // sectors get scrambled briefly and then put back as-was.
                     let original: Vec<u8> = chunk.to_vec();
-                    if !aacs::decrypt_unit_full(chunk, &uk, rdk) {
+                    if !aacs::decrypt_unit_full(chunk, &uk, rdk.as_ref()) {
                         chunk.copy_from_slice(&original);
                     }
                 }
+            };
+
+            if nthreads <= 1 || nunits < PARALLEL_MIN_UNITS {
+                // Serial path: avoids thread-pool overhead for tiny
+                // buffers; also the only path when caller pinned
+                // single-threaded via FREEMKV_THREADS=1.
+                for chunk in chunks {
+                    decrypt_one(chunk);
+                }
+            } else {
+                // Parallel path via rayon's persistent global pool.
+                // The pool is built once on first use (lazy_static-style)
+                // and reused across every decrypt_sectors call — no
+                // per-call OS thread spawn, no thread-creation latency
+                // amortised per batch. Each unit decrypts independently
+                // (own key derivation), so par_iter is sound.
+                decrypt_pool().install(|| {
+                    chunks.into_par_iter().for_each(|chunk| {
+                        decrypt_one(chunk);
+                    });
+                });
             }
         }
         DecryptKeys::Css { title_key } => {

src/halt.rs

@@ -71,6 +71,20 @@ impl Default for Halt {
     }
 }
 
+/// Shared poll interval for halt-aware loops.
+///
+/// `bounded_syscall` checks the cancellation flag and the deadline
+/// every [`POLL_INTERVAL`] while blocked on a worker; the same
+/// cadence governs `Pipeline::send_with_halt`'s `try_send` retry.
+/// 250 ms is the sweet spot between responsiveness (operator presses
+/// Stop, sees it take effect within ~quarter-second) and waste
+/// (atomic load + clock read is cheap but not free at thousands of
+/// hertz).
+///
+/// Centralised here so the half-dozen halt-polling loops across `io`
+/// can't drift apart silently.
+pub const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(250);
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/io/bounded.rs

@@ -54,15 +54,7 @@ use std::sync::mpsc::{RecvTimeoutError, sync_channel};
 use std::thread;
 use std::time::{Duration, Instant};
 
-use crate::halt::Halt;
+use crate::halt::{Halt, POLL_INTERVAL};
-
-/// Granularity of the halt poll. The receive loop wakes every
-/// [`POLL_INTERVAL`] to (a) check the [`Halt`] token, then (b) check
-/// the overall deadline, then go back to waiting. 250 ms is a
-/// pragmatic balance: short enough that human-driven `/api/stop` feels
-/// responsive (< 0.5 s p99), long enough that the polling overhead is
-/// negligible against multi-second syscalls.
-const POLL_INTERVAL: Duration = Duration::from_millis(250);
 
 /// Failure outcome from a bounded syscall wrapper.
 #[derive(Debug)]

src/io/byte_prefetcher.rs

@@ -0,0 +1,146 @@
+//! `BytePrefetcher` — `std::io::Read` analogue of
+//! [`crate::sector::PrefetchedSectorSource`].
+//!
+//! Spawns a producer thread that fills a bounded pool of `Vec<u8>`
+//! chunks from the underlying reader and ships them through a
+//! channel; the consumer pulls filled chunks, uses them, and sends
+//! the empty `Vec<u8>` back through a recycle channel so the
+//! producer can re-fill in place. Result: zero allocations and zero
+//! cross-thread frees in the steady-state hot loop.
+//!
+//! This is the byte-stream half of the freemkv mux highway —
+//! `BytePrefetcher` feeds [`crate::mux::demux_thread::DemuxThread`]
+//! for `m2ts://`, `network://`, `stdio://`, and any other stream
+//! whose source is an `io::Read` rather than a `SectorSource`.
+
+use crate::halt::Halt;
+use crossbeam_channel::{Receiver, Sender, bounded};
+use std::io::Read;
+use std::thread::JoinHandle;
+
+/// Items flowing through the forward channel.
+pub type Batch = std::io::Result<Vec<u8>>;
+
+/// Forward channel depth — how many filled buffers the producer can
+/// stay ahead by. Two is enough to absorb a moderate consumer stall
+/// without piling up bytes.
+const FORWARD_DEPTH: usize = 2;
+
+/// Recycle channel depth = forward + 1 so the producer always has at
+/// least one buffer to fill while the consumer holds one.
+const RECYCLE_DEPTH: usize = FORWARD_DEPTH + 1;
+
+/// Default chunk size — 16 MiB matches the ISO-mux sector batch and
+/// is large enough that per-chunk overhead is amortised; small
+/// enough that the in-flight memory footprint stays bounded.
+pub const DEFAULT_CHUNK_BYTES: usize = 16 * 1024 * 1024;
+
+/// Returned from [`BytePrefetcher::into_channels`]. Owns the
+/// producer-thread join handle so dropping the shell joins the
+/// producer.
+pub struct PrefetchShell {
+    producer: Option<JoinHandle<()>>,
+}
+
+impl Drop for PrefetchShell {
+    fn drop(&mut self) {
+        if let Some(h) = self.producer.take() {
+            let _ = h.join();
+        }
+    }
+}
+
+/// Spawned byte prefetcher. Drop joins the producer thread.
+pub struct BytePrefetcher {
+    rx: Receiver<Batch>,
+    recycle_tx: Sender<Vec<u8>>,
+    producer: Option<JoinHandle<()>>,
+}
+
+impl BytePrefetcher {
+    /// Spawn the producer thread. `reader` must be `Send` because it
+    /// moves into the thread. `chunk_bytes` is the size of each
+    /// recycled buffer; pick the natural batch size of the
+    /// downstream demuxer (16 MiB for the BD-TS mux pipeline).
+    pub fn new<R: Read + Send + 'static>(
+        mut reader: R,
+        chunk_bytes: usize,
+        halt: Option<Halt>,
+    ) -> Self {
+        let (tx, rx) = bounded::<Batch>(FORWARD_DEPTH);
+        let (recycle_tx, recycle_rx) = bounded::<Vec<u8>>(RECYCLE_DEPTH);
+
+        // Seed the recycle pool. Without these the first
+        // `recycle_rx.recv()` would block forever (no consumer has
+        // returned a buffer yet).
+        for _ in 0..RECYCLE_DEPTH {
+            let _ = recycle_tx.send(vec![0u8; chunk_bytes]);
+        }
+
+        let producer = std::thread::Builder::new()
+            .name("freemkv-byte-prefetch".into())
+            .spawn(move || {
+                loop {
+                    if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) {
+                        return;
+                    }
+                    let mut buf = match recycle_rx.recv() {
+                        Ok(b) => b,
+                        Err(_) => return, // consumer dropped both channels
+                    };
+                    // Re-expose the full extent (previous iteration
+                    // may have truncated after a short read).
+                    if buf.len() < chunk_bytes {
+                        buf.resize(chunk_bytes, 0);
+                    } else {
+                        // SAFETY: capacity is at least chunk_bytes
+                        // after construction.
+                        unsafe { buf.set_len(chunk_bytes) };
+                    }
+                    // Read up to one full chunk. Short reads are
+                    // valid and common — pipe `truncate` so the
+                    // consumer sees only the bytes that arrived.
+                    let n = match reader.read(&mut buf[..]) {
+                        Ok(0) => return, // EOF — drop tx, consumer sees RecvError
+                        Ok(n) => n,
+                        Err(e) => {
+                            let _ = tx.send(Err(e));
+                            return;
+                        }
+                    };
+                    buf.truncate(n);
+                    if tx.send(Ok(buf)).is_err() {
+                        return; // consumer dropped
+                    }
+                }
+            })
+            .expect("freemkv-byte-prefetch thread spawn failed");
+
+        Self {
+            rx,
+            recycle_tx,
+            producer: Some(producer),
+        }
+    }
+
+    /// Peel off the channels for zero-copy pipeline consumption. The
+    /// caller (typically [`crate::mux::demux_thread::DemuxThread`])
+    /// drains `rx`, runs the demuxer in place on each filled buffer,
+    /// and recycles back through `recycle_tx`.
+    pub fn into_channels(self) -> (Receiver<Batch>, Sender<Vec<u8>>, PrefetchShell) {
+        let mut me = self;
+        let producer = me.producer.take();
+        let rx = me.rx.clone();
+        let recycle = me.recycle_tx.clone();
+        std::mem::forget(me);
+        (rx, recycle, PrefetchShell { producer })
+    }
+}
+
+impl Drop for BytePrefetcher {
+    fn drop(&mut self) {
+        if let Some(h) = self.producer.take() {
+            let _ = h.join();
+        }
+    }
+}

src/io/file_sector_source/linux.rs

@@ -44,3 +44,22 @@ pub(super) fn drop_window(file: &File, start: u64, len: u64) {
         );
     }
 }
+
+/// Async-prefetch `len` bytes at `offset` into the page cache. The
+/// kernel `readahead(2)` syscall queues the I/O and returns
+/// immediately — it does NOT wait for completion. Called right after
+/// each consumed read so the next batch's I/O overlaps with the
+/// caller's processing of the current batch (decrypt + demux + mux).
+///
+/// Without this hint, with a synchronous demux consumer running at
+/// ~50 MB/s and a single-spindle disk capable of ~150 MB/s, the disk
+/// sits idle ~70% of each iteration because kernel readahead alone
+/// (capped at `/sys/block/<dev>/queue/read_ahead_kb`, default 128 KB)
+/// can only pre-stage a tiny slice of the next batch. An explicit
+/// `readahead()` of the same size as the current batch tells the
+/// kernel to queue the full next-batch read now.
+pub(super) fn prefetch(file: &File, offset: u64, len: u64) {
+    unsafe {
+        libc::readahead(file.as_raw_fd(), offset as i64, len as usize);
+    }
+}

src/io/file_sector_source/macos.rs

@@ -44,3 +44,19 @@ pub(super) fn hint_sequential(file: &File, len_bytes: u64) {
 /// less prone to the pin-everything pathology that triggers the
 /// regression on Linux NFS clients.
 pub(super) fn drop_window(_file: &File, _start: u64, _len: u64) {}
+
+/// Async-prefetch the byte range `[offset, offset+len)`. macOS uses
+/// the same `fcntl(F_RDADVISE, &radvisory)` primitive as the open-
+/// time sequential hint, just targeted at a moving window instead of
+/// the whole file. The kernel queues I/O for the requested range and
+/// returns immediately.
+pub(super) fn prefetch(file: &File, offset: u64, len: u64) {
+    let bytes = (len as i64).min(RDADVISE_MAX_BYTES);
+    let mut ra = RadAdvisory {
+        ra_offset: offset as libc::off_t,
+        ra_count: bytes as libc::c_int,
+    };
+    unsafe {
+        libc::fcntl(file.as_raw_fd(), F_RDADVISE, &mut ra);
+    }
+}

src/io/file_sector_source/mod.rs

@@ -1,39 +1,43 @@
 //! [`FileSectorSource`] — read 2048-byte sectors from an ISO file on
-//! disk, with an internal 32 MiB read-ahead buffer.
+//! disk via direct `seek + read_exact` (`pread`-equivalent) calls,
+//! letting the kernel's own readahead policy manage prefetch.
 //!
-//! ## Why the buffer
+//! ## Why no app-level buffer
 //!
-//! On NFS-mounted ISOs, an unbuffered `pread(2048)` per sector pays an
+//! Pre-0.21.3 this source held a 32 MiB (later 4 MiB) read-ahead
-//! NFS round-trip for every sector. With `rsize=1 MiB` and a 100-150 ms
+//! buffer to amortise per-sector NFS round-trips. Empirically that
-//! NFS RTT, that's three orders of magnitude more round trips than
+//! buffer hurt: 32 MiB refills bursted the NFS TCP connection hard
-//! necessary — the muxer goes read-bound on every read, even though
+//! enough to starve the concurrent writer, and even a 4 MiB window
-//! the local NFS client could deliver MB/s on bigger requests.
+//! gave the kernel less freedom to pipeline reads with writes. Direct
+//! pread per call lets Linux's readahead widen as it detects the
+//! sequential pattern, and naturally interleaves with writeback.
 //!
-//! Internally this source keeps a [`READAHEAD_BUF_BYTES`] (32 MiB)
+//! ## DONTNEED on the consumed window
-//! window pre-read from the file. `read_sectors(lba, count)` slices
-//! into the window if `[lba, lba+count)` is contained in it; otherwise
-//! the window is refilled (full-size aligned to the requested LBA's
-//! buffer position).
 //!
-//! ## Access pattern assumption
+//! Without page-cache eviction an 85 GB streaming ISO read pins the
+//! entire file in memory, starves the concurrent writer, and collapses
+//! mux throughput (observed: 2.7 MB/s mux on 0.21.5 vs. 70 MB/s
+//! isolated NFS reads). Every [`READ_DROP_CHUNK_BYTES`] of consumed
+//! bytes we call `posix_fadvise(DONTNEED)` over that window, mirroring
+//! the write-side [`crate::io::writeback::WritebackPipeline`] policy.
 //!
-//! The buffer is sized for **forward-sequential** reads (sweep, mux).
+//! ## Platform open hint
-//! Reverse-mode patch is range-local, so a refill per range works out
-//! fine (the buffer covers the whole range for typical bad-range
-//! sizes). Random-access reads thrash the buffer — at which point the
-//! 32 MiB pre-read is wasted work. We accept that: the use case is
-//! mux + sweep, both forward-sequential.
-//!
-//! Backward seeks rebuffer from the new LBA; partial reads at EOF
-//! return only the bytes that exist (the underlying file is shorter
-//! than a full buffer slot).
-//!
-//! ## Platform open hints
 //!
 //! On `open()` each platform issues its "sequential access expected"
-//! hint to the kernel so OS-level readahead widens. The hint lives in
+//! hint so OS-level readahead widens. The hint and the DONTNEED call
-//! a per-OS sibling module ([`linux::hint_sequential`] et al.) — no
+//! live in per-OS sibling modules ([`linux::hint_sequential`] et al.)
-//! inline `#[cfg]` in this file.
+//! — no inline `#[cfg]` in this file.
+//!
+//! ## Read-ahead prefetch
+//!
+//! After every consumed read we issue an OS-level prefetch hint for
+//! the next equivalent-sized window (`platform::prefetch`). The
+//! kernel queues that I/O asynchronously and returns immediately, so
+//! the next batch's read overlaps with the caller's processing of
+//! the current batch (decrypt + demux + mux). Without this the disk
+//! sits idle ~70% of each iteration because kernel SEQUENTIAL
+//! readahead alone (capped at `read_ahead_kb`, default 128 KB) is
+//! far smaller than our 16 MiB app-level batch.
 
 #[cfg(target_os = "linux")]
 mod linux;
@@ -60,66 +64,48 @@ use std::path::Path;
 use crate::error::{Error, Result};
 use crate::sector::SectorSource;
 
-/// Internal read-ahead buffer size. 32 MiB amortises one NFS round
-/// trip across ~16 k sectors — three orders of magnitude fewer trips
-/// than per-sector pread, and large enough to coast through a typical
-/// NFS server commit blip.
-///
-/// 0.21.2: shrunk from 32 MiB → 4 MiB. On NFS-backed ISOs with
-/// concurrent NFS writes (the mux phase), a 32 MiB refill bursts the
-/// TCP connection hard enough to starve the writer thread, observed
-/// empirically as a ~3× drop in sustained mux throughput on the
-/// rip1/unraid-1 setup. 4 MiB matches `rsize=1 MiB` × 4 round-trips
-/// and interleaves cleanly with writes.
-///
-/// Tweakable. Named const, not a magic number.
-pub const READAHEAD_BUF_BYTES: usize = 4 * 1024 * 1024;
-
 const SECTOR_SIZE: usize = 2048;
-/// Sectors per refill: [`READAHEAD_BUF_BYTES`] / [`SECTOR_SIZE`]. The
-/// buffer always tries to hold this many, except at the tail of the
-/// file where less data exists.
-const BUF_SECTORS: u32 = (READAHEAD_BUF_BYTES / SECTOR_SIZE) as u32;
 
-/// SectorSource backed by a file (ISO image) with an internal
-/// `READAHEAD_BUF_BYTES`-sized read-ahead window.
-///
-/// `read_sectors` is satisfied from the buffer when possible; otherwise
-/// a full-buffer refill is issued at the requested LBA's position and
-/// the call is re-tried against the freshly populated window.
 /// Bytes-read threshold per `posix_fadvise(DONTNEED)` drop on the
 /// read side. Mirrors `WRITEBACK_CHUNK_BYTES` so the read-side page
 /// cache stays bounded the same way the write side does.
 ///
-/// 0.21.6: re-added after empirical discovery that Phase 1 had silently
+/// 32 MiB is the empirically tuned value on the rip1 test bed (single
-/// dropped this from the pre-Phase-1 (0.20.7) hot path. Without it,
+/// 7200rpm HDD via SATA): smaller windows (8 / 16 MiB) shorten the
-/// 85 GB of streaming ISO reads pin the entire file in the kernel page
+/// kernel-readahead overlap and slow the producer; larger windows
-/// cache, starving the MKV writeback and collapsing mux throughput
+/// (64 / 128 MiB) let the page cache pin enough of the ISO to
-/// (observed: 2.7 MB/s mux on 0.21.5 vs. 70 MB/s isolated NFS reads).
+/// pressure concurrent writes. Override via `FREEMKV_READ_DROP_CHUNK_MIB`.
-const READ_DROP_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
+const READ_DROP_CHUNK_BYTES_DEFAULT: u64 = 32 * 1024 * 1024;
 
+fn read_drop_chunk_bytes() -> u64 {
+    std::env::var("FREEMKV_READ_DROP_CHUNK_MIB")
+        .ok()
+        .and_then(|v| v.parse::<u64>().ok())
+        .filter(|&n| n > 0)
+        .map(|n| n * 1024 * 1024)
+        .unwrap_or(READ_DROP_CHUNK_BYTES_DEFAULT)
+}
+
+/// SectorSource backed by a file (ISO image). Every `read_sectors`
+/// call is a direct `seek + read_exact` against the underlying file
+/// — kernel readahead handles prefetch, and every
+/// [`READ_DROP_CHUNK_BYTES_DEFAULT`] bytes of consumed data the
+/// platform's `DONTNEED` hook drops the consumed window from the
+/// page cache to bound memory pressure.
 pub struct FileSectorSource {
     file: File,
     /// Total file size in sectors. Constant after construction;
     /// surfaced via [`SectorSource::capacity_sectors`].
     capacity: u32,
-    /// 0.21.3+: the app-level buffer is no longer touched on the hot
+    /// Bytes read since the last DONTNEED drop. Drives the per-
-    /// path (every `read_sectors` is a direct pread). The fields are
+    /// [`read_drop_chunk_bytes`] page-cache eviction in read_sectors.
-    /// retained so a future per-source-type policy (e.g. a local-disk
-    /// source where batched reads ARE beneficial) can re-enable
-    /// buffering cleanly without re-plumbing the struct.
-    #[allow(dead_code)]
-    buf: Box<[u8]>,
-    #[allow(dead_code)]
-    buf_start_lba: u32,
-    buf_len_sectors: u32,
-    /// 0.21.6: bytes read since the last DONTNEED drop. Drives the
-    /// per-`READ_DROP_CHUNK_BYTES` page-cache eviction in read_sectors.
     bytes_read_since_drop: u64,
-    /// 0.21.6: file offset at which the current drop window starts.
+    /// File offset at which the current drop window starts. The next
-    /// The next DONTNEED drops from `drop_window_start` for
+    /// DONTNEED drops from `drop_window_start` for
     /// `bytes_read_since_drop` bytes.
     drop_window_start: u64,
+    /// Cached drop chunk size (resolved from env once at open).
+    drop_chunk_bytes: u64,
 }
 
 impl FileSectorSource {
@@ -148,59 +134,14 @@ impl FileSectorSource {
         // FS doesn't honour it).
         platform::hint_sequential(&file, len);
 
-        // Pre-allocate the buffer once. `vec![0u8; N].into_boxed_slice()`
-        // is the canonical way to fix the allocation size up-front;
-        // `Vec::with_capacity` would leave `len == 0` and force callers
-        // to do unsafe length manipulation to write into it.
-        let buf = vec![0u8; READAHEAD_BUF_BYTES].into_boxed_slice();
-
         Ok(Self {
             file,
             capacity,
-            buf,
-            buf_start_lba: 0,
-            buf_len_sectors: 0,
             bytes_read_since_drop: 0,
             drop_window_start: 0,
+            drop_chunk_bytes: read_drop_chunk_bytes(),
         })
     }
-
-    /// True if `[lba, lba + count)` is wholly inside the current
-    /// buffer window. `count == 0` is vacuously true.
-    #[allow(dead_code)]
-    fn buffer_covers(&self, lba: u32, count: u32) -> bool {
-        if self.buf_len_sectors == 0 {
-            return false;
-        }
-        let end = match lba.checked_add(count) {
-            Some(e) => e,
-            None => return false,
-        };
-        let buf_end = self.buf_start_lba.saturating_add(self.buf_len_sectors);
-        lba >= self.buf_start_lba && end <= buf_end
-    }
-
-    /// Refill the buffer so it starts at `lba`. Read as many sectors
-    /// as we have buffer space AND file capacity for. Caller has
-    /// already checked `lba < capacity`.
-    #[allow(dead_code)]
-    fn refill(&mut self, lba: u32) -> Result<()> {
-        debug_assert!(lba < self.capacity, "refill past capacity");
-        // Don't read past EOF — clamp the request to remaining
-        // sectors. partial-buffer-at-EOF behaviour is intentional.
-        let want = BUF_SECTORS.min(self.capacity - lba);
-        let want_bytes = want as usize * SECTOR_SIZE;
-        let offset = lba as u64 * SECTOR_SIZE as u64;
-        self.file
-            .seek(SeekFrom::Start(offset))
-            .map_err(|e| Error::IoError { source: e })?;
-        self.file
-            .read_exact(&mut self.buf[..want_bytes])
-            .map_err(|e| Error::IoError { source: e })?;
-        self.buf_start_lba = lba;
-        self.buf_len_sectors = want;
-        Ok(())
-    }
 }
 
 impl SectorSource for FileSectorSource {
@@ -226,20 +167,6 @@ impl SectorSource for FileSectorSource {
         if count == 0 {
             return Ok(0);
         }
-        // 0.21.3: bypass the application-level buffer entirely.
-        //
-        // Empirically the 32 MiB readahead window (0.21.0–0.21.1) and the
-        // 4 MiB shrink (0.21.2) both regressed mux throughput vs the
-        // pre-Phase-1 0.20.7 baseline on NFS bidirectional workloads
-        // (sweep ~25 MB/s OK; mux dropped from 18 → 7-8 → 5-6 MB/s).
-        // Direct pread per call lets the kernel's own readahead policy
-        // run, which interleaves naturally with concurrent NFS writes on
-        // the same TCP connection.
-        //
-        // Buffer fields are retained (currently unused on this path) so
-        // any future per-source policy can be reintroduced without
-        // re-plumbing structure. `refill` / `buffer_covers` are kept too
-        // (still exercised by the tests so the API contract is locked).
         let offset = lba as u64 * SECTOR_SIZE as u64;
         self.file
             .seek(SeekFrom::Start(offset))
@@ -247,15 +174,21 @@ impl SectorSource for FileSectorSource {
         self.file
             .read_exact(&mut out[..bytes])
             .map_err(|e| Error::IoError { source: e })?;
-        self.buf_len_sectors = 0;
 
-        // 0.21.6: periodic page-cache eviction on the read side. Without
+        // Queue the next batch's read with the kernel before the
+        // caller starts processing what we just returned. readahead()
+        // is non-blocking — it queues I/O and returns, so the kernel
+        // pulls those pages into cache while the consumer (decrypt +
+        // demux + mux) runs. Next read_sectors call hits a warm cache.
+        platform::prefetch(&self.file, offset + bytes as u64, bytes as u64);
+
+        // Periodic page-cache eviction on the read side. Without
         // this, an 85 GB streaming ISO read pins the entire file in
-        // kernel page cache, which starves concurrent NFS writes (the
+        // the kernel page cache, which starves concurrent writes and
-        // MKV output) and collapses mux throughput. Mirrors the
+        // collapses mux throughput. Mirrors the write-side
-        // write-side WritebackPipeline's DONTNEED policy.
+        // WritebackPipeline's DONTNEED policy.
         self.bytes_read_since_drop += bytes as u64;
-        if self.bytes_read_since_drop >= READ_DROP_CHUNK_BYTES {
+        if self.bytes_read_since_drop >= self.drop_chunk_bytes {
             let drop_start = self.drop_window_start;
             let drop_len = self.bytes_read_since_drop;
             platform::drop_window(&self.file, drop_start, drop_len);
@@ -287,11 +220,15 @@ mod tests {
         f.flush().unwrap();
     }
 
+    /// Sectors used by spanning-boundary tests. Pick something that
+    /// exercises multi-megabyte reads without making test ISOs huge.
+    /// 8192 sectors = 16 MiB — large enough to cross any readahead
+    /// chunk size we set the kernel hint to.
+    const TEST_SPAN_SECTORS: u32 = 8192;
+
     #[test]
     fn sequential_reads_match_file() {
-        // Two full buffer windows + a tail = exercise refill across
+        let total = TEST_SPAN_SECTORS * 2 + 17;
-        // boundaries.
-        let total = BUF_SECTORS * 2 + 17;
         let dir = tempdir().unwrap();
         let path = dir.path().join("seq.iso");
         make_iso(&path, total);
@@ -311,28 +248,15 @@ mod tests {
     }
 
     #[test]
-    fn multi_sector_read_spanning_buffer_boundary() {
+    fn multi_sector_read_across_chunk_boundary() {
-        // A read that lands exactly on the last sector of the buffer
+        let total = TEST_SPAN_SECTORS * 2;
-        // plus the first sector of the next refill must rebuffer
-        // mid-read. Bypass path triggers when count > BUF_SECTORS; we
-        // want the in-window path, so count stays small but
-        // straddles the boundary.
-        let total = BUF_SECTORS * 2;
         let dir = tempdir().unwrap();
         let path = dir.path().join("span.iso");
         make_iso(&path, total);
 
         let mut src = FileSectorSource::open(&path).unwrap();
 
-        // Prime: read sector 0. (0.21.3+: app-level buffer is bypassed,
+        let span_lba = TEST_SPAN_SECTORS - 2;
-        // so we don't assert internal buf state here — just exercise
-        // the read path.)
-        let mut got = vec![0u8; SECTOR_SIZE];
-        src.read_sectors(0, 1, &mut got, false).unwrap();
-
-        // Now read 4 sectors crossing what used to be the buffer
-        // boundary. Still a valid SectorSource-contract test.
-        let span_lba = BUF_SECTORS - 2;
         let mut buf4 = vec![0u8; SECTOR_SIZE * 4];
         src.read_sectors(span_lba, 4, &mut buf4, false).unwrap();
         for i in 0..4 {
@@ -345,10 +269,10 @@ mod tests {
     }
 
     #[test]
-    fn backward_seek_rebuffers() {
+    fn backward_seek_reads_correct_bytes() {
-        // Read forward across two windows, then jump back to sector
+        // Read forward then jump back: the SectorSource contract is
-        // 0. Buffer must refill from the start.
+        // byte-correctness regardless of access pattern.
-        let total = BUF_SECTORS * 2 + 5;
+        let total = TEST_SPAN_SECTORS * 2 + 5;
         let dir = tempdir().unwrap();
         let path = dir.path().join("back.iso");
         make_iso(&path, total);
@@ -356,24 +280,17 @@ mod tests {
         let mut src = FileSectorSource::open(&path).unwrap();
         let mut got = vec![0u8; SECTOR_SIZE];
 
-        // Forward to the second window.
+        src.read_sectors(TEST_SPAN_SECTORS + 1, 1, &mut got, false)
-        src.read_sectors(BUF_SECTORS + 1, 1, &mut got, false)
             .unwrap();
-
-        // Backward to sector 0. (0.21.3+: app-level buffer is bypassed
-        // so we only assert the byte-level contract, not internal
-        // buffer state.)
         src.read_sectors(0, 1, &mut got, false).unwrap();
         assert!(got.iter().all(|b| *b == 0));
     }
 
     #[test]
-    fn partial_buffer_at_eof() {
+    fn read_at_eof_returns_correct_bytes() {
-        // File is smaller than one buffer window. The buffer must
+        // File smaller than the readahead chunk — reads near EOF must
-        // populate with only the available sectors and reads must
+        // still return correct bytes.
-        // still succeed.
         let total: u32 = 100;
-        assert!(total < BUF_SECTORS);
         let dir = tempdir().unwrap();
         let path = dir.path().join("small.iso");
         make_iso(&path, total);
@@ -382,39 +299,27 @@ mod tests {
         assert_eq!(src.capacity_sectors(), total);
 
         let mut got = vec![0u8; SECTOR_SIZE];
-        // First read at sector 0.
         src.read_sectors(0, 1, &mut got, false).unwrap();
-
-        // Read the very last sector. (0.21.3+: app-level buffer is
-        // bypassed; the test still verifies that EOF-region reads
-        // return correct bytes.)
         src.read_sectors(total - 1, 1, &mut got, false).unwrap();
         let expected = ((total - 1) & 0xff) as u8;
         assert!(got.iter().all(|b| *b == expected));
     }
 
     #[test]
-    fn oversized_read_bypasses_buffer() {
+    fn large_single_read() {
-        // A request larger than the buffer must not deadlock the
+        // A multi-MB single read must work — the implementation has
-        // refill (which only loads BUF_SECTORS at a time). Bypass
+        // no app-level chunking, so this just exercises the direct
-        // path handles it via direct pread.
+        // pread path on a larger request.
-        let total = BUF_SECTORS + 100;
+        let total = TEST_SPAN_SECTORS + 100;
         let dir = tempdir().unwrap();
-        let path = dir.path().join("over.iso");
+        let path = dir.path().join("big.iso");
         make_iso(&path, total);
 
         let mut src = FileSectorSource::open(&path).unwrap();
-        // Read more than BUF_SECTORS in one call. count is u16, so we
+        let req = (TEST_SPAN_SECTORS + 1) as u16;
-        // can't actually exceed BUF_SECTORS (16k) — but the path also
-        // triggers via `out.len() / SECTOR_SIZE > BUF_SECTORS` check
-        // implicitly because count > BUF_SECTORS. BUF_SECTORS for
-        // 32 MiB is 16384, which does fit in u16 (max 65535). Cap
-        // at BUF_SECTORS + 1 to exercise the bypass.
-        let req = (BUF_SECTORS + 1) as u16;
         let req_bytes = req as usize * SECTOR_SIZE;
         let mut big = vec![0u8; req_bytes];
         src.read_sectors(0, req, &mut big, false).unwrap();
-        // Spot-check sector 0 and the last requested sector.
         assert!(big[..SECTOR_SIZE].iter().all(|b| *b == 0));
         let last_lba = req as u32 - 1;
         let exp = (last_lba & 0xff) as u8;
@@ -425,4 +330,32 @@ mod tests {
                 .all(|b| *b == exp)
         );
     }
+
+    #[test]
+    fn drop_chunk_size_env_override() {
+        // Explicit 8 MiB via env var.
+        // SAFETY: tests in this crate are single-threaded per the
+        // default cargo test harness, but std::env::set_var is
+        // declared `unsafe` since Rust 2024 (it can race with other
+        // threads / TLS). For a test that runs in-process before any
+        // FileSectorSource construction this is safe in practice.
+        unsafe {
+            std::env::set_var("FREEMKV_READ_DROP_CHUNK_MIB", "8");
+        }
+        assert_eq!(read_drop_chunk_bytes(), 8 * 1024 * 1024);
+
+        unsafe {
+            std::env::remove_var("FREEMKV_READ_DROP_CHUNK_MIB");
+        }
+        assert_eq!(read_drop_chunk_bytes(), READ_DROP_CHUNK_BYTES_DEFAULT);
+
+        // Garbage env value falls back to default.
+        unsafe {
+            std::env::set_var("FREEMKV_READ_DROP_CHUNK_MIB", "not-a-number");
+        }
+        assert_eq!(read_drop_chunk_bytes(), READ_DROP_CHUNK_BYTES_DEFAULT);
+        unsafe {
+            std::env::remove_var("FREEMKV_READ_DROP_CHUNK_MIB");
+        }
+    }
 }

src/io/file_sector_source/other.rs

@@ -7,3 +7,5 @@ use std::fs::File;
 pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {}
 
 pub(super) fn drop_window(_file: &File, _start: u64, _len: u64) {}
+
+pub(super) fn prefetch(_file: &File, _offset: u64, _len: u64) {}

src/io/file_sector_source/windows.rs

@@ -22,3 +22,9 @@ pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {
 /// equivalent. The kernel does its own working-set management. No-op
 /// for now.
 pub(super) fn drop_window(_file: &File, _start: u64, _len: u64) {}
+
+/// Windows async-prefetch hint. With FILE_FLAG_SEQUENTIAL_SCAN at
+/// open the kernel already prefetches aggressively, so there's no
+/// per-range hint we'd add on top. No-op stub for parity with the
+/// posix platforms.
+pub(super) fn prefetch(_file: &File, _offset: u64, _len: u64) {}

src/io/mod.rs

@@ -11,9 +11,12 @@
 //! to exhibit the same pathology for this access pattern.
 //!
 //! `FileSectorSource` is the read-side dual — it implements
-//! [`crate::sector::SectorSource`] for an ISO file with an internal
+//! [`crate::sector::SectorSource`] for an ISO file using direct
-//! 32 MiB read-ahead buffer that amortises NFS round-trip latency
+//! `pread`-equivalent calls so the kernel's own readahead policy runs
-//! across thousands of sector reads.
+//! (which interleaves naturally with the concurrent writeback). It
+//! pairs that with periodic `posix_fadvise(DONTNEED)` drops on the
+//! consumed window so an 85 GB streaming ISO read doesn't fill the
+//! page cache and starve the concurrent MKV write.
 //!
 //! `Pipeline` + `Sink` (0.18) is the generic producer/consumer primitive
 //! used by sweep, patch, and mux to overlap reads with writes via a
@@ -25,21 +28,21 @@
 
 pub(crate) mod bounded;
 pub mod byte_channel;
+pub mod byte_prefetcher;
 pub mod file_sector_source;
 pub mod sink;
 mod writeback;
 mod writeback_file;
 
+#[cfg(target_os = "macos")]
+pub(crate) mod platform_macos;
+
 pub mod pipeline;
 
 pub(crate) use writeback_file::WritebackFile;
 
-// Re-exports for the 0.18 redesign. Sweep + patch are both wired up
+// Re-exports for the 0.18 redesign. Sweep, patch, and mux are all
-// (disc/sweep.rs, disc/patch.rs); mux migrates separately in autorip.
+// wired up (disc/sweep.rs, disc/patch.rs, autorip's ripper/mux.rs).
-// `WRITE_THROUGH_DEPTH` is patch-specific and has no other in-tree
-// caller — the targeted `#[allow]` keeps the re-export visible without
-// dragging the rest of the module under `dead_code`.
-#[allow(unused_imports)]
 pub use pipeline::{
     DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, READ_PIPELINE_DEPTH, Sink, WRITE_PIPELINE_DEPTH,
     WRITE_THROUGH_DEPTH,

src/io/pipeline.rs

@@ -53,26 +53,26 @@ use crate::halt::Halt;
 /// caller has already lost the rip.
 pub const JOIN_TIMEOUT_SECS: u64 = 600;
 
-/// Polling slice for the halt-aware send/finish loops. Mirrors the
-/// `bounded_syscall` cadence (250 ms) so halt observation feels equally
-/// responsive across both primitives.
-const POLL_INTERVAL: Duration = Duration::from_millis(250);
-
 /// Halt-check cadence for the send loop. Producer blocks on
 /// [`crossbeam_channel::Sender::send_timeout`] for this slice — the
 /// kernel wakes it the instant the consumer drains a slot, so on the
 /// happy path there's no throughput cap from this primitive at all
 /// (the cap is whatever the underlying medium can sustain). When the
-/// consumer is genuinely wedged, the timeout fires every 250 ms and
+/// consumer is genuinely wedged, the timeout fires every
-/// the producer checks the halt token; that's the latency a stop
+/// [`crate::halt::POLL_INTERVAL`] and the producer checks the halt
-/// request will observe.
+/// token; that's the latency a stop request will observe.
+///
+/// Single source of truth lives in [`crate::halt::POLL_INTERVAL`]
+/// (also used by `bounded_syscall`). Aliased here for readability of
+/// the send/finish call sites below.
 ///
 /// 0.21.7 replaced an old `std::sync::mpsc::sync_channel` + 50 ms
 /// `thread::sleep` polling loop that capped mux throughput at
 /// ~20 frames/sec ≈ 1 MB/s on saturated channels. See
 /// (internal)/memory/feedback_send_with_halt_poll_throttle.md
 /// for the multi-day diagnostic that surfaced it.
-const SEND_HALT_CHECK_INTERVAL: Duration = Duration::from_millis(250);
+use crate::halt::POLL_INTERVAL;
+const SEND_HALT_CHECK_INTERVAL: Duration = POLL_INTERVAL;
 
 /// Check if verbose debug logging is enabled via FREEMKV_DEBUG env var.
 pub fn debug_enabled() -> bool {
@@ -100,8 +100,7 @@ pub const WRITE_PIPELINE_DEPTH: usize = 16;
 /// Channel depth for write-through pipelines. Each `send` fully
 /// drains before the next can enqueue. Use this when the producer
 /// must observe consumer side-effects (e.g. mapfile state) before
-/// emitting the next item.
+/// emitting the next item. Currently used by `disc::patch`.
-#[allow(dead_code)]
 pub const WRITE_THROUGH_DEPTH: usize = 1;
 
 /// Outcome of [`Sink::apply`]: either keep feeding items
@@ -162,11 +161,10 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// propagated rather than panicked.
     ///
     /// Sweep uses [`Pipeline::spawn_named`] directly so the consumer
-    /// thread shows up as `freemkv-sweep-consumer`; this function has
+    /// thread shows up as `freemkv-sweep-consumer`; mux uses
-    /// no in-tree caller yet. Patch and mux migrate in later 0.18
+    /// `freemkv-mux-consumer`. `Pipeline::spawn` (this function, with
-    /// slices. The targeted `#[allow]` is removed when one of them
+    /// the default name) is used by `disc::patch` and by the unit
-    /// lands on the default name.
+    /// tests in this module.
-    #[allow(dead_code)]
     pub fn spawn<S: Sink<I, Output = R>>(depth: usize, sink: S) -> Result<Self, Error> {
         Self::spawn_named("freemkv-pipeline-consumer", depth, sink)
     }

src/io/platform_macos.rs

@@ -0,0 +1,38 @@
+//! Shared macOS `fcntl(F_PREALLOCATE)` definitions.
+//!
+//! The `libc` crate doesn't expose these symbols across all macOS SDK
+//! versions, so we define them locally with values from
+//! `/usr/include/sys/fcntl.h`. Two call sites (
+//! [`crate::io::writeback_file`] and [`crate::io::sink::preallocate`])
+//! need the same constants and `fstore_t` layout — keeping a single
+//! source of truth here prevents the two copies from drifting.
+//!
+//! Module-level cfg gate lives in the parent (`io/mod.rs`); this file
+//! is only compiled on macOS, so no inner `#![cfg]` is needed.
+
+/// `fcntl(F_PREALLOCATE)` command number from `sys/fcntl.h`.
+pub(crate) const F_PREALLOCATE: libc::c_int = 42;
+
+/// Anchor preallocation at the current physical EOF.
+pub(crate) const F_PEOFPOSMODE: libc::c_int = 3;
+
+/// Prefer a contiguous allocation. Try this first; on `EINVAL` (no
+/// contiguous run of that size), fall back to `F_ALLOCATEALL`.
+pub(crate) const F_ALLOCATECONTIG: libc::c_uint = 0x0000_0002;
+
+/// Allow non-contiguous allocation. Stronger guarantee than just
+/// asking for `F_ALLOCATECONTIG` because the kernel will piece
+/// together fragments rather than failing.
+pub(crate) const F_ALLOCATEALL: libc::c_uint = 0x0000_0004;
+
+/// `fstore_t` from `sys/fcntl.h`. `repr(C)` because we hand it to
+/// `fcntl(F_PREALLOCATE)` which writes through the pointer.
+#[repr(C)]
+#[derive(Clone, Copy)]
+pub(crate) struct Fstore {
+    pub fst_flags: libc::c_uint,
+    pub fst_posmode: libc::c_int,
+    pub fst_offset: libc::off_t,
+    pub fst_length: libc::off_t,
+    pub fst_bytesalloc: libc::off_t,
+}

src/io/sink/preallocate/macos.rs

@@ -7,23 +7,9 @@
 use std::fs::File;
 use std::os::unix::io::AsRawFd;
 
-// Mirror the Darwin `fstore_t` struct from `<sys/fcntl.h>`. libc on
+use crate::io::platform_macos::{
-// some Rust toolchains/versions doesn't ship this binding, so define
+    F_ALLOCATEALL, F_ALLOCATECONTIG, F_PEOFPOSMODE, F_PREALLOCATE, Fstore,
-// it locally with the layout the kernel ABI requires.
+};
-#[repr(C)]
-struct Fstore {
-    fst_flags: libc::c_uint,
-    fst_posmode: libc::c_int,
-    fst_offset: libc::off_t,
-    fst_length: libc::off_t,
-    fst_bytesalloc: libc::off_t,
-}
-
-// Constants from <sys/fcntl.h>.
-const F_PREALLOCATE: libc::c_int = 42;
-const F_ALLOCATECONTIG: libc::c_uint = 0x0000_0002;
-const F_ALLOCATEALL: libc::c_uint = 0x0000_0004;
-const F_PEOFPOSMODE: libc::c_int = 3;
 
 pub(super) fn preallocate_impl(file: &File, size_bytes: u64) {
     let fd = file.as_raw_fd();

src/io/writeback/linux.rs

@@ -300,35 +300,22 @@ impl WritebackPipeline {
     }
 }
 
-/// Probe whether `fd` lives on an NFS mount via `fstatfs`. Returns
+/// Probe whether `fd` lives on an NFS mount. Thin wrapper around
-/// `false` on any error — we fail open, not closed: better to run the
+/// [`crate::platform::fs_type::detect_fd`] so writeback policy and
-/// normal local-storage path on a misdetected NFS mount (and surface
+/// general-purpose fs-type classification stay in sync (same magic
-/// the freeze loudly via the timeout) than to needlessly disable
+/// numbers, same musl-vs-glibc cast handling).
-/// writeback bounding on every local file because of a transient
+///
-/// stat error.
+/// Fails open: any classification other than NFS counts as "not NFS"
+/// (including `Unknown` on `fstatfs` error) — better to run the
+/// normal local-storage path on a misdetected NFS mount and surface
+/// the freeze loudly via [`WAIT_AFTER_TIMEOUT`] than to needlessly
+/// disable writeback bounding on every local file because of a
+/// transient stat error.
 fn detect_nfs(fd: RawFd) -> bool {
-    // `libc::statfs` is repr(C) with a fixed layout; zeroing is the
+    matches!(
-    // documented init pattern for the kernel uapi struct.
+        crate::platform::fs_type::detect_fd(fd),
-    let mut buf: libc::statfs = unsafe { std::mem::zeroed() };
+        crate::platform::fs_type::FsType::Nfs
-    let rc = unsafe { libc::fstatfs(fd, &mut buf) };
+    )
-    if rc != 0 {
-        let errno = std::io::Error::last_os_error();
-        tracing::warn!(
-            target: "mux",
-            "WritebackPipeline fstatfs(fd={fd}) failed: {errno} — defaulting is_nfs=false",
-        );
-        return false;
-    }
-    // `f_type` is signed (`__fsword_t`) on glibc and unsigned
-    // (`c_ulong`) on musl. Cast both sides to i64 for a portable
-    // comparison. On glibc x86_64 both already are i64 — clippy flags
-    // the cast as unnecessary on that target only, but we need it for
-    // musl, so silence the lint.
-    #[allow(clippy::unnecessary_cast)]
-    let f_type = buf.f_type as i64;
-    #[allow(clippy::unnecessary_cast)]
-    let nfs_magic = libc::NFS_SUPER_MAGIC as i64;
-    f_type == nfs_magic
 }
 
 /// Run `sync_file_range(WAIT_AFTER)` on a worker thread and wait up
@@ -356,3 +343,117 @@ fn wait_after_with_timeout(fd: RawFd, off: u64, len: u64) -> Option<u64> {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::NamedTempFile;
+
+    /// Helper: build a `WritebackPipeline` over a local tempfile. On
+    /// every test rig (linux dev box, CI) the tempfile lives on a
+    /// local FS, so `is_nfs=false` and `skip_wait` returns false until
+    /// we explicitly mark the pipeline degraded.
+    fn local_pipeline(chunk_bytes: u64) -> (NamedTempFile, WritebackPipeline) {
+        let f = NamedTempFile::new().expect("tempfile create");
+        let pipeline = WritebackPipeline::new(f.as_file(), 0, chunk_bytes);
+        (f, pipeline)
+    }
+
+    #[test]
+    fn new_pipeline_starts_active() {
+        let (_f, p) = local_pipeline(32 * 1024 * 1024);
+        assert!(!p.is_nfs, "local tempfile must not classify as NFS");
+        assert!(!p.degraded.load(Ordering::Relaxed));
+        assert!(!p.skip_wait(), "fresh local pipeline must not skip wait");
+    }
+
+    #[test]
+    fn degraded_flag_short_circuits_wait() {
+        let (_f, p) = local_pipeline(32 * 1024 * 1024);
+        assert!(!p.skip_wait());
+        p.degraded.store(true, Ordering::Relaxed);
+        assert!(
+            p.skip_wait(),
+            "degraded flag must force the wait+dontneed bypass"
+        );
+    }
+
+    #[test]
+    fn record_wait_grows_chunk_on_high_p95() {
+        let (_f, mut p) = local_pipeline(16 * 1024 * 1024);
+        // Fill the window with samples above the grow threshold.
+        for _ in 0..ADAPTIVE_WINDOW {
+            p.record_wait(ADAPTIVE_GROW_MS + 50);
+        }
+        assert!(
+            p.chunk_bytes > 16 * 1024 * 1024,
+            "chunk should have grown; got {}",
+            p.chunk_bytes
+        );
+        assert!(p.chunk_bytes <= CHUNK_BYTES_MAX);
+    }
+
+    #[test]
+    fn record_wait_shrinks_chunk_on_low_p95() {
+        let (_f, mut p) = local_pipeline(64 * 1024 * 1024);
+        for _ in 0..ADAPTIVE_WINDOW {
+            p.record_wait(1); // well under ADAPTIVE_SHRINK_MS
+        }
+        assert!(
+            p.chunk_bytes < 64 * 1024 * 1024,
+            "chunk should have shrunk; got {}",
+            p.chunk_bytes
+        );
+        assert!(p.chunk_bytes >= CHUNK_BYTES_MIN);
+    }
+
+    #[test]
+    fn record_wait_no_op_below_window_fill() {
+        let (_f, mut p) = local_pipeline(16 * 1024 * 1024);
+        let initial = p.chunk_bytes;
+        // Only push a few samples; window not full → no adaptation.
+        for _ in 0..(ADAPTIVE_WINDOW - 1) {
+            p.record_wait(ADAPTIVE_GROW_MS + 100);
+        }
+        assert_eq!(
+            p.chunk_bytes, initial,
+            "chunk must not change before window is full"
+        );
+    }
+
+    #[test]
+    fn record_wait_clamps_to_chunk_bounds() {
+        // Grow past the max.
+        let (_f, mut p) = local_pipeline(CHUNK_BYTES_MAX);
+        for _ in 0..ADAPTIVE_WINDOW {
+            p.record_wait(ADAPTIVE_GROW_MS + 1000);
+        }
+        assert_eq!(p.chunk_bytes, CHUNK_BYTES_MAX, "must clamp to MAX");
+
+        // Shrink past the min.
+        let (_f, mut p) = local_pipeline(CHUNK_BYTES_MIN);
+        for _ in 0..ADAPTIVE_WINDOW {
+            p.record_wait(0);
+        }
+        assert_eq!(p.chunk_bytes, CHUNK_BYTES_MIN, "must clamp to MIN");
+    }
+
+    #[test]
+    fn detect_nfs_local_file_is_false() {
+        // Local tempfile must not classify as NFS. This locks in the
+        // consolidation through `crate::platform::fs_type::detect_fd`.
+        let f = NamedTempFile::new().expect("tempfile create");
+        use std::os::unix::io::AsRawFd;
+        assert!(!detect_nfs(f.as_file().as_raw_fd()));
+    }
+
+    #[test]
+    fn note_progress_below_chunk_is_noop() {
+        let (_f, mut p) = local_pipeline(32 * 1024 * 1024);
+        // No-op return before crossing the first chunk boundary.
+        let before = p.chunk_count;
+        p.note_progress(1024); // < 32 MiB
+        assert_eq!(p.chunk_count, before);
+        assert!(p.pending.is_none());
+    }
+}

src/io/writeback_file/macos.rs

@@ -16,31 +16,14 @@ use std::io;
 use std::os::unix::io::AsRawFd;
 use std::time::Duration;
 
-/// libc `F_PREALLOCATE` — not exposed by the `libc` crate on all macOS
+use crate::io::platform_macos::{
-/// SDK versions, so define it here.
+    F_ALLOCATEALL, F_ALLOCATECONTIG, F_PEOFPOSMODE, F_PREALLOCATE, Fstore,
-const F_PREALLOCATE: libc::c_int = 42;
+};
-/// Allocate from current EOF.
-const F_PEOFPOSMODE: libc::c_int = 3;
-/// Hint: contiguous extent preferred.
-const F_ALLOCATECONTIG: libc::c_uint = 0x00000002;
-/// Allocate all the requested bytes (fall back to non-contig if needed).
-const F_ALLOCATEALL: libc::c_uint = 0x00000004;
 
 /// `fcntl(F_FULLFSYNC)` opcode. Documented in `man 2 fcntl` on macOS;
 /// not in the `libc` crate as a named constant.
 const F_FULLFSYNC: libc::c_int = 51;
 
-/// `fstore_t` layout matches `sys/fcntl.h`. Repr is C-stable so we can
-/// build it manually.
-#[repr(C)]
-struct Fstore {
-    fst_flags: libc::c_uint,
-    fst_posmode: libc::c_int,
-    fst_offset: libc::off_t,
-    fst_length: libc::off_t,
-    fst_bytesalloc: libc::off_t,
-}
-
 pub(super) fn preallocate(file: &File, size_bytes: u64) {
     let mut fst = Fstore {
         fst_flags: F_ALLOCATECONTIG | F_ALLOCATEALL,

src/io/writeback_file/mod.rs

@@ -73,10 +73,21 @@ use std::path::Path;
 use super::writeback::WritebackPipeline;
 
 /// Granularity at which the Linux writeback pipeline issues
-/// `sync_file_range` pairs. 32 MiB is the empirically best value:
+/// `sync_file_range` pairs. 32 MiB is the empirically best value on
-/// iter8 = 28.7, iter9 (64 MiB) = 27.5, iter11 (128 MiB) = 16.6,
+/// the rip1 test bed (NFS to unraid-1 over 1 GbE, single-disk SAS):
-/// iter6 (8 MiB) = 15.8. Locked.
+/// 8 MiB / 64 MiB / 128 MiB all measured worse in the 0.21.x mux
-const WRITEBACK_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
+/// iteration runs. Override via `FREEMKV_WRITEBACK_CHUNK_MIB` —
+/// faster backends (NVMe, RAID) may tolerate larger windows.
+const WRITEBACK_CHUNK_BYTES_DEFAULT: u64 = 32 * 1024 * 1024;
+
+fn writeback_chunk_bytes() -> u64 {
+    std::env::var("FREEMKV_WRITEBACK_CHUNK_MIB")
+        .ok()
+        .and_then(|v| v.parse::<u64>().ok())
+        .filter(|&n| n > 0)
+        .map(|n| n * 1024 * 1024)
+        .unwrap_or(WRITEBACK_CHUNK_BYTES_DEFAULT)
+}
 
 pub(crate) struct WritebackFile {
     file: File,
@@ -91,7 +102,7 @@ impl WritebackFile {
     /// or appended files).
     pub(crate) fn new(mut file: File) -> io::Result<Self> {
         let pos = file.stream_position()?;
-        let pipeline = WritebackPipeline::new(&file, pos, WRITEBACK_CHUNK_BYTES);
+        let pipeline = WritebackPipeline::new(&file, pos, writeback_chunk_bytes());
         Ok(Self {
             file,
             pipeline,

src/lib.rs

@@ -154,7 +154,7 @@ pub use profile::DriveProfile;
 // don't touch `DecryptKeys` directly — `DiscStream::new(reader, title, keys, …)`
 // accepts whatever `Disc::decrypt_keys()` returned. `decrypt_sectors()` is
 // for callers that operate on raw sector buffers (e.g. ISO patching).
-pub use decrypt::{DecryptKeys, decrypt_sectors};
+pub use decrypt::{DecryptKeys, decrypt_sectors, decrypt_threads, set_decrypt_threads};
 
 // ─── Disc structure ─────────────────────────────────────────────────────────
 //
@@ -213,7 +213,8 @@ pub use mux::{InputOptions, StreamUrl, input, output, parse_url};
 // `SectorSource` to get plaintext sectors out.
 pub use scsi::{DriveInfo, ScsiSense, ScsiTransport, drive_has_disc, list_drives};
 pub use sector::{
-    DecryptingSectorSource, FileSectorSink, FileSectorSource, SectorSink, SectorSource,
+    DecryptingSectorSource, FileSectorSink, FileSectorSource, PrefetchedSectorSource, SectorSink,
+    SectorSource,
 };
 pub use speed::DriveSpeed;
 pub use udf::{UdfFs, read_filesystem};

src/mux/codec/h264.rs

@@ -176,14 +176,16 @@ impl<'a> Iterator for NalIterator<'a> {
 }
 
 /// Find the position of the next start code (00 00 01) at or after `from`.
+///
+/// Backed by `memchr::memmem::find` for SIMD-accelerated bytestring
+/// search. On AVX2-capable x86_64 this runs ~5–10× the byte-by-byte
+/// scan that preceded it; on a 200 KB UHD HEVC frame the saving is
+/// in the hundreds of microseconds per call.
 pub fn find_start_code(data: &[u8], from: usize) -> Option<usize> {
     if data.len() < from + 3 {
         return None;
     }
-    // Range excludes last 2 bytes since we read 3 bytes at each position.
+    memchr::memmem::find(&data[from..], b"\x00\x00\x01").map(|rel| from + rel)
-    // data.len()-2 as exclusive upper bound means last checked index is data.len()-3,
-    // which accesses data[len-3], data[len-2], data[len-1] — all valid.
-    (from..data.len() - 2).find(|&i| data[i] == 0x00 && data[i + 1] == 0x00 && data[i + 2] == 0x01)
 }
 
 /// Skip past the start code at position `pos`, returning the first byte after it.

src/mux/codec/hevc.rs

@@ -52,7 +52,10 @@ impl CodecParser for HevcParser {
         let pts_ns = pes.dts.or(pes.pts).map(pts_to_ns).unwrap_or(0);
         let data = &pes.data;
         let mut keyframe = false;
-        let mut frame_data = Vec::new();
+        // Pre-size: output is ~input bytes with a few 4-byte length
+        // prefixes added. UHD frames are 150-300 KB; the unsized Vec
+        // growth chain otherwise reallocs 5-7× per frame.
+        let mut frame_data = Vec::with_capacity(data.len() + 64);
 
         // Single-pass NAL scan: extract params, detect keyframes, build length-prefixed output
         let mut pos = 0;

src/mux/demux_thread.rs

@@ -0,0 +1,350 @@
+//! `DemuxThread` — runs the read+decrypt+demux pipeline on a
+//! dedicated thread, feeding completed `PesPacket` batches to the
+//! caller via a bounded channel.
+//!
+//! ## Why a second worker thread
+//!
+//! With [`crate::sector::PrefetchedSectorSource`] alone, read+decrypt
+//! already runs on a producer thread; the *consumer* (main) thread
+//! still serialises `ts_demuxer.feed` (M2TS parsing) with the codec
+//! parsers. Profiling on the rip1 testbed showed feed at ~37 % and
+//! codec parse at ~44 % of consumer wall time — i.e. feed is heavy
+//! enough that pipelining it with parse pays for itself.
+//!
+//! Splitting them: feed runs in [`DemuxThread`]; the consumer thread
+//! receives `Vec<PesPacket>` batches and runs codec parse + frame
+//! emission only. Total throughput becomes `1/max(feed, parse)`
+//! instead of `1/(feed + parse)`.
+//!
+//! ## Lifecycle
+//!
+//! [`DemuxThread::spawn`] takes ownership of the inner reader and the
+//! demuxer state, returns a handle plus a `Receiver<DemuxBatch>`.
+//! Dropping the handle closes the channel which signals the thread
+//! to exit; the join in `Drop::drop` is bounded.
+
+use crate::halt::Halt;
+use crate::sector::SectorSource;
+use crossbeam_channel::{Receiver, Sender, bounded};
+use std::thread::JoinHandle;
+
+/// Output channel depth. Two batches in flight keeps the consumer
+/// (codec parser) busy without piling up demuxed bytes if it stalls.
+const DEMUX_CHANNEL_DEPTH: usize = 2;
+
+/// One demuxed batch flowing from the demux thread to the consumer.
+pub enum DemuxBatch {
+    /// Successfully demuxed PesPackets — non-empty.
+    Ts(Vec<super::ts::PesPacket>),
+    Ps(Vec<super::ps::PsPacket>),
+    /// Underlying reader returned an error. Terminal.
+    Err(std::io::Error),
+}
+
+/// Spawned demux thread. Drop joins.
+///
+/// In zero-copy mode the thread also owns an opaque
+/// `producer_shell: Option<Box<dyn Send>>` — the join handle of the
+/// upstream producer (sector or byte prefetcher). Dropping the
+/// `DemuxThread` runs the shell's `Drop`, which joins the producer.
+/// `Box<dyn Send>` rather than a concrete type so the same demux
+/// worker can be wired behind either prefetcher kind.
+pub struct DemuxThread {
+    handle: Option<JoinHandle<()>>,
+    #[allow(dead_code)]
+    producer_shell: Option<Box<dyn Send>>,
+}
+
+impl DemuxThread {
+    /// Spawn the demux thread. Returns the thread handle and a
+    /// receiver for [`DemuxBatch`] items.
+    ///
+    /// `reader` is the fully-composed read+decrypt stack (e.g.
+    /// [`PrefetchedSectorSource`](crate::sector::PrefetchedSectorSource)
+    /// wrapping
+    /// [`DecryptingSectorSource`](crate::sector::DecryptingSectorSource)).
+    /// `extents` is what the thread walks; it issues one
+    /// `read_sectors` per batch of `batch_sectors` sectors (aligned
+    /// to 3-sector AACS units when possible).
+    pub fn spawn<S: SectorSource + Send + 'static>(
+        mut reader: S,
+        extents: Vec<crate::disc::Extent>,
+        batch_sectors: u16,
+        halt: Option<Halt>,
+        ts: Option<super::ts::TsDemuxer>,
+        ps: Option<super::ps::PsDemuxer>,
+    ) -> (Self, Receiver<DemuxBatch>) {
+        let (tx, rx) = bounded::<DemuxBatch>(DEMUX_CHANNEL_DEPTH);
+        let mut ts = ts;
+        let mut ps = ps;
+
+        let handle = std::thread::Builder::new()
+            .name("freemkv-demux".into())
+            .spawn(move || {
+                let mut buf = vec![0u8; batch_sectors as usize * 2048];
+                let mut ext_idx = 0usize;
+                let mut offset: u32 = 0;
+                let prof = std::env::var_os("FREEMKV_PROFILE").is_some();
+                let mut prof_started = std::time::Instant::now();
+                let mut prof_last_dump = prof_started;
+                let mut prof_read_ns: u128 = 0;
+                let mut prof_feed_ns: u128 = 0;
+                let mut prof_send_ns: u128 = 0;
+                let mut prof_bytes: u64 = 0;
+                while ext_idx < extents.len() {
+                    if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) {
+                        return;
+                    }
+                    let ext = &extents[ext_idx];
+                    let remaining = ext.sector_count.saturating_sub(offset);
+                    if remaining == 0 {
+                        ext_idx += 1;
+                        offset = 0;
+                        continue;
+                    }
+                    let mut sectors = remaining.min(batch_sectors as u32) as u16;
+                    if sectors >= 3 {
+                        sectors -= sectors % 3;
+                    }
+                    let bytes = sectors as usize * 2048;
+                    if buf.len() < bytes {
+                        buf.resize(bytes, 0);
+                    }
+                    let lba = ext.start_lba + offset;
+                    let t0 = if prof {
+                        Some(std::time::Instant::now())
+                    } else {
+                        None
+                    };
+                    let n = match reader.read_sectors(lba, sectors, &mut buf[..bytes], false) {
+                        Ok(n) => n,
+                        Err(e) => {
+                            let _ = tx.send(DemuxBatch::Err(e.into()));
+                            return;
+                        }
+                    };
+                    let t1 = if prof {
+                        Some(std::time::Instant::now())
+                    } else {
+                        None
+                    };
+                    offset += sectors as u32;
+
+                    // Demux this batch immediately so the channel
+                    // carries already-parsed PesPackets, not raw
+                    // sector bytes.
+                    if let Some(ref mut d) = ts {
+                        let pkts = d.feed(&buf[..n]);
+                        let t2 = if prof {
+                            Some(std::time::Instant::now())
+                        } else {
+                            None
+                        };
+                        if !pkts.is_empty() && tx.send(DemuxBatch::Ts(pkts)).is_err() {
+                            return; // consumer dropped
+                        }
+                        let t3 = if prof {
+                            Some(std::time::Instant::now())
+                        } else {
+                            None
+                        };
+                        if prof {
+                            prof_read_ns += t1.unwrap().duration_since(t0.unwrap()).as_nanos();
+                            prof_feed_ns += t2.unwrap().duration_since(t1.unwrap()).as_nanos();
+                            prof_send_ns += t3.unwrap().duration_since(t2.unwrap()).as_nanos();
+                            prof_bytes += n as u64;
+                            let now = t3.unwrap();
+                            if now.duration_since(prof_last_dump)
+                                >= std::time::Duration::from_secs(5)
+                            {
+                                let el = now.duration_since(prof_started).as_millis().max(1);
+                                let mbps = prof_bytes as u128 * 1000 / 1_000_000 / el;
+                                eprintln!(
+                                    "[demux] elapsed={}ms in={}MB/s read={}% feed={}% send={}%",
+                                    el,
+                                    mbps,
+                                    prof_read_ns / 10_000 / el,
+                                    prof_feed_ns / 10_000 / el,
+                                    prof_send_ns / 10_000 / el,
+                                );
+                                prof_last_dump = now;
+                                prof_started = now;
+                                prof_read_ns = 0;
+                                prof_feed_ns = 0;
+                                prof_send_ns = 0;
+                                prof_bytes = 0;
+                            }
+                        }
+                    } else if let Some(ref mut d) = ps {
+                        let pkts = d.feed(&buf[..n]);
+                        if !pkts.is_empty() && tx.send(DemuxBatch::Ps(pkts)).is_err() {
+                            return;
+                        }
+                    }
+                }
+                // EOF — emit any flushed packets too.
+                if let Some(ref mut d) = ts {
+                    let tail = d.flush();
+                    if !tail.is_empty() {
+                        let _ = tx.send(DemuxBatch::Ts(tail));
+                    }
+                } else if let Some(ref mut d) = ps {
+                    let tail = d.flush();
+                    if !tail.is_empty() {
+                        let _ = tx.send(DemuxBatch::Ps(tail));
+                    }
+                }
+                // Sender drops here -> consumer sees RecvError → EOF.
+            })
+            .expect("freemkv-demux thread spawn failed");
+
+        (
+            Self {
+                handle: Some(handle),
+                producer_shell: None,
+            },
+            rx,
+        )
+    }
+
+    /// Zero-copy variant. Instead of taking a `SectorSource` and
+    /// memcpy-ing through its `read_sectors` API, this constructor
+    /// consumes the prefetch channels directly: filled buffers come
+    /// in via `prefetch_rx`, the demux thread feeds them, then
+    /// returns them to `recycle_tx` for the producer to re-fill.
+    /// Eliminates the 16 MiB memcpy per batch that the SectorSource
+    /// adapter incurred (and, with the producer-side recycling pool,
+    /// also eliminates the per-batch heap alloc / cross-thread free
+    /// that was costing 40 %+ of demux-thread time before).
+    ///
+    /// `producer_shell` is an opaque handle whose only purpose is to
+    /// outlive the demux thread and join the upstream producer when
+    /// dropped. Both
+    /// [`crate::sector::PrefetchedSectorSource::into_channels`] and
+    /// [`crate::io::byte_prefetcher::BytePrefetcher::into_channels`]
+    /// hand back a shell that fits — pass either.
+    pub fn spawn_zero_copy<S: Send + 'static>(
+        prefetch_rx: Receiver<std::io::Result<Vec<u8>>>,
+        recycle_tx: Sender<Vec<u8>>,
+        producer_shell: S,
+        halt: Option<Halt>,
+        ts: Option<super::ts::TsDemuxer>,
+        ps: Option<super::ps::PsDemuxer>,
+    ) -> (Self, Receiver<DemuxBatch>) {
+        let (tx, rx) = bounded::<DemuxBatch>(DEMUX_CHANNEL_DEPTH);
+        let mut ts = ts;
+        let mut ps = ps;
+
+        let handle = std::thread::Builder::new()
+            .name("freemkv-demux".into())
+            .spawn(move || {
+                let prof = std::env::var_os("FREEMKV_PROFILE").is_some();
+                let mut prof_started = std::time::Instant::now();
+                let mut prof_last_dump = prof_started;
+                let mut prof_read_ns: u128 = 0;
+                let mut prof_feed_ns: u128 = 0;
+                let mut prof_bytes: u64 = 0;
+                loop {
+                    if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) {
+                        return;
+                    }
+                    let t0 = if prof {
+                        Some(std::time::Instant::now())
+                    } else {
+                        None
+                    };
+                    let buf = match prefetch_rx.recv() {
+                        Ok(Ok(b)) => b,
+                        Ok(Err(e)) => {
+                            let _ = tx.send(DemuxBatch::Err(e));
+                            return;
+                        }
+                        Err(_) => break, // producer done → EOF
+                    };
+                    let t1 = if prof {
+                        Some(std::time::Instant::now())
+                    } else {
+                        None
+                    };
+                    let n = buf.len();
+                    if let Some(ref mut d) = ts {
+                        let pkts = d.feed(&buf);
+                        let t2 = if prof {
+                            Some(std::time::Instant::now())
+                        } else {
+                            None
+                        };
+                        // Recycle the buffer back to the producer
+                        // before pushing the demuxed packets. If the
+                        // recycle channel is closed the producer has
+                        // exited; we drop the buffer and continue.
+                        let _ = recycle_tx.send(buf);
+                        if !pkts.is_empty() && tx.send(DemuxBatch::Ts(pkts)).is_err() {
+                            return;
+                        }
+                        if prof {
+                            prof_read_ns += t1.unwrap().duration_since(t0.unwrap()).as_nanos();
+                            prof_feed_ns += t2.unwrap().duration_since(t1.unwrap()).as_nanos();
+                            prof_bytes += n as u64;
+                            let now = std::time::Instant::now();
+                            if now.duration_since(prof_last_dump)
+                                >= std::time::Duration::from_secs(5)
+                            {
+                                let el = now.duration_since(prof_started).as_millis().max(1);
+                                let mbps = prof_bytes as u128 * 1000 / 1_000_000 / el;
+                                eprintln!(
+                                    "[demux] elapsed={}ms in={}MB/s read={}% feed={}%",
+                                    el,
+                                    mbps,
+                                    prof_read_ns / 10_000 / el,
+                                    prof_feed_ns / 10_000 / el,
+                                );
+                                prof_last_dump = now;
+                                prof_started = now;
+                                prof_read_ns = 0;
+                                prof_feed_ns = 0;
+                                prof_bytes = 0;
+                            }
+                        }
+                    } else if let Some(ref mut d) = ps {
+                        let pkts = d.feed(&buf);
+                        let _ = recycle_tx.send(buf);
+                        if !pkts.is_empty() && tx.send(DemuxBatch::Ps(pkts)).is_err() {
+                            return;
+                        }
+                    } else {
+                        let _ = recycle_tx.send(buf);
+                    }
+                }
+                // Flush tail packets at EOF.
+                if let Some(ref mut d) = ts {
+                    let tail = d.flush();
+                    if !tail.is_empty() {
+                        let _ = tx.send(DemuxBatch::Ts(tail));
+                    }
+                } else if let Some(ref mut d) = ps {
+                    let tail = d.flush();
+                    if !tail.is_empty() {
+                        let _ = tx.send(DemuxBatch::Ps(tail));
+                    }
+                }
+            })
+            .expect("freemkv-demux thread spawn failed");
+
+        (
+            Self {
+                handle: Some(handle),
+                producer_shell: Some(Box::new(producer_shell)),
+            },
+            rx,
+        )
+    }
+}
+
+impl Drop for DemuxThread {
+    fn drop(&mut self) {
+        if let Some(h) = self.handle.take() {
+            let _ = h.join();
+        }
+    }
+}

src/mux/disc.rs

@@ -1,7 +1,7 @@
 //! DiscStream — read any disc (physical drive or ISO file) → PES frames.
 //!
 //! One stream type for all disc sources. The source is a SectorSource —
-//! Drive (hardware) or IsoSectorReader (file). DiscStream doesn't care.
+//! Drive (hardware) or FileSectorSource (file). DiscStream doesn't care.
 //!
 //! Read-only. For disc→ISO (raw sector copy), use `Disc::copy()`.
 
@@ -149,12 +149,28 @@ pub struct DiscStream {
     // a percent without a separate API call.
     bytes_total_extents: u64,
 
-    // PES output
+    // PES output. `ts_demuxer` and `ps_demuxer` are `None` when the
+    // stream is in pipeline mode — the demux state lives inside a
+    // [`super::demux_thread::DemuxThread`] and PesPackets arrive
+    // through `demux_rx` already parsed.
     ts_demuxer: Option<super::ts::TsDemuxer>,
     ps_demuxer: Option<super::ps::PsDemuxer>,
     parsers: Vec<(u16, Box<dyn super::codec::CodecParser>)>,
     pending_frames: std::collections::VecDeque<crate::pes::PesFrame>,
     pid_to_track: Vec<(u16, usize)>,
+
+    // Pipeline mode: when `Some`, the read+decrypt+demux pipeline
+    // runs on a dedicated thread; this stream's `read()` just pulls
+    // PesPacket batches from `demux_rx` and runs codec parse on the
+    // caller thread. See [`super::demux_thread`].
+    //
+    // `demux_thread` is kept solely so that `Drop` joins the worker
+    // before this stream is dropped — direct reads happen through
+    // `demux_rx`. The `allow(dead_code)` keeps the optimizer happy
+    // since the field is only used at drop time.
+    #[allow(dead_code)]
+    demux_thread: Option<super::demux_thread::DemuxThread>,
+    demux_rx: Option<crossbeam_channel::Receiver<super::demux_thread::DemuxBatch>>,
 }
 
 impl DiscStream {
@@ -235,6 +251,118 @@ impl DiscStream {
             parsers,
             pending_frames: std::collections::VecDeque::new(),
             pid_to_track,
+            demux_thread: None,
+            demux_rx: None,
+        }
+    }
+
+    /// Pipeline-mode constructor. Moves the read+decrypt+demux work
+    /// onto a [`super::demux_thread::DemuxThread`] so the caller's
+    /// `read()` thread only does codec parse + frame emission.
+    ///
+    /// `reader` is a [`crate::sector::PrefetchedSectorSource`] — the
+    /// prefetched producer thread already runs read+decrypt on its
+    /// own thread; this constructor peels off its channels for the
+    /// demux thread to consume in zero-copy mode (no buffer memcpy
+    /// across thread boundary, recycled-pool of two buffers, no
+    /// allocator activity in the hot loop).
+    ///
+    /// Pipeline mode is the preferred wiring for ISO file mux on a
+    /// multi-core host; it gives a ~2× consumer throughput in the
+    /// `null://` benchmark vs the single-thread inline path.
+    pub fn new_pipeline(
+        reader: crate::sector::PrefetchedSectorSource,
+        title: DiscTitle,
+        decrypt_keys: crate::decrypt::DecryptKeys,
+        batch_sectors: u16,
+        content_format: crate::disc::ContentFormat,
+        halt: Option<Halt>,
+    ) -> Self {
+        let extents = title.extents.clone();
+        let bytes_total_extents: u64 = extents.iter().map(|e| e.sector_count as u64 * 2048).sum();
+
+        let mut pids = Vec::new();
+        let mut parsers = Vec::new();
+        let mut pid_to_track = Vec::new();
+        for (idx, s) in title.streams.iter().enumerate() {
+            let (pid, codec) = match s {
+                crate::disc::Stream::Video(v) => (v.pid, v.codec),
+                crate::disc::Stream::Audio(a) => (a.pid, a.codec),
+                crate::disc::Stream::Subtitle(s) => (s.pid, s.codec),
+            };
+            pids.push(pid);
+            pid_to_track.push((pid, idx));
+            parsers.push((pid, super::codec::parser_for_codec(codec, None)));
+        }
+
+        let (ts, ps) = match content_format {
+            crate::disc::ContentFormat::MpegPs => (None, Some(super::ps::PsDemuxer::new())),
+            crate::disc::ContentFormat::BdTs => {
+                let ts_pids: Vec<u16> = pids.clone();
+                if ts_pids.is_empty() {
+                    (None, None)
+                } else {
+                    (Some(super::ts::TsDemuxer::new(&ts_pids)), None)
+                }
+            }
+        };
+
+        let (prefetch_rx, recycle_tx, shell) = reader.into_channels();
+        let (handle, rx) = super::demux_thread::DemuxThread::spawn_zero_copy(
+            prefetch_rx,
+            recycle_tx,
+            shell,
+            halt.clone(),
+            ts,
+            ps,
+        );
+
+        // The DiscStream's own reader is a no-op pass-through — the
+        // real reader lives inside the demux thread. We need *some*
+        // Box<dyn SectorSource> to satisfy the field type; use a
+        // tiny stub. fill_extents won't be called in pipeline mode.
+        struct NullSource;
+        impl SectorSource for NullSource {
+            fn capacity_sectors(&self) -> u32 {
+                0
+            }
+            fn read_sectors(
+                &mut self,
+                _: u32,
+                _: u16,
+                _: &mut [u8],
+                _: bool,
+            ) -> crate::error::Result<usize> {
+                Ok(0)
+            }
+        }
+        let dummy: Box<dyn SectorSource> = Box::new(NullSource);
+
+        Self {
+            reader: DecryptingSectorSource::new(dummy, crate::decrypt::DecryptKeys::None),
+            title,
+            disc: None,
+            decrypt_keys,
+            extents,
+            current_extent: 0,
+            current_offset: 0,
+            read_buf: Vec::new(),
+            buf_valid: 0,
+            adaptive: AdaptiveBatch::new(batch_sectors),
+            errors: 0,
+            skip_errors: false,
+            halt,
+            event_fn: None,
+            eof: false,
+            bytes_read_total: 0,
+            bytes_total_extents,
+            ts_demuxer: None,
+            ps_demuxer: None,
+            parsers,
+            pending_frames: std::collections::VecDeque::new(),
+            pid_to_track,
+            demux_thread: Some(handle),
+            demux_rx: Some(rx),
         }
     }
 
@@ -403,6 +531,153 @@ impl DiscStream {
     }
 }
 
+/// Per-stage profiling state — populated only when `FREEMKV_PROFILE`
+/// is set. Dumps a percentage breakdown to stderr every
+/// [`PROFILE_INTERVAL`]. Zero overhead in normal runs (Option check
+/// is the only added cost).
+struct StageProf {
+    started: std::time::Instant,
+    last_dump: std::time::Instant,
+    fill_ns: u128,
+    feed_ns: u128,
+    consume_ns: u128,
+    bytes_in: u64,
+}
+
+const PROFILE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5);
+
+thread_local! {
+    static STAGE_PROF: std::cell::RefCell<Option<StageProf>> = const { std::cell::RefCell::new(None) };
+}
+
+fn prof_active() -> bool {
+    std::env::var_os("FREEMKV_PROFILE").is_some()
+}
+
+fn prof_tick(stage: &str, ns: u128, bytes: u64) {
+    STAGE_PROF.with(|cell| {
+        let mut slot = cell.borrow_mut();
+        if slot.is_none() {
+            if !prof_active() {
+                return;
+            }
+            let now = std::time::Instant::now();
+            *slot = Some(StageProf {
+                started: now,
+                last_dump: now,
+                fill_ns: 0,
+                feed_ns: 0,
+                consume_ns: 0,
+                bytes_in: 0,
+            });
+        }
+        let p = slot.as_mut().unwrap();
+        match stage {
+            "fill" => p.fill_ns += ns,
+            "feed" => p.feed_ns += ns,
+            "consume" => p.consume_ns += ns,
+            _ => {}
+        }
+        p.bytes_in += bytes;
+        let now = std::time::Instant::now();
+        if now.duration_since(p.last_dump) < PROFILE_INTERVAL {
+            return;
+        }
+        let elapsed_ms = now.duration_since(p.started).as_millis().max(1);
+        let fill_pct = p.fill_ns / 10_000 / elapsed_ms;
+        let feed_pct = p.feed_ns / 10_000 / elapsed_ms;
+        let consume_pct = p.consume_ns / 10_000 / elapsed_ms;
+        let mbps = p.bytes_in as u128 * 1000 / 1_000_000 / elapsed_ms;
+        eprintln!(
+            "[profile] elapsed={}ms in={}MB/s fill={}% feed={}% consume={}%",
+            elapsed_ms, mbps, fill_pct, feed_pct, consume_pct,
+        );
+        p.last_dump = now;
+    });
+}
+
+impl DiscStream {
+    /// Pipeline-mode `read()` helper: pull one PesPacket batch from
+    /// the demux thread, run codec parse on each PES, enqueue the
+    /// resulting PesFrames, return the first one.
+    fn read_pipeline(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
+        use super::demux_thread::DemuxBatch;
+        let rx = self.demux_rx.as_ref().expect("read_pipeline without rx");
+        match rx.recv() {
+            Ok(DemuxBatch::Ts(packets)) => {
+                let skip_parse = std::env::var_os("FREEMKV_SKIP_PARSE").is_some();
+                for pes in packets {
+                    if let Some((_, track)) = self
+                        .pid_to_track
+                        .iter()
+                        .find(|(pid, _)| *pid == pes.pid)
+                        .copied()
+                    {
+                        if skip_parse {
+                            self.pending_frames.push_back(crate::pes::PesFrame {
+                                track,
+                                pts: pes.pts.map(super::codec::pts_to_ns).unwrap_or(0),
+                                keyframe: false,
+                                data: pes.data,
+                            });
+                        } else if let Some((_, parser)) =
+                            self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid)
+                        {
+                            for frame in parser.parse(&pes) {
+                                self.pending_frames.push_back(
+                                    crate::pes::PesFrame::from_codec_frame(track, frame),
+                                );
+                            }
+                        }
+                    }
+                }
+                Ok(self.pending_frames.pop_front())
+            }
+            Ok(DemuxBatch::Ps(packets)) => {
+                for ps in packets {
+                    let track = match ps.stream_id {
+                        0xE0..=0xEF => 0,
+                        0xC0..=0xDF => 1,
+                        0xBD => ps
+                            .sub_stream_id
+                            .map(|s| (s & 0x1F) as usize + 1)
+                            .unwrap_or(1),
+                        _ => continue,
+                    };
+                    if track >= self.title.streams.len() {
+                        continue;
+                    }
+                    let pid = self
+                        .pid_to_track
+                        .iter()
+                        .find(|(_, idx)| *idx == track)
+                        .map(|(p, _)| *p)
+                        .unwrap_or(0);
+                    let pes = super::ts::PesPacket {
+                        pid,
+                        pts: ps.pts.map(|p| p as i64),
+                        dts: ps.dts.map(|d| d as i64),
+                        data: ps.data,
+                    };
+                    if let Some((_, parser)) = self.parsers.iter_mut().find(|(p, _)| *p == pid) {
+                        for frame in parser.parse(&pes) {
+                            self.pending_frames
+                                .push_back(crate::pes::PesFrame::from_codec_frame(track, frame));
+                        }
+                    }
+                }
+                Ok(self.pending_frames.pop_front())
+            }
+            Ok(DemuxBatch::Err(e)) => Err(e),
+            Err(_) => {
+                // Channel closed → demux thread finished. EOF.
+                self.eof = true;
+                Ok(None)
+            }
+        }
+    }
+}
+
 impl crate::pes::Stream for DiscStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
         if let Some(frame) = self.pending_frames.pop_front() {
@@ -413,7 +688,21 @@ impl crate::pes::Stream for DiscStream {
             return Ok(None);
         }
 
+        // Pipeline mode: read+decrypt+demux all happen on the demux
+        // thread. Pull a batch of PesPackets and run codec parse on
+        // this thread; loop until we have at least one frame OR EOF.
+        if self.demux_rx.is_some() {
+            loop {
+                match self.read_pipeline()? {
+                    Some(f) => return Ok(Some(f)),
+                    None if self.eof => return Ok(None),
+                    None => continue,
+                }
+            }
+        }
+
         loop {
+            let t0 = std::time::Instant::now();
             if !self.fill_extents()? {
                 self.eof = true;
                 // Flush demuxer — last PES packet may still be in the assembler
@@ -475,6 +764,8 @@ impl crate::pes::Stream for DiscStream {
             }
 
             let bytes = self.buf_valid;
+            let t1 = std::time::Instant::now();
+            prof_tick("fill", t1.duration_since(t0).as_nanos(), bytes as u64);
             // Plaintext: the wrapped reader (DecryptingSectorSource)
             // applied AACS / CSS in-place during fill_extents'
             // read_sectors call. The pre-0.18 inline decrypt step
@@ -482,21 +773,41 @@ impl crate::pes::Stream for DiscStream {
 
             if let Some(ref mut demuxer) = self.ts_demuxer {
                 let packets = demuxer.feed(&self.read_buf[..bytes]);
-                for pes in &packets {
+                let t2 = std::time::Instant::now();
-                    if let Some((_, track)) =
+                prof_tick("feed", t2.duration_since(t1).as_nanos(), 0);
-                        self.pid_to_track.iter().find(|(pid, _)| *pid == pes.pid)
+                let skip_parse = std::env::var_os("FREEMKV_SKIP_PARSE").is_some();
+                for pes in packets {
+                    if let Some((_, track)) = self
+                        .pid_to_track
+                        .iter()
+                        .find(|(pid, _)| *pid == pes.pid)
+                        .copied()
                     {
-                        if let Some((_, parser)) =
+                        if skip_parse {
+                            // Profiling escape hatch — bypass the codec
+                            // parser and pass the raw PES bytes straight
+                            // through as a single PesFrame. Lets us
+                            // attribute consumer-thread time to
+                            // "demux + framing" vs "codec parse".
+                            self.pending_frames.push_back(crate::pes::PesFrame {
+                                track,
+                                pts: pes.pts.map(super::codec::pts_to_ns).unwrap_or(0),
+                                keyframe: false,
+                                data: pes.data,
+                            });
+                        } else if let Some((_, parser)) =
                             self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid)
                         {
-                            for frame in parser.parse(pes) {
+                            for frame in parser.parse(&pes) {
                                 self.pending_frames.push_back(
-                                    crate::pes::PesFrame::from_codec_frame(*track, frame),
+                                    crate::pes::PesFrame::from_codec_frame(track, frame),
                                 );
                             }
                         }
                     }
                 }
+                let t3 = std::time::Instant::now();
+                prof_tick("consume", t3.duration_since(t2).as_nanos(), 0);
             } else if let Some(ref mut demuxer) = self.ps_demuxer {
                 let packets = demuxer.feed(&self.read_buf[..bytes]);
                 for ps in &packets {
@@ -570,6 +881,13 @@ impl crate::pes::Stream for DiscStream {
     }
 
     fn headers_ready(&self) -> bool {
+        // FREEMKV_SKIP_PARSE bypasses codec parsers entirely for
+        // bottleneck profiling, so codec_private is never populated.
+        // Pretend headers are ready immediately in that mode so the
+        // CLI loop doesn't hang waiting for them.
+        if std::env::var_os("FREEMKV_SKIP_PARSE").is_some() {
+            return true;
+        }
         for (idx, s) in self.title.streams.iter().enumerate() {
             if let crate::disc::Stream::Video(v) = s {
                 if !v.secondary && self.codec_private(idx).is_none() {

src/mux/iso.rs

@@ -1,101 +0,0 @@
-//! ISO sector reader — file-backed SectorSource for Blu-ray ISO images.
-//!
-//! An ISO is a flat image of 2048-byte sectors. Sector N starts at byte offset N * 2048.
-//! Used by DiscStream::open_iso() and Disc::scan_image().
-
-use crate::error::{Error, Result};
-use crate::sector::SectorSource;
-use std::fs::File;
-use std::io::{Read, Seek, SeekFrom};
-use std::path::Path;
-
-const SECTOR_SIZE: u64 = 2048;
-
-/// File-backed sector reader for ISO images.
-pub struct IsoSectorReader {
-    file: File,
-    capacity: u32,
-}
-
-impl IsoSectorReader {
-    pub fn open(path: &str) -> std::io::Result<Self> {
-        let file = File::open(Path::new(path))?;
-        let size = file.metadata()?.len();
-        let sectors = size / SECTOR_SIZE;
-        if sectors > u32::MAX as u64 {
-            return Err(crate::error::Error::IsoTooLarge {
-                path: path.to_string(),
-            }
-            .into());
-        }
-        let capacity = sectors as u32;
-        Ok(Self { file, capacity })
-    }
-
-    pub fn capacity_sectors(&self) -> u32 {
-        self.capacity
-    }
-}
-
-impl SectorSource for IsoSectorReader {
-    fn read_sectors(
-        &mut self,
-        lba: u32,
-        count: u16,
-        buf: &mut [u8],
-        _recovery: bool,
-    ) -> Result<usize> {
-        let bytes = count as usize * SECTOR_SIZE as usize;
-        self.file
-            .seek(SeekFrom::Start(lba as u64 * SECTOR_SIZE))
-            .map_err(|e| Error::IoError { source: e })?;
-        self.file
-            .read_exact(&mut buf[..bytes])
-            .map_err(|e| Error::IoError { source: e })?;
-        Ok(bytes)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn iso_reader_read_sectors() {
-        let mut data = vec![0u8; 4 * SECTOR_SIZE as usize];
-        for i in 0..4u8 {
-            let offset = i as usize * SECTOR_SIZE as usize;
-            data[offset] = i + 1;
-            data[offset + 2047] = i + 100;
-        }
-
-        let dir = std::env::temp_dir().join("freemkv_test_iso_read");
-        std::fs::write(&dir, &data).unwrap();
-
-        let mut reader = IsoSectorReader::open(dir.to_str().unwrap()).unwrap();
-        assert_eq!(reader.capacity_sectors(), 4);
-
-        let mut buf = [0u8; 2048];
-        reader.read_sectors(0, 1, &mut buf, true).unwrap();
-        assert_eq!(buf[0], 1);
-        assert_eq!(buf[2047], 100);
-
-        reader.read_sectors(2, 1, &mut buf, true).unwrap();
-        assert_eq!(buf[0], 3);
-        assert_eq!(buf[2047], 102);
-
-        std::fs::remove_file(&dir).ok();
-    }
-
-    #[test]
-    fn iso_reader_capacity() {
-        let data = vec![0u8; 10 * SECTOR_SIZE as usize];
-        let dir = std::env::temp_dir().join("freemkv_test_iso_cap");
-        std::fs::write(&dir, &data).unwrap();
-
-        let reader = IsoSectorReader::open(dir.to_str().unwrap()).unwrap();
-        assert_eq!(reader.capacity_sectors(), 10);
-
-        std::fs::remove_file(&dir).ok();
-    }
-}

src/mux/m2ts.rs

@@ -1,57 +1,21 @@
-//! M2tsStream — BD transport stream with embedded metadata header.
+//! M2tsStream — BD transport stream write sink.
 //!
-//! Write: prepends FMKV metadata header, then muxes PES frames into BD-TS.
+//! Write: prepends FMKV metadata header, then muxes PES frames into
-//! Read: extracts metadata header (or scans PMT), then demuxes BD-TS into PES frames.
+//! BD-TS. The read direction lives on the pipeline highway —
+//! `m2ts://` URLs route through
+//! [`super::resolve::input`] → `build_m2ts_pipeline` →
+//! [`super::pipelined_stream::PipelinedPesStream`], so this type is
+//! write-only.
 
-use super::{meta, ts};
+use super::meta;
 use crate::disc::{DiscTitle, Stream as DiscStream};
-use std::io::{self, Read, Write};
+use std::io::{self, Write};
 
-type PesSetup = (
+/// BD transport stream write sink with embedded FMKV metadata
-    Vec<u16>,
+/// header.
-    Vec<(u16, Box<dyn super::codec::CodecParser>)>,
-    Vec<(u16, usize)>,
-);
-
-/// Size of initial scan buffer for PMT/stream detection.
-const SCAN_SIZE: usize = 1024 * 1024;
-
-enum Mode {
-    Write {
-        muxer: super::tsmux::TsMuxer<Box<dyn Write + Send>>,
-    },
-    Read {
-        reader: Box<dyn Read + Send>,
-    },
-}
-
-/// Read as many bytes as possible into buf (multiple read calls if needed).
-/// Bounded by buf.len() — caller controls max bytes read.
-fn read_fill(r: &mut impl Read, buf: &mut [u8]) -> io::Result<usize> {
-    let mut total = 0;
-    while total < buf.len() {
-        match r.read(&mut buf[total..]) {
-            Ok(0) => break,
-            Ok(n) => total += n,
-            Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
-            Err(e) => return Err(e),
-        }
-    }
-    Ok(total)
-}
-
-/// BD transport stream with embedded metadata.
 pub struct M2tsStream {
     disc_title: DiscTitle,
-    mode: Mode,
+    muxer: super::tsmux::TsMuxer<Box<dyn Write + Send>>,
-    // PES support
-    demuxer: Option<ts::TsDemuxer>,
-    parsers: Vec<(u16, Box<dyn super::codec::CodecParser>)>,
-    pending_frames: std::collections::VecDeque<crate::pes::PesFrame>,
-    pid_to_track: Vec<(u16, usize)>,
-    pes_eof: bool,
-    /// Codec private data per stream (from FMKV header).
-    stored_codec_privates: Vec<Option<Vec<u8>>>,
 }
 
 impl M2tsStream {
@@ -81,208 +45,38 @@ impl M2tsStream {
         }
         Ok(Self {
             disc_title: title.clone(),
-            mode: Mode::Write { muxer },
+            muxer,
-            demuxer: None,
-            parsers: Vec::new(),
-            pending_frames: std::collections::VecDeque::new(),
-            pid_to_track: Vec::new(),
-            pes_eof: false,
-            stored_codec_privates: Vec::new(),
-        })
-    }
-
-    fn setup_pes(streams: &[DiscStream]) -> PesSetup {
-        let mut pids = Vec::new();
-        let mut parsers: Vec<(u16, Box<dyn super::codec::CodecParser>)> = Vec::new();
-        let mut pid_to_track = Vec::new();
-        for (i, s) in streams.iter().enumerate() {
-            let (pid, codec) = match s {
-                DiscStream::Video(v) => (v.pid, v.codec),
-                DiscStream::Audio(a) => (a.pid, a.codec),
-                DiscStream::Subtitle(s) => (s.pid, s.codec),
-            };
-            pids.push(pid);
-            pid_to_track.push((pid, i));
-            parsers.push((pid, super::codec::parser_for_codec(codec, None)));
-        }
-        (pids, parsers, pid_to_track)
-    }
-
-    /// Open an M2TS stream for reading. Takes any Read source — file, pipe, socket.
-    ///
-    /// Tries FMKV metadata header first. Falls back to PMT scan of first 1 MB.
-    pub fn open(mut reader: impl Read + Send + 'static) -> io::Result<Self> {
-        // Read first chunk — enough for FMKV header or PMT scan
-        let mut head = vec![0u8; SCAN_SIZE];
-        let head_len = read_fill(&mut reader, &mut head)?;
-        head.truncate(head_len);
-
-        // Try FMKV metadata header from the buffered head
-        let mut cursor = io::Cursor::new(&head);
-        if let Ok(Some(m)) = meta::read_header(&mut cursor) {
-            let header_end = cursor.position() as usize;
-            let title = m.to_title();
-            let (pids, parsers, pid_to_track) = Self::setup_pes(&title.streams);
-            // Chain: remaining head bytes + rest of reader
-            let remaining_head = &head[header_end..];
-            let chain: Box<dyn Read + Send> =
-                Box::new(io::Cursor::new(remaining_head.to_vec()).chain(reader));
-            return Ok(Self {
-                disc_title: title.clone(),
-                mode: Mode::Read { reader: chain },
-                demuxer: if pids.is_empty() {
-                    None
-                } else {
-                    Some(ts::TsDemuxer::new(&pids))
-                },
-                parsers,
-                pending_frames: std::collections::VecDeque::new(),
-                pid_to_track,
-                pes_eof: false,
-                stored_codec_privates: title.codec_privates,
-            });
-        }
-
-        // No FMKV header — scan head for PMT
-        let streams = ts::scan_streams(&head)
-            .ok_or_else(|| -> io::Error { crate::error::Error::NoStreams.into() })?;
-
-        let (pids, parsers, pid_to_track) = Self::setup_pes(&streams);
-
-        // Chain: full head (it's all TS data) + rest of reader
-        let chain: Box<dyn Read + Send> = Box::new(io::Cursor::new(head).chain(reader));
-
-        Ok(Self {
-            disc_title: DiscTitle {
-                duration_secs: 0.0, // unknown without seeking
-                streams,
-                ..DiscTitle::empty()
-            },
-            mode: Mode::Read { reader: chain },
-            demuxer: if pids.is_empty() {
-                None
-            } else {
-                Some(ts::TsDemuxer::new(&pids))
-            },
-            parsers,
-            pending_frames: std::collections::VecDeque::new(),
-            pid_to_track,
-            pes_eof: false,
-            stored_codec_privates: Vec::new(),
         })
     }
 }
 
 impl crate::pes::Stream for M2tsStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
-        if let Some(frame) = self.pending_frames.pop_front() {
+        // Write-only sink. The m2ts:// read direction is served by
-            return Ok(Some(frame));
+        // `super::resolve::build_m2ts_pipeline` →
-        }
+        // `PipelinedPesStream`; routing through this type for reads
-        if self.pes_eof {
+        // was removed when the highway became the only ingress.
-            return Ok(None);
+        Err(crate::error::Error::StreamWriteOnly.into())
-        }
-
-        loop {
-            let reader = match &mut self.mode {
-                Mode::Read { reader } => reader,
-                _ => return Err(crate::error::Error::StreamWriteOnly.into()),
-            };
-            let mut buf = vec![0u8; 192 * 1024];
-            let n = reader.read(&mut buf)?;
-            if n == 0 {
-                self.pes_eof = true;
-                // Flush demuxer — last PES packet may still be in the assembler
-                if let Some(ref mut demuxer) = self.demuxer {
-                    for pes in &demuxer.flush() {
-                        if let Some((_, track)) =
-                            self.pid_to_track.iter().find(|(pid, _)| *pid == pes.pid)
-                        {
-                            if let Some((_, parser)) =
-                                self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid)
-                            {
-                                for frame in parser.parse(pes) {
-                                    self.pending_frames.push_back(
-                                        crate::pes::PesFrame::from_codec_frame(*track, frame),
-                                    );
-                                }
-                            }
-                        }
-                    }
-                }
-                return Ok(self.pending_frames.pop_front());
-            }
-
-            if let Some(ref mut demuxer) = self.demuxer {
-                let packets = demuxer.feed(&buf[..n]);
-                for pes in &packets {
-                    if let Some((_, track)) =
-                        self.pid_to_track.iter().find(|(pid, _)| *pid == pes.pid)
-                    {
-                        if let Some((_, parser)) =
-                            self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid)
-                        {
-                            for frame in parser.parse(pes) {
-                                self.pending_frames.push_back(
-                                    crate::pes::PesFrame::from_codec_frame(*track, frame),
-                                );
-                            }
-                        }
-                    }
-                }
-            }
-
-            if let Some(frame) = self.pending_frames.pop_front() {
-                return Ok(Some(frame));
-            }
-        }
     }
 
     fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
-        match &mut self.mode {
+        self.muxer
-            Mode::Write { muxer } => {
+            .write_frame(frame.track, frame.pts, frame.keyframe, &frame.data)
-                muxer.write_frame(frame.track, frame.pts, frame.keyframe, &frame.data)
-            }
-            Mode::Read { .. } => Err(crate::error::Error::StreamReadOnly.into()),
-        }
     }
 
     fn finish(&mut self) -> io::Result<()> {
-        match &mut self.mode {
+        self.muxer.finish()
-            Mode::Write { muxer } => muxer.finish(),
-            Mode::Read { .. } => Ok(()),
-        }
     }
 
     fn info(&self) -> &crate::disc::DiscTitle {
         &self.disc_title
     }
 
-    fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
+    fn codec_private(&self, _track: usize) -> Option<Vec<u8>> {
-        // First check stored codec_privates from FMKV header
+        // Write side doesn't have parsers; codec_private flows in
-        if let Some(Some(cp)) = self.stored_codec_privates.get(track) {
+        // via the title metadata at `create` time and gets baked
-            return Some(cp.clone());
+        // into the FMKV header. Nothing to surface back here.
-        }
+        None
-        // Fall back to parser-extracted codec_private
-        let pid = self
-            .pid_to_track
-            .iter()
-            .find(|(_, idx)| *idx == track)
-            .map(|(pid, _)| *pid)?;
-        self.parsers
-            .iter()
-            .find(|(p, _)| *p == pid)
-            .and_then(|(_, parser)| parser.codec_private())
-    }
-
-    fn headers_ready(&self) -> bool {
-        for (idx, s) in self.disc_title.streams.iter().enumerate() {
-            if let crate::disc::Stream::Video(v) = s {
-                if !v.secondary && self.codec_private(idx).is_none() {
-                    return false;
-                }
-            }
-        }
-        true
     }
 }
 

src/mux/mod.rs

@@ -17,8 +17,9 @@
 
 // Public modules — types here are intentionally part of the consumable API.
 pub mod codec;
+pub mod demux_thread;
 pub mod disc;
-pub mod iso;
+pub mod pipelined_stream;
 pub mod resolve;
 
 // Internal modules — implementation details. Their *types* are re-exported
@@ -54,11 +55,12 @@ pub(crate) mod ts;
 pub(crate) mod tsmux;
 
 pub use disc::DiscStream;
-pub use iso::IsoSectorReader;
 pub use m2ts::M2tsStream;
 pub use mkvstream::MkvStream;
 pub use network::NetworkStream;
 pub use null::NullStream;
+pub use pipelined_stream::PipelinedPesStream;
+pub use resolve::build_iso_pipeline;
 pub use resolve::{InputOptions, StreamUrl, input, output, parse_url};
 pub use stdio::StdioStream;
 

src/mux/pipelined_stream.rs

@@ -0,0 +1,225 @@
+//! `PipelinedPesStream` — the read-side of the freemkv mux
+//! highway.
+//!
+//! Given a [`crate::mux::demux_thread::DemuxThread`] (which has the
+//! producer + demux workers already spawned), a set of codec
+//! parsers, and the title metadata, this struct implements
+//! [`crate::pes::Stream`] by running codec parse on the caller's
+//! thread and emitting `PesFrame`s one at a time.
+//!
+//! The pipeline runs three threads in parallel:
+//!
+//! ```text
+//! Thread A: read + decrypt   (PrefetchedSectorSource / BytePrefetcher)
+//! Thread B: M2TS demux       (DemuxThread)
+//! Thread C: codec parse      (this struct, on the caller's thread)
+//! ```
+//!
+//! Communication between A→B and B→C is via bounded channels with
+//! recycled buffer pools — no allocations or memcpys in the steady-
+//! state hot loop.
+//!
+//! This is the *only* read-side `Stream` impl in tree. Both ISO file
+//! mux ([`crate::mux::resolve`]) and BD-TS file mux ([`crate::mux::M2tsStream`])
+//! return a `PipelinedPesStream`; the differences are in how the
+//! producer thread (A) is configured — sector-aligned reads with
+//! AACS decrypt for ISO, raw byte reads for M2TS.
+
+use super::codec::CodecParser;
+use super::demux_thread::{DemuxBatch, DemuxThread};
+use super::ts::PesPacket;
+use crate::disc::DiscTitle;
+use crate::pes::{PesFrame, Stream};
+use crossbeam_channel::Receiver;
+use std::io;
+
+/// Stream impl that consumes pre-demuxed `PesPacket` batches from a
+/// [`DemuxThread`] and runs codec parse on the caller's thread.
+pub struct PipelinedPesStream {
+    title: DiscTitle,
+    parsers: Vec<(u16, Box<dyn CodecParser>)>,
+    pid_to_track: Vec<(u16, usize)>,
+
+    demux_rx: Receiver<DemuxBatch>,
+    /// Kept alive so dropping this stream joins the demux + producer
+    /// workers deterministically. Never poked directly after spawn.
+    #[allow(dead_code)]
+    demux_thread: DemuxThread,
+
+    pending_frames: std::collections::VecDeque<PesFrame>,
+    eof: bool,
+}
+
+impl PipelinedPesStream {
+    /// Wire up the stream. Caller has already spawned the
+    /// `DemuxThread` (which in turn owns the producer); we take the
+    /// receiver end + the join handle bundle so cleanup is bounded
+    /// on drop.
+    pub fn new(
+        demux_thread: DemuxThread,
+        demux_rx: Receiver<DemuxBatch>,
+        title: DiscTitle,
+        parsers: Vec<(u16, Box<dyn CodecParser>)>,
+        pid_to_track: Vec<(u16, usize)>,
+    ) -> Self {
+        Self {
+            title,
+            parsers,
+            pid_to_track,
+            demux_rx,
+            demux_thread,
+            pending_frames: std::collections::VecDeque::new(),
+            eof: false,
+        }
+    }
+
+    /// Pull one batch of `PesPacket`s from the demux thread, run
+    /// codec parse on each, enqueue resulting `PesFrame`s on
+    /// `pending_frames`. Returns Ok(true) on success, Ok(false) on
+    /// EOF (channel closed cleanly), Err on demuxer error.
+    fn pump_one_batch(&mut self) -> io::Result<bool> {
+        match self.demux_rx.recv() {
+            Ok(DemuxBatch::Ts(packets)) => {
+                self.consume_ts(packets);
+                Ok(true)
+            }
+            Ok(DemuxBatch::Ps(packets)) => {
+                self.consume_ps(packets);
+                Ok(true)
+            }
+            Ok(DemuxBatch::Err(e)) => Err(e),
+            Err(_) => Ok(false),
+        }
+    }
+
+    fn consume_ts(&mut self, packets: Vec<PesPacket>) {
+        let skip_parse = std::env::var_os("FREEMKV_SKIP_PARSE").is_some();
+        for pes in packets {
+            if let Some((_, track)) = self
+                .pid_to_track
+                .iter()
+                .find(|(pid, _)| *pid == pes.pid)
+                .copied()
+            {
+                if skip_parse {
+                    // Profiling escape hatch — bypass codec parser.
+                    self.pending_frames.push_back(PesFrame {
+                        track,
+                        pts: pes.pts.map(super::codec::pts_to_ns).unwrap_or(0),
+                        keyframe: false,
+                        data: pes.data,
+                    });
+                } else if let Some((_, parser)) =
+                    self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid)
+                {
+                    for frame in parser.parse(&pes) {
+                        self.pending_frames
+                            .push_back(PesFrame::from_codec_frame(track, frame));
+                    }
+                }
+            }
+        }
+    }
+
+    fn consume_ps(&mut self, packets: Vec<super::ps::PsPacket>) {
+        for ps in packets {
+            let track = match ps.stream_id {
+                0xE0..=0xEF => 0,
+                0xC0..=0xDF => 1,
+                0xBD => ps
+                    .sub_stream_id
+                    .map(|s| (s & 0x1F) as usize + 1)
+                    .unwrap_or(1),
+                _ => continue,
+            };
+            if track >= self.title.streams.len() {
+                continue;
+            }
+            let pid = self
+                .pid_to_track
+                .iter()
+                .find(|(_, idx)| *idx == track)
+                .map(|(p, _)| *p)
+                .unwrap_or(0);
+            let pes = PesPacket {
+                pid,
+                pts: ps.pts.map(|p| p as i64),
+                dts: ps.dts.map(|d| d as i64),
+                data: ps.data,
+            };
+            if let Some((_, parser)) = self.parsers.iter_mut().find(|(p, _)| *p == pid) {
+                for frame in parser.parse(&pes) {
+                    self.pending_frames
+                        .push_back(PesFrame::from_codec_frame(track, frame));
+                }
+            }
+        }
+    }
+}
+
+impl Stream for PipelinedPesStream {
+    fn read(&mut self) -> io::Result<Option<PesFrame>> {
+        if let Some(frame) = self.pending_frames.pop_front() {
+            return Ok(Some(frame));
+        }
+        if self.eof {
+            return Ok(None);
+        }
+        loop {
+            match self.pump_one_batch()? {
+                true => {
+                    if let Some(frame) = self.pending_frames.pop_front() {
+                        return Ok(Some(frame));
+                    }
+                    // Batch contained no trackable packets — pull again.
+                }
+                false => {
+                    self.eof = true;
+                    return Ok(self.pending_frames.pop_front());
+                }
+            }
+        }
+    }
+
+    fn write(&mut self, _: &PesFrame) -> io::Result<()> {
+        Err(crate::error::Error::StreamReadOnly.into())
+    }
+
+    fn finish(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+
+    fn info(&self) -> &DiscTitle {
+        &self.title
+    }
+
+    fn headers_ready(&self) -> bool {
+        // Match the previous DiscStream semantics: video tracks need
+        // codec_private before the consumer can write the container
+        // header. FREEMKV_SKIP_PARSE forces ready (no parser ever
+        // populates codec_private in that mode).
+        if std::env::var_os("FREEMKV_SKIP_PARSE").is_some() {
+            return true;
+        }
+        for (idx, s) in self.title.streams.iter().enumerate() {
+            if let crate::disc::Stream::Video(v) = s {
+                if !v.secondary && self.codec_private(idx).is_none() {
+                    return false;
+                }
+            }
+        }
+        true
+    }
+
+    fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
+        let pid = self
+            .pid_to_track
+            .iter()
+            .find(|(_, idx)| *idx == track)
+            .map(|(p, _)| *p)?;
+        self.parsers
+            .iter()
+            .find(|(p, _)| *p == pid)
+            .and_then(|(_, parser)| parser.codec_private())
+    }
+}

src/mux/resolve.rs

@@ -15,11 +15,13 @@
 //! Bare paths without a scheme are rejected.
 //! For disc→ISO (raw sector copy), use `Disc::copy()` instead.
 
-use super::disc::DiscStream;
 use super::network::NetworkStream;
 use super::null::NullStream;
+use super::pipelined_stream::PipelinedPesStream;
 use super::stdio::StdioStream;
 use super::{M2tsStream, MkvStream};
+use crate::disc::{ContentFormat, DiscTitle};
+use crate::sector::SectorSource;
 use std::io;
 use std::path::{Path, PathBuf};
 
@@ -185,7 +187,12 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
                 },
                 None => crate::disc::ScanOptions::default(),
             };
-            let mut reader = super::iso::IsoSectorReader::open(&path.to_string_lossy())?;
+            // FileSectorSource is the sole file-backed sector source.
+            // It carries the platform-tuned SEQUENTIAL fadvise hint
+            // (so the kernel readahead window widens) and the periodic
+            // DONTNEED page-cache eviction that bounds memory pressure
+            // when the mux output is being written to the same disk.
+            let mut reader = crate::io::file_sector_source::FileSectorSource::open(path)?;
             let capacity = reader.capacity_sectors();
             let disc = crate::disc::Disc::scan_image(&mut reader, capacity, &scan_opts)
                 .map_err(|e| -> io::Error { e.into() })?;
@@ -203,21 +210,39 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
             let title = disc.titles[idx].clone();
             let keys = disc.decrypt_keys();
             let format = disc.content_format;
-            // ISO file: use large batch size (16 MB) — sequential read from fast storage, no bad sectors.
+            // ISO file: 16 MiB batch — sequential read from fast
-            // Physical drives need small batches for adaptive error handling and retry logic.
+            // storage, no bad sectors. Measured optimum on the rip1
+            // testbed; bumping to 32 MiB regressed (more cache
+            // pressure, longer per-batch latency starves the consumer
+            // between iterations). Physical drives keep smaller
+            // batches for adaptive error handling.
             const ISO_MUX_BATCH_SECTORS: u16 = 8192;
-            let mut stream =
+
-                DiscStream::new(Box::new(reader), title, keys, ISO_MUX_BATCH_SECTORS, format);
+            // Pass `DecryptKeys::None` to the decrypt decorator when
-            if opts.raw {
+            // --raw is set — the read stack still flows through the
-                stream.set_raw();
+            // same producer+demux+parse pipeline, just without the
-            }
+            // AACS / CSS step. Single highway for all ISO reads.
+            let effective_keys = if opts.raw {
+                crate::decrypt::DecryptKeys::None
+            } else {
+                keys
+            };
+            let stream = build_iso_pipeline(
+                reader,
+                title,
+                effective_keys,
+                ISO_MUX_BATCH_SECTORS,
+                format,
+                None,
+            );
             Ok(Box::new(stream))
         }
         StreamUrl::M2ts { ref path } => {
             validate_file_path(path, "m2ts")?;
             let file = std::fs::File::open(path)?;
             let reader = std::io::BufReader::with_capacity(IO_BUF_SIZE, file);
-            Ok(Box::new(M2tsStream::open(reader)?))
+            let stream = build_m2ts_pipeline(reader)?;
+            Ok(Box::new(stream))
         }
         StreamUrl::Mkv { ref path } => {
             validate_file_path(path, "mkv")?;
@@ -281,3 +306,135 @@ pub fn output(
         }
     }
 }
+
+/// Demuxer-side state derived from a `DiscTitle`: the codec parser
+/// table (keyed by PID), the PID-to-track index map, and an initial
+/// `TsDemuxer` / `PsDemuxer` (whichever the content format calls
+/// for).
+type DemuxState = (
+    Vec<(u16, Box<dyn super::codec::CodecParser>)>,
+    Vec<(u16, usize)>,
+    Option<super::ts::TsDemuxer>,
+    Option<super::ps::PsDemuxer>,
+);
+
+/// Build the title's codec parser table + initial `TsDemuxer` /
+/// `PsDemuxer`. Used by both the ISO and M2TS pipeline builders.
+fn build_demux_state(title: &DiscTitle, format: ContentFormat) -> DemuxState {
+    let mut pids = Vec::new();
+    let mut parsers = Vec::new();
+    let mut pid_to_track = Vec::new();
+    for (idx, s) in title.streams.iter().enumerate() {
+        let (pid, codec) = match s {
+            crate::disc::Stream::Video(v) => (v.pid, v.codec),
+            crate::disc::Stream::Audio(a) => (a.pid, a.codec),
+            crate::disc::Stream::Subtitle(s) => (s.pid, s.codec),
+        };
+        pids.push(pid);
+        pid_to_track.push((pid, idx));
+        parsers.push((pid, super::codec::parser_for_codec(codec, None)));
+    }
+    let (ts, ps) = match format {
+        ContentFormat::MpegPs => (None, Some(super::ps::PsDemuxer::new())),
+        ContentFormat::BdTs => {
+            if pids.is_empty() {
+                (None, None)
+            } else {
+                (Some(super::ts::TsDemuxer::new(&pids)), None)
+            }
+        }
+    };
+    (parsers, pid_to_track, ts, ps)
+}
+
+/// Assemble the ISO mux pipeline (read+decrypt → demux → parse) for
+/// a `FileSectorSource`-backed reader. Returns the resulting
+/// `PipelinedPesStream`.
+pub fn build_iso_pipeline<S: SectorSource + Send + 'static>(
+    reader: S,
+    title: DiscTitle,
+    keys: crate::decrypt::DecryptKeys,
+    batch_sectors: u16,
+    format: ContentFormat,
+    halt: Option<crate::halt::Halt>,
+) -> PipelinedPesStream {
+    let extents = title.extents.clone();
+    let decrypting =
+        crate::sector::DecryptingSectorSource::new(Box::new(reader) as Box<dyn SectorSource>, keys);
+    let prefetched = crate::sector::PrefetchedSectorSource::new(
+        decrypting,
+        extents,
+        batch_sectors,
+        halt.clone(),
+    );
+    let (rx, recycle_tx, shell) = prefetched.into_channels();
+
+    let (parsers, pid_to_track, ts, ps) = build_demux_state(&title, format);
+    let (demux_thread, demux_rx) =
+        super::demux_thread::DemuxThread::spawn_zero_copy(rx, recycle_tx, shell, halt, ts, ps);
+    PipelinedPesStream::new(demux_thread, demux_rx, title, parsers, pid_to_track)
+}
+
+/// Assemble the M2TS file mux pipeline (read → demux → parse) for a
+/// byte-stream reader. Scans the head for FMKV header or PMT/PAT,
+/// rebuilds the title metadata, then wraps a chained reader (head +
+/// remainder) in a `BytePrefetcher` feeding the demux + parse
+/// threads.
+fn build_m2ts_pipeline<R: std::io::Read + Send + 'static>(
+    mut reader: R,
+) -> io::Result<PipelinedPesStream> {
+    use super::meta;
+    use std::io::Read;
+
+    const M2TS_SCAN_BYTES: usize = 1024 * 1024;
+    let mut head = vec![0u8; M2TS_SCAN_BYTES];
+    let head_len = {
+        let mut filled = 0;
+        while filled < head.len() {
+            match reader.read(&mut head[filled..])? {
+                0 => break,
+                n => filled += n,
+            }
+        }
+        filled
+    };
+    head.truncate(head_len);
+
+    // Try FMKV metadata header first; fall back to PMT scan.
+    let mut cursor = io::Cursor::new(&head);
+    let (title, head_consumed) = if let Ok(Some(m)) = meta::read_header(&mut cursor) {
+        (m.to_title(), cursor.position() as usize)
+    } else {
+        let streams = super::ts::scan_streams(&head)
+            .ok_or_else(|| -> io::Error { crate::error::Error::NoStreams.into() })?;
+        let t = DiscTitle {
+            duration_secs: 0.0,
+            streams,
+            ..DiscTitle::empty()
+        };
+        (t, 0)
+    };
+
+    // Chain: any un-consumed head bytes + the remainder of the
+    // reader. The demuxer sees a contiguous M2TS byte stream.
+    let remaining_head = head[head_consumed..].to_vec();
+    let chained: Box<dyn Read + Send> = Box::new(io::Cursor::new(remaining_head).chain(reader));
+
+    let prefetcher = crate::io::byte_prefetcher::BytePrefetcher::new(
+        chained,
+        crate::io::byte_prefetcher::DEFAULT_CHUNK_BYTES,
+        None,
+    );
+    let (rx, recycle_tx, shell) = prefetcher.into_channels();
+
+    let (parsers, pid_to_track, ts, ps) = build_demux_state(&title, ContentFormat::BdTs);
+    let (demux_thread, demux_rx) =
+        super::demux_thread::DemuxThread::spawn_zero_copy(rx, recycle_tx, shell, None, ts, ps);
+    Ok(PipelinedPesStream::new(
+        demux_thread,
+        demux_rx,
+        title,
+        parsers,
+        pid_to_track,
+    ))
+}

src/mux/ts.rs

@@ -37,11 +37,20 @@ struct PesAssembler {
     active: bool,
 }
 
+/// Initial capacity for a fresh PES buffer. Sized to cover the
+/// common BD-TS audio / subtitle PES outright (a few KB to ~16 KB).
+/// Video PES (typically 150–300 KB on UHD) will grow this via the
+/// standard Vec doubling, but the doublings hit the allocator's
+/// slab caches instead of the 64-page first-touch faults that the
+/// previous `Vec::with_capacity(256 * 1024)` triggered on every PES
+/// boundary.
+const PES_BUFFER_INIT_CAP: usize = 16 * 1024;
+
 impl PesAssembler {
     fn new(pid: u16) -> Self {
         Self {
             pid,
-            buffer: Vec::with_capacity(256 * 1024),
+            buffer: Vec::with_capacity(PES_BUFFER_INIT_CAP),
             pts: None,
             dts: None,
             active: false,
@@ -55,7 +64,7 @@ impl PesAssembler {
                 pid: self.pid,
                 pts: self.pts,
                 dts: self.dts,
-                data: std::mem::replace(&mut self.buffer, Vec::with_capacity(256 * 1024)),
+                data: std::mem::replace(&mut self.buffer, Vec::with_capacity(PES_BUFFER_INIT_CAP)),
             })
         } else {
             self.buffer.clear();
@@ -124,97 +133,54 @@ impl TsDemuxer {
         }
     }
 
-    /// Feed a chunk of BD transport stream data. Handles non-192-byte-aligned input
+    /// Feed a chunk of BD transport stream data. Handles non-192-byte-
-    /// by buffering leftover bytes between calls. Returns completed PES packets.
+    /// aligned input by buffering leftover bytes between calls. Returns
+    /// completed PES packets.
+    ///
+    /// 16 MiB ISO batches never divide evenly into 192-byte BD-TS
+    /// packets, so every call after the first carries a ~64-byte
+    /// remainder. The pre-0.24 implementation handled this by building
+    /// a `combined` Vec containing remainder + the entire new input —
+    /// a 16 MiB+ memcpy on every call. Now we splice exactly one
+    /// boundary packet from a stack buffer, then process the rest of
+    /// `data` in place. Zero-copy on the bulk path; one 192-byte copy
+    /// on the boundary.
     pub fn feed(&mut self, data: &[u8]) -> Vec<PesPacket> {
         let mut completed = Vec::with_capacity(4);
-
-        // Prepend any remainder from previous call
-        let mut combined: Vec<u8> = Vec::new();
-        let work: &[u8] = if !self.remainder.is_empty() {
-            combined.reserve(self.remainder.len() + data.len());
-            combined.extend_from_slice(&self.remainder);
-            combined.extend_from_slice(data);
-            self.remainder.clear();
-            &combined
-        } else {
-            data
-        };
-
         let mut offset = 0;
 
-        while offset + BD_TS_PACKET_SIZE <= work.len() {
+        // Boundary packet: if a partial packet was left from the last
-            let packet = &work[offset..offset + BD_TS_PACKET_SIZE];
+        // call, complete it from the head of `data` without touching
-            offset += BD_TS_PACKET_SIZE;
+        // the rest of `data`.
-
+        if !self.remainder.is_empty() {
-            // Skip 4-byte TP_extra_header, check sync byte
+            let need = BD_TS_PACKET_SIZE - self.remainder.len();
-            if packet[4] != SYNC_BYTE {
+            if data.len() < need {
-                continue;
+                // Still not a full packet — accumulate and wait.
-            }
+                self.remainder.extend_from_slice(data);
-
+                return completed;
-            let ts = &packet[4..]; // 188-byte standard TS packet
-
-            // Parse TS header
-            let pid = (((ts[1] & 0x1F) as u16) << 8) | ts[2] as u16;
-            let pusi = ts[1] & 0x40 != 0; // Payload Unit Start Indicator
-            let adaptation = (ts[3] >> 4) & 0x03;
-
-            // Check if we're tracking this PID
-            let idx = if (pid as usize) < self.pid_index.len() {
-                self.pid_index[pid as usize]
-            } else {
-                -1
-            };
-            if idx < 0 {
-                continue;
-            }
-            let asm = &mut self.assemblers[idx as usize];
-
-            // Find payload start (skip adaptation field if present)
-            let payload_start = if adaptation == 0x03 || adaptation == 0x02 {
-                // Adaptation field present
-                let af_len = ts[4] as usize;
-                if af_len > 183 {
-                    continue; // Malformed: AF length exceeds TS payload
-                }
-                5 + af_len
-            } else {
-                4
-            };
-
-            if payload_start >= TS_PACKET_SIZE {
-                continue;
-            }
-
-            // No payload
-            if adaptation == 0x02 {
-                continue;
-            }
-
-            let payload = &ts[payload_start..];
-
-            if pusi {
-                // New PES packet starts here — parse PES header
-                let (pts, dts, pes_data_start) = parse_pes_header(payload);
-                if let Some(prev) = asm.start(pts, dts) {
-                    completed.push(prev);
-                }
-                if pes_data_start < payload.len() {
-                    asm.push(&payload[pes_data_start..]);
-                }
-            } else {
-                // Continuation of current PES packet
-                asm.push(payload);
             }
+            let mut boundary = [0u8; BD_TS_PACKET_SIZE];
+            boundary[..self.remainder.len()].copy_from_slice(&self.remainder);
+            boundary[self.remainder.len()..].copy_from_slice(&data[..need]);
+            self.remainder.clear();
+            self.process_packet(&boundary, &mut completed);
+            offset = need;
         }
 
-        // Save leftover bytes for next call (cap at one packet to prevent unbounded growth)
+        // Aligned-packets fast path — reads directly out of `data`.
-        if offset < work.len() {
+        while offset + BD_TS_PACKET_SIZE <= data.len() {
-            let leftover = &work[offset..];
+            let packet = &data[offset..offset + BD_TS_PACKET_SIZE];
+            offset += BD_TS_PACKET_SIZE;
+            self.process_packet(packet, &mut completed);
+        }
+
+        // Save leftover bytes for next call (cap at one packet to
+        // prevent unbounded growth on a desynchronised stream).
+        if offset < data.len() {
+            let leftover = &data[offset..];
             if leftover.len() < BD_TS_PACKET_SIZE {
                 self.remainder.extend_from_slice(leftover);
             } else {
-                // More than one full packet leftover — something is wrong, discard
                 self.remainder.clear();
             }
         }
@@ -222,6 +188,65 @@ impl TsDemuxer {
         completed
     }
 
+    /// Demux a single 192-byte BD-TS packet (4-byte TP_extra_header +
+    /// 188-byte TS). Routes payload bytes into the per-PID
+    /// `PesAssembler`; completed PES packets are pushed onto
+    /// `completed` so the caller's allocation amortises across the
+    /// batch.
+    fn process_packet(&mut self, packet: &[u8], completed: &mut Vec<PesPacket>) {
+        // Sync byte check skips malformed packets.
+        if packet[4] != SYNC_BYTE {
+            return;
+        }
+        let ts = &packet[4..]; // 188-byte standard TS packet
+
+        let pid = (((ts[1] & 0x1F) as u16) << 8) | ts[2] as u16;
+        let pusi = ts[1] & 0x40 != 0; // Payload Unit Start Indicator
+        let adaptation = (ts[3] >> 4) & 0x03;
+
+        let idx = if (pid as usize) < self.pid_index.len() {
+            self.pid_index[pid as usize]
+        } else {
+            -1
+        };
+        if idx < 0 {
+            return;
+        }
+        let asm = &mut self.assemblers[idx as usize];
+
+        let payload_start = if adaptation == 0x03 || adaptation == 0x02 {
+            let af_len = ts[4] as usize;
+            if af_len > 183 {
+                return; // Malformed: AF length exceeds TS payload
+            }
+            5 + af_len
+        } else {
+            4
+        };
+
+        if payload_start >= TS_PACKET_SIZE {
+            return;
+        }
+        // adaptation == 0x02 → AF only, no payload.
+        if adaptation == 0x02 {
+            return;
+        }
+
+        let payload = &ts[payload_start..];
+
+        if pusi {
+            let (pts, dts, pes_data_start) = parse_pes_header(payload);
+            if let Some(prev) = asm.start(pts, dts) {
+                completed.push(prev);
+            }
+            if pes_data_start < payload.len() {
+                asm.push(&payload[pes_data_start..]);
+            }
+        } else {
+            asm.push(payload);
+        }
+    }
+
     /// Flush all assemblers, returning any remaining PES packets.
     pub fn flush(&mut self) -> Vec<PesPacket> {
         let mut completed = Vec::new();

src/platform/fs_type/linux.rs

@@ -6,6 +6,7 @@
 
 use std::ffi::CString;
 use std::os::unix::ffi::OsStrExt;
+use std::os::unix::io::RawFd;
 use std::path::Path;
 
 use super::FsType;
@@ -18,6 +19,27 @@ const XFS_SUPER_MAGIC: i64 = 0x5846_5342;
 const BTRFS_SUPER_MAGIC: i64 = 0x9123_683E;
 const TMPFS_MAGIC: i64 = 0x0102_1994;
 
+/// Classify an `f_type` magic from `statfs`/`fstatfs`. Single source
+/// of truth for the magic comparisons used by both the path-based
+/// (`detect_impl`) and fd-based (`detect_fd_impl`) entry points.
+///
+/// Cast to `i64` because `statfs::f_type` is signed `__fsword_t` on
+/// glibc and unsigned `c_ulong` on musl; a portable comparison needs a
+/// common type. On glibc x86_64 both already are i64 — clippy flags
+/// the cast as unnecessary on that target only, but we need it for
+/// musl.
+#[allow(clippy::unnecessary_cast)]
+fn classify_f_type(f_type: i64) -> FsType {
+    let nfs_magic = libc::NFS_SUPER_MAGIC as i64;
+    if f_type == nfs_magic {
+        return FsType::Nfs;
+    }
+    match f_type {
+        EXT2_SUPER_MAGIC | XFS_SUPER_MAGIC | BTRFS_SUPER_MAGIC | TMPFS_MAGIC => FsType::Local,
+        _ => FsType::Unknown,
+    }
+}
+
 pub(super) fn detect_impl(path: &Path) -> FsType {
     let cpath = match CString::new(path.as_os_str().as_bytes()) {
         Ok(c) => c,
@@ -30,20 +52,19 @@ pub(super) fn detect_impl(path: &Path) -> FsType {
     if rc != 0 {
         return FsType::Unknown;
     }
-    // `f_type` is signed (`__fsword_t`) on glibc and unsigned
-    // (`c_ulong`) on musl. Cast both sides to i64 for a portable
-    // comparison. On glibc x86_64 both already are i64 — clippy flags
-    // the cast as unnecessary on that target only, but we need it for
-    // musl, so silence the lint.
     #[allow(clippy::unnecessary_cast)]
-    let f_type = buf.f_type as i64;
+    classify_f_type(buf.f_type as i64)
+}
+
+/// fd-based variant of [`detect_impl`]. Same return semantics. Used
+/// by the writeback pipeline, which knows the open `File` but not its
+/// original path.
+pub(super) fn detect_fd_impl(fd: RawFd) -> FsType {
+    let mut buf: libc::statfs = unsafe { std::mem::zeroed() };
+    let rc = unsafe { libc::fstatfs(fd, &mut buf) };
+    if rc != 0 {
+        return FsType::Unknown;
+    }
     #[allow(clippy::unnecessary_cast)]
-    let nfs_magic = libc::NFS_SUPER_MAGIC as i64;
+    classify_f_type(buf.f_type as i64)
-    if f_type == nfs_magic {
-        return FsType::Nfs;
-    }
-    match f_type {
-        EXT2_SUPER_MAGIC | XFS_SUPER_MAGIC | BTRFS_SUPER_MAGIC | TMPFS_MAGIC => FsType::Local,
-        _ => FsType::Unknown,
-    }
 }

src/platform/fs_type/mod.rs

@@ -51,6 +51,9 @@ use other::detect_impl;
 #[cfg(target_os = "windows")]
 use windows::detect_impl;
 
+#[cfg(target_os = "linux")]
+use linux::detect_fd_impl;
+
 /// Best-effort classification of the filesystem under `path`.
 ///
 /// Falls back to [`FsType::Unknown`] on any syscall error or unrecognised
@@ -60,6 +63,25 @@ pub fn detect(path: &Path) -> FsType {
     detect_impl(path)
 }
 
+/// fd-based classification. Same return semantics as [`detect`], but
+/// takes a `RawFd` so callers that only have an open file (notably
+/// [`crate::io::writeback::WritebackPipeline`]) don't have to
+/// round-trip through the path.
+///
+/// Only implemented on Linux; other platforms return
+/// [`FsType::Unknown`] (none of them have a writeback policy that
+/// keys off this classification today).
+#[cfg(target_os = "linux")]
+pub fn detect_fd(fd: std::os::unix::io::RawFd) -> FsType {
+    detect_fd_impl(fd)
+}
+
+#[cfg(not(target_os = "linux"))]
+#[allow(dead_code)] // API parity with the linux impl; callers cfg-gate.
+pub fn detect_fd(_fd: i32) -> FsType {
+    FsType::Unknown
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/sector/mod.rs

@@ -6,7 +6,7 @@
 //! ISO images are opened for read OR write at construction time.
 //!
 //! - [`SectorSource`] is implemented by `Drive` (hardware) and
-//!   [`FileSectorSource`] / `IsoSectorReader` (file-backed).
+//!   [`FileSectorSource`] (file-backed).
 //! - [`SectorSink`] is implemented by [`FileSectorSink`]
 //!   (ISO-backed) and sweep/patch consumer adapters.
 //! - [`DecryptingSectorSource`] is a decorator that wraps any
@@ -15,6 +15,7 @@
 
 pub mod decrypting;
 pub mod file;
+pub mod prefetched;
 
 use crate::error::Result;
 
@@ -112,3 +113,4 @@ pub trait SectorSink: Send {
 pub use crate::io::file_sector_source::FileSectorSource;
 pub use decrypting::DecryptingSectorSource;
 pub use file::FileSectorSink;
+pub use prefetched::PrefetchedSectorSource;

src/sector/prefetched.rs

@@ -0,0 +1,250 @@
+//! `PrefetchedSectorSource` — runs the wrapped read+decrypt in a
+//! dedicated producer thread and surfaces the prepared plaintext
+//! buffers on demand via a bounded channel.
+//!
+//! ## Why
+//!
+//! The mux consumer (demux + codec parsing + frame output) is
+//! single-threaded by nature (streams are sequential). The mux
+//! producer (read sectors + AACS decrypt) is also single-threaded
+//! per-call but does CPU-heavy work (AES per 6144-byte unit). Running
+//! both on the same thread means the disk and decrypt cores sit idle
+//! while the demux runs, and vice versa.
+//!
+//! Splitting them across two threads with a bounded channel between
+//! lets both run in parallel — peak throughput becomes
+//! `min(producer_rate, consumer_rate)` instead of
+//! `1 / (1/producer + 1/consumer)`.
+//!
+//! ## Lifecycle
+//!
+//! The producer thread is spawned by [`PrefetchedSectorSource::new`].
+//! It walks the supplied extent list in order, reads the configured
+//! batch size at each LBA, and sends the resulting plaintext buffer
+//! into a [`crossbeam_channel::bounded`] channel of small depth (so
+//! the producer stays a couple of batches ahead without unbounded
+//! memory growth).
+//!
+//! When the channel sender drops (either because all extents were
+//! served or because the [`Halt`] token cancelled), the consumer
+//! observes `RecvError` on the next `read_sectors` and treats it as
+//! end-of-stream. Errors from the underlying reader are forwarded
+//! verbatim through the channel.
+//!
+//! ## Read API
+//!
+//! `read_sectors` ignores its `lba`/`count` arguments — the producer
+//! has already chosen what to read, in the order the extents dictate.
+//! This is sound for the mux read path, which always walks extents
+//! sequentially and never seeks. For random-access callers (sweep
+//! patch retries) this wrapper is the wrong tool — they should keep
+//! reading the underlying source directly.
+
+use crate::error::Result;
+use crate::halt::Halt;
+use crate::sector::SectorSource;
+use crossbeam_channel::{Receiver, Sender, bounded};
+use std::thread::JoinHandle;
+
+const PREFETCH_CHANNEL_DEPTH: usize = 2;
+
+/// Smallest sector source the producer will issue per read. AACS
+/// alignment requires multiples of 3 sectors so a unit doesn't span
+/// two reads.
+const SECTOR_ALIGNMENT: u16 = 3;
+
+/// Item flowing through the prefetch forward channel.
+pub type Batch = std::result::Result<Vec<u8>, std::io::Error>;
+
+/// Producer-thread-backed [`SectorSource`] decorator. Construct it
+/// with the real reader, the extent list to walk, and the batch
+/// size; the wrapper spawns the producer immediately and starts
+/// filling the channel.
+pub struct PrefetchedSectorSource {
+    rx: Receiver<Batch>,
+    /// Recycle channel — consumer returns drained buffers here; the
+    /// producer re-fills them in place. Lets the producer/consumer
+    /// reuse a fixed pool of `PREFETCH_CHANNEL_DEPTH+1` buffers
+    /// instead of `Vec::new()`-ing one per batch (musl mallocng
+    /// cross-thread alloc/free was the dominant cost in the demux
+    /// thread before this).
+    recycle_tx: Sender<Vec<u8>>,
+    /// Joined on drop so producer cleanup runs deterministically.
+    producer: Option<JoinHandle<()>>,
+    /// Cumulative bytes drained by `read_sectors` calls. Exposed via
+    /// [`capacity_sectors`] indirectly: the consumer-side state needs
+    /// this to advance its position bookkeeping in lockstep with what
+    /// the producer fed.
+    total_sectors: u32,
+}
+
+impl PrefetchedSectorSource {
+    /// Spawn the producer thread. `reader` must already be the fully
+    /// composed read+decrypt stack (e.g.
+    /// [`DecryptingSectorSource`](crate::sector::DecryptingSectorSource))
+    /// — every byte the producer emits is what the consumer's demux
+    /// will feed to its codec parsers.
+    pub fn new<S>(
+        mut reader: S,
+        extents: Vec<crate::disc::Extent>,
+        batch_sectors: u16,
+        halt: Option<Halt>,
+    ) -> Self
+    where
+        S: SectorSource + Send + 'static,
+    {
+        let total_sectors: u32 = extents.iter().map(|e| e.sector_count).sum();
+        let (tx, rx) = bounded::<Batch>(PREFETCH_CHANNEL_DEPTH);
+        let (recycle_tx, recycle_rx) = bounded::<Vec<u8>>(PREFETCH_CHANNEL_DEPTH + 1);
+        let batch_bytes = batch_sectors as usize * 2048;
+
+        // Seed the recycle pool so the producer always has a buffer
+        // to fill on the first iteration. Without these, the first
+        // `recycle_rx.recv()` would block forever (no consumer has
+        // returned a buffer yet).
+        for _ in 0..(PREFETCH_CHANNEL_DEPTH + 1) {
+            let _ = recycle_tx.send(vec![0u8; batch_bytes]);
+        }
+
+        let producer = std::thread::Builder::new()
+            .name("freemkv-prefetch".into())
+            .spawn(move || {
+                let mut ext_idx = 0usize;
+                let mut offset: u32 = 0;
+                while ext_idx < extents.len() {
+                    if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) {
+                        return;
+                    }
+                    let extent = &extents[ext_idx];
+                    let remaining = extent.sector_count.saturating_sub(offset);
+                    if remaining == 0 {
+                        ext_idx += 1;
+                        offset = 0;
+                        continue;
+                    }
+                    let mut sectors = remaining.min(batch_sectors as u32) as u16;
+                    if sectors >= SECTOR_ALIGNMENT {
+                        sectors -= sectors % SECTOR_ALIGNMENT;
+                    }
+                    let bytes = sectors as usize * 2048;
+                    let mut buf = match recycle_rx.recv() {
+                        Ok(b) => b,
+                        Err(_) => return, // consumer dropped both channels
+                    };
+                    if buf.len() < bytes {
+                        buf.resize(bytes, 0);
+                    } else {
+                        // Re-expose the full extent; previous truncate
+                        // shrank the visible len without freeing pages.
+                        // SAFETY: capacity is at least `bytes` after
+                        // construction with `vec![0u8; batch_bytes]`.
+                        unsafe { buf.set_len(bytes) };
+                    }
+                    let lba = extent.start_lba + offset;
+                    match reader.read_sectors(lba, sectors, &mut buf[..bytes], false) {
+                        Ok(n) => {
+                            buf.truncate(n);
+                            if tx.send(Ok(buf)).is_err() {
+                                return; // consumer dropped
+                            }
+                            offset += sectors as u32;
+                        }
+                        Err(e) => {
+                            let _ = tx.send(Err(e.into()));
+                            return;
+                        }
+                    }
+                }
+                // Drop tx implicitly — consumer sees RecvError → EOF.
+            })
+            .expect("freemkv-prefetch producer spawn failed");
+
+        Self {
+            rx,
+            recycle_tx,
+            producer: Some(producer),
+            total_sectors,
+        }
+    }
+
+    /// Peel off the receivers for zero-copy pipeline mode. The
+    /// caller (typically [`super::super::mux::demux_thread::DemuxThread`])
+    /// pulls buffers from `rx`, consumes them, and pushes the empty
+    /// `Vec<u8>` back through `recycle_tx` so the producer can
+    /// re-fill it. The producer-thread `JoinHandle` stays with the
+    /// returned `PrefetchedSectorSource` shell; drop that to join.
+    ///
+    /// Returns `(forward_rx, recycle_tx, shell)`. The shell only
+    /// holds the join handle and total_sectors for `capacity_sectors`
+    /// queries; its `SectorSource` impl becomes invalid after this
+    /// call (data has been moved out).
+    pub fn into_channels(self) -> (Receiver<Batch>, Sender<Vec<u8>>, PrefetchShell) {
+        let total = self.total_sectors;
+        // Drop the SectorSource side; transfer the producer join
+        // handle to a shell that just waits on Drop.
+        let mut me = self;
+        let producer = me.producer.take();
+        let rx = me.rx.clone();
+        let recycle = me.recycle_tx.clone();
+        std::mem::forget(me);
+        (rx, recycle, PrefetchShell { producer, total })
+    }
+}
+
+/// Returned from [`PrefetchedSectorSource::into_channels`]. Owns the
+/// producer thread join handle so dropping the shell joins the
+/// producer, even though the channels have been peeled off.
+pub struct PrefetchShell {
+    producer: Option<JoinHandle<()>>,
+    #[allow(dead_code)]
+    total: u32,
+}
+
+impl Drop for PrefetchShell {
+    fn drop(&mut self) {
+        if let Some(h) = self.producer.take() {
+            let _ = h.join();
+        }
+    }
+}
+
+impl Drop for PrefetchedSectorSource {
+    fn drop(&mut self) {
+        // Dropping the receiver closes the channel, which makes the
+        // next producer `send` return Err and exits the loop. Joining
+        // here gives us a deterministic shutdown — no detached thread
+        // can outlive the source.
+        if let Some(h) = self.producer.take() {
+            let _ = h.join();
+        }
+    }
+}
+
+impl SectorSource for PrefetchedSectorSource {
+    fn capacity_sectors(&self) -> u32 {
+        self.total_sectors
+    }
+
+    fn read_sectors(
+        &mut self,
+        _lba: u32,
+        _count: u16,
+        buf: &mut [u8],
+        _recovery: bool,
+    ) -> Result<usize> {
+        // The producer has already decided the next batch. lba/count
+        // are advisory; the consumer's fill_extents will advance its
+        // own bookkeeping using the returned byte count, not the
+        // requested count.
+        match self.rx.recv() {
+            Ok(Ok(filled)) => {
+                let n = filled.len().min(buf.len());
+                buf[..n].copy_from_slice(&filled[..n]);
+                Ok(n)
+            }
+            Ok(Err(e)) => Err(crate::error::Error::IoError { source: e }),
+            // Channel closed (producer finished or panicked).
+            Err(_) => Ok(0),
+        }
+    }
+}