diff --git a/src/io/byte_channel.rs b/src/io/byte_channel.rs new file mode 100644 index 0000000..ef92f96 --- /dev/null +++ b/src/io/byte_channel.rs @@ -0,0 +1,330 @@ +//! Byte-sized bounded producer/consumer channel. +//! +//! Wraps `std::sync::mpsc::sync_channel` with a byte-accounting +//! `Mutex + Condvar` cap. Sender blocks (cooperatively) when +//! `used_bytes + item.byte_size() > capacity_bytes`. Receiver +//! decrements `used_bytes` when it takes the item. +//! +//! Why: the existing producer→consumer channel between `DiscStream` +//! (PES producer) and `MuxSink` (PES consumer) is bounded by frame +//! count. Frame sizes vary 100× between metadata and keyframes, so a +//! count-based cap either starves on small frames or buffers far too +//! much memory on big ones. Byte-sized accounting sizes the buffer for +//! the worst-case input stall (NFS read p99 ≈ 1–2 s × ~15 MB/s peak +//! compressed bitrate ≈ ~30 MB) directly. +//! +//! The underlying mpsc channel is created with a very large slot count +//! so the byte cap (not the slot count) is the real backpressure. Slot +//! count is only there to give the kernel a small chunk to wake on. +//! +//! See `(internal)/memory/project_buffering_architecture.md` § +//! Pipeline channel — sizing. + +use std::sync::mpsc::{Receiver as MpscReceiver, RecvError, SendError, SyncSender, sync_channel}; +use std::sync::{Arc, Condvar, Mutex}; + +/// Default byte cap for the muxer's input channel. Sized to hide a +/// worst-case ~2 s NFS read refill at UHD peak compressed bitrate +/// (~15 MB/s); 64 MiB gives headroom. Tweakable; not magic. +pub const BYTE_CHANNEL_DEFAULT_CAPACITY: usize = 64 * 1024 * 1024; + +/// Slot capacity of the inner `sync_channel`. Large so the byte cap is +/// the real backpressure mechanism — the mpsc slot count only exists +/// to give the kernel a chunk to wake on. PES frames are typically +/// ~700 B each, so 64 MiB ≈ 90 k frames; 200 k is comfortable headroom. +const INNER_SLOT_CAPACITY: usize = 200_000; + +/// Anything whose in-memory cost can be accounted by a single +/// `usize`. Implement on the item type sent through [`Sender`]. +pub trait HasByteSize { + /// Bytes this item contributes to the channel's used budget. + /// Must be > 0 to make progress (a 0-byte item would never + /// block the sender no matter the cap; see send_blocks_at_capacity + /// test). + fn byte_size(&self) -> usize; +} + +impl HasByteSize for crate::pes::PesFrame { + fn byte_size(&self) -> usize { + // Frame data + the fixed header overhead the serializer + // writes (track + pts + keyframe + len). The `Vec` heap + // allocation also has alloc-header overhead but that's + // <0.1 % at typical frame sizes — folding it in would just + // add noise to the budget. + self.data.len() + 14 + } +} + +/// Shared book-keeping between [`Sender`] and [`Receiver`]. Wrapped in +/// an `Arc` because both halves hold it independently. +struct Accounting { + used: Mutex, + cv: Condvar, + capacity: usize, +} + +/// Send half of the byte-bounded channel. +/// +/// `send` blocks (on a `Condvar`) when adding the item would push +/// `used_bytes` past `capacity_bytes`. Unblocks when the receiver +/// `recv`s items out and notifies. Returns `Err(item)` if the +/// receiver has been dropped — mirrors `mpsc::SyncSender::send`. +pub struct Sender { + tx: SyncSender, + acct: Arc, +} + +impl Clone for Sender { + fn clone(&self) -> Self { + Sender { + tx: self.tx.clone(), + acct: self.acct.clone(), + } + } +} + +impl Sender { + /// Push one item. Blocks until adding it would not exceed the + /// capacity, then sends through the inner mpsc channel. + pub fn send(&self, item: T) -> Result<(), SendError> { + let sz = item.byte_size(); + // Reserve capacity first. The reservation is observable to + // other senders via `used`; only after we win the slot do we + // hand the item to the inner mpsc channel. That ordering means + // `used` is always a conservative upper bound on what's in the + // mpsc queue + about-to-be-sent. + { + let mut used = self.acct.used.lock().expect("byte_channel poisoned"); + // An item bigger than the whole capacity will never fit; let + // it through anyway as a one-shot reservation, otherwise the + // sender deadlocks forever waiting for `used == 0` AND + // nothing in flight. The receiver will drain it on the + // other side. Same behaviour as `std::sync::mpsc` for + // arbitrarily large messages. + while *used + sz > self.acct.capacity && *used > 0 { + used = self.acct.cv.wait(used).expect("byte_channel cv poisoned"); + } + *used += sz; + } + match self.tx.send(item) { + Ok(()) => Ok(()), + Err(SendError(returned)) => { + // Receiver dropped — refund the reservation so a later + // sender on a clone doesn't observe phantom used bytes + // (the receiver is gone so nobody will decrement). + let mut used = self.acct.used.lock().expect("byte_channel poisoned"); + *used = used.saturating_sub(sz); + self.acct.cv.notify_all(); + Err(SendError(returned)) + } + } + } +} + +/// Receive half of the byte-bounded channel. +/// +/// `recv` blocks on the inner mpsc until an item is available, then +/// decrements the byte-accounting and wakes any sender waiting on +/// capacity. +pub struct Receiver { + rx: MpscReceiver, + acct: Arc, +} + +impl Receiver { + /// Take the next item. Returns `Err(RecvError)` when all senders + /// have been dropped and the channel is empty. + pub fn recv(&self) -> Result { + let item = self.rx.recv()?; + let sz = item.byte_size(); + let mut used = self.acct.used.lock().expect("byte_channel poisoned"); + *used = used.saturating_sub(sz); + // Notify all so multi-sender setups wake every blocked sender, + // not just one. Wasted wakeups are cheap; missed wakeups would + // be a deadlock. + self.acct.cv.notify_all(); + Ok(item) + } +} + +/// Create a byte-bounded channel with the given capacity in bytes. +/// Returns a `(Sender, Receiver)` pair; clone the `Sender` for +/// multi-producer setups. +pub fn channel(capacity_bytes: usize) -> (Sender, Receiver) { + let (tx, rx) = sync_channel::(INNER_SLOT_CAPACITY); + let acct = Arc::new(Accounting { + used: Mutex::new(0), + cv: Condvar::new(), + capacity: capacity_bytes, + }); + ( + Sender { + tx, + acct: acct.clone(), + }, + Receiver { rx, acct }, + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::thread; + use std::time::{Duration, Instant}; + + /// Test payload — its `byte_size` returns whatever we passed at + /// construction so capacity math is exact and predictable. + #[derive(Clone, Debug, PartialEq, Eq)] + struct Item { + sz: usize, + tag: u32, + } + + impl HasByteSize for Item { + fn byte_size(&self) -> usize { + self.sz + } + } + + #[test] + fn send_recv_round_trip() { + let (tx, rx) = channel::(1024); + for i in 0..5 { + tx.send(Item { sz: 100, tag: i }).unwrap(); + } + for i in 0..5 { + let got = rx.recv().unwrap(); + assert_eq!(got, Item { sz: 100, tag: i }); + } + } + + #[test] + fn byte_accounting_decrements_on_recv() { + // Internal book-keeping check via observable side-effect: after + // sending K items totalling N bytes and receiving them all, a + // subsequent send of an N-byte item must NOT block (no items + // in flight, all capacity refunded). + let (tx, rx) = channel::(1024); + for _ in 0..4 { + tx.send(Item { sz: 256, tag: 0 }).unwrap(); + } + for _ in 0..4 { + rx.recv().unwrap(); + } + // Cap is now fully available again. Send a 1024-byte item; the + // `used > 0` guard means it goes through alone (no wait). + let start = Instant::now(); + tx.send(Item { sz: 1024, tag: 99 }).unwrap(); + assert!(start.elapsed() < Duration::from_millis(100)); + let got = rx.recv().unwrap(); + assert_eq!(got.tag, 99); + } + + #[test] + fn send_blocks_at_capacity_unblocks_on_recv() { + // Cap = 200 bytes, item = 100 bytes. First two sends fit + // exactly; the third must block until a recv frees capacity. + let (tx, rx) = channel::(200); + tx.send(Item { sz: 100, tag: 0 }).unwrap(); + tx.send(Item { sz: 100, tag: 1 }).unwrap(); + + let tx2 = tx.clone(); + let sent_at = Arc::new(Mutex::new(None::)); + let sent_at2 = sent_at.clone(); + let h = thread::spawn(move || { + tx2.send(Item { sz: 100, tag: 2 }).unwrap(); + *sent_at2.lock().unwrap() = Some(Instant::now()); + }); + + // Give the sender thread a head start; it should be parked in + // `cv.wait` because used (200) + 100 > capacity (200). + thread::sleep(Duration::from_millis(100)); + assert!( + sent_at.lock().unwrap().is_none(), + "third send should be blocked at capacity" + ); + + // Drain one. Sender wakes and completes. + let recv_at = Instant::now(); + let got = rx.recv().unwrap(); + assert_eq!(got.tag, 0); + h.join().unwrap(); + + let sent_when = sent_at.lock().unwrap().unwrap(); + assert!( + sent_when >= recv_at, + "sender must complete AFTER receiver freed capacity" + ); + + // Drain the remaining two. + assert_eq!(rx.recv().unwrap().tag, 1); + assert_eq!(rx.recv().unwrap().tag, 2); + } + + #[test] + fn item_larger_than_capacity_still_goes_through() { + // Pathological case: a single item bigger than the capacity. + // The guard `*used > 0` lets it through when the channel is + // empty (otherwise the sender deadlocks forever). Matches + // `mpsc::SyncSender` semantics for oversize messages. + let (tx, rx) = channel::(100); + tx.send(Item { sz: 1000, tag: 7 }).unwrap(); + let got = rx.recv().unwrap(); + assert_eq!(got, Item { sz: 1000, tag: 7 }); + } + + #[test] + fn concurrent_send_recv_stress() { + // 4 sender threads × 1k items each, 1 receiver. Verify byte + // accounting stays sane (channel never deadlocks, every item + // arrives exactly once) under contention. + const SENDERS: u32 = 4; + const PER_SENDER: u32 = 1000; + const TOTAL: u32 = SENDERS * PER_SENDER; + + let (tx, rx) = channel::(8 * 1024); + let sent = Arc::new(AtomicUsize::new(0)); + let mut handles = Vec::new(); + for s in 0..SENDERS { + let tx = tx.clone(); + let sent = sent.clone(); + handles.push(thread::spawn(move || { + for i in 0..PER_SENDER { + // Vary item size so accounting actually has to + // multiplex differently-sized blockers. 1B → 256B. + let sz = 1 + ((i as usize) % 256); + tx.send(Item { + sz, + tag: s * PER_SENDER + i, + }) + .unwrap(); + sent.fetch_add(1, Ordering::SeqCst); + } + })); + } + // Drop our local sender so the receiver can eventually see + // RecvError once all sender clones are done. Cloning the + // sender into each producer means each clone Drop'd separately. + drop(tx); + + let mut received = 0u32; + while let Ok(_item) = rx.recv() { + received += 1; + } + for h in handles { + h.join().unwrap(); + } + assert_eq!(received, TOTAL); + assert_eq!(sent.load(Ordering::SeqCst) as u32, TOTAL); + } + + #[test] + fn send_after_recv_dropped_returns_err() { + let (tx, rx) = channel::(1024); + drop(rx); + let r = tx.send(Item { sz: 10, tag: 0 }); + assert!(r.is_err()); + } +} diff --git a/src/io/file_sector_source/linux.rs b/src/io/file_sector_source/linux.rs new file mode 100644 index 0000000..439affa --- /dev/null +++ b/src/io/file_sector_source/linux.rs @@ -0,0 +1,15 @@ +//! Linux: hint the kernel that this fd will be read sequentially so +//! readahead widens. `posix_fadvise(POSIX_FADV_SEQUENTIAL)` is a hint, +//! not a guarantee — the kernel still owns the policy decision. + +use std::fs::File; +use std::os::unix::io::AsRawFd; + +pub(super) fn hint_sequential(file: &File, _len_bytes: u64) { + // Best-effort: return value ignored. A fadvise failure has no + // user-observable consequence (reads still work, just without the + // widened readahead window). + unsafe { + libc::posix_fadvise(file.as_raw_fd(), 0, 0, libc::POSIX_FADV_SEQUENTIAL); + } +} diff --git a/src/io/file_sector_source/macos.rs b/src/io/file_sector_source/macos.rs new file mode 100644 index 0000000..0ef2350 --- /dev/null +++ b/src/io/file_sector_source/macos.rs @@ -0,0 +1,38 @@ +//! macOS: hint the kernel to prefetch a generous chunk. macOS has no +//! direct `POSIX_FADV_SEQUENTIAL` equivalent; the idiomatic hint is +//! `fcntl(F_RDADVISE, &radvisory)` describing the byte range you +//! intend to read soon. We point it at the whole file (clamped to a +//! ceiling so a multi-TB ISO doesn't ask the kernel to prefetch +//! everything at once). + +use std::fs::File; +use std::os::unix::io::AsRawFd; + +/// `F_RDADVISE` opcode — not in libc's named constants on all SDKs. +const F_RDADVISE: libc::c_int = 44; + +/// Cap on the byte length we pass to `F_RDADVISE`. Asking for a +/// multi-GB readahead window is counterproductive — the OS doesn't +/// have that much cache to throw at one fd. 64 MiB is generous for +/// our use case (sweep, mux) and matches the byte-channel cap so the +/// kernel's prefetch ≥ our app-level pipeline depth. +const RDADVISE_MAX_BYTES: i64 = 64 * 1024 * 1024; + +/// `radvisory` per ``. repr(C) layout is stable. +#[repr(C)] +struct RadAdvisory { + ra_offset: libc::off_t, + ra_count: libc::c_int, +} + +pub(super) fn hint_sequential(file: &File, len_bytes: u64) { + let bytes = (len_bytes as i64).min(RDADVISE_MAX_BYTES); + let mut ra = RadAdvisory { + ra_offset: 0, + ra_count: bytes as libc::c_int, + }; + // Best-effort. + 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 new file mode 100644 index 0000000..dfcf6b4 --- /dev/null +++ b/src/io/file_sector_source/mod.rs @@ -0,0 +1,393 @@ +//! [`FileSectorSource`] — read 2048-byte sectors from an ISO file on +//! disk, with an internal 32 MiB read-ahead buffer. +//! +//! ## Why the 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. +//! +//! 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). +//! +//! ## Access pattern assumption +//! +//! 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 +//! +//! 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. + +#[cfg(target_os = "linux")] +mod linux; +#[cfg(target_os = "macos")] +mod macos; +#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] +mod other; +#[cfg(target_os = "windows")] +mod windows; + +#[cfg(target_os = "linux")] +use linux as platform; +#[cfg(target_os = "macos")] +use macos as platform; +#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] +use other as platform; +#[cfg(target_os = "windows")] +use windows as platform; + +use std::fs::File; +use std::io::{Read, Seek, SeekFrom}; +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. +/// +/// Tweakable. Named const, not a magic number. +pub const READAHEAD_BUF_BYTES: usize = 32 * 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. +pub struct FileSectorSource { + file: File, + /// Total file size in sectors. Constant after construction; + /// surfaced via [`SectorSource::capacity_sectors`]. + capacity: u32, + /// Heap-allocated read-ahead buffer. Boxed slice (not `Vec`) so + /// the allocation is sized exactly once and the layout is + /// stable. Length is at most [`READAHEAD_BUF_BYTES`]. + buf: Box<[u8]>, + /// First LBA covered by `buf`. `buf_len_sectors == 0` means the + /// buffer is empty (e.g. immediately after construction or a + /// backward seek); the value of `buf_start_lba` is then + /// meaningless. + buf_start_lba: u32, + /// Number of valid sectors in `buf`. `[buf_start_lba, + /// buf_start_lba + buf_len_sectors)` is the half-open LBA range + /// the buffer currently caches. + buf_len_sectors: u32, +} + +impl FileSectorSource { + /// Open an existing ISO file for reading. Capacity is derived + /// from `metadata().len() / 2048`. Returns + /// [`Error::IsoTooLarge`] if the file would exceed the 32-bit + /// LBA address space (~8 TB). + /// + /// Issues the platform's "sequential access expected" hint on the + /// fd (Linux `posix_fadvise(SEQUENTIAL)`, macOS `fcntl(F_RDADVISE)`, + /// Windows TODO stub) so the kernel's readahead widens. + pub fn open(path: &Path) -> std::io::Result { + let file = File::open(path)?; + let len = file.metadata()?.len(); + let sectors = len / SECTOR_SIZE as u64; + if sectors > u32::MAX as u64 { + return Err(Error::IsoTooLarge { + path: path.to_string_lossy().into_owned(), + } + .into()); + } + let capacity = sectors as u32; + + // Best-effort sequential hint. Ignored on platforms without + // an equivalent primitive (or where the API exists but the + // 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, + }) + } + + /// True if `[lba, lba + count)` is wholly inside the current + /// buffer window. `count == 0` is vacuously true. + 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`. + 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 { + fn capacity_sectors(&self) -> u32 { + self.capacity + } + + fn read_sectors( + &mut self, + lba: u32, + count: u16, + out: &mut [u8], + _recovery: bool, + ) -> Result { + let count = count as u32; + let bytes = count as usize * SECTOR_SIZE; + debug_assert!( + out.len() >= bytes, + "FileSectorSource::read_sectors: out len {} < requested {}", + out.len(), + bytes + ); + if count == 0 { + return Ok(0); + } + // Refill if the requested range isn't entirely buffered. + // `buffer_covers` also handles the empty-buffer case + // (buf_len_sectors == 0). + if !self.buffer_covers(lba, count) { + // A request larger than the buffer itself can never fit; + // fall back to a one-shot direct pread for that pathological + // case so callers can't deadlock the source. + if count > BUF_SECTORS { + 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 out[..bytes]) + .map_err(|e| Error::IoError { source: e })?; + // Invalidate buffer state — we bypassed it, the + // window is no longer authoritative for this LBA. + self.buf_len_sectors = 0; + return Ok(bytes); + } + self.refill(lba)?; + } + + // Slice the buffer at the requested LBA's offset within it. + let off_sectors = (lba - self.buf_start_lba) as usize; + let off_bytes = off_sectors * SECTOR_SIZE; + out[..bytes].copy_from_slice(&self.buf[off_bytes..off_bytes + bytes]); + Ok(bytes) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use tempfile::tempdir; + + /// Build a deterministic ISO of `sectors` sectors where sector `n` + /// is filled with the byte pattern `((n & 0xff) as u8)`. Lets us + /// verify any sector by content alone. + fn make_iso(path: &std::path::Path, sectors: u32) { + let mut f = std::fs::File::create(path).unwrap(); + let mut chunk = vec![0u8; SECTOR_SIZE]; + for n in 0..sectors { + let b = (n & 0xff) as u8; + chunk.iter_mut().for_each(|c| *c = b); + f.write_all(&chunk).unwrap(); + } + f.flush().unwrap(); + } + + #[test] + fn sequential_reads_match_file() { + // Two full buffer windows + a tail = exercise refill across + // boundaries. + let total = BUF_SECTORS * 2 + 17; + let dir = tempdir().unwrap(); + let path = dir.path().join("seq.iso"); + make_iso(&path, total); + + let mut src = FileSectorSource::open(&path).unwrap(); + assert_eq!(src.capacity_sectors(), total); + + let mut got = vec![0u8; SECTOR_SIZE]; + for lba in 0..total { + src.read_sectors(lba, 1, &mut got, false).unwrap(); + let expected = (lba & 0xff) as u8; + assert!( + got.iter().all(|b| *b == expected), + "sector {lba} content mismatch: expected 0x{expected:02x}" + ); + } + } + + #[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; + 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 to populate buffer #0. + let mut got = vec![0u8; SECTOR_SIZE]; + src.read_sectors(0, 1, &mut got, false).unwrap(); + assert_eq!(src.buf_start_lba, 0); + assert!(src.buf_len_sectors > 0); + + // Now read 4 sectors crossing the buffer boundary at + // BUF_SECTORS - 2 → BUF_SECTORS + 1. Spans the refill. + let span_lba = BUF_SECTORS - 2; + let mut buf4 = vec![0u8; SECTOR_SIZE * 4]; + src.read_sectors(span_lba, 4, &mut buf4, false).unwrap(); + for i in 0..4 { + let lba = span_lba + i as u32; + let expected = (lba & 0xff) as u8; + for b in &buf4[i * SECTOR_SIZE..(i + 1) * SECTOR_SIZE] { + assert_eq!(*b, expected, "byte mismatch at sub-sector {i}"); + } + } + } + + #[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; + let dir = tempdir().unwrap(); + let path = dir.path().join("back.iso"); + make_iso(&path, total); + + 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) + .unwrap(); + let start_after_forward = src.buf_start_lba; + assert!(start_after_forward >= BUF_SECTORS); + + // Backward to sector 0. The current buffer doesn't cover it + // → refill must happen. + src.read_sectors(0, 1, &mut got, false).unwrap(); + assert_eq!(src.buf_start_lba, 0); + 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. + let total: u32 = 100; + assert!(total < BUF_SECTORS); + let dir = tempdir().unwrap(); + let path = dir.path().join("small.iso"); + make_iso(&path, total); + + let mut src = FileSectorSource::open(&path).unwrap(); + assert_eq!(src.capacity_sectors(), total); + + let mut got = vec![0u8; SECTOR_SIZE]; + // First read triggers refill clamped to `total`. + src.read_sectors(0, 1, &mut got, false).unwrap(); + assert_eq!(src.buf_len_sectors, total); + + // Read the very last sector — still inside the buffer. + 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; + let dir = tempdir().unwrap(); + let path = dir.path().join("over.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_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; + let last_off = (req as usize - 1) * SECTOR_SIZE; + assert!( + big[last_off..last_off + SECTOR_SIZE] + .iter() + .all(|b| *b == exp) + ); + } +} diff --git a/src/io/file_sector_source/other.rs b/src/io/file_sector_source/other.rs new file mode 100644 index 0000000..b6e4f65 --- /dev/null +++ b/src/io/file_sector_source/other.rs @@ -0,0 +1,7 @@ +//! Fallback for targets without a known sequential-readahead hint +//! (BSDs, illumos, etc.). No-op — reads still work, they just don't +//! get the OS-level prefetch widening. + +use std::fs::File; + +pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {} diff --git a/src/io/file_sector_source/windows.rs b/src/io/file_sector_source/windows.rs new file mode 100644 index 0000000..6e4b036 --- /dev/null +++ b/src/io/file_sector_source/windows.rs @@ -0,0 +1,19 @@ +//! Windows: the canonical sequential-access hint is +//! `FILE_FLAG_SEQUENTIAL_SCAN` passed to `CreateFile` at open time — +//! it cannot be set after the fact via `SetFileInformationByHandle`. +//! Routing the open call through this module would mean a custom +//! `File::from_raw_handle` plumb for every `FileSectorSource::open` +//! caller, which is more invasive than the Phase 1 scope. +//! +//! TODO: replumb `FileSectorSource::open` to take an +//! `OpenOptions`-style builder so the Windows path can flip the flag +//! at open time. For now this is a no-op stub. + +use std::fs::File; + +pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) { + tracing::debug!( + target: "mux", + "FileSectorSource hint_sequential: windows stub (TODO: FILE_FLAG_SEQUENTIAL_SCAN at open)" + ); +} diff --git a/src/io/mod.rs b/src/io/mod.rs index 348cd98..c153013 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -10,11 +10,22 @@ //! no-op pipeline — their default cache policies have not been shown //! 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. +//! //! `Pipeline` + `Sink` (0.18) is the generic producer/consumer primitive //! used by sweep, patch, and mux to overlap reads with writes via a //! bounded channel + dedicated consumer thread. +//! +//! `byte_channel` is a byte-sized producer/consumer channel for the +//! mux pipeline, sized to absorb worst-case input read stalls (see +//! `(internal)/memory/project_buffering_architecture.md`). pub(crate) mod bounded; +pub mod byte_channel; +pub mod file_sector_source; mod writeback; mod writeback_file; diff --git a/src/io/writeback_file/linux.rs b/src/io/writeback_file/linux.rs new file mode 100644 index 0000000..f7204a1 --- /dev/null +++ b/src/io/writeback_file/linux.rs @@ -0,0 +1,66 @@ +//! Linux platform impl for [`super::WritebackFile`]. +//! +//! - `preallocate`: `fallocate(FALLOC_FL_KEEP_SIZE)` — reserve extents +//! without growing the reported file size. Reduces extent +//! fragmentation on large sequential writes (mux output on NFS in +//! particular). +//! - `durable_sync`: `fsync` wrapped in +//! [`crate::io::bounded::bounded_syscall`] with a 60 s deadline so a +//! wedged NFS server can't trap the calling thread indefinitely. + +use std::fs::File; +use std::io; +use std::os::unix::io::AsRawFd; +use std::time::Duration; + +/// Pre-reserve extents for `size_bytes` of upcoming sequential writes. +/// Best-effort: a non-zero rc is logged but not propagated, since the +/// caller would just continue with the unreserved file anyway. +pub(super) fn preallocate(file: &File, size_bytes: u64) { + // FALLOC_FL_KEEP_SIZE = 0x01 — keep the reported file size at 0 + // (writes grow it normally) while still pre-reserving the extents. + let rc = unsafe { + libc::fallocate( + file.as_raw_fd(), + libc::FALLOC_FL_KEEP_SIZE, + 0, + size_bytes as i64, + ) + }; + tracing::debug!( + target: "mux", + "WritebackFile fallocate size_hint={size_bytes} rc={rc} ok={}", + rc == 0 + ); +} + +/// Run `fsync` on `file` with a 60 s deadline. On timeout we log loudly +/// and return `Ok(())` — the kernel will still flush on close, so the +/// data is best-effort durable; the alternative (trap the thread for +/// the rest of the rip) defeats `/api/stop`. +pub(super) fn durable_sync(file: &File) -> io::Result<()> { + let fd = file.as_raw_fd(); + match crate::io::bounded::bounded_syscall( + None, + Duration::from_secs(60), + move || -> io::Result<()> { + let rc = unsafe { libc::fsync(fd) }; + if rc == 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) + } + }, + ) { + Ok(inner) => inner, + Err(crate::io::bounded::BoundedError::Timeout) => { + tracing::error!( + target: "mux", + "WritebackFile::sync_all fsync timed out after 60s; kernel will flush on close (best-effort)" + ); + Ok(()) + } + Err(crate::io::bounded::BoundedError::Halted) => Ok(()), + Err(crate::io::bounded::BoundedError::WorkerLost) => Ok(()), + } +} diff --git a/src/io/writeback_file/macos.rs b/src/io/writeback_file/macos.rs new file mode 100644 index 0000000..4959298 --- /dev/null +++ b/src/io/writeback_file/macos.rs @@ -0,0 +1,104 @@ +//! macOS platform impl for [`super::WritebackFile`]. +//! +//! - `preallocate`: `fcntl(F_PREALLOCATE)` — macOS's fallocate-equiv. +//! Reserves a contiguous extent when possible, falling back to a +//! non-contiguous reservation if the FS can't satisfy it. Reported +//! file size is unchanged (`F_ALLOCATEALL` is not set, so allocation +//! is "best effort up to length"; growth happens via writes). +//! - `durable_sync`: `fcntl(F_FULLFSYNC)` wrapped in +//! [`crate::io::bounded::bounded_syscall`] with a 60 s deadline. +//! F_FULLFSYNC is HFS+/APFS's true-fsync (flushes the disk's own +//! write cache) — what `fsync` should have been on macOS. Falls back +//! to plain `fsync` if F_FULLFSYNC returns ENOTSUP. + +use std::fs::File; +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; + +/// `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, + fst_posmode: F_PEOFPOSMODE, + fst_offset: 0, + fst_length: size_bytes as libc::off_t, + fst_bytesalloc: 0, + }; + // First attempt: contiguous. + let mut rc = unsafe { libc::fcntl(file.as_raw_fd(), F_PREALLOCATE, &mut fst) }; + if rc == -1 { + // Fall back: drop the contiguous hint, allow scattered extents. + fst.fst_flags = F_ALLOCATEALL; + rc = unsafe { libc::fcntl(file.as_raw_fd(), F_PREALLOCATE, &mut fst) }; + } + tracing::debug!( + target: "mux", + "WritebackFile F_PREALLOCATE size_hint={size_bytes} rc={rc} bytes_allocated={} ok={}", + fst.fst_bytesalloc, + rc != -1 + ); +} + +pub(super) fn durable_sync(file: &File) -> io::Result<()> { + let fd = file.as_raw_fd(); + match crate::io::bounded::bounded_syscall( + None, + Duration::from_secs(60), + move || -> io::Result<()> { + // Try F_FULLFSYNC first. If it isn't supported on this + // filesystem (older HFS, some network mounts) fall back to + // plain fsync — better than nothing. + let rc = unsafe { libc::fcntl(fd, F_FULLFSYNC, 0) }; + if rc == 0 { + return Ok(()); + } + let err = io::Error::last_os_error(); + if err.raw_os_error() == Some(libc::ENOTSUP) { + let rc = unsafe { libc::fsync(fd) }; + if rc == 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) + } + } else { + Err(err) + } + }, + ) { + Ok(inner) => inner, + Err(crate::io::bounded::BoundedError::Timeout) => { + tracing::error!( + target: "mux", + "WritebackFile::sync_all F_FULLFSYNC timed out after 60s; kernel will flush on close (best-effort)" + ); + Ok(()) + } + Err(crate::io::bounded::BoundedError::Halted) => Ok(()), + Err(crate::io::bounded::BoundedError::WorkerLost) => Ok(()), + } +} diff --git a/src/io/writeback_file.rs b/src/io/writeback_file/mod.rs similarity index 71% rename from src/io/writeback_file.rs rename to src/io/writeback_file/mod.rs index 4702e05..aeb533e 100644 --- a/src/io/writeback_file.rs +++ b/src/io/writeback_file/mod.rs @@ -20,6 +20,34 @@ //! //! See `super::writeback::linux` for the underlying pathology and the //! strategy. +//! +//! ## Platform split +//! +//! The platform-specific pieces of this wrapper — extent preallocation +//! (Linux `fallocate(KEEP_SIZE)`, macOS `F_PREALLOCATE`, Windows +//! `SetFileValidData`) and the durable-flush primitive (Linux/macOS +//! `fsync`/`F_FULLFSYNC` wrapped in a bounded syscall, Windows +//! `FlushFileBuffers`) — live in per-OS sibling modules. The dispatch +//! happens once at the bottom of this file via cfg-gated `mod` decls. +//! No inline `#[cfg(target_os = "...")]` in the business-logic above. + +#[cfg(target_os = "linux")] +mod linux; +#[cfg(target_os = "macos")] +mod macos; +#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] +mod other; +#[cfg(target_os = "windows")] +mod windows; + +#[cfg(target_os = "linux")] +use linux as platform; +#[cfg(target_os = "macos")] +use macos as platform; +#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] +use other as platform; +#[cfg(target_os = "windows")] +use windows as platform; use std::fs::{File, OpenOptions}; use std::io::{self, Seek, SeekFrom, Write}; @@ -65,43 +93,20 @@ impl WritebackFile { } /// Like [`Self::create`] but pre-reserves `size_bytes` of disk - /// space via `fallocate(FALLOC_FL_KEEP_SIZE)` on Linux. The - /// reported file size is unchanged (writes still grow the file - /// naturally) — only the on-disk extent allocation is preallocated, - /// which reduces extent fragmentation on large sequential writes - /// (mux output, especially on slow storage / NFS). + /// space via the platform's extent-preallocation primitive (Linux + /// `fallocate(KEEP_SIZE)`, macOS `F_PREALLOCATE`, Windows + /// `SetFileValidData` stub). The reported file size is unchanged + /// (writes still grow the file naturally) — only the on-disk extent + /// allocation is preallocated, which reduces extent fragmentation + /// on large sequential writes (mux output, especially on slow + /// storage / NFS). /// - /// On macOS / Windows the size hint is ignored and this is - /// equivalent to `create`. + /// On platforms without an extent-preallocation primitive this is + /// equivalent to `create` — the size hint is dropped after a debug + /// log. pub(crate) fn create_with_size_hint(path: &Path, size_bytes: u64) -> io::Result { let file = File::create(path)?; - #[cfg(target_os = "linux")] - { - use std::os::unix::io::AsRawFd; - // FALLOC_FL_KEEP_SIZE = 0x01 — keep the reported file - // size at 0 (writes grow it normally) while still - // pre-reserving the extents. - let rc = unsafe { - libc::fallocate( - file.as_raw_fd(), - libc::FALLOC_FL_KEEP_SIZE, - 0, - size_bytes as i64, - ) - }; - tracing::debug!( - target: "mux", - "WritebackFile fallocate size_hint={size_bytes} rc={rc} ok={}", - rc == 0 - ); - } - #[cfg(not(target_os = "linux"))] - { - tracing::debug!( - target: "mux", - "WritebackFile fallocate size_hint={size_bytes} skipped (non-linux)" - ); - } + platform::preallocate(&file, size_bytes); Self::new(file) } @@ -118,7 +123,8 @@ impl WritebackFile { /// place of `File::sync_all`. /// /// The final fsync is wrapped in - /// [`crate::io::bounded::bounded_syscall`] with a 60 s deadline. + /// [`crate::io::bounded::bounded_syscall`] with a 60 s deadline on + /// platforms that have a usable bounded primitive (Linux + macOS). /// fsync on a wedged NFS server (or a degraded local disk) can /// hang the calling thread; the wrapper ensures the worst case is /// 60 s + log-and-continue rather than indefinite. On timeout the @@ -126,39 +132,7 @@ impl WritebackFile { /// best effort, but bounded. pub(crate) fn sync_all(&mut self) -> io::Result<()> { self.pipeline.finalize(); - #[cfg(unix)] - { - use std::os::unix::io::AsRawFd; - use std::time::Duration; - let fd = self.file.as_raw_fd(); - match crate::io::bounded::bounded_syscall( - None, - Duration::from_secs(60), - move || -> io::Result<()> { - let rc = unsafe { libc::fsync(fd) }; - if rc == 0 { - Ok(()) - } else { - Err(io::Error::last_os_error()) - } - }, - ) { - Ok(inner) => inner, - Err(crate::io::bounded::BoundedError::Timeout) => { - tracing::error!( - target: "mux", - "WritebackFile::sync_all fsync timed out after 60s; kernel will flush on close (best-effort)" - ); - Ok(()) - } - Err(crate::io::bounded::BoundedError::Halted) => Ok(()), - Err(crate::io::bounded::BoundedError::WorkerLost) => Ok(()), - } - } - #[cfg(not(unix))] - { - self.file.sync_all() - } + platform::durable_sync(&self.file) } } diff --git a/src/io/writeback_file/other.rs b/src/io/writeback_file/other.rs new file mode 100644 index 0000000..aee4013 --- /dev/null +++ b/src/io/writeback_file/other.rs @@ -0,0 +1,23 @@ +//! Fallback platform impl for [`super::WritebackFile`] on targets +//! without a dedicated implementation (BSDs, illumos, etc.). +//! +//! - `preallocate` is a logged no-op. +//! - `durable_sync` calls `File::sync_all` directly (no bounded-syscall +//! wrapper — the wrapper depends on Linux/macOS unix idioms that +//! aren't universally portable). If a future BSD impl needs the +//! 60-s deadline, it should land in its own per-OS file rather than +//! bloat this fallback. + +use std::fs::File; +use std::io; + +pub(super) fn preallocate(_file: &File, size_bytes: u64) { + tracing::debug!( + target: "mux", + "WritebackFile preallocate size_hint={size_bytes} skipped (no impl on this target)" + ); +} + +pub(super) fn durable_sync(file: &File) -> io::Result<()> { + file.sync_all() +} diff --git a/src/io/writeback_file/windows.rs b/src/io/writeback_file/windows.rs new file mode 100644 index 0000000..f1de892 --- /dev/null +++ b/src/io/writeback_file/windows.rs @@ -0,0 +1,32 @@ +//! Windows platform impl for [`super::WritebackFile`]. +//! +//! TODO: this stub matches the design's "validate without a Windows +//! build env, leave a stub" carve-out. The real impl should use: +//! +//! - `SetEndOfFile` + `SetFileValidData` for extent preallocation +//! (caller needs `SE_MANAGE_VOLUME_NAME` privilege; if unavailable +//! fall back to a write-zero path or just skip). +//! - `FlushFileBuffers` for fsync-equivalent durable flush. +//! +//! Until then: preallocate is a debug-logged no-op; durable_sync calls +//! the std `File::sync_all` (which on Windows maps to +//! `FlushFileBuffers` internally). + +use std::fs::File; +use std::io; + +pub(super) fn preallocate(_file: &File, size_bytes: u64) { + tracing::debug!( + target: "mux", + "WritebackFile preallocate size_hint={size_bytes} skipped (windows stub; TODO: SetFileValidData)" + ); +} + +pub(super) fn durable_sync(file: &File) -> io::Result<()> { + // `File::sync_all` on Windows is `FlushFileBuffers`. Acceptable + // for now; the bounded-syscall wrapper is not used here because + // the stub also skips the worker-thread + leak machinery (the + // wrapper would need an `unsafe impl Send` for `RawHandle`, and + // designing that without a Windows test env is asking for it). + file.sync_all() +} diff --git a/src/sector/file.rs b/src/sector/file.rs index 520118e..485c4f5 100644 --- a/src/sector/file.rs +++ b/src/sector/file.rs @@ -1,148 +1,19 @@ -//! File-backed sector I/O — read and write 2048-byte sectors against -//! an ISO image on disk. +//! File-backed sector sink — write 2048-byte sectors to an ISO image +//! on disk. //! -//! [`FileSectorSource`] is the read side (open-only). [`FileSectorSink`] -//! is the write side (create or open-rw); writes go through -//! [`crate::io::WritebackFile`] so big sequential ISO writes share -//! the same bounded-cache writeback pipeline used by sweep / patch / -//! mux. +//! The read-side counterpart ([`crate::io::file_sector_source::FileSectorSource`]) +//! lives under `io/` because its internals (read-ahead buffer, per-OS +//! `fadvise`/`F_RDADVISE` hints) are I/O infrastructure rather than +//! sector-trait business logic. Both types remain re-exported at +//! [`crate::sector`] for ergonomic imports. -use std::fs::{File, OpenOptions}; -use std::io::{Read, Seek, SeekFrom, Write}; +use std::fs::OpenOptions; +use std::io::{Seek, SeekFrom, Write}; use std::path::Path; use crate::error::{Error, Result}; -use super::{SectorSink, SectorSource}; - -/// Bytes-read threshold per `posix_fadvise(DONTNEED)` drop on the -/// read side. Mirrors the writeback chunk size so the read-side -/// page cache stays bounded the same way the write side does. -#[cfg(target_os = "linux")] -const READ_DROP_CHUNK_BYTES: u64 = 32 * 1024 * 1024; - -/// SectorSource backed by a file (ISO image). -/// -/// Seeks to `lba * 2048`, reads `count * 2048` bytes per call. The -/// file is held directly: every `read_sectors` call performs an -/// absolute seek, so a wrapping `BufReader` would have its buffer -/// invalidated on every call (its internal cursor moves with the -/// `Seek` impl) — pure overhead. Callers that benefit from buffered -/// reads should compose their own `BufReader` at the `read_sectors` -/// granularity they care about. -pub struct FileSectorSource { - file: File, - capacity: u32, - /// Bytes read since the last `posix_fadvise(DONTNEED)` drop. - /// Only updated on Linux; on other targets it stays at 0. - #[cfg(target_os = "linux")] - bytes_read_since_drop: u64, - /// Byte offset at which the current drop window starts (the - /// next `posix_fadvise(DONTNEED)` call drops from here). - #[cfg(target_os = "linux")] - drop_window_start: u64, -} - -impl FileSectorSource { - /// Open an existing ISO file for reading. Capacity is derived - /// from `metadata().len() / 2048`. Returns - /// [`Error::IsoTooLarge`] if the file would exceed the 32-bit - /// LBA address space (~8 TB). - pub fn open(path: &Path) -> std::io::Result { - let file = File::open(path)?; - let len = file.metadata()?.len(); - let sectors = len / 2048; - if sectors > u32::MAX as u64 { - return Err(Error::IsoTooLarge { - path: path.to_string_lossy().into_owned(), - } - .into()); - } - let capacity = sectors as u32; - - // Hint sequential access on Linux so the kernel's readahead - // window widens for the ISO sweep. Best-effort: return value - // is ignored. On macOS / Windows this is a no-op. - #[cfg(target_os = "linux")] - { - use std::os::unix::io::AsRawFd; - unsafe { - libc::posix_fadvise(file.as_raw_fd(), 0, 0, libc::POSIX_FADV_SEQUENTIAL); - } - } - - Ok(Self { - file, - capacity, - #[cfg(target_os = "linux")] - bytes_read_since_drop: 0, - #[cfg(target_os = "linux")] - drop_window_start: 0, - }) - } -} - -// Implement the legacy `SectorSource` trait. The blanket impl in -// `super` produces the `SectorSource` impl automatically — no need -// to write both, and writing both would conflict. This keeps the -// 0.17 method-resolution path intact (callers with `SectorSource` -// in scope can still write `fsr.read_sectors(..)` against a -// `FileSectorSource`). -impl SectorSource for FileSectorSource { - fn capacity_sectors(&self) -> u32 { - self.capacity - } - - fn read_sectors( - &mut self, - lba: u32, - count: u16, - buf: &mut [u8], - _recovery: bool, - ) -> Result { - let offset = lba as u64 * 2048; - let bytes = count as usize * 2048; - self.file - .seek(SeekFrom::Start(offset)) - .map_err(|e| Error::IoError { source: e })?; - self.file - .read_exact(&mut buf[..bytes]) - .map_err(|e| Error::IoError { source: e })?; - - // On Linux, periodically drop the just-read region from the - // page cache to keep cache pressure bounded during multi-GB - // sequential ISO reads. Mirrors the write-side pipeline. - #[cfg(target_os = "linux")] - { - use std::os::unix::io::AsRawFd; - self.bytes_read_since_drop += bytes as u64; - if self.bytes_read_since_drop >= READ_DROP_CHUNK_BYTES { - let drop_start = self.drop_window_start; - let drop_len = self.bytes_read_since_drop; - let t0 = std::time::Instant::now(); - unsafe { - libc::posix_fadvise( - self.file.as_raw_fd(), - drop_start as i64, - drop_len as i64, - libc::POSIX_FADV_DONTNEED, - ); - } - let elapsed_ms = t0.elapsed().as_millis(); - let start_lba = drop_start / 2048; - let end_lba = (drop_start + drop_len) / 2048; - tracing::trace!( - target: "mux", - "FileSectorSource fadvise DONTNEED lba=[{start_lba}..{end_lba}) bytes={drop_len} elapsed_ms={elapsed_ms}" - ); - self.drop_window_start = drop_start + drop_len; - self.bytes_read_since_drop = 0; - } - } - - Ok(bytes) - } -} +use super::SectorSink; /// SectorSink backed by a file (ISO image). /// @@ -211,13 +82,8 @@ impl SectorSink for FileSectorSink { #[cfg(test)] mod tests { - // Bring the 0.18 trait into scope (not super::*: the super - // module also re-exports the legacy `SectorSource`, and - // having both `SectorSource::read_sectors` and - // `SectorSource::read_sectors` visible would force every - // call site to disambiguate). External consumers see the - // same surface this test exercises. - use super::{FileSectorSink, FileSectorSource}; + use super::FileSectorSink; + use crate::io::file_sector_source::FileSectorSource; use crate::sector::{SectorSink, SectorSource}; use tempfile::tempdir; diff --git a/src/sector/mod.rs b/src/sector/mod.rs index 6c9b3ec..17e6f09 100644 --- a/src/sector/mod.rs +++ b/src/sector/mod.rs @@ -109,5 +109,6 @@ pub trait SectorSink: Send { fn finish(self: Box) -> Result<()>; } +pub use crate::io::file_sector_source::FileSectorSource; pub use decrypting::DecryptingSectorSource; -pub use file::{FileSectorSink, FileSectorSource}; +pub use file::FileSectorSink;