diff --git a/Cargo.lock b/Cargo.lock index 89ab7b4478..4a53f9c9f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1647,15 +1647,22 @@ name = "fspy_shared" version = "0.0.0" dependencies = [ "allocator-api2", + "assert2", "base64 0.22.1", "bincode", "bstr", "bytemuck", + "ctor 0.6.0", "derive-where", + "fspy_test_utils", "libc", + "memmap2", "nix 0.30.1", "os_str_bytes", "phf", + "shared_memory", + "tracing", + "uuid", "winapi", "winsafe 0.0.24", ] @@ -1677,6 +1684,15 @@ dependencies = [ "stackalloc", ] +[[package]] +name = "fspy_test_utils" +version = "0.0.0" +dependencies = [ + "base64 0.22.1", + "bincode", + "ctor 0.6.0", +] + [[package]] name = "futures" version = "0.3.31" @@ -2548,6 +2564,15 @@ dependencies = [ "libc", ] +[[package]] +name = "memoffset" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5aa361d4faea93603064a027415f07bd8e1d5c88c9fbf68bf56a285428fd79ce" +dependencies = [ + "autocfg", +] + [[package]] name = "memoffset" version = "0.9.1" @@ -2665,6 +2690,19 @@ version = "1.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086" +[[package]] +name = "nix" +version = "0.23.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f3790c00a0150112de0f4cd161e3d7fc4b2d8a5542ffc35f099a2562aecb35c" +dependencies = [ + "bitflags 1.3.2", + "cc", + "cfg-if", + "libc", + "memoffset 0.6.5", +] + [[package]] name = "nix" version = "0.28.0" @@ -2687,7 +2725,7 @@ dependencies = [ "cfg-if", "cfg_aliases 0.2.1", "libc", - "memoffset", + "memoffset 0.9.1", ] [[package]] @@ -3330,6 +3368,8 @@ version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404" dependencies = [ + "libc", + "rand_chacha 0.3.1", "rand_core 0.6.4", ] @@ -3339,10 +3379,20 @@ version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" dependencies = [ - "rand_chacha", + "rand_chacha 0.9.0", "rand_core 0.9.3", ] +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core 0.6.4", +] + [[package]] name = "rand_chacha" version = "0.9.0" @@ -3358,6 +3408,9 @@ name = "rand_core" version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom 0.2.16", +] [[package]] name = "rand_core" @@ -3922,6 +3975,19 @@ dependencies = [ "libc", ] +[[package]] +name = "shared_memory" +version = "0.12.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba8593196da75d9dc4f69349682bd4c2099f8cde114257d1ef7ef1b33d1aba54" +dependencies = [ + "cfg-if", + "libc", + "nix 0.23.2", + "rand 0.8.5", + "win-sys", +] + [[package]] name = "shell-escape" version = "0.1.5" @@ -5130,6 +5196,15 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dd7cf3379ca1aac9eea11fba24fd7e315d621f8dfe35c8d7d2be8b793726e07d" +[[package]] +name = "win-sys" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b7b128a98c1cfa201b09eb49ba285887deb3cbe7466a98850eb1adabb452be5" +dependencies = [ + "windows", +] + [[package]] name = "winapi" version = "0.3.9" @@ -5161,6 +5236,19 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windows" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "45296b64204227616fdbf2614cefa4c236b98ee64dfaaaa435207ed99fe7829f" +dependencies = [ + "windows_aarch64_msvc 0.34.0", + "windows_i686_gnu 0.34.0", + "windows_i686_msvc 0.34.0", + "windows_x86_64_gnu 0.34.0", + "windows_x86_64_msvc 0.34.0", +] + [[package]] name = "windows-link" version = "0.2.0" @@ -5248,6 +5336,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "86b8d5f90ddd19cb4a147a5fa63ca848db3df085e25fee3cc10b39b6eebae764" +[[package]] +name = "windows_aarch64_msvc" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "17cffbe740121affb56fad0fc0e421804adf0ae00891205213b5cecd30db881d" + [[package]] name = "windows_aarch64_msvc" version = "0.52.6" @@ -5260,6 +5354,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7651a1f62a11b8cbd5e0d42526e55f2c99886c77e007179efff86c2b137e66c" +[[package]] +name = "windows_i686_gnu" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2564fde759adb79129d9b4f54be42b32c89970c18ebf93124ca8870a498688ed" + [[package]] name = "windows_i686_gnu" version = "0.52.6" @@ -5284,6 +5384,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ce6ccbdedbf6d6354471319e781c0dfef054c81fbc7cf83f338a4296c0cae11" +[[package]] +name = "windows_i686_msvc" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9cd9d32ba70453522332c14d38814bceeb747d80b3958676007acadd7e166956" + [[package]] name = "windows_i686_msvc" version = "0.52.6" @@ -5296,6 +5402,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "581fee95406bb13382d2f65cd4a908ca7b1e4c2f1917f143ba16efe98a589b5d" +[[package]] +name = "windows_x86_64_gnu" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfce6deae227ee8d356d19effc141a509cc503dfd1f850622ec4b0f84428e1f4" + [[package]] name = "windows_x86_64_gnu" version = "0.52.6" @@ -5320,6 +5432,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0a6e035dd0599267ce1ee132e51c27dd29437f63325753051e71dd9e42406c57" +[[package]] +name = "windows_x86_64_msvc" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d19538ccc21819d01deaf88d6a17eae6596a12e9aafdbb97916fb49896d89de9" + [[package]] name = "windows_x86_64_msvc" version = "0.52.6" diff --git a/Cargo.toml b/Cargo.toml index c520a1214f..2dd4d3ada2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -73,6 +73,7 @@ fspy_preload_windows = { path = "crates/fspy_preload_windows", artifact = "cdyli fspy_seccomp_unotify = { path = "crates/fspy_seccomp_unotify" } fspy_shared = { path = "crates/fspy_shared" } fspy_shared_unix = { path = "crates/fspy_shared_unix" } +fspy_test_utils = { path = "crates/fspy_test_utils" } futures = "0.3.31" futures-core = "0.3.31" futures-util = "0.3.31" @@ -109,6 +110,7 @@ serde_yml = "0.0.12" serial_test = "3.2.0" sha1 = "0.10.6" sha2 = "0.10.9" +shared_memory = "0.12.4" shell-escape = "0.1.5" slab = "0.4.9" smallvec = { version = "2.0.0-alpha.11", features = ["std"] } diff --git a/bench/Cargo.toml b/bench/Cargo.toml index 9925c6e8f6..21d3a24f2d 100644 --- a/bench/Cargo.toml +++ b/bench/Cargo.toml @@ -4,8 +4,8 @@ version = "0.1.0" edition = "2024" [dependencies] -vite_task = { workspace = true } vite_path = { workspace = true } +vite_task = { workspace = true } [dev-dependencies] criterion = { workspace = true } diff --git a/crates/fspy/Cargo.toml b/crates/fspy/Cargo.toml index e4efb36cd5..0f36b8b4c5 100644 --- a/crates/fspy/Cargo.toml +++ b/crates/fspy/Cargo.toml @@ -9,6 +9,7 @@ allocator-api2 = { workspace = true, features = ["alloc"] } bincode = { workspace = true } bstr = { workspace = true, default-features = false } bumpalo = { workspace = true } +const_format = { workspace = true, features = ["fmt"] } fspy_shared = { workspace = true } futures-util = { workspace = true } libc = { workspace = true } @@ -18,6 +19,7 @@ slab = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["net", "process", "io-util", "sync"] } which = { workspace = true } +xxhash-rust = { workspace = true } [target.'cfg(target_os = "linux")'.dependencies] arrayvec = { workspace = true } @@ -39,10 +41,6 @@ passfd = { git = "https://github.com/polachok/passfd", features = ["async"] } [target.'cfg(target_os = "macos")'.dependencies] phf = { workspace = true } -[target.'cfg(any(target_os = "macos", windows))'.dependencies] -const_format = { workspace = true, features = ["fmt"] } -xxhash-rust = { workspace = true } - [target.'cfg(target_os = "windows")'.dependencies] fspy_detours_sys = { workspace = true } fspy_preload_windows = { workspace = true } diff --git a/crates/fspy/examples/cli.rs b/crates/fspy/examples/cli.rs index 03582e4e74..85c0d1144f 100644 --- a/crates/fspy/examples/cli.rs +++ b/crates/fspy/examples/cli.rs @@ -23,6 +23,8 @@ async fn main() -> io::Result<()> { let TrackedChild { mut tokio_child, accesses_future } = command.spawn().await?; + let output = tokio_child.wait().await?; + let accesses = accesses_future.await?; let mut path_count = 0usize; @@ -47,7 +49,6 @@ async fn main() -> io::Result<()> { } csv_writer.flush().await?; - let output = tokio_child.wait().await?; eprintln!("\nfspy: {path_count} paths accessed. {output}"); Ok(()) } diff --git a/crates/fspy/src/fixture.rs b/crates/fspy/src/fixture.rs index 08e78bc779..17685fcbbf 100644 --- a/crates/fspy/src/fixture.rs +++ b/crates/fspy/src/fixture.rs @@ -26,6 +26,7 @@ macro_rules! fixture { pub use fixture; impl Fixture { + #[cfg(not(target_os = "linux"))] pub const fn new(name: &'static str, content: &'static [u8], hash: &'static str) -> Self { Self { name, content, hash } } diff --git a/crates/fspy/src/lib.rs b/crates/fspy/src/lib.rs index 7763c7d404..d4674995eb 100644 --- a/crates/fspy/src/lib.rs +++ b/crates/fspy/src/lib.rs @@ -1,9 +1,7 @@ #![cfg_attr(target_os = "windows", feature(windows_process_extensions_main_thread_handle))] #![feature(once_cell_try)] -// Windows and macOS both need to persist the injected DLL/shared library somewhere in the filesystem. -// Linux doesn't need this. Instead we use `memfd_create` to create an in-memory shared library. -#[cfg(not(target_os = "linux"))] +// Persist the injected DLL/shared library somewhere in the filesystem. mod fixture; #[cfg(unix)] @@ -17,9 +15,7 @@ mod os_impl; mod arena; mod command; -#[cfg(not(target_os = "linux"))] -use std::{env::temp_dir, fs::create_dir}; -use std::{ffi::OsStr, io, sync::OnceLock}; +use std::{env::temp_dir, ffi::OsStr, fs::create_dir, io, sync::OnceLock}; pub use command::Command; pub use fspy_shared::ipc::{AccessMode, PathAccess}; @@ -30,23 +26,19 @@ use tokio::process::Child; pub struct TrackedChild { pub tokio_child: Child, - pub accesses_future: BoxFuture<'static, io::Result>, + /// This future lazily locks the IPC channel when it's polled. + /// Do not `await` it until the child process has exited. + pub accesses_future: BoxFuture<'static, io::Result>, } pub struct Spy(SpyInner); impl Spy { - #[cfg(not(target_os = "linux"))] pub fn new() -> io::Result { let tmp_dir = temp_dir().join("fspy"); let _ = create_dir(&tmp_dir); Ok(Self(SpyInner::init_in(&tmp_dir)?)) } - #[cfg(target_os = "linux")] - pub fn new() -> io::Result { - Ok(Self(SpyInner::init()?)) - } - pub fn global() -> io::Result<&'static Self> { static GLOBAL_SPY: OnceLock = OnceLock::new(); GLOBAL_SPY.get_or_try_init(Self::new) diff --git a/crates/fspy/src/unix/mod.rs b/crates/fspy/src/unix/mod.rs index edda017011..8bf5101403 100644 --- a/crates/fspy/src/unix/mod.rs +++ b/crates/fspy/src/unix/mod.rs @@ -8,23 +8,15 @@ mod syscall_handler; #[cfg(target_os = "macos")] mod macos_fixtures; -#[cfg(target_os = "macos")] -use std::path::Path; -#[cfg(target_os = "linux")] -use std::{fs::File, io::Write, sync::Arc}; -use std::{ - io::{self}, - iter, - os::fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd, OwnedFd, RawFd}, - sync::atomic::{AtomicU8, Ordering, fence}, -}; +use std::{io, path::Path}; use bincode::borrow_decode_from_slice; #[cfg(target_os = "linux")] use fspy_seccomp_unotify::supervisor::supervise; -#[cfg(target_os = "macos")] -use fspy_shared::ipc::NativeString; -use fspy_shared::ipc::{BINCODE_CONFIG, PathAccess}; +use fspy_shared::ipc::{ + BINCODE_CONFIG, NativeString, PathAccess, + channel::{Receiver, ReceiverLockGuard, channel}, +}; #[cfg(target_os = "macos")] use fspy_shared_unix::payload::Fixtures; use fspy_shared_unix::{ @@ -32,94 +24,66 @@ use fspy_shared_unix::{ payload::{Payload, encode_payload}, spawn::handle_exec, }; -use futures_util::{FutureExt, future::try_join}; -use memmap2::Mmap; -use nix::fcntl::{FcntlArg, FdFlag, fcntl}; -#[cfg(target_os = "linux")] -use nix::sys::memfd::{MFdFlags, memfd_create}; -use passfd::tokio::FdPassingExt; +use futures_util::FutureExt; #[cfg(target_os = "linux")] use syscall_handler::SyscallHandler; -use tokio::net::UnixStream; +use tokio::task::spawn_blocking; use crate::{Command, TrackedChild, arena::PathAccessArena}; #[derive(Debug, Clone)] pub struct SpyInner { - #[cfg(target_os = "linux")] - preload_lib_memfd: Arc, - #[cfg(target_os = "macos")] fixtures: Fixtures, - #[cfg(target_os = "macos")] preload_path: NativeString, } const PRELOAD_CDYLIB_BINARY: &[u8] = include_bytes!(env!("CARGO_CDYLIB_FILE_FSPY_PRELOAD_UNIX")); impl SpyInner { - #[cfg(target_os = "linux")] - pub fn init() -> io::Result { - let preload_lib_memfd = memfd_create("fspy_preload", MFdFlags::MFD_CLOEXEC)?; - let mut execve_host_memfile = File::from(preload_lib_memfd); - execve_host_memfile.write_all(PRELOAD_CDYLIB_BINARY)?; - - let preload_lib_memfd = duplicate_until_safe(OwnedFd::from(execve_host_memfile))?; + /// Initialize the fs access spy by writing the preload library on disk + pub fn init_in(dir: &Path) -> io::Result { + use const_format::formatcp; + use xxhash_rust::const_xxh3::xxh3_128; - Ok(Self { preload_lib_memfd: Arc::new(preload_lib_memfd) }) - } + use crate::fixture::Fixture; - #[cfg(target_os = "macos")] - pub fn init_in(dir: &Path) -> io::Result { const PRELOAD_CDYLIB: Fixture = Fixture { name: "fspy_preload", content: PRELOAD_CDYLIB_BINARY, hash: formatcp!("{:x}", xxh3_128(PRELOAD_CDYLIB_BINARY)), }; - use const_format::formatcp; - use xxhash_rust::const_xxh3::xxh3_128; - - use crate::fixture::Fixture; - let coreutils_path = macos_fixtures::COREUTILS_BINARY.write_to(dir, "")?; - let bash_path = macos_fixtures::OILS_BINARY.write_to(dir, "")?; - let preload_cdylib_path = PRELOAD_CDYLIB.write_to(dir, ".dylib")?; - let fixtures = Fixtures { - bash_path: bash_path.as_path().into(), //Path::new("/opt/homebrew/bin/bash"),//brush.as_path(), - coreutils_path: coreutils_path.as_path().into(), - }; - Ok(Self { fixtures, preload_path: preload_cdylib_path.as_path().into() }) + Ok(Self { + preload_path: preload_cdylib_path.as_path().into(), + #[cfg(target_os = "macos")] + fixtures: { + let coreutils_path = macos_fixtures::COREUTILS_BINARY.write_to(dir, "")?; + let bash_path = macos_fixtures::OILS_BINARY.write_to(dir, "")?; + Fixtures { + bash_path: bash_path.as_path().into(), + coreutils_path: coreutils_path.as_path().into(), + } + }, + }) } } -fn unset_fd_flag(fd: BorrowedFd<'_>, flag_to_remove: FdFlag) -> io::Result<()> { - fcntl( - fd, - FcntlArg::F_SETFD({ - let mut fd_flag = FdFlag::from_bits_retain(fcntl(fd, FcntlArg::F_GETFD)?); - fd_flag.remove(flag_to_remove); - fd_flag - }), - )?; - Ok(()) +#[ouroboros::self_referencing] +struct OwnedReceiverLockGuard { + /// Owns the shared memory + receiver: Receiver, + /// Borrows the shared memory and owns the file lock + #[borrows(receiver)] + #[covariant] + lock_guard: ReceiverLockGuard<'this>, } -// fn unset_fl_flag(fd: BorrowedFd<'_>, flag_to_remove: OFlag) -> io::Result<()> { -// fcntl( -// fd, -// FcntlArg::F_SETFL({ -// let mut fd_flag = OFlag::from_bits_retain(fcntl(fd, FcntlArg::F_GETFL)?); -// fd_flag.remove(flag_to_remove); -// fd_flag -// }), -// )?; -// Ok(()) -// } pub struct PathAccessIterable { arenas: Vec, - shm_mmaps: Vec, + ipc_receiver_lock_guard: OwnedReceiverLockGuard, } impl PathAccessIterable { @@ -127,79 +91,45 @@ impl PathAccessIterable { let accesses_in_arena = self.arenas.iter().flat_map(|arena| arena.borrow_accesses().iter()).copied(); - let accesses_in_shm = self.shm_mmaps.iter().flat_map(|mmap| { - let buf = &**mmap; - let mut position = 0usize; - iter::from_fn(move || { - let (flag_buf, data_buf) = buf[position..].split_first()?; - let atomic_flag = - unsafe { AtomicU8::from_ptr(std::ptr::from_ref::(flag_buf).cast_mut()) }; - let flag = atomic_flag.load(Ordering::Acquire); - if flag == 0 { - return None; - } - fence(Ordering::Acquire); + let accesses_in_shm = + self.ipc_receiver_lock_guard.borrow_lock_guard().iter_frames().map(|frame| { let (path_access, decoded_size) = - borrow_decode_from_slice::, _>(data_buf, BINCODE_CONFIG) - .unwrap(); - - position += decoded_size + 1; - - Some(path_access) - }) - }); + borrow_decode_from_slice::, _>(frame, BINCODE_CONFIG).unwrap(); + assert_eq!(decoded_size, frame.len()); + path_access + }); accesses_in_shm.chain(accesses_in_arena) } } -// https://github.com/nodejs/node/blob/5794e644b724c6c6cac02d306d87a4d6b78251e5/deps/uv/src/unix/core.c#L803-L808 -fn duplicate_until_safe(mut fd: OwnedFd) -> io::Result { - const SAFE_FD_NUM: RawFd = 17; - - let mut fds: Vec = vec![]; - while fd.as_raw_fd() < SAFE_FD_NUM { - let new_fd = fd.try_clone()?; - fds.push(fd); - fd = new_fd; - } - Ok(fd) -} - -pub async fn spawn_impl(mut command: Command) -> io::Result { - let (shm_fd_sender, shm_fd_receiver) = UnixStream::pair()?; - - let shm_fd_sender = shm_fd_sender.into_std()?; - shm_fd_sender.set_nonblocking(false)?; - let shm_fd_sender = duplicate_until_safe(OwnedFd::from(shm_fd_sender))?; +// Shared memory size for storing path accesses. +// 4 GiB is large enough to store path accesses in almost any realistic scenario. +// This doesn't allocate physical memory until it's actually used. +const SHM_CAPACITY: usize = 4 * 1024 * 1024 * 1024; +pub(crate) async fn spawn_impl(mut command: Command) -> io::Result { #[cfg(target_os = "linux")] let supervisor = supervise::()?; #[cfg(target_os = "linux")] let supervisor_pre_exec = supervisor.pre_exec; + let (ipc_channel_conf, ipc_receiver) = channel(SHM_CAPACITY)?; + let payload = Payload { - ipc_fd: shm_fd_sender.as_raw_fd(), + ipc_channel_conf, #[cfg(target_os = "macos")] fixtures: command.spy_inner.fixtures.clone(), - #[cfg(target_os = "macos")] preload_path: command.spy_inner.preload_path.clone(), - #[cfg(target_os = "linux")] - preload_path: format!("/proc/self/fd/{}", command.spy_inner.preload_lib_memfd.as_raw_fd()) - .into(), - #[cfg(target_os = "linux")] seccomp_payload: supervisor.payload, }; let encoded_payload = encode_payload(payload); - #[cfg(target_os = "linux")] - let preload_lib_memfd = Arc::clone(&command.spy_inner.preload_lib_memfd); - let mut exec = command.get_exec(); let mut exec_resolve_accesses = PathAccessArena::default(); let pre_exec = handle_exec( @@ -216,10 +146,6 @@ pub async fn spawn_impl(mut command: Command) -> io::Result { unsafe { tokio_command.pre_exec(move || { - #[cfg(target_os = "linux")] - unset_fd_flag(preload_lib_memfd.as_fd(), FdFlag::FD_CLOEXEC)?; - unset_fd_flag(shm_fd_sender.as_fd(), FdFlag::FD_CLOEXEC)?; - #[cfg(target_os = "linux")] supervisor_pre_exec.run()?; if let Some(pre_exec) = pre_exec.as_ref() { @@ -230,11 +156,9 @@ pub async fn spawn_impl(mut command: Command) -> io::Result { } let child = tokio_command.spawn()?; - // drop channel_sender in the parent process, - // so that channel_receiver reaches eof as soon as the last descendant process exits. + drop(tokio_command); - // #[cfg(target_os = "linux")] let arenas_future = async move { let arenas = std::iter::once(exec_resolve_accesses); #[cfg(target_os = "linux")] @@ -243,30 +167,14 @@ pub async fn spawn_impl(mut command: Command) -> io::Result { io::Result::Ok(arenas.collect::>()) }; - let shm_future = async move { - let mut shm_fds = Vec::::new(); - loop { - let shm_fd = match shm_fd_receiver.recv_fd().await { - Ok(fd) => unsafe { OwnedFd::from_raw_fd(fd) }, - Err(err) => { - if err.kind() == io::ErrorKind::UnexpectedEof { - break; - } - return Err(err); - } - }; - shm_fds.push(shm_fd); - } - io::Result::Ok(shm_fds) - }; - let accesses_future = async move { - let (arenas, shm_fds) = try_join(arenas_future, shm_future).await?; - let shm_mmaps = shm_fds - .into_iter() - .map(|fd| unsafe { Mmap::map(&fd) }) - .collect::>>()?; - Ok(PathAccessIterable { arenas, shm_mmaps }) + let arenas = arenas_future.await?; + // `receiver.lock()` blocks. Run it inside `spawn_blocking` to avoid blocking the tokio runtime. + let ipc_receiver_lock_guard = spawn_blocking(move || { + OwnedReceiverLockGuard::try_new(ipc_receiver, |receiver| receiver.lock()) + }) + .await??; + Ok(PathAccessIterable { arenas, ipc_receiver_lock_guard }) } .boxed(); diff --git a/crates/fspy/src/windows/mod.rs b/crates/fspy/src/windows/mod.rs index 0add51f3cd..7a5979623f 100644 --- a/crates/fspy/src/windows/mod.rs +++ b/crates/fspy/src/windows/mod.rs @@ -126,7 +126,13 @@ pub(crate) async fn spawn_impl(command: Command) -> io::Result { connect_fut.await?; - let accesses_future = async move { + // Temporary workaround before switching to shared memory IPC on Windows. + // The shared memory IPC requires `accesses_future` to be polled after the child process has exited. + // The test code has updated but the Windows implementation here has not yet adopted that model. + // As a result, accesses_future didn't get polled making the pipe_sender to block indefinitely. + // To unblock the pipe_sender, we spawn a separate task to keep reading pipe_receiver. + // This workaround can be removed once the shared memory IPC is used on Windows. + let accesses_future = tokio::task::spawn(async move { let mut arena = PathAccessArena::default(); let mut buf = [0u8; MESSAGE_MAX_LEN]; @@ -142,8 +148,8 @@ pub(crate) async fn spawn_impl(command: Command) -> io::Result { arena.add(path_access); } io::Result::Ok(PathAccessIterable { arena }) - } - .boxed(); + }); + let accesses_future = async move { accesses_future.await.unwrap() }.boxed(); // let path_access_stream = PathAccessIterable { pipe_receiver }; diff --git a/crates/fspy/tests/node_fs.rs b/crates/fspy/tests/node_fs.rs index 8a8b57e05f..ceaf0e8e78 100644 --- a/crates/fspy/tests/node_fs.rs +++ b/crates/fspy/tests/node_fs.rs @@ -15,9 +15,8 @@ async fn track_node_script(script: &str) -> io::Result { .envs(vars_os()) // https://github.com/jdx/mise/discussions/5968 .arg(script); let TrackedChild { mut tokio_child, accesses_future } = command.spawn().await?; - - let accesses = accesses_future.await?; let status = tokio_child.wait().await?; + let accesses = accesses_future.await?; assert!(status.success()); Ok(accesses) } diff --git a/crates/fspy/tests/test_utils.rs b/crates/fspy/tests/test_utils.rs index a69678ffb0..8d83934a40 100644 --- a/crates/fspy/tests/test_utils.rs +++ b/crates/fspy/tests/test_utils.rs @@ -55,8 +55,8 @@ pub async fn _spawn_with_id(id: &str) -> io::Result { command.arg(id); let TrackedChild { mut tokio_child, accesses_future } = command.spawn().await?; - let accesses = accesses_future.await?; let status = tokio_child.wait().await?; + let accesses = accesses_future.await?; assert!(status.success()); Ok(accesses) } diff --git a/crates/fspy_e2e/src/main.rs b/crates/fspy_e2e/src/main.rs index 9cb065e0f9..0d702fc36e 100644 --- a/crates/fspy_e2e/src/main.rs +++ b/crates/fspy_e2e/src/main.rs @@ -8,7 +8,6 @@ use std::{ }; use fspy::{AccessMode, PathAccess}; -use futures_util::future::try_join; use serde::{Deserialize, Serialize}; #[derive(Serialize, Deserialize)] @@ -89,10 +88,9 @@ async fn main() { let tracked_child = cmd.spawn().await.unwrap(); - let (accesses, output) = - try_join(tracked_child.accesses_future, tracked_child.tokio_child.wait_with_output()) - .await - .unwrap(); + let output = tracked_child.tokio_child.wait_with_output().await.unwrap(); + let accesses = tracked_child.accesses_future.await.unwrap(); + if !output.status.success() { eprintln!("----- stdout begin -----"); stderr().write_all(&output.stdout).unwrap(); diff --git a/crates/fspy_preload_unix/src/client/mod.rs b/crates/fspy_preload_unix/src/client/mod.rs index 8703bd7143..cc7104189f 100644 --- a/crates/fspy_preload_unix/src/client/mod.rs +++ b/crates/fspy_preload_unix/src/client/mod.rs @@ -1,66 +1,21 @@ pub mod convert; pub mod raw_exec; -use std::{ - cell::RefCell, - fmt::Debug, - io, - os::fd::AsRawFd, - sync::{ - OnceLock, - atomic::{AtomicU8, AtomicUsize, Ordering, fence}, - }, -}; +use std::{fmt::Debug, num::NonZeroUsize, sync::OnceLock}; -use anyhow::Context; use bincode::{enc::write::SizeWriter, encode_into_slice, encode_into_writer}; use convert::{ToAbsolutePath, ToAccessMode}; -use fspy_shared::ipc::{BINCODE_CONFIG, PathAccess}; +use fspy_shared::ipc::{BINCODE_CONFIG, PathAccess, channel::Sender}; use fspy_shared_unix::{ exec::ExecResolveConfig, payload::EncodedPayload, spawn::{PreExec, handle_exec}, }; -use libc::off_t; -use memmap2::MmapMut; -use nix::{ - fcntl::OFlag, - sys::{ - mman::{shm_open, shm_unlink}, - stat::Mode, - }, - unistd::{ftruncate, getpid}, -}; -use passfd::FdPassingExt; use raw_exec::RawExec; -#[derive(Debug)] -struct ShmCursor { - mmap_mut: MmapMut, - position: usize, -} -impl ShmCursor { - pub fn advance(&mut self, len: usize) -> Option<&mut [u8]> { - let new_position = self.position.checked_add(len)?; - if new_position > self.mmap_mut.len() { - return None; - } - let buf = &mut self.mmap_mut[self.position..new_position]; - self.position = new_position; - Some(buf) - } -} - -thread_local! { - static SHM_CURSOR: RefCell> = const { RefCell::new(None) }; -} - pub struct Client { encoded_payload: EncodedPayload, - shm_id: AtomicUsize, - - #[cfg(target_os = "macos")] - posix_spawn_file_actions: OnceLock, + ipc_sender: Option, } #[cfg(target_os = "macos")] @@ -74,72 +29,32 @@ impl Debug for Client { } } -const SHM_CHUNK_SIZE: off_t = 256 * 1024; - impl Client { #[cfg(not(test))] fn from_env() -> Self { use fspy_shared_unix::payload::decode_payload_from_env; let encoded_payload = decode_payload_from_env().unwrap(); - Self { - shm_id: AtomicUsize::new(0), - encoded_payload, - #[cfg(target_os = "macos")] - posix_spawn_file_actions: OnceLock::new(), - } - } - - fn new_shm(&self) -> io::Result { - let shm_name = format!( - "/fspy_shm_{}_{}", - getpid().as_raw(), - self.shm_id.fetch_add(1, Ordering::Relaxed), - ); - let shm_fd = shm_open( - shm_name.as_str(), - OFlag::O_CLOEXEC | OFlag::O_RDWR | OFlag::O_CREAT | OFlag::O_EXCL, - Mode::empty(), - )?; - shm_unlink(shm_name.as_str())?; - self.encoded_payload.payload.ipc_fd.send_fd(shm_fd.as_raw_fd())?; - ftruncate(&shm_fd, SHM_CHUNK_SIZE)?; - let mmap_mut = unsafe { MmapMut::map_mut(&shm_fd) }?; - Ok(ShmCursor { mmap_mut, position: 0 }) - } - - fn with_shm_buf( - &self, - len: usize, - f: impl FnOnce(&mut [u8]) -> anyhow::Result<()>, - ) -> anyhow::Result<()> { - let result = SHM_CURSOR.try_with(|shm_cursor| { - let mut shm_cursor = shm_cursor.borrow_mut(); - let shm_buf = if let Some(some) = shm_cursor.as_mut() { - some - } else { - shm_cursor.insert(self.new_shm()?) - }; - if let Some(buf) = shm_buf.advance(len) { - f(buf) - } else { - *shm_buf = self.new_shm()?; - let buf = shm_buf.advance(len).with_context(|| { - format!( - "The requested buf ({len}) is greater than the shm chunk size ({SHM_CHUNK_SIZE})" - ) - })?; - f(buf) + let ipc_sender = match encoded_payload.payload.ipc_channel_conf.sender() { + Ok(sender) => Some(sender), + Err(err) => { + // this can happen if the process is started after the root target process has exited. + // By that time the channel would have been closed in the receiver side. + // In this case we just leave a message and skip sending any path accesses. + eprintln!("fspy: failed to create ipc sender: {}", err); + None } - }); - match result { - Ok(ok) => ok, - Err(_) => Ok(()), // Ignore AccessError. TODO(fix): handle AccessError - } + }; + + Self { ipc_sender, encoded_payload } } fn send(&self, path_access: PathAccess<'_>) -> anyhow::Result<()> { + let Some(ipc_sender) = &self.ipc_sender else { + // ipc channel not available, skip sending + return Ok(()); + }; let path = path_access.path.as_bstr(); if path.starts_with(b"/dev/") || (cfg!(target_os = "linux") @@ -150,16 +65,14 @@ impl Client { let mut size_writer = SizeWriter::default(); encode_into_writer(path_access, &mut size_writer, BINCODE_CONFIG)?; - self.with_shm_buf(1 + size_writer.bytes_written, |buf| { - let data_buf = &mut buf[1..]; - let written_size = encode_into_slice(path_access, data_buf, BINCODE_CONFIG)?; - debug_assert_eq!(written_size, size_writer.bytes_written); + let frame_size = NonZeroUsize::new(size_writer.bytes_written) + .expect("fspy: encoded PathAccess should never be empty"); - let flag_ptr = buf.as_mut_ptr().cast::(); - fence(Ordering::Release); - unsafe { AtomicU8::from_ptr(flag_ptr) }.store(1, Ordering::Release); - Ok(()) - })?; + let mut frame = ipc_sender + .claim_frame(frame_size) + .expect("fspy: failed to claim frame in shared memory"); + let written_size = encode_into_slice(&path_access, &mut frame, BINCODE_CONFIG)?; + assert_eq!(written_size, size_writer.bytes_written); Ok(()) } @@ -194,79 +107,6 @@ impl Client { Ok(()) } - - #[cfg(not(target_os = "macos"))] - pub unsafe fn handle_posix_spawn_opts( - &self, - _file_actions: &mut *const libc::posix_spawn_file_actions_t, - _attrp: *const libc::posix_spawnattr_t, - ) -> nix::Result<()> { - Ok(()) - } - - #[cfg(target_os = "macos")] - pub unsafe fn handle_posix_spawn_opts( - &self, - file_actions: &mut *const libc::posix_spawn_file_actions_t, - attrp: *const libc::posix_spawnattr_t, - ) -> nix::Result<()> { - unsafe extern "C" { - unsafe fn posix_spawn_file_actions_addinherit_np( - actions: *mut libc::posix_spawn_file_actions_t, - fd: libc::c_int, - ) -> libc::c_int; - } - - use core::mem::zeroed; - - use libc::c_short; - let cloexec_default = if attrp.is_null() { - false - } else { - let mut flags = 0; - let ret = unsafe { libc::posix_spawnattr_getflags(attrp, &raw mut flags) }; - if ret != 0 { - return Err(nix::Error::from_raw(ret)); - } - (flags & (libc::POSIX_SPAWN_CLOEXEC_DEFAULT as c_short)) != 0 - }; - - if !cloexec_default { - return Ok(()); - } - - // ensure ipc fd is inherited when POSIX_SPAWN_CLOEXEC_DEFAULT is set. - if (*file_actions).is_null() { - let shared_file_actions = self.posix_spawn_file_actions.get_or_init(|| { - let mut fa: libc::posix_spawn_file_actions_t = unsafe { zeroed() }; - let ret = unsafe { libc::posix_spawn_file_actions_init(&raw mut fa) }; - assert_eq!(ret, 0); - let ret = unsafe { - posix_spawn_file_actions_addinherit_np( - &raw mut fa, - self.encoded_payload.payload.ipc_fd, - ) - }; - assert_eq!(ret, 0); - fa - }); - *file_actions = shared_file_actions; - } else { - // this makes `file_actions` list grow indefinitely if it keeps being reused to spawn processes, - // but I can't think of a better way. (no way to inspect or clone `file_actions`) - let ret = unsafe { - posix_spawn_file_actions_addinherit_np( - (*file_actions).cast_mut(), - self.encoded_payload.payload.ipc_fd, - ) - }; - if ret != 0 { - return Err(nix::Error::from_raw(ret)); - } - } - Ok(()) - // posix_spawn_file_actions_addclose(actions, fd, path, oflag, mode) - } } static CLIENT: OnceLock = OnceLock::new(); @@ -284,23 +124,5 @@ pub unsafe fn handle_open(path: impl ToAbsolutePath, mode: impl ToAccessMode) { #[cfg(not(test))] #[ctor::ctor] fn init_client() { - unsafe extern "C" fn reset_shm_atfork() { - let Some(client) = global_client() else { - return; - }; - let _ = SHM_CURSOR.try_with(|shm_cursor| { - // Move the shm cursor to the end so that the next time it's used a new one will be created. - let mut shm_cursor = shm_cursor.borrow_mut(); - let Some(shm_cursor) = shm_cursor.as_mut() else { - return; - }; - shm_cursor.position = shm_cursor.mmap_mut.len(); - }); - } - - use libc::pthread_atfork; - CLIENT.set(Client::from_env()).unwrap(); - let ret = unsafe { pthread_atfork(None, None, Some(reset_shm_atfork)) }; - assert!((ret == 0), "pthread_atfork failed: {ret}"); } diff --git a/crates/fspy_preload_unix/src/interceptions/spawn/posix_spawn.rs b/crates/fspy_preload_unix/src/interceptions/spawn/posix_spawn.rs index b38738c76a..3dd6f0ad1e 100644 --- a/crates/fspy_preload_unix/src/interceptions/spawn/posix_spawn.rs +++ b/crates/fspy_preload_unix/src/interceptions/spawn/posix_spawn.rs @@ -22,7 +22,7 @@ unsafe fn handle_posix_spawn( original: PosixSpawnFn, pid: *mut libc::pid_t, file: *const c_char, - mut file_actions: *const libc::posix_spawn_file_actions_t, + file_actions: *const libc::posix_spawn_file_actions_t, attrp: *const libc::posix_spawnattr_t, argv: *const *mut c_char, envp: *const *mut c_char, @@ -33,9 +33,6 @@ unsafe fn handle_posix_spawn( let client = global_client() .expect("posix_spawn(p) unexpectedly called before client initialized in ctor"); - if let Err(errno) = unsafe { client.handle_posix_spawn_opts(&mut file_actions, attrp) } { - return errno as _; - } let result = unsafe { client.handle_exec::( config, diff --git a/crates/fspy_preload_windows/src/windows/detours/create_process.rs b/crates/fspy_preload_windows/src/windows/detours/create_process.rs index 505b8a3d13..eceb543c6c 100644 --- a/crates/fspy_preload_windows/src/windows/detours/create_process.rs +++ b/crates/fspy_preload_windows/src/windows/detours/create_process.rs @@ -198,7 +198,7 @@ static DETOUR_CREATE_PROCESS_A: Detour< unsafe { sender.send(PathAccess { mode: AccessMode::Read, - path: NativeStr::from_bytes(CStr::from_ptr(lp_application_name).to_bytes()), + path: NativeStr::from_ansi(CStr::from_ptr(lp_application_name).to_bytes()), }); } } diff --git a/crates/fspy_shared/Cargo.toml b/crates/fspy_shared/Cargo.toml index 1c19b84da6..916d6b0af5 100644 --- a/crates/fspy_shared/Cargo.toml +++ b/crates/fspy_shared/Cargo.toml @@ -8,6 +8,11 @@ publish = false allocator-api2 = { workspace = true } bincode = { workspace = true } bstr = { workspace = true } +bytemuck = { workspace = true, features = ["must_cast"] } +memmap2 = { workspace = true } +shared_memory = { workspace = true } +tracing = { workspace = true } +uuid = { workspace = true, features = ["v4"] } [target.'cfg(target_os = "windows")'.dependencies] bytemuck = { workspace = true } @@ -22,8 +27,15 @@ phf = { workspace = true } base64 = { workspace = true } derive-where = { workspace = true } libc = { workspace = true } -nix = { workspace = true } +nix = { workspace = true, features = ["mman"] } # [features] # supervisor = ["dep:tokio", "dep:passfd"] # target = [] + +[dev-dependencies] +assert2 = { workspace = true } +base64 = { workspace = true } +ctor = { workspace = true } +fspy_test_utils = { workspace = true } +shared_memory = { workspace = true } diff --git a/crates/fspy_shared/src/ipc/channel/mod.rs b/crates/fspy_shared/src/ipc/channel/mod.rs new file mode 100644 index 0000000000..3b0b922ca0 --- /dev/null +++ b/crates/fspy_shared/src/ipc/channel/mod.rs @@ -0,0 +1,229 @@ +//! Fast mpsc IPC channel implementation based on shared memory. + +mod shm_io; + +use std::{env::temp_dir, fs::File, io, ops::Deref, path::PathBuf, sync::Arc}; + +use bincode::{Decode, Encode}; +use shared_memory::{Shmem, ShmemConf}; +pub use shm_io::FrameMut; +use shm_io::{ShmReader, ShmWriter}; +use tracing::debug; +use uuid::Uuid; + +use super::NativeString; + +/// Serializable configuration to create channel senders. +#[derive(Encode, Decode, Clone, Debug)] +pub struct ChannelConf { + lock_file_path: NativeString, + shm_id: Arc, + shm_size: usize, +} + +/// Creates a mpsc IPC channel with one receiver and a `ChannelConf` that can be passed around processes and used to create multiple senders +pub fn channel(capacity: usize) -> io::Result<(ChannelConf, Receiver)> { + // Initialize the lock file with a unique name. + let lock_file_path = temp_dir().join(format!("fspy_ipc_{}.lock", Uuid::new_v4())); + + // Initialize the shared memory with a unique id. + let shm = ShmemConf::new() + .size(capacity) + .create() + .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?; + + let conf = ChannelConf { + lock_file_path: lock_file_path.as_os_str().into(), + shm_id: shm.get_os_id().into(), + shm_size: capacity, + }; + + let receiver = Receiver::new(lock_file_path, shm)?; + Ok((conf, receiver)) +} + +impl ChannelConf { + /// Creates a sender. + /// + /// This doesn't block on the file lock. Instead it returns immediately with error if the receiver is locked or dropped. + pub fn sender(&self) -> io::Result { + let lock_file = File::open(self.lock_file_path.to_cow_os_str())?; + lock_file.try_lock_shared()?; + let shm = ShmemConf::new() + .size(self.shm_size) + .os_id(&self.shm_id) + .open() + .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?; + let writer = unsafe { ShmWriter::new(shm) }; + Ok(Sender { writer, lock_file, lock_file_path: self.lock_file_path.clone() }) + } +} + +pub struct Sender { + writer: ShmWriter, + lock_file_path: NativeString, + lock_file: File, +} + +impl Drop for Sender { + fn drop(&mut self) { + if let Err(err) = self.lock_file.unlock() { + debug!("Failed to unlock the shared IPC lock {:?}: {}", self.lock_file_path, err); + } + } +} + +impl Deref for Sender { + type Target = ShmWriter; + + fn deref(&self) -> &Self::Target { + &self.writer + } +} + +/// Safety: `Sender` holds a shared file lock that ensures there's no reader, so `shm` can be safely written to. +unsafe impl Send for Sender {} + +/// Safety: `Sender` holds a shared file lock that ensures there's no reader, so `shm` can be safely written to. +unsafe impl Sync for Sender {} + +/// The unique receiver side of an IPC channel. +/// Owns the lock file and removes it on drop. +pub struct Receiver { + lock_file_path: PathBuf, + lock_file: File, + shm: Shmem, +} + +/// Safety: Receiver doesn't read or write `shm`. It only pass it to ReceiverLockGuard under the lock. +unsafe impl Send for Receiver {} + +/// Safety: Receiver doesn't read or write `shm`. It only pass it to ReceiverLockGuard under the lock. +unsafe impl Sync for Receiver {} + +impl Drop for Receiver { + fn drop(&mut self) { + if let Err(err) = std::fs::remove_file(&self.lock_file_path) { + debug!("Failed to remove IPC lock file {:?}: {}", self.lock_file_path, err); + } + } +} + +impl Receiver { + fn new(lock_file_path: PathBuf, shm: Shmem) -> io::Result { + let lock_file = File::create(&lock_file_path)?; + Ok(Self { lock_file_path, lock_file, shm }) + } + + /// Lock the shared memory for unique read access. + /// Blocks until all the senders have dropped (or processes owning them have all exited) so the shared memory can be safely read. + /// During the lifetime of returned `ReceiverReadGuard`, no new senders can be created (ChannelConf::sender would fail). + pub fn lock(&self) -> io::Result> { + self.lock_file.lock()?; + let reader = ShmReader::new(unsafe { self.shm.as_slice() }); + Ok(ReceiverLockGuard { reader, lock_file: &self.lock_file }) + } +} + +pub struct ReceiverLockGuard<'a> { + reader: ShmReader<&'a [u8]>, + lock_file: &'a File, +} + +impl<'a> Drop for ReceiverLockGuard<'a> { + fn drop(&mut self) { + if let Err(err) = self.lock_file.unlock() { + debug!("Failed to unlock IPC lock file: {}", err); + } + } +} +impl<'a> Deref for ReceiverLockGuard<'a> { + type Target = ShmReader<&'a [u8]>; + + fn deref(&self) -> &Self::Target { + &self.reader + } +} + +#[cfg(test)] +mod tests { + use std::{num::NonZeroUsize, str::from_utf8}; + + use bstr::B; + use fspy_test_utils::command_executing; + + use super::*; + + #[test] + fn smoke() { + let (conf, receiver) = channel(100).unwrap(); + let mut cmd = command_executing!(conf, |conf: ChannelConf| { + let sender = conf.sender().unwrap(); + let frame_size = NonZeroUsize::new(2).unwrap(); + let mut frame = sender.claim_frame(frame_size).unwrap(); + frame.copy_from_slice(&[4, 2]); + }); + assert!(cmd.status().unwrap().success()); + + let lock = receiver.lock().unwrap(); + let mut frames = lock.iter_frames(); + + let received_frame = frames.next().unwrap(); + assert_eq!(received_frame, &[4, 2]); + + assert!(frames.next().is_none()); + } + + #[test] + fn forbid_new_senders_after_locked() { + let (conf, receiver) = channel(42).unwrap(); + let _lock = receiver.lock().unwrap(); + + let mut cmd = command_executing!(conf, |conf: ChannelConf| { + print!("{}", conf.sender().is_ok()); + }); + let output = cmd.output().unwrap(); + assert_eq!(B(&output.stdout), B("false")); + } + + #[test] + fn forbid_new_senders_after_receiver_dropped() { + let (conf, receiver) = channel(42).unwrap(); + drop(receiver); + + let mut cmd = command_executing!(conf, |conf: ChannelConf| { + print!("{}", conf.sender().is_ok()); + }); + let output = cmd.output().unwrap(); + assert_eq!(B(&output.stdout), B("false")); + } + + #[test] + fn concurrent_senders() { + let (conf, receiver) = channel(8192).unwrap(); + for i in 0u16..200 { + let mut cmd = command_executing!((conf.clone(), i), |(conf, i): (ChannelConf, u16)| { + let sender = conf.sender().unwrap(); + let data_to_send = i.to_string(); + sender + .claim_frame(NonZeroUsize::new(data_to_send.len()).unwrap()) + .unwrap() + .copy_from_slice(data_to_send.as_bytes()); + }); + let output = cmd.output().unwrap(); + assert!( + output.status.success(), + "Failed to send in iteration {}: {:?}", + i, + B(&output.stderr) + ); + } + let lock = receiver.lock().unwrap(); + let mut received_values: Vec = lock + .iter_frames() + .map(|frame| from_utf8(frame).unwrap().parse::().unwrap()) + .collect(); + received_values.sort_unstable(); + assert_eq!(received_values, (0u16..200).collect::>()); + } +} diff --git a/crates/fspy_shared/src/ipc/channel/shm_io.rs b/crates/fspy_shared/src/ipc/channel/shm_io.rs new file mode 100644 index 0000000000..e6f5ea5343 --- /dev/null +++ b/crates/fspy_shared/src/ipc/channel/shm_io.rs @@ -0,0 +1,654 @@ +//! Provides lock-free concurrent writing and reading of frames in a shared memory region. + +use core::iter::from_fn; +use std::{ + num::NonZeroUsize, + ops::{Deref, DerefMut}, + ptr::slice_from_raw_parts_mut, + sync::atomic::{AtomicI32, AtomicUsize, Ordering, fence}, +}; + +use bytemuck::must_cast; +use shared_memory::Shmem; + +// `ShmWriter` writes headers using atomic operations to prevent partial writes due to crashes, +// while `ShmReader` reads headers by simple pointer dereferences. +// This is safe because `ShmReader` is only used after all writing is done and visible to the calling thread (see docs of `ShmReader::new`). +// To ensure that the layouts of atomic types and their non-atomic counterparts are the same: +const _: () = { + assert!(size_of::() == size_of::()); + assert!(align_of::() == align_of::()); + assert!(size_of::() == size_of::()); + assert!(align_of::() == align_of::()); +}; + +/// A trait to borrow a raw memory region. +pub trait AsRawSlice { + fn as_raw_slice(&self) -> *mut [u8]; +} + +impl AsRawSlice for Shmem { + fn as_raw_slice(&self) -> *mut [u8] { + slice_from_raw_parts_mut(self.as_ptr(), self.len()) + } +} + +/// A concurrent shared memory writer. +/// +/// It's lock-free and safe to use across multiple threads/processes at the same time. +/// Internally it uses atomic operations to ensure that multiple writers can write to the shared memory without +/// overwriting each other's data. +pub struct ShmWriter { + /* + Layout of the whole shared memory: + | total byte size of frames(AtomicUsize) | frame 1 | frame 2 | ..... | + + Possible layout states of each frame: + - | 0(AtomicI32) | 0000...... | all zero. This happens when the thread/process crashed right after the frame is claimed. + - | byte size of the frame (AtomicI32) | partially written data | extra 0s to align to next frame header | This happens when the thread/process crashed during writing. + - | negative byte size of the frame (AtomicI32) | fully written data | extra 0s to align to next frame header | This is the normal case (negative size indicates completion). + */ + mem: M, + + #[cfg(test)] + fail_on_claim: bool, +} + +// unsafe impl Send for ShmWriter {} +// unsafe impl Sync for ShmWriter {} + +#[track_caller] +fn assert_alignment(ptr: *const u8) { + // Assert that the header of the shm is aligned to usize + assert_eq!(ptr as usize % align_of::(), 0); + // Assert that the content after whole shm header is aligned to i32 + assert_eq!((ptr as usize + size_of::()) % align_of::(), 0); +} + +fn roundup_to_align_frame_header(mut size: usize) -> usize { + // round up new_end so that the next frame header is aligned + const FRAME_HEADER_ALIGN: usize = align_of::(); + if size % FRAME_HEADER_ALIGN != 0 { + size += FRAME_HEADER_ALIGN - (size % FRAME_HEADER_ALIGN); + } + size +} + +pub struct FrameMut<'a> { + header: &'a AtomicI32, + content: &'a mut [u8], +} +impl Deref for FrameMut<'_> { + type Target = [u8]; + + fn deref(&self) -> &Self::Target { + self.content + } +} +impl DerefMut for FrameMut<'_> { + fn deref_mut(&mut self) -> &mut Self::Target { + self.content + } +} + +impl Drop for FrameMut<'_> { + fn drop(&mut self) { + // Prevents compiler from ordering memory operations. Ensure the data is visible before marking as fully written + fence(Ordering::Release); + + // Mark as fully written (negative size indicates completion) + let frame_size_i32 = + i32::try_from(self.content.len()).expect("frame size checked in `append_frame`"); + self.header.store(-frame_size_i32, Ordering::Relaxed); + } +} + +impl ShmWriter { + /// Create a new `ShmWriter` backed by a shared memory region. + /// + /// # Safety + /// - `mem.as_raw_slice()` must return a stable valid pointer to a memory region of `total` bytes, + /// - the memory region must only be accessed via `ShmWriter` across all the processes. + /// - The unused region of the shared memory must be initialized to zero. + pub unsafe fn new(mem: M) -> Self { + assert_alignment(mem.as_raw_slice() as *const u8); + Self { + mem, + #[cfg(test)] + fail_on_claim: false, + } + } + + // Unwrap `self` and return the underlying memory. + #[cfg(test)] + pub fn into_memory(self) -> M { + self.mem + } + + #[cfg(test)] + fn set_fail_on_claim(&mut self, fail_on_claim: bool) { + self.fail_on_claim = fail_on_claim; + } + + /// Claim a frame of size `frame_size`. + /// + /// Returns `None` if there is no sufficient remaining space (or simulated crash in tests) + /// frame_size must be non-zero because frame header being 0 would be ambiguous. + pub fn claim_frame(&self, frame_size: NonZeroUsize) -> Option> { + let shm_slice: *mut [u8] = self.mem.as_raw_slice(); + let shm_ptr = shm_slice as *mut u8; + let shm_len = self.mem.as_raw_slice().len(); + + let frame_size = frame_size.get(); + let Ok(frame_size_i32) = i32::try_from(frame_size) else { + // The frame header uses a signed 32-bit integer (i32) to store the frame size. + // Negative values are reserved to indicate completion, so only positive values are valid. + // Therefore, the maximum allowed frame size is i32::MAX (2^31-1), approximately 2GB. + // Attempting to claim a frame larger than this will fail. + return None; + }; + + // Get the atomic value of the end position (first 8 bytes of shared memory) + let atomic_header = unsafe { AtomicUsize::from_ptr(shm_ptr.cast()) }; + + let frame_with_header_size = size_of::() + frame_size; + + // Try to atomically claim the space + // Different writers only share the header, not each other's content. so relaxed ordering is sufficient. + let current_end = + atomic_header.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current_end| { + let new_end = roundup_to_align_frame_header(current_end + frame_with_header_size); + + // Check if we have enough space + if size_of::() + new_end > shm_len { + return None; + } + + Some(new_end) + }); + + let Ok(current_end) = current_end else { + return None; // Not enough space + }; + + #[cfg(test)] + if self.fail_on_claim { + // Simulate crash right after claiming the space + return None; + } + + // Successfully claimed the space, now write the data + + let frame_start = unsafe { + shm_ptr.add(/* shm header */ size_of::() + current_end) + }; + + let frame_header = unsafe { AtomicI32::from_ptr(frame_start.cast()) }; + + // Mark as partially written with positive size + // Atomic operations on the frame header is only for preventing partial writes of the frame header itself (possibly due to crashes), + // not for synchronization of frame contents, so relaxed ordering is sufficient + frame_header.store(frame_size_i32, Ordering::Relaxed); + + // Prevents compiler from re-ordering memory operations. Ensure the size is visible before writing the data + fence(Ordering::Release); + + let frame_content_ptr = unsafe { frame_start.add(size_of::()) }; // skip the frame header + Some(FrameMut { + header: frame_header, + content: unsafe { std::slice::from_raw_parts_mut(frame_content_ptr, frame_size) }, + }) + } + + #[cfg(test)] + pub fn append_frame(&self, frame: &[u8]) -> bool { + let Some(frame_size) = NonZeroUsize::new(frame.len()) else { + return false; + }; + let Some(mut frame_mut) = self.claim_frame(frame_size) else { + return false; + }; + frame_mut.copy_from_slice(frame); + true + } +} + +/// Reader of frames in shared memory created by `ShmWriter`. +pub struct ShmReader> { + mem: M, +} + +impl> ShmReader { + /// The content of `mem` should be created by `ShmWriter`. + /// Failing to do so may result in panics (mostly out-of-bounds), but won't trigger undefined behavior. + /// + /// The ShmReader must be created after all writing to the shared memory is done and visible to the calling thread. + /// This is guaranteed by `M: AsRef<[u8]>`, which means the memory region is immutable during the lifetime of `ShmReader`, + /// so no need to mark `ShmReader::new` as unsafe, but care must be taken to create a safe `M` from the shared memory. + pub fn new(mem: M) -> Self { + assert_alignment(mem.as_ref().as_ptr()); + Self { mem } + } + + /// Iterate over all the frames in the shared memory. + pub fn iter_frames(&self) -> impl Iterator { + let mem = self.mem.as_ref(); + let (header, content) = mem + .split_first_chunk::<{ size_of::() }>() + .expect("mem too small to contain header"); + let content_size: usize = must_cast(*header); + let mut remaining_content = &content[..content_size]; + + from_fn(move || { + let frame_size = loop { + // looking for the next valid frame + let (frame_header, next_remaining_content) = + remaining_content.split_first_chunk::<{ size_of::() }>()?; + remaining_content = next_remaining_content; + let frame_header: i32 = must_cast(*frame_header); + match frame_header { + 0 => { + // frame was claimed but never written (crashed process) + // Keep reading until we find a non-zero header + continue; + } + 1.. => { + // Partially written frame - skip it and continue + let size = usize::try_from(frame_header).unwrap(); + remaining_content = + &remaining_content[roundup_to_align_frame_header(size)..]; + continue; + } + ..0 => { + // Fully written frame (negative size indicates completion) + break usize::try_from(-frame_header).unwrap(); + } + } + }; + + let (frame_with_padding, next_remaining_content) = + remaining_content.split_at(roundup_to_align_frame_header(frame_size)); + remaining_content = next_remaining_content; + + Some(&frame_with_padding[..frame_size]) + }) + } +} + +#[cfg(test)] +mod tests { + use std::{ + collections::HashSet, + env::current_exe, + process::{Child, Command}, + sync::Arc, + thread, + }; + + use assert2::assert; + use bstr::BStr; + + use super::*; + + /// A mocked shared memory region for testing. + /// + /// To be testable for miri, the shared memory is allocated using `Arc` instead of real shared memory APIs. + #[derive(Clone)] + struct MockedShm { + // Why usize: to ensure alignment + // + // Why not Arc<[usize]>: + // According to miri, from the perspective of data racing, incrementing ref count of Arc<[T]> + // is considered the same as reading the content of [T], which conflicts with writing to [T] by `ShmWriter`. + // This problem is unrelated to real shared memory. + mem: Arc>, + /// The actual requested byte length. + /// + /// over-allocation might happen to ensure alignment of `usize`, so mem.len() might be inaccurate. + len: usize, + } + unsafe impl Send for MockedShm {} + unsafe impl Sync for MockedShm {} + impl MockedShm { + fn alloc(len: usize) -> Self { + // allocates this many of usize to fit the requested byte size + let size_in_usize = len / size_of::() + 1; + + let mem: Vec = core::iter::repeat(0usize).take(size_in_usize).collect(); + + Self { mem: Arc::new(mem), len } + } + } + impl AsRef<[u8]> for MockedShm { + fn as_ref(&self) -> &[u8] { + unsafe { std::slice::from_raw_parts(Vec::as_ptr(&self.mem).cast(), self.len) } + } + } + + impl AsRawSlice for MockedShm { + fn as_raw_slice(&self) -> *mut [u8] { + slice_from_raw_parts_mut(Vec::as_ptr(&self.mem).cast::().cast_mut(), self.len) + } + } + + #[test] + fn single_thread_basic() { + let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; + assert!(writer.append_frame(b"hello")); + assert!(writer.append_frame(b"world")); + assert!(writer.append_frame(b"this is a test")); + assert!(!writer.append_frame(&vec![0u8; 2048])); // too large + + let reader = ShmReader::new(writer.into_memory()); + let mut frames = reader.iter_frames(); + assert_eq!(frames.next().unwrap(), b"hello"); + assert_eq!(frames.next().unwrap(), b"world"); + assert_eq!(frames.next().unwrap(), b"this is a test"); + assert_eq!(frames.next(), None); + } + #[test] + fn single_thread_empty() { + let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; + assert!(writer.append_frame(b"hello")); + assert!(!writer.append_frame(b"")); + assert!(writer.append_frame(b"this is a test")); + + let reader = ShmReader::new(writer.into_memory()); + let mut frames = reader.iter_frames(); + assert_eq!(frames.next().unwrap(), b"hello"); + assert_eq!(frames.next().unwrap(), b"this is a test"); + assert_eq!(frames.next(), None); + } + + #[test] + fn single_thread_crash_after_claim() { + let mut writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; + assert!(writer.append_frame(b"foo")); + + // Simulate crash during writing + writer.set_fail_on_claim(true); + assert!(!writer.append_frame(b"hello")); + + writer.set_fail_on_claim(false); + assert!(writer.append_frame(b"bar")); + + let reader = ShmReader::new(writer.into_memory()); + let mut frames = reader.iter_frames(); + assert_eq!(frames.next().unwrap(), b"foo"); + assert_eq!(frames.next().unwrap(), b"bar"); + assert_eq!(frames.next(), None); + } + + #[test] + fn single_thread_crash_partial_write() { + let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; + assert!(writer.append_frame(b"foo")); + + // Simulate crash during writing + let mut frame = writer.claim_frame(5.try_into().unwrap()).unwrap(); + frame[..3].copy_from_slice(b"wor"); + std::mem::forget(frame); + + assert!(writer.append_frame(b"bar")); + + let reader = ShmReader::new(writer.into_memory()); + let mut frames = reader.iter_frames(); + assert_eq!(frames.next().unwrap(), b"foo"); + assert_eq!(frames.next().unwrap(), b"bar"); + assert_eq!(frames.next(), None); + } + + #[test] + fn single_thread_two_crashes_after_claim_and_partial_write() { + // This test verifies that ShmReader::iter correctly handles MULTIPLE consecutive + // invalid frames by continuing the loop. It's crucial for testing + // that the reader doesn't stop at the first invalid frame but keeps processing + // through multiple crash scenarios to find valid frames beyond them. + + let mut writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; + + assert!(writer.append_frame(b"foo")); + + // First crash: AfterClaim (leaves frame header as 0) + writer.set_fail_on_claim(true); + assert!(!writer.append_frame(b"world")); + writer.set_fail_on_claim(false); + + // Second crash: PartialWrite (leaves positive frame header) + let mut frame = writer.claim_frame(5.try_into().unwrap()).unwrap(); + frame[..3].copy_from_slice(b"wor"); + std::mem::forget(frame); + + assert!(writer.append_frame(b"bar")); + + // ShmReader must skip BOTH invalid frames (0 header + partial header) + // and find the valid frame beyond them - this tests the loop continuation + + let reader = ShmReader::new(writer.into_memory()); + let mut frames = reader.iter_frames(); + assert_eq!(frames.next().unwrap(), b"foo"); + assert_eq!(frames.next().unwrap(), b"bar"); + assert_eq!(frames.next(), None); + } + + #[test] + fn single_thread_two_crashes_partial_write_and_after_claim() { + let mut writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; + // This test verifies the same loop continuation behavior but with crashes + // in reverse order. This ensures the loop correctly handles different + // sequences of invalid frame types (partial write -> after claim). + + assert!(writer.append_frame(b"foo")); + + // First crash: PartialWrite (leaves positive frame header) + let mut frame = writer.claim_frame(5.try_into().unwrap()).unwrap(); + frame[..3].copy_from_slice(b"wor"); + std::mem::forget(frame); + + // Second crash: AfterClaim (leaves frame header as 0) + writer.set_fail_on_claim(true); + assert!(!writer.append_frame(b"world")); + writer.set_fail_on_claim(false); + + assert!(writer.append_frame(b"bar")); + + let reader = ShmReader::new(writer.into_memory()); + // ShmReader must skip BOTH invalid frames in this order and continue + // processing to find valid frames - tests loop robustness + let mut frames = reader.iter_frames(); + assert_eq!(frames.next().unwrap(), b"foo"); + assert_eq!(frames.next().unwrap(), b"bar"); + assert_eq!(frames.next(), None); + } + + #[test] + fn concurrent() { + let shm = MockedShm::alloc(1024 * 4); + + thread::scope(|s| { + for _ in 0..4 { + s.spawn(|| { + let writer = unsafe { ShmWriter::new(shm.clone()) }; + for _ in 0..10 { + assert!(writer.append_frame(b"hello")); + assert!(writer.append_frame(b"foo")); + assert!(writer.append_frame(b"this is a test")); + } + }); + } + }); + let mut count = 0; + let reader = ShmReader::new(shm); + for frame in reader.iter_frames() { + count += 1; + let frame = BStr::new(frame); + assert!(frame == b"hello" || frame == b"foo" || frame == b"this is a test"); + } + assert_eq!(count, 120); + } + + #[test] + fn concurrent_exceeded_size() { + let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; + thread::scope(|s| { + for _ in 0..4 { + s.spawn(|| { + for _ in 0..10 { + writer.append_frame(b"hello"); + writer.append_frame(b"foo"); + writer.append_frame(b"this is a test"); + } + }); + } + }); + let mut count = 0; + let reader = ShmReader::new(writer.into_memory()); + for frame in reader.iter_frames() { + count += 1; + let frame = BStr::new(frame); + assert!(frame == b"hello" || frame == b"foo" || frame == b"this is a test"); + } + assert!(count > 50); + } + + #[test] + fn test_integer_overflow_space_calculation() { + // Test case for potential integer overflow in space calculation + + let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; + + // Try to trigger integer overflow by using maximum values + let large_frame = vec![0u8; (i32::MAX as usize) - 100]; + + // This should fail safely, not cause overflow + assert!(!writer.append_frame(&large_frame)); + + // Small frame should still work + assert!(writer.append_frame(b"test")); + + let reader = ShmReader::new(writer.into_memory()); + let mut frames = reader.iter_frames(); + assert_eq!(frames.next().unwrap(), b"test"); + assert_eq!(frames.next(), None); + } + + #[test] + fn test_space_calculation_race_condition() { + // Test for race condition in space calculation where multiple threads + // might calculate overlapping space requirements + + let writer = unsafe { ShmWriter::new(MockedShm::alloc(200)) }; + + // Very small buffer + thread::scope(|s| { + for _ in 0..10 { + s.spawn(|| { + // Many threads trying to write large-ish frames + writer.append_frame(b"this_is_a_moderately_long_frame_that_might_cause_races"); + }); + } + }); + + // The exact count doesn't matter, but the reader should not panic + // and should handle any race conditions gracefully + + let reader = ShmReader::new(writer.into_memory()); + let mut count = 0; + for _frame in reader.iter_frames() { + count += 1; + } + // At least some but not all writes should succeed + assert!(count > 0); + assert!(count < 10); + } + + #[test] + fn test_alignment_violation_detection() { + struct Misaligned(MockedShm); + impl AsRawSlice for Misaligned { + fn as_raw_slice(&self) -> *mut [u8] { + let raw_slice = self.0.as_raw_slice(); + slice_from_raw_parts_mut( + unsafe { raw_slice.cast::().add(1) }, + raw_slice.len() - 1, + ) + } + } + // Test that alignment violations are properly detected + + // Allocate memory with proper alignment first + let shm = MockedShm::alloc(64); + + // Create a deliberately misaligned pointer by adding 1 byte + // This ensures the pointer is NOT aligned to usize boundary + let misaligned_shm = Misaligned(shm); + + // Verify the pointer is actually misaligned + assert_ne!(misaligned_shm.as_raw_slice().cast::() as usize % align_of::(), 0); + + // This should panic due to alignment assertion + let result = std::panic::catch_unwind(|| { + unsafe { ShmWriter::new(misaligned_shm) }; + }); + + // Verify that the alignment check properly caught the violation + assert!(result.is_err(), "Should panic on misaligned pointer"); + } + + #[test] + #[cfg(not(miri))] + fn real_shm_across_processes() { + use shared_memory::ShmemConf; + + const CHILD_PROCESS_ENV: &str = "FSPY_SHM_IO_TEST_CHILD_PROCESS"; + const CHILD_COUNT: usize = 12; + const FRAME_COUNT_EACH_CHILD: usize = 100; + + #[ctor::ctor] + fn child_process() { + if std::env::var_os(CHILD_PROCESS_ENV).is_none() { + return; + } + let mut args = std::env::args_os(); + args.next().unwrap(); // exe path + let shm_name = args.next().expect("shm name arg").into_string().unwrap(); + let child_index = args.next().expect("child name").into_string().unwrap(); + + let shm = ShmemConf::new().os_id(shm_name).open().unwrap(); + let writer = unsafe { ShmWriter::new(shm) }; + for i in 0..FRAME_COUNT_EACH_CHILD { + assert!(writer.append_frame(format!("{child_index} {i}").as_bytes())); + } + std::process::exit(0); + } + + let shm = ShmemConf::new().size(1024 * 1024).create().unwrap(); + let shm_name = shm.get_os_id(); + + let children: Vec = (0..CHILD_COUNT) + .map(|child_index| { + Command::new(current_exe().unwrap()) + .env(CHILD_PROCESS_ENV, "1") + .arg(shm_name) + .arg(child_index.to_string()) + .spawn() + .unwrap() + }) + .collect(); + + for mut c in children { + let status = c.wait().unwrap(); + assert!(status.success()); + } + + let shm = unsafe { shm.as_slice() }; + let reader = ShmReader::new(shm); + let frames = reader.iter_frames().map(BStr::new).collect::>(); + assert_eq!(frames.len(), CHILD_COUNT * FRAME_COUNT_EACH_CHILD); + for child_index in 0..CHILD_COUNT { + for i in 0..FRAME_COUNT_EACH_CHILD { + assert!(frames.contains(&BStr::new(format!("{child_index} {i}").as_bytes()))); + } + } + } +} diff --git a/crates/fspy_shared/src/ipc/mod.rs b/crates/fspy_shared/src/ipc/mod.rs index 4f0f42f268..fe575a7441 100644 --- a/crates/fspy_shared/src/ipc/mod.rs +++ b/crates/fspy_shared/src/ipc/mod.rs @@ -1,10 +1,8 @@ +pub mod channel; mod native_str; -pub mod shm; use bincode::{BorrowDecode, Encode, config::Configuration}; -pub use native_str::NativeStr; -#[cfg(unix)] -pub use native_str::NativeString; +pub use native_str::{NativeStr, NativeString}; pub const BINCODE_CONFIG: Configuration = bincode::config::standard(); diff --git a/crates/fspy_shared/src/ipc/native_str.rs b/crates/fspy_shared/src/ipc/native_str.rs index 95fe75148d..37fe496af0 100644 --- a/crates/fspy_shared/src/ipc/native_str.rs +++ b/crates/fspy_shared/src/ipc/native_str.rs @@ -1,21 +1,19 @@ #[cfg(windows)] use std::ffi::OsString; -#[cfg(unix)] -use std::sync::Arc; use std::{ borrow::Cow, ffi::OsStr, fmt::Debug, path::{Path, StripPrefixError}, + sync::Arc, }; use allocator_api2::alloc::Allocator; +use bincode::{BorrowDecode, Decode, Encode}; #[cfg(unix)] -use bincode::Decode; -use bincode::{BorrowDecode, Encode}; use bstr::BStr; -/// Similar to `OsStr`, but requires no copy for `encode/borrow_decode` +/// Similar to `OsStr`, but requires zero-copy to construct from either ansi or wide characters on Windows. #[derive(Encode, BorrowDecode, Clone, Copy, PartialEq, Eq)] pub struct NativeStr<'a> { #[cfg(windows)] @@ -55,13 +53,16 @@ impl<'a> NativeStr<'a> { } } + #[cfg(unix)] #[must_use] pub const fn from_bytes(bytes: &'a [u8]) -> Self { - Self { - #[cfg(windows)] - is_wide: false, - data: bytes, - } + Self { data: bytes } + } + + #[cfg(windows)] + #[must_use] + pub const fn from_ansi(bytes: &'a [u8]) -> Self { + Self { is_wide: false, data: bytes } } #[cfg(windows)] @@ -151,6 +152,7 @@ fn strip_windows_path_prefix(p: &OsStr) -> &OsStr { } } +#[cfg(unix)] impl<'a> From<&'a BStr> for NativeStr<'a> { fn from(value: &'a BStr) -> Self { Self::from_bytes(value) @@ -163,57 +165,63 @@ impl Debug for NativeStr<'_> { } } -#[cfg(unix)] +/// Similar to `OsString`, but can be losslessly encoded/decoded using bincode. +/// `Encode`/`Decoded` implementations for `OsString` requires it to be valid UTF-8. This does not. #[derive(Encode, Decode, Clone, Hash)] pub struct NativeString { + #[cfg(unix)] data: Arc<[u8]>, + #[cfg(windows)] + data: Arc<[u16]>, +} + +impl NativeString { + #[cfg(unix)] + pub fn as_os_str(&self) -> &OsStr { + use std::os::unix::ffi::OsStrExt as _; + OsStr::from_bytes(&self.data) + } + + #[cfg(windows)] + pub fn to_os_string(&self) -> OsString { + use std::os::windows::ffi::OsStringExt as _; + OsString::from_wide(&self.data) + } + + pub fn to_cow_os_str(&self) -> Cow<'_, OsStr> { + #[cfg(unix)] + return Cow::Borrowed(self.as_os_str()); + #[cfg(windows)] + return Cow::Owned(self.to_os_string()); + } } -#[cfg(unix)] impl<'a> Debug for NativeString { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - ::fmt(self.as_os_str(), f) + ::fmt(&self.to_cow_os_str(), f) } } -#[cfg(unix)] impl<'a> From<&'a OsStr> for NativeString { + #[cfg(unix)] fn from(value: &'a OsStr) -> Self { - use std::os::unix::ffi::OsStrExt; + use std::os::unix::ffi::OsStrExt as _; Self { data: value.as_bytes().into() } } -} -#[cfg(unix)] -impl<'a> From for NativeString { - fn from(value: String) -> Self { - Self { data: value.as_bytes().into() } + + #[cfg(windows)] + fn from(value: &'a OsStr) -> Self { + use std::os::windows::ffi::OsStrExt as _; + Self { data: value.encode_wide().collect() } } } -#[cfg(unix)] + impl<'a> From<&'a std::path::Path> for NativeString { fn from(value: &'a std::path::Path) -> Self { value.as_os_str().into() } } -#[cfg(unix)] -impl std::ops::Deref for NativeString { - type Target = OsStr; - - fn deref(&self) -> &Self::Target { - self.as_os_str() - } -} - -#[cfg(unix)] -impl NativeString { - #[must_use] - pub fn as_os_str(&self) -> &OsStr { - use std::os::unix::ffi::OsStrExt as _; - OsStr::from_bytes(&self.data) - } -} - #[cfg(test)] mod tests { #[cfg(windows)] @@ -221,11 +229,11 @@ mod tests { #[cfg(windows)] #[test] - fn test_from_asni() { - let asni_str = "hello"; - let native_str = NativeStr::from_bytes(asni_str.as_bytes()); + fn test_from_ansi() { + let ansi_str = "hello"; + let native_str = NativeStr::from_ansi(ansi_str.as_bytes()); let os_string = native_str.to_os_string(); - assert_eq!(os_string.to_str().unwrap(), asni_str); + assert_eq!(os_string.to_str().unwrap(), ansi_str); } #[cfg(windows)] diff --git a/crates/fspy_shared/src/ipc/shm/mod.rs b/crates/fspy_shared/src/ipc/shm/mod.rs deleted file mode 100644 index 70b786d12e..0000000000 --- a/crates/fspy_shared/src/ipc/shm/mod.rs +++ /dev/null @@ -1 +0,0 @@ -// TODO diff --git a/crates/fspy_shared_unix/src/payload.rs b/crates/fspy_shared_unix/src/payload.rs index 53a2254085..925b714af9 100644 --- a/crates/fspy_shared_unix/src/payload.rs +++ b/crates/fspy_shared_unix/src/payload.rs @@ -1,13 +1,14 @@ -use std::os::{fd::RawFd, unix::ffi::OsStringExt}; +use std::os::unix::ffi::OsStringExt; use base64::{Engine as _, prelude::BASE64_STANDARD_NO_PAD}; use bincode::{Decode, Encode, config::standard}; use bstr::BString; -use fspy_shared::ipc::NativeString; +use fspy_shared::ipc::{NativeString, channel::ChannelConf}; #[derive(Debug, Encode, Decode)] pub struct Payload { - pub ipc_fd: RawFd, + pub ipc_channel_conf: ChannelConf, + pub preload_path: NativeString, #[cfg(target_os = "macos")] diff --git a/crates/fspy_shared_unix/src/spawn/linux/mod.rs b/crates/fspy_shared_unix/src/spawn/linux/mod.rs index f6ea8d16e9..8797b3c4b5 100644 --- a/crates/fspy_shared_unix/src/spawn/linux/mod.rs +++ b/crates/fspy_shared_unix/src/spawn/linux/mod.rs @@ -47,7 +47,7 @@ pub fn handle_exec( ensure_env( &mut command.envs, LD_PRELOAD, - encoded_payload.payload.preload_path.as_bytes(), + encoded_payload.payload.preload_path.as_os_str().as_bytes(), )?; ensure_env(&mut command.envs, PAYLOAD_ENV_NAME, &encoded_payload.encoded_string)?; Ok(None) diff --git a/crates/fspy_shared_unix/src/spawn/macos.rs b/crates/fspy_shared_unix/src/spawn/macos.rs index 648f58e6ea..cbbb14cfa4 100644 --- a/crates/fspy_shared_unix/src/spawn/macos.rs +++ b/crates/fspy_shared_unix/src/spawn/macos.rs @@ -52,10 +52,10 @@ pub fn handle_exec( } else if matches!(parent.as_os_str().as_bytes(), b"/bin" | b"/usr/bin") { let fixtures = &encoded_payload.payload.fixtures; if matches!(file_name.as_bytes(), b"sh" | b"bash") { - command.program = fixtures.bash_path.as_bytes().into(); + command.program = fixtures.bash_path.as_os_str().as_bytes().into(); true } else if COREUTILS_FUNCTIONS.contains(file_name.as_bytes()) { - command.program = fixtures.coreutils_path.as_bytes().into(); + command.program = fixtures.coreutils_path.as_os_str().as_bytes().into(); true } else { false @@ -71,7 +71,7 @@ pub fn handle_exec( ensure_env( &mut command.envs, DYLD_INSERT_LIBRARIES, - encoded_payload.payload.preload_path.as_bytes(), + encoded_payload.payload.preload_path.as_os_str().as_bytes(), )?; ensure_env(&mut command.envs, PAYLOAD_ENV_NAME, &encoded_payload.encoded_string)?; } else { diff --git a/crates/fspy_test_utils/Cargo.toml b/crates/fspy_test_utils/Cargo.toml new file mode 100644 index 0000000000..5333f36266 --- /dev/null +++ b/crates/fspy_test_utils/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "fspy_test_utils" +version = "0.0.0" +authors.workspace = true +edition.workspace = true +license.workspace = true +publish = false +rust-version.workspace = true + +[dependencies] +base64 = { workspace = true } +bincode = { workspace = true } +ctor = { workspace = true } + +[lints] +workspace = true diff --git a/crates/fspy_test_utils/README.md b/crates/fspy_test_utils/README.md new file mode 100644 index 0000000000..b7074b0605 --- /dev/null +++ b/crates/fspy_test_utils/README.md @@ -0,0 +1,3 @@ +# fspy_test_utils + +This crate provides shared test utilities for fspy-related tests. It is intended to be used only in test configurations of other crates. diff --git a/crates/fspy_test_utils/src/lib.rs b/crates/fspy_test_utils/src/lib.rs new file mode 100644 index 0000000000..d3e15d2f38 --- /dev/null +++ b/crates/fspy_test_utils/src/lib.rs @@ -0,0 +1,75 @@ +use std::{env::current_exe, process::Command}; + +use base64::{Engine, prelude::BASE64_STANDARD_NO_PAD}; +use bincode::{Decode, Encode, config}; + +/// Creates a `std::process::Command` that only executes the provided function. +/// +/// - $arg: The argument to pass to the function, must implement `Encode` and `Decode`. +/// - $f: The function to run in the separate process, takes one argument of the type of $arg. +#[macro_export] +macro_rules! command_executing { + ($arg: expr, $f: expr) => {{ + // Generate a unique ID for every invocation of this macro. + const ID: &str = + ::core::concat!(::core::file!(), ":", ::core::line!(), ":", ::core::column!()); + + fn assert_arg_type(_arg: &A, _f: impl FnOnce(A)) {} + assert_arg_type(&$arg, $f); + + // Register an initializer that runs the provided function when the process is started + #[ctor::ctor] + unsafe fn init() { + $crate::init_impl(ID, $f); + } + // Create the command + $crate::create_command(ID, $arg) + }}; +} + +#[doc(hidden)] +pub fn init_impl>(expected_id: &str, f: impl FnOnce(A)) { + let mut args = ::std::env::args(); + // + let (Some(_program), Some(current_id), Some(arg_base64)) = + (args.next(), args.next(), args.next()) + else { + return; + }; + if current_id != expected_id { + return; + } + let arg_bytes = BASE64_STANDARD_NO_PAD.decode(arg_base64).expect("Failed to decode base64 arg"); + let arg: A = bincode::decode_from_slice(&arg_bytes, config::standard()) + .expect("Failed to decode bincode arg") + .0; + f(arg); + std::process::exit(0); +} + +#[doc(hidden)] +pub fn create_command(id: &str, arg: impl Encode) -> Command { + let mut command = Command::new(current_exe().unwrap()); + let arg_bytes = bincode::encode_to_vec(&arg, config::standard()).expect("Failed to encode arg"); + let arg_base64 = BASE64_STANDARD_NO_PAD.encode(&arg_bytes); + command.arg(id).arg(arg_base64); + + command +} + +#[cfg(test)] +mod tests { + use std::str::from_utf8; + + use super::*; + + #[test] + fn test_command_executing() { + let mut command = command_executing!(42u32, |arg: u32| { + print!("{}", arg); + }); + let output = command.output().unwrap(); + assert_eq!(from_utf8(&output.stdout), Ok("42")); + assert!(output.status.success()); + } +} diff --git a/crates/vite_task/src/execute.rs b/crates/vite_task/src/execute.rs index 5ac6afb71b..8a1562f6d3 100644 --- a/crates/vite_task/src/execute.rs +++ b/crates/vite_task/src/execute.rs @@ -10,7 +10,7 @@ use std::{ use bincode::{Decode, Encode}; use fspy::{AccessMode, Spy, TrackedChild}; -use futures_util::future::{try_join3, try_join4}; +use futures_util::future::try_join3; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use supports_color::{Stream, on}; @@ -457,10 +457,9 @@ pub async fn execute_task( Ok::<_, Error>((path_reads, path_writes)) }; - let ((), (), (path_reads, path_writes), (exit_status, duration)) = try_join4( + let ((), (), (exit_status, duration)) = try_join3( collect_std_outputs(&outputs, child_stdout, OutputKind::StdOut), collect_std_outputs(&outputs, child_stderr, OutputKind::StdErr), - path_accesses_fut, async move { let start = Instant::now(); let exit_status = child.wait().await?; @@ -469,6 +468,8 @@ pub async fn execute_task( ) .await?; + let (path_reads, path_writes) = path_accesses_fut.await?; + let outputs = outputs.into_inner().unwrap(); tracing::debug!( "executed task finished, path_reads: {}, path_writes: {}, outputs: {}, exit_status: {}", diff --git a/fspy/.gitignore b/fspy/.gitignore deleted file mode 100644 index b43eca9088..0000000000 --- a/fspy/.gitignore +++ /dev/null @@ -1,3 +0,0 @@ -/target -.DS_Store -/ecosystem diff --git a/fspy/node_spawn.mjs b/fspy/node_spawn.mjs deleted file mode 100755 index 1dbb106c09..0000000000 --- a/fspy/node_spawn.mjs +++ /dev/null @@ -1,6 +0,0 @@ -#!/usr/bin/env node - -import cp from 'node:child_process' -import fs from 'node:fs' -cp.execFileSync('/home/vscode/esbuild', ['a.js'], { stdio: 'inherit'}); -fs.readdirSync('/workspaces'); diff --git a/packages/cli/snap-tests/vitest-browser-mode/steps.json b/packages/cli/snap-tests/vitest-browser-mode/steps.json index 07ef1a0f11..559014f882 100644 --- a/packages/cli/snap-tests/vitest-browser-mode/steps.json +++ b/packages/cli/snap-tests/vitest-browser-mode/steps.json @@ -1,5 +1,4 @@ { - "ignoredPlatforms": ["linux"], "env": { "VITE_DISABLE_AUTO_INSTALL": "1" },