Switch to cleaned up non-async fuser fork

Our current fuser fork has a messy commit history and does two big
changes at once: (1) concurrent filesystems and (2) async filesystems.
We don't actually need the async version right now -- it was incomplete
and we were just running it on blocking threads anyway.

This change redirects the fuser fork to our `fuser/fork` branch, which
is a cleaned up history of the fork but without async support. Instead,
we push the async support down into our implementation via `block_on`.
This performs the same, but with a smaller change to fuser.

In the future we'll still want to async-ify fuser so the entire stack
can run on top of (the CRT's?) async runtime. But there's more work
involved in correctly async-ifying fuser, like doing async IO on the
FUSE device, so we can defer that for now.

This change temporarily breaks the `--thread-count` command-line
argument and so will tank concurrent performance tests. I'll fix that in
a followup as I don't want this diff to get any bigger, and will fix
unmount support (#93) at the same time.

Signed-off-by: James Bornholt <bornholt@amazon.com>

mountpoint-s3/Cargo.toml

@@ -13,7 +13,6 @@ mountpoint-s3-crt = { path = "../mountpoint-s3-crt" }
 anyhow = { version = "1.0.64", features = ["backtrace"] }
 async-channel = "1.8.0"
 async-lock = "2.6.0"
-async-trait = "0.1.57"
 bytes = "1.2.1"
 clap = { version = "3.2.12", features = ["derive"] }
 ctrlc = "3.2.3"

mountpoint-s3/src/fuse.rs

@@ -1,8 +1,8 @@
-use async_trait::async_trait;
+use futures::executor::block_on;
 use futures::task::Spawn;
 use std::ffi::OsStr;
 use std::time::Duration;
-use tracing::instrument;
+use tracing::{instrument, Instrument};
 
 use crate::fs::{DirectoryReplier, InodeNo, ReadReplier, S3Filesystem, S3FilesystemConfig};
 use fuser::{
@@ -28,43 +28,42 @@ where
     }
 }
 
-#[async_trait]
 impl<Client, Runtime> Filesystem for S3FuseFilesystem<Client, Runtime>
 where
     Client: ObjectClient + Send + Sync + 'static,
     Runtime: Spawn + Send + Sync,
 {
     #[instrument(level = "debug", skip_all)]
-    async fn init(&self, _req: &Request<'_>, config: &mut KernelConfig) -> Result<(), libc::c_int> {
-        self.fs.init(config).await
+    fn init(&self, _req: &Request<'_>, config: &mut KernelConfig) -> Result<(), libc::c_int> {
+        block_on(self.fs.init(config).in_current_span())
     }
 
     #[instrument(level="debug", skip_all, fields(req=_req.unique(), ino=parent, name=?name))]
-    async fn lookup(&self, _req: &Request<'_>, parent: InodeNo, name: &OsStr, reply: ReplyEntry) {
-        match self.fs.lookup(parent, name).await {
+    fn lookup(&self, _req: &Request<'_>, parent: InodeNo, name: &OsStr, reply: ReplyEntry) {
+        match block_on(self.fs.lookup(parent, name).in_current_span()) {
             Ok(entry) => reply.entry(&entry.ttl, &entry.attr, entry.generation),
             Err(e) => reply.error(e),
         }
     }
 
     #[instrument(level="debug", skip_all, fields(req=_req.unique(), ino=ino))]
-    async fn getattr(&self, _req: &Request<'_>, ino: InodeNo, reply: ReplyAttr) {
-        match self.fs.getattr(ino).await {
+    fn getattr(&self, _req: &Request<'_>, ino: InodeNo, reply: ReplyAttr) {
+        match block_on(self.fs.getattr(ino).in_current_span()) {
             Ok(attr) => reply.attr(&attr.ttl, &attr.attr),
             Err(e) => reply.error(e),
         }
     }
 
     #[instrument(level="debug", skip_all, fields(req=_req.unique(), ino=ino))]
-    async fn open(&self, _req: &Request<'_>, ino: InodeNo, flags: i32, reply: ReplyOpen) {
-        match self.fs.open(ino, flags).await {
+    fn open(&self, _req: &Request<'_>, ino: InodeNo, flags: i32, reply: ReplyOpen) {
+        match block_on(self.fs.open(ino, flags).in_current_span()) {
             Ok(opened) => reply.opened(opened.fh, opened.flags),
             Err(e) => reply.error(e),
         }
     }
 
     #[instrument(level="debug", skip_all, fields(req=_req.unique(), ino=ino, fh=fh, offset=offset, size=size))]
-    async fn read(
+    fn read(
         &self,
         _req: &Request<'_>,
         ino: InodeNo,
@@ -103,29 +102,26 @@ where
             inner: reply,
             bytes_sent: &mut bytes_sent,
         };
-        self.fs.read(ino, fh, offset, size, flags, lock, replier).await;
+        block_on(
+            self.fs
+                .read(ino, fh, offset, size, flags, lock, replier)
+                .in_current_span(),
+        );
         // return value of read is proof a reply was sent
 
         metrics::counter!("fuse.bytes_read", bytes_sent as u64);
     }
 
     #[instrument(level="debug", skip_all, fields(req=_req.unique(), ino=parent))]
-    async fn opendir(&self, _req: &Request<'_>, parent: InodeNo, flags: i32, reply: ReplyOpen) {
-        match self.fs.opendir(parent, flags).await {
+    fn opendir(&self, _req: &Request<'_>, parent: InodeNo, flags: i32, reply: ReplyOpen) {
+        match block_on(self.fs.opendir(parent, flags).in_current_span()) {
             Ok(opened) => reply.opened(opened.fh, opened.flags),
             Err(e) => reply.error(e),
         }
     }
 
     #[instrument(level="debug", skip_all, fields(req=_req.unique(), ino=parent, fh=fh, offset=offset))]
-    async fn readdir(
-        &self,
-        _req: &Request<'_>,
-        parent: InodeNo,
-        fh: u64,
-        offset: i64,
-        mut reply: fuser::ReplyDirectory,
-    ) {
+    fn readdir(&self, _req: &Request<'_>, parent: InodeNo, fh: u64, offset: i64, mut reply: fuser::ReplyDirectory) {
         struct ReplyDirectory<'a> {
             inner: &'a mut fuser::ReplyDirectory,
         }
@@ -146,14 +142,14 @@ where
 
         let replier = ReplyDirectory { inner: &mut reply };
 
-        match self.fs.readdir(parent, fh, offset, replier).await {
+        match block_on(self.fs.readdir(parent, fh, offset, replier).in_current_span()) {
             Ok(_) => reply.ok(),
             Err(e) => reply.error(e),
         }
     }
 
     #[instrument(level="debug", skip_all, fields(req=_req.unique(), ino=parent, fh=fh, offset=offset))]
-    async fn readdirplus(
+    fn readdirplus(
         &self,
         _req: &Request<'_>,
         parent: InodeNo,
@@ -181,14 +177,14 @@ where
 
         let replier = ReplyDirectoryPlus { inner: &mut reply };
 
-        match self.fs.readdir(parent, fh, offset, replier).await {
+        match block_on(self.fs.readdir(parent, fh, offset, replier).in_current_span()) {
             Ok(_) => reply.ok(),
             Err(e) => reply.error(e),
         }
     }
 
     #[instrument(level="debug", skip_all, fields(req=_req.unique(), ino=ino, fh=fh))]
-    async fn release(
+    fn release(
         &self,
         _req: &Request<'_>,
         ino: InodeNo,
@@ -198,14 +194,14 @@ where
         flush: bool,
         reply: ReplyEmpty,
     ) {
-        match self.fs.release(ino, fh, flags, lock_owner, flush).await {
+        match block_on(self.fs.release(ino, fh, flags, lock_owner, flush).in_current_span()) {
             Ok(()) => reply.ok(),
             Err(e) => reply.error(e),
         }
     }
 
     #[instrument(level="debug", skip_all, fields(req=_req.unique(), parent=parent, name=?name))]
-    async fn mknod(
+    fn mknod(
         &self,
         _req: &Request<'_>,
         parent: InodeNo,
@@ -218,14 +214,14 @@ where
         // mode_t is u32 on Linux but u16 on macOS, so cast it here
         let mode = mode as libc::mode_t;
 
-        match self.fs.mknod(parent, name, mode, umask, rdev).await {
+        match block_on(self.fs.mknod(parent, name, mode, umask, rdev).in_current_span()) {
             Ok(entry) => reply.entry(&entry.ttl, &entry.attr, entry.generation),
             Err(e) => reply.error(e),
         }
     }
 
     #[instrument(level="debug", skip_all, fields(req=_req.unique(), ino=ino, fh=fh, offset=offset, length=data.len()))]
-    async fn write(
+    fn write(
         &self,
         _req: &Request<'_>,
         ino: InodeNo,
@@ -237,11 +233,11 @@ where
         lock_owner: Option<u64>,
         reply: ReplyWrite,
     ) {
-        match self
-            .fs
-            .write(ino, fh, offset, data, write_flags, flags, lock_owner)
-            .await
-        {
+        match block_on(
+            self.fs
+                .write(ino, fh, offset, data, write_flags, flags, lock_owner)
+                .in_current_span(),
+        ) {
             Ok(bytes_written) => reply.written(bytes_written),
             Err(e) => reply.error(e),
         }

mountpoint-s3/src/main.rs

@@ -355,12 +355,8 @@ fn mount(args: CliArgs) -> anyhow::Result<BackgroundSession> {
 
     let session = Session::new(fs, &args.mount_point, &options).context("Failed to create FUSE session")?;
 
-    let session = if let Some(thread_count) = args.thread_count {
-        BackgroundSession::new_multi_thread(session, thread_count as usize)
-    } else {
-        BackgroundSession::new(session)
-    };
-    let session = session.context("Failed to start FUSE session")?;
+    // TODO correctly handle multi-threading and unmounting
+    let session = BackgroundSession::new(session).context("Failed to start FUSE session")?;
 
     tracing::info!("successfully mounted {:?}", args.mount_point);
 

vendor-fuser.sh

@@ -13,11 +13,11 @@ fi
 
 rm -rf $FUSER_FULL_PATH
 
-git clone --branch fuser/async ssh://git@github.com/awslabs/mountpoint-s3.git $FUSER_FULL_PATH
+git clone --branch fuser/fork ssh://git@github.com/awslabs/mountpoint-s3.git $FUSER_FULL_PATH
 COMMIT=$(git -C $FUSER_FULL_PATH rev-parse --short HEAD)
 
 rm -rf $FUSER_FULL_PATH/.git
 
 git add $FUSER_FULL_PATH
 
-git commit -m "Update vendored fuser to $COMMIT"
+git commit -m "Update vendored fuser to $COMMIT" -s