Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,12 @@
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()`.
- [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`,
Expand All @@ -49,6 +55,12 @@
- 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.
- [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`,
Expand Down
4 changes: 4 additions & 0 deletions crates/fuse-backend-core/src/abi/fuse_abi_linux.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
39 changes: 36 additions & 3 deletions crates/fuse-backend-core/src/api/server/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down
80 changes: 66 additions & 14 deletions crates/fuse-backend-core/src/api/server/sync_io.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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 = {
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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>()];
Expand Down Expand Up @@ -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() {
Expand Down
95 changes: 65 additions & 30 deletions crates/fuse-backend-core/src/common/async_runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,10 @@
//! By default the runtime type is auto-detected: `tokio-uring` is used if io-uring is
//! available, otherwise the tokio current-thread runtime is used. The detection result
//! may be overridden with the `FUSE_BACKEND_RS_ASYNC_RUNTIME` environment variable,
//! which accepts `tokio` or `uring`. This is useful when asynchronous futures created
//! by this crate (e.g. `FuseDevTask::poll_handler()`) are driven by the application's
//! own tokio runtime, where `tokio-uring` objects can't be polled.
//! which accepts `tokio` or `uring`, or by calling [`set_runtime_type()`] before any
//! other function of this module or `async_file`. This is useful when the application's
//! own tokio runtime drives futures from this crate, such as `FuseDevTask::poll_handler()`,
//! because a plain tokio runtime can't poll `tokio-uring` objects.
//!
//! Note: creating io-uring rings locks kernel memory accounted against `RLIMIT_MEMLOCK`
//! (see `account_memlock()` in the kernel). If the limit is too low (e.g. the default
Expand All @@ -18,7 +19,8 @@
//! runtime with `FUSE_BACKEND_RS_ASYNC_RUNTIME=tokio` instead.

use std::future::Future;
use std::sync::LazyLock;
use std::io;
use std::sync::{LazyLock, OnceLock};

/// Environment variable to select the asynchronous runtime type, `tokio` or `uring`.
pub const RUNTIME_TYPE_ENV: &str = "FUSE_BACKEND_RS_ASYNC_RUNTIME";
Expand All @@ -33,44 +35,77 @@ const RING_ENTRIES: u32 = 256;

pub(crate) static RUNTIME_TYPE: LazyLock<RuntimeType> = LazyLock::new(RuntimeType::new);

pub(crate) enum RuntimeType {
/// The runtime type requested with `set_runtime_type()`, which takes precedence over the
/// environment variable.
static REQUESTED_RUNTIME_TYPE: OnceLock<RuntimeType> = OnceLock::new();

/// Type of the asynchronous runtime used by this crate.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RuntimeType {
/// The tokio current-thread runtime, with plain syscalls for file IO.
Tokio,
/// The tokio-uring runtime, with io-uring for file IO.
#[cfg(target_os = "linux")]
Uring,
}

/// Select the runtime type instead of detecting it.
///
/// The crate fixes the type on first use, so call this before creating a [`Runtime`], an
/// `async_file::File` or a `FuseDevTask`. It overrides `FUSE_BACKEND_RS_ASYNC_RUNTIME`. It
/// returns an error if the type in effect is a different one, for example because io-uring
/// isn't available.
pub fn set_runtime_type(runtime_type: RuntimeType) -> io::Result<()> {
let _ = REQUESTED_RUNTIME_TYPE.set(runtime_type);
let current = *RUNTIME_TYPE;
if current == runtime_type {
Ok(())
} else {
Err(io::Error::other(format!(
"async runtime type is {current:?}, not {runtime_type:?}"
)))
}
}

/// Get the runtime type used by this crate, detecting it if it isn't fixed yet.
pub fn runtime_type() -> RuntimeType {
*RUNTIME_TYPE
}

impl RuntimeType {
fn new() -> Self {
if let Ok(val) = std::env::var(RUNTIME_TYPE_ENV) {
match Self::parse_env(&val) {
Some(RuntimeType::Tokio) => return Self::Tokio,
let requested = REQUESTED_RUNTIME_TYPE
.get()
.copied()
.or_else(Self::from_env);
match requested {
Some(RuntimeType::Tokio) => Self::Tokio,
#[cfg(target_os = "linux")]
_ if Self::probe_io_uring() => Self::Uring,
#[cfg(target_os = "linux")]
Some(RuntimeType::Uring) => {
warn!("'uring' is requested but io-uring isn't available, falling back to tokio");
Self::Tokio
}
None => {
#[cfg(target_os = "linux")]
Some(RuntimeType::Uring) => {
if Self::probe_io_uring() {
return Self::Uring;
}
warn!(
"'uring' is requested via {} but io-uring isn't available, falling back to tokio",
RUNTIME_TYPE_ENV
);
}
None => {
warn!(
"unknown value '{}' for environment variable {}, ignored",
val, RUNTIME_TYPE_ENV
);
}
warn!("io-uring isn't available, falling back to the tokio runtime");
Self::Tokio
}
}
}

#[cfg(target_os = "linux")]
{
if Self::probe_io_uring() {
return Self::Uring;
}
warn!("io-uring isn't available, falling back to the tokio runtime");
/// The runtime type requested with `FUSE_BACKEND_RS_ASYNC_RUNTIME`, if any.
fn from_env() -> Option<Self> {
let val = std::env::var(RUNTIME_TYPE_ENV).ok()?;
let requested = Self::parse_env(&val);
if requested.is_none() {
warn!(
"unknown value '{}' for environment variable {}, ignored",
val, RUNTIME_TYPE_ENV
);
}
Self::Tokio
requested
}

fn parse_env(val: &str) -> Option<Self> {
Expand Down
22 changes: 22 additions & 0 deletions crates/fuse-backend-core/tests/runtime_type.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
// SPDX-License-Identifier: Apache-2.0

//! The crate fixes the runtime type once per process, so this test runs in a binary of its own.

#[cfg(feature = "async-io")]
#[test]
fn test_set_runtime_type() {
use fuse_backend_core::async_file::File;
use fuse_backend_core::async_runtime::{self, Runtime, RuntimeType};

async_runtime::set_runtime_type(RuntimeType::Tokio).unwrap();
assert_eq!(async_runtime::runtime_type(), RuntimeType::Tokio);
// Asking for the type in effect again succeeds.
async_runtime::set_runtime_type(RuntimeType::Tokio).unwrap();
// The type can't change once it is fixed.
#[cfg(target_os = "linux")]
assert!(async_runtime::set_runtime_type(RuntimeType::Uring).is_err());

assert!(matches!(Runtime::new(), Runtime::Tokio(..)));
let file = File::from_std_file(std::fs::File::open("/dev/null").unwrap());
assert!(matches!(file, File::Tokio(_)));
}
3 changes: 2 additions & 1 deletion crates/fuse-backend-fusedev/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ libc = "0.2.68"
log = "0.4.6"
mio = { version = "0.8", features = ["os-poll", "os-ext"] }
nix = "0.24"
tokio = { version = "1", optional = true }
tokio = { version = "1.53.3", optional = true }
vm-memory = { workspace = true }
vmm-sys-util = { workspace = true }

Expand All @@ -49,6 +49,7 @@ async-io = [
"futures-util",
"tokio/rt",
"tokio/macros",
"tokio/net",
"fuse-backend-core/async-io",
# The session tests mount a real `PassthroughFs`, and core's
# `BackendFileSystem` requires `AsyncFileSystem` once core's `async-io`
Expand Down
Loading
Loading