[NEEDS #270] Make file IO on the tokio runtime block less

CHANGELOG.md

@@ -32,6 +32,14 @@
   FUSE-over-io_uring transport (kernel 6.14+). Both join the `FuseServing` interface, so a daemon
   can pick its transport at runtime behind `Box<dyn FuseServing>` and fall back to the classic
   transport on `Error::UringNotSupported`.
+- `async_runtime::set_runtime_type()` selects the asynchronous runtime type in code, because
+  setting `FUSE_BACKEND_RS_ASYNC_RUNTIME` from a running program isn't safe.
+  `async_runtime::runtime_type()` returns the type in effect, and `RuntimeType` is now public.
+- `Server::init_state()` and `Server::restore_init()` let a new server take over a connection
+  that another server initialized. `abi::fuse_abi::UNIQUE_RESEND` is the bit that the kernel
+  sets in requests it sends again after `notify_resend()`.
+- `RuntimeType::file_io_blocks()` tells whether async file IO on a runtime type blocks the
+  calling thread, as it does on tokio.
 - [269](https://github.com/cloud-hypervisor/fuse-backend-rs/pull/269): `AsyncFileSystem` methods default to calling their `FileSystem`
   counterparts, so a file system only implements the handlers it serves asynchronously.
 - [269](https://github.com/cloud-hypervisor/fuse-backend-rs/pull/269): `FuseSession::detach()` lazily unmounts a Linux session through `&self`,
@@ -49,6 +57,31 @@
 - The FUSE-over-io_uring transport answers opcodes it doesn't know with ENOSYS instead of EIO.
   Kernels that send `FUSE_STATX` only fall back to `FUSE_GETATTR` after ENOSYS, so `statx()`
   calls that ask for more than the basic fields, like the birth time `stat(1)` shows, failed.
+- With the tokio runtime, `FuseDevTask` no longer retries reads in a loop. It waits for requests
+  through the IO driver of the tokio runtime that polls it, and checks its state flag every 100ms
+  while idle. Polling it on a tokio runtime without IO enabled panics. `fuse-backend-fusedev`
+  needs tokio 1.53.3.
+- The async fusedev server retries a read that fails with `ENOENT` and stops the task on read
+  errors it doesn't know, like the synchronous server. It used to log them and read again.
+- `async-io` builds on macOS, where it runs on tokio. `fuse-backend-core` now depends on
+  `io-uring` only on Linux, like `tokio-uring`.
+- If io-uring isn't available, `set_runtime_type(RuntimeType::Uring)` returns the probe's
+  error, and the fallback warning includes it.
+- On the tokio runtime, the vectored reads and writes of `async_file::File` make one
+  `preadv()`/`pwritev()` call for all buffers instead of one call per buffer.
+  `async_file::preadv()` and `pwritev()` accept more than 1024 buffers and make one call per
+  1024.
+- `PassthroughFs::enable_async_thread_pool(true)` also moves READ and WRITE
+  requests that would block the runtime thread to the blocking thread pool.
+  These are the O_DIRECT and WRITE_KILL_PRIV requests, and on the tokio
+  runtime also READ cache misses and every WRITE. On a file system without
+  `RWF_NOWAIT`, such as tmpfs, overlayfs or FUSE, every READ counts as a cache
+  miss. Each of these requests costs one more copy and a thread handoff.
+- The umbrella crate's `async-io` no longer turns on the `fusedev` and `virtiofs`
+  transports. Code that uses the items they gate, such as `abi::virtio_fs`, has
+  to enable them. A crate that depends on a transport crate directly must enable
+  that crate's `async-io` too, or its writers fail asynchronous replies with
+  `EINVAL`.
 - [254](https://github.com/cloud-hypervisor/fuse-backend-rs/pull/254): `fuse-backend-rs` is now a thin facade re-exporting the sub-crates. Every
   historical `fuse_backend_rs::{abi, api, buffer, common, transport, passthrough,
   overlayfs}` import path and cargo feature name (`fusedev`, `virtiofs`,

Cargo.toml

@@ -69,17 +69,18 @@ default = ["fusedev"]
 # its own. The `dep:` prefix keeps the dependency crate names themselves out of
 # the public feature list.
 #
-# Forwarding `async-io` to the sub-crates with the `pkg/feature` syntax also
-# activates those optional dependencies, so an async-io-only build compiles
-# the sub-crates even when no transport feature is selected. The umbrella's
-# own re-exports stay gated on the transport features below, so its public
-# surface is unchanged; resolver-level weak dependencies would avoid the
-# extra compilation but require `resolver = "2"`.
+# The weak `pkg?/feature` syntax passes `async-io` only to the transports that
+# a transport feature enables. This keeps the Linux-only virtiofs transport out
+# of macOS builds. A crate that depends on a transport directly must enable the
+# transport's `async-io` itself, or the transport's writers fail asynchronous
+# replies with EINVAL. The drivers always get `async-io`, because core's
+# `async-io` makes `BackendFileSystem` require `AsyncFileSystem`. They are
+# Linux-only dependencies, so macOS builds don't see them.
 async-io = [
     "fuse-backend-core/async-io",
     "fuse-backend-vfs/async-io",
-    "fuse-backend-fusedev/async-io",
-    "fuse-backend-virtiofs/async-io",
+    "fuse-backend-fusedev?/async-io",
+    "fuse-backend-virtiofs?/async-io",
     "fuse-backend-passthrough/async-io",
     "fuse-backend-overlayfs/async-io",
 ]

Makefile

@@ -119,6 +119,7 @@ bench-smoke:
 build-macos:
 	${CARGO} build --features="fusedev"
 	${CARGO} build --features="fusedev,fuse-t"
+	${CARGO} build --features="fusedev,async-io"
 	${CARGO} build -p fuse-backend-core --no-default-features
 	${CARGO} build -p fuse-backend-vfs --no-default-features
 	${CARGO} build -p fuse-backend-vfs --no-default-features --features="fuse-backend-vfs/persist"
@@ -129,8 +130,12 @@ check-macos: build-macos
 	${CARGO} test --features="fusedev" -- --nocapture --skip integration
 	${CARGO} clippy --features="fusedev,fuse-t" -- -Dwarnings
 	${CARGO} test --features="fusedev,fuse-t" -- --nocapture --skip integration
+	${CARGO} clippy --features="fusedev,async-io" -- -Dwarnings
+	${CARGO} test --features="fusedev,async-io" -- --nocapture --skip integration
 	${CARGO} clippy -p fuse-backend-core --no-default-features -- -Dwarnings
 	${CARGO} test -p fuse-backend-core --no-default-features -- --nocapture --skip integration
+	${CARGO} clippy -p fuse-backend-core --no-default-features --features="async-io" -- -Dwarnings
+	${CARGO} test -p fuse-backend-core --no-default-features --features="async-io" -- --nocapture --skip integration
 	${CARGO} clippy -p fuse-backend-vfs --no-default-features -- -Dwarnings
 	${CARGO} test -p fuse-backend-vfs --no-default-features -- --nocapture --skip integration
 	${CARGO} test -p fuse-backend-vfs --no-default-features --features="fuse-backend-vfs/persist" -- --nocapture --skip integration

README.md

@@ -77,9 +77,11 @@ fuse-backend-rs = { git = "https://github.com/cloud-hypervisor/fuse-backend-rs",
 The `async-io` feature is not part of a released crate version yet, so depend
 on the git repository directly. Please note that the feature is still
 **experimental**:
-- It depends on [tokio-uring](https://github.com/tokio-rs/tokio-uring) and
-  [io_uring](https://kernel.dk/io_uring.pdf), so it's only available on Linux.
-  Builds with the feature enabled will fail on other platforms.
+- On Linux it runs on [tokio-uring](https://github.com/tokio-rs/tokio-uring)
+  when the kernel supports [io_uring](https://kernel.dk/io_uring.pdf), and on a
+  [tokio](https://tokio.rs) runtime otherwise. Other platforms always use tokio.
+  The asynchronous fusedev transport (`FuseDevTask`, `AsyncFuseServing`) is
+  Linux-only.
 - The asynchronous interfaces and behavior may change in future releases.
 - The asynchronous handlers of `PassthroughFs` currently relay requests to their
   synchronous counterparts, so the blocking syscalls run in the context of the

crates/fuse-backend-core/Cargo.toml

@@ -20,13 +20,14 @@ documentation = "https://docs.rs/fuse-backend-core"
 [dependencies]
 async-trait = { version = "0.1.42", optional = true }
 bitflags = "1.1"
-io-uring = { version = "0.5.8", optional = true }
 libc = "0.2.68"
 log = "0.4.6"
 tokio = { version = "1", optional = true }
 vm-memory = { workspace = true, features = ["backend-mmap"] }
 
+# io_uring only exists on Linux, so `async-io` uses tokio elsewhere.
 [target.'cfg(target_os = "linux")'.dependencies]
+io-uring = { version = "0.5.8", optional = true }
 tokio-uring = { version = "0.4.0", optional = true }
 
 [dev-dependencies]

crates/fuse-backend-core/README.md

@@ -16,7 +16,7 @@ without any transport or driver.
 
 | Feature | Effect |
 | --- | --- |
-| `async-io` | Experimental asynchronous IO path (tokio-uring / io_uring; Linux-only). |
+| `async-io` | Experimental asynchronous IO path. It runs on tokio-uring on Linux when io_uring is available, and on tokio otherwise. |
 | `fusedev`, `virtiofs`, `fusedev-uring`, `fuse-t` | Dependency-free cfg flags forwarded from the umbrella and transport crates so code historically gated on those names keeps compiling. Enable them through the transport crates, not directly. |
 
 ## Usage

crates/fuse-backend-core/src/abi/fuse_abi_linux.rs

@@ -821,6 +821,10 @@ impl From<u32> for Opcode {
     }
 }
 
+/// Bit that the kernel sets in the unique id of a request it sends again after a
+/// `NotifyOpcode::Resend` notification.
+pub const UNIQUE_RESEND: u64 = 1 << 63;
+
 #[repr(u32)]
 #[derive(Debug, Copy, Clone)]
 pub enum NotifyOpcode {

crates/fuse-backend-core/src/api/server/mod.rs

@@ -20,15 +20,13 @@ use std::ffi::CStr;
 use std::io::{self, Read};
 use std::marker::PhantomData;
 use std::mem::size_of;
-use std::sync::atomic::AtomicU64;
+use std::sync::atomic::{AtomicU64, Ordering};
 
 use crate::abi::fuse_abi::*;
 use crate::api::filesystem::{Context, FileSystem, ZeroCopyReader, ZeroCopyWriter};
 use crate::buffer::{Reader, Writer};
 use crate::file_traits::FileReadWriteVolatile;
 use crate::{bytes_to_cstr, BitmapSlice, Error, Result};
-#[cfg(all(target_os = "linux", feature = "fusedev-uring"))]
-use std::sync::atomic::Ordering;
 
 #[cfg(feature = "async-io")]
 mod async_io;
@@ -81,6 +79,41 @@ impl<F: FileSystem + Sync> Server<F> {
         }
     }
 
+    /// Get the protocol version and the options enabled at INIT, for `restore_init()`. Before
+    /// INIT, the options are empty.
+    pub fn init_state(&self) -> (ServerVersion, FsOptions) {
+        let version = decode_version(self.vers.load(Ordering::Acquire));
+        let options = FsOptions::from_bits_truncate(self.options.load(Ordering::Acquire));
+        (version, options)
+    }
+
+    /// Restore the protocol version and options from another server's `init_state()`.
+    ///
+    /// A server that takes over a FUSE connection never sees INIT, so call this before it
+    /// serves requests. This doesn't call `FileSystem::init()`, so the file system has to
+    /// restore its own INIT state. For a `Vfs`, `Vfs::restore_from_bytes()` restores the state
+    /// of the `Vfs` itself. Then call `init()` on each file system mounted on it, with the
+    /// restored `VfsOptions::out_opts`, before `Vfs::restore_mount()` mounts it again at its old
+    /// index. The inode numbers that the kernel holds refer to that index. For a file system
+    /// served without a `Vfs`, call `init()` with the options from `init_state()`.
+    ///
+    /// Then send `notify_resend()` to get back the requests that the old server didn't reply
+    /// to. Send it after the old server stops handling requests and before the new one reads
+    /// any. The kernel sends again every request read from the connection and not replied to
+    /// yet, so a request that a server is still handling would run twice. The kernel ends the
+    /// requests read from a FUSE device file description once that description is closed, so
+    /// the new server needs the file descriptions that the old one read from.
+    pub fn restore_init(&self, version: ServerVersion, options: FsOptions) {
+        self.vers.store(
+            encode_version(version.major, version.minor),
+            Ordering::Release,
+        );
+        self.options.store(options.bits(), Ordering::Release);
+        #[cfg(all(target_os = "linux", feature = "fusedev-uring"))]
+        self.negotiated_init_flags
+            .store(options.bits(), Ordering::Release);
+    }
+
     /// Request serving FUSE requests over io_uring (experimental).
     ///
     /// Must be called before the session is mounted: the request is carried

crates/fuse-backend-core/src/api/server/sync_io.rs

@@ -113,6 +113,9 @@ pub(super) fn parse_create_extensions(_options: &FsOptions, _tail: &[u8]) -> Res
 impl<F: FileSystem + Sync> Server<F> {
     #[cfg(feature = "fusedev")]
     /// Use to send notify msg to kernel fuse
+    ///
+    /// The write waits for the lock of `parent`, which a lookup in it holds until its reply, so
+    /// don't call this from a thread that serves requests. Any fd of the connection works.
     pub fn notify_inval_entry<W: Writer>(
         &self,
         mut w: W,
@@ -178,6 +181,8 @@ impl<F: FileSystem + Sync> Server<F> {
     /// the crash recovery routine. Given that FUSE initialization does not occur again during recovery,
     /// the capability to support resend notifications may not be automatically detected. It is the responsibility
     /// of the upper layers to verify and persist the kernel's support for this feature upon the initial FUSE setup.
+    ///
+    /// Resent requests have `UNIQUE_RESEND` set in their unique id, which the reply keeps.
     pub fn notify_resend<W: Writer>(&self, mut w: W) -> Result<()> {
         let mut buffer_writer = w.split_at(0).map_err(Error::FailedToSplitWriter)?;
         let header = {
@@ -1726,10 +1731,10 @@ mod tests {
     use super::super::test_util::{MockFS, TestWriter};
     use super::*;
 
-    fn prepare_srvcontext<'a>(
+    fn prepare_srvcontext<'a, F: FileSystem>(
         read_buf: &'a mut [u8],
         write_buf: &'a mut [u8],
-    ) -> SrvContext<'a, MockFS, (), TestWriter<'a>> {
+    ) -> SrvContext<'a, F, (), TestWriter<'a>> {
         let reader = Reader::<()>::from_slice(read_buf);
         let writer = TestWriter::new(write_buf);
         let in_header = InHeader::default();
@@ -1963,21 +1968,21 @@ mod tests {
         assert_eq!(res, 80);
     }
 
-    #[test]
-    fn test_server_init_echoes_init_ext_for_64bit_caps() {
-        // A file system negotiating FUSE_CREATE_SUPP_GROUP (bit 34) when the
-        // kernel offers it, like Vfs and PassthroughFs do.  Its other
-        // operations keep the default implementations.
-        struct SuppGroupFs;
-        impl FileSystem for SuppGroupFs {
-            type Inode = u64;
-            type Handle = u64;
+    // A file system negotiating FUSE_CREATE_SUPP_GROUP (bit 34) when the
+    // kernel offers it, like Vfs and PassthroughFs do.  Its other
+    // operations keep the default implementations.
+    struct SuppGroupFs;
+    impl FileSystem for SuppGroupFs {
+        type Inode = u64;
+        type Handle = u64;
 
-            fn init(&self, capable: FsOptions) -> io::Result<FsOptions> {
-                Ok(capable & FsOptions::CREATE_SUPP_GROUP)
-            }
+        fn init(&self, capable: FsOptions) -> io::Result<FsOptions> {
+            Ok(capable & FsOptions::CREATE_SUPP_GROUP)
         }
+    }
 
+    #[test]
+    fn test_server_init_echoes_init_ext_for_64bit_caps() {
         // A kernel >= 6.3 sends the full 64-byte fuse_init_in: INIT_EXT in
         // flags and FUSE_CREATE_SUPP_GROUP in flags2.
         let mut read_buf = [0u8; size_of::<InitIn>() + size_of::<InitIn2>()];
@@ -2019,6 +2024,53 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_server_restore_init() {
+        let init_in = InitIn {
+            major: KERNEL_VERSION,
+            minor: 40,
+            max_readahead: 0,
+            flags: FsOptions::INIT_EXT.bits() as u32,
+        };
+        let init_in2 = InitIn2 {
+            flags2: ((FsOptions::CREATE_SUPP_GROUP | FsOptions::HAS_RESEND).bits() >> 32) as u32,
+            unused: [0; 11],
+        };
+        let mut read_buf = [init_in.as_slice(), init_in2.as_slice()].concat();
+        let mut write_buf = [0u8; 4096];
+        let old = Server::new(SuppGroupFs);
+        old.init(prepare_srvcontext(&mut read_buf, &mut write_buf), |_| {})
+            .unwrap();
+        let (version, options) = old.init_state();
+        assert_eq!(version.minor, 40);
+        assert!(options.contains(FsOptions::CREATE_SUPP_GROUP | FsOptions::INIT_EXT));
+
+        let new = Server::new(SuppGroupFs);
+        assert!(new.init_state().1.is_empty());
+        new.restore_init(version.clone(), options);
+        assert_eq!(new.init_state(), (version, options));
+    }
+
+    #[test]
+    fn test_server_reply_keeps_unique_resend() {
+        let server = Server::new(MockFS);
+        let unique = UNIQUE_RESEND | 42;
+        let header = InHeader {
+            len: (size_of::<InHeader>() + size_of::<GetattrIn>()) as u32,
+            opcode: Opcode::Getattr as u32,
+            unique,
+            nodeid: ROOT_ID,
+            ..Default::default()
+        };
+        let mut read_buf = [header.as_slice(), GetattrIn::default().as_slice()].concat();
+        let mut write_buf = [0u8; 4096];
+        let reader = Reader::<()>::from_slice(&mut read_buf);
+        let writer = TestWriter::new(&mut write_buf);
+        server.handle_message(reader, writer, None, None).unwrap();
+        let out = OutHeader::from_slice(&write_buf[..size_of::<OutHeader>()]).unwrap();
+        assert_eq!(out.unique, unique);
+    }
+
     #[cfg(feature = "fusedev-uring")]
     #[test]
     fn test_server_init_uring_negotiation() {