From befccd899e9c6e5e45294a309f5c55f019b15f92 Mon Sep 17 00:00:00 2001 From: wan9chi Date: Tue, 28 Jul 2026 19:01:55 +0800 Subject: [PATCH] refactor(fspy): simplify shm_io to a plain frame arena MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gated API reads shared memory only when the gate is closed and nothing was in flight, which proves every claimed frame was fully written and published. The three-state frame header (`0` / `+size` / `-size`), the atomic header stores and the release fences existed so a reader could tolerate crashed or in-flight writers — states that are unreachable through the gate. Delete them. A frame is now `| size: u32 | content | padding |`, written once at claim time with a plain store, and the arena keeps exactly one atomic: its end offset. Crash exclusion lives entirely in the gate, where a leaked count fails closed — the run is not cached — instead of being parsed around. The arena also stops being a surface of its own and becomes an internal of the gated module. No public API change: `GatedShmWriter`, `GatedShmReceiver`, `GatedShmReader`, `FrameMut`, `ClaimError` and `StillWriting` keep their signatures, so the channel layer, the fspy crate and both preload clients are untouched. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 1 - crates/fspy_shared/Cargo.toml | 1 - crates/fspy_shared/src/ipc/channel/gated.rs | 233 +++++- .../src/ipc/channel/gated/shm_io.rs | 202 +++++ crates/fspy_shared/src/ipc/channel/mod.rs | 1 - crates/fspy_shared/src/ipc/channel/shm_io.rs | 721 ------------------ crates/fspy_shm/README.md | 21 +- crates/fspy_shm/src/file_backed.rs | 13 - 8 files changed, 423 insertions(+), 770 deletions(-) create mode 100644 crates/fspy_shared/src/ipc/channel/gated/shm_io.rs delete mode 100644 crates/fspy_shared/src/ipc/channel/shm_io.rs diff --git a/Cargo.lock b/Cargo.lock index 366d5ceb..31ffcb5d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1340,7 +1340,6 @@ dependencies = [ name = "fspy_shared" version = "0.0.0" dependencies = [ - "assert2", "bitflags 2.10.0", "bstr", "bumpalo", diff --git a/crates/fspy_shared/Cargo.toml b/crates/fspy_shared/Cargo.toml index 481396fd..439ceb32 100644 --- a/crates/fspy_shared/Cargo.toml +++ b/crates/fspy_shared/Cargo.toml @@ -21,7 +21,6 @@ bytemuck = { workspace = true } winapi = { workspace = true, features = ["std"] } [dev-dependencies] -assert2 = { workspace = true } ctor = { workspace = true } rustc-hash = { workspace = true } subprocess_test = { workspace = true } diff --git a/crates/fspy_shared/src/ipc/channel/gated.rs b/crates/fspy_shared/src/ipc/channel/gated.rs index 3923d082..69ddeedc 100644 --- a/crates/fspy_shared/src/ipc/channel/gated.rs +++ b/crates/fspy_shared/src/ipc/channel/gated.rs @@ -27,12 +27,14 @@ use std::{ sync::{Arc, atomic::AtomicU64}, }; -use wincode::{SchemaWrite, config::DefaultConfig}; +use wincode::{SchemaWrite, Serialize as _, config::DefaultConfig}; -use super::{ - gate::{EnterError, Gate, GateGuard}, - shm_io::{self, AsRawSlice, ShmReader, ShmWriter}, -}; +use self::shm_io::{ShmReader, ShmWriter}; +use super::gate::{EnterError, Gate, GateGuard}; + +mod shm_io; + +pub use shm_io::AsRawSlice; /// Bytes reserved in front of the arena for the gate word. const GATE_REGION_SIZE: usize = 64; @@ -125,8 +127,8 @@ impl GatedShmWriter { /// [`ClaimError::Capacity`] when the arena cannot fit the frame. pub fn claim_frame(&self, size: NonZeroUsize) -> Result, ClaimError> { let gate = self.gate().enter()?; - let frame = self.arena.claim_frame(size).ok_or(ClaimError::Capacity)?; - Ok(FrameMut { frame, _gate: gate }) + let content = self.arena.claim_frame(size).ok_or(ClaimError::Capacity)?; + Ok(FrameMut { content, _gate: gate }) } /// Appends an encoded value as a single frame. @@ -134,20 +136,36 @@ impl GatedShmWriter { /// # Errors /// /// Returns [`ClaimError::Closed`] wrapped in [`WriteEncodedError::Claim`] - /// once the read end has closed the gate, and otherwise whatever the arena - /// reports. + /// once the read end has closed the gate, [`WriteEncodedError::ZeroSizedFrame`] + /// for a value that encodes to nothing, and otherwise the encoder's error. + /// + /// # Panics + /// + /// Panics if the encoder does not fill exactly the size it asked for, which + /// would mean the schema is inconsistent with itself. pub fn write_encoded>( &self, value: &T, ) -> Result<(), WriteEncodedError> { - // The guard is held across the inner write and released only afterwards, - // so the frame is complete before the gate count drops. - let _gate = self.gate().enter().map_err(ClaimError::from)?; - self.arena.write_encoded(value).map_err(|error| match error { - shm_io::WriteEncodedError::EncodeError(error) => WriteEncodedError::Encode(error), - shm_io::WriteEncodedError::ZeroSizedFrame => WriteEncodedError::ZeroSizedFrame, - shm_io::WriteEncodedError::InsufficientSpace => ClaimError::Capacity.into(), - }) + // Enter the gate before touching the value, so a post-close call + // reports `Claim(Closed)` no matter what the value encodes to. + let gate = self.gate().enter().map_err(ClaimError::from)?; + + let serialized_size = + usize::try_from(T::serialized_size(value)?).expect("serialized size exceeds usize"); + let Some(size) = NonZeroUsize::new(serialized_size) else { + return Err(WriteEncodedError::ZeroSizedFrame); + }; + + let content = self.arena.claim_frame(size).ok_or(ClaimError::Capacity)?; + // The frame holds the gate guard, so it is released only after the + // content is written. + let mut frame = FrameMut { content, _gate: gate }; + let mut writer: &mut [u8] = &mut frame; + T::serialize_into(&mut writer, value)?; + assert_eq!(writer.len(), 0); + + Ok(()) } fn gate(&self) -> Gate<'_> { @@ -158,10 +176,11 @@ impl GatedShmWriter { /// An exclusively owned frame together with the gate guard that keeps the region /// open while it is written. /// -/// The frame is declared first so that its completion lands before the gate is -/// released. +/// Dropping it releases the gate, which is what publishes the content to a +/// reader; the arena's own header was already written when the frame was +/// claimed. pub struct FrameMut<'a> { - frame: shm_io::FrameMut<'a>, + content: &'a mut [u8], _gate: GateGuard<'a>, } @@ -169,13 +188,13 @@ impl Deref for FrameMut<'_> { type Target = [u8]; fn deref(&self) -> &Self::Target { - &self.frame + self.content } } impl DerefMut for FrameMut<'_> { fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.frame + self.content } } @@ -407,6 +426,18 @@ mod tests { assert!(reader.iter_frames().all(|frame| frame == b"filler")); } + #[test] + fn write_encoded_after_close_reports_closed_for_any_value() { + let (writer, receiver) = endpoints(1024); + let _frames = receiver.close().unwrap(); + + // A zero-sized value must still report the close, not its own size. + assert!(matches!( + writer.write_encoded(&()), + Err(WriteEncodedError::Claim(ClaimError::Closed)) + )); + } + #[test] fn write_encoded_roundtrips_through_the_gate() { let (writer, receiver) = endpoints(1024); @@ -452,4 +483,162 @@ mod tests { assert_eq!(seen.len(), WRITERS * PER_WRITER); assert_eq!(seen[0], "0 0"); } + + /// Writers that run out of room must fail cleanly rather than corrupt the + /// arena for the ones that fitted. + #[test] + fn concurrent_writers_exceeding_the_arena() { + let (writer, receiver) = endpoints(1024); + + thread::scope(|scope| { + for _ in 0..4 { + scope.spawn(|| { + for _ in 0..10 { + let _ = write_frame(&writer, b"hello"); + let _ = write_frame(&writer, b"foo"); + let _ = write_frame(&writer, b"this is a test"); + } + }); + } + }); + + let reader = receiver.close().unwrap(); + let mut count = 0; + for frame in reader.iter_frames() { + count += 1; + assert!( + frame == b"hello" || frame == b"foo" || frame == b"this is a test", + "unexpected frame {:?}", + from_utf8(frame) + ); + } + assert!(count > 20, "expected most writes to fit, got {count}"); + } + + /// Several threads racing for the last few bytes must all agree on who got + /// them, and the reader must walk whatever survived without panicking. + #[test] + fn concurrent_writers_racing_for_the_last_bytes() { + let (writer, receiver) = endpoints(GATE_REGION_SIZE + 200); + + thread::scope(|scope| { + for _ in 0..10 { + scope.spawn(|| { + let _ = write_frame( + &writer, + b"this_is_a_moderately_long_frame_that_might_cause_races", + ); + }); + } + }); + + let reader = receiver.close().unwrap(); + let count = reader.iter_frames().count(); + // Some but not all of them fit into 200 bytes of arena. + assert!(count > 0); + assert!(count < 10); + } + + /// A frame larger than the size header can describe is refused, and the + /// arena stays usable afterwards. + #[test] + fn a_frame_too_large_for_the_header_is_refused() { + let (writer, receiver) = endpoints(1024); + + let oversized = NonZeroUsize::new(usize::try_from(u32::MAX).unwrap() + 1).unwrap(); + assert!(matches!(writer.claim_frame(oversized), Err(ClaimError::Capacity))); + + write_frame(&writer, b"test").unwrap(); + let reader = receiver.close().unwrap(); + assert_eq!(reader.iter_frames().collect::>(), vec![b"test".as_slice()]); + } + + #[test] + fn a_misaligned_region_is_rejected() { + struct Misaligned(MockRegion); + + 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: shifting by one byte deliberately misaligns the + // region and stays inside the over-allocated backing buffer. + unsafe { raw_slice.cast::().add(1) }, + raw_slice.len() - 1, + ) + } + } + + let misaligned = Misaligned(MockRegion::alloc(1024)); + assert_ne!(misaligned.as_raw_slice().cast::() as usize % align_of::(), 0); + + let result = std::panic::catch_unwind(|| { + // SAFETY: the region is deliberately misaligned to check that the + // constructor catches it; the assertion fires before any access. + unsafe { GatedShmWriter::new(misaligned) }; + }); + + assert!(result.is_err(), "a misaligned region must be rejected"); + } + + /// The same protocol over a real shared mapping, written by several + /// processes at once. + #[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; + const SHM_SIZE: usize = 1024 * 1024; + + let (keeper, handle) = fspy_shm::create(SHM_SIZE).unwrap(); + let shm_id = keeper.id().to_str().expect("test temp dir is UTF-8").to_owned(); + let mapping = handle.map().unwrap(); + // SAFETY: the backing file was just created zero-initialized, the + // mapping stays valid while the receiver holds it, and only this + // protocol reaches it. + let receiver = unsafe { GatedShmReceiver::new(mapping) }; + + let children: Vec = (0..CHILD_COUNT) + .map(|child_index| { + let cmd = command_for_fn!((shm_id.clone(), child_index), |( + shm_id, + child_index, + ): ( + String, + usize + )| { + let mapping = + fspy_shm::open(std::ffi::OsStr::new(&shm_id)).unwrap().map().unwrap(); + // SAFETY: the mapping was created under the same + // contract and only this protocol reaches it. + let writer = unsafe { GatedShmWriter::new(mapping) }; + for i in 0..FRAME_COUNT_EACH_CHILD { + let frame_data = format!("{child_index} {i}"); + let size = NonZeroUsize::new(frame_data.len()).unwrap(); + writer.claim_frame(size).unwrap().copy_from_slice(frame_data.as_bytes()); + } + }); + Command::from(cmd).spawn().unwrap() + }) + .collect(); + + for mut child in children { + assert!(child.wait().unwrap().success()); + } + + let reader = receiver.close().unwrap(); + let frames: FxHashSet<&str> = + reader.iter_frames().map(|frame| from_utf8(frame).unwrap()).collect(); + assert_eq!(frames.len(), CHILD_COUNT * FRAME_COUNT_EACH_CHILD); + for child_index in 0..CHILD_COUNT { + for i in 0..FRAME_COUNT_EACH_CHILD { + assert!(frames.contains(format!("{child_index} {i}").as_str())); + } + } + } } diff --git a/crates/fspy_shared/src/ipc/channel/gated/shm_io.rs b/crates/fspy_shared/src/ipc/channel/gated/shm_io.rs new file mode 100644 index 00000000..599c83b2 --- /dev/null +++ b/crates/fspy_shared/src/ipc/channel/gated/shm_io.rs @@ -0,0 +1,202 @@ +//! A lock-free frame arena in a shared memory region. +//! +//! Region layout: +//! +//! ```text +//! | total byte size of frames: AtomicUsize | frame | frame | ... | +//! ``` +//! +//! and every frame is +//! +//! ```text +//! | size: u32 | content | padding to the next 4-byte boundary | +//! ``` +//! +//! One atomic word — the end offset — carries the whole arena. Writers claim +//! disjoint ranges out of it with a single `fetch_update`, then write their +//! header once with a plain store: the claim already made the range exclusive, +//! and what publishes it to the reader is the gate release in the parent module, +//! not the header store. +//! +//! There is exactly one frame state, so nothing here tolerates a crashed or +//! half-finished writer. It does not have to: [`super::GatedShmReceiver::close`] +//! hands out a reader only when every claim it admitted ran to completion, and a +//! writer that dies mid-frame leaks its gate count instead, so the run is never +//! read at all. + +use core::iter::from_fn; +use std::{ + num::NonZeroUsize, + ptr::slice_from_raw_parts_mut, + slice, + sync::atomic::{AtomicUsize, Ordering}, +}; + +use bytemuck::must_cast; +use fspy_shm::Mapping; + +/// Type of a frame's size header. +type FrameSize = u32; + +// The end offset is written with atomic operations and read back with a plain +// load, so the two layouts have to agree. +const _: () = { + 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()) + } +} + +#[track_caller] +fn assert_alignment(ptr: *const u8) { + // The arena's end offset is a `usize`. + assert_eq!(ptr as usize % align_of::(), 0); + // Every frame header that follows it is a `FrameSize`. + assert_eq!((ptr as usize + size_of::()) % align_of::(), 0); +} + +const fn roundup_to_align_frame_header(size: usize) -> usize { + size.next_multiple_of(align_of::()) +} + +/// A concurrent frame arena writer. +/// +/// Lock-free and safe to use from several threads and processes at once: the end +/// offset gives every claim a range of its own. +pub(super) struct ShmWriter { + mem: M, +} + +impl ShmWriter { + /// Creates a writer over a shared memory region. + /// + /// # Safety + /// - `mem.as_raw_slice()` must return a stable valid pointer to the region, + /// - the region must only be accessed via `ShmWriter` and `ShmReader` across + /// all the processes, + /// - the region must be zero-initialized before first use. + pub(super) unsafe fn new(mem: M) -> Self { + assert_alignment(mem.as_raw_slice() as *const u8); + Self { mem } + } + + /// Claims `frame_size` bytes of frame content and returns them. + /// + /// Returns `None` when the arena has no room left, or when the frame is + /// larger than a size header can describe. + /// + /// The header is written before returning. Publishing the content to a + /// reader is the caller's job, and its gate guard does that on drop. + #[expect( + clippy::mut_from_ref, + reason = "every claim carves out a range no other claim can reach, which is what makes \ + the exclusive borrow sound" + )] + #[expect( + clippy::cast_ptr_alignment, + reason = "`new` asserts the base alignment and the claim keeps frame headers aligned" + )] + pub(super) fn claim_frame(&self, frame_size: NonZeroUsize) -> Option<&mut [u8]> { + let shm_slice: *mut [u8] = self.mem.as_raw_slice(); + let shm_ptr = shm_slice.cast::(); + let shm_len = shm_slice.len(); + + let frame_size = frame_size.get(); + // The header describes a frame's size in a `FrameSize`, so that is also + // the largest frame this arena can hold. + let Ok(frame_size_header) = FrameSize::try_from(frame_size) else { + return None; + }; + + // The end position of all frames lives in the region's first machine word. + // SAFETY: `shm_ptr` points to the start of the region, which is aligned to + // `usize` (verified by `assert_alignment` in `new`) and large enough to + // contain at least a `usize` header. + let end_offset = unsafe { AtomicUsize::from_ptr(shm_ptr.cast()) }; + + let frame_with_header_size = size_of::() + frame_size; + + // Claim the space. Writers share only this word, never each other's + // content, so relaxed ordering is enough. + let current_end = end_offset + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current_end| { + // Round up so that the next frame's header stays aligned. + let new_end = roundup_to_align_frame_header(current_end + frame_with_header_size); + if size_of::() + new_end > shm_len { + return None; + } + Some(new_end) + }) + .ok()?; + + // SAFETY: the atomic claim above guaranteed that + // `size_of::() + current_end` is within the region's bounds, so + // this pointer arithmetic stays inside the allocation. + let frame_start = unsafe { shm_ptr.add(size_of::() + current_end) }; + + // SAFETY: `frame_start` is aligned for a `FrameSize` (ensured by + // `roundup_to_align_frame_header`) and lies inside the range this writer + // just claimed exclusively, so nothing else can race this store. + unsafe { frame_start.cast::().write(frame_size_header) }; + + // SAFETY: skipping the header stays inside the claimed range. + let content_ptr = unsafe { frame_start.add(size_of::()) }; + // SAFETY: `content_ptr` is valid for `frame_size` bytes (guaranteed by the + // atomic claim), aligned for `u8`, and no other writer will touch this + // range because every writer claims a range of its own. + Some(unsafe { slice::from_raw_parts_mut(content_ptr, frame_size) }) + } +} + +/// Reader of the frames an [`ShmWriter`] appended. +/// +/// `mem` must cover everything a writer produced and must no longer be written. +/// `M: AsRef<[u8]>` makes the local half of that borrow-checked; the +/// cross-process half is the caller's responsibility. Here it is discharged by +/// [`super::GatedShmReceiver::close`], which produces a reader only once the gate +/// proved that every claimed frame was fully written and published. +pub(super) struct ShmReader> { + mem: M, +} + +impl> ShmReader { + pub(super) fn new(mem: M) -> Self { + assert_alignment(mem.as_ref().as_ptr()); + Self { mem } + } + + /// Iterates over every frame in the arena, in claim order. + pub(super) 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_header, next_remaining_content) = + remaining_content.split_first_chunk::<{ size_of::() }>()?; + let frame_size = usize::try_from(must_cast::<_, FrameSize>(*frame_header)) + .expect("frame size exceeds usize"); + // Claims are `NonZeroUsize`, so a zero header means the region does + // not hold what this reader was promised. Say so rather than loop. + assert!(frame_size != 0, "shared memory holds a frame of size zero"); + + let (frame_with_padding, next_remaining_content) = + next_remaining_content.split_at(roundup_to_align_frame_header(frame_size)); + remaining_content = next_remaining_content; + + Some(&frame_with_padding[..frame_size]) + }) + } +} diff --git a/crates/fspy_shared/src/ipc/channel/mod.rs b/crates/fspy_shared/src/ipc/channel/mod.rs index 61ea485d..23024b8b 100644 --- a/crates/fspy_shared/src/ipc/channel/mod.rs +++ b/crates/fspy_shared/src/ipc/channel/mod.rs @@ -13,7 +13,6 @@ mod gate; mod gated; -mod shm_io; use std::io; 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 42970486..00000000 --- a/crates/fspy_shared/src/ipc/channel/shm_io.rs +++ /dev/null @@ -1,721 +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.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current_end| { - let new_end = roundup_to_align_frame_header(current_end + frame_with_header_size); - - // Check if we have enough space - if size_of::() + new_end > shm_len { - return None; - } - - Some(new_end) - }); - - let Ok(current_end) = current_end else { - return None; // Not enough space - }; - - #[cfg(test)] - if self.fail_on_claim { - // Simulate crash right after claiming the space - return None; - } - - // Successfully claimed the space, now write the data - - // SAFETY: The atomic fetch_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 fetch_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 (keeper, handle) = fspy_shm::create(SHM_SIZE).unwrap(); - let shm_name = keeper.id().to_str().expect("test temp dir is UTF-8").to_owned(); - // 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 mapping = - fspy_shm::open(std::ffi::OsStr::new(&shm_name)).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_shm/README.md b/crates/fspy_shm/README.md index d63fb935..0e93f742 100644 --- a/crates/fspy_shm/README.md +++ b/crates/fspy_shm/README.md @@ -8,17 +8,16 @@ The public API is defined in [`src/lib.rs`](src/lib.rs). -| API | Contract | -| --------------------- | --------------------------------------------------------------------------------------- | -| `create(size)` | Creates a zero-initialized backing file and returns its `ShmKeeper` and an `ShmHandle`. | -| `open(id)` | Opens an `ShmHandle` on the shared memory identified by `id`. | -| `ShmKeeper::id()` | Returns the identifier another process passes to `open`. | -| `ShmHandle::map()` | Maps the shared bytes. Callable more than once. | -| `Mapping::len()` | Returns the mapped size. | -| `Mapping::as_ptr()` | Returns a mutable raw pointer to the first byte. | -| `Mapping::as_slice()` | Returns the bytes as a shared slice. The caller must prevent mutation for its lifetime. | - -`ShmKeeper` is the name: while it lives, `open` succeeds, and dropping it removes the backing file. `ShmHandle` is the opened file: `create` returns one so the creator never looks its own file up by name, and `open` returns one to everybody else. `Mapping` is the bytes: it keeps them alive until dropped and can do nothing else. None of the three synchronizes memory access. The fspy channel adds that on top with atomic frame headers and a close gate stored in the mapping itself: every writer holds a gate guard while it writes, and the receiver closes the gate with one atomic operation, which refuses every later writer and reports whether any write was still in flight. +| API | Contract | +| ------------------- | --------------------------------------------------------------------------------------- | +| `create(size)` | Creates a zero-initialized backing file and returns its `ShmKeeper` and an `ShmHandle`. | +| `open(id)` | Opens an `ShmHandle` on the shared memory identified by `id`. | +| `ShmKeeper::id()` | Returns the identifier another process passes to `open`. | +| `ShmHandle::map()` | Maps the shared bytes. Callable more than once. | +| `Mapping::len()` | Returns the mapped size. | +| `Mapping::as_ptr()` | Returns a mutable raw pointer to the first byte. | + +`ShmKeeper` is the name: while it lives, `open` succeeds, and dropping it removes the backing file. `ShmHandle` is the opened file: `create` returns one so the creator never looks its own file up by name, and `open` returns one to everybody else. `Mapping` is the bytes: it keeps them alive until dropped and can do nothing else. None of the three synchronizes memory access or dereferences the mapping, so the crate has no safety conditions of its own. All access rules live one layer up, in the fspy channel: a single atomic end-offset word hands every writer a frame range of its own, and a close gate stored in the same mapping tracks writers. Every writer holds a gate guard while it writes. The receiver closes the gate with one atomic operation, which refuses every later writer and reports whether any write was still in flight. Every byte in a mapping returned by `create` is initially zero. `open` exposes the mapping's current contents and does not reinitialize them. diff --git a/crates/fspy_shm/src/file_backed.rs b/crates/fspy_shm/src/file_backed.rs index 778bd5eb..844e6a84 100644 --- a/crates/fspy_shm/src/file_backed.rs +++ b/crates/fspy_shm/src/file_backed.rs @@ -186,19 +186,6 @@ impl Mapping { pub fn as_ptr(&self) -> *mut u8 { self.raw.as_mut_ptr() } - - /// Returns the mapped bytes as a shared slice. - /// - /// # Safety - /// - /// The caller must ensure that no process or thread mutates the mapping for - /// the lifetime of the returned slice. - #[must_use] - pub unsafe fn as_slice(&self) -> &[u8] { - // SAFETY: The mapping is valid for its full length, and the caller - // guarantees that it is not mutated while the slice is borrowed. - unsafe { std::slice::from_raw_parts(self.as_ptr().cast_const(), self.len()) } - } } #[cfg(test)]