Use multi-threaded FuseSession in tests

Signed-off-by: Alessandro Passaro <[email protected]>

Author: passaro

mountpoint-s3-fs/src/fuse/session.rs

@@ -24,7 +24,7 @@ pub struct FuseSession {
     on_close: Vec<OnClose>,
 }
 
-type OnClose = Box<dyn FnOnce()>;
+type OnClose = Box<dyn FnOnce() + Send>;
 
 impl FuseSession {
     /// Create a new multi-threaded FUSE session.
@@ -47,7 +47,7 @@ impl FuseSession {
     }
 
     /// Create worker threads to dispatch requests for a FUSE session.
-    fn from_session<FS: Filesystem + Send + Sync + 'static>(
+    pub fn from_session<FS: Filesystem + Send + Sync + 'static>(
         mut session: Session<FS>,
         max_worker_threads: usize,
     ) -> anyhow::Result<Self> {

mountpoint-s3-fs/tests/common/fuse.rs

@@ -4,12 +4,13 @@ use std::os::fd::{AsFd, AsRawFd};
 use std::path::Path;
 use std::sync::Arc;
 
-use fuser::{BackgroundSession, Mount, MountOption, Session};
+use fuser::{Mount, MountOption, Session};
 use mountpoint_s3_client::ObjectClient;
 use mountpoint_s3_client::checksums::crc32c;
 use mountpoint_s3_client::config::S3ClientAuthConfig;
 use mountpoint_s3_client::types::{Checksum, PutObjectSingleParams, UploadChecksum};
 use mountpoint_s3_fs::data_cache::DataCache;
+use mountpoint_s3_fs::fuse::session::FuseSession;
 use mountpoint_s3_fs::fuse::{ErrorLogger, S3FuseFilesystem};
 #[cfg(feature = "manifest")]
 use mountpoint_s3_fs::manifest::{Manifest, ManifestMetablock};
@@ -68,6 +69,7 @@ pub struct TestSessionConfig {
     // If true, the test session will be created by opening and passing
     // FUSE device using `Session::from_fd`, otherwise `Session::new` will be used.
     pub pass_fuse_fd: bool,
+    pub max_worker_threads: usize,
     pub error_logger: Option<Box<dyn ErrorLogger + Send + Sync>>,
     pub cache_block_size: usize,
     #[cfg(feature = "manifest")]
@@ -84,6 +86,7 @@ impl Default for TestSessionConfig {
             filesystem_config: Default::default(),
             auth_config: Default::default(),
             pass_fuse_fd: false,
+            max_worker_threads: 16,
             error_logger: None,
             cache_block_size,
             #[cfg(feature = "manifest")]
@@ -110,15 +113,15 @@ pub struct TestSession {
     test_client: Box<dyn TestClient>,
     prefix: String,
     // Option so we can explicitly unmount
-    session: Option<BackgroundSession>,
+    session: Option<FuseSession>,
     // Only set if `pass_fuse_fd` is true, will unmount the filesystem on drop.
     mount: Option<Mount>,
 }
 
 impl TestSession {
     pub fn new(
         mount_dir: TempDir,
-        session: BackgroundSession,
+        session: FuseSession,
         test_client: impl TestClient + 'static,
         prefix: String,
         mount: Option<Mount>,
@@ -150,8 +153,13 @@ impl Drop for TestSession {
         // If the session created with a pre-existing mount (e.g., with `pass_fuse_fd`),
         // this will unmount it explicitly...
         drop(self.mount.take());
-        // ...if not, the background session will have a mount here, and dropping it will unmount it.
-        self.session.take();
+        // ...if not, the fuse session will unmount on shutdown.
+        if let Some(session) = self.session.take() {
+            session.shutdown_fn()();
+            if let Err(error) = session.join() {
+                tracing::warn!(?error, "error while unmounting");
+            }
+        }
     }
 }
 
@@ -213,7 +221,7 @@ pub fn create_fuse_session<Client>(
     s3_path: S3Path,
     mount_dir: &Path,
     test_config: TestSessionConfig,
-) -> (BackgroundSession, Option<Mount>)
+) -> (FuseSession, Option<Mount>)
 where
     Client: ObjectClient + Clone + Send + Sync + 'static,
 {
@@ -245,7 +253,10 @@ where
         (Session::new(fs, mount_dir, &options).unwrap(), None)
     };
 
-    (BackgroundSession::new(session).unwrap(), mount)
+    (
+        FuseSession::from_session(session, test_config.max_worker_threads).unwrap(),
+        mount,
+    )
 }
 
 /// Open `/dev/fuse` and call `mount` syscall with given `mount_point`.

mountpoint-s3-fs/tests/fuse_tests/cache_test.rs

@@ -6,12 +6,12 @@ use crate::common::s3::{get_test_prefix, get_test_s3_path};
 use mountpoint_s3_client::S3CrtClient;
 use mountpoint_s3_fs::Runtime;
 use mountpoint_s3_fs::data_cache::{DataCache, DiskDataCache, DiskDataCacheConfig};
+use mountpoint_s3_fs::fuse::session::FuseSession;
 use mountpoint_s3_fs::memory::PagedPool;
 use mountpoint_s3_fs::object::ObjectId;
 use mountpoint_s3_fs::prefetch::Prefetcher;
 use mountpoint_s3_fs::s3::S3Path;
 
-use fuser::BackgroundSession;
 use rand::{Rng, RngCore, SeedableRng};
 use rand_chacha::ChaChaRng;
 use std::fs;
@@ -409,12 +409,7 @@ fn get_random_key(key_prefix: &str, key_suffix: &str, min_size_in_bytes: usize)
     format!("{last_key_part}{padding}")
 }
 
-fn mount_bucket<Cache>(
-    client: S3CrtClient,
-    cache: Cache,
-    pool: PagedPool,
-    s3_path: S3Path,
-) -> (TempDir, BackgroundSession)
+fn mount_bucket<Cache>(client: S3CrtClient, cache: Cache, pool: PagedPool, s3_path: S3Path) -> (TempDir, FuseSession)
 where
     Cache: DataCache + Send + Sync + 'static,
 {

mountpoint-s3-fs/tests/fuse_tests/write_test.rs

@@ -1089,6 +1089,7 @@ fn overwrite_truncate_test(creator_fn: impl TestSessionCreator, prefix: &str, rw
     };
     let test_config = TestSessionConfig {
         filesystem_config,
+        max_worker_threads: 1, // avoid concurrency issues with read after write.
         ..Default::default()
     };
     let test_session = creator_fn(prefix, test_config);