fusedev: end the connection when a uring request handler panics

A panic in a request handler unwound the uring worker thread, which
dropped its ring entries while the kernel still held them and could
copy the next request into the freed memory. The worker's queues were
left without entries, so requests from those CPUs waited forever, and so
did the request that panicked.

The worker now catches the panic and stops as it does on an error: it
ends the connection through the fusectl abort file, waits for the kernel
to give the entries back, and then lets the panic go on. It logs the
panic first, because nothing joins the worker thread's panic.

A new test serves a file system whose LOOKUP panics, and checks that
the lookup fails and FuseServing::wait() returns.

Signed-off-by: Ilan Joselevich <personal@ilanjoselevich.com>

Author: Kranzes

CHANGELOG.md

@@ -92,6 +92,9 @@
   its fusectl `abort` file. Its queues had no entries left, so requests from those CPUs waited
   forever, and so did requests it had taken but not answered. Now they fail, the other workers
   stop, and `FuseServing::wait()` returns.
+- A `UringFuseServing` worker whose request handler panics ends the connection the same way and
+  waits for the kernel to give its ring entries back before the panic goes on. Before, unwinding
+  freed entries the kernel could still copy a request into.
 - 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

crates/fuse-backend-fusedev/src/uring_session.rs

@@ -707,6 +707,9 @@ impl<F: FileSystem + Send + Sync + 'static> UringWorker<F> {
         registered?;
         // The number of entries whose command the kernel hasn't completed.
         let mut armed = entries.len();
+        // A handler that panics stops the worker like an error, and the
+        // panic goes on once the kernel has given the entries back.
+        let mut panic = None;
 
         let result = 'serve: loop {
             if let Err(e) = submit_and_wait(&mut ring, 1) {
@@ -743,9 +746,17 @@ impl<F: FileSystem + Send + Sync + 'static> UringWorker<F> {
                     break 'serve Err(io::Error::from_raw_os_error(-res));
                 }
 
-                if let Err(e) = self.process(entry) {
-                    warn!("uring worker: failed to handle request: {e}");
-                    entry.stage_error_reply(libc::EIO);
+                match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| self.process(entry)))
+                {
+                    Ok(Ok(())) => {}
+                    Ok(Err(e)) => {
+                        warn!("uring worker: failed to handle request: {e}");
+                        entry.stage_error_reply(libc::EIO);
+                    }
+                    Err(payload) => {
+                        panic = Some(payload);
+                        break 'serve Err(io::Error::other("a request handler panicked"));
+                    }
                 }
 
                 let ent: FuseUringEntInOut = entry.header().ring_ent_in_out;
@@ -783,6 +794,10 @@ impl<F: FileSystem + Send + Sync + 'static> UringWorker<F> {
             self.abort.abort();
         }
         release_entries(ring, entries, armed);
+        if let Some(payload) = panic {
+            error!("uring worker: a request handler panicked");
+            std::panic::resume_unwind(payload);
+        }
         result
     }
 
@@ -1777,6 +1792,79 @@ mod tests {
         assert_eq!(mount_minor(info, "/tmp/a b".as_ref()), Some(88));
     }
 
+    /// A file system whose LOOKUP panics.
+    struct PanicFs;
+
+    impl FileSystem for PanicFs {
+        type Inode = u64;
+        type Handle = u64;
+
+        fn lookup(
+            &self,
+            _ctx: &fuse_backend_core::api::filesystem::Context,
+            _parent: u64,
+            _name: &std::ffi::CStr,
+        ) -> io::Result<fuse_backend_core::api::filesystem::Entry> {
+            panic!("LOOKUP panics");
+        }
+
+        fn getattr(
+            &self,
+            _ctx: &fuse_backend_core::api::filesystem::Context,
+            inode: u64,
+            _handle: Option<u64>,
+        ) -> io::Result<(fuse_backend_core::abi::fuse_abi::stat64, Duration)> {
+            // Safe because stat64 is plain data.
+            let mut st: fuse_backend_core::abi::fuse_abi::stat64 = unsafe { std::mem::zeroed() };
+            st.st_ino = inode;
+            st.st_nlink = 1;
+            st.st_mode = libc::S_IFDIR | 0o755;
+            Ok((st, Duration::ZERO))
+        }
+    }
+
+    /// A worker whose handler panics ends the connection, so the request
+    /// fails instead of waiting forever, and `wait()` returns.
+    #[test]
+    fn test_uring_serving_handler_panic() {
+        let mnt = TempDir::new().unwrap();
+        let server = Arc::new(Server::new(PanicFs));
+        server.set_uring(true);
+        let mut se = FuseSession::new(mnt.as_path(), "uring_panic_test", "", false).unwrap();
+        se.mount().unwrap();
+        if ConnectionAbort::new(se.mountpoint()).path.is_none() {
+            eprintln!("fusectl isn't mounted, skipping");
+            return;
+        }
+        let cfg = UringConfig {
+            workers: 1,
+            entries_per_queue: 1,
+        };
+        let serving = match UringFuseServing::new(se, server, cfg) {
+            Ok(serving) => Arc::new(serving),
+            Err(UringNotSupported) => {
+                eprintln!("FUSE-over-io_uring is not available, skipping");
+                return;
+            }
+            Err(e) => panic!("uring serving failed: {}", e),
+        };
+
+        let (tx, rx) = channel();
+        let path = mnt.as_path().join("panics");
+        std::thread::spawn(move || {
+            let _ = tx.send(std::fs::metadata(path).is_err());
+        });
+        assert_eq!(rx.recv_timeout(Duration::from_secs(5)), Ok(true));
+        let (tx, rx) = channel();
+        let waiting = serving.clone();
+        std::thread::spawn(move || {
+            waiting.wait();
+            let _ = tx.send(());
+        });
+        rx.recv_timeout(Duration::from_secs(5))
+            .expect("the worker didn't end the connection after the panic");
+    }
+
     /// Ending the connection, as a failed worker does, stops every worker,
     /// so `wait()` returns instead of serving on with a hanging queue.
     #[test]