Add option to use FUSE_DEV_IOC_CLONE.

This mimics functionality from libfuse. When enabled, it will create a
new file descriptor for each fuse thread by opening /dev/fuse again and
using the FUSE_DEV_IOC_CLONE ioctl to associate the file handle with the
original filesystem.

Signed-off-by: Andrew Peace <adpeace@amazon.com>

Author: adpeace

Cargo.toml

@@ -3,7 +3,7 @@ name = "mountpoint-s3-fuser"
 description = "A fork of fuser - Filesystem in Userspace (FUSE) for Rust - only for use in Mountpoint for Amazon S3"
 license = "MIT"
 homepage = "https://github.com/awslabs/mountpoint-s3/tree/fuser/fork"
-version = "0.1.0"
+version = "0.1.1"
 edition = "2021"
 readme = "README.md"
 keywords = ["fuse", "filesystem", "system", "bindings"]
@@ -21,7 +21,7 @@ page_size = "0.6.0"
 serde = { version = "1.0.102", features = ["std", "derive"], optional = true }
 smallvec = "1.6.1"
 zerocopy = { version = "0.8", features = ["derive"] }
-nix = { version = "0.29.0", features = ["fs", "user"] }
+nix = { version = "0.29.0", features = ["fs", "user", "ioctl"] }
 
 [dev-dependencies]
 env_logger = "0.11.3"

src/channel.rs

@@ -1,5 +1,5 @@
 use std::{
-    fs::File,
+    fs::{File, OpenOptions},
     io,
     os::{
         fd::{AsFd, BorrowedFd},
@@ -8,9 +8,21 @@ use std::{
     sync::Arc,
 };
 
+use crate::reply::ReplySender;
 use libc::{c_int, c_void, size_t};
+use nix::ioctl_read;
+use nix::Result as NixResult;
 
-use crate::reply::ReplySender;
+/// FUSE_DEV_IOC_CLONE ioctl constant matching the constant defined by the FUSE kernel module in fuse.h
+const IOCTL_FUSE_DEV_IOC_CLONE: u8 = 229;
+
+ioctl_read!(
+    /// Ioctl to clone a fuse session onto a new file handle
+    fuse_dev_ioc_clone,
+    IOCTL_FUSE_DEV_IOC_CLONE,
+    0,
+    libc::c_int
+);
 
 /// A raw communication channel to the FUSE kernel driver
 #[derive(Debug)]
@@ -30,6 +42,31 @@ impl Channel {
         Self(device)
     }
 
+    /// Create a worker channel by opening a new /dev/fuse file descriptor and
+    /// associating it with the session using FUSE_DEV_IOC_CLONE ioctl.
+    pub fn clone_channel(&self) -> io::Result<Self> {
+        let worker_file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open("/dev/fuse")?;
+        let worker_fd = worker_file.as_raw_fd();
+        let mut session_fd = self.0.as_raw_fd();
+
+        // Associate the worker fd with the session fd using FUSE_DEV_IOC_CLONE
+        // SAFETY: `session_fd` is a valid open file descriptor. The ioctl
+        // FUSE_DEV_IOC_CLONE expects a pointer to an int containing the fd
+        // to clone from. The pointer is valid for the duration of the call.
+        let result: NixResult<libc::c_int> = unsafe {
+            let val = &mut session_fd as *mut libc::c_int;
+            fuse_dev_ioc_clone(worker_fd, val)
+        };
+
+        if let Err(err) = result {
+            return Err(io::Error::new(io::ErrorKind::Other, err));
+        }
+        Ok(Self(Arc::new(worker_file)))
+    }
+
     /// Receives data up to the capacity of the given buffer (can block).
     pub fn receive(&self, buffer: &mut [u8]) -> io::Result<usize> {
         let rc = unsafe {

src/session.rs

@@ -141,15 +141,47 @@ impl<FS: Filesystem> Session<FS> {
     /// Run the session loop that receives kernel requests and dispatches them to method
     /// calls into the filesystem.
     pub fn run(&self) -> io::Result<()> {
-        self.run_with_callbacks(|_| {}, |_| {})
+        self.run_with_callbacks(|_| {}, |_| {}, false)
     }
 
     /// Run the session loop that receives kernel requests and dispatches them to method
     /// calls into the filesystem.
     /// This version also notifies callers of kernel requests before and after they
     /// are dispatched to the filesystem.
-    pub fn run_with_callbacks<FA, FB>(&self, mut before_dispatch: FB, mut after_dispatch: FA) -> io::Result<()> 
-    where 
+    pub fn run_with_callbacks<FA, FB>(
+        &self,
+        before_dispatch: FB,
+        after_dispatch: FA,
+        clone_fuse_fd: bool,
+    ) -> io::Result<()>
+    where
+        FB: FnMut(&Request<'_>),
+        FA: FnMut(&Request<'_>),
+    {
+        info!(
+            "FUSE channel cloning: {}",
+            if clone_fuse_fd { "enabled" } else { "disabled" }
+        );
+
+        if clone_fuse_fd {
+            // Create a worker channel for this thread using FUSE_DEV_IOC_CLONE
+            // This allows multiple threads to read from the FUSE device without contention
+            let worker_channel = self.ch.clone_channel()?;
+            self.process_requests(&worker_channel, before_dispatch, after_dispatch)
+        } else {
+            // Use the original channel without cloning
+            self.process_requests(&self.ch, before_dispatch, after_dispatch)
+        }
+    }
+
+    /// Process requests using the given channel
+    fn process_requests<FA, FB>(
+        &self,
+        channel: &Channel,
+        mut before_dispatch: FB,
+        mut after_dispatch: FA,
+    ) -> io::Result<()>
+    where
         FB: FnMut(&Request<'_>),
         FA: FnMut(&Request<'_>),
     {
@@ -163,8 +195,8 @@ impl<FS: Filesystem> Session<FS> {
         loop {
             // Read the next request from the given channel to kernel driver
             // The kernel driver makes sure that we get exactly one request per read
-            match self.ch.receive(buf) {
-                Ok(size) => match Request::new(self.ch.sender(), &buf[..size]) {
+            match channel.receive(buf) {
+                Ok(size) => match Request::new(channel.sender(), &buf[..size]) {
                     // Dispatch request
                     Some(req) => {
                         before_dispatch(&req);