From 92c34772031b01eb3f3c13d82210cbe3549ae8b6 Mon Sep 17 00:00:00 2001 From: thomasjm Date: Sat, 14 Nov 2026 03:36:51 -0800 Subject: [PATCH 1/4] Split writeback into --writeback-cache (off) and --write-pipeline --- README.md | 5 +- src/fuse9p.rs | 301 +++++++++++++++++++++++------------------- src/lib.rs | 34 +++++ src/main.rs | 25 +++- src/write_pipeline.rs | 143 ++++++++++++++++++++ tests/diod_fuse.rs | 213 +++++++++++++++++++++--------- 6 files changed, 519 insertions(+), 202 deletions(-) create mode 100644 src/write_pipeline.rs diff --git a/README.md b/README.md index 6ee2064..8460a9f 100644 --- a/README.md +++ b/README.md @@ -89,8 +89,9 @@ multiple clients write concurrently. | `attr_ttl` | 60s | How long the kernel trusts a cached `getattr`. `0` = round-trip every stat. | | `entry_ttl` | 60s | How long the kernel trusts a cached name→inode lookup. | | `negative_ttl` | 5s | Cache "no such file" lookups (skip round-trips on `$PATH`/include probes). | -| `writeback` | on | Pipeline writes via the FUSE writeback cache instead of one round-trip per write. | -| `wb_depth` | 16 | Max in-flight `Twrite`s per open file when write-back is on. | +| `write_pipeline` | on | Acknowledge writes at once and send their `Twrite`s in the background instead of one round-trip per write. | +| `wb_depth` | 16 | Max in-flight `Twrite`s per file when the write pipeline is on. | +| `writeback_cache` | off | Let the kernel buffer writes in its page cache. Faster for small writes, but only safe when nothing else writes to the export: the kernel then keeps its own file sizes, so a file grown from outside the mount reads back truncated while its inode is cached, even after invalidation. | | `readdirplus` | off | Prefetch entry attrs on `readdir`. Off by default — usually net-negative. | ## Custom transports diff --git a/src/fuse9p.rs b/src/fuse9p.rs index 4f4b625..4cdbb2e 100644 --- a/src/fuse9p.rs +++ b/src/fuse9p.rs @@ -8,6 +8,7 @@ use crate::client::NineClient; use crate::ninep::Attr; +use crate::write_pipeline::{write_all, InodeWrites, Write, WriteError}; use fuser::{ FileAttr, FileType, Filesystem, MountOption, ReplyAttr, ReplyCreate, ReplyData, ReplyDirectory, ReplyDirectoryPlus, ReplyEmpty, ReplyEntry, ReplyOpen, ReplyWrite, Request, TimeOrNow, @@ -15,10 +16,9 @@ use fuser::{ use std::collections::HashMap; use std::ffi::OsStr; use std::path::Path; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::runtime::Handle; -use tokio::sync::Semaphore; const ROOT_INO: u64 = 1; const AT_REMOVEDIR: u32 = 0x200; @@ -44,11 +44,18 @@ pub struct Tuning { /// a just-listed file is a kernel-cache hit instead of a round-trip -- the big lever for /// `ls`/`find`/`git`/build tree-walks. Off = plain `readdir` (kernel stats each entry separately). pub readdirplus: bool, - /// Pipeline writes: defer each `Twrite` to a background task (depth `wb_depth`) and let the kernel - /// buffer/flush via FUSE writeback cache, turning latency-bound sequential writes into throughput. - /// Off = each `write` is one synchronous `Twrite` round-trip (the original behavior). - pub writeback: bool, - /// Max concurrent in-flight `Twrite`s per open file when `writeback` is on. Bounds memory + /// Advertise `FUSE_WRITEBACK_CACHE`: the kernel buffers writes in its page cache and flushes them + /// in large background batches. Unsafe when the export is also changed from outside the mount: + /// under this capability the kernel treats its own `i_size` as authoritative, so a file grown + /// behind the mount reads back truncated to the old size for as long as its inode stays cached, + /// invalidation notwithstanding. Only turn it on when this mount is the sole writer. + pub writeback_cache: bool, + /// Pipeline writes: acknowledge each FUSE `write` at once and send its `Twrite` from a background + /// task, up to `wb_depth` in flight per file, turning latency-bound sequential writes into + /// throughput. Errors surface at flush/fsync/release. Off = each `write` is one synchronous + /// `Twrite` round-trip. + pub write_pipeline: bool, + /// Max concurrent in-flight `Twrite`s per file when `write_pipeline` is on. Bounds memory /// (~`wb_depth` * msize) and sets the pipeline depth. pub wb_depth: usize, } @@ -66,7 +73,8 @@ impl Default for Tuning { // the attr cache. The caching knobs deliver the v9fs-`cache=loose` speed; readdirplus // doesn't. Kept as a knob in case a future impl walks per-entry fids concurrently. readdirplus: false, - writeback: true, + writeback_cache: false, + write_pipeline: true, wb_depth: 16, } } @@ -84,18 +92,12 @@ struct Inode { lookups: u64, } -/// Write-back state for one open writable file: a semaphore bounding in-flight `Twrite`s (the -/// pipeline depth) and the first error seen by any background write, surfaced at flush/fsync/release. -struct WriteBack { - sem: Arc, - depth: usize, - err: Mutex>, -} - -/// An open file/dir handle: the opened 9p fid, plus (for writable files) the write-back tracker. +/// An open file/dir handle: the opened 9p fid and its inode, plus, for a handle whose writes go +/// through the inode's write pipeline, the slot for the first error one of them hits. struct FileHandle { fid: u32, - wb: Option>, + ino: u64, + wb: Option, } pub struct Fuse9p { @@ -108,6 +110,8 @@ pub struct Fuse9p { /// Open file/dir handles, keyed by fh. handles: HashMap, next_fh: u64, + /// The write pipeline of each inode with a pipelined handle open, keyed by inode. + writes: HashMap>, } /// Map a 9p qid.path to a FUSE inode number. We use the qid.path directly (it's a stable unique id @@ -132,39 +136,74 @@ impl Fuse9p { self.inodes.get(&ino).map(|i| i.fid) } - /// Register an opened fid as a new file/dir handle. Writable file handles get a write-back - /// tracker (when the `writeback` knob is on) so their writes can pipeline; read-only/dir handles - /// and the writeback-off case don't, so writes fall back to synchronous round-trips. - fn insert_handle(&mut self, fid: u32, writable: bool) -> u64 { + /// The flags to open a file's fid with for the kernel's `flags`, and whether writes through it + /// may be pipelined. + fn open_flags(&self, flags: i32) -> (u32, bool) { + let mut oflags = (flags as u32) & !(libc::O_CLOEXEC as u32); + let acc = oflags & libc::O_ACCMODE as u32; + // With the FUSE writeback cache the kernel issues READs on the handle for partial-page + // read-modify-write even when the file was opened write-only, which the server would + // refuse with EBADF on a write-only fid; so open those read-write. + if self.tuning.writeback_cache && acc == libc::O_WRONLY as u32 { + oflags = (oflags & !(libc::O_ACCMODE as u32)) | libc::O_RDWR as u32; + } + let pipelined = acc != libc::O_RDONLY as u32 && self.tuning.write_pipeline; + (oflags, pipelined) + } + + /// Register an opened fid as a new file/dir handle. A `pipelined` handle sends its writes + /// through the inode's write pipeline; others write synchronously. + fn insert_handle(&mut self, fid: u32, ino: u64, pipelined: bool) -> u64 { let fh = self.next_fh; self.next_fh += 1; - let wb = if writable && self.tuning.writeback { - let depth = self.tuning.wb_depth.max(1); - Some(Arc::new(WriteBack { - sem: Arc::new(Semaphore::new(depth)), - depth, - err: Mutex::new(None), - })) + let wb = if pipelined { + let (client, chunk, depth) = ( + self.client.clone(), + self.write_chunk(), + self.tuning.wb_depth, + ); + self.writes + .entry(ino) + .or_insert_with(|| InodeWrites::new(client, chunk, depth)); + Some(WriteError::default()) } else { None }; - self.handles.insert(fh, FileHandle { fid, wb }); + self.handles.insert(fh, FileHandle { fid, ino, wb }); fh } - /// Block until every in-flight write-back task for `wb` has completed, returning the first error - /// any of them hit. Acquiring all `depth` permits is only possible once all outstanding writes - /// have released theirs. - fn drain_writeback(&self, wb: Arc) -> Option { - let depth = wb.depth as u32; + /// The most bytes one `Twrite` carries: the negotiated msize less the header. + fn write_chunk(&self) -> usize { + self.client.msize.saturating_sub(24) as usize + } + + /// Wait for the pipelined writes still unsent or in flight on `ino` (through any handle), or on + /// every inode when `ino` is `None`, so a server op that reports a file's size or contents sees + /// them, and a synchronous write or truncate lands after them. Without the writeback cache the + /// kernel takes size and data from us, and the server does not know about writes we have + /// acknowledged but not yet sent. Errors stay recorded for that handle's flush/fsync/release. + fn settle_writes(&self, ino: Option) { + let busy: Vec> = match ino { + Some(ino) => self.writes.get(&ino).cloned().into_iter().collect(), + None => self.writes.values().cloned().collect(), + }; + let busy: Vec> = busy.into_iter().filter(|w| w.busy()).collect(); + if busy.is_empty() { + return; + } self.rt.block_on(async move { - let _all = wb - .sem - .acquire_many(depth) - .await - .expect("write-back semaphore closed"); - wb.err.lock().unwrap().take() - }) + for w in busy { + w.settle().await; + } + }); + } + + /// Wait for every pipelined write on the handle's inode, returning the first error any write + /// through this handle hit since the last time it was asked. + fn drain_writeback(&self, ino: u64, err: &WriteError) -> Option { + self.settle_writes(Some(ino)); + err.lock().unwrap().take() } } @@ -190,8 +229,6 @@ fn negative_attr() -> FileAttr { } } -/// Write `data` to `fid` starting at byte `off`, split into `chunk`-sized `Twrite`s. A short write of -/// 0 is treated as an error so callers don't spin. /// Read up to `cap` bytes at `off`, reassembling short replies. /// /// A 9p server may answer a Tread with fewer bytes than asked for even mid-file -- a short `pread` @@ -213,27 +250,6 @@ async fn read_full(client: &NineClient, fid: u32, off: u64, cap: u32) -> Result< Ok(out) } -async fn write_all( - client: &NineClient, - fid: u32, - off: u64, - data: &[u8], - chunk: usize, -) -> Result<(), i32> { - let mut pos = 0usize; - let mut o = off; - while pos < data.len() { - let end = (pos + chunk.max(1)).min(data.len()); - let n = client.write(fid, o, &data[pos..end]).await?; - if n == 0 { - return Err(libc::EIO); - } - pos += n as usize; - o += n as u64; - } - Ok(()) -} - fn unixtime((sec, nsec): (u64, u64)) -> SystemTime { UNIX_EPOCH + Duration::new(sec, (nsec as u32).min(999_999_999)) } @@ -313,6 +329,10 @@ impl Fuse9p { // Give up on a 9p request unanswered for this long: the transport is declared dead, the // mount exits, and the supervisor remounts. Zero disables the watchdog. stall_timeout: std::time::Duration, + // Where invalidation requests arrive, one per line, and where their replies go (see + // `invalidation_loop`). The binary passes its stdin and stdout. + invalidations: Box, + replies: Box, ) -> Result<(), Box> { tracing::info!(?tuning, "mount9p-fuse: tuning"); // Attach as `uid` so the server acts as that user for file ops (a multiuser server like diod @@ -345,6 +365,7 @@ impl Fuse9p { root_qid_path, handles: HashMap::new(), next_fh: 1, + writes: HashMap::new(), }; // No AutoUnmount: that path wants the libfuse feature we deliberately dropped. The kernel @@ -371,8 +392,10 @@ impl Fuse9p { // letting us cache aggressively AND stay coherent when the backing store changes underneath. let mut session = fuser::Session::new(fs, &mp, &options)?; let notifier = session.notifier(); - // Read invalidation paths (one per line) from stdin and drop them from the kernel cache. + // Read invalidation paths (one per line) and drop them from the kernel cache. tokio::spawn(invalidation_loop( + invalidations, + replies, notifier, inval_client, root_fid, @@ -406,12 +429,14 @@ impl Fuse9p { } } -/// Reads invalidation paths from stdin (one path per line, relative to the mount root) and drops them +/// Reads invalidation paths (one path per line, relative to the mount root) and drops them /// from the kernel's caches via FUSE notifies. Feed this from whatever knows the backing store changed /// out-of-band, to keep aggressive caching coherent. Resolving a path to its inode is a plain 9p walk /// (inode == qid.path), so no shared state with the FUSE filesystem is needed. Best-effort: a path the /// kernel never cached simply isn't there to drop. async fn invalidation_loop( + invalidations: Box, + replies: Box, notifier: fuser::Notifier, client: Arc, root_fid: u32, @@ -425,14 +450,14 @@ async fn invalidation_loop( // complete the LOOKUP is never processed -> the kernel never releases the lock -> the notify // never returns). Run every notify on a blocking thread so the runtime keeps servicing 9p. let notifier = std::sync::Arc::new(notifier); - let mut lines = BufReader::new(tokio::io::stdin()).lines(); - let mut out = tokio::io::stdout(); + let mut lines = BufReader::new(invalidations).lines(); + let mut out = replies; loop { let line = match lines.next_line().await { Ok(Some(l)) => l, Ok(None) => break, Err(e) => { - tracing::warn!(?e, "mount9p-fuse: error reading invalidation stdin"); + tracing::warn!(?e, "mount9p-fuse: error reading invalidations"); break; } }; @@ -450,7 +475,7 @@ async fn invalidation_loop( e, "mount9p-fuse: undecodable invalidation path" ); - reply(&mut out, &format!("err {id} {e}")).await; + reply(&mut *out, &format!("err {id} {e}")).await; continue; } } @@ -513,11 +538,11 @@ async fn invalidation_loop( } } match &reply_id { - Some(id) => reply(&mut out, &format!("ok {id}")).await, - None => reply(&mut out, &format!("ok {path}")).await, + Some(id) => reply(&mut *out, &format!("ok {id}")).await, + None => reply(&mut *out, &format!("ok {path}")).await, } } - tracing::info!("mount9p-fuse: invalidation stdin closed"); + tracing::info!("mount9p-fuse: invalidation input closed"); } /// Percent-decode a request's path. The client percent-encodes it because a filename may contain @@ -533,7 +558,7 @@ fn decode_path(s: &str) -> Result { /// orders work after an invalidation needs this, since writing to our stdin only says the path was /// queued. Replies are tagged so a client can tell one from anything else that ever reaches stdout, /// and match it to the request it answers rather than relying on position. -async fn reply(out: &mut tokio::io::Stdout, line: &str) { +async fn reply(out: &mut (dyn tokio::io::AsyncWrite + Send + Unpin), line: &str) { use tokio::io::AsyncWriteExt; let mut buf = String::with_capacity(line.len() + 1); buf.push_str(line); @@ -563,7 +588,7 @@ impl Filesystem for Fuse9p { const FUSE_WRITEBACK_CACHE: u32 = 1 << 16; // kernel buffers + async-flushes writes const FUSE_DO_READDIRPLUS: u32 = 1 << 13; // kernel issues READDIRPLUS (readdir+lookup) ops const FUSE_READDIRPLUS_AUTO: u32 = 1 << 14; // ...adaptively, only when it pays off - if self.tuning.writeback { + if self.tuning.writeback_cache { // Writeback caching lets the kernel buffer the app's writes and flush them async, // coalesced and several in flight (max_background, default 16) -- which, with our async // write-back handler, turns the latency-bound write path into a pipelined one. @@ -585,8 +610,15 @@ impl Filesystem for Fuse9p { let newfid = self.client.alloc_fid(); let client = self.client.clone(); // Walk first, and only own (and so only clunk) newfid once the walk has established it. + let walked = self + .rt + .block_on(async move { client.walk(parent_fid, newfid, &[&name]).await }); + if let Ok(Some(q)) = walked.as_ref().map(|qids| qids.last()) { + self.settle_writes(Some(self.intern(q.path))); + } + let client = self.client.clone(); let res = self.rt.block_on(async move { - client.walk(parent_fid, newfid, &[&name]).await?; + walked?; match client.getattr(newfid).await { Ok(a) => Ok(a), // The fid exists but we are not returning it, so release it. @@ -645,6 +677,7 @@ impl Filesystem for Fuse9p { Some(f) => f, None => return reply.error(libc::ENOENT), }; + self.settle_writes(Some(ino)); let client = self.client.clone(); match self.rt.block_on(async move { client.getattr(fid).await }) { Ok(a) => reply.attr(&self.tuning.attr_ttl, &to_fileattr(ino, &a)), @@ -713,6 +746,9 @@ impl Filesystem for Fuse9p { } p9mtime = time_or_now(t); } + // A truncate must land after the writes that came before it, and the size we report back + // has to include them. + self.settle_writes(Some(ino)); let client = self.client.clone(); let res = self.rt.block_on(async move { client @@ -732,13 +768,10 @@ impl Filesystem for Fuse9p { None => return reply.error(libc::ENOENT), }; let newfid = self.client.alloc_fid(); - let mut oflags = (flags as u32) & !(libc::O_CLOEXEC as u32); - // With the FUSE writeback cache the kernel issues READs on the handle for partial-page - // read-modify-write even when the file was opened write-only; the server would then answer - // those reads with EBADF on a write-only fid. Upgrade write-only opens to O_RDWR so the - // kernel's RMW reads succeed. (Harmless when off, but only needed with writeback on.) - if self.tuning.writeback && (oflags & libc::O_ACCMODE as u32) == libc::O_WRONLY as u32 { - oflags = (oflags & !(libc::O_ACCMODE as u32)) | libc::O_RDWR as u32; + let (oflags, pipelined) = self.open_flags(flags); + if oflags & libc::O_TRUNC as u32 != 0 { + // The server truncates at open; that must land after writes already queued. + self.settle_writes(Some(ino)); } let client = self.client.clone(); let res = self.rt.block_on(async move { @@ -755,9 +788,7 @@ impl Filesystem for Fuse9p { }); match res { Ok(()) => { - // Give writable opens a write-back tracker so writes can pipeline. - let writable = (oflags & libc::O_ACCMODE as u32) != libc::O_RDONLY as u32; - let fh = self.insert_handle(newfid, writable); + let fh = self.insert_handle(newfid, ino, pipelined); reply.opened(fh, 0); } Err(e) => reply.error(e), @@ -767,7 +798,7 @@ impl Filesystem for Fuse9p { fn read( &mut self, _req: &Request, - _ino: u64, + ino: u64, fh: u64, offset: i64, size: u32, @@ -779,6 +810,7 @@ impl Filesystem for Fuse9p { Some(h) => h.fid, None => return reply.error(libc::EBADF), }; + self.settle_writes(Some(ino)); let client = self.client.clone(); let cap = size.min(self.client.msize.saturating_sub(24)); match self @@ -803,54 +835,41 @@ impl Filesystem for Fuse9p { _lock: Option, reply: ReplyWrite, ) { - let (fid, wb) = match self.handles.get(&fh) { - Some(h) => (h.fid, h.wb.clone()), + let (fid, ino, wb) = match self.handles.get(&fh) { + Some(h) => (h.fid, h.ino, h.wb.clone()), None => return reply.error(libc::EBADF), }; - // Chunk to the negotiated msize (leave headroom for the Twrite header). - let chunk = self.client.msize.saturating_sub(24) as usize; - let client = self.client.clone(); let data = data.to_vec(); let len = data.len() as u32; let off = offset as u64; - let wb = match wb { - // Writable handle: defer the Twrite(s) to a background task so successive writes pipeline - // over the ws (depth-bounded by the semaphore). Ack the bytes optimistically; any error - // is recorded and surfaced at flush/fsync/release -- standard write-back semantics. - Some(wb) => wb, - // Read-only handle being written (shouldn't happen): fall back to a synchronous write. - None => { + match (wb, self.writes.get(&ino).cloned()) { + // Pipelined handle: queue the write and acknowledge the bytes at once. An error is + // recorded and surfaced at flush/fsync/release -- standard write-back semantics. + (Some(err), Some(pipeline)) => { + let w = Write { + fid, + off, + data, + err, + }; + self.rt.block_on(async move { pipeline.write(w).await }); + reply.written(len); + } + // Synchronous handle: one round-trip, after any pipelined writes to the same file. + _ => { + self.settle_writes(Some(ino)); + let client = self.client.clone(); + let chunk = self.write_chunk(); let res = self .rt .block_on(async move { write_all(&client, fid, off, &data, chunk).await }); - return match res { + match res { Ok(()) => reply.written(len), Err(e) => reply.error(e), - }; - } - }; - - self.rt.block_on(async { - // Backpressure: block here once wb_depth writes are already in flight. - let permit = wb - .sem - .clone() - .acquire_owned() - .await - .expect("write-back semaphore closed"); - let wb2 = wb.clone(); - tokio::spawn(async move { - let _permit = permit; // released (slot freed) when this write finishes - if let Err(e) = write_all(&client, fid, off, &data, chunk).await { - let mut g = wb2.err.lock().unwrap(); - if g.is_none() { - *g = Some(e); - } } - }); - }); - reply.written(len); + } + } } fn release( @@ -866,10 +885,19 @@ impl Filesystem for Fuse9p { if let Some(h) = self.handles.remove(&fh) { // Finish any deferred writes before clunking the fid (and surface a write error here if // the app never called fsync/flush). - let err = h.wb.and_then(|wb| self.drain_writeback(wb)); + let err = h.wb.as_ref().and_then(|wb| self.drain_writeback(h.ino, wb)); let fid = h.fid; let c = self.client.clone(); let _ = self.rt.block_on(async move { c.clunk(fid).await }); + // Drop the inode's pipeline with its last pipelined handle; it is idle after the drain. + if h.wb.is_some() + && !self + .handles + .values() + .any(|o| o.ino == h.ino && o.wb.is_some()) + { + self.writes.remove(&h.ino); + } if let Some(e) = err { return reply.error(e); } @@ -899,7 +927,7 @@ impl Filesystem for Fuse9p { }); match res { Ok(()) => { - let fh = self.insert_handle(newfid, false); + let fh = self.insert_handle(newfid, ino, false); reply.opened(fh, 0); } Err(e) => reply.error(e), @@ -967,6 +995,8 @@ impl Filesystem for Fuse9p { Err(e) => return reply.error(e), }; + // The prefetched attrs include sizes, so they must not miss our own pending writes. + self.settle_writes(None); // One fid per entry; walk+getattr them all concurrently (the client multiplexes by tag). let fids: Vec = entries.iter().map(|_| self.client.alloc_fid()).collect(); let names: Vec = entries.iter().map(|e| e.name.clone()).collect(); @@ -1070,7 +1100,7 @@ impl Filesystem for Fuse9p { let gid = req.gid(); let openfid = self.client.alloc_fid(); let basefid = self.client.alloc_fid(); - let oflags = (flags as u32) & !(libc::O_CLOEXEC as u32); + let (oflags, pipelined) = self.open_flags(flags); let client = self.client.clone(); let res = self.rt.block_on(async move { // Clone the parent into openfid and create+open the new file through it. Each step @@ -1107,8 +1137,7 @@ impl Filesystem for Fuse9p { lookups: 1, }, ); - // The created file is opened for writing -> give it a write-back tracker. - let fh = self.insert_handle(openfid, true); + let fh = self.insert_handle(openfid, ino, pipelined); reply.created(&self.tuning.entry_ttl, &to_fileattr(ino, &attr), 0, fh, 0); } Err(e) => reply.error(e), @@ -1171,8 +1200,12 @@ impl Filesystem for Fuse9p { fn flush(&mut self, _req: &Request, _ino: u64, fh: u64, _lock: u64, reply: ReplyEmpty) { // flush() runs on every close(2). Drain deferred writes so an error reaches the app's // close() return, matching POSIX write-back expectations. - if let Some(wb) = self.handles.get(&fh).and_then(|h| h.wb.clone()) { - if let Some(e) = self.drain_writeback(wb) { + if let Some((ino, wb)) = self + .handles + .get(&fh) + .and_then(|h| h.wb.clone().map(|wb| (h.ino, wb))) + { + if let Some(e) = self.drain_writeback(ino, &wb) { return reply.error(e); } } @@ -1180,13 +1213,13 @@ impl Filesystem for Fuse9p { } fn fsync(&mut self, _req: &Request, _ino: u64, fh: u64, _datasync: bool, reply: ReplyEmpty) { - let (fid, wb) = match self.handles.get(&fh) { - Some(h) => (h.fid, h.wb.clone()), + let (fid, ino, wb) = match self.handles.get(&fh) { + Some(h) => (h.fid, h.ino, h.wb.clone()), None => return reply.ok(), }; // Make deferred writes durable: first drain our in-flight Twrites, then Tfsync on the server. if let Some(wb) = wb { - if let Some(e) = self.drain_writeback(wb) { + if let Some(e) = self.drain_writeback(ino, &wb) { return reply.error(e); } } diff --git a/src/lib.rs b/src/lib.rs index f7d9462..19086fc 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -26,6 +26,7 @@ pub mod fuse9p; pub mod mount9p; pub mod ninep; pub mod transport; +mod write_pipeline; pub use client::NineClient; pub use fuse9p::{Fuse9p, Tuning}; @@ -42,6 +43,9 @@ use std::path::Path; /// - `uid` is the identity to attach as (`n_uname`); the server acts as this user for file ops. /// - `aname` is the export name to attach (must match the server's export, e.g. `"/export"`). /// - `tuning` controls the caching / write-back knobs (see [`Tuning`]). +/// +/// Invalidation requests are read from the process's stdin and answered on its stdout; use +/// [`mount_with_invalidations`] to use other streams. pub async fn mount( transport: Box, mountpoint: &Path, @@ -49,6 +53,34 @@ pub async fn mount( uid: u32, aname: &str, tuning: Tuning, +) -> Result<(), Box> { + mount_with_invalidations( + transport, + mountpoint, + msize, + uid, + aname, + tuning, + Box::new(tokio::io::stdin()), + Box::new(tokio::io::stdout()), + ) + .await +} + +/// Like [`mount`], but reads invalidation requests from `invalidations` and writes their replies to +/// `replies`. Each request names a path relative to the mount root and drops its cached entry, +/// attributes and data from the kernel, so the next access sees changes made to the export from +/// outside the mount; its reply is written once that is done. +#[allow(clippy::too_many_arguments)] +pub async fn mount_with_invalidations( + transport: Box, + mountpoint: &Path, + msize: u32, + uid: u32, + aname: &str, + tuning: Tuning, + invalidations: Box, + replies: Box, ) -> Result<(), Box> { Fuse9p::run( transport, @@ -60,6 +92,8 @@ pub async fn mount( true, true, std::time::Duration::from_secs(30), + invalidations, + replies, ) .await } diff --git a/src/main.rs b/src/main.rs index 637a106..1f5c5fa 100644 --- a/src/main.rs +++ b/src/main.rs @@ -94,11 +94,21 @@ enum Cmd { /// experiment. Metadata speed comes from the attr/entry caches above, not this. #[arg(long, default_value_t = false, action = clap::ArgAction::Set)] readdirplus: bool, - /// Pipeline writes (async write-back + kernel writeback cache). Pass false for a synchronous - /// round-trip per write (the un-tuned baseline). + /// Let the kernel buffer writes in its page cache (FUSE writeback cache). OFF by default: with + /// it on, a file grown from outside the mount reads back truncated to its old size while the + /// kernel holds its inode, even after invalidation. Only enable it when nothing else writes + /// to the export. + #[arg(long, default_value_t = false, action = clap::ArgAction::Set)] + writeback_cache: bool, + /// Acknowledge writes immediately and keep up to --wb-depth Twrites per file in flight. Pass + /// false for a synchronous round-trip per write (the un-tuned baseline). #[arg(long, default_value_t = true, action = clap::ArgAction::Set)] - writeback: bool, - /// Max concurrent in-flight Twrites per file when writeback is on (pipeline depth). + write_pipeline: bool, + /// Deprecated: sets both --writeback-cache and --write-pipeline to the given value. + #[arg(long, hide = true, action = clap::ArgAction::Set, + conflicts_with_all = ["writeback_cache", "write_pipeline"])] + writeback: Option, + /// Max concurrent in-flight Twrites per file when --write-pipeline is on (pipeline depth). #[arg(long, default_value_t = 16)] wb_depth: usize, @@ -165,6 +175,8 @@ async fn main() -> Result<(), Box> { entry_ttl, negative_ttl, readdirplus, + writeback_cache, + write_pipeline, writeback, wb_depth, detach_on_transport_loss, @@ -179,7 +191,8 @@ async fn main() -> Result<(), Box> { negative_ttl: (negative_ttl > 0) .then(|| std::time::Duration::from_secs(negative_ttl)), readdirplus, - writeback, + writeback_cache: writeback.unwrap_or(writeback_cache), + write_pipeline: writeback.unwrap_or(write_pipeline), wb_depth, }; fuse9p::Fuse9p::run( @@ -192,6 +205,8 @@ async fn main() -> Result<(), Box> { detach_on_transport_loss, default_permissions, std::time::Duration::from_secs(stall_timeout), + Box::new(tokio::io::stdin()), + Box::new(tokio::io::stdout()), ) .await } diff --git a/src/write_pipeline.rs b/src/write_pipeline.rs new file mode 100644 index 0000000..2ed4b05 --- /dev/null +++ b/src/write_pipeline.rs @@ -0,0 +1,143 @@ +//! Pipelined writes for one inode. A FUSE `write` is acknowledged as soon as it is queued here; the +//! `Twrite`s go out from background tasks, up to `depth` at a time. +//! +//! The FUSE session calls `write` and `settle` from one thread at a time, so only the background +//! tasks run concurrently with them. + +use crate::client::NineClient; +use std::sync::{Arc, Mutex}; +use tokio::sync::{watch, OwnedSemaphorePermit, Semaphore}; + +/// The first error hit by a background write through one handle, reported at its next +/// flush/fsync/release. +pub(crate) type WriteError = Arc>>; + +/// One application write: the bytes, where they go, and the fid and error slot of its handle. +pub(crate) struct Write { + pub fid: u32, + pub off: u64, + pub data: Vec, + pub err: WriteError, +} + +/// The write pipeline of one inode, shared by all its open handles. +pub(crate) struct InodeWrites { + client: Arc, + /// The most bytes one `Twrite` carries. + chunk: usize, + /// One permit per write in flight. + slots: Arc, + state: Mutex, +} + +#[derive(Default)] +struct State { + /// Writes handed to a task and not yet finished, oldest first. + inflight: Vec, + next_id: u64, +} + +struct InFlight { + id: u64, + /// Its sender is dropped when the write finishes, which wakes everything waiting on it. + done: watch::Receiver<()>, +} + +impl InodeWrites { + pub(crate) fn new(client: Arc, chunk: usize, depth: usize) -> Arc { + Arc::new(InodeWrites { + client, + chunk: chunk.max(1), + slots: Arc::new(Semaphore::new(depth.max(1))), + state: Mutex::new(State::default()), + }) + } + + /// Whether any write is still unsent or in flight. + pub(crate) fn busy(&self) -> bool { + let st = self.state.lock().unwrap(); + !st.inflight.is_empty() + } + + /// Queue `w`. Returns once it is handed to a background task; blocks while every slot is busy. + pub(crate) async fn write(self: &Arc, w: Write) { + let permit = self.acquire().await; + self.start(permit, w); + } + + /// Wait until every write on this inode has finished. + pub(crate) async fn settle(&self) { + let waits: Vec> = { + let st = self.state.lock().unwrap(); + st.inflight.iter().map(|f| f.done.clone()).collect() + }; + for mut done in waits { + let _ = done.changed().await; + } + } + + async fn acquire(&self) -> OwnedSemaphorePermit { + self.slots + .clone() + .acquire_owned() + .await + .expect("write slots semaphore closed") + } + + fn start(self: &Arc, permit: OwnedSemaphorePermit, w: Write) { + let started = { + let mut st = self.state.lock().unwrap(); + register(&mut st) + }; + tokio::spawn(self.clone().run(permit, w, started)); + } + + async fn run(self: Arc, permit: OwnedSemaphorePermit, w: Write, started: Started) { + if let Err(e) = write_all(&self.client, w.fid, w.off, &w.data, self.chunk).await { + w.err.lock().unwrap().get_or_insert(e); + } + self.state + .lock() + .unwrap() + .inflight + .retain(|f| f.id != started.id); + drop(permit); + } +} + +/// A write just added to `inflight`: its id and the sender whose drop marks it finished. +struct Started { + id: u64, + _done: watch::Sender<()>, +} + +fn register(st: &mut State) -> Started { + let (tx, rx) = watch::channel(()); + let id = st.next_id; + st.next_id += 1; + st.inflight.push(InFlight { id, done: rx }); + Started { id, _done: tx } +} + +/// Write `data` to `fid` starting at byte `off`, split into `chunk`-sized `Twrite`s. A short write of +/// 0 is treated as an error so callers don't spin. +pub(crate) async fn write_all( + client: &NineClient, + fid: u32, + off: u64, + data: &[u8], + chunk: usize, +) -> Result<(), i32> { + let mut pos = 0usize; + let mut o = off; + while pos < data.len() { + let end = (pos + chunk.max(1)).min(data.len()); + let n = client.write(fid, o, &data[pos..end]).await?; + if n == 0 { + return Err(libc::EIO); + } + pos += n as usize; + o += n as u64; + } + Ok(()) +} diff --git a/tests/diod_fuse.rs b/tests/diod_fuse.rs index 5fb0bbf..b0899f6 100644 --- a/tests/diod_fuse.rs +++ b/tests/diod_fuse.rs @@ -12,9 +12,10 @@ use futures_util::{SinkExt, StreamExt}; use p9fuse::{ - mount, NineClient, NineTransport, TcpTransport, Tuning, UnixTransport, WebSocketTransport, + mount_with_invalidations, NineClient, NineTransport, TcpTransport, Tuning, UnixTransport, + WebSocketTransport, }; -use std::io::Write; +use std::io::{BufRead, Write}; use std::os::unix::fs::PermissionsExt; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; @@ -277,12 +278,28 @@ async fn ws_tcp_bridge( tokio::select! { _ = c2s => {}, _ = s2c => {} } } +/// What a test body gets: the mountpoint, the export directory behind it (to change files from +/// outside the mount), and a socket to send the mount invalidation requests and read their replies. +struct Mounted<'a> { + mp: &'a Path, + export: &'a Path, + inval: std::os::unix::net::UnixStream, +} + /// Start diod, reach it via `via`, FUSE-mount the export with `tuning`, wait until the mount is live, /// run `body` (on a blocking thread, with the mountpoint path), then unmount and tear down. Any panic /// in `body` propagates to fail the test. async fn with_mount_via(via: Via, tuning: Tuning, body: F) where F: FnOnce(&Path) + Send + 'static, +{ + with_mount_full(via, tuning, |m| body(m.mp)).await; +} + +/// [`with_mount_via`], but the body also gets the export directory and the invalidation input. +async fn with_mount_full(via: Via, tuning: Tuning, body: F) +where + F: FnOnce(Mounted) + Send + 'static, { let diod = Diod::start(); // Seed a readiness sentinel in the export; it appears through the mount once FUSE is live. @@ -309,6 +326,12 @@ where } }; + let (inval_tx, inval_rx) = std::os::unix::net::UnixStream::pair().unwrap(); + inval_rx.set_nonblocking(true).unwrap(); + let (inval_rx, inval_replies) = tokio::net::UnixStream::from_std(inval_rx) + .unwrap() + .into_split(); + let mountpoint = tempfile::tempdir().unwrap(); let mp = mountpoint.path().to_path_buf(); let mp_task = mp.clone(); @@ -316,9 +339,18 @@ where let task = tokio::spawn(async move { let transport = connect_transport(connect).await?; // `mount` blocks until unmounted; the error (if any) surfaces after umount below. - mount(transport, &mp_task, 512_000, uid, &aname, tuning) - .await - .map_err(|e| e.to_string()) + mount_with_invalidations( + transport, + &mp_task, + 512_000, + uid, + &aname, + tuning, + Box::new(inval_rx), + Box::new(inval_replies), + ) + .await + .map_err(|e| e.to_string()) }); // Wait for the mount to become live (sentinel visible through FUSE). @@ -337,7 +369,15 @@ where } let mp_body = mp.clone(); - let outcome = tokio::task::spawn_blocking(move || body(&mp_body)).await; + let export = diod.export().to_path_buf(); + let outcome = tokio::task::spawn_blocking(move || { + body(Mounted { + mp: &mp_body, + export: &export, + inval: inval_tx, + }) + }) + .await; umount(&mp); let mount_result = task.await.expect("mount task join"); @@ -453,7 +493,7 @@ async fn rename_then_read() { } /// A multi-megabyte write with write-back pipelining round-trips byte-for-byte, and the file's size -/// is correct on the server side. +/// is correct on the server side. Run with the kernel writeback cache both off and on. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn writeback_large_file() { if let Some(reason) = skip_reason() { @@ -465,23 +505,29 @@ async fn writeback_large_file() { let payload: Vec = (0..4u64 * 1024 * 1024) .map(|i| i.wrapping_mul(2654435761).rotate_right(13) as u8) .collect(); - let expected = payload.clone(); - - with_mount(Tuning::default(), move |mp| { - let path = mp.join("big.bin"); - { - let mut f = std::fs::File::create(&path).unwrap(); - // Write in odd-sized chunks to exercise the write-back pipeline across offsets. - for chunk in payload.chunks(7919) { - f.write_all(chunk).unwrap(); + for writeback_cache in [false, true] { + let payload = payload.clone(); + let expected = payload.clone(); + let tuning = Tuning { + writeback_cache, + ..Tuning::default() + }; + with_mount(tuning, move |mp| { + let path = mp.join("big.bin"); + { + let mut f = std::fs::File::create(&path).unwrap(); + // Write in odd-sized chunks to exercise the write-back pipeline across offsets. + for chunk in payload.chunks(7919) { + f.write_all(chunk).unwrap(); + } + f.sync_all().unwrap(); // flush the pipeline } - f.sync_all().unwrap(); // flush the pipeline - } - let got = std::fs::read(&path).unwrap(); - assert_eq!(got.len(), expected.len(), "size mismatch after write-back"); - assert!(got == expected, "content mismatch after write-back"); - }) - .await; + let got = std::fs::read(&path).unwrap(); + assert_eq!(got.len(), expected.len(), "size mismatch after write-back"); + assert!(got == expected, "content mismatch after write-back"); + }) + .await; + } } /// `mv src dst` where `dst` already exists: the overwritten file's inode is discarded and `dst` now @@ -560,7 +606,7 @@ async fn client_rename_then_walk() { ); } -/// With write-back disabled, each write is a synchronous round-trip -- data must still be intact. +/// With the write pipeline off, each write is a synchronous round-trip -- data must still be intact. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn synchronous_writes() { if let Some(reason) = skip_reason() { @@ -568,7 +614,7 @@ async fn synchronous_writes() { return; } let tuning = Tuning { - writeback: false, + write_pipeline: false, ..Tuning::default() }; with_mount(tuning, move |mp| { @@ -582,6 +628,69 @@ async fn synchronous_writes() { .await; } +/// A file grown from outside the mount reads back whole once invalidated, even while an open fd keeps +/// its inode cached. Under the FUSE writeback cache the kernel keeps its own stale size for the +/// pinned inode and the read comes back cut to the old length, which is why the cache is off by +/// default. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn external_growth_with_pinned_inode() { + if let Some(reason) = skip_reason() { + eprintln!("SKIP external_growth_with_pinned_inode: {reason}"); + return; + } + with_mount_full(Via::Tcp, Tuning::default(), |mut m| { + let long: &[u8] = b"#let title = \"Changed in the export\""; + std::fs::write(m.export.join("f"), b"#let title = \"Original\"").unwrap(); + assert_eq!( + std::fs::read(m.mp.join("f")).unwrap(), + b"#let title = \"Original\"" + ); + + let _pin = std::fs::File::open(m.mp.join("f")).unwrap(); + std::fs::write(m.export.join("f"), long).unwrap(); + writeln!(m.inval, "inval 1 f").unwrap(); + let mut reply = String::new(); + std::io::BufReader::new(&m.inval) + .read_line(&mut reply) + .unwrap(); + assert_eq!(reply, "ok 1\n"); + + let got = std::fs::read(m.mp.join("f")).unwrap(); + assert_eq!(String::from_utf8_lossy(&got), String::from_utf8_lossy(long)); + }) + .await; +} + +/// `stat` and a read through a second fd see every write made so far on a still-open writer, even +/// though the write pipeline has acknowledged writes the server may not have yet. Without the +/// writeback cache the kernel takes size and data from us, so we have to wait for those writes. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn stat_and_read_see_pipelined_writes() { + if let Some(reason) = skip_reason() { + eprintln!("SKIP stat_and_read_see_pipelined_writes: {reason}"); + return; + } + with_mount(Tuning::default(), |mp| { + const LEN: u64 = 64 * 131072; + for trial in 0..10 { + let p = mp.join(format!("t{trial}")); + let mut w = std::fs::File::create(&p).unwrap(); + for i in 0..64u8 { + w.write_all(&[b'A' + i % 26; 131072]).unwrap(); + } + + let size = std::fs::metadata(&p).unwrap().len(); + let read = std::fs::read(&p).unwrap().len() as u64; + assert_eq!( + (size, read), + (LEN, LEN), + "trial {trial}: (stat, read) lengths" + ); + } + }) + .await; +} + /// The same core ops over the Unix-socket transport (relayed to diod's TCP port). #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn unix_transport() { @@ -690,49 +799,31 @@ async fn set_modified_time() { /// Overwriting at an offset on a *write-only* handle: with the writeback cache on this is a /// partial-page read-modify-write, so the kernel issues reads on the write-only handle -- which only -/// works because `open` upgrades write-only opens to O_RDWR under writeback (else the server EBADFs). +/// works because `open` upgrades write-only opens to O_RDWR under the cache (else the server EBADFs). +/// Run with the cache both on and off. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn offset_writes() { if let Some(reason) = skip_reason() { eprintln!("SKIP offset_writes: {reason}"); return; } - with_mount(Tuning::default(), |mp| { - use std::io::{Seek, SeekFrom, Write}; - let p = mp.join("o"); - std::fs::write(&p, b"hello world").unwrap(); - let mut f = std::fs::OpenOptions::new().write(true).open(&p).unwrap(); - f.seek(SeekFrom::Start(6)).unwrap(); - f.write_all(b"WORLD").unwrap(); - drop(f); - assert_eq!(std::fs::read(&p).unwrap(), b"hello WORLD"); - }) - .await; -} - -/// Appending (O_APPEND) lands at end-of-file. Tested with write-back off: under the FUSE writeback -/// cache the kernel handles O_APPEND itself, which is not reliable, so append semantics are only -/// well-defined in the synchronous (writeback-off) mode. -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn append_writes() { - if let Some(reason) = skip_reason() { - eprintln!("SKIP append_writes: {reason}"); - return; + for writeback_cache in [false, true] { + let tuning = Tuning { + writeback_cache, + ..Tuning::default() + }; + with_mount(tuning, |mp| { + use std::io::{Seek, SeekFrom, Write}; + let p = mp.join("o"); + std::fs::write(&p, b"hello world").unwrap(); + let mut f = std::fs::OpenOptions::new().write(true).open(&p).unwrap(); + f.seek(SeekFrom::Start(6)).unwrap(); + f.write_all(b"WORLD").unwrap(); + drop(f); + assert_eq!(std::fs::read(&p).unwrap(), b"hello WORLD"); + }) + .await; } - let tuning = Tuning { - writeback: false, - ..Tuning::default() - }; - with_mount(tuning, |mp| { - use std::io::Write; - let p = mp.join("a"); - std::fs::write(&p, b"hello").unwrap(); - let mut f = std::fs::OpenOptions::new().append(true).open(&p).unwrap(); - f.write_all(b" world").unwrap(); - drop(f); - assert_eq!(std::fs::read(&p).unwrap(), b"hello world"); - }) - .await; } /// A directory with enough entries to span multiple `Treaddir` round-trips lists every entry. From 88cf2a4964f846c19c2348af8dcfa82fd708b07d Mon Sep 17 00:00:00 2001 From: thomasjm Date: Sat, 14 Nov 2026 03:37:38 -0800 Subject: [PATCH 2/4] Keep overlapping and O_APPEND pipelined writes in order --- README.md | 2 +- src/fuse9p.rs | 28 +++++-- src/write_pipeline.rs | 43 +++++++++-- tests/diod_fuse.rs | 171 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 229 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 8460a9f..e07bb9c 100644 --- a/README.md +++ b/README.md @@ -89,7 +89,7 @@ multiple clients write concurrently. | `attr_ttl` | 60s | How long the kernel trusts a cached `getattr`. `0` = round-trip every stat. | | `entry_ttl` | 60s | How long the kernel trusts a cached name→inode lookup. | | `negative_ttl` | 5s | Cache "no such file" lookups (skip round-trips on `$PATH`/include probes). | -| `write_pipeline` | on | Acknowledge writes at once and send their `Twrite`s in the background instead of one round-trip per write. | +| `write_pipeline` | on | Acknowledge writes at once and send their `Twrite`s in the background instead of one round-trip per write. Overlapping writes stay in order. | | `wb_depth` | 16 | Max in-flight `Twrite`s per file when the write pipeline is on. | | `writeback_cache` | off | Let the kernel buffer writes in its page cache. Faster for small writes, but only safe when nothing else writes to the export: the kernel then keeps its own file sizes, so a file grown from outside the mount reads back truncated while its inode is cached, even after invalidation. | | `readdirplus` | off | Prefetch entry attrs on `readdir`. Off by default — usually net-negative. | diff --git a/src/fuse9p.rs b/src/fuse9p.rs index 4cdbb2e..e7031cc 100644 --- a/src/fuse9p.rs +++ b/src/fuse9p.rs @@ -52,8 +52,9 @@ pub struct Tuning { pub writeback_cache: bool, /// Pipeline writes: acknowledge each FUSE `write` at once and send its `Twrite` from a background /// task, up to `wb_depth` in flight per file, turning latency-bound sequential writes into - /// throughput. Errors surface at flush/fsync/release. Off = each `write` is one synchronous - /// `Twrite` round-trip. + /// throughput. A write waits for overlapping writes still in flight. Errors surface at + /// flush/fsync/release. Off = each `write` is one synchronous `Twrite` round-trip. O_APPEND + /// handles write synchronously unless `writeback_cache` is on. pub write_pipeline: bool, /// Max concurrent in-flight `Twrite`s per file when `write_pipeline` is on. Bounds memory /// (~`wb_depth` * msize) and sets the pipeline depth. @@ -141,13 +142,24 @@ impl Fuse9p { fn open_flags(&self, flags: i32) -> (u32, bool) { let mut oflags = (flags as u32) & !(libc::O_CLOEXEC as u32); let acc = oflags & libc::O_ACCMODE as u32; - // With the FUSE writeback cache the kernel issues READs on the handle for partial-page - // read-modify-write even when the file was opened write-only, which the server would - // refuse with EBADF on a write-only fid; so open those read-write. - if self.tuning.writeback_cache && acc == libc::O_WRONLY as u32 { - oflags = (oflags & !(libc::O_ACCMODE as u32)) | libc::O_RDWR as u32; + let append = oflags & libc::O_APPEND as u32 != 0; + if self.tuning.writeback_cache { + // With the FUSE writeback cache the kernel issues READs on the handle for partial-page + // read-modify-write even when the file was opened write-only, which the server would + // refuse with EBADF on a write-only fid; so open those read-write. The kernel also + // places O_APPEND writes itself and may write back any dirty page through any of the + // file's handles, so the server must honour our offsets: drop O_APPEND. + if acc == libc::O_WRONLY as u32 { + oflags = (oflags & !(libc::O_ACCMODE as u32)) | libc::O_RDWR as u32; + } + oflags &= !(libc::O_APPEND as u32); } - let pipelined = acc != libc::O_RDONLY as u32 && self.tuning.write_pipeline; + // Without the cache an O_APPEND fid stays O_APPEND, so the server appends at its real end + // of file -- but then it ignores our offsets, and two appends in flight could land in either + // order. Those handles write synchronously. + let writable = acc != libc::O_RDONLY as u32; + let pipelined = + writable && self.tuning.write_pipeline && (self.tuning.writeback_cache || !append); (oflags, pipelined) } diff --git a/src/write_pipeline.rs b/src/write_pipeline.rs index 2ed4b05..5fc0b95 100644 --- a/src/write_pipeline.rs +++ b/src/write_pipeline.rs @@ -1,5 +1,7 @@ //! Pipelined writes for one inode. A FUSE `write` is acknowledged as soon as it is queued here; the -//! `Twrite`s go out from background tasks, up to `depth` at a time. +//! `Twrite`s go out from background tasks, up to `depth` at a time. A write never overtakes an +//! earlier write it overlaps: the server may apply concurrent `Twrite`s in any order, so a write +//! waits for every overlapping write still in flight before it is sent. //! //! The FUSE session calls `write` and `settle` from one thread at a time, so only the background //! tasks run concurrently with them. @@ -20,6 +22,12 @@ pub(crate) struct Write { pub err: WriteError, } +impl Write { + fn end(&self) -> u64 { + self.off + self.data.len() as u64 + } +} + /// The write pipeline of one inode, shared by all its open handles. pub(crate) struct InodeWrites { client: Arc, @@ -39,6 +47,8 @@ struct State { struct InFlight { id: u64, + start: u64, + end: u64, /// Its sender is dropped when the write finishes, which wakes everything waiting on it. done: watch::Receiver<()>, } @@ -87,12 +97,16 @@ impl InodeWrites { fn start(self: &Arc, permit: OwnedSemaphorePermit, w: Write) { let started = { let mut st = self.state.lock().unwrap(); - register(&mut st) + register(&mut st, &w) }; tokio::spawn(self.clone().run(permit, w, started)); } + /// Send `w` once the writes it overlaps have finished. async fn run(self: Arc, permit: OwnedSemaphorePermit, w: Write, started: Started) { + for mut done in started.waits { + let _ = done.changed().await; + } if let Err(e) = write_all(&self.client, w.fid, w.off, &w.data, self.chunk).await { w.err.lock().unwrap().get_or_insert(e); } @@ -105,18 +119,35 @@ impl InodeWrites { } } -/// A write just added to `inflight`: its id and the sender whose drop marks it finished. +/// A write just added to `inflight`: its id, the overlapping writes it must wait for, and the +/// sender whose drop marks it finished. struct Started { id: u64, + waits: Vec>, _done: watch::Sender<()>, } -fn register(st: &mut State) -> Started { +fn register(st: &mut State, w: &Write) -> Started { + let waits = st + .inflight + .iter() + .filter(|f| f.start < w.end() && w.off < f.end) + .map(|f| f.done.clone()) + .collect(); let (tx, rx) = watch::channel(()); let id = st.next_id; st.next_id += 1; - st.inflight.push(InFlight { id, done: rx }); - Started { id, _done: tx } + st.inflight.push(InFlight { + id, + start: w.off, + end: w.end(), + done: rx, + }); + Started { + id, + waits, + _done: tx, + } } /// Write `data` to `fid` starting at byte `off`, split into `chunk`-sized `Twrite`s. A short write of diff --git a/tests/diod_fuse.rs b/tests/diod_fuse.rs index b0899f6..c37a200 100644 --- a/tests/diod_fuse.rs +++ b/tests/diod_fuse.rs @@ -826,6 +826,177 @@ async fn offset_writes() { } } +/// The write modes: the default (pipeline, no kernel cache), the same with a pipeline one write +/// deep, pipeline plus kernel writeback cache, and fully synchronous writes. +fn write_modes() -> [(&'static str, Tuning); 4] { + [ + ("pipeline", Tuning::default()), + ( + "pipeline depth 1", + Tuning { + wb_depth: 1, + ..Tuning::default() + }, + ), + ( + "pipeline+cache", + Tuning { + writeback_cache: true, + ..Tuning::default() + }, + ), + ( + "sync", + Tuning { + write_pipeline: false, + ..Tuning::default() + }, + ), + ] +} + +/// A small deterministic PRNG (xorshift64*), so the randomized tests need no extra dependency and +/// a failure reproduces. +struct Rng(u64); + +impl Rng { + fn next(&mut self) -> u64 { + self.0 ^= self.0 >> 12; + self.0 ^= self.0 << 25; + self.0 ^= self.0 >> 27; + self.0.wrapping_mul(0x2545F4914F6CDD1D) + } + + fn below(&mut self, n: u64) -> u64 { + self.next() % n + } +} + +/// Many O_APPEND writes land at end-of-file in the order they were made, in every write mode. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn append_writes() { + if let Some(reason) = skip_reason() { + eprintln!("SKIP append_writes: {reason}"); + return; + } + for (mode, tuning) in write_modes() { + with_mount(tuning, move |mp| { + use std::io::Write; + let p = mp.join("a"); + std::fs::write(&p, b"start:").unwrap(); + let mut f = std::fs::OpenOptions::new().append(true).open(&p).unwrap(); + let mut expected = b"start:".to_vec(); + for i in 0..64u8 { + let chunk = [b'A' + i % 26; 8192]; + f.write_all(&chunk).unwrap(); + expected.extend_from_slice(&chunk); + } + drop(f); + + let got = std::fs::read(&p).unwrap(); + assert!(got == expected, "{mode}: appends out of order or lost"); + }) + .await; + } +} + +/// Rewriting the same region over and over leaves the last write's data, in every write mode. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn overlapping_writes_keep_order() { + if let Some(reason) = skip_reason() { + eprintln!("SKIP overlapping_writes_keep_order: {reason}"); + return; + } + for (mode, tuning) in write_modes() { + with_mount(tuning, move |mp| { + use std::os::unix::fs::FileExt; + for trial in 0..10 { + let p = mp.join(format!("o{trial}")); + let f = std::fs::File::create(&p).unwrap(); + for i in 0..64u8 { + f.write_all_at(&[i; 262_144], 0).unwrap(); + } + drop(f); + + let got = std::fs::read(&p).unwrap(); + assert!( + got == [63u8; 262_144], + "{mode}: trial {trial}: an older write won" + ); + } + }) + .await; + } +} + +/// Random writes (small and large, overlapping, through two handles on the same file), truncates +/// and mid-way stat/read checks all agree with an in-memory copy of the file, in every write mode. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn random_writes_match_model() { + if let Some(reason) = skip_reason() { + eprintln!("SKIP random_writes_match_model: {reason}"); + return; + } + for (mode, tuning) in write_modes() { + with_mount(tuning, move |mp| { + use std::os::unix::fs::FileExt; + let p = mp.join("r"); + let a = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(true) + .open(&p) + .unwrap(); + let b = std::fs::OpenOptions::new().write(true).open(&p).unwrap(); + let mut model: Vec = Vec::new(); + let mut rng = Rng(0x9E3779B97F4A7C15); + for step in 0..3000u32 { + let f = if rng.below(4) == 0 { &b } else { &a }; + match rng.below(40) { + 0 => { + let len = rng.below(300_000) as usize; + f.set_len(len as u64).unwrap(); + model.resize(len, 0); + } + 1 | 2 => { + let size = std::fs::metadata(&p).unwrap().len(); + let got = std::fs::read(&p).unwrap(); + assert_eq!(size, model.len() as u64, "{mode}: step {step}: stat size"); + assert!(got == model, "{mode}: step {step}: contents differ"); + } + _ => { + // Mostly runs of small sequential writes, with some big ones and jumps. + let off = if rng.below(3) == 0 { + rng.below(262_144) as usize + } else { + model.len().min(262_144) + }; + let len = 1 + if rng.below(5) == 0 { + rng.below(131_072) + } else { + rng.below(4096) + } as usize; + let data: Vec = (0..len).map(|i| (step as usize + i) as u8).collect(); + f.write_all_at(&data, off as u64).unwrap(); + if model.len() < off + len { + model.resize(off + len, 0); + } + model[off..off + len].copy_from_slice(&data); + } + } + } + drop(a); + drop(b); + + let got = std::fs::read(&p).unwrap(); + assert_eq!(got.len(), model.len(), "{mode}: final size"); + assert!(got == model, "{mode}: final contents differ"); + }) + .await; + } +} + /// A directory with enough entries to span multiple `Treaddir` round-trips lists every entry. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn large_directory_readdir() { From 809453c61b36f3ac071533613da9fa5b73e6bbee Mon Sep 17 00:00:00 2001 From: thomasjm Date: Sat, 14 Nov 2026 03:38:22 -0800 Subject: [PATCH 3/4] Merge small sequential writes while the write pipeline is full --- README.md | 2 +- src/fuse9p.rs | 3 +- src/write_pipeline.rs | 108 +++++++++++++++++++++++++++++++++--------- tests/diod_fuse.rs | 42 +++++++++++++++- 4 files changed, 129 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index e07bb9c..fd779d7 100644 --- a/README.md +++ b/README.md @@ -89,7 +89,7 @@ multiple clients write concurrently. | `attr_ttl` | 60s | How long the kernel trusts a cached `getattr`. `0` = round-trip every stat. | | `entry_ttl` | 60s | How long the kernel trusts a cached name→inode lookup. | | `negative_ttl` | 5s | Cache "no such file" lookups (skip round-trips on `$PATH`/include probes). | -| `write_pipeline` | on | Acknowledge writes at once and send their `Twrite`s in the background instead of one round-trip per write. Overlapping writes stay in order. | +| `write_pipeline` | on | Acknowledge writes at once and send their `Twrite`s in the background instead of one round-trip per write. Overlapping writes stay in order; adjacent small writes are merged while the pipeline is full. | | `wb_depth` | 16 | Max in-flight `Twrite`s per file when the write pipeline is on. | | `writeback_cache` | off | Let the kernel buffer writes in its page cache. Faster for small writes, but only safe when nothing else writes to the export: the kernel then keeps its own file sizes, so a file grown from outside the mount reads back truncated while its inode is cached, even after invalidation. | | `readdirplus` | off | Prefetch entry attrs on `readdir`. Off by default — usually net-negative. | diff --git a/src/fuse9p.rs b/src/fuse9p.rs index e7031cc..d94a26b 100644 --- a/src/fuse9p.rs +++ b/src/fuse9p.rs @@ -52,7 +52,8 @@ pub struct Tuning { pub writeback_cache: bool, /// Pipeline writes: acknowledge each FUSE `write` at once and send its `Twrite` from a background /// task, up to `wb_depth` in flight per file, turning latency-bound sequential writes into - /// throughput. A write waits for overlapping writes still in flight. Errors surface at + /// throughput. A write waits for overlapping writes still in flight, and when the pipeline is + /// full, adjacent small writes are merged into one `Twrite`. Errors surface at /// flush/fsync/release. Off = each `write` is one synchronous `Twrite` round-trip. O_APPEND /// handles write synchronously unless `writeback_cache` is on. pub write_pipeline: bool, diff --git a/src/write_pipeline.rs b/src/write_pipeline.rs index 5fc0b95..59fc67c 100644 --- a/src/write_pipeline.rs +++ b/src/write_pipeline.rs @@ -1,7 +1,13 @@ //! Pipelined writes for one inode. A FUSE `write` is acknowledged as soon as it is queued here; the -//! `Twrite`s go out from background tasks, up to `depth` at a time. A write never overtakes an -//! earlier write it overlaps: the server may apply concurrent `Twrite`s in any order, so a write -//! waits for every overlapping write still in flight before it is sent. +//! `Twrite`s go out from background tasks, up to `depth` at a time. Two rules keep the file's +//! contents what the application wrote: +//! +//! - A write never overtakes an earlier write it overlaps. The server may apply concurrent `Twrite`s +//! in any order, so a write waits for every overlapping write still in flight before it is sent. +//! - When every slot is busy, a write that continues exactly where the previous one ended is merged +//! into a single pending buffer (up to one `Twrite`'s worth) instead of waiting for a slot, and the +//! first write to finish sends it. Runs of small sequential writes over a slow link then go out +//! as a few large `Twrite`s; on an idle pipeline nothing is held back. //! //! The FUSE session calls `write` and `settle` from one thread at a time, so only the background //! tasks run concurrently with them. @@ -31,9 +37,10 @@ impl Write { /// The write pipeline of one inode, shared by all its open handles. pub(crate) struct InodeWrites { client: Arc, - /// The most bytes one `Twrite` carries. + /// The most bytes one `Twrite` carries; also the cap on a merged pending buffer. chunk: usize, - /// One permit per write in flight. + /// One permit per `Twrite` in flight. Every held permit belongs to a running task, which is + /// what guarantees a pending buffer is eventually sent. slots: Arc, state: Mutex, } @@ -42,6 +49,9 @@ pub(crate) struct InodeWrites { struct State { /// Writes handed to a task and not yet finished, oldest first. inflight: Vec, + /// Newer than everything in `inflight`; sent by the next task to finish, or by the next write + /// or settle that cannot merge into it. + pending: Option, next_id: u64, } @@ -66,17 +76,54 @@ impl InodeWrites { /// Whether any write is still unsent or in flight. pub(crate) fn busy(&self) -> bool { let st = self.state.lock().unwrap(); - !st.inflight.is_empty() + st.pending.is_some() || !st.inflight.is_empty() } - /// Queue `w`. Returns once it is handed to a background task; blocks while every slot is busy. + /// Queue `w`. Returns once it is merged into the pending buffer or handed to a background task; + /// blocks only while every slot is busy and `w` cannot be merged. pub(crate) async fn write(self: &Arc, w: Write) { - let permit = self.acquire().await; + let earlier = { + let mut st = self.state.lock().unwrap(); + if let Some(p) = st.pending.as_mut() { + if p.fid == w.fid && p.end() == w.off && p.data.len() + w.data.len() <= self.chunk { + p.data.extend_from_slice(&w.data); + return; + } + } + st.pending.take() + }; + // The pending buffer is older than `w`, so it goes out first. + if let Some(p) = earlier { + let permit = self.acquire().await; + self.start(permit, p); + } + let permit = { + let mut st = self.state.lock().unwrap(); + // Checked under the lock that tasks release their permits under: if this fails, some + // task still holds a slot and will see the pending buffer when it finishes. + match self.slots.clone().try_acquire_owned() { + Ok(permit) => Some(permit), + Err(_) if w.data.len() < self.chunk => { + st.pending = Some(w); + return; + } + Err(_) => None, + } + }; + let permit = match permit { + Some(permit) => permit, + None => self.acquire().await, + }; self.start(permit, w); } - /// Wait until every write on this inode has finished. - pub(crate) async fn settle(&self) { + /// Send the pending buffer and wait until every write on this inode has finished. + pub(crate) async fn settle(self: &Arc) { + let earlier = self.state.lock().unwrap().pending.take(); + if let Some(p) = earlier { + let permit = self.acquire().await; + self.start(permit, p); + } let waits: Vec> = { let st = self.state.lock().unwrap(); st.inflight.iter().map(|f| f.done.clone()).collect() @@ -102,20 +149,35 @@ impl InodeWrites { tokio::spawn(self.clone().run(permit, w, started)); } - /// Send `w` once the writes it overlaps have finished. - async fn run(self: Arc, permit: OwnedSemaphorePermit, w: Write, started: Started) { - for mut done in started.waits { - let _ = done.changed().await; - } - if let Err(e) = write_all(&self.client, w.fid, w.off, &w.data, self.chunk).await { - w.err.lock().unwrap().get_or_insert(e); + /// Send `w` once the writes it overlaps have finished, then keep the slot for the pending buffer + /// if there is one. + async fn run( + self: Arc, + permit: OwnedSemaphorePermit, + mut w: Write, + mut started: Started, + ) { + loop { + for mut done in std::mem::take(&mut started.waits) { + let _ = done.changed().await; + } + if let Err(e) = write_all(&self.client, w.fid, w.off, &w.data, self.chunk).await { + w.err.lock().unwrap().get_or_insert(e); + } + let mut st = self.state.lock().unwrap(); + st.inflight.retain(|f| f.id != started.id); + match st.pending.take() { + Some(next) => { + started = register(&mut st, &next); + w = next; + } + None => { + // Released under the lock; see `write`. + drop(permit); + return; + } + } } - self.state - .lock() - .unwrap() - .inflight - .retain(|f| f.id != started.id); - drop(permit); } } diff --git a/tests/diod_fuse.rs b/tests/diod_fuse.rs index c37a200..9d6f504 100644 --- a/tests/diod_fuse.rs +++ b/tests/diod_fuse.rs @@ -691,6 +691,45 @@ async fn stat_and_read_see_pipelined_writes() { .await; } +/// Writes held back for merging still reach the server when the application stops writing and +/// neither closes nor stats the file: the write that was in flight sends them when it finishes. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn merged_writes_reach_server_without_close() { + if let Some(reason) = skip_reason() { + eprintln!("SKIP merged_writes_reach_server_without_close: {reason}"); + return; + } + let tuning = Tuning { + wb_depth: 1, + ..Tuning::default() + }; + with_mount_full(Via::Tcp, tuning, |m| { + let mut f = std::fs::File::create(m.mp.join("m")).unwrap(); + let mut expected = Vec::new(); + // Small writes, then a large one immediately followed by a small one: the last write + // arrives while the large one is in flight, so it is held back. + let mut chunks: Vec> = (0..256u32).map(|i| vec![(i % 251) as u8; 1000]).collect(); + chunks.push(vec![b'L'; 400_000]); + chunks.push(vec![b'E'; 1000]); + for chunk in chunks { + f.write_all(&chunk).unwrap(); + expected.extend_from_slice(&chunk); + } + + // Read the export directly, so nothing on the mount side forces the writes out. + let deadline = Instant::now() + Duration::from_secs(5); + let mut got = std::fs::read(m.export.join("m")).unwrap(); + while got != expected && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(50)); + got = std::fs::read(m.export.join("m")).unwrap(); + } + assert_eq!(got.len(), expected.len(), "bytes on the server"); + assert!(got == expected, "contents on the server"); + drop(f); + }) + .await; +} + /// The same core ops over the Unix-socket transport (relayed to diod's TCP port). #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn unix_transport() { @@ -827,7 +866,8 @@ async fn offset_writes() { } /// The write modes: the default (pipeline, no kernel cache), the same with a pipeline one write -/// deep, pipeline plus kernel writeback cache, and fully synchronous writes. +/// deep (so nearly every write arriving while another is in flight is merged), pipeline plus kernel +/// writeback cache, and fully synchronous writes. fn write_modes() -> [(&'static str, Tuning); 4] { [ ("pipeline", Tuning::default()), From 8000bea19e8ad7ee43d6a0ebdd1aab67c916501b Mon Sep 17 00:00:00 2001 From: thomasjm Date: Sat, 14 Nov 2026 04:23:47 -0800 Subject: [PATCH 4/4] Stop the diod test harness hanging or racing for ports --- tests/diod_fuse.rs | 126 ++++++++++++++++++++++++++++++++------------- 1 file changed, 89 insertions(+), 37 deletions(-) diff --git a/tests/diod_fuse.rs b/tests/diod_fuse.rs index 9d6f504..fdcdc03 100644 --- a/tests/diod_fuse.rs +++ b/tests/diod_fuse.rs @@ -107,42 +107,41 @@ struct Diod { } impl Diod { - // The child is reaped in `Drop` (kill + wait); clippy can't see through the struct. - #[allow(clippy::zombie_processes)] + /// Start diod on a free port. The port is only free when we look: another test may take it + /// before diod binds it, so wait for diod itself to say it is listening, and try another port + /// if it couldn't bind. fn start() -> Diod { let diod = find_diod().expect("diod (checked by skip_reason)"); let export = tempfile::tempdir().unwrap(); - let port = free_port(); - let child = Command::new(&diod) - .args([ - "-f", // foreground - "-n", // no auth - "-N", // bypass userdb, so a rootless diod needs no passwd/group entry for our uid - "-L", - "stderr", - "-l", - &format!("127.0.0.1:{port}"), - "-e", - export.path().to_str().unwrap(), - ]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .expect("spawn diod"); - - // Wait for diod to accept connections before anyone attaches. - let deadline = Instant::now() + Duration::from_secs(10); - while Instant::now() < deadline { - if std::net::TcpStream::connect(("127.0.0.1", port)).is_ok() { + for _ in 0..20 { + let port = free_port(); + let mut child = Command::new(&diod) + .args([ + "-f", // foreground + "-n", // no auth + "-N", // bypass userdb, so a rootless diod needs no passwd/group entry for our uid + "-L", + "stderr", + "-l", + &format!("127.0.0.1:{port}"), + "-e", + export.path().to_str().unwrap(), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn diod"); + if diod_listening(&mut child) { return Diod { child, export, port, }; } - std::thread::sleep(Duration::from_millis(50)); + let _ = child.kill(); + let _ = child.wait(); } - panic!("diod did not start listening on 127.0.0.1:{port}"); + panic!("diod could not start listening on a free port"); } fn export(&self) -> &Path { @@ -150,6 +149,29 @@ impl Diod { } } +/// Read diod's log until it says it is listening (true) or exits without doing so (false). The rest +/// of the log is drained in the background so diod never blocks writing it. +fn diod_listening(child: &mut Child) -> bool { + let stderr = child.stderr.take().expect("diod stderr piped"); + let (tx, rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let mut lines = std::io::BufReader::new(stderr).lines(); + for line in lines.by_ref() { + match line { + Ok(l) if l.starts_with("Listening on ") => { + let _ = tx.send(true); + break; + } + Ok(_) => {} + Err(_) => break, + } + } + let _ = tx.send(false); + for _ in lines {} + }); + rx.recv_timeout(Duration::from_secs(10)).unwrap_or(false) +} + impl Drop for Diod { fn drop(&mut self) { let _ = self.child.kill(); @@ -157,19 +179,47 @@ impl Drop for Diod { } } +/// Unmount `mp`. Retries for a few seconds first: a desktop service probing new mounts (GNOME's +/// gvfsd-trash looks up a trash folder in each one) holds the mount busy while its lookup is in +/// flight. Then falls back to a lazy unmount, which detaches the mount at once. fn umount(mp: &Path) { - if let Some(bin) = fusermount_bin() { - let ok = Command::new(bin) - .arg("-u") + let Some(bin) = fusermount_bin() else { + let _ = Command::new("umount").arg(mp).status(); + return; + }; + let run = |args: &[&str]| { + Command::new(bin) + .args(args) .arg(mp) + .stderr(Stdio::null()) .status() - .map(|s| s.success()) - .unwrap_or(false); - if ok { + .is_ok_and(|s| s.success()) + }; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + if run(&["-u"]) { return; } + std::thread::sleep(Duration::from_millis(100)); + } + if !run(&["-u", "-z"]) { + eprintln!("couldn't unmount {}", mp.display()); + } +} + +/// Unmount `mp` and wait for its mount task to end, failing instead of hanging if it doesn't. +async fn unmount_and_join( + mp: &Path, + task: tokio::task::JoinHandle>, +) -> Result<(), String> { + umount(mp); + match tokio::time::timeout(Duration::from_secs(30), task).await { + Ok(joined) => joined.expect("mount task join"), + Err(_) => panic!( + "mount of {} still running 30s after unmounting", + mp.display() + ), } - let _ = Command::new("umount").arg(mp).status(); } /// Which transport to reach diod through. Non-TCP transports use a tiny in-test relay that bridges @@ -360,9 +410,12 @@ where if tokio::fs::try_exists(&ready_path).await.unwrap_or(false) { break; } + if task.is_finished() { + let result = task.await.expect("mount task join"); + panic!("mount ended before it became ready: {result:?}"); + } if Instant::now() >= deadline { - umount(&mp); - let _ = task.await; + let _ = unmount_and_join(&mp, task).await; panic!("mount did not become ready within 15s"); } tokio::time::sleep(Duration::from_millis(50)).await; @@ -379,8 +432,7 @@ where }) .await; - umount(&mp); - let mount_result = task.await.expect("mount task join"); + let mount_result = unmount_and_join(&mp, task).await; for h in relays { h.abort(); }