From b24ba53322dd74623434c259e470dd76cdb909e6 Mon Sep 17 00:00:00 2001 From: Matthew Jackson Date: Sat, 16 May 2026 17:02:15 -0700 Subject: [PATCH] Reapply "io/writeback_file: restore Phase 2.5 (writer thread + bounded ring)" This reverts commit 212cc70cea9b90459fb20d5db2ea43f5951e0611. --- src/io/writeback_file/mod.rs | 891 +++++++++++++++++++++++++++++++---- 1 file changed, 807 insertions(+), 84 deletions(-) diff --git a/src/io/writeback_file/mod.rs b/src/io/writeback_file/mod.rs index 9362b9a..9f63a44 100644 --- a/src/io/writeback_file/mod.rs +++ b/src/io/writeback_file/mod.rs @@ -31,22 +31,67 @@ //! happens once at the bottom of this file via cfg-gated `mod` decls. //! No inline `#[cfg(target_os = "...")]` in the business-logic above. //! -//! ## Write path +//! ## Phase 2.5 — write-side flatness (writer thread) //! -//! Writes are direct passthrough to the underlying `File` (no writer -//! thread, no ring, no batching). Empirically the Phase-2.5 -//! writer-thread architecture introduced a ~60% mux throughput -//! regression on NFS bidirectional workloads; reverting the write path -//! to direct passthrough restores the 0.20.7 baseline. The writeback -//! pipeline still runs (it's called inline from `write` / `write_all` / -//! `seek`) so the bounded-cache invariant on Linux is preserved. +//! `WritebackFile` is split into a thin muxer-facing handle and a +//! dedicated writer thread that owns the real `File` + writeback +//! pipeline. The muxer's `Write::write` and `Seek::seek` calls return as +//! soon as the byte handoff to a bounded SPSC ring completes; the writer +//! thread executes the real syscalls (incl. `sync_file_range(WAIT_AFTER)` +//! on Linux) without ever blocking the muxer on a kernel commit. //! -//! ## Halt-safety +//! ### Backpressure //! -//! `sync_all` runs the per-OS durable-flush primitive, which on -//! Linux/macOS is wrapped in [`crate::io::bounded::bounded_syscall`] -//! with a 60 s deadline. A wedged NFS server cannot trap the muxer -//! indefinitely on the final fsync. +//! The ring is byte-bounded at [`RING_CAPACITY_BYTES`]. When the ring is +//! full the muxer's `write` blocks on a condvar until the writer thread +//! drains enough bytes to admit the next chunk. **Backpressure on +//! ring-full deliberately blocks the muxer rather than dropping bytes** +//! — archival workflows cannot afford byte loss, and the kernel page +//! cache is already a second buffering layer underneath the writer +//! thread. +//! +//! ### MKV seek-back semantics +//! +//! The MKV container backpatches cluster size headers shortly after +//! emitting them. The writer thread maintains an +//! [`ActiveClusterBuffer`] tracking the last +//! [`ACTIVE_CLUSTER_WINDOW_BYTES`] bytes by absolute file position. When +//! the muxer issues `Seek(pos)` for a `pos` inside that window, the +//! writer issues a real `file.seek` (cheap) but **skips +//! `pipeline.handle_seek()`** — no `sync_file_range(WAIT_AFTER)` drain +//! is forced for the current chunk. This is the dominant case: every +//! cluster backpatch is within the current 32 MiB writeback chunk. +//! +//! For seeks **outside** the window (rare — Cues index write at the end +//! of mux, Segment header backpatch right before close) the writer +//! falls back to the pre-Phase-2.5 behaviour: drain the in-flight +//! writeback via `pipeline.handle_seek()`, then issue the real seek. +//! +//! ### Halt-safe +//! +//! All real `sync_file_range(WAIT_AFTER)` and `fsync` calls on the +//! writer thread route through [`crate::io::bounded::bounded_syscall`] +//! with a 60 s deadline (already in place in the per-OS modules pre- +//! Phase-2.5). A wedged NFS server cannot freeze the writer thread +//! indefinitely; the muxer keeps queueing into the ring; the kernel +//! page cache absorbs. +//! +//! ### `sync_all` semantics +//! +//! `sync_all` is synchronous: it drains the ring through the writer +//! thread, then runs the per-OS durable-flush primitive, and returns +//! the result to the caller. This is the API contract — callers +//! (sweep/patch consumers, mux finalisation) rely on it. +//! +//! ### `speed_mbs` reporting +//! +//! Speed measurements taken at the muxer side (bytes handed off into +//! the ring) reflect ring-handoff throughput, **not** bytes committed +//! to durable storage. This is the correct number for muxer flatness +//! reporting; sweep/patch use mapfile-based progress which is unrelated. +//! Autorip's UI is unaffected — speed is calculated outside this +//! module — but the distinction is worth noting in release notes if the +//! UI ever exposes "throughput vs commit rate" separately. #[cfg(target_os = "linux")] mod linux; @@ -66,9 +111,13 @@ use other as platform; #[cfg(target_os = "windows")] use windows as platform; +use std::collections::VecDeque; use std::fs::{File, OpenOptions}; use std::io::{self, Seek, SeekFrom, Write}; use std::path::Path; +use std::sync::mpsc::{SyncSender, sync_channel}; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread::{self, JoinHandle}; use super::writeback::WritebackPipeline; @@ -77,25 +126,206 @@ use super::writeback::WritebackPipeline; /// historical default — bounded-cache pressure stays at ~2 × this size. const WRITEBACK_CHUNK_BYTES: u64 = 32 * 1024 * 1024; +/// Maximum bytes outstanding in the muxer → writer-thread ring. Sized +/// to cover ~4 s of muxer output at a 32 MB/s peak — enough to absorb a +/// short NFS commit stall without dropping the muxer's effective +/// throughput, but not so large that the resident-memory footprint +/// grows unbounded under a long writeback stall. +const RING_CAPACITY_BYTES: usize = 128 * 1024 * 1024; + +/// Bytes the writer thread keeps in [`ActiveClusterBuffer`] for the +/// in-window seek-then-patch fast path. Matches [`WRITEBACK_CHUNK_BYTES`] +/// so that a cluster backpatch landing inside the most recently written +/// (but not yet WAIT_AFTER'd) writeback chunk doesn't force a drain. +const ACTIVE_CLUSTER_WINDOW_BYTES: u64 = WRITEBACK_CHUNK_BYTES; + +/// Thread name used for the writer thread. Visible to OS-level tooling +/// (`ps -L`, `top -H`) so operators can correlate the muxer's flatness +/// with this thread's activity. +const WRITER_THREAD_NAME: &str = "freemkv-writeback-writer"; + +/// Single chunk size cap for batching: when a `Write` command arrives +/// the muxer copies its slice into a `Vec` to hand ownership over +/// the ring. We keep the allocation a single contiguous buffer — no +/// internal segmentation — so the writer thread can pass the slice +/// straight to `File::write_all` and the kernel can coalesce. +const MAX_WRITE_CHUNK_BYTES: usize = RING_CAPACITY_BYTES; // soft cap: a single command may not exceed the ring + +/// One command on the muxer → writer-thread ring. +enum Cmd { + /// Write `buf.len()` bytes at the writer's current logical + /// position, then advance. + Write(Vec), + /// Seek to `from` against the writer-side `File`. + Seek(SeekFrom), + /// Flush the muxer-side `Write::flush()` request (rarely useful; + /// kept for trait completeness). The writer ignores it — the real + /// flushing happens on `SyncAll`. + Flush, + /// Drain the ring then run the per-OS durable-sync primitive and + /// signal `done` with the result. + SyncAll { done: SyncSender> }, + /// Drain the ring (final pipeline finalize for chunk tail). Signal + /// `done` so `Drop` can wait synchronously. No fsync — that's what + /// `SyncAll` is for. + Finish { done: SyncSender<()> }, +} + +/// Ring state behind the muxer/writer condvar. `bytes_inflight` tracks +/// the sum of `Write(buf).len()` bytes currently queued so backpressure +/// can be enforced on a byte budget rather than a per-command count. +struct RingState { + queue: VecDeque, + /// Total `Write` payload bytes currently in `queue`. Non-write + /// commands (`Seek`, `Flush`, `SyncAll`, `Finish`) don't count + /// against the budget. + bytes_inflight: usize, + /// Set by the writer thread when it observes a fatal error (a + /// failed `write_all` or `seek` on the underlying file). Once set, + /// the muxer's next `write` / `seek` returns the error and stops + /// queueing. + sticky_error: Option, + /// Set when the writer thread has exited (clean Finish, panic, or + /// channel closed). Muxer-side ops surface this as a broken-pipe. + writer_gone: bool, +} + +struct Shared { + state: Mutex, + /// Notified when the writer dequeues something (bytes free up) or + /// when the writer exits. + space_available: Condvar, + /// Notified when the muxer pushes a new command. + work_available: Condvar, +} + +impl Shared { + fn new() -> Self { + Self { + state: Mutex::new(RingState { + queue: VecDeque::new(), + bytes_inflight: 0, + sticky_error: None, + writer_gone: false, + }), + space_available: Condvar::new(), + work_available: Condvar::new(), + } + } +} + +/// Tiny ring of recently-written bytes indexed by absolute file +/// position. Used by the writer thread to decide whether a `Seek` +/// target falls inside the active writeback chunk; if so, the seek +/// proceeds without forcing the pipeline to drain (the dominant case +/// for MKV cluster-size backpatches). +/// +/// The bytes themselves are kept in a contiguous `VecDeque` whose +/// front corresponds to file position [`lo`]. The data on disk is the +/// authoritative copy — `ActiveClusterBuffer` is a read-only mirror +/// used for in-window patch verification in tests and (potentially) for +/// future in-buffer mutation, NOT a write-through cache. +/// +/// [`lo`]: Self::lo +struct ActiveClusterBuffer { + /// Window capacity in bytes. The ring trims from the front to stay + /// at or below this size after every `push`. + cap: u64, + /// Absolute file position of the byte at `data.front()`. + lo: u64, + data: VecDeque, +} + +impl ActiveClusterBuffer { + fn new(cap: u64) -> Self { + Self { + cap, + lo: 0, + data: VecDeque::with_capacity(cap as usize), + } + } + + /// Reset the window — used after an out-of-window seek where the + /// previous data is no longer adjacent to the new position. + fn reset(&mut self, new_lo: u64) { + self.data.clear(); + self.lo = new_lo; + } + + /// One contiguous logical span of file positions currently held. + fn hi(&self) -> u64 { + self.lo + self.data.len() as u64 + } + + /// True if `pos` is in `[lo, hi]` (hi is exclusive of bytes but + /// inclusive of the seek-to-end-of-cluster boundary). + fn contains(&self, pos: u64) -> bool { + pos >= self.lo && pos <= self.hi() + } + + /// Append `bytes` at absolute file position `start`. If `start` is + /// contiguous with `hi()`, the bytes extend the window; otherwise + /// the window is reset (the previous data is no longer adjacent and + /// would corrupt the position index). + fn push(&mut self, start: u64, bytes: &[u8]) { + let h = self.hi(); + if start == h { + // Contiguous append. + self.data.extend(bytes.iter().copied()); + } else if start >= self.lo && start <= h { + // Patch landing inside the window: overwrite from + // (start - lo) for bytes.len(), then extend if it spills + // past `hi`. + let offset = (start - self.lo) as usize; + let mut bi = 0usize; + while bi < bytes.len() && offset + bi < self.data.len() { + self.data[offset + bi] = bytes[bi]; + bi += 1; + } + if bi < bytes.len() { + self.data.extend(bytes[bi..].iter().copied()); + } + } else { + // Non-contiguous: drop the window and reseat. + self.data.clear(); + self.lo = start; + self.data.extend(bytes.iter().copied()); + } + // Trim from the front so the window stays at or below `cap`. + while self.data.len() as u64 > self.cap { + self.data.pop_front(); + self.lo += 1; + } + } +} + +/// Muxer-facing handle. Holds a sender into the bounded ring and a +/// `JoinHandle` for the writer thread. `Write`/`Seek`/`sync_all`/`Drop` +/// route through the ring; the muxer thread is never trapped on a +/// commit syscall. pub(crate) struct WritebackFile { - file: File, - pipeline: WritebackPipeline, - pos: u64, + shared: Arc, + /// Joined on `Drop` after `Finish` so the writer thread's exit is + /// observed and any panic is surfaced loudly. `Option` so `Drop` + /// can `take()` it. + writer: Option>, + /// Logical file position from the muxer's point of view. Updated on + /// `write`/`write_all` (advanced by the count) and on `seek` + /// (replaced by the new position). Mirrors what the writer thread + /// will end up at once it has drained all queued commands — + /// callers that need a stream_position can read this without + /// blocking on the writer. + muxer_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). + /// once so the writer thread 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, WRITEBACK_CHUNK_BYTES); - Ok(Self { - file, - pipeline, - pos, - }) + Ok(Self::spawn(file, pos)) } /// Create a new file at `path` (truncating any existing contents) @@ -139,84 +369,432 @@ impl WritebackFile { Self::new(file) } + /// Spawn the writer thread for `file` starting at logical + /// position `start_pos`. The writer takes ownership of the `File`; + /// the muxer keeps the handle. + fn spawn(file: File, start_pos: u64) -> Self { + let shared = Arc::new(Shared::new()); + let shared_w = Arc::clone(&shared); + let writer = thread::Builder::new() + .name(WRITER_THREAD_NAME.into()) + .spawn(move || { + writer_thread_main(file, start_pos, shared_w); + }) + .expect("writer thread spawn"); + Self { + shared, + writer: Some(writer), + muxer_pos: start_pos, + } + } + /// Drain in-flight writeback then issue a full fsync. Use this in /// place of `File::sync_all`. /// - /// The final durable flush is wrapped in - /// [`crate::io::bounded::bounded_syscall`] (per the per-OS module) - /// with a 60 s deadline on Linux/macOS — a wedged NFS server cannot - /// trap the calling thread indefinitely. On timeout the page cache - /// is left to the kernel's normal flush-on-close path — best - /// effort, but bounded. + /// Blocks the calling thread until the ring is fully drained AND + /// the per-OS durable-flush primitive has returned. The flush + /// itself runs on the writer thread, wrapped in + /// [`crate::io::bounded::bounded_syscall`] (60 s deadline on + /// Linux + macOS); a wedged NFS server cannot trap the muxer. pub(crate) fn sync_all(&mut self) -> io::Result<()> { - self.pipeline.finalize(); - platform::durable_sync(&self.file) + let (tx, rx) = sync_channel::>(0); + self.push_command(Cmd::SyncAll { done: tx }, 0)?; + // recv() blocks until the writer thread drains the ring up to + // the SyncAll command, runs the per-OS durable-sync, and sends + // the result back. + match rx.recv() { + Ok(r) => r, + Err(_) => { + // Writer thread exited without sending. Surface a + // distinct error kind so the caller can distinguish + // "writer panicked" from a normal fsync failure. + Err(io::Error::from(io::ErrorKind::BrokenPipe)) + } + } + } + + /// Push a single command onto the ring. `bytes_charge` is the + /// number of bytes this command contributes to the byte-budget + /// backpressure check; only `Write` commands contribute. + fn push_command(&mut self, cmd: Cmd, bytes_charge: usize) -> io::Result<()> { + let mut guard = self.shared.state.lock().unwrap(); + // Surface any sticky error from the writer thread before + // queueing more work. The muxer should stop pushing once the + // writer has reported a failure. + if let Some(kind) = guard.sticky_error { + return Err(io::Error::from(kind)); + } + if guard.writer_gone { + return Err(io::Error::from(io::ErrorKind::BrokenPipe)); + } + // Byte-budget backpressure: wait for space if this would + // overflow the cap. A single command larger than the cap is + // admitted regardless (the cap is a soft target for batching; + // a giant single write still fits because the channel itself + // is unbounded count-wise). + while bytes_charge > 0 + && guard.bytes_inflight + bytes_charge > RING_CAPACITY_BYTES + && guard.bytes_inflight > 0 + { + guard = self.shared.space_available.wait(guard).unwrap(); + if let Some(kind) = guard.sticky_error { + return Err(io::Error::from(kind)); + } + if guard.writer_gone { + return Err(io::Error::from(io::ErrorKind::BrokenPipe)); + } + } + guard.queue.push_back(cmd); + guard.bytes_inflight += bytes_charge; + drop(guard); + self.shared.work_available.notify_one(); + Ok(()) } } 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); + // `Write::write` is allowed to be partial; we always accept + // the full slice (handoff is in-process) and report `buf.len()`. + // Callers that need the partial-write semantic still get the + // strict guarantee documented on `write_all`. + let n = buf.len(); + if n == 0 { + return Ok(0); + } + if n > MAX_WRITE_CHUNK_BYTES { + // Defensive: a single command bigger than the ring can't + // be admitted by the backpressure check above without + // deadlocking against itself. Split into ring-sized + // chunks. + let mut off = 0; + while off < n { + let take = (n - off).min(MAX_WRITE_CHUNK_BYTES); + self.push_command(Cmd::Write(buf[off..off + take].to_vec()), take)?; + off += take; + } + } else { + self.push_command(Cmd::Write(buf.to_vec()), n)?; + } + self.muxer_pos += n as u64; 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); + // `Write::write_all` default delegates to `write` in a loop. We + // can do better: a single handoff per call, never partial. Same + // chunk-split for the absurd-large case. + let n = buf.len(); + if n == 0 { + return Ok(()); + } + if n > MAX_WRITE_CHUNK_BYTES { + let mut off = 0; + while off < n { + let take = (n - off).min(MAX_WRITE_CHUNK_BYTES); + self.push_command(Cmd::Write(buf[off..off + take].to_vec()), take)?; + off += take; + } + } else { + self.push_command(Cmd::Write(buf.to_vec()), n)?; + } + self.muxer_pos += n as u64; Ok(()) } fn flush(&mut self) -> io::Result<()> { - self.file.flush() + // The writer thread's view of `Flush` is a no-op (real flushing + // happens on `SyncAll`). We still send it so a future change + // could intercept it; for now the queue ordering is the only + // observable effect. + self.push_command(Cmd::Flush, 0) } } 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 { - // Diagnostic for the NFS mux hang: the MKV format requires - // the muxer to seek back occasionally (cluster size - // patching, Cues index write, Segment header backpatch). - // Each such seek invalidates the writeback chunk tracking - // and forces a finalize → WAIT_AFTER on the in-flight - // chunk. Logging the seek delta lets us correlate hang - // offsets with specific muxer operations. - let from_pos = self.pos; - let to_pos = p; - let delta: i64 = (to_pos as i64).wrapping_sub(from_pos as i64); - tracing::debug!( - target: "mux", - "WritebackFile seek from={from_pos} to={to_pos} delta={delta}" - ); - self.pipeline.handle_seek(p); - self.pos = p; - } - Ok(p) + // The muxer's logical position must update synchronously so + // subsequent `write` calls advance from the right base, but the + // writer thread is the only place that has the authoritative + // OS file position. We model the muxer's `muxer_pos` purely + // from `SeekFrom::Start(n)` (the dominant case for MKV + // backpatch) and bounce other variants through to the writer + // by querying its current position via a synchronous round. + let new_pos = match from { + SeekFrom::Start(n) => n, + SeekFrom::Current(d) => { + let base = self.muxer_pos as i64; + let p = base + .checked_add(d) + .ok_or_else(|| io::Error::from(io::ErrorKind::InvalidInput))?; + if p < 0 { + return Err(io::Error::from(io::ErrorKind::InvalidInput)); + } + p as u64 + } + SeekFrom::End(_) => { + // SeekFrom::End requires the OS file's current EOF. We + // do not maintain that on the muxer side; this branch + // is not used by the MKV muxer (it always seeks with + // `SeekFrom::Start`). If a future caller needs it, the + // path is: SyncAll → real seek → query position back. + // Reject explicitly so a regression is loud. + return Err(io::Error::from(io::ErrorKind::Unsupported)); + } + }; + self.push_command(Cmd::Seek(SeekFrom::Start(new_pos)), 0)?; + self.muxer_pos = new_pos; + Ok(new_pos) } } impl Drop for WritebackFile { fn drop(&mut self) { - // Run the pipeline's tail finalize so the last in-flight chunk - // gets its `WAIT_AFTER` + `posix_fadvise(DONTNEED)`. Without - // this, callers that drop a `WritebackFile` without calling - // `sync_all` (panic, early-return, idiomatic `let _ = w;`) - // leave the trailing chunk in cache; the kernel still flushes - // on close, but the bounded-cache invariant fails at the tail. - // We deliberately do *not* call `self.file.sync_all()` here — - // close already triggers a flush, and an `fsync` from `Drop` - // would silently swallow its `io::Error` anyway. `finalize` is - // idempotent so an explicit `sync_all` followed by drop is - // still safe. + // Send a final `Finish` so the writer drains the ring (running + // the pipeline's tail finalize for the last in-flight chunk) + // before exiting. `sync_all` is *not* called from here — the + // existing pre-Phase-2.5 contract is "Drop runs finalize, not + // fsync", and Drop returning an io::Error is impossible anyway. + let (tx, rx) = sync_channel::<()>(0); + // If the writer thread already exited (sticky error path, or + // an earlier panic), the queue push will fail with + // BrokenPipe; we treat that as "nothing to drain" and proceed + // to join. + let push_ok = { + let mut guard = match self.shared.state.lock() { + Ok(g) => g, + Err(poison) => { + // The writer panicked; recover the mutex so we + // can still observe `writer_gone`. + poison.into_inner() + } + }; + if guard.writer_gone { + false + } else { + guard.queue.push_back(Cmd::Finish { done: tx }); + self.shared.work_available.notify_one(); + true + } + }; + if push_ok { + // Block until the writer signals Finish completed (ring is + // drained, pipeline finalize ran). The wait is bounded + // only by the writer's per-syscall deadlines; if the + // writer panicked between the push and the recv, the + // sender is dropped and `recv` returns Err — proceed to + // join. + let _ = rx.recv(); + } + if let Some(jh) = self.writer.take() { + // `join` surfaces a panic. We re-raise it loudly: a + // writer-thread panic indicates an io-layer bug, and + // swallowing it would mask data loss. + if let Err(panic) = jh.join() { + tracing::error!( + target: "mux", + "WritebackFile writer thread panicked during Drop; data may be lost" + ); + // Re-raise during drop is allowed (terminates the + // process), but doing so from Drop can cause a double- + // panic if the caller is already unwinding. Compromise: + // log loudly and resume_unwind only outside of an + // ongoing unwind. + if !std::thread::panicking() { + std::panic::resume_unwind(panic); + } + } + } + } +} + +/// Writer-thread entry. Owns the `File` and the `WritebackPipeline`; +/// pulls commands from the shared ring and executes them. Exits when a +/// `Finish` command is observed (clean shutdown from `Drop`) or when +/// the muxer side disconnects (every `Arc` cloned by the +/// handle is dropped — only happens on a forgotten-handle bug, which +/// the panic-surface in `Drop::join` catches). +fn writer_thread_main(file: File, start_pos: u64, shared: Arc) { + let pipeline = WritebackPipeline::new(&file, start_pos, WRITEBACK_CHUNK_BYTES); + let mut state = WriterState { + file, + pipeline, + pos: start_pos, + active: ActiveClusterBuffer::new(ACTIVE_CLUSTER_WINDOW_BYTES), + shared: Arc::clone(&shared), + }; + state.run(); +} + +/// All writer-thread-owned state. Methods here run exclusively on the +/// writer thread — no Send/Sync concerns inside the body. +struct WriterState { + file: File, + pipeline: WritebackPipeline, + /// Authoritative OS file position. Tracked locally so we can decide + /// whether a `Seek` is a no-op (target equals current position). + pos: u64, + active: ActiveClusterBuffer, + shared: Arc, +} + +impl WriterState { + fn run(&mut self) { + loop { + let cmd = match self.dequeue() { + Some(c) => c, + None => { + // All senders dropped (handle leaked); mark + // writer_gone and exit. The Drop join will surface + // this if anyone cares. + self.mark_writer_gone(); + return; + } + }; + match cmd { + Cmd::Write(buf) => { + if let Err(e) = self.do_write(&buf) { + self.publish_error(e.kind()); + } + } + Cmd::Seek(from) => { + if let Err(e) = self.do_seek(from) { + self.publish_error(e.kind()); + } + } + Cmd::Flush => { + // No-op for now (see note on `Cmd::Flush`). + } + Cmd::SyncAll { done } => { + let r = self.do_sync_all(); + // Ignore send errors: if the muxer dropped the + // receiver (cancelled wait), there's nothing to + // do. + let _ = done.send(r); + } + Cmd::Finish { done } => { + // Drain pipeline tail; do not fsync. Mark + // `writer_gone` so any racing `push_command` after + // this returns BrokenPipe instead of queueing into + // a dead writer. + self.pipeline.finalize(); + self.mark_writer_gone(); + let _ = done.send(()); + return; + } + } + } + } + + /// Block until at least one command is available, then return it. + /// Notifies the muxer side that bytes are free. + fn dequeue(&mut self) -> Option { + let mut guard = self.shared.state.lock().unwrap(); + loop { + if let Some(cmd) = guard.queue.pop_front() { + if let Cmd::Write(ref buf) = cmd { + guard.bytes_inflight = guard.bytes_inflight.saturating_sub(buf.len()); + } + drop(guard); + self.shared.space_available.notify_all(); + return Some(cmd); + } + // Queue is empty. Wait for new work. + guard = self.shared.work_available.wait(guard).unwrap(); + } + } + + fn do_write(&mut self, buf: &[u8]) -> io::Result<()> { + let start = self.pos; + self.file.write_all(buf)?; + self.pos += buf.len() as u64; + self.pipeline.note_progress(self.pos); + self.active.push(start, buf); + Ok(()) + } + + fn do_seek(&mut self, from: SeekFrom) -> io::Result<()> { + // We only ever push `SeekFrom::Start(n)` from the handle. + let target = match from { + SeekFrom::Start(n) => n, + // Defensive: should not occur on the wire, but handle + // gracefully. + SeekFrom::Current(_) | SeekFrom::End(_) => { + let p = self.file.seek(from)?; + self.pos = p; + self.active.reset(p); + self.pipeline.handle_seek(p); + return Ok(()); + } + }; + if target == self.pos { + // No-op seek (common: sweep emits `seek(Current(pos))` + // before every write). Skip the syscall. + return Ok(()); + } + if self.active.contains(target) { + // In-window seek: the target is inside the current + // writeback chunk's data. The kernel page cache already + // has those bytes; we issue the real `seek` (cheap, no + // commit syscall) but **skip** the pipeline's + // `handle_seek` so no `sync_file_range(WAIT_AFTER)` drain + // is forced. Subsequent writes still call + // `pipeline.note_progress` from the new position, so the + // writeback chunk accounting stays coherent — the chunk + // simply gets "re-emitted" data over its tail bytes, + // which is what the MKV backpatch is. + tracing::trace!( + target: "mux", + "WritebackFile in-window seek pos={} -> {} window=[{},{}]", + self.pos, + target, + self.active.lo, + self.active.hi(), + ); + self.file.seek(SeekFrom::Start(target))?; + self.pos = target; + } else { + // Out-of-window seek: rare (segment-header / Cues backpatch + // at end of mux). Drain in-flight writeback so the kernel + // doesn't carry dirty pages across the seek discontinuity, + // then do the real seek. + tracing::debug!( + target: "mux", + "WritebackFile out-of-window seek pos={} -> {} window=[{},{}]", + self.pos, + target, + self.active.lo, + self.active.hi(), + ); + self.pipeline.handle_seek(target); + self.file.seek(SeekFrom::Start(target))?; + self.pos = target; + self.active.reset(target); + } + Ok(()) + } + + fn do_sync_all(&mut self) -> io::Result<()> { self.pipeline.finalize(); + platform::durable_sync(&self.file) + } + + fn publish_error(&self, kind: io::ErrorKind) { + let mut guard = self.shared.state.lock().unwrap(); + if guard.sticky_error.is_none() { + guard.sticky_error = Some(kind); + } + drop(guard); + // Wake any muxer thread waiting on space — it will observe + // sticky_error and return. + self.shared.space_available.notify_all(); + } + + fn mark_writer_gone(&self) { + let mut guard = self.shared.state.lock().unwrap(); + guard.writer_gone = true; + drop(guard); + self.shared.space_available.notify_all(); } } @@ -239,13 +817,13 @@ mod tests { { let mut w = WritebackFile::create(&p).unwrap(); w.write_all(b"hello world").unwrap(); - // Drop drains the pipeline tail. + // Drop drains the ring. } assert_eq!(read_back(&p), b"hello world"); } #[test] - fn sync_all_drains_and_flushes() { + fn sync_all_blocks_until_ring_drains() { let dir = tempfile::tempdir().unwrap(); let p = dir.path().join("b.bin"); let mut w = WritebackFile::create(&p).unwrap(); @@ -253,8 +831,7 @@ mod tests { w.write_all(&[0x5au8; 1024]).unwrap(); } // After sync_all, the bytes MUST be visible to a separate - // reader. The pipeline has been finalised and durable-sync has - // run. + // reader. The ring has been drained and durable-sync has run. w.sync_all().unwrap(); let bytes = read_back(&p); assert_eq!(bytes.len(), 32 * 1024); @@ -263,17 +840,20 @@ mod tests { } #[test] - fn seek_then_patch_roundtrip() { - // Write A; seek back; patch with B; read back; the patch lands - // at the right offset. + fn in_window_seek_then_patch_roundtrip() { + // Write A; seek back inside the active-cluster window; patch + // with B; read back; the patch lands at the right offset. let dir = tempfile::tempdir().unwrap(); let p = dir.path().join("c.bin"); let mut w = WritebackFile::create(&p).unwrap(); + // 4 KiB of 'A' (well within ACTIVE_CLUSTER_WINDOW_BYTES = + // 32 MiB, so the seek-back is guaranteed in-window). let big = vec![b'A'; 4096]; w.write_all(&big).unwrap(); // Seek back to offset 1000 and overwrite 8 bytes. w.seek(SeekFrom::Start(1000)).unwrap(); w.write_all(b"PATCHED!").unwrap(); + // Seek to end so subsequent reads see the right size. w.sync_all().unwrap(); drop(w); let bytes = read_back(&p); @@ -284,10 +864,67 @@ mod tests { assert_eq!(bytes[1008], b'A'); } + #[test] + fn out_of_window_seek_then_patch_roundtrip() { + // Write enough bytes that a seek to offset 0 is outside the + // active-cluster window (which is 32 MiB). To keep the test + // bounded, we hammer the ActiveClusterBuffer's `cap` field + // directly via the public Write path — 33 MiB of payload is + // sufficient. + let dir = tempfile::tempdir().unwrap(); + let p = dir.path().join("d.bin"); + let mut w = WritebackFile::create(&p).unwrap(); + // Write 33 MiB of 'A'; the first 1 MiB is now outside the + // 32 MiB active-cluster window. + let chunk = vec![b'A'; 1024 * 1024]; + for _ in 0..33 { + w.write_all(&chunk).unwrap(); + } + // Seek to offset 100 (definitely outside the window) and + // patch. + w.seek(SeekFrom::Start(100)).unwrap(); + w.write_all(b"OUTSIDE!").unwrap(); + w.sync_all().unwrap(); + drop(w); + let bytes = read_back(&p); + assert_eq!(bytes.len(), 33 * 1024 * 1024); + assert_eq!(&bytes[100..108], b"OUTSIDE!"); + // Surrounding bytes are still 'A'. + assert_eq!(bytes[99], b'A'); + assert_eq!(bytes[108], b'A'); + } + + #[test] + fn backpressure_blocks_when_ring_full() { + // Submit more bytes than RING_CAPACITY_BYTES; if backpressure + // works, the call sequence still completes once the writer + // drains. We measure that the total written matches and the + // calling thread did not panic / loop forever. + let dir = tempfile::tempdir().unwrap(); + let p = dir.path().join("e.bin"); + let mut w = WritebackFile::create(&p).unwrap(); + // 4 × RING_CAPACITY_BYTES of payload, in chunks small enough + // that several can fit in the ring at once and backpressure + // triggers naturally. + let total = RING_CAPACITY_BYTES.saturating_mul(2) + (RING_CAPACITY_BYTES / 2); + let chunk = vec![0u8; 1024 * 1024]; + let mut written = 0; + while written < total { + let take = (total - written).min(chunk.len()); + w.write_all(&chunk[..take]).unwrap(); + written += take; + } + w.sync_all().unwrap(); + drop(w); + let meta = std::fs::metadata(&p).unwrap(); + assert_eq!(meta.len() as usize, total); + } + #[test] fn flush_is_observed_in_order() { - // `Write::flush` should not panic or reorder; verify the bytes - // land in order through interleaved flushes. + // `Write::flush` is a no-op on the writer side but must not + // panic or leak. Run an interleaved sequence and verify the + // bytes still land in order. let dir = tempfile::tempdir().unwrap(); let p = dir.path().join("f.bin"); let mut w = WritebackFile::create(&p).unwrap(); @@ -300,4 +937,90 @@ mod tests { drop(w); assert_eq!(read_back(&p), b"onetwothree"); } + + #[test] + fn active_cluster_buffer_contiguous_append_and_trim() { + let mut b = ActiveClusterBuffer::new(8); + b.push(0, b"abcd"); + assert_eq!(b.lo, 0); + assert_eq!(b.hi(), 4); + b.push(4, b"efgh"); + assert_eq!(b.lo, 0); + assert_eq!(b.hi(), 8); + // Push past the cap — front trims. + b.push(8, b"ij"); + assert_eq!(b.lo, 2); + assert_eq!(b.hi(), 10); + assert!(b.contains(2)); + assert!(b.contains(10)); + assert!(!b.contains(1)); + assert!(!b.contains(11)); + } + + #[test] + fn active_cluster_buffer_in_window_patch() { + let mut b = ActiveClusterBuffer::new(16); + b.push(100, b"AAAAAAAA"); + assert!(b.contains(104)); + // Patch in the middle. + b.push(102, b"BB"); + let collected: Vec = b.data.iter().copied().collect(); + assert_eq!(collected, b"AABBAAAA"); + assert_eq!(b.lo, 100); + assert_eq!(b.hi(), 108); + } + + #[test] + fn active_cluster_buffer_non_contiguous_reseats() { + let mut b = ActiveClusterBuffer::new(16); + b.push(0, b"abcd"); + b.push(1000, b"XYZ"); + assert_eq!(b.lo, 1000); + assert_eq!(b.hi(), 1003); + } + + #[test] + fn writer_thread_panic_surfaces_on_drop() { + // Simulate a writer-side panic by writing to a read-only file + // — the underlying `file.write_all` will return EBADF / + // PermissionDenied. The writer publishes sticky_error and + // exits via the next dequeue; subsequent push_command returns + // the error. Drop joins cleanly (no panic from the writer + // thread itself; it returned through the error path). + // + // We deliberately use a closed-FD strategy: open a file, + // truncate the kernel's view by closing it, then write — this + // is hard to force without unsafe. Easier: write to a path + // and then forcibly close the underlying File via shutdown of + // the writer thread. Since we don't expose the inner File, + // pick the read-only-mode approach: open the file in + // read-only mode and try to write. + let dir = tempfile::tempdir().unwrap(); + let p = dir.path().join("ro.bin"); + std::fs::write(&p, b"seed").unwrap(); + let f = OpenOptions::new().read(true).open(&p).unwrap(); + let mut w = WritebackFile::new(f).unwrap(); + // First write may succeed depending on platform; loop until + // an error surfaces. On Linux a write on an O_RDONLY fd + // returns EBADF immediately. + let mut saw_error = false; + for _ in 0..32 { + match w.write_all(b"x") { + Ok(()) => { + // Give the writer thread a moment to surface the + // error then retry. + std::thread::sleep(std::time::Duration::from_millis(10)); + } + Err(_) => { + saw_error = true; + break; + } + } + } + assert!( + saw_error, + "expected the writer to publish an error on a read-only fd" + ); + drop(w); + } }