WIP: Add Multithreading Support

CHANGELOG.md

@@ -1,5 +1,11 @@
 # FUSE for Rust - Changelog
 
+## 0.15 - 2024-08-20
+* **Major** `Filesystem` is bound by `Clone` to enable multithreading
+* Add `multithreading` feature dependent on `libfuse3`
+* Add `spawn_mount2_threaded` method
+* `BackgroundSession::new` accepts a thread count. Will throw an error if the Multithreading feature is not enabled and thread count is greater than one
+
 ## 0.14.0 - 2023-11-04
 * Add support for poll
 * Add support for notifications

Cargo.toml

@@ -5,7 +5,7 @@ license = "MIT"
 repository = "https://github.com/cberner/fuser"
 documentation = "https://docs.rs/fuser"
 homepage = "https://github.com/cberner/fuser"
-version = "0.14.0"
+version = "0.15.0"
 edition = "2021"
 readme = "README.md"
 authors = ["Christopher Berner <christopherberner@gmail.com>"]
@@ -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.7", features = ["derive"] }
-nix = { version = "0.28.0", features = ["fs", "user"] }
+nix = { version = "0.28.0", features = ["fs", "user", "ioctl",] }
 
 [dev-dependencies]
 env_logger = "0.11.3"
@@ -61,6 +61,7 @@ abi-7-28 = ["abi-7-27"]
 abi-7-29 = ["abi-7-28"]
 abi-7-30 = ["abi-7-29"]
 abi-7-31 = ["abi-7-30"]
+multithreading = []
 
 [[example]]
 name = "poll"

examples/hello.rs

@@ -47,6 +47,7 @@ const HELLO_TXT_ATTR: FileAttr = FileAttr {
     blksize: 512,
 };
 
+#[derive(Clone)]
 struct HelloFS;
 
 impl Filesystem for HelloFS {
@@ -145,5 +146,6 @@ fn main() {
     if matches.get_flag("allow-root") {
         options.push(MountOption::AllowRoot);
     }
-    fuser::mount2(HelloFS, mountpoint, &options).unwrap();
+    let s = fuser::spawn_mount2_threaded(HelloFS, mountpoint, &options, 2).unwrap();
+    s.guard.join();
 }

examples/null.rs

@@ -1,6 +1,7 @@
 use fuser::{Filesystem, MountOption};
 use std::env;
 
+#[derive(Clone)]
 struct NullFS;
 
 impl Filesystem for NullFS {}

examples/simple.rs

@@ -32,6 +32,7 @@ use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use std::{env, fs, io};
+use std::sync::Arc;
 
 const BLOCK_SIZE: u64 = 512;
 const MAX_NAME_LENGTH: u32 = 255;
@@ -243,9 +244,10 @@ impl From<InodeAttributes> for fuser::FileAttr {
 
 // Stores inode metadata data in "$data_dir/inodes" and file contents in "$data_dir/contents"
 // Directory data is stored in the file's contents, as a serialized DirectoryDescriptor
+#[derive(Clone)]
 struct SimpleFS {
     data_dir: String,
-    next_file_handle: AtomicU64,
+    next_file_handle: Arc<AtomicU64>,
     direct_io: bool,
     suid_support: bool,
 }
@@ -260,7 +262,7 @@ impl SimpleFS {
         {
             SimpleFS {
                 data_dir,
-                next_file_handle: AtomicU64::new(1),
+                next_file_handle: Arc::new(AtomicU64::new(1)),
                 direct_io,
                 suid_support,
             }
@@ -269,7 +271,7 @@ impl SimpleFS {
         {
             SimpleFS {
                 data_dir,
-                next_file_handle: AtomicU64::new(1),
+                next_file_handle: Arc::new(AtomicU64::new(1)),
                 direct_io,
                 suid_support: false,
             }

src/channel.rs

@@ -1,8 +1,46 @@
+use crate::reply::ReplySender;
+use libc::{c_int, c_void, size_t};
 use std::{fs::File, io, os::unix::prelude::AsRawFd, sync::Arc};
+use log::{debug, trace};
+use std::os::fd::FromRawFd;
+use crate::mnt::Mount;
 
-use libc::{c_int, c_void, size_t};
+/// The implementation of fuse fd clone. Taken from (Datenlord)[https://github.com/datenlord/datenlord/blob/master/src/async_fuse/fuse/session.rs#L73 under the MIT License.
+/// This module is just for avoiding the `missing_docs` of `ioctl_read` macro.
+#[allow(missing_docs)] // Raised by `ioctl_read!`
+#[allow(dead_code)]
+mod _fuse_fd_clone {
+    use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, OwnedFd, RawFd};
 
-use crate::reply::ReplySender;
+    use nix::fcntl::{self, FcntlArg, FdFlag, OFlag};
+    use nix::ioctl_read;
+    use nix::sys::stat::Mode;
+    ioctl_read!(fuse_fd_clone_impl, 229, 0, u32);
+
+    /// Clones a FUSE session fd into a FUSE worker fd.
+    ///
+    /// # Safety
+    /// Behavior is undefined if any of the following conditions are violated:
+    ///
+    /// - `session_fd` must be a valid file descriptor to an open FUSE device.
+    #[allow(clippy::unnecessary_safety_comment)]
+    pub unsafe fn fuse_fd_clone(session_fd: RawFd) -> nix::Result<RawFd> {
+        let devname = "/dev/fuse";
+        let cloned_fd = fcntl::open(devname, OFlag::O_RDWR | OFlag::O_CLOEXEC, Mode::empty())?;
+        // use `OwnedFd` here is just to release the fd when error occurs
+        // SAFETY: the `cloned_fd` is just opened
+        let cloned_fd = OwnedFd::from_raw_fd(cloned_fd);
+
+        fcntl::fcntl(cloned_fd.as_raw_fd(), FcntlArg::F_SETFD(FdFlag::FD_CLOEXEC))?;
+
+        let mut result_fd: u32 = session_fd.try_into().unwrap();
+        // SAFETY: `cloned_fd` is ensured to be valid, and `&mut result_fd` is a valid
+        // pointer to a value on stack
+        fuse_fd_clone_impl(cloned_fd.as_raw_fd(), &mut result_fd)?;
+        Ok(cloned_fd.into_raw_fd()) // use `into_raw_fd` to transfer the
+                                    // ownership of the fd
+    }
+}
 
 /// A raw communication channel to the FUSE kernel driver
 #[derive(Debug)]
@@ -13,9 +51,35 @@ impl Channel {
     /// given path. The kernel driver will delegate filesystem operations of
     /// the given path to the channel.
     pub(crate) fn new(device: Arc<File>) -> Self {
+        trace!("established channel to kernel driver. device={:?}", device);
         Self(device)
     }
 
+    /// Create a new communication channel to the kernel driver by calling ['_fuse_fd_clone::fuse_fd_clone']
+    /// with the Session FD. This will create a new communication channel to
+    /// the kernel driver attached to the parent session.
+    #[cfg(all(feature = "multithreading", feature = "libfuse3"))]
+    pub(crate) fn worker(mount: &Mount) -> (Self, c_int) {
+        let session_fd = mount.session_fd();
+
+        // SAFETY: `session_fd` is ensured to be valid as it is returned from the Mount
+        let fd = unsafe { _fuse_fd_clone::fuse_fd_clone(session_fd) };
+
+        let fd = match fd {
+            Ok(fd) => fd,
+            Err(err) => {
+                panic!("fuse: failed to clone device fd: {:?}", err);
+            }
+        };
+
+        debug!("established worker fd '{}' from session fd '{}'", fd, session_fd);
+
+        // SAFETY: `fd` is created above and is validated before this point
+        let device = unsafe { File::from_raw_fd(fd) };
+
+        (Self(Arc::new(device)), fd)
+    }
+
     /// 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 {
@@ -28,6 +92,7 @@ impl Channel {
         if rc < 0 {
             Err(io::Error::last_os_error())
         } else {
+            trace!("received {} bytes", rc);
             Ok(rc as usize)
         }
     }
@@ -58,6 +123,7 @@ impl ReplySender for ChannelSender {
             Err(io::Error::last_os_error())
         } else {
             debug_assert_eq!(bufs.iter().map(|b| b.len()).sum::<usize>(), rc as usize);
+            trace!("sent {} bytes", rc);
             Ok(())
         }
     }

src/lib.rs

@@ -291,7 +291,7 @@ impl KernelConfig {
 /// implementations are provided here to get a mountable filesystem that does
 /// nothing.
 #[allow(clippy::too_many_arguments)]
-pub trait Filesystem {
+pub trait Filesystem: Clone {
     /// Initialize filesystem.
     /// Called before any other filesystem method.
     /// The kernel module connection can be configured using the KernelConfig object
@@ -1036,7 +1036,7 @@ pub fn spawn_mount<'a, FS: Filesystem + Send + 'static + 'a, P: AsRef<Path>>(
         .map(|x| Some(MountOption::from_str(x.to_str()?)))
         .collect();
     let options = options.ok_or(ErrorKind::InvalidData)?;
-    Session::new(filesystem, mountpoint.as_ref(), options.as_ref()).and_then(|se| se.spawn())
+    Session::new(filesystem, mountpoint.as_ref(), options.as_ref()).and_then(|se| se.spawn(1))
 }
 
 /// Mount the given filesystem to the given mountpoint. This function spawns
@@ -1047,10 +1047,24 @@ pub fn spawn_mount<'a, FS: Filesystem + Send + 'static + 'a, P: AsRef<Path>>(
 ///
 /// NOTE: This is the corresponding function to mount2.
 pub fn spawn_mount2<'a, FS: Filesystem + Send + 'static + 'a, P: AsRef<Path>>(
+    filesystem: FS,
+    mountpoint: P,
+    options: &[MountOption]
+) -> io::Result<BackgroundSession> {
+    check_option_conflicts(options)?;
+    Session::new(filesystem, mountpoint.as_ref(), options).and_then(|se| se.spawn(1))
+}
+
+/// Mount the given filesystem to the given mountpoint; spawning n number of worker threads. 
+/// There is an assumption that the [`Filesystem`] given is thread-safe, and has proper internal 
+/// synchronization to prevent deadlocks. 
+#[cfg(all(feature = "multithreading", feature = "libfuse3"))]
+pub fn spawn_mount2_threaded<'a, FS: Filesystem + Send + 'static + 'a, P: AsRef<Path>>(
     filesystem: FS,
     mountpoint: P,
     options: &[MountOption],
+    threads: u8
 ) -> io::Result<BackgroundSession> {
     check_option_conflicts(options)?;
-    Session::new(filesystem, mountpoint.as_ref(), options).and_then(|se| se.spawn())
+    Session::new(filesystem, mountpoint.as_ref(), options).and_then(|se| se.spawn(threads))
 }

src/ll/mod.rs

@@ -11,7 +11,7 @@ use std::{convert::TryInto, num::NonZeroI32, time::SystemTime};
 
 pub use reply::Response;
 pub use request::{
-    AnyRequest, FileHandle, INodeNo, Lock, Operation, Request, RequestError, RequestId, Version,
+    AnyRequest, FileHandle, INodeNo, Lock, Operation, Request, RequestId, Version,
 };
 
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]

src/mnt/fuse3.rs

@@ -12,6 +12,7 @@ use std::{
     ptr,
     sync::Arc,
 };
+use libc::c_int;
 
 /// Ensures that an os error is never 0/Success
 fn ensure_last_os_error() -> io::Error {
@@ -50,6 +51,11 @@ impl Mount {
             Ok((Arc::new(file), mount))
         })
     }
+
+    #[cfg(feature = "multithreading")]
+    pub fn session_fd(&self) -> c_int {
+        unsafe { fuse_session_fd(self.fuse_session) }
+    }
 }
 impl Drop for Mount {
     fn drop(&mut self) {

src/request.rs

@@ -11,7 +11,7 @@ use std::convert::TryFrom;
 #[cfg(feature = "abi-7-28")]
 use std::convert::TryInto;
 use std::path::Path;
-
+use std::sync::atomic::Ordering;
 use crate::channel::ChannelSender;
 use crate::ll::Request as _;
 #[cfg(feature = "abi-7-21")]
@@ -168,22 +168,22 @@ impl<'a> Request<'a> {
                     config.max_readahead,
                     config.max_write
                 );
-                se.initialized = true;
+                se.initialized.store(true, Ordering::Relaxed);
                 return Ok(Some(x.reply(&config)));
             }
             // Any operation is invalid before initialization
-            _ if !se.initialized => {
+            _ if !se.initialized.load(Ordering::Relaxed) => {
                 warn!("Ignoring FUSE operation before init: {}", self.request);
                 return Err(Errno::EIO);
             }
             // Filesystem destroyed
             ll::Operation::Destroy(x) => {
                 se.filesystem.destroy();
-                se.destroyed = true;
+                se.destroyed.store(true, Ordering::Relaxed);
                 return Ok(Some(x.reply()));
             }
             // Any operation is invalid after destroy
-            _ if se.destroyed => {
+            _ if se.destroyed.load(Ordering::Relaxed) => {
                 warn!("Ignoring FUSE operation after destroy: {}", self.request);
                 return Err(Errno::EIO);
             }