Add multi-reader FUSE support via FUSE_DEV_IOC_CLONE

Adds support for multi-threaded FUSE request processing using cloned
file descriptors. This enables parallel request handling for improved
throughput.

API additions:
- Session::clone_fd() - clones FUSE fd using FUSE_DEV_IOC_CLONE ioctl
- Session::from_fd_initialized() - creates session from cloned fd,
  skips INIT handshake since primary session already completed it
- Session::run() made public for external multi-reader use
- Conditional handshake skip when proto_version is pre-set

Usage pattern:
  let session = Session::new(fs, mountpoint, &Config::default())?;
  let cloned_fd = session.clone_fd()?;
  let reader = Session::from_fd_initialized(fs2, cloned_fd, SessionACL::All);
  thread::spawn(move || reader.run());
  session.run()?;

Platform: Linux only (#[cfg(target_os = "linux")])

Tests: Added clone_fd_multi_reader and from_fd_initialized_works
integration tests verifying multi-reader request distribution.

Author: ejc3

src/channel.rs

@@ -1,13 +1,23 @@
 use std::io;
 use std::os::fd::AsFd;
+use std::os::fd::AsRawFd;
 use std::os::fd::BorrowedFd;
+use std::os::fd::FromRawFd;
+use std::os::fd::OwnedFd;
 use std::sync::Arc;
 
 use nix::errno::Errno;
 
 use crate::dev_fuse::DevFuse;
 use crate::passthrough::BackingId;
 
+/// FUSE_DEV_IOC_CLONE ioctl number for cloning /dev/fuse file descriptors.
+/// This is _IOW('E', 0, __u32) = 0x8004e500 on Linux.
+#[cfg(all(target_os = "linux", target_env = "musl"))]
+const FUSE_DEV_IOC_CLONE: libc::c_int = 0x8004e500u32 as libc::c_int;
+#[cfg(all(target_os = "linux", not(target_env = "musl")))]
+const FUSE_DEV_IOC_CLONE: libc::c_ulong = 0x8004e500;
+
 /// A raw communication channel to the FUSE kernel driver
 #[derive(Debug)]
 pub(crate) struct Channel(Arc<DevFuse>);
@@ -55,6 +65,32 @@ impl Channel {
         // a sender by using the same file and use it in other threads.
         ChannelSender(self.0.clone())
     }
+
+    /// Clone the FUSE file descriptor for multi-threaded request processing.
+    ///
+    /// Uses the `FUSE_DEV_IOC_CLONE` ioctl to create a new fd that shares
+    /// the same FUSE connection. Each cloned fd can independently read
+    /// requests and must send responses on the same fd that read the request.
+    #[cfg(target_os = "linux")]
+    pub(crate) fn clone_fd(&self) -> io::Result<OwnedFd> {
+        // Open a new /dev/fuse fd
+        let fd = unsafe { libc::open(c"/dev/fuse".as_ptr(), libc::O_RDWR) };
+        if fd < 0 {
+            return Err(io::Error::last_os_error());
+        }
+        // SAFETY: fd is valid, we just opened it successfully
+        let new_fd = unsafe { OwnedFd::from_raw_fd(fd) };
+
+        // Clone the session onto the new fd
+        let original_fd = self.0.as_raw_fd() as u32;
+        // SAFETY: ioctl with FUSE_DEV_IOC_CLONE expects a pointer to u32 containing the source fd
+        let ret = unsafe { libc::ioctl(new_fd.as_raw_fd(), FUSE_DEV_IOC_CLONE, &original_fd) };
+        if ret < 0 {
+            return Err(io::Error::last_os_error());
+        }
+
+        Ok(new_fd)
+    }
 }
 
 #[derive(Clone, Debug)]

src/session.rs

@@ -197,6 +197,42 @@ impl<FS: Filesystem> Session<FS> {
         }
     }
 
+    /// Create a session from a cloned FUSE file descriptor for multi-reader setups.
+    ///
+    /// Use this with fds obtained from [`Session::clone_fd()`] when the primary
+    /// session has already completed the FUSE INIT handshake.
+    ///
+    /// # Arguments
+    /// * `filesystem` - The filesystem implementation to handle requests
+    /// * `fd` - A cloned fd from [`Session::clone_fd()`]
+    /// * `acl` - Access control settings for the session
+    ///
+    /// # Important
+    /// This session skips the FUSE INIT protocol. Using this with an uninitialized
+    /// fd will cause all requests to fail with EIO.
+    ///
+    /// Each cloned fd handles its own request/response pairs - the FUSE kernel
+    /// requires that the fd which reads a request is the same fd that sends the response.
+    pub fn from_fd_initialized(filesystem: FS, fd: OwnedFd, acl: SessionACL) -> Self {
+        let ch = Channel::new(Arc::new(DevFuse(File::from(fd))));
+        Session {
+            filesystem: FilesystemHolder {
+                fs: Some(filesystem),
+            },
+            ch,
+            mount: UmountOnDrop {
+                mount: Arc::new(Mutex::new(None)),
+            },
+            allowed: acl,
+            session_owner: geteuid(),
+            // Skip INIT - caller guarantees mount is initialized
+            proto_version: Some(Version(
+                abi::FUSE_KERNEL_VERSION,
+                abi::FUSE_KERNEL_MINOR_VERSION,
+            )),
+        }
+    }
+
     /// Run the session loop in a background thread. If the returned handle is dropped,
     /// the filesystem is unmounted and the given session ends.
     pub fn spawn(self) -> io::Result<BackgroundSession> {
@@ -217,8 +253,11 @@ impl<FS: Filesystem> Session<FS> {
     /// may run concurrent by spawning threads.
     /// # Errors
     /// Returns any final error when the session comes to an end.
-    pub(crate) fn run(mut self) -> io::Result<()> {
-        self.handshake()?;
+    pub fn run(mut self) -> io::Result<()> {
+        // Skip handshake if session is already initialized (e.g., from_fd_initialized)
+        if self.proto_version.is_none() {
+            self.handshake()?;
+        }
 
         let ret = self.event_loop();
 
@@ -429,6 +468,35 @@ impl<FS: Filesystem> Session<FS> {
     pub fn notifier(&self) -> Notifier {
         Notifier::new(self.ch.sender())
     }
+
+    /// Clone the FUSE file descriptor for multi-threaded request processing.
+    ///
+    /// Creates a new fd that can independently read FUSE requests, enabling
+    /// multi-threaded request handling. The cloned fd shares the same FUSE
+    /// connection but can be used by a separate thread to read requests in parallel.
+    ///
+    /// # Usage
+    /// ```ignore
+    /// // Primary session handles INIT and runs in main thread
+    /// let session = Session::new(fs, mountpoint, options)?;
+    ///
+    /// // Clone fd for additional reader threads
+    /// let cloned_fd = session.clone_fd()?;
+    /// let reader_session = Session::from_fd_initialized(fs_clone, cloned_fd, acl);
+    /// std::thread::spawn(move || reader_session.run());
+    ///
+    /// session.run()?;
+    /// ```
+    ///
+    /// # Platform Support
+    /// This is only available on Linux.
+    ///
+    /// # Errors
+    /// Returns an error if `/dev/fuse` cannot be opened or the ioctl fails.
+    #[cfg(target_os = "linux")]
+    pub fn clone_fd(&self) -> io::Result<OwnedFd> {
+        self.ch.clone_fd()
+    }
 }
 
 #[derive(Debug)]

tests/integration_tests.rs

@@ -0,0 +1,245 @@
+use std::os::unix::fs::PermissionsExt;
+use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+use std::thread;
+use std::time::Duration;
+
+use fuser::Config;
+use fuser::Errno;
+use fuser::FileHandle;
+use fuser::Filesystem;
+use fuser::INodeNo;
+use fuser::Session;
+use fuser::SessionACL;
+use tempfile::TempDir;
+
+/// Test that clone_fd creates a working file descriptor for multi-reader setups.
+#[cfg(target_os = "linux")]
+#[test]
+fn clone_fd_multi_reader() {
+    use std::os::fd::AsRawFd;
+
+    // Simple filesystem that tracks how many times getattr is called
+    struct CountingFS {
+        count: Arc<AtomicUsize>,
+    }
+
+    impl Filesystem for CountingFS {
+        fn getattr(
+            &self,
+            _req: &fuser::Request,
+            ino: INodeNo,
+            _fh: Option<FileHandle>,
+            reply: fuser::ReplyAttr,
+        ) {
+            self.count.fetch_add(1, Ordering::SeqCst);
+            if ino == INodeNo::ROOT {
+                // Root directory
+                reply.attr(
+                    &Duration::from_secs(1),
+                    &fuser::FileAttr {
+                        ino: INodeNo::ROOT,
+                        size: 0,
+                        blocks: 0,
+                        atime: std::time::UNIX_EPOCH,
+                        mtime: std::time::UNIX_EPOCH,
+                        ctime: std::time::UNIX_EPOCH,
+                        crtime: std::time::UNIX_EPOCH,
+                        kind: fuser::FileType::Directory,
+                        perm: 0o755,
+                        nlink: 2,
+                        uid: 0,
+                        gid: 0,
+                        rdev: 0,
+                        blksize: 4096,
+                        flags: 0,
+                    },
+                );
+            } else {
+                reply.error(Errno::ENOENT);
+            }
+        }
+    }
+
+    let tmpdir: TempDir = tempfile::tempdir().unwrap();
+    let count = Arc::new(AtomicUsize::new(0));
+
+    let session = Session::new(
+        CountingFS {
+            count: count.clone(),
+        },
+        tmpdir.path(),
+        &Config::default(),
+    )
+    .unwrap();
+
+    // Clone the fd - this should succeed
+    let cloned_fd = session.clone_fd().expect("clone_fd should succeed");
+
+    // Verify it's a valid fd (different from the original)
+    assert!(cloned_fd.as_raw_fd() >= 0);
+
+    // Clean up
+    drop(cloned_fd);
+    drop(session);
+}
+
+/// Test that from_fd_initialized creates a session that can process requests.
+/// Verifies both readers receive requests and metadata returns expected values.
+#[cfg(target_os = "linux")]
+#[test]
+fn from_fd_initialized_works() {
+    // Filesystem that tracks request count per instance with artificial delay
+    // to ensure kernel dispatches to both readers
+    struct SlowCountingFS {
+        count: Arc<AtomicUsize>,
+    }
+
+    impl Filesystem for SlowCountingFS {
+        fn getattr(
+            &self,
+            _req: &fuser::Request,
+            ino: INodeNo,
+            _fh: Option<FileHandle>,
+            reply: fuser::ReplyAttr,
+        ) {
+            self.count.fetch_add(1, Ordering::SeqCst);
+
+            // Add delay so while one reader is processing, the kernel
+            // will dispatch concurrent requests to the other reader
+            thread::sleep(Duration::from_millis(50));
+
+            if ino == INodeNo::ROOT {
+                reply.attr(
+                    &Duration::from_secs(0), // No caching to ensure requests reach FUSE
+                    &fuser::FileAttr {
+                        ino: INodeNo::ROOT,
+                        size: 0,
+                        blocks: 0,
+                        atime: std::time::UNIX_EPOCH,
+                        mtime: std::time::UNIX_EPOCH,
+                        ctime: std::time::UNIX_EPOCH,
+                        crtime: std::time::UNIX_EPOCH,
+                        kind: fuser::FileType::Directory,
+                        perm: 0o755,
+                        nlink: 2,
+                        uid: 0,
+                        gid: 0,
+                        rdev: 0,
+                        blksize: 4096,
+                        flags: 0,
+                    },
+                );
+            } else {
+                reply.error(Errno::ENOENT);
+            }
+        }
+    }
+
+    let tmpdir: TempDir = tempfile::tempdir().unwrap();
+
+    // Separate counters to track which reader handled requests
+    let primary_count = Arc::new(AtomicUsize::new(0));
+    let reader_count = Arc::new(AtomicUsize::new(0));
+
+    let session = Session::new(
+        SlowCountingFS {
+            count: primary_count.clone(),
+        },
+        tmpdir.path(),
+        &Config::default(),
+    )
+    .unwrap();
+
+    // Clone fd for second reader BEFORE spawning the primary (spawn takes ownership)
+    let cloned_fd = session.clone_fd().expect("clone_fd should succeed");
+
+    // Save path for concurrent access (before session is moved)
+    let path = tmpdir.path().to_path_buf();
+
+    // Spawn primary session in background
+    let primary_bg = session.spawn().unwrap();
+
+    // Start second reader in a thread
+    let reader_count_clone = reader_count.clone();
+    let reader_handle = thread::spawn(move || {
+        let reader_session = Session::from_fd_initialized(
+            SlowCountingFS {
+                count: reader_count_clone,
+            },
+            cloned_fd,
+            SessionACL::All,
+        );
+        // Spawn in background - the thread will run until ENODEV when primary unmounts
+        let bg = reader_session.spawn().unwrap();
+        // Keep BackgroundSession alive - when dropped it will wait for the thread
+        // The thread exits on ENODEV when primary unmounts
+        drop(bg);
+    });
+
+    // Give readers time to start processing
+    thread::sleep(Duration::from_millis(100));
+
+    // Generate concurrent requests from multiple threads
+    // With 50ms delay per request and concurrent threads, the kernel should
+    // dispatch to both readers
+    let request_threads: Vec<_> = (0..4)
+        .map(|_| {
+            let p = path.clone();
+            thread::spawn(move || {
+                for _ in 0..5 {
+                    let meta = std::fs::metadata(&p);
+                    // Verify metadata returns expected values
+                    if let Ok(m) = meta {
+                        assert!(m.is_dir(), "root should be a directory");
+                        assert_eq!(
+                            m.permissions().mode() & 0o777,
+                            0o755,
+                            "permissions should be 0o755"
+                        );
+                    }
+                }
+            })
+        })
+        .collect();
+
+    // Wait for all request threads
+    for t in request_threads {
+        t.join().unwrap();
+    }
+
+    // Let any in-flight requests complete
+    thread::sleep(Duration::from_millis(200));
+
+    // Unmount by dropping the primary BackgroundSession
+    // This will cause the secondary to exit with ENODEV
+    drop(primary_bg);
+
+    // Wait for reader thread to finish
+    let _ = reader_handle.join();
+
+    // Verify both readers processed requests
+    let primary = primary_count.load(Ordering::SeqCst);
+    let reader = reader_count.load(Ordering::SeqCst);
+    let total = primary + reader;
+
+    eprintln!(
+        "Request distribution: primary={}, reader={}, total={}",
+        primary, reader, total
+    );
+
+    // Total should be > 0 (requests were processed)
+    assert!(total > 0, "expected some requests to be processed");
+
+    // With 50ms delay per request and 4 concurrent threads, both readers
+    // should handle some requests. The kernel dispatches to whichever
+    // reader is blocked in read(), and with the delay, both should be available.
+    assert!(
+        primary > 0 && reader > 0,
+        "expected both readers to process requests: primary={}, reader={}. \
+         This verifies multi-threaded request handling works.",
+        primary,
+        reader
+    );
+}