From c4efc19f9b2110c104d47af681aa9b8fa986fbd5 Mon Sep 17 00:00:00 2001 From: Ilan Joselevich Date: Mon, 19 Oct 2026 21:09:56 +0200 Subject: [PATCH] fusedev: free uring entries only after the kernel gives them back A uring worker freed its ring entries as soon as it stopped. When it stopped on an error while the connection was still up, such as a failed command or an interrupted io_uring_enter(), its other entries were still registered, and the kernel could copy the next request into freed memory. Closing the ring doesn't help in time, because the kernel cancels the commands later, from a workqueue. The worker now counts the entries whose command the kernel hasn't completed, and waits for those completions before it frees them. Once the connection is torn down they arrive at once. If they don't arrive within a second, the worker leaks the entries instead of freeing memory the kernel may still write to. A signal that interrupts io_uring_enter() no longer stops the worker, whether it was waiting or only submitting. If the wait in the drain fails with another error, the worker logs that error. The new mount test serves a PassthroughFs over io_uring and checks that dropping the serving layer still returns well within that second. It can't reproduce the race itself. Signed-off-by: Ilan Joselevich --- CHANGELOG.md | 4 + .../fuse-backend-fusedev/src/uring_session.rs | 171 ++++++++++++++++-- 2 files changed, 163 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f076eeb..e0adaa8a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -88,6 +88,10 @@ - READDIR and READDIRPLUS answer ENOMEM when the reply can't hold the reply header plus the requested size. The check left out the header, so a writer short by up to 16 bytes started the listing and failed partway. +- A `UringFuseServing` worker that stops frees its ring entries only after the kernel has given + them all back, and leaks them if that takes more than a second. A worker that stopped on an + error while the connection was up freed entries the kernel could still copy requests into. A + signal that interrupts the worker's `io_uring_enter()` calls no longer stops it. - [278](https://github.com/cloud-hypervisor/fuse-backend-rs/pull/278): Treat `EOPNOTSUPP`/`ENOTSUP` and `ENOSYS` from `getxattr` as "not opaque" in `Layer::is_opaque()`, so `OverlayFs::import()` no longer aborts when a layer's backing filesystem has no xattr support at all (tmpfs answers `EOPNOTSUPP` to diff --git a/crates/fuse-backend-fusedev/src/uring_session.rs b/crates/fuse-backend-fusedev/src/uring_session.rs index d0171a0f..4afb69a7 100644 --- a/crates/fuse-backend-fusedev/src/uring_session.rs +++ b/crates/fuse-backend-fusedev/src/uring_session.rs @@ -24,6 +24,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::{channel, Sender}; use std::sync::Arc; use std::thread; +use std::time::{Duration, Instant}; use io_uring::{cqueue, opcode, squeue, types, IoUring}; use vm_memory::ByteValued; @@ -599,22 +600,26 @@ impl UringWorker { let mut sq = ring.submission(); for (idx, entry) in entries.iter().enumerate() { let sqe = entry.cmd_sqe(fd, FUSE_IO_URING_CMD_REGISTER, 0, idx as u64); - // Safe because entries live for the whole lifetime of the loop. + // Safe because release_entries() keeps the entries alive + // while the kernel holds them. unsafe { sq.push(&sqe) } .map_err(|e| io::Error::other(format!("submission queue full: {e}")))?; } drop(sq); - ring.submit() - })() - .map(|_| ()); + submit(&mut ring) + })(); let _ = ready_tx.send(match ®istered { Ok(()) => Ok(()), Err(e) => Err(io::Error::other(e.to_string())), }); registered?; + // The number of entries whose command the kernel hasn't completed. + let mut armed = entries.len(); - loop { - ring.submit_and_wait(1)?; + let result = 'serve: loop { + if let Err(e) = submit_and_wait(&mut ring, 1) { + break 'serve Err(e); + } // Drain the completion queue first so that its borrow is // released before entries are processed and commits submitted. @@ -625,6 +630,11 @@ impl UringWorker { if completed.is_empty() { continue; } + let returned = completed + .iter() + .filter(|(idx, _)| *idx < entries.len()) + .count(); + armed = armed.saturating_sub(returned); for (idx, res) in completed { if idx >= entries.len() { @@ -634,11 +644,11 @@ impl UringWorker { let entry = &mut entries[idx]; if res < 0 { if exit.load(Ordering::Relaxed) { - return Ok(()); + break 'serve Ok(()); } // Registration/commit failures leave the entry in an // undefined state; stop serving rather than guessing. - return Err(io::Error::from_raw_os_error(-res)); + break 'serve Err(io::Error::from_raw_os_error(-res)); } if let Err(e) = self.process(entry) { @@ -656,7 +666,8 @@ impl UringWorker { loop { let pushed = { let mut sq = ring.submission(); - // Safe because the entry outlives the ring usage. + // Safe because release_entries() keeps the entries + // alive while the kernel holds them. unsafe { sq.push(&sqe) }.is_ok() }; if pushed { @@ -665,11 +676,18 @@ impl UringWorker { // The submission queue can never hold more outstanding // commits than completed entries, so submitting must // eventually make room. - ring.submit()?; + if let Err(e) = submit(&mut ring) { + break 'serve Err(e); + } } + // Count the commit once it is queued, because + // release_entries() submits anything still queued. + armed += 1; } - ring.submit()?; - } + // The wait at the top of the loop submits the queued commits. + }; + release_entries(ring, entries, armed); + result } /// Handle one delivered request: reassemble the header/args in the @@ -798,6 +816,86 @@ impl UringWorker { } } +/// Submit what is queued on `ring`. A signal that interrupts the submit is +/// no error, because whatever it left queued goes with the next submit. +fn submit(ring: &mut IoUring) -> io::Result<()> { + match ring.submit() { + Ok(_) => Ok(()), + Err(e) if e.raw_os_error() == Some(libc::EINTR) => Ok(()), + Err(e) => Err(e), + } +} + +/// Submit what is queued on `ring` and wait for `want` completions, retrying +/// when a signal interrupts the wait. +fn submit_and_wait( + ring: &mut IoUring, + want: usize, +) -> io::Result<()> { + loop { + match ring.submit_and_wait(want) { + Ok(_) => return Ok(()), + Err(e) if e.raw_os_error() == Some(libc::EINTR) => {} + Err(e) => return Err(e), + } + } +} + +/// How long a stopping worker waits for the kernel to give back the ring +/// entries it still holds. +const RELEASE_TIMEOUT: Duration = Duration::from_secs(1); + +/// Drop `ring` and `entries` once the kernel holds none of the entries. +/// +/// The kernel may copy a request into an entry until it completes the +/// entry's command, and closing the ring only cancels the commands later, +/// from a workqueue. So wait for the `armed` outstanding completions first, +/// which arrive quickly once the connection is torn down. If they don't +/// arrive within `RELEASE_TIMEOUT`, leak the entries instead of freeing +/// memory the kernel may still write to. +fn release_entries( + mut ring: IoUring, + entries: Vec, + mut armed: usize, +) { + let deadline = Instant::now() + RELEASE_TIMEOUT; + while armed > 0 { + let left = deadline.saturating_duration_since(Instant::now()); + if left.is_zero() { + break; + } + let ts = types::Timespec::new() + .sec(left.as_secs()) + .nsec(left.subsec_nanos()); + let args = types::SubmitArgs::new().timespec(&ts); + // Not submit_and_wait(): a timeout or a full completion queue is no + // reason to stop waiting either. + match ring.submitter().submit_with_args(1, &args) { + Ok(_) => {} + Err(e) + if matches!( + e.raw_os_error(), + Some(libc::ETIME | libc::EINTR | libc::EBUSY | libc::EAGAIN) + ) => {} + Err(e) => { + warn!("uring worker: waiting for entries failed: {e}"); + break; + } + } + let len = entries.len(); + armed = armed.saturating_sub( + ring.completion() + .filter(|cqe| (cqe.user_data() as usize) < len) + .count(), + ); + } + if armed > 0 { + warn!("uring worker: leaking {armed} ring entries that the kernel still holds"); + std::mem::forget(entries); + } + drop(ring); +} + /// Serve FUSE requests over io_uring (experimental). /// /// The serving layer takes ownership of the mounted session: dropping it @@ -1514,4 +1612,53 @@ mod tests { assert_eq!(w.available_bytes(), offset); assert_eq!(other.available_bytes(), 256 - 100); } + + /// Serve a PassthroughFs over io_uring, read and write through the + /// mount, then check that dropping the serving layer gets the ring + /// entries back from the kernel and unmounts. + #[test] + fn test_uring_serving_mount() { + let src = TempDir::new().unwrap(); + std::fs::write(src.as_path().join("hello"), b"over io_uring").unwrap(); + let mnt = TempDir::new().unwrap(); + let fs = PassthroughFs::<()>::new(Config { + root_dir: src.as_path().to_string_lossy().to_string(), + do_import: false, + ..Default::default() + }) + .unwrap(); + fs.import().unwrap(); + let server = Arc::new(Server::new(fs)); + server.set_uring(true); + let mut se = FuseSession::new(mnt.as_path(), "uring_serving_test", "", false).unwrap(); + se.mount().unwrap(); + let cfg = UringConfig { + workers: 2, + entries_per_queue: 2, + }; + let serving = match UringFuseServing::new(se, server, cfg) { + Ok(serving) => serving, + Err(UringNotSupported) => { + eprintln!("FUSE-over-io_uring is not available, skipping"); + return; + } + Err(e) => panic!("uring serving failed: {}", e), + }; + + assert_eq!( + std::fs::read(mnt.as_path().join("hello")).unwrap(), + b"over io_uring" + ); + let created = mnt.as_path().join("created"); + std::fs::write(&created, b"written over io_uring").unwrap(); + assert_eq!(std::fs::read(&created).unwrap(), b"written over io_uring"); + + // Tearing the connection down completes the commands of all armed + // entries, so the workers get their entries back well within + // RELEASE_TIMEOUT instead of waiting it out and leaking them. + let start = Instant::now(); + drop(serving); + assert!(start.elapsed() < RELEASE_TIMEOUT); + assert!(std::fs::read(mnt.as_path().join("hello")).is_err()); + } }