diff --git a/Cargo.toml b/Cargo.toml index 38c0c37..39577c8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "libfreemkv" -version = "0.20.3" +version = "0.20.4" edition = "2024" rust-version = "1.86" license = "AGPL-3.0-only" diff --git a/src/io/writeback/linux.rs b/src/io/writeback/linux.rs index 640e8b8..8ab31cc 100644 --- a/src/io/writeback/linux.rs +++ b/src/io/writeback/linux.rs @@ -14,9 +14,28 @@ //! a full chunk's worth of work to flush — the wait is near-instant. //! Dirty cache stays bounded at ~2 × `chunk_bytes` and writes drain //! continuously instead of in bursts. +//! +//! The chunk size is adaptive: we measure the elapsed time of the +//! `WAIT_AFTER` call over a rolling window of the last 16 chunks and +//! resize the chunk based on the p95. Slow storage (NFS, network +//! shares, HDD) sees larger chunks to amortise per-chunk overhead; +//! fast storage (NVMe) sees smaller chunks to keep cache pressure +//! tight. Bounds: [4 MiB, 256 MiB]. +use std::collections::VecDeque; use std::fs::File; use std::os::unix::io::{AsRawFd, RawFd}; +use std::time::Instant; + +const ADAPTIVE_WINDOW: usize = 16; +const CHUNK_BYTES_MIN: u64 = 4 * 1024 * 1024; +const CHUNK_BYTES_MAX: u64 = 256 * 1024 * 1024; +const ADAPTIVE_GROW_MS: u64 = 200; +const ADAPTIVE_SHRINK_MS: u64 = 20; +/// Every N chunks, emit a `debug!` snapshot of the current chunk +/// size so operators tailing the log can see where the autoscaler +/// settled. +const SIZE_LOG_INTERVAL: u64 = 32; pub(crate) struct WritebackPipeline { /// Aliases the wrapping `WritebackFile::file`. Only valid for the @@ -28,6 +47,11 @@ pub(crate) struct WritebackPipeline { chunk_bytes: u64, last_flush_pos: u64, pending: Option<(u64, u64)>, + /// Rolling window of recent `WAIT_AFTER` elapsed_ms measurements. + wait_after_window: VecDeque, + /// Count of chunks emitted (used to space out periodic + /// `debug!` size snapshots). + chunk_count: u64, } impl WritebackPipeline { @@ -41,6 +65,8 @@ impl WritebackPipeline { chunk_bytes, last_flush_pos: start_pos, pending: None, + wait_after_window: VecDeque::with_capacity(ADAPTIVE_WINDOW), + chunk_count: 0, } } @@ -51,27 +77,83 @@ impl WritebackPipeline { if pos < self.last_flush_pos.saturating_add(self.chunk_bytes) { return; } + let chunk_off = self.last_flush_pos as i64; + let chunk_len = (pos - self.last_flush_pos) as i64; + let mut wait_ms: u64 = 0; + let mut fadvise_ms: u64 = 0; unsafe { - let chunk_off = self.last_flush_pos as i64; - let chunk_len = (pos - self.last_flush_pos) as i64; libc::sync_file_range(self.fd, chunk_off, chunk_len, libc::SYNC_FILE_RANGE_WRITE); if let Some((prev_off, prev_len)) = self.pending.take() { + let t_wait = Instant::now(); libc::sync_file_range( self.fd, prev_off as i64, prev_len as i64, libc::SYNC_FILE_RANGE_WAIT_AFTER, ); + wait_ms = t_wait.elapsed().as_millis() as u64; + let t_fadv = Instant::now(); libc::posix_fadvise( self.fd, prev_off as i64, prev_len as i64, libc::POSIX_FADV_DONTNEED, ); + fadvise_ms = t_fadv.elapsed().as_millis() as u64; + self.record_wait(wait_ms); } self.pending = Some((chunk_off as u64, chunk_len as u64)); } self.last_flush_pos = pos; + self.chunk_count += 1; + tracing::trace!( + target: "mux", + "WritebackPipeline chunk off={} len={} sync_file_range_ms={wait_ms} fadvise_ms={fadvise_ms} chunk_bytes={}", + chunk_off, + chunk_len, + self.chunk_bytes + ); + if self.chunk_count % SIZE_LOG_INTERVAL == 0 { + tracing::debug!( + target: "mux", + "WritebackPipeline chunk_bytes={} after {} chunks", + self.chunk_bytes, + self.chunk_count + ); + } + } + + /// Push a new `WAIT_AFTER` measurement into the rolling window + /// and, if the window is full, adapt `chunk_bytes` based on p95. + fn record_wait(&mut self, wait_ms: u64) { + if self.wait_after_window.len() == ADAPTIVE_WINDOW { + self.wait_after_window.pop_front(); + } + self.wait_after_window.push_back(wait_ms); + if self.wait_after_window.len() < ADAPTIVE_WINDOW { + return; + } + // p95 of 16 samples ≈ sorted[14] (5 % of 16 = 0.8 ≈ 1 above). + let mut sorted: Vec = self.wait_after_window.iter().copied().collect(); + sorted.sort_unstable(); + let p95 = sorted[14]; + let old = self.chunk_bytes; + let new = if p95 > ADAPTIVE_GROW_MS && self.chunk_bytes < CHUNK_BYTES_MAX { + (self.chunk_bytes * 2).min(CHUNK_BYTES_MAX) + } else if p95 < ADAPTIVE_SHRINK_MS && self.chunk_bytes > CHUNK_BYTES_MIN { + (self.chunk_bytes / 2).max(CHUNK_BYTES_MIN) + } else { + self.chunk_bytes + }; + if new != old { + self.chunk_bytes = new; + tracing::info!( + target: "mux", + "WritebackPipeline adaptive chunk_bytes {} -> {} p95_ms={p95}", + old, + new + ); + } } /// Caller is about to seek away from the current write region. diff --git a/src/io/writeback_file.rs b/src/io/writeback_file.rs index 2d083b0..94097ac 100644 --- a/src/io/writeback_file.rs +++ b/src/io/writeback_file.rs @@ -54,11 +54,57 @@ impl WritebackFile { /// and wrap it. Convenience for the common /// `File::create(path)` + `WritebackFile::new(file)` pair so callers /// don't have to assemble a `File` first. + /// + /// Callers that know the target output size should prefer + /// [`Self::create_with_size_hint`] so the kernel can pre-reserve + /// extents. + #[allow(dead_code)] pub(crate) fn create(path: &Path) -> io::Result { let file = File::create(path)?; Self::new(file) } + /// 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). + /// + /// On macOS / Windows the size hint is ignored and this is + /// equivalent to `create`. + 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)" + ); + } + Self::new(file) + } + /// Open an existing file at `path` for writing (no truncation) and /// wrap it. Mirrors `File::open` semantics for the writable case /// — used by patch / resume paths that mutate an existing ISO in diff --git a/src/mux/resolve.rs b/src/mux/resolve.rs index a29ee6c..b7ac8f0 100644 --- a/src/mux/resolve.rs +++ b/src/mux/resolve.rs @@ -250,11 +250,13 @@ pub fn output( // writeback) so a UHD-scale MKV mux to slow / network-attached // staging doesn't hit the dirty-page burst pathology that // sweep already side-steps. BufWriter sits on top to coalesce - // mux's many small EBML element writes. + // mux's many small EBML element writes. Pre-reserve the + // target's worth of extents on Linux via fallocate(KEEP_SIZE) + // to reduce extent fragmentation during the mux. let writer: Box = Box::new(std::io::BufWriter::with_capacity( IO_BUF_SIZE, - crate::io::WritebackFile::create(path)?, + crate::io::WritebackFile::create_with_size_hint(path, title.size_bytes)?, )); Ok(Box::new(MkvStream::create(writer, title)?)) } @@ -262,7 +264,7 @@ pub fn output( validate_file_path(path, "m2ts")?; let writer = std::io::BufWriter::with_capacity( IO_BUF_SIZE, - crate::io::WritebackFile::create(path)?, + crate::io::WritebackFile::create_with_size_hint(path, title.size_bytes)?, ); Ok(Box::new(M2tsStream::create(writer, title)?)) } diff --git a/src/sector/file.rs b/src/sector/file.rs index 4b2a932..520118e 100644 --- a/src/sector/file.rs +++ b/src/sector/file.rs @@ -15,6 +15,12 @@ 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 @@ -27,6 +33,14 @@ use super::{SectorSink, SectorSource}; 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 { @@ -45,7 +59,26 @@ impl FileSectorSource { .into()); } let capacity = sectors as u32; - Ok(Self { file, capacity }) + + // 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, + }) } } @@ -75,6 +108,38 @@ impl SectorSource for FileSectorSource { 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) } }