From d42c7d17bef3dd96c71c4c56e1455288de95339e Mon Sep 17 00:00:00 2001 From: Matthew Jackson Date: Sat, 9 May 2026 08:53:17 -0700 Subject: [PATCH] =?UTF-8?q?0.18=20primitive:=20rename=20crate::io::Writer?= =?UTF-8?q?=20=E2=86=92=20WritebackFile?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The type's job is the bounded-cache writeback pipeline (sync_file_range + posix_fadvise(DONTNEED)) — not generic writing. The 0.17 name was ambiguous; reading `Writer::new(file)` gave no hint about what was special. New name makes the role obvious at every call site. Adds `WritebackFile::create(path)` and `WritebackFile::open(path)` constructors so callers don't have to assemble a `File` first. No alias kept; this is a clean 0.18 rename. See freemkv-private/memory/0_18_redesign.md. Single contributor: MattJackson. --- src/disc/mod.rs | 20 +++---- src/disc/sweep_pipeline.rs | 6 +- src/io/mod.rs | 22 ++++---- src/io/writeback_file.rs | 112 +++++++++++++++++++++++++++++++++++++ src/io/writer.rs | 84 ---------------------------- src/mux/resolve.rs | 12 ++-- 6 files changed, 141 insertions(+), 115 deletions(-) create mode 100644 src/io/writeback_file.rs delete mode 100644 src/io/writer.rs diff --git a/src/disc/mod.rs b/src/disc/mod.rs index c819cc9..3af4c11 100644 --- a/src/disc/mod.rs +++ b/src/disc/mod.rs @@ -1418,10 +1418,11 @@ impl Disc { f }; - // Wrap the raw `File` in our bounded-cache writer (drains - // dirty pages continuously instead of bursting; see - // `crate::io`). The Writer moves into the consumer thread. - let file = crate::io::Writer::new(file).map_err(|e| Error::IoError { source: e })?; + // Wrap the raw `File` in our bounded-cache `WritebackFile` + // (drains dirty pages continuously instead of bursting; see + // `crate::io`). The `WritebackFile` moves into the consumer + // thread. + let file = crate::io::WritebackFile::new(file).map_err(|e| Error::IoError { source: e })?; let batch: u16 = match opts.batch_sectors { Some(b) => b, None if opts.skip_on_error => ecc_sectors(self.format), @@ -1435,7 +1436,7 @@ impl Disc { // sweep finishes are the patch pass's job. let regions: Vec<(u64, u64)> = map.ranges_with(&[mapfile::SectorStatus::NonTried]); - // Spawn the consumer. It owns Writer + Mapfile; the producer + // Spawn the consumer. It owns WritebackFile + Mapfile; the producer // (this thread) keeps `reader`, `read_ctx`, halt + set_speed. let (work_tx, prog_rx, consumer_handle) = spawn_consumer(ConsumerInputs { file, @@ -1978,14 +1979,11 @@ impl Disc { let is_regular = std::fs::metadata(path) .map(|m| m.file_type().is_file()) .unwrap_or(false); - let file = std::fs::OpenOptions::new() - .write(true) - .open(path) - .map_err(|e| Error::IoError { source: e })?; - // Same bounded-cache writeback wrapper sweep uses, so patch's + // Same bounded-cache `WritebackFile` sweep uses, so patch's // recovery writes (sparse but can be many across a damaged region) // get the burst-flush protection on slow / NFS-backed staging. - let mut file = crate::io::Writer::new(file).map_err(|e| Error::IoError { source: e })?; + let mut file = + crate::io::WritebackFile::open(path).map_err(|e| Error::IoError { source: e })?; // Log ISO file size at patch start for write monitoring if let Ok(metadata) = std::fs::metadata(path) { diff --git a/src/disc/sweep_pipeline.rs b/src/disc/sweep_pipeline.rs index d406488..eff66c8 100644 --- a/src/disc/sweep_pipeline.rs +++ b/src/disc/sweep_pipeline.rs @@ -9,7 +9,7 @@ //! both costs. //! //! This module decouples them. A consumer thread owns the -//! [`crate::io::Writer`] (the ISO file) and the +//! [`crate::io::WritebackFile`] (the ISO file) and the //! [`super::mapfile::Mapfile`]. The producer thread (the caller of //! `Disc::sweep`) keeps the [`crate::sector::SectorReader`], the //! [`super::read_error`] state machine, and decrypt — so what enters @@ -112,7 +112,7 @@ pub(super) struct ConsumerSummary { /// happens on the producer side before send, so the consumer never /// sees keys. pub(super) struct ConsumerInputs { - pub file: crate::io::Writer, + pub file: crate::io::WritebackFile, pub map: Mapfile, /// `sync_all`-on-failure-is-an-error iff the output is a regular /// file. `/dev/null` and pipes always fail `sync_all`; that's not @@ -279,7 +279,7 @@ fn apply_item( .file .seek(SeekFrom::Start(pos)) .map_err(|e| Error::IoError { source: e })?; - // Subsequent writes are sequential; `crate::io::Writer`'s + // Subsequent writes are sequential; `crate::io::WritebackFile`'s // seek-elision keeps them on the writeback pipeline path. let mut filled = 0u64; while filled < len { diff --git a/src/io/mod.rs b/src/io/mod.rs index 68a4fd0..17c9266 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -1,16 +1,16 @@ //! File I/O helpers that bound kernel cache pressure on big writes. //! -//! `Writer` is a drop-in wrapper around `std::fs::File` for any call -//! site that performs large sequential writes (sweep, mux, etc.). It -//! implements `Write` and `Seek` so existing code paths can swap -//! `File` for `Writer` with no body changes. Internally it drives a -//! `WritebackPipeline` that, on Linux, drains dirty pages continuously -//! at 32 MB granularity to avoid the kernel's accumulate-then-burst -//! flush behaviour. macOS and Windows use a no-op pipeline — their -//! default cache policies have not been shown to exhibit the same -//! pathology for this access pattern. +//! `WritebackFile` is a drop-in wrapper around `std::fs::File` for any +//! call site that performs large sequential writes (sweep, patch, mux, +//! etc.). It implements `Write` and `Seek` so existing code paths can +//! swap `File` for `WritebackFile` with no body changes. Internally it +//! drives a `WritebackPipeline` that, on Linux, drains dirty pages +//! continuously at 32 MB granularity to avoid the kernel's +//! accumulate-then-burst flush behaviour. macOS and Windows use a +//! no-op pipeline — their default cache policies have not been shown +//! to exhibit the same pathology for this access pattern. mod writeback; -mod writer; +mod writeback_file; -pub(crate) use writer::Writer; +pub(crate) use writeback_file::WritebackFile; diff --git a/src/io/writeback_file.rs b/src/io/writeback_file.rs new file mode 100644 index 0000000..6266f70 --- /dev/null +++ b/src/io/writeback_file.rs @@ -0,0 +1,112 @@ +//! `WritebackFile` — a `File` wrapper whose reason for existing is the +//! bounded-cache writeback pipeline. +//! +//! Why: large sequential writes (sweep, patch, mux on UHD-scale output) +//! left to the kernel's default writeback policy accumulate hundreds of +//! megabytes of dirty pages and then burst-flush, stalling subsequent +//! writes for seconds at a time. `WritebackFile` drives a continuous +//! [`super::writeback::WritebackPipeline`] that on Linux issues +//! incremental `sync_file_range` + `posix_fadvise(DONTNEED)` calls at +//! 32 MB granularity so dirty pages drain at the same rate they're +//! produced. macOS and Windows fall through to a no-op pipeline — their +//! default cache policies have not been shown to exhibit the same +//! pathology for this access pattern. +//! +//! It implements `Write` and `Seek` so any call site that wrote to a +//! plain `File` through those traits (sweep, patch, mux) can swap in +//! `WritebackFile` without touching the body of the loop. The wrapper +//! also tracks the current file position to feed the pipeline with +//! progress + seek boundaries. +//! +//! See `super::writeback::linux` for the underlying pathology and the +//! strategy. + +use std::fs::{File, OpenOptions}; +use std::io::{self, Seek, SeekFrom, Write}; +use std::path::Path; + +use super::writeback::WritebackPipeline; + +const CHUNK_BYTES: u64 = 32 * 1024 * 1024; + +pub(crate) struct WritebackFile { + file: File, + pipeline: WritebackPipeline, + pos: u64, +} + +impl WritebackFile { + /// Wrap an open `File`. The current OS file position is queried + /// once so the pipeline starts tracking from wherever the file + /// already is (typically 0 for fresh files; non-zero for resumed + /// or appended files). + pub(crate) fn new(mut file: File) -> io::Result { + let pos = file.stream_position()?; + let pipeline = WritebackPipeline::new(&file, pos, CHUNK_BYTES); + Ok(Self { + file, + pipeline, + pos, + }) + } + + /// Create a new file at `path` (truncating any existing contents) + /// and wrap it. Convenience for the common + /// `File::create(path)` + `WritebackFile::new(file)` pair so callers + /// don't have to assemble a `File` first. + pub(crate) fn create(path: &Path) -> io::Result { + let file = File::create(path)?; + 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 + /// place. + pub(crate) fn open(path: &Path) -> io::Result { + let file = OpenOptions::new().write(true).open(path)?; + Self::new(file) + } + + /// Drain in-flight writeback then issue a full fsync. Use this in + /// place of `File::sync_all`. + pub(crate) fn sync_all(&mut self) -> io::Result<()> { + self.pipeline.finalize(); + self.file.sync_all() + } +} + +impl Write for WritebackFile { + fn write(&mut self, buf: &[u8]) -> io::Result { + let n = self.file.write(buf)?; + self.pos += n as u64; + self.pipeline.note_progress(self.pos); + Ok(n) + } + + fn write_all(&mut self, buf: &[u8]) -> io::Result<()> { + self.file.write_all(buf)?; + self.pos += buf.len() as u64; + self.pipeline.note_progress(self.pos); + Ok(()) + } + + fn flush(&mut self) -> io::Result<()> { + self.file.flush() + } +} + +impl Seek for WritebackFile { + fn seek(&mut self, from: SeekFrom) -> io::Result { + let p = self.file.seek(from)?; + // Only treat seeks that actually move the position as + // boundaries — sweep does a redundant `seek(Current(pos))` + // before every write, and we don't want that to drain the + // pipeline on every iteration. + if p != self.pos { + self.pipeline.handle_seek(p); + self.pos = p; + } + Ok(p) + } +} diff --git a/src/io/writer.rs b/src/io/writer.rs deleted file mode 100644 index 7844c0d..0000000 --- a/src/io/writer.rs +++ /dev/null @@ -1,84 +0,0 @@ -//! `Writer` — a drop-in `File` wrapper that keeps the kernel page -//! cache bounded during large sequential output. -//! -//! Implements `Write` and `Seek`, so any call site that uses `File` -//! through those traits (sweep, mux, patch) can swap to `Writer` -//! without touching the body of the loop. The wrapper tracks the -//! current file position and forwards each write to a -//! [`super::writeback::WritebackPipeline`], which on Linux schedules -//! incremental `sync_file_range` + `posix_fadvise(DONTNEED)` calls -//! to drain dirty pages continuously instead of letting the kernel -//! burst-flush hundreds of MB at a time. -//! -//! See `super::writeback::linux` for the pathology and the strategy. - -use std::fs::File; -use std::io::{self, Seek, SeekFrom, Write}; - -use super::writeback::WritebackPipeline; - -const CHUNK_BYTES: u64 = 32 * 1024 * 1024; - -pub(crate) struct Writer { - file: File, - pipeline: WritebackPipeline, - pos: u64, -} - -impl Writer { - /// Wrap an open `File`. The current OS file position is queried - /// once so the pipeline starts tracking from wherever the file - /// already is (typically 0 for fresh files; non-zero for resumed - /// or appended files). - pub(crate) fn new(mut file: File) -> io::Result { - let pos = file.stream_position()?; - let pipeline = WritebackPipeline::new(&file, pos, CHUNK_BYTES); - Ok(Self { - file, - pipeline, - pos, - }) - } - - /// Drain in-flight writeback then issue a full fsync. Use this in - /// place of `File::sync_all`. - pub(crate) fn sync_all(&mut self) -> io::Result<()> { - self.pipeline.finalize(); - self.file.sync_all() - } -} - -impl Write for Writer { - fn write(&mut self, buf: &[u8]) -> io::Result { - let n = self.file.write(buf)?; - self.pos += n as u64; - self.pipeline.note_progress(self.pos); - Ok(n) - } - - fn write_all(&mut self, buf: &[u8]) -> io::Result<()> { - self.file.write_all(buf)?; - self.pos += buf.len() as u64; - self.pipeline.note_progress(self.pos); - Ok(()) - } - - fn flush(&mut self) -> io::Result<()> { - self.file.flush() - } -} - -impl Seek for Writer { - fn seek(&mut self, from: SeekFrom) -> io::Result { - let p = self.file.seek(from)?; - // Only treat seeks that actually move the position as - // boundaries — sweep does a redundant `seek(Current(pos))` - // before every write, and we don't want that to drain the - // pipeline on every iteration. - if p != self.pos { - self.pipeline.handle_seek(p); - self.pos = p; - } - Ok(p) - } -} diff --git a/src/mux/resolve.rs b/src/mux/resolve.rs index 465f24e..7bf921a 100644 --- a/src/mux/resolve.rs +++ b/src/mux/resolve.rs @@ -242,23 +242,23 @@ pub fn output( match parsed { StreamUrl::Mkv { ref path } => { validate_file_path(path, "mkv")?; - // Wrap the raw `File` in `crate::io::Writer` (bounded-cache + // Wrap the output in `crate::io::WritebackFile` (bounded-cache // 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. - let file = std::fs::File::create(path)?; let writer: Box = Box::new(std::io::BufWriter::with_capacity( IO_BUF_SIZE, - crate::io::Writer::new(file)?, + crate::io::WritebackFile::create(path)?, )); Ok(Box::new(MkvStream::create(writer, title)?)) } StreamUrl::M2ts { ref path } => { validate_file_path(path, "m2ts")?; - let file = std::fs::File::create(path)?; - let writer = - std::io::BufWriter::with_capacity(IO_BUF_SIZE, crate::io::Writer::new(file)?); + let writer = std::io::BufWriter::with_capacity( + IO_BUF_SIZE, + crate::io::WritebackFile::create(path)?, + ); Ok(Box::new(M2tsStream::create(writer, title)?)) } StreamUrl::Network { ref addr } => {