v0.17.10: bounded-cache writeback pipeline for big sequential writes

Pass 1 sweep speed on a healthy disc previously dipped from ~15 MB/s
to ~1 MB/s every ~30 s on a host with default Linux dirty-page
settings. Empirical cause: the kernel's vm.dirty_ratio (~20% of RAM)
lets hundreds of MB of dirty pages accumulate, then bursts a flush at
99% disk utilisation that blocks app writes for ~1 s. Confirmed on
the BU40N test bed — dirty pages grew 112 → 563 MB between bursts;
lowering vm.dirty_bytes to 64 MB at the host sysctl level eliminated
the dips. Shipping the equivalent inside libfreemkv so users do not
need to tune the host kernel.

- New crate::io::Writer: drop-in File wrapper (impl Write + Seek).
  Wraps a per-platform WritebackPipeline that on Linux schedules
  sync_file_range(WRITE) + lagging sync_file_range(WAIT_AFTER) +
  posix_fadvise(DONTNEED) in 32 MB chunks, bounding dirty cache at
  ~64 MB. macOS and Windows ship a no-op stub.
- Disc::sweep wraps its output File in Writer. Loop body unchanged.
- Module is purpose-built so any large sequential output (patch,
  mux) can adopt the same wrapper as a one-line change later.

src/disc/mod.rs

@@ -1414,7 +1414,11 @@ impl Disc {
             f
         };
 
-        let mut file = file;
+        // Wrap the raw `File` in our bounded-cache writer so the
+        // kernel's writeback queue drains continuously instead of
+        // accumulating hundreds of MB of dirty pages and then bursting
+        // a flush that blocks app writes (see `crate::io`).
+        let mut file = crate::io::Writer::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),

src/io/mod.rs

@@ -0,0 +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.
+
+mod writeback;
+mod writer;
+
+pub(crate) use writer::Writer;

src/io/writeback.rs

@@ -0,0 +1,17 @@
+//! Per-platform writeback pipeline. On Linux, drains dirty pages
+//! continuously at chunk granularity to keep the kernel's writeback
+//! queue bounded. On macOS and Windows, a no-op stub.
+//!
+//! The platform decision lives entirely in this file (the cfg-gated
+//! `pub use` below). Callers — and `DiskWriter` itself — are
+//! platform-independent.
+
+#[cfg(target_os = "linux")]
+mod linux;
+#[cfg(not(target_os = "linux"))]
+mod noop;
+
+#[cfg(target_os = "linux")]
+pub(super) use linux::WritebackPipeline;
+#[cfg(not(target_os = "linux"))]
+pub(super) use noop::WritebackPipeline;

src/io/writeback/linux.rs

@@ -0,0 +1,95 @@
+//! Linux writeback pipeline using `sync_file_range` + `posix_fadvise`.
+//!
+//! Pathology this fixes: the kernel's default `vm.dirty_ratio` (~20 %
+//! of RAM) lets dirty pages accumulate to hundreds of MB during a
+//! big sequential write, then bursts a flush at 99 % disk utilisation.
+//! While the burst runs, app writes block on the writeback queue —
+//! observed empirically as instantaneous speed dropping from ~15 MB/s
+//! to ~1 MB/s every ~30 s during a Pass 1 sweep.
+//!
+//! Strategy: every `chunk_bytes` of new sequential output, kick async
+//! writeback (`SYNC_FILE_RANGE_WRITE`) on the just-completed chunk and
+//! finalise the *previous* chunk via `WAIT_AFTER` + `posix_fadvise
+//! (DONTNEED)`. By the time we finalise, that previous chunk has had
+//! 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.
+
+use std::fs::File;
+use std::os::unix::io::{AsRawFd, RawFd};
+
+pub(crate) struct WritebackPipeline {
+    fd: RawFd,
+    chunk_bytes: u64,
+    last_flush_pos: u64,
+    pending: Option<(u64, u64)>,
+}
+
+impl WritebackPipeline {
+    pub(crate) fn new(file: &File, start_pos: u64, chunk_bytes: u64) -> Self {
+        Self {
+            fd: file.as_raw_fd(),
+            chunk_bytes,
+            last_flush_pos: start_pos,
+            pending: None,
+        }
+    }
+
+    /// Caller advanced the file position to `pos`. If a chunk boundary
+    /// was crossed, kick async writeback for the just-completed chunk
+    /// and finalise the previous one.
+    pub(crate) fn note_progress(&mut self, pos: u64) {
+        if pos < self.last_flush_pos.saturating_add(self.chunk_bytes) {
+            return;
+        }
+        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() {
+                libc::sync_file_range(
+                    self.fd,
+                    prev_off as i64,
+                    prev_len as i64,
+                    libc::SYNC_FILE_RANGE_WAIT_AFTER,
+                );
+                libc::posix_fadvise(
+                    self.fd,
+                    prev_off as i64,
+                    prev_len as i64,
+                    libc::POSIX_FADV_DONTNEED,
+                );
+            }
+            self.pending = Some((chunk_off as u64, chunk_len as u64));
+        }
+        self.last_flush_pos = pos;
+    }
+
+    /// Caller is about to seek away from the current write region.
+    /// Drain any in-flight chunk and reset tracking.
+    pub(crate) fn handle_seek(&mut self, new_pos: u64) {
+        self.finalize();
+        self.last_flush_pos = new_pos;
+    }
+
+    /// Drain any in-flight chunk. Idempotent. Call before `sync_all()`
+    /// or when discarding the pipeline.
+    pub(crate) fn finalize(&mut self) {
+        if let Some((prev_off, prev_len)) = self.pending.take() {
+            unsafe {
+                libc::sync_file_range(
+                    self.fd,
+                    prev_off as i64,
+                    prev_len as i64,
+                    libc::SYNC_FILE_RANGE_WAIT_AFTER,
+                );
+                libc::posix_fadvise(
+                    self.fd,
+                    prev_off as i64,
+                    prev_len as i64,
+                    libc::POSIX_FADV_DONTNEED,
+                );
+            }
+        }
+    }
+}

src/io/writeback/noop.rs

@@ -0,0 +1,18 @@
+//! No-op writeback pipeline for non-Linux targets. macOS and Windows
+//! page cache policies have not been shown to exhibit the Linux
+//! accumulate-then-burst flush pathology for our access pattern.
+//! If that changes, replace this stub with a real implementation
+//! (e.g. `F_NOCACHE` on macOS, `FILE_FLAG_WRITE_THROUGH` on Windows).
+
+use std::fs::File;
+
+pub(crate) struct WritebackPipeline;
+
+impl WritebackPipeline {
+    pub(crate) fn new(_file: &File, _start_pos: u64, _chunk_bytes: u64) -> Self {
+        Self
+    }
+    pub(crate) fn note_progress(&mut self, _pos: u64) {}
+    pub(crate) fn handle_seek(&mut self, _new_pos: u64) {}
+    pub(crate) fn finalize(&mut self) {}
+}

src/io/writer.rs

@@ -0,0 +1,88 @@
+//! `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, Metadata};
+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<Self> {
+        let pos = file.stream_position()?;
+        let pipeline = WritebackPipeline::new(&file, pos, CHUNK_BYTES);
+        Ok(Self {
+            file,
+            pipeline,
+            pos,
+        })
+    }
+
+    pub(crate) fn metadata(&self) -> io::Result<Metadata> {
+        self.file.metadata()
+    }
+
+    /// 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<usize> {
+        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<u64> {
+        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)
+    }
+}

src/lib.rs

@@ -81,6 +81,7 @@ pub mod error;
 pub mod event;
 pub(crate) mod identity;
 pub(crate) mod ifo;
+pub(crate) mod io;
 pub mod keydb;
 pub(crate) mod labels;
 pub(crate) mod mpls;