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

io/writeback_file: coalesce consecutive Cmd::Writes in the writer thread

Browse Source 
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<Vec<u8>>)`, 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.
This commit is contained in:
matthew
2026-05-14 22:18:23 -07:00
parent a69ad202ab
commit c4fa65a905
1 changed files with 101 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/io/writeback_file/mod.rs
+101 -18
View File
@@ -638,11 +638,41 @@ struct WriterState {
shared: Arc<Shared>,
}
/// 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<Vec<u8>>),
/// 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<Cmd> {
/// 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<DequeuedWork> {
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<u8>> = 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();
}
}