From fc13268dc6d27b487aa3c9b4efd6f77b5745fe95 Mon Sep 17 00:00:00 2001 From: MattJackson <1085847+MattJackson@users.noreply.github.com> Date: Thu, 14 May 2026 22:18:23 -0700 Subject: [PATCH] io/writeback_file: coalesce consecutive Cmd::Writes in the writer thread MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pre-coalescing the Phase-2.5 writer thread issued one `file.write_all` syscall per `Cmd::Write` dequeued. The mux side calls `WritebackFile::write_all(buf)` per PES frame, typically 30-200 KB. On NFS that translates to one RPC per syscall, capping per-thread throughput at `(wsize / rtt) × inflight` — well below what the same disk delivers under a 1 MiB `dd oflag=direct` workload (empirical 2026-05-15: dd 71 MB/s vs mux ~25 MB/s sustained, with instantaneous samples bursting 7→108 MB/s as the kernel page cache filled and drained on its own cadence). Coalesce instead: dequeue drains consecutive `Cmd::Write` items off the ring up to a 1 MiB byte budget, returns them as `DequeuedWork::Writes(Vec>)`, and the run loop concatenates into one contiguous buffer and issues a single `file.write_all`. Non-write commands (Seek, Flush, SyncAll, Finish) break the run and are returned one at a time as `DequeuedWork::Other`, preserving their ordering relative to the writes. Single-buffer fast path avoids the concat allocation when only one write is in the queue at dequeue time. A single oversize write (e.g. the rare matroska cluster larger than 1 MiB) is admitted alone so it still makes progress — the kernel splits internally. This is generic across mediums: bigger app writes are at-least-as- good on local SSD, HDD, or NFS. On fast storage the ring rarely fills so coalescing is mostly a no-op; on slow storage with significant per-RPC overhead it materially improves throughput. Tests in `tests/` (write_then_drop_persists_bytes, sync_all_drains, seek_then_patch_roundtrip, flush_is_observed_in_order) still pass — ordering and durability semantics are unchanged. --- src/io/writeback_file/mod.rs | 119 +++++++++++++++++++++++++++++------ 1 file changed, 101 insertions(+), 18 deletions(-) diff --git a/src/io/writeback_file/mod.rs b/src/io/writeback_file/mod.rs index 9f63a44..086ec4f 100644 --- a/src/io/writeback_file/mod.rs +++ b/src/io/writeback_file/mod.rs @@ -638,11 +638,41 @@ struct WriterState { shared: Arc, } +/// Target byte budget for coalescing consecutive `Cmd::Write` commands +/// into a single `file.write_all` syscall. Sized to match NFS wsize on +/// rip1 (1 MiB), which is also the kernel default for most filesystems +/// and a reasonable upper bound for a single block-layer write on any +/// medium. The medium-specific syscall path may split further; that's +/// the kernel's job, not ours. +/// +/// 0.21.11 added coalescing because pre-coalescing the writer thread +/// emitted one syscall per muxer `Write` call (typically 30-200 KB PES +/// frames), and on NFS that translates to one RPC per syscall, capping +/// throughput at `(wsize / rtt) × inflight` — well below what the same +/// disk delivers under a 1 MiB `dd oflag=direct` workload (~71 MB/s +/// empirical 2026-05-15 vs ~25 MB/s sustained mux). Bigger app writes +/// = fewer NFS RPCs = better throughput on the same hardware. +const WRITE_COALESCE_TARGET_BYTES: usize = 1024 * 1024; + +/// One unit of work pulled off the ring. Either a coalesced run of +/// consecutive `Cmd::Write` commands (batched into a single +/// `file.write_all`) or a single non-write command. +enum DequeuedWork { + /// At least one consecutive `Cmd::Write` buffer, total length + /// bounded by [`WRITE_COALESCE_TARGET_BYTES`] (except when a + /// single `Cmd::Write` already exceeds the budget — in which case + /// it's returned alone). Coalesced and issued to the kernel as one + /// write_all. + Writes(Vec>), + /// A single non-`Write` command (Seek / Flush / SyncAll / Finish). + Other(Cmd), +} + impl WriterState { fn run(&mut self) { loop { - let cmd = match self.dequeue() { - Some(c) => c, + let work = match self.dequeue_work() { + Some(w) => w, None => { // All senders dropped (handle leaked); mark // writer_gone and exit. The Drop join will surface @@ -651,28 +681,48 @@ impl WriterState { return; } }; - match cmd { - Cmd::Write(buf) => { - if let Err(e) = self.do_write(&buf) { + match work { + DequeuedWork::Writes(bufs) => { + // Single-buffer fast path: skip the concat alloc. + // For the multi-buffer case, concatenate into a + // single contiguous slice so the kernel sees one + // write_all syscall — on NFS this becomes one RPC + // (per inflight slot) instead of N small RPCs, + // which is the whole point. + let result = if bufs.len() == 1 { + self.do_write(&bufs[0]) + } else { + let total: usize = bufs.iter().map(|b| b.len()).sum(); + let mut concat = Vec::with_capacity(total); + for b in &bufs { + concat.extend_from_slice(b); + } + self.do_write(&concat) + }; + if let Err(e) = result { self.publish_error(e.kind()); } } - Cmd::Seek(from) => { + DequeuedWork::Other(Cmd::Write(_)) => { + // dequeue_work places all Writes into DequeuedWork::Writes. + unreachable!("DequeuedWork::Other never wraps Cmd::Write"); + } + DequeuedWork::Other(Cmd::Seek(from)) => { if let Err(e) = self.do_seek(from) { self.publish_error(e.kind()); } } - Cmd::Flush => { + DequeuedWork::Other(Cmd::Flush) => { // No-op for now (see note on `Cmd::Flush`). } - Cmd::SyncAll { done } => { + DequeuedWork::Other(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 } => { + DequeuedWork::Other(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 @@ -686,21 +736,54 @@ impl WriterState { } } - /// Block until at least one command is available, then return it. - /// Notifies the muxer side that bytes are free. - fn dequeue(&mut self) -> Option { + /// Block until at least one command is available, then drain a + /// unit of work. Consecutive `Cmd::Write` commands at the front of + /// the queue are coalesced into a single `DequeuedWork::Writes` + /// (up to [`WRITE_COALESCE_TARGET_BYTES`] total). Non-write + /// commands break the run and are returned as `DequeuedWork::Other` + /// one at a time, preserving their ordering relative to writes. + /// Notifies the muxer side once after the dequeue completes so + /// ring-full waiters wake. + fn dequeue_work(&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()); + if guard.queue.is_empty() { + guard = self.shared.work_available.wait(guard).unwrap(); + continue; + } + // Branch on whether the front is a Write or something else. + if matches!(guard.queue.front(), Some(Cmd::Write(_))) { + let mut bufs: Vec> = Vec::new(); + let mut total_bytes: usize = 0; + while let Some(front_len) = guard.queue.front().and_then(|c| match c { + Cmd::Write(b) => Some(b.len()), + _ => None, + }) { + // Always admit the first write (even if oversize) + // so a giant single buffer still makes progress. + // Stop before exceeding the target on subsequent + // additions. + if !bufs.is_empty() && total_bytes + front_len > WRITE_COALESCE_TARGET_BYTES { + break; + } + match guard.queue.pop_front() { + Some(Cmd::Write(b)) => { + total_bytes += b.len(); + guard.bytes_inflight = guard.bytes_inflight.saturating_sub(b.len()); + bufs.push(b); + } + _ => unreachable!("front was Cmd::Write per the peek above"), + } } drop(guard); self.shared.space_available.notify_all(); - return Some(cmd); + return Some(DequeuedWork::Writes(bufs)); + } else { + let cmd = guard.queue.pop_front().expect("front was non-empty"); + drop(guard); + self.shared.space_available.notify_all(); + return Some(DequeuedWork::Other(cmd)); } - // Queue is empty. Wait for new work. - guard = self.shared.work_available.wait(guard).unwrap(); } }