core: let a server take over an initialized connection

A daemon can hand its FUSE device fds to a new process, which then
serves the connection without a FUSE_INIT. The new Server has no way to
learn what the old one negotiated, so it parses requests as if no
option were enabled, for example the FUSE_EXT_GROUPS extension of
create requests.

Add Server::init_state(), which returns the protocol version and the
options enabled at INIT, and Server::restore_init(), which sets them on
the new server. Add abi::fuse_abi::UNIQUE_RESEND, the bit the kernel
sets in requests that notify_resend() makes it send again. The server
already replies with the unique id unchanged, and a test covers it.

Document that notify_inval_entry() waits for the parent's lock, which
a lookup in it holds until its reply, so a thread that serves requests
must not call it.

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

Author: Kranzes

crates/fuse-backend-core/src/abi/fuse_abi_linux.rs

@@ -821,6 +821,10 @@ impl From<u32> for Opcode {
     }
 }
 
+/// Bit that the kernel sets in the unique id of a request it sends again after a
+/// `NotifyOpcode::Resend` notification.
+pub const UNIQUE_RESEND: u64 = 1 << 63;
+
 #[repr(u32)]
 #[derive(Debug, Copy, Clone)]
 pub enum NotifyOpcode {

crates/fuse-backend-core/src/api/server/mod.rs

@@ -20,15 +20,13 @@ use std::ffi::CStr;
 use std::io::{self, Read};
 use std::marker::PhantomData;
 use std::mem::size_of;
-use std::sync::atomic::AtomicU64;
+use std::sync::atomic::{AtomicU64, Ordering};
 
 use crate::abi::fuse_abi::*;
 use crate::api::filesystem::{Context, FileSystem, ZeroCopyReader, ZeroCopyWriter};
 use crate::buffer::{Reader, Writer};
 use crate::file_traits::FileReadWriteVolatile;
 use crate::{bytes_to_cstr, BitmapSlice, Error, Result};
-#[cfg(all(target_os = "linux", feature = "fusedev-uring"))]
-use std::sync::atomic::Ordering;
 
 #[cfg(feature = "async-io")]
 mod async_io;
@@ -81,6 +79,37 @@ impl<F: FileSystem + Sync> Server<F> {
         }
     }
 
+    /// Get the protocol version and the options enabled at INIT, for `restore_init()`. Before
+    /// INIT, the options are empty.
+    pub fn init_state(&self) -> (ServerVersion, FsOptions) {
+        let version = decode_version(self.vers.load(Ordering::Acquire));
+        let options = FsOptions::from_bits_truncate(self.options.load(Ordering::Acquire));
+        (version, options)
+    }
+
+    /// Restore the protocol version and options from another server's `init_state()`.
+    ///
+    /// A server that takes over a FUSE connection never sees INIT, so call this before it
+    /// serves requests. It doesn't call `FileSystem::init()`, so the file system has to get its
+    /// INIT state back on its own. `Vfs::restore_from_bytes()` does that for a `Vfs`, and
+    /// `Vfs::mount()` then initializes each file system mounted on it with the restored
+    /// options. A file system served without a `Vfs` can have `init()` called with the options
+    /// from `init_state()`.
+    ///
+    /// Then send `notify_resend()` to get the requests that the old server didn't reply to.
+    /// The kernel ends the requests read from a FUSE device file description once it is closed,
+    /// so the new server needs the file descriptions that the old one read from.
+    pub fn restore_init(&self, version: ServerVersion, options: FsOptions) {
+        self.vers.store(
+            encode_version(version.major, version.minor),
+            Ordering::Release,
+        );
+        self.options.store(options.bits(), Ordering::Release);
+        #[cfg(all(target_os = "linux", feature = "fusedev-uring"))]
+        self.negotiated_init_flags
+            .store(options.bits(), Ordering::Release);
+    }
+
     /// Request serving FUSE requests over io_uring (experimental).
     ///
     /// Must be called before the session is mounted: the request is carried

crates/fuse-backend-core/src/api/server/sync_io.rs

@@ -113,6 +113,9 @@ pub(super) fn parse_create_extensions(_options: &FsOptions, _tail: &[u8]) -> Res
 impl<F: FileSystem + Sync> Server<F> {
     #[cfg(feature = "fusedev")]
     /// Use to send notify msg to kernel fuse
+    ///
+    /// The write waits for the lock of `parent`, which a lookup in it holds until its reply, so
+    /// don't call this from a thread that serves requests. Any fd of the connection works.
     pub fn notify_inval_entry<W: Writer>(
         &self,
         mut w: W,
@@ -178,6 +181,8 @@ impl<F: FileSystem + Sync> Server<F> {
     /// the crash recovery routine. Given that FUSE initialization does not occur again during recovery,
     /// the capability to support resend notifications may not be automatically detected. It is the responsibility
     /// of the upper layers to verify and persist the kernel's support for this feature upon the initial FUSE setup.
+    ///
+    /// Resent requests have `UNIQUE_RESEND` set in their unique id, which the reply keeps.
     pub fn notify_resend<W: Writer>(&self, mut w: W) -> Result<()> {
         let mut buffer_writer = w.split_at(0).map_err(Error::FailedToSplitWriter)?;
         let header = {
@@ -2019,6 +2024,67 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_server_restore_init() {
+        struct SuppGroupFs;
+        impl FileSystem for SuppGroupFs {
+            type Inode = u64;
+            type Handle = u64;
+
+            fn init(&self, capable: FsOptions) -> io::Result<FsOptions> {
+                Ok(capable & FsOptions::CREATE_SUPP_GROUP)
+            }
+        }
+
+        let init_in = InitIn {
+            major: KERNEL_VERSION,
+            minor: 40,
+            max_readahead: 0,
+            flags: FsOptions::INIT_EXT.bits() as u32,
+        };
+        let init_in2 = InitIn2 {
+            flags2: ((FsOptions::CREATE_SUPP_GROUP | FsOptions::HAS_RESEND).bits() >> 32) as u32,
+            unused: [0; 11],
+        };
+        let mut read_buf = [init_in.as_slice(), init_in2.as_slice()].concat();
+        let mut write_buf = [0u8; 4096];
+        let ctx = {
+            let reader = Reader::<()>::from_slice(&mut read_buf);
+            let writer = TestWriter::new(&mut write_buf);
+            SrvContext::new(InHeader::default(), reader, writer)
+        };
+        let old = Server::new(SuppGroupFs);
+        old.init(ctx, |_| {}).unwrap();
+        let (version, options) = old.init_state();
+        assert_eq!(version.minor, 40);
+        assert!(options.contains(FsOptions::CREATE_SUPP_GROUP | FsOptions::INIT_EXT));
+
+        let new = Server::new(SuppGroupFs);
+        assert!(new.init_state().1.is_empty());
+        new.restore_init(version.clone(), options);
+        assert_eq!(new.init_state(), (version, options));
+    }
+
+    #[test]
+    fn test_server_reply_keeps_unique_resend() {
+        let server = Server::new(MockFS);
+        let unique = UNIQUE_RESEND | 42;
+        let header = InHeader {
+            len: (size_of::<InHeader>() + size_of::<GetattrIn>()) as u32,
+            opcode: Opcode::Getattr as u32,
+            unique,
+            nodeid: ROOT_ID,
+            ..Default::default()
+        };
+        let mut read_buf = [header.as_slice(), GetattrIn::default().as_slice()].concat();
+        let mut write_buf = [0u8; 4096];
+        let reader = Reader::<()>::from_slice(&mut read_buf);
+        let writer = TestWriter::new(&mut write_buf);
+        server.handle_message(reader, writer, None, None).unwrap();
+        let out = OutHeader::from_slice(&write_buf[..size_of::<OutHeader>()]).unwrap();
+        assert_eq!(out.unique, unique);
+    }
+
     #[cfg(feature = "fusedev-uring")]
     #[test]
     fn test_server_init_uring_negotiation() {