diff --git a/Cargo.toml b/Cargo.toml index 9881523..61cac28 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/decrypt.rs b/src/decrypt.rs index b8131d9..b84d479 100644 --- a/src/decrypt.rs +++ b/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>>` 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>> = 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 { + // 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::().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 = 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 } => { diff --git a/src/halt.rs b/src/halt.rs index e357b15..06bf932 100644 --- a/src/halt.rs +++ b/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::*; diff --git a/src/io/bounded.rs b/src/io/bounded.rs index 3daf685..45b83f9 100644 --- a/src/io/bounded.rs +++ b/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; - -/// 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); +use crate::halt::{Halt, POLL_INTERVAL}; /// Failure outcome from a bounded syscall wrapper. #[derive(Debug)] diff --git a/src/io/byte_prefetcher.rs b/src/io/byte_prefetcher.rs new file mode 100644 index 0000000..bd8aa8e --- /dev/null +++ b/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` +//! chunks from the underlying reader and ships them through a +//! channel; the consumer pulls filled chunks, uses them, and sends +//! the empty `Vec` 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>; + +/// 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>, +} + +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, + recycle_tx: Sender>, + producer: Option>, +} + +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( + mut reader: R, + chunk_bytes: usize, + halt: Option, + ) -> Self { + let (tx, rx) = bounded::(FORWARD_DEPTH); + let (recycle_tx, recycle_rx) = bounded::>(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, Sender>, 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(); + } + } +} diff --git a/src/io/file_sector_source/linux.rs b/src/io/file_sector_source/linux.rs index 0355cee..8a7c0eb 100644 --- a/src/io/file_sector_source/linux.rs +++ b/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//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); + } +} diff --git a/src/io/file_sector_source/macos.rs b/src/io/file_sector_source/macos.rs index a196d6c..d3519cd 100644 --- a/src/io/file_sector_source/macos.rs +++ b/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); + } +} diff --git a/src/io/file_sector_source/mod.rs b/src/io/file_sector_source/mod.rs index 99b2b98..3d6624c 100644 --- a/src/io/file_sector_source/mod.rs +++ b/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 -//! NFS round-trip for every sector. With `rsize=1 MiB` and a 100-150 ms -//! NFS RTT, that's three orders of magnitude more round trips than -//! necessary — the muxer goes read-bound on every read, even though -//! the local NFS client could deliver MB/s on bigger requests. +//! Pre-0.21.3 this source held a 32 MiB (later 4 MiB) read-ahead +//! buffer to amortise per-sector NFS round-trips. Empirically that +//! buffer hurt: 32 MiB refills bursted the NFS TCP connection hard +//! enough to starve the concurrent writer, and even a 4 MiB window +//! 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) -//! 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). +//! ## DONTNEED on the consumed window //! -//! ## 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). -//! 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 +//! ## Platform open hint //! //! On `open()` each platform issues its "sequential access expected" -//! hint to the kernel so OS-level readahead widens. The hint lives in -//! a per-OS sibling module ([`linux::hint_sequential`] et al.) — no -//! inline `#[cfg]` in this file. +//! hint so OS-level readahead widens. The hint and the DONTNEED call +//! live in per-OS sibling modules ([`linux::hint_sequential`] et al.) +//! — 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 -/// dropped this from the pre-Phase-1 (0.20.7) hot path. Without it, -/// 85 GB of streaming ISO reads pin the entire file in the kernel page -/// cache, starving the MKV writeback and collapsing mux throughput -/// (observed: 2.7 MB/s mux on 0.21.5 vs. 70 MB/s isolated NFS reads). -const READ_DROP_CHUNK_BYTES: u64 = 32 * 1024 * 1024; +/// 32 MiB is the empirically tuned value on the rip1 test bed (single +/// 7200rpm HDD via SATA): smaller windows (8 / 16 MiB) shorten the +/// kernel-readahead overlap and slow the producer; larger windows +/// (64 / 128 MiB) let the page cache pin enough of the ISO to +/// pressure concurrent writes. Override via `FREEMKV_READ_DROP_CHUNK_MIB`. +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::().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 - /// path (every `read_sectors` is a direct pread). The fields are - /// 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 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. - /// The next DONTNEED drops from `drop_window_start` for + /// File offset at which the current drop window starts. The next + /// 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 - // MKV output) and collapses mux throughput. Mirrors the - // write-side WritebackPipeline's DONTNEED policy. + // the kernel page cache, which starves concurrent writes and + // collapses mux throughput. Mirrors the write-side + // 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 - // boundaries. - let total = BUF_SECTORS * 2 + 17; + let total = TEST_SPAN_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() { - // A read that lands exactly on the last sector of the buffer - // 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; + fn multi_sector_read_across_chunk_boundary() { + let total = TEST_SPAN_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, - // 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 span_lba = TEST_SPAN_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() { - // Read forward across two windows, then jump back to sector - // 0. Buffer must refill from the start. - let total = BUF_SECTORS * 2 + 5; + fn backward_seek_reads_correct_bytes() { + // Read forward then jump back: the SectorSource contract is + // byte-correctness regardless of access pattern. + 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(BUF_SECTORS + 1, 1, &mut got, false) + src.read_sectors(TEST_SPAN_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() { - // File is smaller than one buffer window. The buffer must - // populate with only the available sectors and reads must - // still succeed. + fn read_at_eof_returns_correct_bytes() { + // File smaller than the readahead chunk — reads near EOF must + // still return correct bytes. 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() { - // A request larger than the buffer must not deadlock the - // refill (which only loads BUF_SECTORS at a time). Bypass - // path handles it via direct pread. - let total = BUF_SECTORS + 100; + fn large_single_read() { + // A multi-MB single read must work — the implementation has + // no app-level chunking, so this just exercises the direct + // pread path on a larger request. + 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 - // 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 = (TEST_SPAN_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"); + } + } } diff --git a/src/io/file_sector_source/other.rs b/src/io/file_sector_source/other.rs index 805192a..97ca955 100644 --- a/src/io/file_sector_source/other.rs +++ b/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) {} diff --git a/src/io/file_sector_source/windows.rs b/src/io/file_sector_source/windows.rs index 0cc76f3..81fbf94 100644 --- a/src/io/file_sector_source/windows.rs +++ b/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) {} diff --git a/src/io/mod.rs b/src/io/mod.rs index d7233d6..993502c 100644 --- a/src/io/mod.rs +++ b/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 -//! 32 MiB read-ahead buffer that amortises NFS round-trip latency -//! across thousands of sector reads. +//! [`crate::sector::SectorSource`] for an ISO file using direct +//! `pread`-equivalent calls so the kernel's own readahead policy runs +//! (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 -// (disc/sweep.rs, disc/patch.rs); mux migrates separately in autorip. -// `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)] +// Re-exports for the 0.18 redesign. Sweep, patch, and mux are all +// wired up (disc/sweep.rs, disc/patch.rs, autorip's ripper/mux.rs). pub use pipeline::{ DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, READ_PIPELINE_DEPTH, Sink, WRITE_PIPELINE_DEPTH, WRITE_THROUGH_DEPTH, diff --git a/src/io/pipeline.rs b/src/io/pipeline.rs index eb9d47f..2916797 100644 --- a/src/io/pipeline.rs +++ b/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 -/// the producer checks the halt token; that's the latency a stop -/// request will observe. +/// consumer is genuinely wedged, the timeout fires every +/// [`crate::halt::POLL_INTERVAL`] and the producer checks the halt +/// 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. -#[allow(dead_code)] +/// emitting the next item. Currently used by `disc::patch`. pub const WRITE_THROUGH_DEPTH: usize = 1; /// Outcome of [`Sink::apply`]: either keep feeding items @@ -162,11 +161,10 @@ impl Pipeline { /// propagated rather than panicked. /// /// Sweep uses [`Pipeline::spawn_named`] directly so the consumer - /// thread shows up as `freemkv-sweep-consumer`; this function has - /// no in-tree caller yet. Patch and mux migrate in later 0.18 - /// slices. The targeted `#[allow]` is removed when one of them - /// lands on the default name. - #[allow(dead_code)] + /// thread shows up as `freemkv-sweep-consumer`; mux uses + /// `freemkv-mux-consumer`. `Pipeline::spawn` (this function, with + /// the default name) is used by `disc::patch` and by the unit + /// tests in this module. pub fn spawn>(depth: usize, sink: S) -> Result { Self::spawn_named("freemkv-pipeline-consumer", depth, sink) } diff --git a/src/io/platform_macos.rs b/src/io/platform_macos.rs new file mode 100644 index 0000000..26b692f --- /dev/null +++ b/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, +} diff --git a/src/io/sink/preallocate/macos.rs b/src/io/sink/preallocate/macos.rs index dd7bc13..f4c8360 100644 --- a/src/io/sink/preallocate/macos.rs +++ b/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 ``. libc on -// some Rust toolchains/versions doesn't ship this binding, so define -// 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 . -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; +use crate::io::platform_macos::{ + F_ALLOCATEALL, F_ALLOCATECONTIG, F_PEOFPOSMODE, F_PREALLOCATE, Fstore, +}; pub(super) fn preallocate_impl(file: &File, size_bytes: u64) { let fd = file.as_raw_fd(); diff --git a/src/io/writeback/linux.rs b/src/io/writeback/linux.rs index 0d1ede3..2aee7f0 100644 --- a/src/io/writeback/linux.rs +++ b/src/io/writeback/linux.rs @@ -300,35 +300,22 @@ impl WritebackPipeline { } } -/// Probe whether `fd` lives on an NFS mount via `fstatfs`. Returns -/// `false` on any error — we fail open, not closed: better to run the -/// normal local-storage path on a misdetected NFS mount (and surface -/// the freeze loudly via the timeout) than to needlessly disable -/// writeback bounding on every local file because of a transient -/// stat error. +/// Probe whether `fd` lives on an NFS mount. Thin wrapper around +/// [`crate::platform::fs_type::detect_fd`] so writeback policy and +/// general-purpose fs-type classification stay in sync (same magic +/// numbers, same musl-vs-glibc cast handling). +/// +/// 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 - // documented init pattern for the kernel uapi struct. - let mut buf: libc::statfs = unsafe { std::mem::zeroed() }; - 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 + matches!( + crate::platform::fs_type::detect_fd(fd), + crate::platform::fs_type::FsType::Nfs + ) } /// 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 { } } } + +#[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()); + } +} diff --git a/src/io/writeback_file/macos.rs b/src/io/writeback_file/macos.rs index 4959298..ed9aefb 100644 --- a/src/io/writeback_file/macos.rs +++ b/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 -/// SDK versions, so define it here. -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; +use crate::io::platform_macos::{ + F_ALLOCATEALL, F_ALLOCATECONTIG, F_PEOFPOSMODE, F_PREALLOCATE, Fstore, +}; /// `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, diff --git a/src/io/writeback_file/mod.rs b/src/io/writeback_file/mod.rs index f9b226f..89b00b7 100644 --- a/src/io/writeback_file/mod.rs +++ b/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: -/// iter8 = 28.7, iter9 (64 MiB) = 27.5, iter11 (128 MiB) = 16.6, -/// iter6 (8 MiB) = 15.8. Locked. -const WRITEBACK_CHUNK_BYTES: u64 = 32 * 1024 * 1024; +/// `sync_file_range` pairs. 32 MiB is the empirically best value on +/// the rip1 test bed (NFS to unraid-1 over 1 GbE, single-disk SAS): +/// 8 MiB / 64 MiB / 128 MiB all measured worse in the 0.21.x mux +/// 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::().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 { 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, diff --git a/src/lib.rs b/src/lib.rs index 24b2de9..adc42f8 100644 --- a/src/lib.rs +++ b/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}; diff --git a/src/mux/codec/h264.rs b/src/mux/codec/h264.rs index a674237..4c4670e 100644 --- a/src/mux/codec/h264.rs +++ b/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 { if data.len() < from + 3 { return None; } - // Range excludes last 2 bytes since we read 3 bytes at each position. - // 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) + memchr::memmem::find(&data[from..], b"\x00\x00\x01").map(|rel| from + rel) } /// Skip past the start code at position `pos`, returning the first byte after it. diff --git a/src/mux/codec/hevc.rs b/src/mux/codec/hevc.rs index 08d3345..5899fe0 100644 --- a/src/mux/codec/hevc.rs +++ b/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; diff --git a/src/mux/demux_thread.rs b/src/mux/demux_thread.rs new file mode 100644 index 0000000..d4f9816 --- /dev/null +++ b/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` 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`. +//! 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), + Ps(Vec), + /// 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>` — the join handle of the +/// upstream producer (sector or byte prefetcher). Dropping the +/// `DemuxThread` runs the shell's `Drop`, which joins the producer. +/// `Box` rather than a concrete type so the same demux +/// worker can be wired behind either prefetcher kind. +pub struct DemuxThread { + handle: Option>, + #[allow(dead_code)] + producer_shell: Option>, +} + +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( + mut reader: S, + extents: Vec, + batch_sectors: u16, + halt: Option, + ts: Option, + ps: Option, + ) -> (Self, Receiver) { + let (tx, rx) = bounded::(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( + prefetch_rx: Receiver>>, + recycle_tx: Sender>, + producer_shell: S, + halt: Option, + ts: Option, + ps: Option, + ) -> (Self, Receiver) { + let (tx, rx) = bounded::(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(); + } + } +} diff --git a/src/mux/disc.rs b/src/mux/disc.rs index 539b723..b45070a 100644 --- a/src/mux/disc.rs +++ b/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, ps_demuxer: Option, parsers: Vec<(u16, Box)>, pending_frames: std::collections::VecDeque, 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, + demux_rx: Option>, } 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, + ) -> 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 = 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 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 { + Ok(0) + } + } + let dummy: Box = 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> = 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> { + 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> { 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 { - if let Some((_, track)) = - self.pid_to_track.iter().find(|(pid, _)| *pid == pes.pid) + let t2 = std::time::Instant::now(); + prof_tick("feed", t2.duration_since(t1).as_nanos(), 0); + 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() { diff --git a/src/mux/iso.rs b/src/mux/iso.rs deleted file mode 100644 index 5e82f1a..0000000 --- a/src/mux/iso.rs +++ /dev/null @@ -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 { - 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 { - 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(); - } -} diff --git a/src/mux/m2ts.rs b/src/mux/m2ts.rs index bea1925..88b5c28 100644 --- a/src/mux/m2ts.rs +++ b/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. -//! Read: extracts metadata header (or scans PMT), then demuxes BD-TS into PES frames. +//! Write: prepends FMKV metadata header, then muxes PES frames into +//! 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 = ( - Vec, - Vec<(u16, Box)>, - 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>, - }, - Read { - reader: Box, - }, -} - -/// 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 { - 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. +/// BD transport stream write sink with embedded FMKV metadata +/// header. pub struct M2tsStream { disc_title: DiscTitle, - mode: Mode, - // PES support - demuxer: Option, - parsers: Vec<(u16, Box)>, - pending_frames: std::collections::VecDeque, - pid_to_track: Vec<(u16, usize)>, - pes_eof: bool, - /// Codec private data per stream (from FMKV header). - stored_codec_privates: Vec>>, + muxer: super::tsmux::TsMuxer>, } impl M2tsStream { @@ -81,208 +45,38 @@ impl M2tsStream { } Ok(Self { disc_title: title.clone(), - mode: Mode::Write { 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)> = 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 { - // 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 = - 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 = 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(), + muxer, }) } } impl crate::pes::Stream for M2tsStream { fn read(&mut self) -> io::Result> { - if let Some(frame) = self.pending_frames.pop_front() { - return Ok(Some(frame)); - } - if self.pes_eof { - return Ok(None); - } - - 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)); - } - } + // Write-only sink. The m2ts:// read direction is served by + // `super::resolve::build_m2ts_pipeline` → + // `PipelinedPesStream`; routing through this type for reads + // was removed when the highway became the only ingress. + Err(crate::error::Error::StreamWriteOnly.into()) } fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> { - match &mut self.mode { - Mode::Write { muxer } => { - muxer.write_frame(frame.track, frame.pts, frame.keyframe, &frame.data) - } - Mode::Read { .. } => Err(crate::error::Error::StreamReadOnly.into()), - } + self.muxer + .write_frame(frame.track, frame.pts, frame.keyframe, &frame.data) } fn finish(&mut self) -> io::Result<()> { - match &mut self.mode { - Mode::Write { muxer } => muxer.finish(), - Mode::Read { .. } => Ok(()), - } + self.muxer.finish() } fn info(&self) -> &crate::disc::DiscTitle { &self.disc_title } - fn codec_private(&self, track: usize) -> Option> { - // First check stored codec_privates from FMKV header - if let Some(Some(cp)) = self.stored_codec_privates.get(track) { - return Some(cp.clone()); - } - // 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 + fn codec_private(&self, _track: usize) -> Option> { + // Write side doesn't have parsers; codec_private flows in + // via the title metadata at `create` time and gets baked + // into the FMKV header. Nothing to surface back here. + None } } diff --git a/src/mux/mod.rs b/src/mux/mod.rs index 537a09c..e95d105 100644 --- a/src/mux/mod.rs +++ b/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; diff --git a/src/mux/pipelined_stream.rs b/src/mux/pipelined_stream.rs new file mode 100644 index 0000000..9eb46be --- /dev/null +++ b/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)>, + pid_to_track: Vec<(u16, usize)>, + + demux_rx: Receiver, + /// 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, + 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, + title: DiscTitle, + parsers: Vec<(u16, Box)>, + 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 { + 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) { + 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) { + 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> { + 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> { + 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()) + } +} diff --git a/src/mux/resolve.rs b/src/mux/resolve.rs index b7ac8f0..c6aa0cd 100644 --- a/src/mux/resolve.rs +++ b/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 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 { 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)>, + Vec<(u16, usize)>, + Option, + Option, +); + +/// 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( + reader: S, + title: DiscTitle, + keys: crate::decrypt::DecryptKeys, + batch_sectors: u16, + format: ContentFormat, + halt: Option, +) -> PipelinedPesStream { + let extents = title.extents.clone(); + let decrypting = + crate::sector::DecryptingSectorSource::new(Box::new(reader) as Box, 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( + mut reader: R, +) -> io::Result { + 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 = 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, + )) +} diff --git a/src/mux/ts.rs b/src/mux/ts.rs index e55383e..3f4b864 100644 --- a/src/mux/ts.rs +++ b/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 - /// by buffering leftover bytes between calls. Returns completed PES packets. + /// Feed a chunk of BD transport stream data. Handles non-192-byte- + /// 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 { let mut completed = Vec::with_capacity(4); - - // Prepend any remainder from previous call - let mut combined: Vec = 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() { - let packet = &work[offset..offset + BD_TS_PACKET_SIZE]; - offset += BD_TS_PACKET_SIZE; - - // Skip 4-byte TP_extra_header, check sync byte - if packet[4] != SYNC_BYTE { - continue; - } - - 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); + // Boundary packet: if a partial packet was left from the last + // call, complete it from the head of `data` without touching + // the rest of `data`. + if !self.remainder.is_empty() { + let need = BD_TS_PACKET_SIZE - self.remainder.len(); + if data.len() < need { + // Still not a full packet — accumulate and wait. + self.remainder.extend_from_slice(data); + return completed; } + 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) - if offset < work.len() { - let leftover = &work[offset..]; + // Aligned-packets fast path — reads directly out of `data`. + while offset + BD_TS_PACKET_SIZE <= data.len() { + 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) { + // 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 { let mut completed = Vec::new(); diff --git a/src/platform/fs_type/linux.rs b/src/platform/fs_type/linux.rs index 591018e..f5df6a5 100644 --- a/src/platform/fs_type/linux.rs +++ b/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; - 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, - } + classify_f_type(buf.f_type as i64) } diff --git a/src/platform/fs_type/mod.rs b/src/platform/fs_type/mod.rs index d0db78a..cc6c4ea 100644 --- a/src/platform/fs_type/mod.rs +++ b/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::*; diff --git a/src/sector/mod.rs b/src/sector/mod.rs index 17e6f09..67df098 100644 --- a/src/sector/mod.rs +++ b/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; diff --git a/src/sector/prefetched.rs b/src/sector/prefetched.rs new file mode 100644 index 0000000..1a555ee --- /dev/null +++ b/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, 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, + /// 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>, + /// Joined on drop so producer cleanup runs deterministically. + producer: Option>, + /// 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( + mut reader: S, + extents: Vec, + batch_sectors: u16, + halt: Option, + ) -> Self + where + S: SectorSource + Send + 'static, + { + let total_sectors: u32 = extents.iter().map(|e| e.sector_count).sum(); + let (tx, rx) = bounded::(PREFETCH_CHANNEL_DEPTH); + let (recycle_tx, recycle_rx) = bounded::>(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` 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, Sender>, 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>, + #[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 { + // 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), + } + } +}