diff --git a/cli/Cargo.lock b/cli/Cargo.lock index c1b7bab3..df8ccc0d 100644 --- a/cli/Cargo.lock +++ b/cli/Cargo.lock @@ -3163,9 +3163,9 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "turso" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1918eba268f456241f21ba50d73e2ddfef0fb34027bde12f3e77d9df051f4ea2" +checksum = "6f5bb728195957fa231f1713a51d20eb6032d0d50ff9f886e220e36a1ac64c28" dependencies = [ "bytes", "http-body-util", @@ -3183,9 +3183,9 @@ dependencies = [ [[package]] name = "turso_core" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cb93aace5ab3ddfc45375ded38107e33b326beb957161456961ac82982c47e18" +checksum = "4b7772de274fbdf70da2c256e7c1bc26029512bdf4e5d6001dd0e5939973e44d" dependencies = [ "aegis", "aes", @@ -3238,9 +3238,9 @@ dependencies = [ [[package]] name = "turso_ext" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "01e1b93164b6438208d7506b12466c44bd8c2de1a584f66c20dfc68286ad9869" +checksum = "b407a4f1369005b0dd9aff499c57eb15bb32391ccd9478633d50513a0c1e8c76" dependencies = [ "chrono", "getrandom 0.3.4", @@ -3249,9 +3249,9 @@ dependencies = [ [[package]] name = "turso_macros" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "302307f3722f23be569036ff53bd0019160aab0c9e1f0ffe45fb3c345d30ee67" +checksum = "bdf248386af09f478d5973011b3853dfc965872332134c8a9f89c8ea46f6304e" dependencies = [ "proc-macro2", "quote", @@ -3260,9 +3260,9 @@ dependencies = [ [[package]] name = "turso_parser" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c70ef6c11754e676549790dc2938221e55696737be09972c5f3b4d6cf2566a1" +checksum = "72611dab70787d2b4d953b94d5a54ab4a57823b75ea2549f9e8abeef066b9fad" dependencies = [ "bitflags 2.10.0", "memchr", @@ -3275,9 +3275,9 @@ dependencies = [ [[package]] name = "turso_sdk_kit" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "28bc2e2689a68dce996f33dd59438475a099b742627452c9c4a337f7a281b5b5" +checksum = "738831ed986f66dbb14dac188b01496f1d683d8943616174df6cd6ac7df1a77c" dependencies = [ "bindgen", "env_logger", @@ -3290,9 +3290,9 @@ dependencies = [ [[package]] name = "turso_sdk_kit_macros" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "663005789da0020a17af745e041ea3009dcd1ce8baa5e3ab1611ca45335a5a0f" +checksum = "681908196c8bb86bec77416e90c1a5a3eb959795d815039a3aa606ef152e89e1" dependencies = [ "proc-macro2", "quote", @@ -3301,9 +3301,9 @@ dependencies = [ [[package]] name = "turso_sync_engine" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "56e73c29342e418fb10faa3e4ebec73fcdf76dfb652be4fe5a795627204c73a3" +checksum = "f1d7412557ec1c87e80305dd2300510a3bc7b298fa2e6848bd33f4779a3fee93" dependencies = [ "base64", "bytes", @@ -3323,9 +3323,9 @@ dependencies = [ [[package]] name = "turso_sync_sdk_kit" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49d93006dae55a8c177b1b832305afcb2432b650372b04f6e2e0e6082b07d883" +checksum = "1c429193511cefff2255739c88c9ab8243a51dad97aca14a8d5a31200e9be3bd" dependencies = [ "bindgen", "env_logger", diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 937d77c9..3c4edfb9 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -26,7 +26,7 @@ agentfs-sdk = { path = "../sdk/rust" } tokio = { version = "1", features = ["full"] } clap = { version = "4", features = ["derive"] } anyhow = "1.0" -turso = { version = "0.4.3-pre.2", features = ["sync"] } +turso = { version = "0.4.4-pre.1", features = ["sync"] } serde = { version = "1.0", features = ["derive"] } parking_lot = "0.12.5" clap_complete = { version = "=4.5.61", features = ["unstable-dynamic"] } diff --git a/cli/src/cmd/mount.rs b/cli/src/cmd/mount.rs index 16fdaa57..469de085 100644 --- a/cli/src/cmd/mount.rs +++ b/cli/src/cmd/mount.rs @@ -75,46 +75,48 @@ pub fn mount(args: MountArgs) -> Result<()> { let mount = move || { let rt = crate::get_runtime(); - let agentfs = rt.block_on(open_agentfs(opts))?; - - // Check for overlay configuration - let fs: Arc = rt.block_on(async { - let conn = agentfs.get_connection().await?; - - // Check if fs_overlay_config table exists and has base_path - let query = "SELECT value FROM fs_overlay_config WHERE key = 'base_path'"; - let base_path: Option = match conn.query(query, ()).await { - Ok(mut rows) => { - if let Ok(Some(row)) = rows.next().await { - row.get_value(0).ok().and_then(|v| { - if let Value::Text(s) = v { - Some(s.clone()) - } else { - None - } - }) - } else { - None + rt.block_on(async { + let agentfs = open_agentfs(opts).await?; + + // Check for overlay configuration + let fs: Arc = { + let conn = agentfs.get_connection().await?; + + // Check if fs_overlay_config table exists and has base_path + let query = "SELECT value FROM fs_overlay_config WHERE key = 'base_path'"; + let base_path: Option = match conn.query(query, ()).await { + Ok(mut rows) => { + if let Ok(Some(row)) = rows.next().await { + row.get_value(0).ok().and_then(|v| { + if let Value::Text(s) = v { + Some(s.clone()) + } else { + None + } + }) + } else { + None + } } + Err(_) => None, // Table doesn't exist or query failed + }; + + if let Some(base_path) = base_path { + // Create OverlayFS with HostFS base + eprintln!("Using overlay filesystem with base: {}", base_path); + let hostfs = HostFS::new(&base_path)?; + #[cfg(target_family = "unix")] + let hostfs = { hostfs.with_fuse_mountpoint(mountpoint_ino) }; + let overlay = OverlayFS::new(Arc::new(hostfs), agentfs.fs); + Arc::new(overlay) as Arc + } else { + // Plain AgentFS + Arc::new(agentfs.fs) as Arc } - Err(_) => None, // Table doesn't exist or query failed }; - if let Some(base_path) = base_path { - // Create OverlayFS with HostFS base - eprintln!("Using overlay filesystem with base: {}", base_path); - let hostfs = HostFS::new(&base_path)?; - #[cfg(target_family = "unix")] - let hostfs = { hostfs.with_fuse_mountpoint(mountpoint_ino) }; - let overlay = OverlayFS::new(Arc::new(hostfs), agentfs.fs); - Ok::, anyhow::Error>(Arc::new(overlay)) - } else { - // Plain AgentFS - Ok(Arc::new(agentfs.fs) as Arc) - } - })?; - - crate::fuse::mount(fs, fuse_opts, rt) + crate::fuse::mount(fs, fuse_opts).await + }) }; if args.foreground { diff --git a/cli/src/fuse.rs b/cli/src/fuse.rs index 5af125bc..b099b03f 100644 --- a/cli/src/fuse.rs +++ b/cli/src/fuse.rs @@ -9,6 +9,7 @@ use crate::fuser::{ }; use agentfs_sdk::error::Error as SdkError; use agentfs_sdk::{BoxedFile, FileSystem, Stats}; +use async_trait::async_trait; use parking_lot::Mutex; use std::{ collections::HashMap, @@ -20,7 +21,6 @@ use std::{ }, time::{Duration, SystemTime, UNIX_EPOCH}, }; -use tokio::runtime::Runtime; /// Convert an SDK error to an errno code for FUSE replies. /// @@ -33,6 +33,7 @@ fn error_to_errno(e: &SdkError) -> i32 { SdkError::Fs(fs_err) => fs_err.to_errno(), SdkError::Io(io_err) => io_err.raw_os_error().unwrap_or(libc::EIO), SdkError::Database(turso::Error::Busy(_)) => libc::EAGAIN, + SdkError::Database(turso::Error::BusySnapshot(_)) => libc::EAGAIN, SdkError::ConnectionPoolTimeout => libc::EAGAIN, _ => libc::EIO, } @@ -67,10 +68,9 @@ struct OpenFile { struct AgentFSFuse { fs: Arc, - runtime: Runtime, - path_cache: Arc>>, + path_cache: Mutex>, /// Maps file handle -> open file state - open_files: Arc>>, + open_files: Mutex>, /// Next file handle to allocate next_fh: AtomicU64, /// User ID to report for all files (set at mount time) @@ -86,6 +86,7 @@ struct AgentFSFuse { mountpoint_path: String, } +#[async_trait] impl Filesystem for AgentFSFuse { /// Initialize the filesystem and enable performance optimizations. /// @@ -99,7 +100,7 @@ impl Filesystem for AgentFSFuse { /// for symlink resolution. /// - No opendir support: skips opendir/releasedir calls since we don't track /// directory handles, reducing round-trips for directory operations. - fn init(&mut self, _req: &Request, config: &mut KernelConfig) -> Result<(), libc::c_int> { + async fn init(&self, _req: &Request<'_>, config: &mut KernelConfig) -> Result<(), libc::c_int> { let _ = config.add_capabilities( FUSE_ASYNC_READ | FUSE_WRITEBACK_CACHE @@ -118,20 +119,15 @@ impl Filesystem for AgentFSFuse { /// /// Resolves `name` under the directory identified by `parent` inode, stats the /// resulting path, and caches the inode-to-path mapping on success. - fn lookup(&mut self, _req: &Request, parent: u64, name: &OsStr, reply: ReplyEntry) { + async fn lookup(&self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEntry) { let Some(path) = self.lookup_path(parent, name) else { reply.error(libc::ENOENT); return; }; - let fs = self.fs.clone(); - let (result, path) = self.runtime.block_on(async move { - let result = fs.lstat(&path).await; - (result, path) - }); - match result { + match self.fs.lstat(&path).await { Ok(Some(stats)) => { let attr = fillattr(&stats, self.uid, self.gid); - self.add_path(attr.ino, path); + self.path_cache.lock().insert(attr.ino, path); reply.entry(&TTL, &attr, 0); } Ok(None) => reply.error(libc::ENOENT), @@ -143,16 +139,13 @@ impl Filesystem for AgentFSFuse { /// /// Returns metadata (size, permissions, timestamps, etc.) for the file or /// directory identified by `ino`. Root inode (1) is handled specially. - fn getattr(&mut self, _req: &Request, ino: u64, _fh: Option, reply: ReplyAttr) { + async fn getattr(&self, _req: &Request<'_>, ino: u64, _fh: Option, reply: ReplyAttr) { let Some(path) = self.get_path(ino) else { reply.error(libc::ENOENT); return; }; - let fs = self.fs.clone(); - let result = self.runtime.block_on(async move { fs.lstat(&path).await }); - - match result { + match self.fs.lstat(&path).await { Ok(Some(stats)) => reply.attr(&TTL, &fillattr(&stats, self.uid, self.gid)), Ok(None) => reply.error(libc::ENOENT), Err(e) => reply.error(error_to_errno(&e)), @@ -163,18 +156,13 @@ impl Filesystem for AgentFSFuse { /// /// Returns the path that the symlink points to. This is called by operations /// like `ls -l` to display symlink targets. - fn readlink(&mut self, _req: &Request, ino: u64, reply: ReplyData) { + async fn readlink(&self, _req: &Request<'_>, ino: u64, reply: ReplyData) { let Some(path) = self.get_path(ino) else { reply.error(libc::ENOENT); return; }; - let fs = self.fs.clone(); - let result = self - .runtime - .block_on(async move { fs.readlink(&path).await }); - - match result { + match self.fs.readlink(&path).await { Ok(Some(target)) => reply.data(target.as_bytes()), Ok(None) => reply.error(libc::ENOENT), Err(e) => reply.error(error_to_errno(&e)), @@ -185,9 +173,9 @@ impl Filesystem for AgentFSFuse { /// /// Currently `size` changes (truncate) and `mode` changes (chmod) are supported. /// Other attribute changes (uid, gid, timestamps) are accepted but ignored. - fn setattr( - &mut self, - _req: &Request, + async fn setattr( + &self, + _req: &Request<'_>, ino: u64, mode: Option, _uid: Option, @@ -210,12 +198,7 @@ impl Filesystem for AgentFSFuse { return; }; - let fs = self.fs.clone(); - let result = self - .runtime - .block_on(async move { fs.chmod(&path, new_mode).await }); - - if let Err(e) = result { + if let Err(e) = self.fs.chmod(&path, new_mode).await { reply.error(error_to_errno(&e)); return; } @@ -231,8 +214,7 @@ impl Filesystem for AgentFSFuse { }; if let Some(file) = file { - self.runtime - .block_on(async move { file.truncate(new_size).await }) + file.truncate(new_size).await } else { reply.error(libc::EBADF); return; @@ -244,11 +226,14 @@ impl Filesystem for AgentFSFuse { return; }; - let fs = self.fs.clone(); - self.runtime.block_on(async move { - let file = fs.open(&path).await?; - file.truncate(new_size).await - }) + let file = match self.fs.open(&path).await { + Ok(file) => file, + Err(e) => { + reply.error(error_to_errno(&e)); + return; + } + }; + file.truncate(new_size).await }; if let Err(e) = result { @@ -263,10 +248,7 @@ impl Filesystem for AgentFSFuse { return; }; - let fs = self.fs.clone(); - let result = self.runtime.block_on(async move { fs.stat(&path).await }); - - match result { + match self.fs.stat(&path).await { Ok(Some(stats)) => reply.attr(&TTL, &fillattr(&stats, self.uid, self.gid)), Ok(None) => reply.error(libc::ENOENT), Err(e) => reply.error(error_to_errno(&e)), @@ -284,9 +266,9 @@ impl Filesystem for AgentFSFuse { /// /// Uses readdir_plus to fetch entries with stats in a single query, /// avoiding N+1 database queries. - fn readdir( - &mut self, - _req: &Request, + async fn readdir( + &self, + _req: &Request<'_>, ino: u64, _fh: u64, offset: i64, @@ -297,13 +279,7 @@ impl Filesystem for AgentFSFuse { return; }; - let fs = self.fs.clone(); - let (entries_result, path) = self.runtime.block_on(async move { - let result = fs.readdir_plus(&path).await; - (result, path) - }); - - let entries = match entries_result { + let entries = match self.fs.readdir_plus(&path).await { Ok(Some(entries)) => entries, Ok(None) => { reply.error(libc::ENOENT); @@ -334,11 +310,7 @@ impl Filesystem for AgentFSFuse { if parent_path == "/" { 1 } else { - let fs = self.fs.clone(); - match self - .runtime - .block_on(async move { fs.stat(&parent_path).await }) - { + match self.fs.stat(&parent_path).await { Ok(Some(stats)) => stats.ino as u64, _ => 1, // Fallback to root if parent lookup fails } @@ -383,9 +355,9 @@ impl Filesystem for AgentFSFuse { /// This is an optimized version that returns both directory entries and /// their attributes in a single call, reducing kernel/userspace round trips. /// Uses readdir_plus to fetch entries with stats in a single database query. - fn readdirplus( - &mut self, - _req: &Request, + async fn readdirplus( + &self, + _req: &Request<'_>, ino: u64, _fh: u64, offset: i64, @@ -396,13 +368,7 @@ impl Filesystem for AgentFSFuse { return; }; - let fs = self.fs.clone(); - let (entries_result, path) = self.runtime.block_on(async move { - let result = fs.readdir_plus(&path).await; - (result, path) - }); - - let entries = match entries_result { + let entries = match self.fs.readdir_plus(&path).await { Ok(Some(entries)) => entries, Ok(None) => { reply.error(libc::ENOENT); @@ -415,13 +381,7 @@ impl Filesystem for AgentFSFuse { }; // Get current directory stats for "." - let fs = self.fs.clone(); - let path_for_stat = path.clone(); - let dir_stats = self - .runtime - .block_on(async move { fs.stat(&path_for_stat).await }) - .ok() - .flatten(); + let dir_stats = self.fs.stat(&path).await.ok().flatten(); // Determine parent inode and stats for ".." entry let (parent_ino, parent_stats) = if ino == 1 { @@ -440,20 +400,10 @@ impl Filesystem for AgentFSFuse { .unwrap_or_else(|| "/".to_string()); if parent_path == "/" { - let fs = self.fs.clone(); - let parent_stats = self - .runtime - .block_on(async move { fs.stat(&parent_path).await }) - .ok() - .flatten(); + let parent_stats = self.fs.stat(&parent_path).await.ok().flatten(); (1u64, parent_stats) } else { - let fs = self.fs.clone(); - let parent_stats = self - .runtime - .block_on(async move { fs.stat(&parent_path).await }) - .ok() - .flatten(); + let parent_stats = self.fs.stat(&parent_path).await.ok().flatten(); let parent_ino = parent_stats.as_ref().map(|s| s.ino as u64).unwrap_or(1); (parent_ino, parent_stats) } @@ -523,9 +473,9 @@ impl Filesystem for AgentFSFuse { /// /// Creates a directory at `name` under `parent`, then stats it to return /// proper attributes and cache the inode mapping. - fn mkdir( - &mut self, - _req: &Request, + async fn mkdir( + &self, + _req: &Request<'_>, parent: u64, name: &OsStr, _mode: u32, @@ -537,25 +487,13 @@ impl Filesystem for AgentFSFuse { return; }; - let fs = self.fs.clone(); - let (result, path) = self.runtime.block_on(async move { - let result = fs.mkdir(&path).await; - (result, path) - }); - - if let Err(e) = result { + if let Err(e) = self.fs.mkdir(&path).await { reply.error(error_to_errno(&e)); return; } // Get the new directory's stats - let fs = self.fs.clone(); - let (stat_result, path) = self.runtime.block_on(async move { - let result = fs.stat(&path).await; - (result, path) - }); - - match stat_result { + match self.fs.stat(&path).await { Ok(Some(stats)) => { let attr = fillattr(&stats, self.uid, self.gid); self.add_path(attr.ino, path); @@ -574,20 +512,14 @@ impl Filesystem for AgentFSFuse { /// /// Verifies the target is a directory and is empty before removal. /// Returns `ENOTDIR` if not a directory, `ENOTEMPTY` if not empty. - fn rmdir(&mut self, _req: &Request, parent: u64, name: &OsStr, reply: ReplyEmpty) { + async fn rmdir(&self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEmpty) { let Some(path) = self.lookup_path(parent, name) else { reply.error(libc::ENOENT); return; }; // Verify target is a directory - let fs = self.fs.clone(); - let (stat_result, path) = self.runtime.block_on(async move { - let result = fs.lstat(&path).await; - (result, path) - }); - - let stats = match stat_result { + let stats = match self.fs.lstat(&path).await { Ok(Some(s)) => s, Ok(None) => { reply.error(libc::ENOENT); @@ -605,13 +537,7 @@ impl Filesystem for AgentFSFuse { } // Verify directory is empty - let fs = self.fs.clone(); - let (readdir_result, path) = self.runtime.block_on(async move { - let result = fs.readdir(&path).await; - (result, path) - }); - - match readdir_result { + match self.fs.readdir(&path).await { Ok(Some(entries)) if !entries.is_empty() => { reply.error(libc::ENOTEMPTY); return; @@ -629,10 +555,7 @@ impl Filesystem for AgentFSFuse { // Remove the directory let ino = stats.ino as u64; - let fs = self.fs.clone(); - let result = self.runtime.block_on(async move { fs.remove(&path).await }); - - match result { + match self.fs.remove(&path).await { Ok(()) => { self.drop_path(ino); reply.ok(); @@ -649,9 +572,9 @@ impl Filesystem for AgentFSFuse { /// /// Creates an empty file at `name` under `parent`, allocates a file handle, /// and returns both the file attributes and handle for immediate use. - fn create( - &mut self, - _req: &Request, + async fn create( + &self, + _req: &Request<'_>, parent: u64, name: &OsStr, mode: u32, @@ -665,13 +588,7 @@ impl Filesystem for AgentFSFuse { }; // Create file with mode, get stats and file handle in one operation - let fs = self.fs.clone(); - let path_for_create = path.clone(); - let result = self - .runtime - .block_on(async move { fs.create_file(&path_for_create, mode).await }); - - match result { + match self.fs.create_file(&path, mode).await { Ok((stats, file)) => { let attr = fillattr(&stats, self.uid, self.gid); self.add_path(attr.ino, path); @@ -690,9 +607,9 @@ impl Filesystem for AgentFSFuse { /// Creates a symbolic link. /// /// Creates a symlink at `name` under `parent` pointing to `link`. - fn symlink( - &mut self, - _req: &Request, + async fn symlink( + &self, + _req: &Request<'_>, parent: u64, link_name: &OsStr, target: &Path, @@ -708,26 +625,13 @@ impl Filesystem for AgentFSFuse { return; }; - let fs = self.fs.clone(); - let target_owned = target_str.to_string(); - let (result, path) = self.runtime.block_on(async move { - let result = fs.symlink(&target_owned, &path).await; - (result, path) - }); - - if let Err(e) = result { + if let Err(e) = self.fs.symlink(target_str, &path).await { reply.error(error_to_errno(&e)); return; } // Get the new symlink's stats - let fs = self.fs.clone(); - let (stat_result, path) = self.runtime.block_on(async move { - let result = fs.lstat(&path).await; - (result, path) - }); - - match stat_result { + match self.fs.lstat(&path).await { Ok(Some(stats)) => { let attr = fillattr(&stats, self.uid, self.gid); self.add_path(attr.ino, path); @@ -746,9 +650,9 @@ impl Filesystem for AgentFSFuse { /// /// Creates a new directory entry `newname` under `newparent` that refers to the /// same inode as `ino`. The link count of the inode is incremented. - fn link( - &mut self, - _req: &Request, + async fn link( + &self, + _req: &Request<'_>, ino: u64, newparent: u64, newname: &OsStr, @@ -766,25 +670,13 @@ impl Filesystem for AgentFSFuse { return; }; - let fs = self.fs.clone(); - let (result, newpath) = self.runtime.block_on(async move { - let result = fs.link(&oldpath, &newpath).await; - (result, newpath) - }); - - if let Err(e) = result { + if let Err(e) = self.fs.link(&oldpath, &newpath).await { reply.error(error_to_errno(&e)); return; } // Get the new link's stats - let fs = self.fs.clone(); - let (stat_result, newpath) = self.runtime.block_on(async move { - let result = fs.lstat(&newpath).await; - (result, newpath) - }); - - match stat_result { + match self.fs.lstat(&newpath).await { Ok(Some(stats)) => { let attr = fillattr(&stats, self.uid, self.gid); self.add_path(attr.ino, newpath); @@ -802,27 +694,21 @@ impl Filesystem for AgentFSFuse { /// Removes a file (unlinks it from the directory). /// /// Gets the file's inode before removal to clean up the path cache. - fn unlink(&mut self, _req: &Request, parent: u64, name: &OsStr, reply: ReplyEmpty) { + async fn unlink(&self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEmpty) { let Some(path) = self.lookup_path(parent, name) else { reply.error(libc::ENOENT); return; }; // Get inode before removing so we can uncache - let fs = self.fs.clone(); - let (stat_result, path) = self.runtime.block_on(async move { - let result = fs.lstat(&path).await; - (result, path) - }); - - let stats = match &stat_result { + let stats = match self.fs.lstat(&path).await { Ok(Some(s)) => s, Ok(None) => { reply.error(libc::ENOENT); return; } Err(e) => { - reply.error(error_to_errno(e)); + reply.error(error_to_errno(&e)); return; } }; @@ -835,8 +721,7 @@ impl Filesystem for AgentFSFuse { let ino = stats.ino as u64; let nlink = stats.nlink; - let fs = self.fs.clone(); - let result = self.runtime.block_on(async move { fs.remove(&path).await }); + let result = self.fs.remove(&path).await; match result { Ok(()) => { @@ -856,9 +741,9 @@ impl Filesystem for AgentFSFuse { /// /// Moves `name` from `parent` to `newname` under `newparent`. Updates the /// path cache accordingly, removing any replaced destination entry. - fn rename( - &mut self, - _req: &Request, + async fn rename( + &self, + _req: &Request<'_>, parent: u64, name: &OsStr, newparent: u64, @@ -877,31 +762,13 @@ impl Filesystem for AgentFSFuse { }; // Get source inode before rename so we can update cache - let fs = self.fs.clone(); - let (src_stat, from_path) = self.runtime.block_on(async move { - let result = fs.stat(&from_path).await; - (result, from_path) - }); - - let src_ino = src_stat.ok().flatten().map(|s| s.ino as u64); + let src_ino = self.fs.stat(&from_path).await.ok().flatten().map(|s| s.ino as u64); // Check if destination exists and get its inode for cache cleanup - let fs = self.fs.clone(); - let (dst_stat, to_path) = self.runtime.block_on(async move { - let result = fs.stat(&to_path).await; - (result, to_path) - }); - - let dst_ino = dst_stat.ok().flatten().map(|s| s.ino as u64); + let dst_ino = self.fs.stat(&to_path).await.ok().flatten().map(|s| s.ino as u64); // Perform the rename - let fs = self.fs.clone(); - let (result, to_path) = self.runtime.block_on(async move { - let result = fs.rename(&from_path, &to_path).await; - (result, to_path) - }); - - match result { + match self.fs.rename(&from_path, &to_path).await { Ok(()) => { // Update path cache: remove old path, add new path if let Some(ino) = src_ino { @@ -925,19 +792,13 @@ impl Filesystem for AgentFSFuse { /// Opens a file for reading or writing. /// /// Allocates a file handle and opens the file in the filesystem layer. - fn open(&mut self, _req: &Request, ino: u64, _flags: i32, reply: ReplyOpen) { + async fn open(&self, _req: &Request<'_>, ino: u64, _flags: i32, reply: ReplyOpen) { let Some(path) = self.get_path(ino) else { reply.error(libc::ENOENT); return; }; - let fs = self.fs.clone(); - let path_clone = path.clone(); - let result = self - .runtime - .block_on(async move { fs.open(&path_clone).await }); - - match result { + match self.fs.open(&path).await { Ok(file) => { let fh = self.alloc_fh(); self.open_files.lock().insert(fh, OpenFile { file }); @@ -948,9 +809,9 @@ impl Filesystem for AgentFSFuse { } /// Reads data using the file handle. - fn read( - &mut self, - _req: &Request, + async fn read( + &self, + _req: &Request<'_>, _ino: u64, fh: u64, offset: i64, @@ -968,20 +829,16 @@ impl Filesystem for AgentFSFuse { open_file.file.clone() }; - let result = self - .runtime - .block_on(async move { file.pread(offset as u64, size as u64).await }); - - match result { + match file.pread(offset as u64, size as u64).await { Ok(data) => reply.data(&data), Err(e) => reply.error(error_to_errno(&e)), } } /// Writes data using the file handle. - fn write( - &mut self, - _req: &Request, + async fn write( + &self, + _req: &Request<'_>, _ino: u64, fh: u64, offset: i64, @@ -1001,12 +858,7 @@ impl Filesystem for AgentFSFuse { }; let data_len = data.len(); - let data_vec = data.to_vec(); - let result = self - .runtime - .block_on(async move { file.pwrite(offset as u64, &data_vec).await }); - - match result { + match file.pwrite(offset as u64, data).await { Ok(()) => reply.written(data_len as u32), Err(e) => reply.error(error_to_errno(&e)), } @@ -1015,7 +867,7 @@ impl Filesystem for AgentFSFuse { /// Flushes data to the backend storage. /// /// Since writes go directly to the database, this is a no-op. - fn flush(&mut self, _req: &Request, _ino: u64, fh: u64, _lock_owner: u64, reply: ReplyEmpty) { + async fn flush(&self, _req: &Request<'_>, _ino: u64, fh: u64, _lock_owner: u64, reply: ReplyEmpty) { let open_files = self.open_files.lock(); if open_files.contains_key(&fh) { reply.ok(); @@ -1028,7 +880,7 @@ impl Filesystem for AgentFSFuse { /// /// This now uses the file handle's fsync which knows which layer(s) the /// file exists in, avoiding errors when a file only exists in one layer. - fn fsync(&mut self, _req: &Request, _ino: u64, fh: u64, _datasync: bool, reply: ReplyEmpty) { + async fn fsync(&self, _req: &Request<'_>, _ino: u64, fh: u64, _datasync: bool, reply: ReplyEmpty) { let file = { let open_files = self.open_files.lock(); match open_files.get(&fh) { @@ -1040,9 +892,7 @@ impl Filesystem for AgentFSFuse { } }; - let result = self.runtime.block_on(async move { file.fsync().await }); - - match result { + match file.fsync().await { Ok(()) => reply.ok(), Err(e) => reply.error(error_to_errno(&e)), } @@ -1052,9 +902,9 @@ impl Filesystem for AgentFSFuse { /// /// Removes the file handle from the open files table. /// Since writes go directly to the database, no flushing is needed. - fn release( - &mut self, - _req: &Request, + async fn release( + &self, + _req: &Request<'_>, _ino: u64, fh: u64, _flags: i32, @@ -1069,15 +919,12 @@ impl Filesystem for AgentFSFuse { /// Returns filesystem statistics. /// /// Queries actual usage from the SDK and reports it to tools like `df`. - fn statfs(&mut self, _req: &Request, _ino: u64, reply: ReplyStatfs) { + async fn statfs(&self, _req: &Request<'_>, _ino: u64, reply: ReplyStatfs) { const BLOCK_SIZE: u64 = 4096; const TOTAL_INODES: u64 = 1_000_000; // Virtual limit const MAX_NAMELEN: u32 = 255; - let fs = self.fs.clone(); - let result = self.runtime.block_on(async move { fs.statfs().await }); - - let (used_blocks, used_inodes) = match result { + let (used_blocks, used_inodes) = match self.fs.statfs().await { Ok(stats) => { let used_blocks = stats.bytes_used.div_ceil(BLOCK_SIZE); (used_blocks, stats.inodes) @@ -1106,23 +953,18 @@ impl Filesystem for AgentFSFuse { impl AgentFSFuse { /// Create a new FUSE filesystem adapter wrapping a FileSystem instance. /// - /// The provided Tokio runtime is used to execute async FileSystem operations - /// from within synchronous FUSE callbacks via `block_on`. - /// /// The uid and gid are used for all file ownership to avoid "dubious ownership" /// errors from tools like git that check file ownership. fn new( fs: Arc, - runtime: Runtime, uid: u32, gid: u32, mountpoint_path: PathBuf, ) -> Self { Self { fs, - runtime, - path_cache: Arc::new(Mutex::new(HashMap::new())), - open_files: Arc::new(Mutex::new(HashMap::new())), + path_cache: Mutex::new(HashMap::new()), + open_files: Mutex::new(HashMap::new()), next_fh: AtomicU64::new(1), uid, gid, @@ -1240,17 +1082,16 @@ fn fillattr(stats: &Stats, uid: u32, gid: u32) -> FileAttr { } } -pub fn mount( +pub async fn mount( fs: Arc, opts: FuseMountOptions, - runtime: Runtime, ) -> anyhow::Result<()> { // Use provided uid/gid or default to current user // This avoids "dubious ownership" errors from git and similar tools let uid = opts.uid.unwrap_or_else(|| unsafe { libc::getuid() }); let gid = opts.gid.unwrap_or_else(|| unsafe { libc::getgid() }); - let fs = AgentFSFuse::new(fs, runtime, uid, gid, opts.mountpoint.clone()); + let fs = AgentFSFuse::new(fs, uid, gid, opts.mountpoint.clone()); fs.add_path(1, "/".to_string()); @@ -1262,7 +1103,7 @@ pub fn mount( mount_opts.push(MountOption::AllowRoot); } - crate::fuser::mount2(fs, &opts.mountpoint, &mount_opts)?; + crate::fuser::mount_async(fs, &opts.mountpoint, &mount_opts).await?; Ok(()) } diff --git a/cli/src/fuser/channel.rs b/cli/src/fuser/channel.rs index 0d84c5a7..235c24a1 100644 --- a/cli/src/fuser/channel.rs +++ b/cli/src/fuser/channel.rs @@ -2,16 +2,20 @@ use std::{ fs::File, io, os::{ - fd::{AsFd, BorrowedFd}, - unix::prelude::AsRawFd, + fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd, IntoRawFd}, + unix::prelude::OwnedFd, }, sync::Arc, }; use libc::{c_int, c_void, size_t}; +use tokio::io::unix::AsyncFd; use super::reply::ReplySender; +/// Size of the buffer for reading a request from the kernel. +pub const BUFFER_SIZE: usize = super::session::MAX_WRITE_SIZE + 4096; + /// A raw communication channel to the FUSE kernel driver #[derive(Debug)] pub struct Channel(Arc); @@ -54,6 +58,77 @@ impl Channel { // a sender by using the same file and use it in other threads. ChannelSender(self.0.clone()) } + + /// Create an async channel from this channel for use with tokio. + /// This duplicates the underlying file descriptor so the original channel + /// remains valid. + pub fn to_async(&self) -> io::Result { + let fd = self.0.as_raw_fd(); + unsafe { + // Duplicate the fd for async use + let dup_fd = libc::dup(fd); + if dup_fd < 0 { + return Err(io::Error::last_os_error()); + } + // Set non-blocking mode on the duplicated fd + let flags = libc::fcntl(dup_fd, libc::F_GETFL); + if flags < 0 { + let err = io::Error::last_os_error(); + libc::close(dup_fd); + return Err(err); + } + if libc::fcntl(dup_fd, libc::F_SETFL, flags | libc::O_NONBLOCK) < 0 { + let err = io::Error::last_os_error(); + libc::close(dup_fd); + return Err(err); + } + let async_fd = AsyncFd::new(OwnedFd::from_raw_fd(dup_fd))?; + Ok(AsyncChannel { inner: async_fd }) + } + } +} + +/// An async communication channel to the FUSE kernel driver. +/// Uses tokio's AsyncFd for non-blocking I/O. +#[derive(Debug)] +pub struct AsyncChannel { + inner: AsyncFd, +} + +impl AsyncChannel { + /// Receives data asynchronously, returning the size read and a buffer containing the data. + pub async fn receive(&self) -> io::Result<(usize, Vec)> { + let mut buffer = vec![0u8; BUFFER_SIZE]; + + loop { + let mut guard = self.inner.readable().await?; + + match guard.try_io(|inner| { + let fd = inner.get_ref().as_raw_fd(); + let rc = unsafe { + libc::read(fd, buffer.as_mut_ptr() as *mut c_void, buffer.len() as size_t) + }; + if rc < 0 { + Err(io::Error::last_os_error()) + } else { + Ok(rc as usize) + } + }) { + Ok(Ok(size)) => return Ok((size, buffer)), + Ok(Err(e)) => return Err(e), + Err(_would_block) => continue, + } + } + } + + /// Returns a sender object for this channel. + pub fn sender(&self) -> ChannelSender { + let fd = self.inner.get_ref().as_raw_fd(); + // Create a new Arc from the fd (note: this doesn't duplicate the fd, + // but since writev is thread-safe and we're only writing, this is safe) + let file = unsafe { Arc::new(File::from_raw_fd(libc::dup(fd))) }; + ChannelSender(file) + } } #[derive(Clone, Debug)] diff --git a/cli/src/fuser/mod.rs b/cli/src/fuser/mod.rs index 1c74ab18..12eb0f34 100644 --- a/cli/src/fuser/mod.rs +++ b/cli/src/fuser/mod.rs @@ -3,6 +3,8 @@ //! This is an improved rewrite of the FUSE userspace library (lowlevel interface) to fully take //! advantage of Rust's architecture. This version uses a pure-Rust mounting implementation //! and does not require libfuse. +//! +//! This version has been modified to support async filesystem operations natively. #![allow( missing_docs, @@ -14,6 +16,7 @@ clippy::io_other_error )] +use async_trait::async_trait; use libc::{c_int, ENOSYS, EPERM}; use log::warn; use mnt::mount_options::parse_options_from_args; @@ -265,44 +268,47 @@ impl KernelConfig { } } -/// Filesystem trait. +/// Async Filesystem trait. /// /// This trait must be implemented to provide a userspace filesystem via FUSE. +/// All methods are async and take `&self` (not `&mut self`) to allow concurrent +/// handling of filesystem operations. #[allow(clippy::too_many_arguments)] -pub trait Filesystem { +#[async_trait] +pub trait Filesystem: Send + Sync + 'static { /// Initialize filesystem. - fn init(&mut self, _req: &Request<'_>, _config: &mut KernelConfig) -> Result<(), c_int> { + async fn init(&self, _req: &Request<'_>, _config: &mut KernelConfig) -> Result<(), c_int> { Ok(()) } /// Clean up filesystem. - fn destroy(&mut self) {} + async fn destroy(&self) {} /// Look up a directory entry by name and get its attributes. - fn lookup(&mut self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEntry) { + async fn lookup(&self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEntry) { warn!("[Not Implemented] lookup(parent: {parent:#x?}, name {name:?})"); reply.error(ENOSYS); } /// Forget about an inode. - fn forget(&mut self, _req: &Request<'_>, _ino: u64, _nlookup: u64) {} + async fn forget(&self, _req: &Request<'_>, _ino: u64, _nlookup: u64) {} /// Like forget, but take multiple forget requests at once for performance. - fn batch_forget(&mut self, req: &Request<'_>, nodes: &[fuse_forget_one]) { + async fn batch_forget(&self, req: &Request<'_>, nodes: &[fuse_forget_one]) { for node in nodes { - self.forget(req, node.nodeid, node.nlookup); + self.forget(req, node.nodeid, node.nlookup).await; } } /// Get file attributes. - fn getattr(&mut self, _req: &Request<'_>, ino: u64, fh: Option, reply: ReplyAttr) { + async fn getattr(&self, _req: &Request<'_>, ino: u64, fh: Option, reply: ReplyAttr) { warn!("[Not Implemented] getattr(ino: {ino:#x?}, fh: {fh:#x?})"); reply.error(ENOSYS); } /// Set file attributes. - fn setattr( - &mut self, + async fn setattr( + &self, _req: &Request<'_>, ino: u64, mode: Option, @@ -327,14 +333,14 @@ pub trait Filesystem { } /// Read symbolic link. - fn readlink(&mut self, _req: &Request<'_>, ino: u64, reply: ReplyData) { + async fn readlink(&self, _req: &Request<'_>, ino: u64, reply: ReplyData) { warn!("[Not Implemented] readlink(ino: {ino:#x?})"); reply.error(ENOSYS); } /// Create file node. - fn mknod( - &mut self, + async fn mknod( + &self, _req: &Request<'_>, parent: u64, name: &OsStr, @@ -351,8 +357,8 @@ pub trait Filesystem { } /// Create a directory. - fn mkdir( - &mut self, + async fn mkdir( + &self, _req: &Request<'_>, parent: u64, name: &OsStr, @@ -367,20 +373,20 @@ pub trait Filesystem { } /// Remove a file. - fn unlink(&mut self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEmpty) { + async fn unlink(&self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEmpty) { warn!("[Not Implemented] unlink(parent: {parent:#x?}, name: {name:?})",); reply.error(ENOSYS); } /// Remove a directory. - fn rmdir(&mut self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEmpty) { + async fn rmdir(&self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEmpty) { warn!("[Not Implemented] rmdir(parent: {parent:#x?}, name: {name:?})",); reply.error(ENOSYS); } /// Create a symbolic link. - fn symlink( - &mut self, + async fn symlink( + &self, _req: &Request<'_>, parent: u64, link_name: &OsStr, @@ -394,8 +400,8 @@ pub trait Filesystem { } /// Rename a file. - fn rename( - &mut self, + async fn rename( + &self, _req: &Request<'_>, parent: u64, name: &OsStr, @@ -412,8 +418,8 @@ pub trait Filesystem { } /// Create a hard link. - fn link( - &mut self, + async fn link( + &self, _req: &Request<'_>, ino: u64, newparent: u64, @@ -427,13 +433,13 @@ pub trait Filesystem { } /// Open a file. - fn open(&mut self, _req: &Request<'_>, _ino: u64, _flags: i32, reply: ReplyOpen) { + async fn open(&self, _req: &Request<'_>, _ino: u64, _flags: i32, reply: ReplyOpen) { reply.opened(0, 0); } /// Read data. - fn read( - &mut self, + async fn read( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -451,8 +457,8 @@ pub trait Filesystem { } /// Write data. - fn write( - &mut self, + async fn write( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -473,14 +479,14 @@ pub trait Filesystem { } /// Flush method. - fn flush(&mut self, _req: &Request<'_>, ino: u64, fh: u64, lock_owner: u64, reply: ReplyEmpty) { + async fn flush(&self, _req: &Request<'_>, ino: u64, fh: u64, lock_owner: u64, reply: ReplyEmpty) { warn!("[Not Implemented] flush(ino: {ino:#x?}, fh: {fh}, lock_owner: {lock_owner:?})"); reply.error(ENOSYS); } /// Release an open file. - fn release( - &mut self, + async fn release( + &self, _req: &Request<'_>, _ino: u64, _fh: u64, @@ -493,19 +499,19 @@ pub trait Filesystem { } /// Synchronize file contents. - fn fsync(&mut self, _req: &Request<'_>, ino: u64, fh: u64, datasync: bool, reply: ReplyEmpty) { + async fn fsync(&self, _req: &Request<'_>, ino: u64, fh: u64, datasync: bool, reply: ReplyEmpty) { warn!("[Not Implemented] fsync(ino: {ino:#x?}, fh: {fh}, datasync: {datasync})"); reply.error(ENOSYS); } /// Open a directory. - fn opendir(&mut self, _req: &Request<'_>, _ino: u64, _flags: i32, reply: ReplyOpen) { + async fn opendir(&self, _req: &Request<'_>, _ino: u64, _flags: i32, reply: ReplyOpen) { reply.opened(0, 0); } /// Read directory. - fn readdir( - &mut self, + async fn readdir( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -517,8 +523,8 @@ pub trait Filesystem { } /// Read directory with attributes. - fn readdirplus( - &mut self, + async fn readdirplus( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -530,8 +536,8 @@ pub trait Filesystem { } /// Release an open directory. - fn releasedir( - &mut self, + async fn releasedir( + &self, _req: &Request<'_>, _ino: u64, _fh: u64, @@ -542,8 +548,8 @@ pub trait Filesystem { } /// Synchronize directory contents. - fn fsyncdir( - &mut self, + async fn fsyncdir( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -555,13 +561,13 @@ pub trait Filesystem { } /// Get file system statistics. - fn statfs(&mut self, _req: &Request<'_>, _ino: u64, reply: ReplyStatfs) { + async fn statfs(&self, _req: &Request<'_>, _ino: u64, reply: ReplyStatfs) { reply.statfs(0, 0, 0, 0, 0, 512, 255, 0); } /// Set an extended attribute. - fn setxattr( - &mut self, + async fn setxattr( + &self, _req: &Request<'_>, ino: u64, name: &OsStr, @@ -578,8 +584,8 @@ pub trait Filesystem { } /// Get an extended attribute. - fn getxattr( - &mut self, + async fn getxattr( + &self, _req: &Request<'_>, ino: u64, name: &OsStr, @@ -591,26 +597,26 @@ pub trait Filesystem { } /// List extended attribute names. - fn listxattr(&mut self, _req: &Request<'_>, ino: u64, size: u32, reply: ReplyXattr) { + async fn listxattr(&self, _req: &Request<'_>, ino: u64, size: u32, reply: ReplyXattr) { warn!("[Not Implemented] listxattr(ino: {ino:#x?}, size: {size})"); reply.error(ENOSYS); } /// Remove an extended attribute. - fn removexattr(&mut self, _req: &Request<'_>, ino: u64, name: &OsStr, reply: ReplyEmpty) { + async fn removexattr(&self, _req: &Request<'_>, ino: u64, name: &OsStr, reply: ReplyEmpty) { warn!("[Not Implemented] removexattr(ino: {ino:#x?}, name: {name:?})"); reply.error(ENOSYS); } /// Check file access permissions. - fn access(&mut self, _req: &Request<'_>, ino: u64, mask: i32, reply: ReplyEmpty) { + async fn access(&self, _req: &Request<'_>, ino: u64, mask: i32, reply: ReplyEmpty) { warn!("[Not Implemented] access(ino: {ino:#x?}, mask: {mask})"); reply.error(ENOSYS); } /// Create and open a file. - fn create( - &mut self, + async fn create( + &self, _req: &Request<'_>, parent: u64, name: &OsStr, @@ -627,8 +633,8 @@ pub trait Filesystem { } /// Test for a POSIX file lock. - fn getlk( - &mut self, + async fn getlk( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -647,8 +653,8 @@ pub trait Filesystem { } /// Acquire, modify or release a POSIX file lock. - fn setlk( - &mut self, + async fn setlk( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -668,14 +674,14 @@ pub trait Filesystem { } /// Map block index within file to block index within device. - fn bmap(&mut self, _req: &Request<'_>, ino: u64, blocksize: u32, idx: u64, reply: ReplyBmap) { + async fn bmap(&self, _req: &Request<'_>, ino: u64, blocksize: u32, idx: u64, reply: ReplyBmap) { warn!("[Not Implemented] bmap(ino: {ino:#x?}, blocksize: {blocksize}, idx: {idx})",); reply.error(ENOSYS); } /// Control device. - fn ioctl( - &mut self, + async fn ioctl( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -694,8 +700,8 @@ pub trait Filesystem { } /// Poll for events. - fn poll( - &mut self, + async fn poll( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -712,8 +718,8 @@ pub trait Filesystem { } /// Preallocate or deallocate space to a file. - fn fallocate( - &mut self, + async fn fallocate( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -730,8 +736,8 @@ pub trait Filesystem { } /// Reposition read/write file offset. - fn lseek( - &mut self, + async fn lseek( + &self, _req: &Request<'_>, ino: u64, fh: u64, @@ -747,8 +753,8 @@ pub trait Filesystem { } /// Copy the specified range from the source inode to the destination inode. - fn copy_file_range( - &mut self, + async fn copy_file_range( + &self, _req: &Request<'_>, ino_in: u64, fh_in: u64, @@ -769,49 +775,20 @@ pub trait Filesystem { } } -/// Mount the given filesystem to the given mountpoint. This function will -/// not return until the filesystem is unmounted. -#[deprecated(note = "use mount2() instead")] -pub fn mount>( - filesystem: FS, - mountpoint: P, - options: &[&OsStr], -) -> io::Result<()> { - let options = parse_options_from_args(options)?; - mount2(filesystem, mountpoint, options.as_ref()) -} - -/// Mount the given filesystem to the given mountpoint. This function will -/// not return until the filesystem is unmounted. -pub fn mount2>( +/// Mount the given filesystem to the given mountpoint asynchronously. +/// This function will not return until the filesystem is unmounted. +pub async fn mount_async>( filesystem: FS, mountpoint: P, options: &[MountOption], ) -> io::Result<()> { check_option_conflicts(options)?; - Session::new(filesystem, mountpoint.as_ref(), options).and_then(|mut se| se.run()) -} - -/// Mount the given filesystem to the given mountpoint. This function spawns -/// a background thread to handle filesystem operations while being mounted -/// and therefore returns immediately. -#[deprecated(note = "use spawn_mount2() instead")] -pub fn spawn_mount<'a, FS: Filesystem + Send + 'static + 'a, P: AsRef>( - filesystem: FS, - mountpoint: P, - options: &[&OsStr], -) -> io::Result { - let options: Option> = options - .iter() - .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(session::Session::spawn) + let session = Session::new(filesystem, mountpoint.as_ref(), options)?; + std::sync::Arc::new(session).run().await } /// Mount the given filesystem to the given mountpoint. This function spawns -/// a background thread to handle filesystem operations while being mounted +/// a background tokio task to handle filesystem operations while being mounted /// and therefore returns immediately. pub fn spawn_mount2<'a, FS: Filesystem + Send + 'static + 'a, P: AsRef>( filesystem: FS, diff --git a/cli/src/fuser/request.rs b/cli/src/fuser/request.rs index a0a4bca0..577f2383 100644 --- a/cli/src/fuser/request.rs +++ b/cli/src/fuser/request.rs @@ -10,6 +10,8 @@ use log::{debug, error, warn}; use std::convert::TryFrom; use std::convert::TryInto; use std::path::Path; +use std::sync::atomic::Ordering; +use std::sync::Arc; use super::channel::ChannelSender; use super::ll::Request as _; @@ -46,14 +48,13 @@ impl<'a> Request<'a> { Some(Self { ch, data, request }) } - /// Dispatch request to the given filesystem. - /// This calls the appropriate filesystem operation method for the - /// request and sends back the returned reply to the kernel - pub(crate) fn dispatch(&self, se: &mut Session) { + /// Dispatch request asynchronously to the given filesystem. + /// This is the async version that calls async filesystem methods. + pub(crate) async fn dispatch_async(&self, se: &Arc>) { debug!("{}", self.request); let unique = self.request.unique(); - let res = match self.dispatch_req(se) { + let res = match self.dispatch_req_async(se).await { Ok(Some(resp)) => resp, Ok(None) => return, Err(errno) => self.request.reply_err(errno), @@ -65,9 +66,10 @@ impl<'a> Request<'a> { } } - fn dispatch_req( + /// Async dispatch implementation + async fn dispatch_req_async( &self, - se: &mut Session, + se: &Arc>, ) -> Result>, Errno> { let op = self.request.operation().map_err(|_| Errno::ENOSYS)?; // Implement allow_root & access check for auto_unmount @@ -79,7 +81,6 @@ impl<'a> Request<'a> { #[cfg(feature = "abi-7-21")] { match op { - // Only allow operations that the kernel may issue without a uid set ll::Operation::Init(_) | ll::Operation::Destroy(_) | ll::Operation::Read(_) @@ -100,7 +101,6 @@ impl<'a> Request<'a> { #[cfg(not(feature = "abi-7-21"))] { match op { - // Only allow operations that the kernel may issue without a uid set ll::Operation::Init(_) | ll::Operation::Destroy(_) | ll::Operation::Read(_) @@ -119,27 +119,21 @@ impl<'a> Request<'a> { } } match op { - // Filesystem initialization ll::Operation::Init(x) => { - // We don't support ABI versions before 7.6 let v = x.version(); if v < ll::Version(7, 6) { error!("Unsupported FUSE ABI version {v}"); return Err(Errno::EPROTO); } - // Remember ABI version supported by kernel - se.proto_major = v.major(); - se.proto_minor = v.minor(); + se.proto_major.store(v.major(), Ordering::Release); + se.proto_minor.store(v.minor(), Ordering::Release); let mut config = KernelConfig::new(x.capabilities(), x.max_readahead()); - // Call filesystem init method and give it a chance to return an error se.filesystem .init(self, &mut config) + .await .map_err(Errno::from_i32)?; - // Reply with our desired version and settings. If the kernel supports a - // larger major version, it'll re-send a matching init message. If it - // supports only lower major versions, we replied with an error above. debug!( "INIT response: ABI {}.{}, flags {:#x}, max readahead {}, max write {}", abi::FUSE_KERNEL_VERSION, @@ -148,28 +142,24 @@ impl<'a> Request<'a> { config.max_readahead, config.max_write ); - se.initialized = true; + se.initialized.store(true, Ordering::Release); return Ok(Some(x.reply(&config))); } - // Any operation is invalid before initialization - _ if !se.initialized => { + _ if !se.initialized.load(Ordering::Acquire) => { 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.filesystem.destroy().await; + se.destroyed.store(true, Ordering::Release); return Ok(Some(x.reply())); } - // Any operation is invalid after destroy - _ if se.destroyed => { + _ if se.destroyed.load(Ordering::Acquire) => { warn!("Ignoring FUSE operation after destroy: {}", self.request); return Err(Errno::EIO); } ll::Operation::Interrupt(_) => { - // TODO: handle FUSE_INTERRUPT return Err(Errno::ENOSYS); } @@ -179,11 +169,11 @@ impl<'a> Request<'a> { self.request.nodeid().into(), x.name().as_ref(), self.reply(), - ); + ).await; } ll::Operation::Forget(x) => { se.filesystem - .forget(self, self.request.nodeid().into(), x.nlookup()); // no reply + .forget(self, self.request.nodeid().into(), x.nlookup()).await; } ll::Operation::GetAttr(_attr) => { se.filesystem.getattr( @@ -191,7 +181,7 @@ impl<'a> Request<'a> { self.request.nodeid().into(), _attr.file_handle().map(std::convert::Into::into), self.reply(), - ); + ).await; } ll::Operation::SetAttr(x) => { se.filesystem.setattr( @@ -210,11 +200,11 @@ impl<'a> Request<'a> { x.bkuptime(), x.flags(), self.reply(), - ); + ).await; } ll::Operation::ReadLink(_) => { se.filesystem - .readlink(self, self.request.nodeid().into(), self.reply()); + .readlink(self, self.request.nodeid().into(), self.reply()).await; } ll::Operation::MkNod(x) => { se.filesystem.mknod( @@ -225,7 +215,7 @@ impl<'a> Request<'a> { x.umask(), x.rdev(), self.reply(), - ); + ).await; } ll::Operation::MkDir(x) => { se.filesystem.mkdir( @@ -235,7 +225,7 @@ impl<'a> Request<'a> { x.mode(), x.umask(), self.reply(), - ); + ).await; } ll::Operation::Unlink(x) => { se.filesystem.unlink( @@ -243,7 +233,7 @@ impl<'a> Request<'a> { self.request.nodeid().into(), x.name().as_ref(), self.reply(), - ); + ).await; } ll::Operation::RmDir(x) => { se.filesystem.rmdir( @@ -251,7 +241,7 @@ impl<'a> Request<'a> { self.request.nodeid().into(), x.name().as_ref(), self.reply(), - ); + ).await; } ll::Operation::SymLink(x) => { se.filesystem.symlink( @@ -260,7 +250,7 @@ impl<'a> Request<'a> { x.link_name().as_ref(), Path::new(x.target()), self.reply(), - ); + ).await; } ll::Operation::Rename(x) => { se.filesystem.rename( @@ -271,7 +261,7 @@ impl<'a> Request<'a> { x.dest().name.as_ref(), 0, self.reply(), - ); + ).await; } ll::Operation::Link(x) => { se.filesystem.link( @@ -280,11 +270,11 @@ impl<'a> Request<'a> { self.request.nodeid().into(), x.dest().name.as_ref(), self.reply(), - ); + ).await; } ll::Operation::Open(x) => { se.filesystem - .open(self, self.request.nodeid().into(), x.flags(), self.reply()); + .open(self, self.request.nodeid().into(), x.flags(), self.reply()).await; } ll::Operation::Read(x) => { se.filesystem.read( @@ -296,7 +286,7 @@ impl<'a> Request<'a> { x.flags(), x.lock_owner().map(std::convert::Into::into), self.reply(), - ); + ).await; } ll::Operation::Write(x) => { se.filesystem.write( @@ -309,7 +299,7 @@ impl<'a> Request<'a> { x.flags(), x.lock_owner().map(std::convert::Into::into), self.reply(), - ); + ).await; } ll::Operation::Flush(x) => { se.filesystem.flush( @@ -318,7 +308,7 @@ impl<'a> Request<'a> { x.file_handle().into(), x.lock_owner().into(), self.reply(), - ); + ).await; } ll::Operation::Release(x) => { se.filesystem.release( @@ -329,7 +319,7 @@ impl<'a> Request<'a> { x.lock_owner().map(std::convert::Into::into), x.flush(), self.reply(), - ); + ).await; } ll::Operation::FSync(x) => { se.filesystem.fsync( @@ -338,11 +328,11 @@ impl<'a> Request<'a> { x.file_handle().into(), x.fdatasync(), self.reply(), - ); + ).await; } ll::Operation::OpenDir(x) => { se.filesystem - .opendir(self, self.request.nodeid().into(), x.flags(), self.reply()); + .opendir(self, self.request.nodeid().into(), x.flags(), self.reply()).await; } ll::Operation::ReadDir(x) => { se.filesystem.readdir( @@ -355,7 +345,7 @@ impl<'a> Request<'a> { self.ch.clone(), x.size() as usize, ), - ); + ).await; } ll::Operation::ReleaseDir(x) => { se.filesystem.releasedir( @@ -364,7 +354,7 @@ impl<'a> Request<'a> { x.file_handle().into(), x.flags(), self.reply(), - ); + ).await; } ll::Operation::FSyncDir(x) => { se.filesystem.fsyncdir( @@ -373,11 +363,11 @@ impl<'a> Request<'a> { x.file_handle().into(), x.fdatasync(), self.reply(), - ); + ).await; } ll::Operation::StatFs(_) => { se.filesystem - .statfs(self, self.request.nodeid().into(), self.reply()); + .statfs(self, self.request.nodeid().into(), self.reply()).await; } ll::Operation::SetXAttr(x) => { se.filesystem.setxattr( @@ -388,7 +378,7 @@ impl<'a> Request<'a> { x.flags(), x.position(), self.reply(), - ); + ).await; } ll::Operation::GetXAttr(x) => { se.filesystem.getxattr( @@ -397,11 +387,11 @@ impl<'a> Request<'a> { x.name(), x.size_u32(), self.reply(), - ); + ).await; } ll::Operation::ListXAttr(x) => { se.filesystem - .listxattr(self, self.request.nodeid().into(), x.size(), self.reply()); + .listxattr(self, self.request.nodeid().into(), x.size(), self.reply()).await; } ll::Operation::RemoveXAttr(x) => { se.filesystem.removexattr( @@ -409,11 +399,11 @@ impl<'a> Request<'a> { self.request.nodeid().into(), x.name(), self.reply(), - ); + ).await; } ll::Operation::Access(x) => { se.filesystem - .access(self, self.request.nodeid().into(), x.mask(), self.reply()); + .access(self, self.request.nodeid().into(), x.mask(), self.reply()).await; } ll::Operation::Create(x) => { se.filesystem.create( @@ -424,7 +414,7 @@ impl<'a> Request<'a> { x.umask(), x.flags(), self.reply(), - ); + ).await; } ll::Operation::GetLk(x) => { se.filesystem.getlk( @@ -437,7 +427,7 @@ impl<'a> Request<'a> { x.lock().typ, x.lock().pid, self.reply(), - ); + ).await; } ll::Operation::SetLk(x) => { se.filesystem.setlk( @@ -451,7 +441,7 @@ impl<'a> Request<'a> { x.lock().pid, false, self.reply(), - ); + ).await; } ll::Operation::SetLkW(x) => { se.filesystem.setlk( @@ -465,7 +455,7 @@ impl<'a> Request<'a> { x.lock().pid, true, self.reply(), - ); + ).await; } ll::Operation::BMap(x) => { se.filesystem.bmap( @@ -474,7 +464,7 @@ impl<'a> Request<'a> { x.block_size(), x.block(), self.reply(), - ); + ).await; } ll::Operation::IoCtl(x) => { @@ -490,7 +480,7 @@ impl<'a> Request<'a> { x.in_data(), x.out_size(), self.reply(), - ); + ).await; } ll::Operation::Poll(x) => { let ph = PollHandle::new(se.ch.sender(), x.kernel_handle()); @@ -503,14 +493,13 @@ impl<'a> Request<'a> { x.events(), x.flags(), self.reply(), - ); + ).await; } ll::Operation::NotifyReply(_) => { - // TODO: handle FUSE_NOTIFY_REPLY return Err(Errno::ENOSYS); } ll::Operation::BatchForget(x) => { - se.filesystem.batch_forget(self, x.nodes()); // no reply + se.filesystem.batch_forget(self, x.nodes()).await; } #[cfg(feature = "abi-7-19")] ll::Operation::FAllocate(x) => { @@ -522,7 +511,7 @@ impl<'a> Request<'a> { x.len(), x.mode(), self.reply(), - ); + ).await; } #[cfg(feature = "abi-7-21")] ll::Operation::ReadDirPlus(x) => { @@ -536,7 +525,7 @@ impl<'a> Request<'a> { self.ch.clone(), x.size() as usize, ), - ); + ).await; } #[cfg(feature = "abi-7-23")] ll::Operation::Rename2(x) => { @@ -548,7 +537,7 @@ impl<'a> Request<'a> { x.to().name.as_ref(), x.flags(), self.reply(), - ); + ).await; } #[cfg(feature = "abi-7-24")] ll::Operation::Lseek(x) => { @@ -559,7 +548,7 @@ impl<'a> Request<'a> { x.offset(), x.whence(), self.reply(), - ); + ).await; } #[cfg(feature = "abi-7-28")] ll::Operation::CopyFileRange(x) => { @@ -575,16 +564,16 @@ impl<'a> Request<'a> { x.len(), x.flags().try_into().unwrap(), self.reply(), - ); + ).await; } #[cfg(target_os = "macos")] ll::Operation::SetVolName(x) => { - se.filesystem.setvolname(self, x.name(), self.reply()); + se.filesystem.setvolname(self, x.name(), self.reply()).await; } #[cfg(target_os = "macos")] ll::Operation::GetXTimes(x) => { se.filesystem - .getxtimes(self, x.nodeid().into(), self.reply()); + .getxtimes(self, x.nodeid().into(), self.reply()).await; } #[cfg(target_os = "macos")] ll::Operation::Exchange(x) => { @@ -596,11 +585,10 @@ impl<'a> Request<'a> { x.to().name.as_ref(), x.options(), self.reply(), - ); + ).await; } ll::Operation::CuseInit(_) => { - // TODO: handle CUSE_INIT return Err(Errno::ENOSYS); } } diff --git a/cli/src/fuser/session.rs b/cli/src/fuser/session.rs index ee0883a1..317d250d 100644 --- a/cli/src/fuser/session.rs +++ b/cli/src/fuser/session.rs @@ -4,27 +4,31 @@ //! point. A session begins by mounting the filesystem and ends by unmounting it. While the //! filesystem is mounted, the session loop receives, dispatches and replies to kernel requests //! for filesystem operations under its mount point. +//! +//! This implementation uses async I/O for high-performance concurrent request handling. use libc::{EAGAIN, EINTR, ENODEV, ENOENT}; -use log::{info, warn}; +use log::{debug, info, warn}; use nix::unistd::geteuid; use std::io; use std::os::fd::{AsFd, BorrowedFd, OwnedFd}; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::{Arc, Mutex}; -use std::thread::{self, JoinHandle}; +use tokio::sync::Semaphore; use super::ll::fuse_abi as abi; use super::request::Request; use super::Filesystem; +use super::KernelConfig; use super::MountOption; -use super::{channel::Channel, mnt::Mount}; +use super::{channel::AsyncChannel, channel::Channel, mnt::Mount}; use super::{channel::ChannelSender, notify::Notifier}; /// The max size of write requests from the kernel. The absolute minimum is 4k, /// FUSE recommends at least 128k, max 16M. The FUSE default is 16M on macOS -/// and 128k on other systems. -pub const MAX_WRITE_SIZE: usize = 16 * 1024 * 1024; +/// and 128k on other systems. We use 128k to minimize buffer allocation overhead. +pub const MAX_WRITE_SIZE: usize = 128 * 1024; /// Size of the buffer for reading a request from the kernel. Since the kernel may send /// up to `MAX_WRITE_SIZE` bytes in a write request, we use that value plus some extra space. @@ -42,11 +46,15 @@ pub enum SessionACL { Owner, } +/// Maximum number of concurrent FUSE requests to process. +/// Using 1 to serialize all requests and avoid database contention. +const MAX_CONCURRENT_REQUESTS: usize = 2; + /// The session data structure #[derive(Debug)] pub struct Session { - /// Filesystem operation implementations - pub(crate) filesystem: FS, + /// Filesystem operation implementations (Arc for sharing across async tasks) + pub(crate) filesystem: Arc, /// Communication channel to the kernel driver pub(crate) ch: Channel, /// Handle to the mount. Dropping this unmounts. @@ -57,13 +65,13 @@ pub struct Session { /// User that launched the fuser process pub(crate) session_owner: u32, /// FUSE protocol major version - pub(crate) proto_major: u32, + pub(crate) proto_major: AtomicU32, /// FUSE protocol minor version - pub(crate) proto_minor: u32, + pub(crate) proto_minor: AtomicU32, /// True if the filesystem is initialized (init operation done) - pub(crate) initialized: bool, + pub(crate) initialized: AtomicBool, /// True if the filesystem was destroyed (destroy operation done) - pub(crate) destroyed: bool, + pub(crate) destroyed: AtomicBool, } impl AsFd for Session { @@ -110,15 +118,15 @@ impl Session { }; Ok(Session { - filesystem, + filesystem: Arc::new(filesystem), ch, mount: Arc::new(Mutex::new(Some((mountpoint.to_owned(), mount)))), allowed, session_owner: geteuid().as_raw(), - proto_major: 0, - proto_minor: 0, - initialized: false, - destroyed: false, + proto_major: AtomicU32::new(0), + proto_minor: AtomicU32::new(0), + initialized: AtomicBool::new(false), + destroyed: AtomicBool::new(false), }) } @@ -127,54 +135,67 @@ impl Session { pub fn from_fd(filesystem: FS, fd: OwnedFd, acl: SessionACL) -> Self { let ch = Channel::new(Arc::new(fd.into())); Session { - filesystem, + filesystem: Arc::new(filesystem), ch, mount: Arc::new(Mutex::new(None)), allowed: acl, session_owner: geteuid().as_raw(), - proto_major: 0, - proto_minor: 0, - initialized: false, - destroyed: false, + proto_major: AtomicU32::new(0), + proto_minor: AtomicU32::new(0), + initialized: AtomicBool::new(false), + destroyed: AtomicBool::new(false), } } - /// Run the session loop that receives kernel requests and dispatches them to method - /// calls into the filesystem. This read-dispatch-loop is non-concurrent to prevent - /// having multiple buffers (which take up much memory), but the filesystem methods - /// may run concurrent by spawning threads. + /// Run the async session loop that receives kernel requests and dispatches them + /// concurrently to the filesystem implementation. + /// + /// This uses a semaphore-based backpressure mechanism to limit concurrent requests + /// and prevent overwhelming the connection pool. + /// /// # Errors /// Returns any final error when the session comes to an end. - pub fn run(&mut self) -> io::Result<()> { - // Buffer for receiving requests from the kernel. Only one is allocated and - // it is reused immediately after dispatching to conserve memory and allocations. - let mut buffer = vec![0; BUFFER_SIZE]; - let buf = aligned_sub_buf(&mut buffer, std::mem::align_of::()); + pub async fn run(self: Arc) -> io::Result<()> { + // Create async channel for non-blocking I/O + let async_ch = self.ch.to_async()?; + let semaphore = Arc::new(Semaphore::new(MAX_CONCURRENT_REQUESTS)); + loop { - // Read the next request from the given channel to kernel driver - // The kernel driver makes sure that we get exactly one request per read - match self.ch.receive(buf) { - Ok(size) => match Request::new(self.ch.sender(), &buf[..size]) { - // Dispatch request - Some(req) => req.dispatch(self), - // Quit loop on illegal request - None => break, - }, + // Read the next request from the kernel driver + let (size, buffer) = match async_ch.receive().await { + Ok(result) => result, Err(err) => match err.raw_os_error() { - Some( - ENOENT // Operation interrupted. Accordingly to FUSE, this is safe to retry - | EINTR // Interrupted system call, retry - | EAGAIN // Explicitly instructed to try again - ) => continue, + Some(ENOENT | EINTR | EAGAIN) => continue, Some(ENODEV) => break, - // Unhandled error _ => return Err(err), }, - } + }; + + // Acquire a permit for backpressure before spawning + let permit = semaphore.clone().acquire_owned().await.unwrap(); + + // Clone what we need for the spawned task + let session = self.clone(); + let sender = async_ch.sender(); + + // Spawn an async task to handle this request + tokio::spawn(async move { + let _permit = permit; // Hold until complete + // Parse request with the owned buffer + if let Some(req) = Request::new(sender, &buffer[..size]) { + req.dispatch_async(&session).await; + } + }); } + Ok(()) } + /// Create an Arc-wrapped session for async operation. + pub fn into_arc(self) -> Arc { + Arc::new(self) + } + /// Unmount the filesystem pub fn unmount(&mut self) { drop(std::mem::take(&mut *self.mount.lock().unwrap())); @@ -225,9 +246,10 @@ impl Session { impl Drop for Session { fn drop(&mut self) { - if !self.destroyed { - self.filesystem.destroy(); - self.destroyed = true; + if !self.destroyed.load(Ordering::Acquire) { + // Note: We can't call async destroy() from drop, so filesystem cleanup + // should be handled explicitly before dropping the session. + self.destroyed.store(true, Ordering::Release); } if let Some((mountpoint, _mount)) = std::mem::take(&mut *self.mount.lock().unwrap()) { @@ -239,8 +261,8 @@ impl Drop for Session { /// The background session data structure #[derive(Debug)] pub struct BackgroundSession { - /// Thread guard of the background session - pub guard: JoinHandle>, + /// Tokio task handle of the background session + pub guard: tokio::task::JoinHandle>, /// Object for creating Notifiers for client use sender: ChannelSender, /// Ensures the filesystem is unmounted when the session ends @@ -249,15 +271,15 @@ pub struct BackgroundSession { impl BackgroundSession { /// Create a new background session for the given session by running its - /// session loop in a background thread. If the returned handle is dropped, + /// session loop as an async task. If the returned handle is dropped, /// the filesystem is unmounted and the given session ends. pub fn new(se: Session) -> io::Result { let sender = se.ch.sender(); // Take the fuse_session, so that we can unmount it let mount = std::mem::take(&mut *se.mount.lock().unwrap()).map(|(_, mount)| mount); - let guard = thread::spawn(move || { - let mut se = se; - se.run() + let session = Arc::new(se); + let guard = tokio::spawn(async move { + session.run().await }); Ok(BackgroundSession { guard, @@ -265,17 +287,18 @@ impl BackgroundSession { _mount: mount, }) } - /// Unmount the filesystem and join the background thread. + + /// Unmount the filesystem and join the background task. /// # Panics - /// Panics if the background thread can't be recovered (e.g., because it panicked). - pub fn join(self) { + /// Panics if the background task can't be recovered (e.g., because it panicked). + pub async fn join(self) { let Self { guard, sender: _, _mount, } = self; drop(_mount); - guard.join().unwrap().unwrap(); + guard.await.unwrap().unwrap(); } /// Returns an object that can be used to send notifications to the kernel diff --git a/cli/src/lib.rs b/cli/src/lib.rs index e1ad8219..980ec4c9 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -15,5 +15,9 @@ pub mod fuser; pub mod nfs; pub fn get_runtime() -> tokio::runtime::Runtime { - tokio::runtime::Runtime::new().expect("Internal error: failed to initialize runtime") + tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .expect("Internal error: failed to initialize runtime") } diff --git a/cli/src/sandbox/linux.rs b/cli/src/sandbox/linux.rs index e524ab86..2227f637 100644 --- a/cli/src/sandbox/linux.rs +++ b/cli/src/sandbox/linux.rs @@ -223,10 +223,9 @@ pub async fn run_cmd( gid: Some(gid), }; - // Start FUSE in a separate thread - let fuse_handle = std::thread::spawn(move || { - let rt = crate::get_runtime(); - crate::fuse::mount(overlay, fuse_opts, rt) + // Start FUSE as an async task + let fuse_handle = tokio::spawn(async move { + crate::fuse::mount(overlay, fuse_opts).await }); // Wait for FUSE mount to be ready @@ -909,7 +908,7 @@ fn run_parent( child_pid: i32, cwd_fd: std::fs::File, fuse_mountpoint: &Path, - _fuse_handle: std::thread::JoinHandle>, + _fuse_handle: tokio::task::JoinHandle>, db_path: &Path, session_id: &str, ) -> ! { diff --git a/sandbox/Cargo.lock b/sandbox/Cargo.lock index c5fc6665..4004c677 100644 --- a/sandbox/Cargo.lock +++ b/sandbox/Cargo.lock @@ -2929,9 +2929,9 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "turso" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1918eba268f456241f21ba50d73e2ddfef0fb34027bde12f3e77d9df051f4ea2" +checksum = "6f5bb728195957fa231f1713a51d20eb6032d0d50ff9f886e220e36a1ac64c28" dependencies = [ "bytes", "http-body-util", @@ -2949,9 +2949,9 @@ dependencies = [ [[package]] name = "turso_core" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cb93aace5ab3ddfc45375ded38107e33b326beb957161456961ac82982c47e18" +checksum = "4b7772de274fbdf70da2c256e7c1bc26029512bdf4e5d6001dd0e5939973e44d" dependencies = [ "aegis", "aes", @@ -3004,9 +3004,9 @@ dependencies = [ [[package]] name = "turso_ext" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "01e1b93164b6438208d7506b12466c44bd8c2de1a584f66c20dfc68286ad9869" +checksum = "b407a4f1369005b0dd9aff499c57eb15bb32391ccd9478633d50513a0c1e8c76" dependencies = [ "chrono", "getrandom 0.3.4", @@ -3015,9 +3015,9 @@ dependencies = [ [[package]] name = "turso_macros" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "302307f3722f23be569036ff53bd0019160aab0c9e1f0ffe45fb3c345d30ee67" +checksum = "bdf248386af09f478d5973011b3853dfc965872332134c8a9f89c8ea46f6304e" dependencies = [ "proc-macro2", "quote", @@ -3026,9 +3026,9 @@ dependencies = [ [[package]] name = "turso_parser" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c70ef6c11754e676549790dc2938221e55696737be09972c5f3b4d6cf2566a1" +checksum = "72611dab70787d2b4d953b94d5a54ab4a57823b75ea2549f9e8abeef066b9fad" dependencies = [ "bitflags 2.10.0", "memchr", @@ -3041,9 +3041,9 @@ dependencies = [ [[package]] name = "turso_sdk_kit" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "28bc2e2689a68dce996f33dd59438475a099b742627452c9c4a337f7a281b5b5" +checksum = "738831ed986f66dbb14dac188b01496f1d683d8943616174df6cd6ac7df1a77c" dependencies = [ "bindgen", "env_logger", @@ -3056,9 +3056,9 @@ dependencies = [ [[package]] name = "turso_sdk_kit_macros" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "663005789da0020a17af745e041ea3009dcd1ce8baa5e3ab1611ca45335a5a0f" +checksum = "681908196c8bb86bec77416e90c1a5a3eb959795d815039a3aa606ef152e89e1" dependencies = [ "proc-macro2", "quote", @@ -3067,9 +3067,9 @@ dependencies = [ [[package]] name = "turso_sync_engine" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "56e73c29342e418fb10faa3e4ebec73fcdf76dfb652be4fe5a795627204c73a3" +checksum = "f1d7412557ec1c87e80305dd2300510a3bc7b298fa2e6848bd33f4779a3fee93" dependencies = [ "base64", "bytes", @@ -3089,9 +3089,9 @@ dependencies = [ [[package]] name = "turso_sync_sdk_kit" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49d93006dae55a8c177b1b832305afcb2432b650372b04f6e2e0e6082b07d883" +checksum = "1c429193511cefff2255739c88c9ab8243a51dad97aca14a8d5a31200e9be3bd" dependencies = [ "bindgen", "env_logger", diff --git a/sdk/rust/Cargo.lock b/sdk/rust/Cargo.lock index df2fd4a0..e966e066 100644 --- a/sdk/rust/Cargo.lock +++ b/sdk/rust/Cargo.lock @@ -2310,9 +2310,9 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "turso" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1918eba268f456241f21ba50d73e2ddfef0fb34027bde12f3e77d9df051f4ea2" +checksum = "6f5bb728195957fa231f1713a51d20eb6032d0d50ff9f886e220e36a1ac64c28" dependencies = [ "bytes", "http-body-util", @@ -2330,9 +2330,9 @@ dependencies = [ [[package]] name = "turso_core" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cb93aace5ab3ddfc45375ded38107e33b326beb957161456961ac82982c47e18" +checksum = "4b7772de274fbdf70da2c256e7c1bc26029512bdf4e5d6001dd0e5939973e44d" dependencies = [ "aegis", "aes", @@ -2385,9 +2385,9 @@ dependencies = [ [[package]] name = "turso_ext" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "01e1b93164b6438208d7506b12466c44bd8c2de1a584f66c20dfc68286ad9869" +checksum = "b407a4f1369005b0dd9aff499c57eb15bb32391ccd9478633d50513a0c1e8c76" dependencies = [ "chrono", "getrandom 0.3.4", @@ -2396,9 +2396,9 @@ dependencies = [ [[package]] name = "turso_macros" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "302307f3722f23be569036ff53bd0019160aab0c9e1f0ffe45fb3c345d30ee67" +checksum = "bdf248386af09f478d5973011b3853dfc965872332134c8a9f89c8ea46f6304e" dependencies = [ "proc-macro2", "quote", @@ -2407,9 +2407,9 @@ dependencies = [ [[package]] name = "turso_parser" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c70ef6c11754e676549790dc2938221e55696737be09972c5f3b4d6cf2566a1" +checksum = "72611dab70787d2b4d953b94d5a54ab4a57823b75ea2549f9e8abeef066b9fad" dependencies = [ "bitflags", "memchr", @@ -2422,9 +2422,9 @@ dependencies = [ [[package]] name = "turso_sdk_kit" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "28bc2e2689a68dce996f33dd59438475a099b742627452c9c4a337f7a281b5b5" +checksum = "738831ed986f66dbb14dac188b01496f1d683d8943616174df6cd6ac7df1a77c" dependencies = [ "bindgen", "env_logger", @@ -2437,9 +2437,9 @@ dependencies = [ [[package]] name = "turso_sdk_kit_macros" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "663005789da0020a17af745e041ea3009dcd1ce8baa5e3ab1611ca45335a5a0f" +checksum = "681908196c8bb86bec77416e90c1a5a3eb959795d815039a3aa606ef152e89e1" dependencies = [ "proc-macro2", "quote", @@ -2448,9 +2448,9 @@ dependencies = [ [[package]] name = "turso_sync_engine" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "56e73c29342e418fb10faa3e4ebec73fcdf76dfb652be4fe5a795627204c73a3" +checksum = "f1d7412557ec1c87e80305dd2300510a3bc7b298fa2e6848bd33f4779a3fee93" dependencies = [ "base64", "bytes", @@ -2470,9 +2470,9 @@ dependencies = [ [[package]] name = "turso_sync_sdk_kit" -version = "0.4.3-pre.2" +version = "0.4.4-pre.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49d93006dae55a8c177b1b832305afcb2432b650372b04f6e2e0e6082b07d883" +checksum = "1c429193511cefff2255739c88c9ab8243a51dad97aca14a8d5a31200e9be3bd" dependencies = [ "bindgen", "env_logger", diff --git a/sdk/rust/Cargo.toml b/sdk/rust/Cargo.toml index aa20983b..12c76b92 100644 --- a/sdk/rust/Cargo.toml +++ b/sdk/rust/Cargo.toml @@ -6,7 +6,7 @@ description = "AgentFS SDK for Rust" license = "MIT" [dependencies] -turso = { version = "0.4.3-pre.2", features = ["sync"] } +turso = { version = "0.4.4-pre.1", features = ["sync"] } tokio = { version = "1", features = ["full"] } async-trait = "0.1" serde = { version = "1.0", features = ["derive"] } diff --git a/sdk/rust/src/connection_pool.rs b/sdk/rust/src/connection_pool.rs index 5e85da82..49d23faa 100644 --- a/sdk/rust/src/connection_pool.rs +++ b/sdk/rust/src/connection_pool.rs @@ -11,11 +11,14 @@ use turso::{Connection, Database}; use crate::error::{Error, Result}; /// Maximum number of connections in the pool. -const MAX_CONNECTIONS: usize = 1; +const MAX_CONNECTIONS: usize = 2; /// Default timeout for acquiring a connection from the pool. const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30); +/// SQLite busy timeout - how long to wait when the database is locked. +const BUSY_TIMEOUT: Duration = Duration::from_secs(30); + /// Database wrapper that supports both regular and sync databases. enum DatabaseType { Local(Database), @@ -94,10 +97,14 @@ impl ConnectionPool { let conn = match conn { Some(c) => c, - None => match &self.inner.db { - DatabaseType::Local(db) => db.connect()?, - DatabaseType::Sync(db) => db.connect().await?, - }, + None => { + let conn = match &self.inner.db { + DatabaseType::Local(db) => db.connect()?, + DatabaseType::Sync(db) => db.connect().await?, + }; + conn.busy_timeout(BUSY_TIMEOUT)?; + conn + } }; Ok(PooledConnection {