From 3b829ef6ff68f821861ba8da16e2fe9e30d1117b Mon Sep 17 00:00:00 2001 From: thomasjm Date: Fri, 4 Sep 2026 15:10:01 -0700 Subject: [PATCH 1/3] mount9p: give up on a stalled 9p request so the mount detaches instead of hanging forever --- src/main.rs | 16 ++++- src/mount9p.rs | 186 ++++++++++++++++++++++++++++++++++++++++++++++++- src/ninep.rs | 28 ++++++++ 3 files changed, 226 insertions(+), 4 deletions(-) diff --git a/src/main.rs b/src/main.rs index e09e522..83eb025 100644 --- a/src/main.rs +++ b/src/main.rs @@ -45,6 +45,12 @@ enum Cmd { /// revalidates, so out-of-band writes show up. #[arg(long, default_value = "mmap")] cache: String, + /// Seconds an outstanding 9p request may go unanswered before the daemon gives up, detaches + /// the mount, and exits (so a supervisor can remount). The kernel v9fs client itself waits + /// forever, so without this a wedged server or silently dead tunnel hangs every process + /// touching the mount. 0 disables the watchdog. + #[arg(long, default_value_t = 30)] + stall_timeout: u64, mountpoint: PathBuf, }, @@ -131,10 +137,18 @@ async fn main() -> Result<(), Box> { headers, msize, cache, + stall_timeout, mountpoint, } => { let transport = build_transport(&connect, &parse_headers(&headers)?).await?; - mount9p::mount9p(transport, &mountpoint, msize, &cache).await + mount9p::mount9p( + transport, + &mountpoint, + msize, + &cache, + std::time::Duration::from_secs(stall_timeout), + ) + .await } Cmd::Mount9pFuse { connect, diff --git a/src/mount9p.rs b/src/mount9p.rs index bcbe061..4e0245e 100644 --- a/src/mount9p.rs +++ b/src/mount9p.rs @@ -12,19 +12,118 @@ //! helper and is not user-namespace mountable) and the `9p`/`9pnet`/`9pnet_fd` kernel modules loaded //! on the node. When those aren't available, use the FUSE bridge (`mount9p-fuse`) instead. +use crate::ninep::{tmsg_name, TFLUSH}; use crate::transport::NineTransport; use futures_util::{SinkExt, StreamExt}; use nix::mount::{mount, umount2, MntFlags, MsFlags}; use nix::sys::socket::{socketpair, AddressFamily, SockFlag, SockType}; +use std::collections::HashMap; use std::io; use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd}; use std::path::Path; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; fn to_io(e: E) -> io::Error { io::Error::other(e.to_string()) } +/// Requests we've forwarded to the server and not yet seen a response for: tag -> (T-type, sent). +type Outstanding = Arc>>; + +/// Incremental scan of a 9p byte stream, reporting each frame's header without buffering payloads. +/// A frame is `size[4] type[1] tag[2] body...`; for `Tflush` the first body field is `oldtag[2]`, +/// so we capture up to 9 bytes per frame and skip the rest. +struct FrameScanner { + header: [u8; 9], + have: usize, + skip: usize, +} + +impl FrameScanner { + fn new() -> Self { + FrameScanner { + header: [0; 9], + have: 0, + skip: 0, + } + } + + /// Feed a chunk, invoking `on_frame(type, tag, oldtag)` once per frame. `oldtag` is only + /// meaningful for `Tflush`; for frames shorter than 9 bytes it is 0. + fn feed(&mut self, mut b: &[u8], mut on_frame: impl FnMut(u8, u16, u16)) { + while !b.is_empty() { + if self.skip > 0 { + let n = self.skip.min(b.len()); + self.skip -= n; + b = &b[n..]; + continue; + } + let target = if self.have < 4 { + 4 + } else { + let size = u32::from_le_bytes(self.header[0..4].try_into().unwrap()) as usize; + if size < 7 { + // Not legal 9p; resync as best we can by skipping the remainder. + tracing::warn!(size, "9p frame scanner: undersized frame"); + self.skip = size.saturating_sub(self.have); + self.have = 0; + continue; + } + size.min(9) + }; + let n = (target - self.have).min(b.len()); + self.header[self.have..self.have + n].copy_from_slice(&b[..n]); + self.have += n; + b = &b[n..]; + if self.have < target || self.have < 7 { + continue; + } + let size = u32::from_le_bytes(self.header[0..4].try_into().unwrap()) as usize; + let typ = self.header[4]; + let tag = u16::from_le_bytes(self.header[5..7].try_into().unwrap()); + let oldtag = match self.have { + 9 => u16::from_le_bytes(self.header[7..9].try_into().unwrap()), + _ => 0, + }; + on_frame(typ, tag, oldtag); + self.skip = size - self.have; + self.have = 0; + } + } +} + +/// Resolves with an error once the oldest outstanding request has gone unanswered for +/// `timeout` (never, when `timeout` is zero). The kernel v9fs client waits forever for a reply, +/// so a server that stops answering -- a silently dead tunnel, or a server wedged on its backing +/// store -- hangs every process touching the mount until we tear it down. +async fn stall_watchdog(outstanding: Outstanding, timeout: Duration) -> io::Error { + if timeout.is_zero() { + return std::future::pending().await; + } + let mut ticker = tokio::time::interval(Duration::from_secs(5)); + loop { + ticker.tick().await; + let oldest = { + let map = outstanding.lock().unwrap(); + map.iter() + .map(|(tag, (typ, at))| (*tag, *typ, at.elapsed())) + .max_by_key(|(_, _, age)| *age) + }; + if let Some((tag, typ, age)) = oldest { + if age > timeout { + let n = outstanding.lock().unwrap().len(); + return io::Error::other(format!( + "9p request stalled: {} (tag {tag}) unanswered for {}s, {n} outstanding", + tmsg_name(typ), + age.as_secs() + )); + } + } + } +} + /// Split `transport`, mount the v9fs client at `mountpoint`, and bridge bytes until either side /// closes. Best-effort unmount on exit. `aname` is fixed to `/export` in the mount options below. /// @@ -35,6 +134,7 @@ pub async fn mount9p( mountpoint: &Path, msize: usize, cache: &str, + stall_timeout: Duration, ) -> Result<(), Box> { // 1. Split the transport into its 9p byte sink/stream. let (mut sink, mut stream) = transport.split(); @@ -56,30 +156,51 @@ pub async fn mount9p( let bridge = tokio::net::UnixStream::from_std(bridge_std)?; let (mut bridge_rd, mut bridge_wr) = tokio::io::split(bridge); - // sock -> transport: forward the kernel's 9p requests. + let outstanding: Outstanding = Arc::new(Mutex::new(HashMap::new())); + + // sock -> transport: forward the kernel's 9p requests, recording each one's tag. + let req_tags = outstanding.clone(); let sock_to_transport = async move { let mut buf = vec![0u8; 1 << 17]; + let mut scan = FrameScanner::new(); loop { let n = bridge_rd.read(&mut buf).await?; if n == 0 { break; } + scan.feed(&buf[..n], |typ, tag, oldtag| { + let mut map = req_tags.lock().unwrap(); + // A Tflush abandons oldtag: the kernel no longer expects its answer. + if typ == TFLUSH { + map.remove(&oldtag); + } + map.insert(tag, (typ, Instant::now())); + }); sink.send(buf[..n].to_vec()).await?; } Ok::<(), io::Error>(()) }; - // transport -> sock: forward the server's 9p responses. + // transport -> sock: forward the server's 9p responses, resolving their tags. + let resp_tags = outstanding.clone(); let transport_to_sock = async move { + let mut scan = FrameScanner::new(); while let Some(item) = stream.next().await { match item { - Ok(chunk) => bridge_wr.write_all(&chunk).await?, + Ok(chunk) => { + scan.feed(&chunk, |_typ, tag, _oldtag| { + resp_tags.lock().unwrap().remove(&tag); + }); + bridge_wr.write_all(&chunk).await? + } Err(e) => return Err(e), } } Ok::<(), io::Error>(()) }; + let watchdog = stall_watchdog(outstanding, stall_timeout); + // 4. Perform mount(2) on a blocking thread while the pumps run on the async runtime. The // v9fs fd transport fget()s the fd and keeps its own reference, so once mount() returns we can // drop our copy of sock_kernel. @@ -115,6 +236,7 @@ pub async fn mount9p( tokio::pin!(sock_to_transport); tokio::pin!(transport_to_sock); + tokio::pin!(watchdog); let result: Result<(), Box> = async { // mount() completes only after the handshake, which needs the pumps running concurrently. @@ -142,12 +264,14 @@ pub async fn mount9p( } r = &mut sock_to_transport => { r.map_err(to_io)?; return Err("9p transport closed before mount completed".into()); } r = &mut transport_to_sock => { r.map_err(to_io)?; return Err("9p transport closed before mount completed".into()); } + e = &mut watchdog => { tracing::error!(%e, "mount9p: exiting so the mount detaches and the supervisor remounts"); return Err(e.into()); } } tracing::info!(?mountpoint, "mount9p: mounted; bridging"); // Keep pumping for the life of the mount. tokio::select! { r = &mut sock_to_transport => r.map_err(to_io)?, r = &mut transport_to_sock => r.map_err(to_io)?, + e = &mut watchdog => { tracing::error!(%e, "mount9p: exiting so the mount detaches and the supervisor remounts"); return Err(e.into()); } } Ok(()) } @@ -157,3 +281,59 @@ pub async fn mount9p( let _ = umount2(mountpoint, MntFlags::MNT_DETACH); result } + +#[cfg(test)] +mod tests { + use super::*; + + /// Build a frame: size[4] type[1] tag[2] + body. + fn frame(typ: u8, tag: u16, body: &[u8]) -> Vec { + let size = (7 + body.len()) as u32; + let mut f = size.to_le_bytes().to_vec(); + f.push(typ); + f.extend_from_slice(&tag.to_le_bytes()); + f.extend_from_slice(body); + f + } + + fn scan_all(scan: &mut FrameScanner, bytes: &[u8], chunk: usize) -> Vec<(u8, u16, u16)> { + let mut seen = Vec::new(); + for c in bytes.chunks(chunk.max(1)) { + scan.feed(c, |typ, tag, oldtag| seen.push((typ, tag, oldtag))); + } + seen + } + + #[test] + fn parses_frames_at_every_chunking() { + let mut bytes = frame(crate::ninep::TGETATTR, 1, &[0u8; 20]); + bytes.extend(frame(TFLUSH, 2, &1u16.to_le_bytes())); // Tflush oldtag=1 + bytes.extend(frame(crate::ninep::RCLUNK, 3, &[])); // 7-byte frame, no body + bytes.extend(frame(crate::ninep::TWRITE, 4, &[0u8; 300])); + for chunk in 1..=bytes.len() { + let mut scan = FrameScanner::new(); + let seen = scan_all(&mut scan, &bytes, chunk); + assert_eq!( + seen, + vec![ + (crate::ninep::TGETATTR, 1, 0), + (TFLUSH, 2, 1), + (crate::ninep::RCLUNK, 3, 0), + (crate::ninep::TWRITE, 4, 0), + ], + "chunk size {chunk}" + ); + } + } + + #[test] + fn oldtag_not_leaked_across_frames() { + // A 9-byte frame leaves bytes in header[7..9]; the following short frame must not + // report them as its oldtag. + let mut bytes = frame(TFLUSH, 1, &7u16.to_le_bytes()); + bytes.extend(frame(crate::ninep::RCLUNK, 2, &[])); + let mut scan = FrameScanner::new(); + let seen = scan_all(&mut scan, &bytes, bytes.len()); + assert_eq!(seen, vec![(TFLUSH, 1, 7), (crate::ninep::RCLUNK, 2, 0)]); + } +} diff --git a/src/ninep.rs b/src/ninep.rs index 06ddebc..79e65ec 100644 --- a/src/ninep.rs +++ b/src/ninep.rs @@ -37,6 +37,8 @@ pub const TVERSION: u8 = 100; pub const RVERSION: u8 = 101; pub const TATTACH: u8 = 104; pub const RATTACH: u8 = 105; +pub const TFLUSH: u8 = 108; +pub const RFLUSH: u8 = 109; pub const TWALK: u8 = 110; pub const RWALK: u8 = 111; pub const TREAD: u8 = 116; @@ -46,6 +48,32 @@ pub const RWRITE: u8 = 119; pub const TCLUNK: u8 = 120; pub const RCLUNK: u8 = 121; +/// Human-readable name for a T-message type, for diagnostics. +pub fn tmsg_name(t: u8) -> &'static str { + match t { + TSTATFS => "Tstatfs", + TLOPEN => "Tlopen", + TLCREATE => "Tlcreate", + TSYMLINK => "Tsymlink", + TREADLINK => "Treadlink", + TGETATTR => "Tgetattr", + TSETATTR => "Tsetattr", + TREADDIR => "Treaddir", + TFSYNC => "Tfsync", + TMKDIR => "Tmkdir", + TRENAMEAT => "Trenameat", + TUNLINKAT => "Tunlinkat", + TVERSION => "Tversion", + TATTACH => "Tattach", + TFLUSH => "Tflush", + TWALK => "Twalk", + TREAD => "Tread", + TWRITE => "Twrite", + TCLUNK => "Tclunk", + _ => "T?", + } +} + // Tsetattr `valid` bitmask. pub const SETATTR_MODE: u32 = 0x0001; pub const SETATTR_UID: u32 = 0x0002; From 4a0ad6908a8c7652a74928ca99a95860d1dc9e3c Mon Sep 17 00:00:00 2001 From: thomasjm Date: Fri, 4 Sep 2026 15:12:35 -0700 Subject: [PATCH 2/3] mount9p-fuse: stall watchdog on the 9p client so a wedged server fails the mount instead of hanging it --- src/client.rs | 61 +++++++++++++++++++++++++++++++++++++++++++--- src/fuse9p.rs | 6 ++++- src/lib.rs | 13 +++++++++- src/main.rs | 7 ++++++ tests/diod_fuse.rs | 8 +++++- 5 files changed, 88 insertions(+), 7 deletions(-) diff --git a/src/client.rs b/src/client.rs index 79a8e8e..b22956a 100644 --- a/src/client.rs +++ b/src/client.rs @@ -14,6 +14,7 @@ use std::collections::HashMap; use std::io; use std::sync::atomic::{AtomicU16, AtomicU32, Ordering}; use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; use tokio::sync::oneshot; const NOTAG: u16 = 0xffff; @@ -25,9 +26,17 @@ struct Frame { body: Vec, } +/// A request that has been written to the transport and is waiting for its reply. +struct PendingReq { + tx: oneshot::Sender, + /// T-message type, so a stuck request can be named when the stall watchdog fires. + mtype: u8, + sent_at: Instant, +} + pub struct NineClient { sink: tokio::sync::Mutex, - pending: Mutex>>, + pending: Mutex>, next_tag: AtomicU16, next_fid: AtomicU32, pub msize: u32, @@ -50,6 +59,7 @@ impl NineClient { msize: u32, n_uname: u32, aname: &str, + stall_timeout: Duration, ) -> Result<(Arc, Qid), Box> { let (sink, mut stream) = transport.split(); @@ -86,6 +96,42 @@ impl NineClient { let _ = pump.transport_gone.send(true); }); + // Stall watchdog: a reply that never arrives -- a silently dead tunnel, or a server wedged + // on its backing store -- would otherwise hang its waiter (and with it the serial FUSE + // session loop) forever with no other symptom. Declare the transport dead instead, which + // fails all waiters and makes the mount exit so its supervisor can remount. + if !stall_timeout.is_zero() { + let watch = client.clone(); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(Duration::from_secs(5)); + loop { + ticker.tick().await; + if *watch.transport_gone.subscribe().borrow() { + return; + } + let oldest = { + let map = watch.pending.lock().unwrap(); + map.iter() + .map(|(tag, p)| (*tag, p.mtype, p.sent_at.elapsed())) + .max_by_key(|(_, _, age)| *age) + }; + if let Some((tag, mtype, age)) = oldest { + if age > stall_timeout { + tracing::error!( + tag, + mtype = tmsg_name(mtype), + age_secs = age.as_secs(), + "9p request stalled; declaring the transport dead" + ); + watch.pending.lock().unwrap().clear(); + let _ = watch.transport_gone.send(true); + return; + } + } + } + }); + } + // Version handshake (tag NOTAG), then attach. let negotiated = client.version(msize).await?; // Re-stamp msize is immutable on the struct; we only ever send <= negotiated, and our reads @@ -111,8 +157,8 @@ impl NineClient { typ: frame.typ, body: frame.body[2..].to_vec(), }; - if let Some(tx) = self.pending.lock().unwrap().remove(&tag) { - let _ = tx.send(payload); + if let Some(p) = self.pending.lock().unwrap().remove(&tag) { + let _ = p.tx.send(payload); } } @@ -133,7 +179,14 @@ impl NineClient { /// transport/protocol failure (mapped to EIO). async fn transact(&self, mtype: u8, tag: u16, body: &[u8]) -> Result { let (tx, rx) = oneshot::channel(); - self.pending.lock().unwrap().insert(tag, tx); + self.pending.lock().unwrap().insert( + tag, + PendingReq { + tx, + mtype, + sent_at: Instant::now(), + }, + ); let size = (4 + 1 + 2 + body.len()) as u32; let mut frame = Vec::with_capacity(size as usize); diff --git a/src/fuse9p.rs b/src/fuse9p.rs index 84b102f..c2160a2 100644 --- a/src/fuse9p.rs +++ b/src/fuse9p.rs @@ -289,11 +289,15 @@ impl Fuse9p { detach_on_transport_loss: bool, // Whether to mount with `default_permissions`. default_permissions: bool, + // 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, ) -> 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 // setfsuids to it per attach), so files are owned by `uid` and chmod works. - let (client, root_qid) = NineClient::connect(transport, msize, uid, aname).await?; + let (client, root_qid) = + NineClient::connect(transport, msize, uid, aname, stall_timeout).await?; // Watch for the 9p transport closing: if it does, the mount is dead (every op would just // return EIO), so we exit below and let whatever supervises this process remount. let mut transport_gone = client.transport_gone(); diff --git a/src/lib.rs b/src/lib.rs index aaf8de2..f7d9462 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -50,5 +50,16 @@ pub async fn mount( aname: &str, tuning: Tuning, ) -> Result<(), Box> { - Fuse9p::run(transport, mountpoint, msize, uid, aname, tuning, true, true).await + Fuse9p::run( + transport, + mountpoint, + msize, + uid, + aname, + tuning, + true, + true, + std::time::Duration::from_secs(30), + ) + .await } diff --git a/src/main.rs b/src/main.rs index 83eb025..637a106 100644 --- a/src/main.rs +++ b/src/main.rs @@ -111,6 +111,11 @@ enum Cmd { /// Let the kernel enforce permissions against the owner/mode the 9p server reports. #[arg(long, default_value_t = true, action = clap::ArgAction::Set)] default_permissions: bool, + /// Seconds an outstanding 9p request may go unanswered before the daemon gives up and exits + /// (so a supervisor can remount). Without this a wedged server or silently dead tunnel hangs + /// every process touching the mount. 0 disables the watchdog. + #[arg(long, default_value_t = 30)] + stall_timeout: u64, mountpoint: PathBuf, }, @@ -164,6 +169,7 @@ async fn main() -> Result<(), Box> { wb_depth, detach_on_transport_loss, default_permissions, + stall_timeout, mountpoint, } => { let transport = build_transport(&connect, &parse_headers(&headers)?).await?; @@ -185,6 +191,7 @@ async fn main() -> Result<(), Box> { tuning, detach_on_transport_loss, default_permissions, + std::time::Duration::from_secs(stall_timeout), ) .await } diff --git a/tests/diod_fuse.rs b/tests/diod_fuse.rs index 699f70d..fa681a7 100644 --- a/tests/diod_fuse.rs +++ b/tests/diod_fuse.rs @@ -517,7 +517,13 @@ async fn client_rename_then_walk() { let gid = unsafe { libc::getegid() }; let transport = Box::new(TcpTransport::connect(&addr).await.unwrap()); - let (client, _root) = NineClient::connect(transport, 512_000, uid, &aname) + let (client, _root) = NineClient::connect( + transport, + 512_000, + uid, + &aname, + std::time::Duration::from_secs(0), + ) .await .unwrap(); let root = client.root_fid; From a728d4bf6394cbd79c7d4696e9d9496a406ab65f Mon Sep 17 00:00:00 2001 From: thomasjm Date: Sun, 6 Sep 2026 18:00:08 -0700 Subject: [PATCH 3/3] Fix rustfmt indentation in the diod test's client connect call --- tests/diod_fuse.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/diod_fuse.rs b/tests/diod_fuse.rs index fa681a7..5fb0bbf 100644 --- a/tests/diod_fuse.rs +++ b/tests/diod_fuse.rs @@ -524,8 +524,8 @@ async fn client_rename_then_walk() { &aname, std::time::Duration::from_secs(0), ) - .await - .unwrap(); + .await + .unwrap(); let root = client.root_fid; // Create "a.txt" (lcreate opens `dfid` on the new file, so clone root into a scratch fid first).