diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ba0419ab..a0d93dcf2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,7 @@ # Changelog +- **Fixed** `vp run` no longer hangs or fails when a task leaves a process running behind it, such as a dev server or a background helper, or when one of a task's processes is killed. The run finishes as soon as the task itself does, and the files the task used are still recorded ([#544](https://github.com/voidzero-dev/vite-task/issues/544), [#675](https://github.com/voidzero-dev/vite-task/pull/675)). +- **Fixed** A task that reads or writes an unusually large number of files now runs to the end instead of being killed partway through. Vite+ reports the run as not cached, because it could not record every file the task used ([#533](https://github.com/voidzero-dev/vite-task/issues/533), [#675](https://github.com/voidzero-dev/vite-task/pull/675)). - **Fixed** Vite+ diagnostics now display individual paths and working directories without Rust debug formatting such as quoted paths or escaped Windows backslashes ([#534](https://github.com/voidzero-dev/vite-task/pull/534)). - **Fixed** Automatic file-access tracking now works inside the default Codex CLI and Claude Code sandboxes ([#562](https://github.com/voidzero-dev/vite-task/issues/562), [#563](https://github.com/voidzero-dev/vite-task/issues/563), [#576](https://github.com/voidzero-dev/vite-task/pull/576), [#569](https://github.com/voidzero-dev/vite-task/pull/569)). - **Fixed** Broad workspace globs no longer discover and run package scripts inside `node_modules` ([#539](https://github.com/voidzero-dev/vite-task/pull/539)). diff --git a/Cargo.lock b/Cargo.lock index f0549dd6b..35ce5bdd0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1306,7 +1306,6 @@ dependencies = [ "itoa", "libc", "nix 0.31.2", - "wincode", ] [[package]] @@ -1435,7 +1434,6 @@ dependencies = [ "subprocess_test", "thiserror 2.0.18", "tokio", - "tracing", "uuid", "vt_path", "winapi", diff --git a/crates/fspy/examples/cli.rs b/crates/fspy/examples/cli.rs index 4ae34790f..8b1331ce4 100644 --- a/crates/fspy/examples/cli.rs +++ b/crates/fspy/examples/cli.rs @@ -27,7 +27,11 @@ async fn main() -> anyhow::Result<()> { let mut csv_writer = csv_async::AsyncWriter::from_writer(out_file); - for acc in termination.path_accesses.iter() { + for acc in termination + .path_accesses + .expect("the tracking region holds every record this run makes") + .iter() + { path_count += 1; let path_str = format!("{:?}", acc.path); let mode_str = format!("{:?}", acc.mode); diff --git a/crates/fspy/src/error.rs b/crates/fspy/src/error.rs index 56f983c70..6df8e4a5d 100644 --- a/crates/fspy/src/error.rs +++ b/crates/fspy/src/error.rs @@ -30,3 +30,15 @@ pub enum SpawnError { #[error("underlying os error: {0}")] OsSpawn(std::io::Error), } + +/// A tracked process could not record a file access it went on to perform, +/// so the accesses collected for the run are a subset of what it really +/// touched. +/// +/// The run itself is unaffected: recording must never stop the program +/// doing the work. What cannot be done is anything that needs every +/// access, caching above all, which has to treat the run as untracked +/// rather than as having touched only the paths that fit. +#[derive(thiserror::Error, Clone, Copy, PartialEq, Eq, Debug)] +#[error("the file-access records did not fit in the tracking channel")] +pub struct TrackingIncomplete; diff --git a/crates/fspy/src/ipc.rs b/crates/fspy/src/ipc.rs index a804fa8c3..fbb1e1485 100644 --- a/crates/fspy/src/ipc.rs +++ b/crates/fspy/src/ipc.rs @@ -1,10 +1,9 @@ -use std::io; - use fspy_shared::ipc::{ - PathAccess, - channel::{Receiver, ReceiverLockGuard}, + ChannelSize, PathAccess, + channel::{FrameReader, Receiver}, }; -use tokio::task::spawn_blocking; + +use crate::error::TrackingIncomplete; /// Shared memory for one tracked run's file-access records. /// @@ -18,43 +17,55 @@ const DEFAULT_SHM_CAPACITY: usize = 4 * 1024 * 1024 * 1024; /// nothing outside this repository should set it. const SHM_CAPACITY_ENV: &str = "VP_RUN_INTERNAL_FSPY_SHM_CAPACITY"; -/// How much shared memory to give the next tracked run. +/// How much shared memory to give the next tracked run, split at one +/// record per 64 bytes: 8 for its descriptor and 56 for its payload. +/// Records run a few hundred bytes each, so payload space runs out well +/// before slots do. /// /// # Panics /// /// When the override is set to something that is not a byte count. It is /// ours to set, so a value we cannot read is a mistake worth stopping for /// rather than quietly ignoring. -pub fn shm_capacity() -> usize { - std::env::var_os(SHM_CAPACITY_ENV).map_or(DEFAULT_SHM_CAPACITY, |value| { +pub fn shm_size() -> ChannelSize { + let capacity = std::env::var_os(SHM_CAPACITY_ENV).map_or(DEFAULT_SHM_CAPACITY, |value| { value.to_str().and_then(|value| value.parse().ok()).unwrap_or_else(|| { panic!("{SHM_CAPACITY_ENV} is not a byte count: {}", value.display()) }) - }) + }); + ChannelSize { capacity, slots: capacity / 64 } } -#[ouroboros::self_referencing] -pub struct OwnedReceiverLockGuard { - /// Owns the shared memory - receiver: Receiver, - /// Borrows the shared memory and owns the file lock - #[borrows(receiver)] - #[covariant] - lock_guard: ReceiverLockGuard<'this>, +/// The path accesses a run reported through the IPC channel. +pub struct ChannelAccesses { + frames: FrameReader, } -impl OwnedReceiverLockGuard { - pub fn lock(receiver: Receiver) -> io::Result { - Self::try_new(receiver, fspy_shared::ipc::channel::Receiver::lock) - } +impl TryFrom for ChannelAccesses { + type Error = TrackingIncomplete; - pub async fn lock_async(receiver: Receiver) -> io::Result { - spawn_blocking(move || Self::lock(receiver)).await.expect("lock task panicked") + /// Closes the channel and takes every record it collected. + /// + /// Never waits for tracked processes: closing reads one counter and + /// shuts the channel's gate (see + /// [`fspy_shared::ipc::channel::Receiver::close`]), so it runs inline + /// however many records were reported. + /// + /// # Errors + /// + /// [`TrackingIncomplete`] when a tracked process could not record + /// something it went on to do. What did arrive is then a subset of + /// what the run really touched, so none of it is handed back. + fn try_from(receiver: Receiver) -> Result { + Ok(Self { frames: receiver.close().map_err(|_| TrackingIncomplete)? }) } +} +impl ChannelAccesses { pub fn iter_path_accesses(&self) -> impl Iterator> { - self.borrow_lock_guard() - .iter_frames() - .map(|frame| wincode::deserialize_exact(frame).unwrap()) + self.frames.iter().map(|frame| { + wincode::deserialize_exact(frame) + .expect("committed frames are complete under the channel protocol") + }) } } diff --git a/crates/fspy/src/lib.rs b/crates/fspy/src/lib.rs index 6c89414ba..5621547f5 100644 --- a/crates/fspy/src/lib.rs +++ b/crates/fspy/src/lib.rs @@ -20,6 +20,7 @@ mod command; use std::{env::temp_dir, fs::create_dir, io, process::ExitStatus, sync::LazyLock}; pub use command::Command; +pub use error::TrackingIncomplete; pub use fspy_shared::ipc::{AccessMode, PathAccess}; use futures_util::future::BoxFuture; pub use os_impl::PathAccessIterable; @@ -30,8 +31,9 @@ use tokio::process::{ChildStderr, ChildStdin, ChildStdout}; pub struct ChildTermination { /// The exit status of the child process. pub status: ExitStatus, - /// The path accesses captured from the child process. - pub path_accesses: PathAccessIterable, + /// The path accesses captured from the child process, or the reason + /// they cannot be trusted to be all of them. + pub path_accesses: Result, } pub struct TrackedChild { diff --git a/crates/fspy/src/unix/mod.rs b/crates/fspy/src/unix/mod.rs index c612d00b7..0ea6b413e 100644 --- a/crates/fspy/src/unix/mod.rs +++ b/crates/fspy/src/unix/mod.rs @@ -25,7 +25,7 @@ use tokio::task::spawn_blocking; use tokio_util::sync::CancellationToken; #[cfg(not(target_env = "musl"))] -use crate::ipc::OwnedReceiverLockGuard; +use crate::ipc::ChannelAccesses; use crate::{ChildTermination, Command, TrackedChild, arena::PathAccessArena, error::SpawnError}; #[derive(Debug)] @@ -80,7 +80,7 @@ impl SpyImpl { #[cfg(not(target_env = "musl"))] let (ipc_channel_conf, ipc_receiver) = - channel(crate::ipc::shm_capacity()).map_err(SpawnError::ChannelCreation)?; + channel(crate::ipc::shm_size()).map_err(SpawnError::ChannelCreation)?; let payload = Payload { #[cfg(not(target_env = "musl"))] @@ -137,7 +137,7 @@ impl SpyImpl { stdout: child.stdout.take(), stderr: child.stderr.take(), // Keep polling for the child to exit in the background even if `wait_handle` is not awaited, - // because we need to stop the supervisor and lock the channel as soon as the child exits. + // because we need to stop the supervisor and close the channel as soon as the child exits. wait_handle: tokio::spawn(async move { let status = tokio::select! { status = child.wait() => status?, @@ -159,16 +159,14 @@ impl SpyImpl { ); let arenas = arenas.collect::>(); - // Lock the ipc channel after the child has exited. + // Close the ipc channel after the child has exited. // We are not interested in path accesses from descendants after the main child has exited. #[cfg(not(target_env = "musl"))] - let ipc_receiver_lock_guard = - OwnedReceiverLockGuard::lock_async(ipc_receiver).await?; - let path_accesses = PathAccessIterable { - arenas, - #[cfg(not(target_env = "musl"))] - ipc_receiver_lock_guard, - }; + #[cfg(not(target_env = "musl"))] + let path_accesses = ChannelAccesses::try_from(ipc_receiver) + .map(|ipc_accesses| PathAccessIterable { arenas, ipc_accesses }); + #[cfg(target_env = "musl")] + let path_accesses = Ok(PathAccessIterable { arenas }); io::Result::Ok(ChildTermination { status, path_accesses }) }) @@ -181,7 +179,7 @@ impl SpyImpl { pub struct PathAccessIterable { arenas: Vec, #[cfg(not(target_env = "musl"))] - ipc_receiver_lock_guard: OwnedReceiverLockGuard, + ipc_accesses: ChannelAccesses, } impl PathAccessIterable { @@ -191,7 +189,7 @@ impl PathAccessIterable { #[cfg(not(target_env = "musl"))] { - let accesses_in_shm = self.ipc_receiver_lock_guard.iter_path_accesses(); + let accesses_in_shm = self.ipc_accesses.iter_path_accesses(); accesses_in_shm.chain(accesses_in_arena) } #[cfg(target_env = "musl")] diff --git a/crates/fspy/src/windows/mod.rs b/crates/fspy/src/windows/mod.rs index 2e6470d47..eb568a9f5 100644 --- a/crates/fspy/src/windows/mod.rs +++ b/crates/fspy/src/windows/mod.rs @@ -21,20 +21,19 @@ use winapi::{ use winsafe::co::{CP, WC}; use crate::{ - ChildTermination, TrackedChild, command::Command, error::SpawnError, - ipc::OwnedReceiverLockGuard, + ChildTermination, TrackedChild, command::Command, error::SpawnError, ipc::ChannelAccesses, }; const INTERPOSE_CDYLIB: Artifact = artifact!("fspy_preload", "CARGO_CDYLIB_FILE_FSPY_PRELOAD_WINDOWS"); pub struct PathAccessIterable { - ipc_receiver_lock_guard: OwnedReceiverLockGuard, + ipc_accesses: ChannelAccesses, } impl PathAccessIterable { pub fn iter(&self) -> impl Iterator> { - self.ipc_receiver_lock_guard.iter_path_accesses() + self.ipc_accesses.iter_path_accesses() } } @@ -85,7 +84,7 @@ impl SpyImpl { command.creation_flags(CREATE_SUSPENDED); let (channel_conf, receiver) = - channel(crate::ipc::shm_capacity()).map_err(SpawnError::ChannelCreation)?; + channel(crate::ipc::shm_size()).map_err(SpawnError::ChannelCreation)?; let mut spawn_success = false; let spawn_success = &mut spawn_success; @@ -156,7 +155,7 @@ impl SpyImpl { stderr: child.stderr.take(), process_handle, // Keep polling for the child to exit in the background even if `wait_handle` is not awaited, - // because we need to stop the supervisor and lock the channel as soon as the child exits. + // because we need to stop the supervisor and close the channel as soon as the child exits. wait_handle: tokio::spawn(async move { let status = tokio::select! { status = child.wait() => status?, @@ -165,10 +164,10 @@ impl SpyImpl { child.wait().await? } }; - // Lock the ipc channel after the child has exited. + // Close the ipc channel after the child has exited. // We are not interested in path accesses from descendants after the main child has exited. - let ipc_receiver_lock_guard = OwnedReceiverLockGuard::lock_async(receiver).await?; - let path_accesses = PathAccessIterable { ipc_receiver_lock_guard }; + let path_accesses = ChannelAccesses::try_from(receiver) + .map(|ipc_accesses| PathAccessIterable { ipc_accesses }); io::Result::Ok(ChildTermination { status, path_accesses }) }) diff --git a/crates/fspy/tests/node_fs.rs b/crates/fspy/tests/node_fs.rs index 96e951487..27574b5bb 100644 --- a/crates/fspy/tests/node_fs.rs +++ b/crates/fspy/tests/node_fs.rs @@ -49,7 +49,9 @@ fn track_script( let child = command.spawn(tokio_util::sync::CancellationToken::new()).await?; let termination = child.wait_handle.await?; assert!(termination.status.success()); - Ok(termination.path_accesses) + Ok(termination + .path_accesses + .expect("the tracking region holds every record this run makes")) }) } diff --git a/crates/fspy/tests/oxlint.rs b/crates/fspy/tests/oxlint.rs index fe4a96291..2c3e06f63 100644 --- a/crates/fspy/tests/oxlint.rs +++ b/crates/fspy/tests/oxlint.rs @@ -51,7 +51,7 @@ async fn track_oxlint(dir: &std::path::Path, args: &[&str]) -> anyhow::Result) -> PathAccessIterable let termination = tracked_child.wait_handle.await.unwrap(); assert!(termination.status.success()); - termination.path_accesses + termination.path_accesses.expect("the tracking region holds every record this run makes") } #[test(tokio::test)] diff --git a/crates/fspy/tests/test_utils/mod.rs b/crates/fspy/tests/test_utils/mod.rs index cfa46c4a9..9487b0d90 100644 --- a/crates/fspy/tests/test_utils/mod.rs +++ b/crates/fspy/tests/test_utils/mod.rs @@ -84,5 +84,5 @@ pub async fn spawn_command(cmd: subprocess_test::Command) -> anyhow::Result PathAccessIterable; +} + +impl TrackedAccesses for PathAccessIterable { + fn tracked(self) -> PathAccessIterable { + self + } +} + +impl TrackedAccesses for Result { + fn tracked(self) -> PathAccessIterable { + self.expect("the tracking region holds every record this run makes") + } +} + async fn validate(target: &OsString, target_args: &[OsString], relative: bool) { let mut command = Command::new(target); command @@ -159,7 +183,7 @@ async fn validate(target: &OsString, target_args: &[OsString], relative: bool) { .await .expect("failed to wait for tracked target"); assert!(termination.status.success(), "benchmark target failed: {}", termination.status); - let captured_missing_access = termination.path_accesses.iter().any(|access| { + let captured_missing_access = termination.path_accesses.tracked().iter().any(|access| { access.path.strip_path_prefix(MISSING_PATH, |result| { result.is_ok_and(|path| path.as_os_str().is_empty()) }) diff --git a/crates/fspy_client_unix/Cargo.toml b/crates/fspy_client_unix/Cargo.toml index a7e42e124..c0b9da26a 100644 --- a/crates/fspy_client_unix/Cargo.toml +++ b/crates/fspy_client_unix/Cargo.toml @@ -14,7 +14,6 @@ libc = { workspace = true } nix = { workspace = true, features = ["fs"] } fspy_nostd = { workspace = true } fspy_nostd_alloc = { workspace = true } -wincode = { workspace = true } [target.'cfg(all(target_os = "linux", not(target_env = "musl")))'.dependencies] itoa = { workspace = true } diff --git a/crates/fspy_client_unix/src/lib.rs b/crates/fspy_client_unix/src/lib.rs index d697257e2..3ba3877a1 100644 --- a/crates/fspy_client_unix/src/lib.rs +++ b/crates/fspy_client_unix/src/lib.rs @@ -8,7 +8,7 @@ pub mod convert; pub mod raw_exec; -use std::{ffi::OsStr, fmt::Debug, num::NonZeroUsize, os::unix::ffi::OsStrExt as _, path::Path}; +use std::{ffi::OsStr, fmt::Debug, os::unix::ffi::OsStrExt as _, path::Path}; use convert::{ToAbsolutePath, ToAccessMode}; use fspy_shared::ipc::{PathAccess, channel::Sender}; @@ -18,7 +18,6 @@ use fspy_shared_unix::{ spawn::{PreExec, handle_exec}, }; use raw_exec::RawExec; -use wincode::Serialize as _; pub struct Client { encoded_payload: EncodedPayload, @@ -45,53 +44,35 @@ impl Client { /// /// # Panics /// - /// Panics when the payload is missing, malformed, or cannot be decoded. - #[expect( - clippy::print_stderr, - reason = "the client intentionally reports an unavailable supervisor channel" - )] + /// Panics when the payload is missing, malformed, or cannot be decoded, + /// and when the channel is there but cannot be attached to (see + /// [`ChannelConf::sender`](fspy_shared::ipc::channel::ChannelConf::sender)). pub fn from_env(envs: impl Iterator) -> Self { let encoded_payload = decode_payload_from_env(envs).unwrap(); - let ipc_sender = match encoded_payload.payload.ipc_channel_conf.sender() { - Ok(sender) => Some(sender), - Err(err) => { - // This can happen if the process starts after the root target - // has exited and the receiver has closed the channel. - eprintln!("fspy: failed to create ipc sender: {err}"); - None - } - }; + // `None` when the channel is already over, which happens when this + // process starts after the root target exited. Nothing is said + // about it: a preload library writing to the traced process's + // stderr corrupts whatever that process is printing. + let ipc_sender = encoded_payload.payload.ipc_channel_conf.sender(); Self { encoded_payload, ipc_sender } } - fn send(&self, mode: fspy_shared::ipc::AccessMode, path: &Path) -> anyhow::Result<()> { + fn send(&self, mode: fspy_shared::ipc::AccessMode, path: &Path) { let Some(ipc_sender) = &self.ipc_sender else { - return Ok(()); + return; }; let path_bytes = path.as_os_str().as_bytes(); if path_bytes.starts_with(b"/dev/") || (cfg!(target_os = "linux") && (path_bytes.starts_with(b"/proc/") || path_bytes.starts_with(b"/sys/"))) { - return Ok(()); + return; } - let path_access = PathAccess { mode, path: path.into() }; - let serialized_size = usize::try_from(PathAccess::serialized_size(&path_access)?) - .expect("serialized size exceeds usize"); - - let frame_size = NonZeroUsize::new(serialized_size) - .expect("fspy: encoded PathAccess should never be empty"); - - let mut frame = ipc_sender - .claim_frame(frame_size) - .expect("fspy: failed to claim frame in shared memory"); - let mut writer: &mut [u8] = &mut frame; - PathAccess::serialize_into(&mut writer, &path_access)?; - assert_eq!(writer.len(), 0); - - Ok(()) + // The interception proceeds whether or not the record could be + // sent — a preload library can never panic its host process. + ipc_sender.send(&PathAccess { mode, path: path.into() }); } /// Resolves and reports an exec before forwarding its transformed arguments. @@ -106,10 +87,6 @@ impl Client { /// /// Returns errors from exec resolution, platform preparation, or the /// forwarding callback. - /// - /// # Panics - /// - /// Panics if reporting the executable path fails. pub unsafe fn handle_exec( &self, config: ExecResolveConfig, @@ -120,7 +97,7 @@ impl Client { // null-terminated arrays, as provided by the caller. let mut exec = unsafe { raw_exec.to_exec() }; let pre_exec = handle_exec(&mut exec, config, &self.encoded_payload, |mode, path| { - self.send(mode, path).unwrap(); + self.send(mode, path); })?; RawExec::from_exec(exec, |raw_command| f(raw_command, pre_exec)) } @@ -147,6 +124,7 @@ impl Client { let Some(abs_path) = path.to_absolute_path(&arena)? else { return Ok(()); }; - self.send(mode, Path::new(OsStr::from_bytes(abs_path.as_units()))) + self.send(mode, Path::new(OsStr::from_bytes(abs_path.as_units()))); + Ok(()) } } diff --git a/crates/fspy_e2e/src/main.rs b/crates/fspy_e2e/src/main.rs index 9d6a30525..ca3d7a25b 100644 --- a/crates/fspy_e2e/src/main.rs +++ b/crates/fspy_e2e/src/main.rs @@ -108,7 +108,11 @@ async fn main() { } let mut collector = AccessCollector::new(dir); - for access in termination.path_accesses.iter() { + for access in termination + .path_accesses + .expect("the tracking region holds every record this run makes") + .iter() + { collector.add(access); } let snap_file = File::create(manifest_dir.join(format!("snaps/{name}.txt"))).unwrap(); diff --git a/crates/fspy_preload_windows/src/windows/client.rs b/crates/fspy_preload_windows/src/windows/client.rs index 48933414e..bbe4e0c11 100644 --- a/crates/fspy_preload_windows/src/windows/client.rs +++ b/crates/fspy_preload_windows/src/windows/client.rs @@ -16,22 +16,11 @@ impl<'a> Client<'a> { pub fn from_payload_bytes(payload_bytes: &'a [u8]) -> Self { let payload: Payload<'a> = wincode::deserialize_exact(payload_bytes).unwrap(); - let ipc_sender = match payload.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. - #[expect( - clippy::print_stderr, - reason = "preload library uses stderr for debug diagnostics" - )] - { - eprintln!("fspy: failed to create ipc sender: {err}"); - } - None - } - }; + // `None` when the channel is already over, which happens when this + // process starts after the root target exited. Nothing is said + // about it: a detours DLL writing to the traced process's stderr + // corrupts whatever that process is printing. + let ipc_sender = payload.channel_conf.sender(); Self { payload, ipc_sender } } @@ -40,7 +29,9 @@ impl<'a> Client<'a> { let Some(sender) = &self.ipc_sender else { return; }; - sender.write_encoded(&access).expect("failed to send path access"); + // The intercepted call proceeds whether or not the record could be + // sent; a detours DLL can never panic its host. + sender.send(&access); } pub unsafe fn prepare_child_process(&self, child_handle: HANDLE) -> BOOL { diff --git a/crates/fspy_shared/Cargo.toml b/crates/fspy_shared/Cargo.toml index 071210f63..213d25011 100644 --- a/crates/fspy_shared/Cargo.toml +++ b/crates/fspy_shared/Cargo.toml @@ -17,7 +17,6 @@ fspy_nostd_alloc = { workspace = true } fspy_shm = { workspace = true } fspy_ipc_str = { workspace = true } thiserror = { workspace = true } -tracing = { workspace = true } uuid = { workspace = true, features = ["v4"] } vt_path = { workspace = true } diff --git a/crates/fspy_shared/src/ipc/channel/mod.rs b/crates/fspy_shared/src/ipc/channel/mod.rs index ad5c5c205..b5fb33f47 100644 --- a/crates/fspy_shared/src/ipc/channel/mod.rs +++ b/crates/fspy_shared/src/ipc/channel/mod.rs @@ -1,20 +1,27 @@ //! Fast mpsc IPC channel implementation based on shared memory. +//! +//! The channel is crash-tolerant and nonblocking on both ends: any sender +//! process may die (or keep running) at any point without preventing the +//! receiver from closing the channel and reading every committed frame. See +//! the `shm_io` module for the underlying protocol. mod shm_io; -use std::{env::temp_dir, ffi::OsStr, fs::File, io, ops::Deref, path::PathBuf}; +use std::{env::temp_dir, ffi::OsStr, io, num::NonZeroUsize, path::PathBuf}; use allocator_api2::alloc::Global; use fspy_nostd::Fat; use fspy_nostd_alloc::OsCString; use fspy_shm::Mapping; -pub use shm_io::FrameMut; -use shm_io::{ShmReader, ShmWriter}; -use tracing::debug; +use shm_io::{SealError, ShmReader, ShmWriter, from_usize, to_usize}; + +/// Reads the committed frames of a sealed channel; borrows the shared +/// mapping, which stays alive (and mapped) until this value drops. +pub type FrameReader = shm_io::ShmReader; use uuid::Uuid; -use wincode::{SchemaRead, SchemaWrite}; +use wincode::{SchemaRead, SchemaWrite, Serialize as _, config::DefaultConfig}; -use super::IpcStr; +use super::{ChannelSize, IpcStr}; /// Prefix of shared-memory backing file names inside the system temporary /// directory. @@ -27,16 +34,16 @@ const SHM_BACKING_PREFIX: &str = "vite-task-fspy-"; /// Serializable configuration to create channel senders. #[derive(SchemaWrite, SchemaRead, Clone, Debug)] pub struct ChannelConf { - lock_file_path: Box, shm_id: Box, + /// The slot count the region was created with, since a sender cannot + /// work it out from the mapping's size alone. + slots: u64, } -/// Creates a mpsc IPC channel with one receiver and a `ChannelConf` that can be passed around processes and used to create multiple senders +/// Creates a mpsc IPC channel with one receiver and a `ChannelConf` that can be passed around processes and used to create multiple senders. #[expect(clippy::missing_errors_doc, reason = "non-vt crate: cannot use vt_str/vt_path types")] -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())); - +pub fn channel(size: ChannelSize) -> io::Result<(ChannelConf, Receiver)> { + let ChannelSize { capacity, slots } = size; let shm_c_path = os_c_string(shm_backing_path()?.as_os_str())?; let handle = fspy_shm::create(shm_c_path.as_c_str().as_thin(), capacity).map_err(shm_error_to_io)?; @@ -44,13 +51,24 @@ pub fn channel(capacity: usize) -> io::Result<(ChannelConf, Receiver)> { let keeper = ShmKeeper { path: shm_c_path }; let mapping = handle.map().map_err(shm_error_to_io)?; + // Prove the region can host the protocol — the same fallible attach + // senders perform — so a size the two halves do not fit in fails the + // task now, not at its first record. + // SAFETY: the region was just created zero-initialized and is only + // accessed through the `shm_io` protocol. + if unsafe { ShmWriter::new(&mapping, slots) }.is_none() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "the shared-memory capacity cannot hold that many slots", + )); + } + let conf = ChannelConf { - lock_file_path: lock_file_path.as_os_str().into(), shm_id: IpcStr::from_os_c_str(keeper.path.as_c_str()).to_boxed(), + slots: from_usize(slots), }; - let receiver = Receiver::new(lock_file_path, keeper, mapping)?; - Ok((conf, receiver)) + Ok((conf, Receiver { _keeper: keeper, mapping, slots })) } /// Encodes `path` as an owned NUL-terminated platform C string. @@ -139,155 +157,221 @@ impl Drop for ShmKeeper { } impl ChannelConf { - /// Creates a sender. + /// Creates a sender, or `None` when the channel is already over. /// - /// This doesn't block on the file lock. Instead it returns immediately with error if the receiver is locked or dropped. - #[expect( - clippy::missing_errors_doc, - reason = "error conditions are self-evident from return type" - )] - pub fn sender(&self) -> io::Result { - let lock_file = File::open(self.lock_file_path.to_cow_os_str())?; - lock_file.try_lock_shared()?; - + /// Never blocks. `None` means the receiver removed the backing file, + /// or sealed the region before removing it and this call caught the + /// gate in between. Either way whatever the caller does next happens + /// past the receiver's boundary, so recording nothing loses nothing. + /// + /// # Panics + /// + /// When the channel is there but cannot be attached to: its path is + /// unreadable, the file refuses to open or map, or the region cannot + /// hold the protocol. A process with no sender has no way to tell the + /// receiver it recorded nothing, and a trace that silently omits every + /// access a process made is worse than no trace, so it stops here. + #[must_use] + pub fn sender(&self) -> Option { // The arena never touches the process heap, so this stays safe in // the preload contexts that create senders (pre-`main` constructors, // the Windows loader lock). let arena = fspy_nostd_alloc::arena(); - let shm_path = self.shm_id.to_os_c_string_in(&arena).ok_or_else(|| { - io::Error::new(io::ErrorKind::InvalidData, "invalid shared-memory path") - })?; - let mapping = fspy_shm::open(shm_path.as_c_str().as_thin()) - .map_err(shm_error_to_io)? - .map() - .map_err(shm_error_to_io)?; - // SAFETY: `mapping` is a freshly mapped shared memory region with valid - // pointer and size. Exclusive write access is ensured by the shared - // file lock held by this sender. - let writer = unsafe { ShmWriter::new(mapping) }; - Ok(Sender { writer, lock_file, lock_file_path: self.lock_file_path.clone() }) + let shm_path = self + .shm_id + .to_os_c_string_in(&arena) + .expect("the channel's shared-memory path is not a valid C string"); + let mapping = match fspy_shm::open(shm_path.as_c_str().as_thin()) { + Ok(handle) => handle.map().expect("cannot map the shared-memory channel"), + Err(error) => { + let error = shm_error_to_io(error); + // The receiver removed the backing file, so it has already + // stopped collecting. + if error.kind() == io::ErrorKind::NotFound { + return None; + } + panic!("cannot open the shared-memory channel: {error}"); + } + }; + // SAFETY: `mapping` is a freshly mapped shared memory region created + // zero-initialized by `channel` and accessed only through the + // `shm_io` protocol by every attached process. + let writer = unsafe { ShmWriter::new(mapping, to_usize(self.slots)) } + .expect("the shared-memory region cannot hold the channel"); + // The receiver sealed the region but has not removed it yet. + if writer.is_closed() { + return None; + } + Some(Sender { writer }) } } pub struct Sender { writer: ShmWriter, - lock_file_path: Box, - lock_file: File, -} - -impl Drop for Sender { - fn drop(&mut self) { - if let Err(err) = self.lock_file.unlock() { - let lock_file_path = self.lock_file_path.to_cow_os_str(); - debug!("Failed to unlock the shared IPC lock {}: {}", lock_file_path.display(), err); - } - } } -impl Deref for Sender { - type Target = ShmWriter; - - fn deref(&self) -> &Self::Target { - &self.writer +impl Sender { + /// Serializes one record into a committed frame. + /// + /// A claim the channel refuses is skipped, because that is all a sender + /// inside an intercepted call can do: the channel has closed, so the + /// record belongs past the receiver's boundary, or the region is full + /// and the failed claim already set the CLOSED gate to say so. + /// + /// # Panics + /// + /// When the record's serialized size disagrees with the bytes it then + /// writes. Nothing the caller passes can cause that, so it is a defect + /// in this crate or its codec, and a trace built on it would be wrong + /// in ways the receiver cannot see. + pub fn send>(&self, value: &T) { + let serialized_size = + T::serialized_size(value).expect("a record cannot report its serialized size"); + let frame_size = usize::try_from(serialized_size) + .ok() + .and_then(NonZeroUsize::new) + .expect("a record reports a serialized size of zero, or one no frame could hold"); + let Ok(mut frame) = self.writer.claim_frame(frame_size) else { + return; + }; + let mut buf: &mut [u8] = &mut frame; + T::serialize_into(&mut buf, value).expect("a record will not serialize into its own frame"); + assert!(buf.is_empty(), "a record wrote fewer bytes than the size it reported"); + frame.finish(); } } -/// SAFETY: `Sender` holds a shared file lock that ensures there's no reader, so `shm` can be safely written to. +// SAFETY: `Sender` only accesses the shared mapping through the `shm_io` +// protocol, which synchronizes concurrent writers and the receiver with +// atomic operations; the mapping's address is stable and independently owned. 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. +// SAFETY: see the `Send` impl; `ShmWriter`'s shared-reference API is +// internally synchronized by the protocol. unsafe impl Sync for Sender {} /// The unique receiver side of an IPC channel. -/// Owns the lock file and removes it on drop. +/// +/// Holds the shared memory and its backing file alive for as long as senders +/// may attach; [`Receiver::close`] (or dropping) removes the backing file. pub struct Receiver { - lock_file_path: PathBuf, - lock_file: File, /// Keeps the shared memory's backing file alive for as long as senders /// may attach. _keeper: ShmKeeper, mapping: Mapping, + /// The slot count the region was created with, needed again to seal it. + slots: usize, } -/// SAFETY: `Receiver` doesn't read or write `shm`. It only passes it to `ReceiverLockGuard` under the lock. +// SAFETY: `Receiver` only holds the mapping; it accesses it exclusively +// through the `shm_io` protocol in `close`, which synchronizes with senders +// via atomic operations. The mapping's address is stable and independently +// owned. unsafe impl Send for Receiver {} -/// SAFETY: `Receiver` doesn't read or write `shm`. It only passes it to `ReceiverLockGuard` under the lock. +// SAFETY: see the `Send` impl. 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.display(), err); - } - } -} - impl Receiver { - fn new(lock_file_path: PathBuf, keeper: ShmKeeper, mapping: Mapping) -> io::Result { - let lock_file = File::create(&lock_file_path)?; - Ok(Self { lock_file_path, lock_file, _keeper: keeper, mapping }) - } - - /// 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). - #[expect( - clippy::missing_errors_doc, - reason = "error conditions are self-evident from return type" - )] - pub fn lock(&self) -> io::Result> { - self.lock_file.lock()?; - // SAFETY: The exclusive file lock is held, so no writers can access the shared memory. - // The lock ensures all prior writes are visible to this thread. - let reader = ShmReader::new(unsafe { self.mapping.as_slice() }); - Ok(ReceiverLockGuard { reader, lock_file: &self.lock_file }) - } -} - -pub struct ReceiverLockGuard<'a> { - reader: ShmReader<&'a [u8]>, - lock_file: &'a File, -} - -impl Drop for ReceiverLockGuard<'_> { - fn drop(&mut self) { - if let Err(err) = self.lock_file.unlock() { - debug!("Failed to unlock IPC lock file: {}", err); + /// Closes the channel and returns every committed frame, borrowed from + /// the shared mapping that moves into the returned [`FrameReader`]. + /// + /// Never blocks on senders: it reads the claim counter once, which + /// fixes how far reading goes, and shuts the gate so no later claim + /// succeeds. Committed frames become readable in place. A sender that + /// was still filling a frame keeps running, and its frame may or may + /// not appear depending on whether it commits before the read reaches + /// that slot — either way the operation it describes happens after the + /// receiver stopped collecting. The mapping is released when the + /// returned [`FrameReader`] drops. + /// + /// # Errors + /// + /// [`RecordsLost`] when a sender could not record something it went on + /// to do. There is no complete set of records then, so none are handed + /// back, and a caller that needs the trace has to treat the run as + /// untracked rather than as having reported nothing. + /// + /// # Panics + /// + /// When the region cannot hold the protocol, which [`channel`] proved + /// it could before any sender saw it. + pub fn close(self) -> Result { + let Self { _keeper: keeper, mapping, slots } = self; + // SAFETY: `mapping` was created zero-initialized by `channel`, its + // address is stable and independently owned, and all attached + // processes access it only through the `shm_io` protocol. + let sealed = unsafe { ShmReader::seal(mapping, slots) }; + // Remove the backing file only after the gate is shut. A process + // that attaches in between finds a closed channel and gives up + // cleanly; one that found the file already gone could not attach at + // all, and so could not report whatever it then failed to record. + drop(keeper); + match sealed { + Ok(reader) => Ok(reader), + // This receiver is the only one that could have sealed, and it + // is gone by now, so the gate can only be a sender's report. + Err(SealError::Closed) => Err(RecordsLost), + Err(SealError::UnsupportedRegion) => { + panic!("the shared-memory region cannot hold the channel") + } } } } -impl<'a> Deref for ReceiverLockGuard<'a> { - type Target = ShmReader<&'a [u8]>; - fn deref(&self) -> &Self::Target { - &self.reader - } -} +/// A sender could not record something, so the receiver has no complete +/// set of records to hand back. +/// +/// The region filled up, or a record came out longer than one frame can +/// hold. Both are the channel running out of room rather than anything the +/// senders did wrong, and both leave the run untracked. +#[derive(thiserror::Error, Clone, Copy, PartialEq, Eq, Debug)] +#[error("a sender ran out of room in the shared-memory channel")] +pub struct RecordsLost; #[cfg(test)] mod tests { use std::{ffi::OsString, fs, num::NonZeroUsize, str::from_utf8}; + use assert2::assert; use bstr::B; use subprocess_test::command_for_fn; use super::*; + use crate::ipc::{AccessMode, IpcPath, PathAccess}; + + /// A gibibyte of sparse address space, so its table costs nothing + /// until slots are touched. + const SIZE: ChannelSize = ChannelSize { capacity: 1 << 30, slots: 1 << 24 }; + + /// A size whose two halves do not fit has to fail here, at creation. + /// Everything downstream treats the region as able to host the + /// protocol: `sender` panics when it cannot, and so does + /// `Receiver::close`. + #[test] + fn a_capacity_too_small_for_the_table_fails_the_channel() { + // The counters alone need sixteen bytes, and each slot eight more. + let Err(error) = channel(ChannelSize { capacity: 8, slots: 1 }) else { + panic!("a region too small for the protocol made a channel"); + }; + assert!(error.kind() == io::ErrorKind::InvalidInput); + } /// The shared-memory path is generated absolute, so a sender in a process /// with a different working directory and a relative temporary directory /// must still attach. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn sender_ignores_changed_temp_and_working_directory() { - let (conf, receiver) = channel(100).unwrap(); + let (conf, receiver) = channel(SIZE).unwrap(); let changed_cwd = temp_dir().join(format!("fspy-ipc-changed-cwd-{}", Uuid::new_v4())); fs::create_dir(&changed_cwd).unwrap(); let mut command = command_for_fn!(conf, |conf: ChannelConf| { let sender = conf.sender().unwrap(); let frame_size = NonZeroUsize::new(2).unwrap(); - let mut frame = sender.claim_frame(frame_size).unwrap(); + let mut frame = sender.writer.claim_frame(frame_size).unwrap(); frame.copy_from_slice(&[4, 2]); + frame.finish(); }); command.cwd = changed_cwd.clone(); for name in ["TMPDIR", "TMP", "TEMP"] { @@ -297,67 +381,126 @@ mod tests { fs::remove_dir(changed_cwd).unwrap(); assert!(succeeded); - let lock = receiver.lock().unwrap(); - assert_eq!(lock.iter_frames().next().unwrap(), &[4, 2]); + let frames = receiver.close().unwrap(); + assert!(frames.iter().next().unwrap() == &[4, 2]); + } + + /// `Sender::send` is the only writer production uses, and the rest of + /// these tests reach past it into `claim_frame`. This one drives it + /// end to end, so that a disagreement between `serialized_size` and + /// `serialize_into` — which would silently drop every record — fails + /// here rather than in a build. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn sender_round_trips_records() { + let (conf, receiver) = channel(SIZE).unwrap(); + let sender = conf.sender().unwrap(); + // A record path carries the platform's own string form: bytes on + // unix, UTF-16 on Windows. + #[cfg(unix)] + let owned = ["/tmp/one", "/tmp/two/three"]; + #[cfg(windows)] + let owned = [r"C:\tmp\one", r"C:\tmp\two\three"] + .map(|path| path.encode_utf16().collect::>()); + #[cfg(unix)] + let paths = owned.map(<&IpcPath>::from); + #[cfg(windows)] + let paths = [IpcPath::from_wide(&owned[0]), IpcPath::from_wide(&owned[1])]; + + for path in paths { + sender.send(&PathAccess::read(path)); + } + drop(sender); + + let frames = receiver.close().unwrap(); + let mut iter = frames.iter(); + for path in paths { + let access: PathAccess<'_> = wincode::deserialize_exact(iter.next().unwrap()).unwrap(); + assert!(access.path == path); + assert!(access.mode == AccessMode::READ); + } + assert!(iter.next().is_none()); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn smoke() { - let (conf, receiver) = channel(100).unwrap(); + let (conf, receiver) = channel(SIZE).unwrap(); let cmd = command_for_fn!(conf, |conf: ChannelConf| { let sender = conf.sender().unwrap(); let frame_size = NonZeroUsize::new(2).unwrap(); - let mut frame = sender.claim_frame(frame_size).unwrap(); + let mut frame = sender.writer.claim_frame(frame_size).unwrap(); frame.copy_from_slice(&[4, 2]); + frame.finish(); }); assert!(std::process::Command::from(cmd).status().unwrap().success()); - let lock = receiver.lock().unwrap(); - let mut frames = lock.iter_frames(); + let frames = receiver.close().unwrap(); + let mut iter = frames.iter(); - let received_frame = frames.next().unwrap(); - assert_eq!(received_frame, &[4, 2]); + let received_frame = iter.next().unwrap(); + assert!(received_frame == &[4, 2]); - assert!(frames.next().is_none()); + assert!(iter.next().is_none()); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[expect(clippy::print_stdout, reason = "test diagnostics")] - async fn forbid_new_senders_after_locked() { - let (conf, receiver) = channel(42).unwrap(); - let _lock = receiver.lock().unwrap(); + async fn forbid_new_senders_after_close() { + let (conf, receiver) = channel(SIZE).unwrap(); + let _frames = receiver.close().unwrap(); let cmd = command_for_fn!(conf, |conf: ChannelConf| { - print!("{}", conf.sender().is_ok()); + print!("{}", conf.sender().is_some()); }); let output = std::process::Command::from(cmd).output().unwrap(); - assert_eq!(B(&output.stdout), B("false")); + assert!(B(&output.stdout) == B("false")); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[expect(clippy::print_stdout, reason = "test diagnostics")] async fn forbid_new_senders_after_receiver_dropped() { - let (conf, receiver) = channel(42).unwrap(); + let (conf, receiver) = channel(SIZE).unwrap(); drop(receiver); let cmd = command_for_fn!(conf, |conf: ChannelConf| { - print!("{}", conf.sender().is_ok()); + print!("{}", conf.sender().is_some()); }); let output = std::process::Command::from(cmd).output().unwrap(); - assert_eq!(B(&output.stdout), B("false")); + assert!(B(&output.stdout) == B("false")); + } + + /// A sender that attached before close keeps its mapping but cannot + /// claim any new frame afterwards. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn attached_sender_cannot_claim_after_close() { + let (conf, receiver) = channel(SIZE).unwrap(); + let sender = conf.sender().unwrap(); + + let mut frame = sender.writer.claim_frame(NonZeroUsize::new(2).unwrap()).unwrap(); + frame.copy_from_slice(&[4, 2]); + frame.finish(); + + let frames = receiver.close().unwrap(); + assert!(frames.iter().next().unwrap() == &[4, 2]); + + assert!( + sender.writer.claim_frame(NonZeroUsize::new(2).unwrap()).unwrap_err() + == shm_io::ClaimError::Closed + ); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn concurrent_senders() { - let (conf, receiver) = channel(8192).unwrap(); + let (conf, receiver) = channel(SIZE).unwrap(); for i in 0u16..200 { let cmd = command_for_fn!((conf.clone(), i), |(conf, i): (ChannelConf, u16)| { let sender = conf.sender().unwrap(); let data_to_send = i.to_string(); - sender + let mut frame = sender + .writer .claim_frame(NonZeroUsize::new(data_to_send.len()).unwrap()) - .unwrap() - .copy_from_slice(data_to_send.as_bytes()); + .unwrap(); + frame.copy_from_slice(data_to_send.as_bytes()); + frame.finish(); }); let output = std::process::Command::from(cmd).output().unwrap(); assert!( @@ -367,12 +510,10 @@ mod tests { 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(); + let frames = receiver.close().unwrap(); + let mut received_values: Vec = + frames.iter().map(|frame| from_utf8(frame).unwrap().parse::().unwrap()).collect(); received_values.sort_unstable(); - assert_eq!(received_values, (0u16..200).collect::>()); + assert!(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 deleted file mode 100644 index 916df30d7..000000000 --- a/crates/fspy_shared/src/ipc/channel/shm_io.rs +++ /dev/null @@ -1,727 +0,0 @@ -//! 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 fspy_shm::Mapping; -use wincode::{SchemaWrite, Serialize as _, config::DefaultConfig}; - -// `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 Mapping { - 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); -} - -const 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.is_multiple_of(FRAME_HEADER_ALIGN) { - 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); - } -} - -#[derive(thiserror::Error, Debug)] -pub enum WriteEncodedError { - #[error("Failed to encode value into shared memory")] - EncodeError(#[from] wincode::error::WriteError), - #[error("Tried to write a frame of zero size into shared memory")] - ZeroSizedFrame, - #[error("Not enough space in shared memory to write the encoded frame")] - InsufficientSpace, -} - -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)] - const 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.cast::(); - 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) - // SAFETY: `shm_ptr` points to the start of the shared memory region, which is properly - // aligned to `usize` (verified by `assert_alignment` in `new`), and the allocation is - // large enough to contain at least a `usize` header. - 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.try_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 - - // SAFETY: The atomic try_update above guaranteed that `size_of::() + current_end` - // is within the shared memory bounds, so this pointer arithmetic stays within the allocation. - let frame_start = unsafe { - shm_ptr.add(/* shm header */ size_of::() + current_end) - }; - - // SAFETY: `frame_start` is properly aligned to `i32` (ensured by `roundup_to_align_frame_header`) - // and points within the shared memory allocation (bounds checked by the atomic try_update). - 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); - - // SAFETY: `frame_start` is within bounds and adding `size_of::()` skips the frame - // header to reach the content area, which is still within the claimed space. - let frame_content_ptr = unsafe { frame_start.add(size_of::()) }; // skip the frame header - Some(FrameMut { - header: frame_header, - // SAFETY: `frame_content_ptr` is valid for `frame_size` bytes (guaranteed by the - // atomic space claim), properly aligned for `u8`, and no other writer will access - // this region because each writer atomically claims a unique range. - content: unsafe { std::slice::from_raw_parts_mut(frame_content_ptr, frame_size) }, - }) - } - - /// Append an encoded value into the shared memory. - pub fn write_encoded>( - &self, - value: &T, - ) -> Result<(), WriteEncodedError> { - let serialized_size = - usize::try_from(T::serialized_size(value)?).expect("serialized size exceeds usize"); - - let Some(frame_size) = NonZeroUsize::new(serialized_size) else { - return Err(WriteEncodedError::ZeroSizedFrame); - }; - let Some(mut frame) = self.claim_frame(frame_size) else { - return Err(WriteEncodedError::InsufficientSpace); - }; - - let mut writer: &mut [u8] = &mut frame; - T::serialize_into(&mut writer, value)?; - assert_eq!(writer.len(), 0); - - Ok(()) - } - - #[cfg(test)] - pub fn try_write_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 - } - 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)..]; - } - ..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::{ - process::{Child, Command}, - sync::Arc, - thread, - }; - - use assert2::assert; - use bstr::BStr; - use rustc_hash::FxHashSet; - - 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, - } - // SAFETY: `MockedShm` uses `Arc>` for its backing memory, which is safe to send - // across threads. The raw pointer access through `AsRawSlice` is synchronized by `ShmWriter`'s - // atomic operations. - unsafe impl Send for MockedShm {} - // SAFETY: Concurrent access to the shared memory is synchronized by `ShmWriter`'s atomic - // operations. The `Arc` wrapper ensures the allocation remains valid. - 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 = std::iter::repeat_n(0usize, size_in_usize).collect(); - - Self { mem: Arc::new(mem), len } - } - } - impl AsRef<[u8]> for MockedShm { - fn as_ref(&self) -> &[u8] { - // SAFETY: `Vec::as_ptr` returns a valid pointer to the vec's buffer. The vec is - // allocated with enough `usize` elements to cover `self.len` bytes, and the pointer - // is valid for reads of `self.len` bytes. The `Arc` ensures the allocation is alive. - 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() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - assert!(writer.try_write_frame(b"hello")); - assert!(writer.try_write_frame(b"world")); - assert!(writer.try_write_frame(b"this is a test")); - assert!(!writer.try_write_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() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - assert!(writer.try_write_frame(b"hello")); - assert!(!writer.try_write_frame(b"")); - assert!(writer.try_write_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() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let mut writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - assert!(writer.try_write_frame(b"foo")); - - // Simulate crash during writing - writer.set_fail_on_claim(true); - assert!(!writer.try_write_frame(b"hello")); - - writer.set_fail_on_claim(false); - assert!(writer.try_write_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() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - assert!(writer.try_write_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.try_write_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. - - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let mut writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - - assert!(writer.try_write_frame(b"foo")); - - // First crash: AfterClaim (leaves frame header as 0) - writer.set_fail_on_claim(true); - assert!(!writer.try_write_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.try_write_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() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - 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.try_write_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.try_write_frame(b"world")); - writer.set_fail_on_claim(false); - - assert!(writer.try_write_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(|| { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized - // allocation. The clone shares the same backing memory, which is safe because - // `ShmWriter` uses atomic operations for concurrent access. - let writer = unsafe { ShmWriter::new(shm.clone()) }; - for _ in 0..10 { - assert!(writer.try_write_frame(b"hello")); - assert!(writer.try_write_frame(b"foo")); - assert!(writer.try_write_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() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - thread::scope(|s| { - for _ in 0..4 { - s.spawn(|| { - for _ in 0..10 { - writer.try_write_frame(b"hello"); - writer.try_write_frame(b"foo"); - writer.try_write_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 - - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - 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.try_write_frame(&large_frame)); - - // Small frame should still work - assert!(writer.try_write_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 - - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - 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 - .try_write_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( - // SAFETY: Adding 1 byte to create a deliberately misaligned pointer for testing. - // The original allocation is large enough that adding 1 byte stays within bounds. - 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(|| { - // SAFETY: Intentionally passing a misaligned pointer to test that the alignment - // assertion in `ShmWriter::new` correctly panics. This is expected to panic. - 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 subprocess_test::command_for_fn; - - const CHILD_COUNT: usize = 12; - const FRAME_COUNT_EACH_CHILD: usize = 100; - - const SHM_SIZE: usize = 1024 * 1024; - - let shm_path = crate::ipc::channel::shm_backing_path().unwrap(); - let shm_name = shm_path.to_str().expect("test temp dir is UTF-8").to_owned(); - let c_path = crate::ipc::channel::os_c_string(shm_path.as_os_str()).unwrap(); - let handle = fspy_shm::create(c_path.as_c_str().as_thin(), SHM_SIZE).unwrap(); - let _keeper = crate::ipc::channel::ShmKeeper { path: c_path }; - // Map before the children run. Windows keeps views coherent while they - // exist at the same time; a view created after every writer exited can - // observe the file before the writers' dirty pages reach it. - let mapping = handle.map().unwrap(); - - let children: Vec = (0..CHILD_COUNT) - .map(|child_index| { - let cmd = command_for_fn!( - (shm_name.clone(), child_index), - |(shm_name, child_index): (String, usize)| { - let c_path = - crate::ipc::channel::os_c_string(std::ffi::OsStr::new(&shm_name)) - .unwrap(); - let mapping = - fspy_shm::open(c_path.as_c_str().as_thin()).unwrap().map().unwrap(); - // SAFETY: `mapping` is a freshly mapped shared memory region with a - // valid pointer and size. Concurrent write access is safe because - // `ShmWriter` uses atomic operations. - let writer = unsafe { ShmWriter::new(mapping) }; - for i in 0..FRAME_COUNT_EACH_CHILD { - let frame_data = std::format!("{child_index} {i}"); - assert!(writer.try_write_frame(frame_data.as_bytes())); - } - } - ); - Command::from(cmd).spawn().unwrap() - }) - .collect(); - - for mut c in children { - let status = c.wait().unwrap(); - assert!(status.success()); - } - - // SAFETY: All child processes have exited (waited above), so no concurrent writers exist. - // The shared memory is valid and fully written. - let shm = unsafe { mapping.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 { - let frame_data = format!("{child_index} {i}"); - assert!(frames.contains(&BStr::new(frame_data.as_bytes()))); - } - } - } -} diff --git a/crates/fspy_shared/src/ipc/channel/shm_io/README.md b/crates/fspy_shared/src/ipc/channel/shm_io/README.md new file mode 100644 index 000000000..dae21e457 --- /dev/null +++ b/crates/fspy_shared/src/ipc/channel/shm_io/README.md @@ -0,0 +1,90 @@ +# shm_io: a crash-tolerant frame channel over shared memory + +One shared-memory region. Many writer processes append variable-length records; one receiver collects them once, when the channel's lifetime ends. + +Three requirements shape the design: + +1. **A writer may die at any instruction.** That must not corrupt the channel or cost another writer its records. +2. **A writer may outlive the channel.** The receiver never waits for one. +3. **The receiver must know whether it got everything.** Either it gets every record writers published, or sealing fails and it gets none. + +## The region + +Any zero-initialized shared memory works. In practice it is a sparse file mapped into every participating process, so only the pages someone writes cost real memory. + +```text +| counters | descriptor table (one slot per frame) | payloads (the rest, grow up) | +``` + +Two `AtomicU64` counters sit at the front, followed by one 8-byte descriptor slot per frame. + +- The **claim counter** counts frames ever claimed. Bit 63 is the CLOSED gate: the receiver sets it when it seals, and so does any writer whose claim failed, which is how the receiver hears about a lost record. +- The **payload counter** counts payload bytes ever reserved. + +The mapping's size and the number of slots are the whole geometry, and every process attaching to a region passes the slot count it was created with. Payloads take what the table leaves. Attaching checks one bound: the counters and the table fit, and what remains is small enough for the descriptors' 32-bit offsets. Splitting 4 GiB at one slot per 64 bytes, say, gives ~67 million slots and ~3.5 GiB of payload room, which is tens of millions of records of a few hundred bytes; payload space runs out first. + +Where the table ends and payloads begin never moves. Claiming needs no retry loop, because a writer checks each counter against a fixed limit using the value `fetch_add` returned. Overshooting a limit costs nothing: no counter says where data is, since every committed descriptor carries its own offset and length. + +## Writing a frame + +1. **Claim.** Two `fetch_add`s reserve payload bytes and a slot. No retry loop, no lock. A claim that does not fit fails after the fact, and sets the CLOSED gate before the writer moves on. +2. **Fill.** The writer serializes into its payload span, which nobody else knows exists. +3. **Commit.** One store puts the payload's offset and length into the slot. Only the writer that claimed a slot ever writes it, so it needs no compare-and-swap. The receiver cannot see the frame before that store, and nobody touches the payload after it. + +`FrameMut::finish` commits. What happens when it never runs is the heart of the design: + +- **The process died,** mid-claim or mid-fill. The slot stays zero and the receiver ignores it. No cleanup code runs, because none exists. +- **The process abandoned the frame** and kept going. The slot stays zero and the receiver ignores that too, since it cannot tell the two apart. + +So the channel asks one thing of its users: **publish a record before performing the action it describes.** A dead writer's missing record then describes an action that never happened, and a record refused after the seal describes one performed after the channel closed. The receiver drops both. A writer that records after acting, or that abandons a frame and acts anyway, breaks the rule and loses records with nothing said. The channel cannot see either one. + +## When the region fills up + +A 4 GiB region holds tens of millions of records, but not endless ones. When a claim asks for more room than the payload area or the table has left, it fails. The writer skips that record and carries on, because recording must never stop the program doing the work. + +The loss is not silent. The failed claim sets the CLOSED gate before it returns, and a receiver that finds that bit already set fails its seal and hands back nothing. + +Setting the bit first matters for the same reason publishing before acting does. If the receiver's read misses the bit, the writer set it after the seal, so the skipped record describes an action performed after the channel closed. And a writer that died before setting it never performed its action. + +Because the bit is also the gate, the first lost record closes the channel and every later claim is refused. Those records would only pile up in a result nobody can use. + +A single frame holds at most `u32::MAX` bytes, since a descriptor cannot describe more. Such a claim is refused and reported the same way. + +## Sealing and reading + +Sealing swaps the CLOSED gate into the claim counter and reads the old value. That one operation draws the boundary and shuts the gate: claims at or before it are in, every later one fails. If the bit was already set, a record was lost or someone sealed earlier, and sealing fails here. + +No slot is touched, so sealing a channel holding ten million frames costs what sealing an empty one costs. + +`ShmReader` owns the mapping and lends out one `&[u8]` per committed span, straight from shared memory. It reads the table when asked, so a writer still filling a frame when the boundary was drawn may appear in a later read and not an earlier one. Dropping the reader releases the mapping. + +## Why this is sound + +- Finding frames never reads payload bytes, and every slot sits at a fixed place, so a half-written payload cannot be mistaken for metadata. +- The receiver reaches a payload only through its committed descriptor. The writer commits with `Release` and the receiver loads with `Acquire`, so a descriptor the receiver sees brings its payload bytes along. +- One writer owns each slot and writes it once. A slot goes from zero to committed and stops. +- No counter has to be exact. Both only climb, refused claims leave their increments behind, and the receiver clamps its snapshot to the table length, so a scribbled counter costs it a walk over empty slots. +- The receiver builds its borrows from a committed descriptor without re-checking it. That trusts the other processes to follow the protocol; one that scribbles random memory is outside the model. It is also why the receiver reads frames straight out of shared memory, with no copies and no checksums. + +## Deployment note + +On Linux, the first touch of the sparse backing file costs a millisecond or two on journalling filesystems. It is the fault path rather than block allocation, so `fallocate` does not help. Whichever side touches the region first pays it, once per channel. A filesystem that does not journal avoids it. + +## Files + +| File | Role | +| ----------- | ------------------------------------------------------------------------------------------------------------- | +| `mod.rs` | Public surface, and integration tests over a mocked region (`cargo miri test -p fspy_shared shm_io`). | +| `writer.rs` | Claim a frame, fill it, finish it. | +| `reader.rs` | Seal the channel, then iterate committed frames, with the argument for why its borrows hold. | +| `layout.rs` | What both sides share: the `repr(C)` shape, the descriptor format, the ordering contract, and `MappedLayout`. | + +```mermaid +graph TD + mod["mod.rs
public surface"] --> writer["writer.rs
claim, fill, finish"] + mod --> reader["reader.rs
seal and iterate"] + writer --> layout["layout.rs
what both sides share"] + reader --> layout +``` + +Each file needs only the ones below it, so read from `layout.rs` upward. diff --git a/crates/fspy_shared/src/ipc/channel/shm_io/layout.rs b/crates/fspy_shared/src/ipc/channel/shm_io/layout.rs new file mode 100644 index 000000000..c5278aa5f --- /dev/null +++ b/crates/fspy_shared/src/ipc/channel/shm_io/layout.rs @@ -0,0 +1,321 @@ +//! What the writer and reader sides share: the region's `repr(C)` shape, +//! the descriptor format, and [`MappedLayout`], which locates the parts of +//! one mapping. The sides live in [`super::writer`] and [`super::reader`]; +//! `README.md` describes the region itself. +//! +//! The payload area never exceeds `u32::MAX` bytes, checked at attach, so +//! an offset and a length always fit a descriptor's 32-bit fields and +//! bounds checks add them in 32 bits. + +use std::{num::NonZeroU32, ptr::NonNull, sync::atomic::AtomicU64}; + +/// The CLOSED gate bit of the claim counter. The receiver sets it when it +/// seals, and so does a claim the region had no room for, which is how it +/// reports the loss (rule 1). Nothing else sets it: what a writer cannot +/// record for its own reasons is not this protocol's to report. +/// +/// A bit rather than a value to compare against, so it survives the +/// increment of a writer that arrives late. Counting cannot reach it: that +/// takes 2^63 claims, and a channel lives for one command. A gate that +/// somehow read as set would fail the seal, which is the cautious +/// answer. +pub const CLOSED: u64 = 1 << 63; + +/// The two counters at the start of the region, as one `repr(C)` struct +/// that starts zeroed. The descriptor table follows them, and payloads +/// take the rest of the mapping. +#[repr(C)] +pub struct Counters { + /// Bit 63 is the CLOSED gate. The low bits count claims that got as + /// far as reserving payload space, which is where a slot is taken. + pub claims: AtomicU64, + /// Payload bytes ever reserved, failed claims included. + pub payload_reserved: AtomicU64, +} + +// The mapping starts at a `u64`-aligned address and is cast to +// `&Counters`, so no field in it may need more alignment than that. This +// is also what makes the check on the start address serve both parts: an +// address aligned for the counters is aligned for a slot. +const _: () = assert!(align_of::() == align_of::()); + +// The descriptor table starts right after the counters, so their size has +// to leave it aligned: an address good for a slot must still be good for +// one this many bytes later. +const _: () = assert!(size_of::().is_multiple_of(align_of::())); + +// Both endpoints keep raw pointers into the region and read through them +// long after the references they came from are gone, which is sound only +// because every byte they point at sits inside an atomic. This catches a +// field being added or padding appearing; it cannot catch a field changing +// type, so keep that in mind when editing the struct. +const _: () = assert!(size_of::() == 2 * size_of::()); + +// --- The descriptor slot codec --------------------------------------------- +// +// One slot is a 64-bit value that publishes a frame: +// +// ```text +// bits 32..=63 bits 0..=31 +// payload length (32) payload offset (32) +// ``` +// +// | Value | State | +// | --------------------- | ----------------------------------------------- | +// | `0` | Unfinished: slot claimed, nothing published yet | +// | length field nonzero | Committed: offset and length of the payload | +// +// A claimed frame is never zero-length, so a committed value always has a +// nonzero length field and can never be read as the zero a fresh slot +// starts at. One writer owns each slot and writes it once, so a slot goes +// from zero to committed and never changes again. +// +// Offsets are counted from the start of the payload area, so a descriptor +// can never point into the counters or the table. + +// --- MappedLayout and the ordering contract -------------------------------- +// `MappedLayout::new` works out where each part sits once, at attach, and +// the endpoint keeps the result beside the mapping. The payload pointer +// stays raw because writers hand out `&mut` slices into it, which must not +// overlap a shared reference. +// +// Neither counter guards its add against wrapping, because neither wrap is +// reachable. The payload counter passes the region only after a claim has +// failed, and every claim after that is refused before it builds a span. +// The claim count needs 2^63 increments to reach the gate bit. +// +// # Memory-ordering contract +// +// 1. **Claim versus seal.** The seal swaps the gate into the claim counter +// and reads the old value in one step, so the boundary and the gate are +// one point in that counter's modification order: claims at or before +// it are in, every later one fails on the gate. Claims publish no +// payload data, so `Relaxed` suffices. Completeness rides the same +// order: a failed claim sets the gate before performing the operation +// whose record it lost, so either the seal sees the bit or the loss +// happened past the boundary. A writer that died before setting it +// never performed its operation. +// 2. **Writer commit.** `FrameMut::finish` stores the descriptor with +// `Release`, so every payload write lands first. The writer that +// claimed the slot is the only one that writes it, so a store is +// enough. +// 3. **Receiver read.** `Iter` loads each descriptor with `Acquire`, so a +// descriptor it sees brings the payload bytes along. + +/// Requires the target's `usize` to be as wide as a `u64`, which is what +/// makes [`to_usize`] and [`from_usize`] lossless. Stated once, and cited +/// by both. +const EQUAL_WIDTHS: () = + assert!(size_of::() == size_of::(), "requires a 64-bit target"); + +/// Converts an integer into a `usize`. +/// +/// Never loses bits: the bound takes only what fits a `u64`, and +/// [`EQUAL_WIDTHS`] lets only targets whose `usize` is that wide build +/// this module. Those asserts are why the casts are safe, so the pair sits +/// here rather than at module scope. They are the only `as` in the +/// protocol. +#[expect(clippy::cast_possible_truncation, reason = "the assert allows only equal widths")] +pub fn to_usize(value: impl Into) -> usize { + const { EQUAL_WIDTHS }; + value.into() as usize +} + +/// Converts a `usize` into a `u64`; the inverse of [`to_usize`]. +#[expect(clippy::as_conversions, reason = "the assert allows only equal widths")] +pub const fn from_usize(value: usize) -> u64 { + const { EQUAL_WIDTHS }; + value as u64 +} + +/// Casts to a pointer of another type, returning `None` when the pointer +/// is not aligned for `U`: a stable stand-in for the still-unstable +/// [`<*mut T>::try_cast_aligned`][std]. +/// +/// [std]: https://doc.rust-lang.org/std/primitive.pointer.html#method.try_cast_aligned +fn try_cast_aligned(ptr: *mut T) -> Option<*mut U> { + if ptr.addr().is_multiple_of(align_of::()) { Some(ptr.cast()) } else { None } +} + +/// Where the counters, the descriptor table and the payload area sit in +/// one mapping, worked out once at attach. The pointers outlive this +/// value, since they point into the mapping rather than into the endpoint +/// holding them. +#[derive(Clone, Copy)] +pub struct MappedLayout { + counters: NonNull, + /// Every descriptor slot the region was created with. + table: NonNull<[AtomicU64]>, + /// Start of the payload area, raw rather than a reference: writers + /// hand out `&mut` slices into it, which must not overlap a shared + /// reference. + pub payload_start: NonNull, + /// Length of the payload area, in the width of a descriptor's offset, + /// so the writer's bounds check needs no conversion. + pub payload_len: u32, +} + +/// A decoded descriptor slot (the codec above). +#[derive(Clone, Copy)] +pub enum SlotState { + /// Nothing is published in the slot: the writer has not finished it + /// yet, or died or gave it up before finishing. The receiver ignores + /// such slots. + Unfinished, + /// A payload is committed: the receiver may read its span, once + /// checked against the payload area's bounds. + Committed { + /// Byte offset of the payload from the start of the payload region. + offset: u32, + /// Byte length of the payload. Nonzero, so a committed value is + /// never the zero a fresh slot starts at, which the offset alone + /// could be. + len: NonZeroU32, + }, +} + +impl SlotState { + /// Decodes a slot value into its state. The two processes sharing a + /// value run on one machine, so native byte order is fine. + pub const fn decode(slot_value: u64) -> Self { + let [offset, len] = bytemuck::must_cast::(slot_value); + let Some(len) = NonZeroU32::new(len) else { + return Self::Unfinished; + }; + Self::Committed { offset, len } + } + + /// Encodes this state as a slot value; the inverse of [`Self::decode`]. + pub const fn encode(self) -> u64 { + let [offset, len] = match self { + Self::Unfinished => [0, 0], + Self::Committed { offset, len } => [offset, len.get()], + }; + bytemuck::must_cast([offset, len]) + } +} + +impl MappedLayout { + /// Locates the parts of a shared mapping whose table holds `slots` + /// descriptors, or returns `None` when the mapping cannot hold the + /// protocol: a null or misaligned start, too little room for the + /// counters and that many slots, or a payload area too long for a + /// 32-bit offset to reach. + /// + /// Both endpoints must pass the `slots` the region was created with. + /// A smaller one reads part of the table as payload; a larger one + /// reads payload bytes as descriptors. + /// + /// # Safety + /// + /// - `mem` must be valid for reads and writes, and its address stable, + /// for as long as the returned pointers (and any copy of them) are + /// used. + /// - The memory must have been zero-initialized when the region was + /// created, and accessed only through this protocol since. + pub unsafe fn new(mem: *mut [u8], slots: usize) -> Option { + let mem_start = mem.cast::(); + // The mapping must hold the counters and the whole table. Their + // sizes together are where payloads begin. + let table_bytes = slots.checked_mul(size_of::())?; + let payloads_at = size_of::().checked_add(table_bytes)?; + let payload_len = mem.len().checked_sub(payloads_at)?; + // A descriptor holds a 32-bit offset, so the payload area can be + // no longer than a `u32`. Keeping the converted value is what lets + // later bounds checks stay in 32 bits. + let payload_len = u32::try_from(payload_len).ok()?; + // These two conversions are the checks on the start address: + // aligned for `Counters`, and not null. + let counters = NonNull::new(try_cast_aligned::<_, Counters>(mem_start)?)?; + + // The table sits right after the counters, and the payload area + // after the table. + // SAFETY: the mapping holds both, checked above. The slots are + // aligned because the start address is (the conversion above) and + // the counters are a whole number of slots wide (the assert near + // `Counters`). Payload bytes need no alignment. + let table_start = NonNull::new(unsafe { mem_start.add(size_of::()) })?; + let table = NonNull::slice_from_raw_parts(table_start.cast::(), slots); + // SAFETY: as above. + let payload_start = NonNull::new(unsafe { mem_start.add(payloads_at) })?; + Some(Self { counters, table, payload_start, payload_len }) + } + + /// The counters at the start of the region. + const fn counters(&self) -> &Counters { + // SAFETY: `new`'s contract keeps the memory valid while any + // pointer is used, and both counters are atomics, so the shared + // borrow is valid even while other threads and processes access + // the same memory through them. + unsafe { self.counters.as_ref() } + } + + /// The claim counter. + pub const fn claims(&self) -> &AtomicU64 { + &self.counters().claims + } + + /// The payload counter. + pub const fn payload_reserved(&self) -> &AtomicU64 { + &self.counters().payload_reserved + } + + /// The descriptor table. + pub const fn table(&self) -> &[AtomicU64] { + // SAFETY: see `counters`; the table is a run of atomics inside the + // same mapping. + unsafe { self.table.as_ref() } + } +} + +#[cfg(test)] +mod tests { + use assert2::assert; + + use super::*; + + #[test] + fn slot_codec_roundtrips() { + for (offset, len) in [(0, 5), (3, 5), (u32::MAX, 1), (0, u32::MAX)] { + let len = NonZeroU32::new(len).unwrap(); + let encoded = SlotState::Committed { offset, len }.encode(); + let SlotState::Committed { offset: o, len: l } = SlotState::decode(encoded) else { + panic!("committed value decoded as unfinished"); + }; + assert!(o == offset && l == len); + } + // Zero length fields publish nothing, whatever the offset half says. + assert!(matches!(SlotState::decode(0), SlotState::Unfinished)); + assert!(matches!(SlotState::decode(42), SlotState::Unfinished)); + } + + /// Every slot the reader and writer touch is dereferenced as an + /// `AtomicU64`, so `new`'s arithmetic has to land the table on that + /// alignment. The const asserts near `Counters` argue it; this checks + /// the pointers `new` actually builds. + #[test] + fn the_table_lands_aligned_for_a_slot() { + let mut mem = [0u64; 64]; + let raw = std::ptr::slice_from_raw_parts_mut(mem.as_mut_ptr().cast::(), 512); + for slots in 0..8 { + // SAFETY: the array is live, aligned and zeroed, and nothing + // else touches it. + let mapped = unsafe { MappedLayout::new(raw, slots) }.unwrap(); + for slot in mapped.table() { + assert!(std::ptr::from_ref(slot).addr().is_multiple_of(align_of::())); + } + } + } + + #[test] + fn a_region_too_small_for_the_table_is_rejected() { + let mut mem = [0u64; 4]; + let raw = std::ptr::slice_from_raw_parts_mut(mem.as_mut_ptr().cast::(), 32); + // Two counters and two slots exactly fill it; a third slot does not. + // SAFETY: the array is live, aligned and zeroed, and nothing else + // touches it. + assert!(unsafe { MappedLayout::new(raw, 2) }.is_some()); + // SAFETY: as above. + assert!(unsafe { MappedLayout::new(raw, 3) }.is_none()); + } +} diff --git a/crates/fspy_shared/src/ipc/channel/shm_io/mod.rs b/crates/fspy_shared/src/ipc/channel/shm_io/mod.rs new file mode 100644 index 000000000..e815eef46 --- /dev/null +++ b/crates/fspy_shared/src/ipc/channel/shm_io/mod.rs @@ -0,0 +1,685 @@ +//! A crash-tolerant, nonblocking frame channel in a shared memory region. +//! +//! Many writer processes append variable-length frames at once. One +//! receiver seals the channel and reads every committed frame without +//! waiting for any of them. A writer may die at any instruction and lose +//! only its own unfinished frame. No writer runs cleanup code, because +//! none exists: no exit hooks, PID checks, heartbeats, or timeouts. +//! +//! The channel asks one thing of its writers: **publish a record before +//! performing the action it describes.** A record that never arrives then +//! describes an action that never happened, and one refused after the seal +//! describes an action performed after the receiver stopped collecting. A +//! writer that gives up on a record and performs the action anyway breaks +//! the rule, and this protocol cannot tell that it did. +//! +//! `README.md` in this directory describes the region, the claim sequence, +//! and why the receiver can borrow frames out of shared memory. [`layout`] +//! carries the memory-ordering contract that the code cites by rule +//! number. + +mod layout; +mod reader; +mod writer; + +use std::ptr::slice_from_raw_parts_mut; + +use fspy_shm::Mapping; +pub use layout::{from_usize, to_usize}; +pub use reader::{SealError, ShmReader}; +// Only tests name a claim's failure; a sender skips the record either +// way, so production matches on `Ok`/`Err` alone. +#[cfg(test)] +pub use writer::ClaimError; +pub use writer::ShmWriter; + +/// A trait to borrow a raw memory region. +pub trait AsRawSlice { + fn as_raw_slice(&self) -> *mut [u8]; +} + +impl AsRawSlice for Mapping { + fn as_raw_slice(&self) -> *mut [u8] { + slice_from_raw_parts_mut(self.as_ptr(), self.len()) + } +} + +impl AsRawSlice for &M { + fn as_raw_slice(&self) -> *mut [u8] { + (**self).as_raw_slice() + } +} + +#[cfg(test)] +mod tests { + use std::{ + sync::{ + Arc, Barrier, + atomic::{AtomicU64, Ordering}, + }, + 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<[T]>: + // According to miri, from the perspective of data racing, incrementing + // the 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, + } + // SAFETY: `MockedShm` uses `Arc>` for its backing memory, which + // is safe to send across threads. The raw pointer access through + // `AsRawSlice` is synchronized by the protocol's atomic operations. + unsafe impl Send for MockedShm {} + // SAFETY: Concurrent access to the shared memory is synchronized by the + // protocol's atomic operations. The `Arc` keeps the allocation alive. + 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 = std::iter::repeat_n(0usize, size_in_usize).collect(); + + Self { mem: Arc::new(mem), len } + } + + /// Reads one raw `u64` of the region, for asserting on protocol + /// state the API deliberately does not expose. + fn peek_u64(&self, byte_offset: usize) -> u64 { + // SAFETY: as for `poke_u64`. + let atomic = unsafe { + AtomicU64::from_ptr(self.as_raw_slice().cast::().add(byte_offset).cast()) + }; + atomic.load(Ordering::Relaxed) + } + + /// Overwrites one raw `u64` of the region, simulating foreign-process + /// corruption of protocol metadata. + fn poke_u64(&self, byte_offset: usize, value: u64) { + // SAFETY: the offsets used by tests lie within the allocation and + // are `u64`-aligned; the atomic store synchronizes with the + // protocol's atomic accesses of the same `u64`. + let atomic = unsafe { + AtomicU64::from_ptr(self.as_raw_slice().cast::().add(byte_offset).cast()) + }; + atomic.store(value, Ordering::Relaxed); + } + } + + 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) + } + } + + /// The table length most tests use; regions add payload room on top. + const S: usize = 15; + + fn collect_frames(shm: &MockedShm) -> ShmReader { + // SAFETY: `MockedShm` provides a stable, zero-initialized allocation + // accessed only through the protocol. + unsafe { ShmReader::seal(shm.clone(), S) }.unwrap() + } + + #[test] + fn single_thread_basic() { + let shm = MockedShm::alloc(1024); + // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, + // zero-initialized allocation. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + assert!(writer.try_write_frame(b"hello")); + assert!(writer.try_write_frame(b"world")); + assert!(writer.try_write_frame(b"this is a test")); + + let frames = collect_frames(&shm); + let mut iter = frames.iter(); + assert!(iter.next().unwrap() == b"hello"); + assert!(iter.next().unwrap() == b"world"); + assert!(iter.next().unwrap() == b"this is a test"); + assert!(iter.next() == None); + } + + #[test] + fn zero_sized_frames_are_rejected() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + assert!(writer.try_write_frame(b"hello")); + assert!(!writer.try_write_frame(b"")); + + let frames = collect_frames(&shm); + let mut iter = frames.iter(); + assert!(iter.next().unwrap() == b"hello"); + assert!(iter.next() == None); + } + + #[test] + fn frame_spanning_many_u64s_roundtrips_exactly() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + let pattern: Vec = (0..=99).collect(); + assert!(writer.try_write_frame(&pattern)); + + let frames = collect_frames(&shm); + let mut iter = frames.iter(); + assert!(iter.next().unwrap() == pattern.as_slice()); + assert!(iter.next() == None); + } + + #[test] + fn full_region_fails_the_seal() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + + assert!(writer.try_write_frame(b"test")); + + // Larger than the payload region: the claim fails and sets the + // gate, which is what tells the receiver a record was lost. + assert!(!writer.try_write_frame(&vec![0u8; 2048])); + // The refused reservation stays counted: four bytes of "test" + // plus the 2048 that did not fit. + assert!(shm.peek_u64(8) == 4 + 2048); + // The loss report replaces the count with the gate. + assert!(shm.peek_u64(0) == 1 << 63); + + // "test" did land, but a lost record means the frames are not all + // of them, so the seal hands back none of them. + // SAFETY: see `collect_frames`. + let sealed = unsafe { ShmReader::seal(shm, S) }; + assert!(sealed.unwrap_err() == SealError::Closed); + } + + #[test] + fn oversized_frame_is_refused_and_fails_the_seal() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + assert!(writer.try_write_frame(b"kept")); + + // No descriptor can describe a frame this long: the claim is + // refused and sets the gate, which shuts out later claims: their + // records would ride on a result the receiver must already + // reject. + let oversized = (layout::to_usize(u32::MAX) + 1).try_into().unwrap(); + assert!(matches!(writer.claim_frame(oversized), Err(ClaimError::Capacity))); + assert!(writer.is_closed()); + assert!(!writer.try_write_frame(b"refused")); + + // SAFETY: see `collect_frames`. + let sealed = unsafe { ShmReader::seal(shm, S) }; + assert!(sealed.unwrap_err() == SealError::Closed); + } + + #[test] + fn crash_after_claim_is_skipped() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + assert!(writer.try_write_frame(b"foo")); + + // A crash right after claiming and an abandoned frame leave the + // identical state: an unfinished slot. + let _ = writer.claim_frame(5.try_into().unwrap()).unwrap(); + + assert!(writer.try_write_frame(b"bar")); + + let frames = collect_frames(&shm); + let mut iter = frames.iter(); + assert!(iter.next().unwrap() == b"foo"); + assert!(iter.next().unwrap() == b"bar"); + assert!(iter.next() == None); + // Death loses no performed operation, so the channel stays complete. + } + + #[test] + fn crash_during_partial_write_is_skipped() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + assert!(writer.try_write_frame(b"foo")); + + // Simulate a crash during writing: the frame is abandoned + // half-filled. + { + let mut frame = writer.claim_frame(5.try_into().unwrap()).unwrap(); + frame[..3].copy_from_slice(b"wor"); + } + + assert!(writer.try_write_frame(b"bar")); + + let frames = collect_frames(&shm); + let mut iter = frames.iter(); + assert!(iter.next().unwrap() == b"foo"); + assert!(iter.next().unwrap() == b"bar"); + assert!(iter.next() == None); + } + + #[test] + fn consecutive_crashes_do_not_hide_later_frames() { + // Two unfinished slots in a row, in both orders, must not prevent the + // receiver from finding the valid frames around them. + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + + assert!(writer.try_write_frame(b"foo")); + + // Crash after claim (slot stays zero, payload untouched). + let _ = writer.claim_frame(5.try_into().unwrap()).unwrap(); + + // Crash mid-write (slot stays zero, payload partially filled). + { + let mut frame = writer.claim_frame(7.try_into().unwrap()).unwrap(); + frame[..3].copy_from_slice(b"wor"); + } + + assert!(writer.try_write_frame(b"bar")); + + let frames = collect_frames(&shm); + let mut iter = frames.iter(); + assert!(iter.next().unwrap() == b"foo"); + assert!(iter.next().unwrap() == b"bar"); + assert!(iter.next() == None); + } + + #[test] + fn abandoned_frame_is_ignored() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + assert!(writer.try_write_frame(b"foo")); + + // Dropping an unfinished frame abandons it: the receiver ignores + // the slot exactly as if the writer had died there. + let _ = writer.claim_frame(5.try_into().unwrap()).unwrap(); + + let frames = collect_frames(&shm); + let mut iter = frames.iter(); + assert!(iter.next().unwrap() == b"foo"); + assert!(iter.next() == None); + } + + #[test] + fn slot_capacity_failure_fails_the_seal() { + // A 1024-byte region has a 15-slot table; the 16th claim must fail + // on the slot side while payload space remains. + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + for _ in 0..15 { + assert!(writer.try_write_frame(b"x")); + } + assert!(writer.claim_frame(1.try_into().unwrap()).unwrap_err() == ClaimError::Capacity); + // The loss report replaces the count with the gate. + assert!(shm.peek_u64(0) == 1 << 63); + + // Fifteen frames landed, but the sixteenth was lost. + // SAFETY: see `collect_frames`. + let sealed = unsafe { ShmReader::seal(shm, S) }; + assert!(sealed.unwrap_err() == SealError::Closed); + } + + #[test] + fn claims_after_seal_are_gated_without_poisoning() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + assert!(writer.try_write_frame(b"foo")); + + assert!(!writer.is_closed()); + let frames = collect_frames(&shm); + let mut iter = frames.iter(); + assert!(iter.next().unwrap() == b"foo"); + assert!(iter.next() == None); + + // The seal set the gate: a late writer's claim fails cleanly and + // does not mark the channel incomplete, since that record belongs + // after the receiver stopped collecting. + assert!(writer.is_closed()); + let before = shm.peek_u64(0); + for _ in 0..100 { + assert!(writer.claim_frame(5.try_into().unwrap()).unwrap_err() == ClaimError::Closed); + } + // Each refusal still counted its claim, and the gate stayed set. + assert!(shm.peek_u64(0) == before + 100); + } + + #[test] + fn commit_after_seal_shows_up_in_a_later_read() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + + let mut frame = writer.claim_frame(5.try_into().unwrap()).unwrap(); + frame.copy_from_slice(b"late!"); + + // The receiver seals while the frame is still unfinished: nothing + // to show yet, and nothing lost either, since the writer has not + // performed the operation this record describes. + let frames = collect_frames(&shm); + assert!(frames.iter().count() == 0); + + // The reader reads the table when asked, so a commit that lands + // after the seal shows up in the very same reader. + frame.finish(); + assert!(frames.iter().count() == 1); + + // A second seal fails: the first one set the gate, and a re-seal + // cannot say what was refused since then. + // SAFETY: see `collect_frames`. + let sealed = unsafe { ShmReader::seal(shm, S) }; + assert!(sealed.unwrap_err() == SealError::Closed); + } + + #[test] + fn concurrent() { + // A 255-slot table for the 120 frames written below, with 16 KiB + // of room for the fixed struct and the payloads. + const S: usize = 255; + let shm = MockedShm::alloc(1024 * 16); + + thread::scope(|s| { + for _ in 0..4 { + s.spawn(|| { + // SAFETY: see `single_thread_basic`. The clone shares the + // same backing memory, which is safe because the protocol + // synchronizes concurrent access with atomics. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + for _ in 0..10 { + assert!(writer.try_write_frame(b"hello")); + assert!(writer.try_write_frame(b"foo")); + assert!(writer.try_write_frame(b"this is a test")); + } + }); + } + }); + + // SAFETY: see `collect_frames`. + let frames = unsafe { ShmReader::seal(shm, S) }.unwrap(); + let mut count = 0; + for frame in &frames { + count += 1; + let frame = BStr::new(frame); + assert!(frame == b"hello" || frame == b"foo" || frame == b"this is a test"); + } + assert!(count == 120); + } + + #[test] + fn concurrent_exceeded_size() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + thread::scope(|s| { + for _ in 0..4 { + s.spawn(|| { + for _ in 0..10 { + writer.try_write_frame(b"hello"); + writer.try_write_frame(b"foo"); + writer.try_write_frame(b"this is a test"); + } + }); + } + }); + + // The table holds 15 slots and 120 writes were attempted, so some + // had to fail on capacity, and one failure is enough to fail the + // seal. The writers all survived it. + assert!(writer.is_closed()); + // SAFETY: see `collect_frames`. + let sealed = unsafe { ShmReader::seal(shm, S) }; + assert!(sealed.unwrap_err() == SealError::Closed); + } + + #[test] + fn seal_races_with_active_writers() { + // Plenty of slots and payload room: capacity must never fail here. + const S: usize = 1023; + let shm = MockedShm::alloc(1024 * 64); + let barrier = Barrier::new(3); + + let (frames, results) = thread::scope(|s| { + let writers = [(); 2].map(|()| { + s.spawn(|| { + // SAFETY: see `concurrent`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + barrier.wait(); + let mut written = 0usize; + // Bounded so the test terminates even if the seal is slow; + // the region is large enough that capacity never fails. + for _ in 0..200 { + match writer.claim_frame(5.try_into().unwrap()) { + Ok(mut frame) => { + frame.copy_from_slice(b"hello"); + frame.finish(); + written += 1; + } + Err(ClaimError::Closed) => break, + Err(ClaimError::Capacity) => panic!("region unexpectedly full"), + } + } + written + }) + }); + + barrier.wait(); + // SAFETY: see `collect_frames`. + let frames = unsafe { ShmReader::seal(shm.clone(), S) }.unwrap(); + let results = writers.map(|writer| writer.join().unwrap()); + (frames, results) + }); + + // Every frame the reader yields is whole: a descriptor becomes + // visible only after its payload is written. + let mut count = 0; + for frame in &frames { + count += 1; + assert!(frame == b"hello"); + } + // Claims taken after the boundary are never read, so the count can + // fall short of what the writers wrote, but nothing can appear + // that was never finished. + let written: usize = results.into_iter().sum(); + assert!(count <= written); + // Writers either finished or cleanly observed `Closed`; nothing was + // abandoned, so completeness holds. + } + + #[test] + fn overshot_claim_counter_clamps_to_the_table() { + let shm = MockedShm::alloc(1024); + // SAFETY: see `single_thread_basic`. + let writer = unsafe { ShmWriter::new(shm.clone(), S) }.unwrap(); + assert!(writer.try_write_frame(b"hello")); + + // A wildly inflated claim counter, from mass claim failures or a + // foreign scribble, degrades to a full-table sweep rather than an + // out-of-bounds slot access: the committed frame survives and the + // untouched slots read as unpublished. + shm.poke_u64(0, (1 << 40) | 1); + + let frames = collect_frames(&shm); + let mut iter = frames.iter(); + assert!(iter.next().unwrap() == b"hello"); + assert!(iter.next() == None); + } + + #[test] + fn misaligned_region_is_rejected() { + 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( + // SAFETY: Adding 1 byte to create a deliberately + // misaligned pointer for testing. The original allocation + // is large enough that adding 1 byte stays within bounds. + unsafe { raw_slice.cast::().add(1) }, + raw_slice.len() - 1, + ) + } + } + + let misaligned_shm = Misaligned(MockedShm::alloc(1024)); + assert!( + !misaligned_shm.as_raw_slice().cast::().addr().is_multiple_of(align_of::()) + ); + + // SAFETY: the wrapped allocation is valid; only its alignment is + // deliberately wrong. + assert!(unsafe { ShmWriter::new(misaligned_shm, S) }.is_none()); + } + + #[test] + #[cfg(not(miri))] + fn real_shm_across_processes() { + use std::process::{Child, Command}; + + use rustc_hash::FxHashSet; + use subprocess_test::command_for_fn; + + const CHILD_COUNT: usize = 12; + const FRAME_COUNT_EACH_CHILD: usize = 100; + + // Room for every child's frames, in slots and in payload bytes. + const S: usize = 16383; + const SHM_SIZE: usize = 1024 * 1024; + + let shm_path = crate::ipc::channel::shm_backing_path().unwrap(); + let shm_name = shm_path.to_str().expect("test temp dir is UTF-8").to_owned(); + let c_path = crate::ipc::channel::os_c_string(shm_path.as_os_str()).unwrap(); + let handle = fspy_shm::create(c_path.as_c_str().as_thin(), SHM_SIZE).unwrap(); + let _keeper = crate::ipc::channel::ShmKeeper { path: c_path }; + // Map before the children run. Windows keeps views coherent while they + // exist at the same time; a view created after every writer exited can + // observe the file before the writers' dirty pages reach it. + let mapping = handle.map().unwrap(); + + let children: Vec = (0..CHILD_COUNT) + .map(|child_index| { + let cmd = command_for_fn!( + (shm_name.clone(), child_index), + |(shm_name, child_index): (String, usize)| { + let c_path = + crate::ipc::channel::os_c_string(std::ffi::OsStr::new(&shm_name)) + .unwrap(); + let mapping = + fspy_shm::open(c_path.as_c_str().as_thin()).unwrap().map().unwrap(); + // SAFETY: `mapping` is a freshly mapped shared memory + // region with a valid pointer and size; the protocol + // synchronizes concurrent access. + let writer = unsafe { ShmWriter::new(mapping, S) }.unwrap(); + for i in 0..FRAME_COUNT_EACH_CHILD { + let frame_data = std::format!("{child_index} {i}"); + assert!(writer.try_write_frame(frame_data.as_bytes())); + } + } + ); + Command::from(cmd).spawn().unwrap() + }) + .collect(); + + for mut c in children { + let status = c.wait().unwrap(); + assert!(status.success()); + } + + // SAFETY: the mapping is a valid shared-memory region created zeroed + // and accessed only through the protocol. + let frames = unsafe { ShmReader::seal(mapping, S) }.unwrap(); + let collected = frames.iter().map(BStr::new).collect::>(); + assert!(collected.len() == CHILD_COUNT * FRAME_COUNT_EACH_CHILD); + for child_index in 0..CHILD_COUNT { + for i in 0..FRAME_COUNT_EACH_CHILD { + let frame_data = format!("{child_index} {i}"); + assert!(collected.contains(&BStr::new(frame_data.as_bytes()))); + } + } + } + + /// A writer killed mid-frame (SIGKILL on Unix, `TerminateProcess` on + /// Windows, both via `Child::kill`) must not lose other writers' frames + /// or completeness: no cleanup code runs in the killed process. + #[test] + #[cfg(not(miri))] + fn killed_writer_does_not_poison_the_channel() { + use std::{ + io::{BufRead as _, BufReader}, + process::{Command, Stdio}, + }; + + use subprocess_test::command_for_fn; + + const SHM_SIZE: usize = 1024 * 1024; + + let shm_path = crate::ipc::channel::shm_backing_path().unwrap(); + let shm_name = shm_path.to_str().expect("test temp dir is UTF-8").to_owned(); + let c_path = crate::ipc::channel::os_c_string(shm_path.as_os_str()).unwrap(); + let handle = fspy_shm::create(c_path.as_c_str().as_thin(), SHM_SIZE).unwrap(); + let _keeper = crate::ipc::channel::ShmKeeper { path: c_path }; + let mapping = handle.map().unwrap(); + + let cmd = command_for_fn!(shm_name, |shm_name: String| { + let c_path = crate::ipc::channel::os_c_string(std::ffi::OsStr::new(&shm_name)).unwrap(); + let child_mapping = fspy_shm::open(c_path.as_c_str().as_thin()).unwrap().map().unwrap(); + // SAFETY: see `real_shm_across_processes`. + let writer = unsafe { ShmWriter::new(child_mapping, S) }.unwrap(); + let mut frame = writer.claim_frame(5.try_into().unwrap()).unwrap(); + frame[..3].copy_from_slice(b"wor"); + // Signal the parent that the frame is claimed and partially + // written, then wait to be killed. + #[expect(clippy::print_stdout, reason = "readiness handshake with the parent")] + { + println!("claimed"); + } + let _ = frame; + // Wait to be killed; nothing ever unparks this thread. + loop { + std::thread::park(); + } + }); + let mut command = Command::from(cmd); + command.stdout(Stdio::piped()); + let mut child = command.spawn().unwrap(); + let mut line = String::new(); + BufReader::new(child.stdout.take().unwrap()).read_line(&mut line).unwrap(); + assert!(line.trim() == "claimed"); + child.kill().unwrap(); + child.wait().unwrap(); + + // A surviving writer keeps working after the kill. It borrows the + // mapping so the seal below can take it over. + // SAFETY: see `real_shm_across_processes`. + let writer = unsafe { ShmWriter::new(&mapping, S) }.unwrap(); + assert!(writer.try_write_frame(b"alive")); + + // SAFETY: see `real_shm_across_processes`. + let frames = unsafe { ShmReader::seal(mapping, S) }.unwrap(); + let mut iter = frames.iter(); + assert!(iter.next().unwrap() == b"alive"); + assert!(iter.next() == None); + // The killed writer left only an unfinished slot; the counters + // stayed within their limits, so the channel is complete. + } +} diff --git a/crates/fspy_shared/src/ipc/channel/shm_io/reader.rs b/crates/fspy_shared/src/ipc/channel/shm_io/reader.rs new file mode 100644 index 000000000..a2b399d2d --- /dev/null +++ b/crates/fspy_shared/src/ipc/channel/shm_io/reader.rs @@ -0,0 +1,214 @@ +//! The reader side: seal the channel, then iterate the committed frames. +//! +//! Neither step walks the table or waits for a writer, and no payload byte +//! is copied. The reader keeps the mapping alive and hands out each +//! committed span on demand. Those borrows hold because nothing writes to +//! a committed span again (committing uses up the writer's frame) and it +//! never overlaps what a live writer may touch. +//! +//! A span's offset and length come from the writer that reserved them, and +//! nothing here re-checks them. That rests on the promise made at attach: +//! only this protocol touches the region. A process that scribbles on it +//! some other way breaks the promise, and this code does not defend +//! against that. + +use std::{ + fmt, + iter::FusedIterator, + ptr::NonNull, + slice, + sync::atomic::{AtomicU64, Ordering}, +}; + +use super::{ + AsRawSlice, + layout::{CLOSED, MappedLayout, SlotState, to_usize}, +}; + +/// Why a channel could not be sealed. +#[derive(thiserror::Error, Clone, Copy, PartialEq, Eq, Debug)] +pub enum SealError { + /// The mapping cannot hold the protocol at all (see + /// [`MappedLayout::new`]). + #[error("the shared-memory region cannot host the channel")] + UnsupportedRegion, + /// The channel was already closed when the seal ran. A claim had + /// failed, or someone sealed earlier and this seal cannot say what was + /// refused since. Either way the frames are not all of them, so the + /// reader hands back none. + #[error("the shared-memory channel was closed before it was sealed")] + Closed, +} + +/// A reader over the committed frames of a sealed channel, serving them +/// straight out of the mapping. It holds no buffer of its own, so sealing +/// allocates nothing; dropping it releases the mapping. +pub struct ShmReader { + /// Where the payload area is, for turning descriptors into spans. + payload_start: NonNull, + /// The slots the seal admitted, in claim order. + table: NonNull<[AtomicU64]>, + /// Owns the region the pointers point into. Declared after them: + /// fields drop in order, and the borrower must go first. + _mem: M, +} + +// SAFETY: the reader only loads counters and descriptors atomically, and +// reads committed spans that nothing writes to any more; the stored +// pointers point into the mapped memory, which is owned separately and +// does not move, not into the reader itself. +unsafe impl Send for ShmReader {} +// SAFETY: see the `Send` impl. +unsafe impl Sync for ShmReader {} + +impl ShmReader { + /// Seals the channel against further records and returns the reader of + /// its committed frames. + /// + /// A reader exists only for a channel that kept everything. If a claim + /// had already failed, or someone sealed earlier, there is no complete + /// set of records and this fails instead. + /// + /// One atomic operation fixes how far iteration goes and shuts the + /// gate, so it neither waits for a writer nor walks the table. A claim + /// taken before that point but committed after it shows up if the + /// store lands before the read reaches its slot; one taken after it + /// lands in a slot iteration never reaches. + /// + /// # Safety + /// + /// Same contract as [`ShmWriter::new`](super::ShmWriter::new): + /// + /// - `mem.as_raw_slice()` must return a stable, valid pointer to the + /// whole region for the reader's lifetime. + /// - The region must have been zero-initialized when it was created and + /// accessed only through this protocol since. + /// - `slots` must be the count the region was created with. + /// + /// # Errors + /// + /// [`SealError`]: the mapping cannot hold the protocol, or the channel + /// was already closed before this call. + pub unsafe fn seal(mem: M, slots: usize) -> Result { + // SAFETY: forwarded from this function's contract, which keeps the + // region valid for as long as the reader lives, and so for every + // use of the pointers, which are stored in the reader and dropped + // with it. + let Some(mapped) = (unsafe { MappedLayout::new(mem.as_raw_slice(), slots) }) else { + return Err(SealError::UnsupportedRegion); + }; + + // Draw the boundary and shut the gate in one step (rule 1): this + // returns the claim count at the instant no later claim can + // succeed. Replacing the count rather than keeping it is fine, + // since nothing reads it from here on: a later claim fails on the + // gate before it uses its slot index, and a later seal only tests + // the bit. + let claims = mapped.claims().swap(CLOSED, Ordering::Relaxed); + // The same value says whether anything was lost. A gate already + // set means a failed claim, which rule 1 puts before this + // boundary, or an earlier seal this one cannot account for. + if claims & CLOSED != 0 { + return Err(SealError::Closed); + } + // The count runs past the table whenever writers attempt more + // claims than there are slots: one bumps the counter, finds its + // index out of range, and only then sets the gate, so a seal + // landing in that window reads the higher count with the gate + // still clear. Such a count names slots that do not exist, so the + // fall-back reads the whole table. + // + // Not an error, either: that writer performs the operation it + // failed to record only after this boundary, and if it died first + // it never performed one at all. + let slots = mapped.table(); + let admitted = slots.get(..to_usize(claims)).unwrap_or(slots); + + // The admitted slots, kept as a raw pointer so the reader needs + // no lifetime. + // SAFETY of every later read through it: it points into the + // mapping this reader owns, and a slot is an `AtomicU64`, so a + // writer storing a descriptor never invalidates it. + let table = NonNull::from_ref(admitted); + Ok(Self { payload_start: mapped.payload_start, table, _mem: mem }) + } + + /// Iterates over the committed frames in claim order. + /// + /// Reads the descriptor table as it goes, so a writer that was still + /// filling a frame when the channel was sealed may appear in a later + /// call and not an earlier one. Everything it yields is a whole frame + /// whose writer finished it. + pub const fn iter(&self) -> Iter<'_> { + Iter { + payload_start: self.payload_start, + // SAFETY: the slots live in the mapping this reader owns, and + // each one is an `AtomicU64`, so writers storing descriptors + // through their own pointers never invalidate this borrow. It + // lasts no longer than `&self`, and so no longer than the + // mapping. + table: unsafe { self.table.as_ref() }, + } + } +} + +impl fmt::Debug for ShmReader { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ShmReader").field("slots", &self.table.len()).finish_non_exhaustive() + } +} + +/// Iterator over a [`ShmReader`]'s committed frames, in claim order. +pub struct Iter<'a> { + /// Where the payload area is, for turning descriptors into spans. + payload_start: NonNull, + /// The admitted slots this iterator has not reached yet. + table: &'a [AtomicU64], +} + +impl<'a> Iterator for Iter<'a> { + type Item = &'a [u8]; + + fn next(&mut self) -> Option { + while let Some((slot, rest)) = self.table.split_first() { + self.table = rest; + // Rule 3: `Acquire`, so a descriptor this load sees brings + // its payload bytes with it. + let slot_value = slot.load(Ordering::Acquire); + // An unfinished slot published nothing. + let SlotState::Committed { offset, len } = SlotState::decode(slot_value) else { + continue; + }; + // SAFETY: a committed descriptor names the span its writer + // reserved inside the payload area, and the attach contract + // says nothing but this protocol writes the region, so these + // are the bits a writer put here. Nothing writes to a + // committed span any more, and the reader borrowed for `'a` + // keeps the mapping alive. + return Some(unsafe { + slice::from_raw_parts( + self.payload_start.add(to_usize(offset)).as_ptr().cast_const(), + to_usize(len.get()), + ) + }); + } + None + } + + fn size_hint(&self) -> (usize, Option) { + // One frame per remaining slot at most; how many are committed is + // only known by reading them. + (0, Some(self.table.len())) + } +} + +impl FusedIterator for Iter<'_> {} + +impl<'a, M: AsRawSlice> IntoIterator for &'a ShmReader { + type IntoIter = Iter<'a>; + type Item = &'a [u8]; + + fn into_iter(self) -> Iter<'a> { + self.iter() + } +} diff --git a/crates/fspy_shared/src/ipc/channel/shm_io/writer.rs b/crates/fspy_shared/src/ipc/channel/shm_io/writer.rs new file mode 100644 index 000000000..9cda0ab33 --- /dev/null +++ b/crates/fspy_shared/src/ipc/channel/shm_io/writer.rs @@ -0,0 +1,213 @@ +//! The writer side: claim a frame, fill it, finish it. + +use std::{ + num::{NonZeroU32, NonZeroUsize}, + ops::{Deref, DerefMut}, + slice, + sync::atomic::{AtomicU64, Ordering}, +}; + +use super::{ + AsRawSlice, + layout::{CLOSED, MappedLayout, SlotState, to_usize}, +}; + +/// A shared-memory frame writer, usable from many threads and processes at +/// once. Each frame is reserved atomically, filled in a span no one else +/// can touch, and published with one atomic write (the ordering contract +/// in [`super::layout`]). +pub struct ShmWriter { + mapped: MappedLayout, + /// Owns the region the pointers point into. Declared after them: + /// fields drop in order, and the borrower must go first. + _mem: M, +} + +// SAFETY: the writer touches the region only through the protocol's +// atomics, which synchronize access from any thread; the stored pointers +// point into the mapped memory, which is owned separately and does not +// move, not into the writer itself. +unsafe impl Send for ShmWriter {} +// SAFETY: see the `Send` impl; the writer's shared-reference API is +// internally synchronized by the protocol. +unsafe impl Sync for ShmWriter {} + +/// Why a frame could not be claimed. +#[derive(thiserror::Error, Clone, Copy, PartialEq, Eq, Debug)] +pub enum ClaimError { + /// The CLOSED gate is set, by a seal or by an earlier failed claim. + /// Dropping the record is right either way: the receiver had stopped + /// collecting, or that same bit already fails its seal. + #[error("the channel has been closed")] + Closed, + /// No room left, or a frame longer than the `u32::MAX` a descriptor + /// can describe. This claim set the CLOSED gate on its way out, so the + /// seal will fail and every later claim is refused. + #[error("no space left in the shared-memory region")] + Capacity, +} + +impl ShmWriter { + /// Creates a writer on a shared-memory region whose table holds + /// `slots` descriptors, or `None` when the region cannot hold the + /// protocol (see [`MappedLayout::new`]), as a truncated or unrelated + /// file cannot. + /// + /// # Safety + /// + /// - `mem.as_raw_slice()` must return a stable, valid pointer to the + /// whole region for the writer's lifetime. + /// - The region must have been zero-initialized when it was created and + /// accessed only through this protocol since. + /// - `slots` must be the count the region was created with. + pub unsafe fn new(mem: M, slots: usize) -> Option { + // SAFETY: forwarded from this function's contract, which keeps the + // region valid, and used only by this protocol, for as long as the + // writer lives, and so for every use of the pointers, which are + // stored in the writer and dropped with it. + let mapped = unsafe { MappedLayout::new(mem.as_raw_slice(), slots) }?; + Some(Self { mapped, _mem: mem }) + } + + /// Whether the CLOSED gate is set, by a seal or by a failed claim. + pub fn is_closed(&self) -> bool { + self.mapped.claims().load(Ordering::Relaxed) & CLOSED != 0 + } + + /// Claims a frame of exactly `frame_size` bytes. Wait-free: two + /// `fetch_add`s, no retry loop (rule 1). + /// + /// The receiver cannot see the frame until [`FrameMut::finish`] + /// commits it. Dropping it gives the claim up, and the receiver + /// ignores the slot as if the writer had died. A claim that does not + /// fit returns [`ClaimError::Capacity`], after setting the CLOSED + /// gate. + pub fn claim_frame(&self, frame_size: NonZeroUsize) -> Result, ClaimError> { + let mapped = self.mapped; + + // The loss report (rule 1): the gate marks the frames incomplete + // and shuts the channel down for later claims. + // + // Storing the gate rather than or-ing it in drops the claim count, + // which nothing reads once the gate is set. The seal fails on the + // bit before it looks at the count, and a claim that reads the + // cleared count reads the gate along with it, so it gives up + // before using a slot index. + let report_loss = || { + mapped.claims().store(CLOSED, Ordering::Relaxed); + ClaimError::Capacity + }; + + // A frame too long for the descriptor's 32-bit length field cannot + // be described, so this conversion is the oversize check. + let Ok(frame_size) = NonZeroU32::try_from(frame_size) else { + return Err(report_loss()); + }; + // The payload space this claim takes, as the counter's `u64`. + // Widening a 32-bit size never loses anything. + let reservation = u64::from(frame_size.get()); + // Payload bytes first, so a payload-capacity failure does not burn a + // slot. + let payload_start = mapped.payload_reserved().fetch_add(reservation, Ordering::Relaxed); + let Some(payload_offset) = + fitted_offset(payload_start, frame_size.get(), mapped.payload_len) + else { + return Err(report_loss()); + }; + + let claims = mapped.claims().fetch_add(1, Ordering::Relaxed); + if claims & CLOSED != 0 { + // Not a loss: the receiver had already stopped collecting when + // this record was refused. + return Err(ClaimError::Closed); + } + let Some(slot) = self.mapped.table().get(to_usize(claims)) else { + return Err(report_loss()); + }; + + // SAFETY: the claim reserved this span for itself, and the check + // above put it inside the region. Other writers reserve spans + // that never overlap, and the receiver reads no payload until it + // sees the committed descriptor, which `finish` writes only by + // taking this borrow. + let content = unsafe { + slice::from_raw_parts_mut( + mapped.payload_start.add(to_usize(payload_offset)).as_ptr(), + to_usize(frame_size.get()), + ) + }; + Ok(FrameMut { + slot, + slot_to_commit: SlotState::Committed { offset: payload_offset, len: frame_size } + .encode(), + content, + }) + } + + #[cfg(test)] + pub fn try_write_frame(&self, frame: &[u8]) -> bool { + let Some(frame_size) = NonZeroUsize::new(frame.len()) else { + return false; + }; + let Ok(mut frame_mut) = self.claim_frame(frame_size) else { + return false; + }; + frame_mut.copy_from_slice(frame); + frame_mut.finish(); + true + } +} + +/// The offset of a `len`-byte frame reserved at `start`, or `None` when it +/// does not fit: the start must be small enough for a descriptor's 32-bit +/// offset, and the frame must end inside the payload area. Both are +/// checked, so if something else scribbled on the counter the claim fails +/// rather than wrapping around into a span outside the region. +fn fitted_offset(start: u64, len: u32, payload_len: u32) -> Option { + let offset = u32::try_from(start).ok()?; + let end = offset.checked_add(len)?; + (end <= payload_len).then_some(offset) +} + +/// An exclusively owned frame, claimed but not yet published. +/// +/// [`FrameMut::finish`] is the only way to show the payload to the +/// receiver. Dropping the frame gives the claim up: the slot stays +/// unfinished and the receiver ignores it, as if the writer had died +/// there. The two look identical from the outside, which is why a writer +/// that drops a frame and performs the operation anyway breaks the rule +/// this channel rests on. +#[derive(Debug)] +pub struct FrameMut<'a> { + slot: &'a AtomicU64, + slot_to_commit: u64, + 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 FrameMut<'_> { + /// Commits the frame, making it visible to the receiver. + /// + /// A receiver that already sealed the channel may or may not show it, + /// depending on whether this store lands before the read reaches that + /// slot. Either answer is truthful, because the operation this record + /// describes had not happened when the receiver drew its line. + pub fn finish(self) { + // Rule 2: `Release` orders every payload write before the + // descriptor. This writer owns the slot, so a store is enough. + self.slot.store(self.slot_to_commit, Ordering::Release); + } +} diff --git a/crates/fspy_shared/src/ipc/mod.rs b/crates/fspy_shared/src/ipc/mod.rs index 9c49e7371..43a127196 100644 --- a/crates/fspy_shared/src/ipc/mod.rs +++ b/crates/fspy_shared/src/ipc/mod.rs @@ -8,6 +8,22 @@ pub use fspy_ipc_str::IpcStr; pub use ipc_path::IpcPath; use wincode::{SchemaRead, SchemaWrite}; +/// How much shared memory a channel gets, and how many records fit in it. +/// +/// Both numbers come from the caller: this crate has no way to guess how +/// many records a workload makes. Both ends of one channel must agree on +/// the slot count, which the receiver passes to `channel` and every sender +/// reads back out of the `ChannelConf`. +#[derive(Clone, Copy, Debug)] +pub struct ChannelSize { + /// Bytes of shared memory. The descriptor table takes the front of it + /// and payloads take the rest. + pub capacity: usize, + /// Descriptor slots, one per record. A record past this many is + /// refused just like one the payload area has no room for. + pub slots: usize, +} + #[derive(SchemaWrite, SchemaRead, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy)] pub struct AccessMode(u8); diff --git a/crates/vt/src/session/event.rs b/crates/vt/src/session/event.rs index be9210e18..b28b2f2d5 100644 --- a/crates/vt/src/session/event.rs +++ b/crates/vt/src/session/event.rs @@ -99,6 +99,12 @@ pub enum CacheNotUpdatedReason { /// A runner-aware tool explicitly requested that this run not be cached /// (e.g. vite dev-server, a watch task). ToolRequested, + /// A tracked process could not record a file access it went on to + /// perform: the channel it reports them through had no room left, or + /// none for a record that long. The accesses that did arrive are a + /// subset of what the task touched, so caching from them would bake in + /// inputs and outputs that are not all of them. + TrackingIncomplete, } #[derive(Debug)] diff --git a/crates/vt/src/session/execute/cache_update.rs b/crates/vt/src/session/execute/cache_update.rs index f13d3f1df..6d320a083 100644 --- a/crates/vt/src/session/execute/cache_update.rs +++ b/crates/vt/src/session/execute/cache_update.rs @@ -89,8 +89,18 @@ pub(super) async fn update_cache( return (CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::NonZeroExitStatus), None); } + // The accesses, or `None` when the task was not tracked at all. An + // `Err` means it made more file accesses than the tracking channel had + // room for, so what arrived is a subset of what it touched, and an entry + // built from a subset would replay with inputs and outputs missing. + #[cfg(fspy)] + let Ok(path_accesses) = outcome.path_accesses.as_ref().map(Result::as_ref).transpose() else { + return (CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::TrackingIncomplete), None); + }; + let fspy_outcome = observe_fspy( - outcome, + #[cfg(fspy)] + path_accesses, metadata, fspy, &ignored_input_rels, @@ -194,7 +204,7 @@ pub(super) async fn update_cache( /// `path_writes` is filtered by user-configured output negatives and /// tool-reported `ignoreOutput` paths before read-write overlap detection. fn observe_fspy( - outcome: &ChildOutcome, + #[cfg(fspy)] path_accesses: Option<&fspy::PathAccessIterable>, metadata: &CacheMetadata, fspy: Option<&super::FspyTracking<'_>>, ignored_input_rels: &FxHashSet, @@ -205,7 +215,7 @@ fn observe_fspy( { use super::tracked_accesses::TrackedPathAccesses; - outcome.path_accesses.as_ref().map(|raw| { + path_accesses.map(|raw| { let tracked = TrackedPathAccesses::from_raw(raw, workspace_root); let filtered_path_reads: HashMap = // fspy can be attached for auto-output-only tasks. In that @@ -254,7 +264,7 @@ fn observe_fspy( } #[cfg(not(fspy))] { - let _ = (outcome, metadata, fspy, ignored_input_rels, ignored_output_rels, workspace_root); + let _ = (metadata, fspy, ignored_input_rels, ignored_output_rels, workspace_root); None } } diff --git a/crates/vt/src/session/execute/spawn.rs b/crates/vt/src/session/execute/spawn.rs index adff8aac9..7e2abb59e 100644 --- a/crates/vt/src/session/execute/spawn.rs +++ b/crates/vt/src/session/execute/spawn.rs @@ -41,9 +41,10 @@ pub struct ChildHandle { /// Result of waiting for a child to exit. pub struct ChildOutcome { pub exit_status: std::process::ExitStatus, - /// Raw fspy accesses. `Some` iff `fspy` was `true` at spawn time. + /// Raw fspy accesses. `Some` iff `fspy` was `true` at spawn time, and + /// `Err` when a tracked process could not record everything it did. #[cfg(fspy)] - pub path_accesses: Option, + pub path_accesses: Option>, } /// Spawn a command with the requested fspy and stdio configuration. diff --git a/crates/vt/src/session/reporter/summary.rs b/crates/vt/src/session/reporter/summary.rs index c9da7dff2..2895e5c80 100644 --- a/crates/vt/src/session/reporter/summary.rs +++ b/crates/vt/src/session/reporter/summary.rs @@ -111,6 +111,12 @@ pub enum SpawnOutcome { /// Rendered message of the IPC server error that caused the cache to /// be skipped, if any. ipc_server_error: Option, + /// `true` when the task made more file accesses than tracking had + /// room for, so the inferred inputs and outputs were a subset of + /// what it touched. Task ran successfully but cache was not + /// updated. + #[serde(default)] + tracking_incomplete: bool, /// Set when a runner-aware tool called `disableCache()`, skipping /// cache update. tool_disabled_cache: bool, @@ -343,6 +349,10 @@ impl TaskResult { cache_update_status, CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::ToolRequested) ); + let tracking_incomplete = matches!( + cache_update_status, + CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::TrackingIncomplete) + ); match cache_status { CacheStatus::Hit { replayed_duration } => { @@ -358,6 +368,7 @@ impl TaskResult { fspy_unsupported, ipc_server_error, tool_disabled_cache, + tracking_incomplete, ), }, CacheStatus::Miss(cache_miss) => Self::Spawned { @@ -371,6 +382,7 @@ impl TaskResult { fspy_unsupported, ipc_server_error, tool_disabled_cache, + tracking_incomplete, ), }, } @@ -385,6 +397,7 @@ fn spawn_outcome_from_execution( fspy_unsupported: bool, ipc_server_error: Option, tool_disabled_cache: bool, + tracking_incomplete: bool, ) -> SpawnOutcome { match (exit_status, saved_error) { // Spawn error — process never ran @@ -396,6 +409,7 @@ fn spawn_outcome_from_execution( fspy_unsupported, ipc_server_error, tool_disabled_cache, + tracking_incomplete, }, // Process exited with non-zero code (Some(status), _) => { @@ -416,6 +430,7 @@ fn spawn_outcome_from_execution( fspy_unsupported: false, ipc_server_error: None, tool_disabled_cache: false, + tracking_incomplete: false, }, } } @@ -554,6 +569,17 @@ impl TaskResult { { return vt_str::format!("→ Not cached: read and wrote '{path}'"); } + // Tracking came up short, so the inferred inputs and outputs would + // have been a subset of what the task touched. + if let Self::Spawned { + outcome: SpawnOutcome::Success { tracking_incomplete: true, .. }, + .. + } = self + { + return Str::from( + "→ Not cached: tracking ran out of room for this task's file accesses", + ); + } // fspy-unsupported-on-this-OS message — same overrides precedence as above if let Self::Spawned { outcome: SpawnOutcome::Success { fspy_unsupported: true, .. }, .. diff --git a/crates/vt_bin/tests/e2e_snapshots/fixtures/fspy_shm_capacity/snapshots.toml b/crates/vt_bin/tests/e2e_snapshots/fixtures/fspy_shm_capacity/snapshots.toml index 6ce6bb943..5ea5c6527 100644 --- a/crates/vt_bin/tests/e2e_snapshots/fixtures/fspy_shm_capacity/snapshots.toml +++ b/crates/vt_bin/tests/e2e_snapshots/fixtures/fspy_shm_capacity/snapshots.toml @@ -1,9 +1,9 @@ [[e2e]] name = "shm_capacity_env_sizes_the_tracking_channel" comment = """ -`VP_RUN_INTERNAL_FSPY_SHM_CAPACITY` sizes the shared memory a tracked task reports its file accesses through. The task makes twenty thousand of them, and 64 MiB holds every one, so the run caches like any other. +`VP_RUN_INTERNAL_FSPY_SHM_CAPACITY` sizes the shared memory a tracked task reports its file accesses through. The task makes twenty thousand of them, and a 64 KiB channel holds a thousand: one descriptor slot per 64 bytes of the region. -Setting the capacity below what the task needs is what the knob exists for. That case has to wait: a channel with no room for a record currently aborts the task process, and the panic it prints carries a thread id, a toolchain path, a backtrace and a platform's own abort code, none of which snapshot the same way twice. +The task runs to the end anyway: recording must never stop the program doing the work, so the accesses past that go unrecorded and the process carries on to a clean exit, printing its last line. What the run cannot claim is that it saw every file the task touched, so it is not cached, and the second run says the same rather than replaying an entry built from part of a trace. Not on musl, which has no preload: those builds collect through the seccomp supervisor, on the runner's own side of the boundary, so they have no shared-memory channel to fill. """ @@ -17,13 +17,18 @@ steps = [ ], envs = [ [ "VP_RUN_INTERNAL_FSPY_SHM_CAPACITY", - "67108864", + "65536", ], - ], comment = "64 MiB, room for every access" }, + ], comment = "64 KiB, a thousand slots for twenty thousand accesses" }, { argv = [ "vt", "run", "-v", "stat", - ], comment = "replayed from the entry the first run stored" }, + ], envs = [ + [ + "VP_RUN_INTERNAL_FSPY_SHM_CAPACITY", + "65536", + ], + ], comment = "nothing was cached to replay" }, ] diff --git a/crates/vt_bin/tests/e2e_snapshots/fixtures/fspy_shm_capacity/snapshots/shm_capacity_env_sizes_the_tracking_channel.md b/crates/vt_bin/tests/e2e_snapshots/fixtures/fspy_shm_capacity/snapshots/shm_capacity_env_sizes_the_tracking_channel.md index f28d778bc..abbc9b792 100644 --- a/crates/vt_bin/tests/e2e_snapshots/fixtures/fspy_shm_capacity/snapshots/shm_capacity_env_sizes_the_tracking_channel.md +++ b/crates/vt_bin/tests/e2e_snapshots/fixtures/fspy_shm_capacity/snapshots/shm_capacity_env_sizes_the_tracking_channel.md @@ -1,14 +1,14 @@ # shm_capacity_env_sizes_the_tracking_channel -`VP_RUN_INTERNAL_FSPY_SHM_CAPACITY` sizes the shared memory a tracked task reports its file accesses through. The task makes twenty thousand of them, and 64 MiB holds every one, so the run caches like any other. +`VP_RUN_INTERNAL_FSPY_SHM_CAPACITY` sizes the shared memory a tracked task reports its file accesses through. The task makes twenty thousand of them, and a 64 KiB channel holds a thousand: one descriptor slot per 64 bytes of the region. -Setting the capacity below what the task needs is what the knob exists for. That case has to wait: a channel with no room for a record currently aborts the task process, and the panic it prints carries a thread id, a toolchain path, a backtrace and a platform's own abort code, none of which snapshot the same way twice. +The task runs to the end anyway: recording must never stop the program doing the work, so the accesses past that go unrecorded and the process carries on to a clean exit, printing its last line. What the run cannot claim is that it saw every file the task touched, so it is not cached, and the second run says the same rather than replaying an entry built from part of a trace. Not on musl, which has no preload: those builds collect through the seccomp supervisor, on the runner's own side of the boundary, so they have no shared-memory channel to fill. -## `VP_RUN_INTERNAL_FSPY_SHM_CAPACITY=67108864 vt run -v stat` +## `VP_RUN_INTERNAL_FSPY_SHM_CAPACITY=65536 vt run -v stat` -64 MiB, room for every access +64 KiB, a thousand slots for twenty thousand accesses ``` $ vtt stat-many 20000 @@ -25,16 +25,16 @@ Performance: 0% cache hit rate Task Details: ──────────────────────────────────────────────── [1] fspy-shm-capacity#stat: $ vtt stat-many 20000 ✓ - → Cache miss: no previous cache entry found + → Not cached: tracking ran out of room for this task's file accesses ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ``` -## `vt run -v stat` +## `VP_RUN_INTERNAL_FSPY_SHM_CAPACITY=65536 vt run -v stat` -replayed from the entry the first run stored +nothing was cached to replay ``` -$ vtt stat-many 20000 ◉ cache hit, replaying +$ vtt stat-many 20000 stat 20000 @@ -42,12 +42,12 @@ stat 20000 Vite+ Task Runner • Execution Summary ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ -Statistics: 1 tasks • 1 cache hits • 0 cache misses -Performance: 100% cache hit rate +Statistics: 1 tasks • 0 cache hits • 1 cache misses +Performance: 0% cache hit rate Task Details: ──────────────────────────────────────────────── [1] fspy-shm-capacity#stat: $ vtt stat-many 20000 ✓ - → Cache hit - output replayed - + → Not cached: tracking ran out of room for this task's file accesses ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ```