pq/libfreemkv
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

io/writeback_file: restore Phase 2.5 (writer thread + bounded ring)

Browse Source 
Reverts 1001084. That revert was made on the premise that Phase 2.5
caused a ~60% mux throughput regression on NFS bidirectional workloads.
The premise was wrong: at the time of measurement the producer was
capped at ~8 MB/s by a 50 ms thread::sleep poll in
Pipeline::send_with_halt (fixed in v0.21.7's io/pipeline change), so
the comparison was measuring the polling cap on both sides.

With the polling cap removed, direct passthrough exposes the kernel's
default dirty-page writeback pathology on NFS: writes accumulate, the
kernel periodically bursts a flush, app writes block for the burst.
Observed empirically on Civil War UHD remux 2026-05-14: 5-45 MB/s
spiking around a ~21 MB/s sustained mean, dominated by burst-flush
back-pressure cycles.

Phase 2.5 decouples the mux thread from the file syscall:
  * mux writes complete instantly into a 128 MiB byte-bounded SPSC ring,
  * a dedicated writer thread executes the real File writes, seeks,
    and sync_file_range calls; can sit in a kernel burst without
    blocking the mux pipeline,
  * backpressure via Condvar notify/wait, no polling primitive,
  * the ActiveClusterBuffer fast-path preserves the original MKV
    cluster-backpatch optimisation so in-window seeks don't drain
    the current writeback chunk.

Halt-safety is preserved: every blocking writeback syscall on the
writer thread still routes through bounded_syscall with a 60 s
deadline. A wedged NFS server cannot trap the writer indefinitely;
the muxer keeps queueing into the ring; the kernel page cache and
the ring together absorb the stall.
This commit is contained in:
matthew
2026-05-14 16:42:00 -07:00
parent 9043a4ba7a
commit 2cdcb4f3e1
1 changed files with 807 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/io/writeback_file/mod.rs
+806 -83
View File
@@ -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<u8>` 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<u8>),
/// 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<io::Result<()>> },
/// 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<Cmd>,
/// 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<io::ErrorKind>,
/// 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<RingState>,
/// 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<u8>` 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<u8>,
}
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<Shared>,
/// 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<JoinHandle<()>>,
/// 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<Self> {
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::<io::Result<()>>(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<usize> {
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<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 {
// 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;
// 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));
}
Ok(p)
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<Shared>` 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<Shared>) {
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<Shared>,
}
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<Cmd> {
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<u8> = 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);
}
}