//! miranda-ipc — Work Order 1: the POSIX shared-memory ring buffer at //! `/dev/shm/miranda_bus`, lock-free atomic head/tail pointers, and C-ABI //! aligned payload structs for audio chunks, 52-channel ARKit blendshape //! frames, 9-coefficient spherical harmonic vectors, and WO-4 //! `KinematicTransformFrame` quaternions for neck/clavicle routing. //! //! # What this is //! //! Four independent single-producer / single-consumer (SPSC) ring buffers //! sharing one memory mapping. Each ring is lock-free: coordination is a pair //! of `AtomicUsize` counters (`head` = read position, `tail` = write //! position), never a mutex. The bus is on the critical path of a 60 FPS //! render loop, so a mutex held by a stalled audio thread would cause the //! renderer to miss its frame deadline. That is a latency constraint, not a //! style preference (WO-1 REQ-3). //! //! # Concurrency contract — read this before using the bus //! //! Each ring is **SPSC**: at most one producer thread and at most one //! consumer thread per ring, concurrently. The three rings are independent, //! so an audio producer, a blendshape producer, and a renderer consumer may //! all run concurrently without coordination. What is **not** supported is //! two threads pushing to the *same* ring — that would race on `tail` and //! silently interleave slot writes. If a future Work Order needs multiple //! producers on one ring, it needs a different algorithm (MPSC), not a //! tweak to this one. //! //! # Layout //! //! ```text //! offset 0 : audio head (AtomicUsize, own 64-byte cache line) //! offset 64 : audio tail (AtomicUsize, own 64-byte cache line) //! offset 128 : AudioChunk slots × 64 //! ... : blendshape head / tail / BlendshapeFrame slots × 128 //! ... : sh head / tail / SphericalHarmonics slots × 128 //! ``` //! //! `head` and `tail` for a given ring are deliberately placed on **separate //! 64-byte cache lines**. If they shared a line, the producer's store to //! `tail` would invalidate the consumer's cached copy of `head` on every //! single push (and vice versa) — "false sharing" — which turns two //! independent atomic counters into a ping-pong of cache-coherency traffic //! and can cost an order of magnitude in throughput. This matters most on //! ARM64/Graviton, where the memory model requires real fence instructions //! rather than x86's near-free TSO ordering. use std::fs::OpenOptions; use std::io; use std::path::Path; use std::sync::atomic::{AtomicUsize, Ordering}; use bytemuck::Pod; use memmap2::MmapMut; use miranda_core::{ AudioChunk, BackpressureError, BlendshapeFrame, KinematicTransformFrame, RingId, SphericalHarmonics, }; /// Canonical path for the production bus. `/dev/shm` is a tmpfs mount /// (RAM-backed), which is what makes the ≤50 μs round-trip target /// achievable — a regular on-disk file cannot meet it (WO-1 REQ-2). pub const BUS_PATH: &str = "/dev/shm/miranda_bus"; /// Cache line size assumed for false-sharing avoidance. 64 bytes on both /// x86-64 and the Graviton (Neoverse) cores this project targets. const CACHE_LINE: usize = 64; /// Bytes reserved for one ring's control block: `head` on its own cache /// line, then `tail` on its own cache line. const CTRL_BYTES: usize = 2 * CACHE_LINE; /// Slot counts. All powers of two so the modulo that maps a monotonic /// counter to a slot index becomes a single bitmask (`counter & (slots - 1)`) /// instead of an integer division on the hot path. const AUDIO_SLOTS: usize = 64; const BLEND_SLOTS: usize = 128; const SH_SLOTS: usize = 128; /// 128 slots × 88 bytes = 11,264 bytes. Matches the blendshape ring depth /// because every blendshape frame should have a companion kinematic frame. const KINEMATIC_SLOTS: usize = 128; /// Rounds `value` up to the next multiple of `align`. `align` must be a /// power of two. const fn align_up(value: usize, align: usize) -> usize { (value + align - 1) & !(align - 1) } // Region offsets, all derived from `size_of` rather than hardcoded — the // design doc's stated byte sizes are informational and one of them (44 for // SphericalHarmonics) is arithmetically wrong once `#[repr(C)]` alignment is // applied. Deriving from `size_of` makes that impossible to get wrong here. const AUDIO_CTRL_OFF: usize = 0; const AUDIO_DATA_OFF: usize = align_up(AUDIO_CTRL_OFF + CTRL_BYTES, CACHE_LINE); const AUDIO_DATA_BYTES: usize = AUDIO_SLOTS * std::mem::size_of::(); const BLEND_CTRL_OFF: usize = align_up(AUDIO_DATA_OFF + AUDIO_DATA_BYTES, CACHE_LINE); const BLEND_DATA_OFF: usize = align_up(BLEND_CTRL_OFF + CTRL_BYTES, CACHE_LINE); const BLEND_DATA_BYTES: usize = BLEND_SLOTS * std::mem::size_of::(); const SH_CTRL_OFF: usize = align_up(BLEND_DATA_OFF + BLEND_DATA_BYTES, CACHE_LINE); const SH_DATA_OFF: usize = align_up(SH_CTRL_OFF + CTRL_BYTES, CACHE_LINE); const SH_DATA_BYTES: usize = SH_SLOTS * std::mem::size_of::(); const KINEMATIC_CTRL_OFF: usize = align_up(SH_DATA_OFF + SH_DATA_BYTES, CACHE_LINE); const KINEMATIC_DATA_OFF: usize = align_up(KINEMATIC_CTRL_OFF + CTRL_BYTES, CACHE_LINE); const KINEMATIC_DATA_BYTES: usize = KINEMATIC_SLOTS * std::mem::size_of::(); /// Total size of the shared-memory region, rounded up to a cache line. pub const BUS_TOTAL_BYTES: usize = align_up(KINEMATIC_DATA_OFF + KINEMATIC_DATA_BYTES, CACHE_LINE); // Compile-time proof of every layout invariant the unsafe code below relies // on. If any of these ever becomes false (someone adds a field, changes a // slot count to a non-power-of-two, reorders a region), this fails the build // rather than producing a misaligned atomic or a torn read at runtime. const _: () = { assert!(AUDIO_SLOTS.is_power_of_two()); assert!(BLEND_SLOTS.is_power_of_two()); assert!(SH_SLOTS.is_power_of_two()); assert!(KINEMATIC_SLOTS.is_power_of_two()); // Control blocks must be aligned for AtomicUsize. assert!(AUDIO_CTRL_OFF % std::mem::align_of::() == 0); assert!(BLEND_CTRL_OFF % std::mem::align_of::() == 0); assert!(SH_CTRL_OFF % std::mem::align_of::() == 0); assert!(KINEMATIC_CTRL_OFF % std::mem::align_of::() == 0); // ...and head/tail must not share a cache line. assert!(CTRL_BYTES >= 2 * CACHE_LINE); assert!(std::mem::size_of::() <= CACHE_LINE); // Every slot region must be aligned for its payload type, and every // slot within it too (which follows if the region is aligned and the // slot size is a multiple of the payload alignment). assert!(AUDIO_DATA_OFF % std::mem::align_of::() == 0); assert!(std::mem::size_of::() % std::mem::align_of::() == 0); assert!(BLEND_DATA_OFF % std::mem::align_of::() == 0); assert!(std::mem::size_of::() % std::mem::align_of::() == 0); assert!(SH_DATA_OFF % std::mem::align_of::() == 0); assert!( std::mem::size_of::() % std::mem::align_of::() == 0 ); assert!(KINEMATIC_DATA_OFF % std::mem::align_of::() == 0); assert!( std::mem::size_of::() % std::mem::align_of::() == 0 ); // Regions must not overlap. assert!(AUDIO_DATA_OFF + AUDIO_DATA_BYTES <= BLEND_CTRL_OFF); assert!(BLEND_DATA_OFF + BLEND_DATA_BYTES <= SH_CTRL_OFF); assert!(SH_DATA_OFF + SH_DATA_BYTES <= KINEMATIC_CTRL_OFF); assert!(KINEMATIC_DATA_OFF + KINEMATIC_DATA_BYTES <= BUS_TOTAL_BYTES); }; /// A 64-byte, 64-byte-aligned block. Used only to obtain a cache-line-aligned /// heap allocation for the in-memory backing (see [`MirandaBus::in_memory`]); /// a plain `Vec` only guarantees 1-byte alignment, which is not enough to /// place an `AtomicUsize` soundly. #[repr(C, align(64))] #[derive(Clone, Copy)] struct CacheLineBlock([u8; CACHE_LINE]); /// Where the bus bytes actually live. /// /// Neither variant's payload is read after construction — `base` is the /// pointer everything else uses. This enum exists purely to keep the /// backing storage (`MmapMut` or `Vec`) alive for as long as `MirandaBus` /// does, so `base` never dangles. `#[allow(dead_code)]` documents that this /// is intentional (an owner held only for its `Drop`/liveness side effect), /// not an oversight — the alternative would be a `Box` that hides /// the same fact behind a vtable instead of stating it plainly. #[allow(dead_code)] enum Backing { /// Production: a `mmap` of a tmpfs file, shareable across processes. Mmap(MmapMut), /// Testing: a private heap allocation. Needed because MIRI cannot /// execute the `mmap` syscall, so the ring-buffer logic would otherwise /// be entirely unverifiable by Rust's undefined-behaviour checker. /// Same layout, same code paths, no syscall. Heap(Vec), } /// The Miranda IPC bus: three lock-free SPSC ring buffers over one shared /// memory region. pub struct MirandaBus { /// Kept alive for as long as the bus exists; `base` points into it. _backing: Backing, /// Base address of the region. Valid for `BUS_TOTAL_BYTES`. base: *mut u8, } // SAFETY: `MirandaBus` contains a raw pointer, which makes it neither `Send` // nor `Sync` by default. Both are sound here: // // - The pointed-to region stays valid and at a fixed address for the whole // life of the `MirandaBus`, because `_backing` owns it and neither variant // stores the bytes inline: `MmapMut` refers to an OS mapping, and // `Vec` refers to a heap allocation. Moving the // `MirandaBus` therefore does not move the bytes, so `base` cannot dangle. // Nothing after construction ever mutates `_backing` (no push/reallocate), // so the heap buffer cannot be reallocated out from under `base`. // - All shared mutation goes through `AtomicUsize` (interior mutability, safe // under `&self`) or through raw-pointer writes to slots that the SPSC // protocol guarantees are exclusively owned by the calling side at that // moment — see the `// SAFETY:` comment in `ring_push`. // // If this were violated — e.g. by adding a method that reallocates the Vec, // or by allowing two producers on one ring — the result would be a dangling // pointer or a data race on slot bytes, i.e. undefined behaviour. unsafe impl Send for MirandaBus {} unsafe impl Sync for MirandaBus {} impl MirandaBus { /// Opens the production bus at [`BUS_PATH`], creating it if absent. /// /// A freshly created file is zero-filled by the kernel, and all-zero is a /// valid initial state for this layout (`head == tail == 0`, i.e. empty; /// all slots zeroed). That means no explicit initialisation pass is /// needed, which conveniently sidesteps the "how do you initialise an /// atomic you did not allocate" problem entirely. /// /// Reopening an existing bus inherits whatever `head`/`tail` the previous /// user left. That is consistent (never torn) but the contents are stale; /// callers that need a pristine bus should remove the file first or use /// [`MirandaBus::open_or_create_at`] with a unique path. pub fn open_or_create() -> io::Result { Self::open_or_create_at(BUS_PATH) } /// Same as [`MirandaBus::open_or_create`] but at an explicit path. /// /// Tests must use this with a unique path per test: `cargo test` runs /// tests in parallel by default, and two tests sharing one bus would /// interleave pushes on the same ring, violating the SPSC contract and /// producing spurious failures that look like a ring-buffer bug. pub fn open_or_create_at>(path: P) -> io::Result { let file = OpenOptions::new() .read(true) .write(true) .create(true) .truncate(false) .open(path)?; file.set_len(BUS_TOTAL_BYTES as u64)?; // SAFETY: `MmapMut::map_mut` is unsafe because the mapping's contents // can be mutated by any other process with access to the same file, // so Rust cannot statically guarantee exclusivity. That is precisely // the intent here — this is an interprocess bus. Every access below // goes through atomics or raw-pointer copies sized within // `BUS_TOTAL_BYTES`, and `set_len` above guarantees the file is at // least that large, so no access can fall outside the mapping. If the // file were shorter than `BUS_TOTAL_BYTES`, touching a slot past the // end would fault (SIGBUS). let mut mmap = unsafe { MmapMut::map_mut(&file)? }; let base = mmap.as_mut_ptr(); Ok(Self { _backing: Backing::Mmap(mmap), base, }) } /// Creates a private, process-local bus on the heap with the identical /// layout and code paths, but no `mmap` syscall. /// /// This exists so the ring-buffer logic is verifiable under MIRI. MIRI /// interprets Rust rather than executing real syscalls, so it cannot run /// `mmap` at all — without this backing, the atomic ordering and pointer /// arithmetic in `ring_push`/`ring_pop` could never be checked for /// undefined behaviour, which for a lock-free structure is the only check /// that really matters. pub fn in_memory() -> Self { let blocks = BUS_TOTAL_BYTES / CACHE_LINE; debug_assert_eq!(BUS_TOTAL_BYTES % CACHE_LINE, 0); let mut backing = vec![CacheLineBlock([0u8; CACHE_LINE]); blocks]; let base = backing.as_mut_ptr() as *mut u8; Self { _backing: Backing::Heap(backing), base, } } /// Borrows the `head` counter (read position) of the ring whose control /// block starts at `ctrl_off`. fn head(&self, ctrl_off: usize) -> &AtomicUsize { // SAFETY: Casting `base + ctrl_off` to `*const AtomicUsize` and // dereferencing it is sound because: // // - Alignment: `base` is page-aligned (mmap) or 64-byte aligned // (`CacheLineBlock`), and `ctrl_off` is one of the three // `*_CTRL_OFF` constants, each asserted at compile time above to be // a multiple of `align_of::()`. So the address is // correctly aligned for an atomic. // - Validity: `ctrl_off + size_of::()` is well inside // `BUS_TOTAL_BYTES` (asserted above: regions do not overlap and all // fit), and the backing is guaranteed to be at least that large. // - Initialisation: the region is zero-filled at creation (kernel for // a fresh tmpfs file, `vec![0u8; _]` for the heap backing), and // every bit pattern is a valid `usize`, so the atomic is never read // uninitialised. // - Aliasing: the returned `&AtomicUsize` permits concurrent // modification through interior mutability, which is exactly what // two threads (or processes) need here. No `&mut` to this address // is ever created. // // If alignment were violated, atomic operations on x86 can tear and // on ARM64 will fault outright; if the offset were out of bounds we // would read or write unmapped memory. unsafe { &*(self.base.add(ctrl_off) as *const AtomicUsize) } } /// Borrows the `tail` counter (write position) of the ring whose control /// block starts at `ctrl_off`. Placed one full cache line after `head` to /// prevent false sharing between producer and consumer. fn tail(&self, ctrl_off: usize) -> &AtomicUsize { // SAFETY: Identical reasoning to `head` above. The address is // `ctrl_off + CACHE_LINE`; `CACHE_LINE` (64) is a multiple of // `align_of::()` (8), so adding it preserves alignment, // and `CTRL_BYTES` (128) reserves room for both counters, asserted at // compile time to be at least `2 * CACHE_LINE`. unsafe { &*(self.base.add(ctrl_off + CACHE_LINE) as *const AtomicUsize) } } /// Pushes one payload into a ring. Returns `Err` if the ring is full — /// never overwrites unread data, never silently drops (WO-1 REQ-6). fn ring_push( &self, ctrl_off: usize, data_off: usize, slots: usize, ring: RingId, value: T, ) -> Result<(), BackpressureError> { let tail = self.tail(ctrl_off); let head = self.head(ctrl_off); // Only this (single) producer writes `tail`, so a Relaxed load of our // own counter is sufficient — nobody else can change it. let current_tail = tail.load(Ordering::Relaxed); // `head` is written by the consumer, so this load must be Acquire. // // Deliberate deviation from design.md, which specifies Relaxed here: // Relaxed would be adequate for the *capacity check* alone (a stale, // smaller `head` only makes the producer more conservative, never less // safe). But this load also has to establish that the consumer has // finished reading the slot we are about to overwrite. The consumer // signals completion with a Release store to `head`; pairing it with // a Relaxed load creates no happens-before edge, so the consumer's // reads and our overwrite of the same slot bytes would be a data race // — undefined behaviour under the Rust/C++ memory model even though // it happens to work on x86's TSO hardware. Acquire creates the edge. // This is the textbook SPSC pairing and still satisfies REQ-3's // "AcqRel/Acquire/Release, never SeqCst" constraint. let current_head = head.load(Ordering::Acquire); if current_tail.wrapping_sub(current_head) >= slots { return Err(BackpressureError { ring, capacity: slots, }); } let index = current_tail & (slots - 1); let slot_off = data_off + index * std::mem::size_of::(); // SAFETY: Writing `size_of::()` bytes at `base + slot_off`. // // - Bounds: `index < slots` because of the bitmask, and the compile- // time assertions guarantee `data_off + slots * size_of::()` // fits inside `BUS_TOTAL_BYTES`, so the whole write is in bounds. // - Exclusivity: the capacity check above proved // `tail - head < slots`, which means slot `tail & (slots-1)` is not // one of the slots in `[head, tail)` that the consumer may be // reading. Combined with the SPSC contract (one producer per ring), // this producer has exclusive access to this slot until it // publishes below. // - Validity of the source: `bytes_of` gives a byte view of a local, // fully-initialised `T: Pod` — `Pod` is the compile-time proof that // `T` has no padding and no invalid bit patterns, so every byte is // meaningful and the round-trip is lossless. // - Non-overlap: source is a stack local, destination is inside the // mapping; they cannot overlap. // // Raw `copy_nonoverlapping` is used rather than building a // `&mut [u8]` over the slot: constructing a Rust reference into memory // another process may hold a reference to would assert exclusivity we // do not actually have. A raw pointer write asserts nothing. // // If exclusivity were violated (two producers, or a wrong capacity // check), the consumer could observe a half-written frame — the exact // silent corruption this module exists to prevent. unsafe { let src = bytemuck::bytes_of(&value); std::ptr::copy_nonoverlapping( src.as_ptr(), self.base.add(slot_off), std::mem::size_of::(), ); } // Release: publishes the slot write above. Any consumer that observes // this new `tail` with an Acquire load is guaranteed to see the fully // written slot, never a partial one. tail.store(current_tail.wrapping_add(1), Ordering::Release); Ok(()) } /// Pops one payload from a ring, or `None` if empty. fn ring_pop(&self, ctrl_off: usize, data_off: usize, slots: usize) -> Option { let tail = self.tail(ctrl_off); let head = self.head(ctrl_off); // Only this (single) consumer writes `head`. let current_head = head.load(Ordering::Relaxed); // Acquire pairs with the producer's Release store to `tail`; this is // what guarantees the slot bytes we are about to read are the fully // written ones. let current_tail = tail.load(Ordering::Acquire); if current_head == current_tail { return None; } let index = current_head & (slots - 1); let slot_off = data_off + index * std::mem::size_of::(); // SAFETY: Reading `size_of::()` bytes at `base + slot_off` into a // zeroed local. // // - Bounds: same argument as `ring_push` — `index < slots` via the // bitmask, region fits inside `BUS_TOTAL_BYTES` by compile-time // assertion. // - Exclusivity: `current_head != current_tail` proved this slot is in // the readable range `[head, tail)`. The producer's capacity check // guarantees it will not overwrite any slot in that range, so this // consumer has exclusive read access until it publishes below. // - Validity: the Acquire load of `tail` above happens-before this // read and after the producer's slot write, so the bytes are fully // initialised. `T: Pod` guarantees any bit pattern is a valid `T`, // so even a stale-but-complete slot yields a well-formed value. // - Destination is a fresh local; source is inside the mapping; no // overlap. // // `copy_nonoverlapping` into `T::zeroed()` is used instead of // `bytemuck::from_bytes`, which would require materialising a // `&[u8]` over shared-memory bytes that another process is permitted // to write concurrently — a shared Rust reference asserts the // referent will not change, which is not true of an interprocess // mapping. The raw copy makes no such assertion. (Deliberate // deviation from design.md's suggested `from_bytes` call, for // soundness; `bytes_of` is still used on the write path as specified.) // // If exclusivity were violated, this would read a torn frame — for // BlendshapeFrame that means EVE's face driven by half of one // expression and half of another. let value = unsafe { let mut out = T::zeroed(); std::ptr::copy_nonoverlapping( self.base.add(slot_off), &mut out as *mut T as *mut u8, std::mem::size_of::(), ); out }; // Release: publishes that this slot is free. A producer that observes // this new `head` with an Acquire load is guaranteed to see our read // as complete, so it may safely overwrite the slot. head.store(current_head.wrapping_add(1), Ordering::Release); Some(value) } /// Pushes an [`AudioChunk`] onto the audio ring. pub fn push_audio(&self, chunk: AudioChunk) -> Result<(), BackpressureError> { self.ring_push( AUDIO_CTRL_OFF, AUDIO_DATA_OFF, AUDIO_SLOTS, RingId::Audio, chunk, ) } /// Pops an [`AudioChunk`] from the audio ring. pub fn pop_audio(&self) -> Option { self.ring_pop(AUDIO_CTRL_OFF, AUDIO_DATA_OFF, AUDIO_SLOTS) } /// Pushes a [`BlendshapeFrame`] onto the blendshape ring. pub fn push_blendshape(&self, frame: BlendshapeFrame) -> Result<(), BackpressureError> { self.ring_push( BLEND_CTRL_OFF, BLEND_DATA_OFF, BLEND_SLOTS, RingId::Blendshape, frame, ) } /// Pops a [`BlendshapeFrame`] from the blendshape ring. pub fn pop_blendshape(&self) -> Option { self.ring_pop(BLEND_CTRL_OFF, BLEND_DATA_OFF, BLEND_SLOTS) } /// Pushes a [`SphericalHarmonics`] vector onto the lighting ring. pub fn push_sh(&self, sh: SphericalHarmonics) -> Result<(), BackpressureError> { self.ring_push( SH_CTRL_OFF, SH_DATA_OFF, SH_SLOTS, RingId::SphericalHarmonics, sh, ) } /// Pops a [`SphericalHarmonics`] vector from the lighting ring. pub fn pop_sh(&self) -> Option { self.ring_pop(SH_CTRL_OFF, SH_DATA_OFF, SH_SLOTS) } /// Pushes a [`KinematicTransformFrame`] onto the kinematic ring. /// /// The kinematic ring runs in lockstep with the blendshape ring: one /// frame per 16.67 ms, same timestamp, consumed together by the renderer /// (WO-5) to drive both the face ARKit weights and the skeletal joint /// rotations. pub fn push_kinematic( &self, frame: KinematicTransformFrame, ) -> Result<(), BackpressureError> { self.ring_push( KINEMATIC_CTRL_OFF, KINEMATIC_DATA_OFF, KINEMATIC_SLOTS, RingId::Kinematic, frame, ) } /// Pops a [`KinematicTransformFrame`] from the kinematic ring. pub fn pop_kinematic(&self) -> Option { self.ring_pop(KINEMATIC_CTRL_OFF, KINEMATIC_DATA_OFF, KINEMATIC_SLOTS) } /// Capacity in slots of each ring, as `(audio, blendshape, sh, kinematic)`. pub const fn capacities() -> (usize, usize, usize, usize) { (AUDIO_SLOTS, BLEND_SLOTS, SH_SLOTS, KINEMATIC_SLOTS) } } #[cfg(test)] mod tests { use super::*; use miranda_core::{BLENDSHAPE_COUNT, SH_COEFF_COUNT}; fn frame(i: u64) -> BlendshapeFrame { BlendshapeFrame { timestamp_us: i, weights: [i as f32; BLENDSHAPE_COUNT], } } /// Single-slot round trip: the most basic correctness property. #[test] fn single_round_trip_preserves_bytes() { let bus = MirandaBus::in_memory(); assert!(bus.pop_blendshape().is_none(), "fresh ring must be empty"); bus.push_blendshape(frame(42)).unwrap(); let got = bus.pop_blendshape().expect("one frame was pushed"); assert_eq!(got.timestamp_us, 42); assert_eq!(got.weights, [42.0f32; BLENDSHAPE_COUNT]); assert!(bus.pop_blendshape().is_none(), "ring must be empty again"); } /// Fills the ring exactly to capacity, confirms the next push is rejected /// with backpressure rather than overwriting, then confirms every queued /// payload still reads back intact — i.e. a full-ring push did not /// corrupt existing contents (WO-1 REQ-6). #[test] fn full_ring_returns_backpressure_without_corruption() { let bus = MirandaBus::in_memory(); for i in 0..BLEND_SLOTS as u64 { bus.push_blendshape(frame(i)) .unwrap_or_else(|e| panic!("push {i} within capacity must succeed: {e}")); } let err = bus .push_blendshape(frame(9999)) .expect_err("push beyond capacity must be rejected"); assert_eq!(err.ring, RingId::Blendshape); assert_eq!(err.capacity, BLEND_SLOTS); for i in 0..BLEND_SLOTS as u64 { let got = bus.pop_blendshape().expect("queued frame must be present"); assert_eq!(got.timestamp_us, i, "FIFO order and contents preserved"); assert_eq!(got.weights, [i as f32; BLENDSHAPE_COUNT]); } assert!(bus.pop_blendshape().is_none()); } /// Drives the monotonic counters well past `slots` so the bitmask /// index wraps many times. Catches off-by-one errors in the /// `counter & (slots - 1)` mapping that a short test would miss. #[test] fn index_wraparound_stays_fifo() { let bus = MirandaBus::in_memory(); for i in 0..(BLEND_SLOTS as u64 * 5) { bus.push_blendshape(frame(i)).unwrap(); let got = bus.pop_blendshape().expect("pushed then popped"); assert_eq!(got.timestamp_us, i, "wraparound must preserve identity"); } } /// The three rings must be fully independent — a full audio ring must not /// affect blendshape or lighting traffic, and payloads must never appear /// on the wrong ring. #[test] fn rings_are_independent() { let bus = MirandaBus::in_memory(); for i in 0..AUDIO_SLOTS as u64 { bus.push_audio(AudioChunk { timestamp_us: i, sample_rate: miranda_core::AUDIO_SAMPLE_RATE_HZ, frame_count: miranda_core::AUDIO_CHUNK_FRAMES as u32, samples: [i as f32; miranda_core::AUDIO_CHUNK_FRAMES], }) .unwrap(); } assert!( bus.push_audio(AudioChunk { timestamp_us: 0, sample_rate: 0, frame_count: 0, samples: [0.0; miranda_core::AUDIO_CHUNK_FRAMES], }) .is_err(), "audio ring should now be full" ); // Other rings unaffected. bus.push_blendshape(frame(7)).unwrap(); bus.push_sh(SphericalHarmonics { timestamp_us: 7, coefficients: [0.5; SH_COEFF_COUNT], _padding: [0; 4], }) .unwrap(); assert_eq!(bus.pop_blendshape().unwrap().timestamp_us, 7); let sh = bus.pop_sh().unwrap(); assert_eq!(sh.timestamp_us, 7); assert_eq!(sh.coefficients, [0.5; SH_COEFF_COUNT]); assert_eq!(bus.pop_audio().unwrap().timestamp_us, 0, "audio intact"); } /// Two real threads, genuinely concurrent, on the heap backing so MIRI /// can check the atomic ordering. MIRI's weak-memory emulation will /// report a data race here if the Acquire/Release pairing is wrong — /// which is the only way to catch that class of bug, since x86's strong /// memory model hides it at runtime. #[test] fn concurrent_spsc_under_miri() { // Kept small: MIRI is orders of magnitude slower than native. const N: u64 = 200; let bus = std::sync::Arc::new(MirandaBus::in_memory()); let producer_bus = std::sync::Arc::clone(&bus); let producer = std::thread::spawn(move || { let mut sent = 0u64; while sent < N { if producer_bus.push_blendshape(frame(sent)).is_ok() { sent += 1; } else { std::thread::yield_now(); } } }); let consumer_bus = std::sync::Arc::clone(&bus); let consumer = std::thread::spawn(move || { let mut received = 0u64; while received < N { match consumer_bus.pop_blendshape() { Some(got) => { assert_eq!( got.timestamp_us, received, "SPSC must deliver in FIFO order with no gaps" ); assert_eq!( got.weights, [received as f32; BLENDSHAPE_COUNT], "payload must be byte-identical, not torn" ); received += 1; } None => std::thread::yield_now(), } } received }); producer.join().expect("producer thread panicked"); let received = consumer.join().expect("consumer thread panicked"); assert_eq!(received, N); } /// WO-1 REQ-5 / acceptance criteria: a real concurrent test, two actual /// OS threads, one pushing 1,000 `BlendshapeFrame`s with incrementing /// timestamps, one popping and asserting byte-identical contents via /// `assert_eq!`. Runs at native speed (unlike `concurrent_spsc_under_miri`, /// which is deliberately small so MIRI's interpreter can finish it) and /// prints per-thread evidence so the interleaving is visible in /// `--nocapture` output, not just asserted silently. #[test] fn test_blendshape_round_trip_concurrent() { const N: u64 = 1_000; let bus = std::sync::Arc::new(MirandaBus::in_memory()); let producer_bus = std::sync::Arc::clone(&bus); let producer = std::thread::Builder::new() .name("blendshape-writer".into()) .spawn(move || { let start = std::time::Instant::now(); let mut sent = 0u64; while sent < N { // weights filled with the frame index cast to f32, per spec. if producer_bus.push_blendshape(frame(sent)).is_ok() { if sent % 250 == 0 { println!( "[{:?}] writer pushed frame {sent} at {:?} since start", std::thread::current().name().unwrap_or("?"), start.elapsed() ); } sent += 1; } else { std::thread::yield_now(); } } println!( "[{:?}] writer done: {N} frames in {:?}", std::thread::current().name().unwrap_or("?"), start.elapsed() ); }) .expect("spawn writer thread"); let consumer_bus = std::sync::Arc::clone(&bus); let consumer = std::thread::Builder::new() .name("blendshape-reader".into()) .spawn(move || { let start = std::time::Instant::now(); let mut received = 0u64; while received < N { match consumer_bus.pop_blendshape() { Some(got) => { assert_eq!( got.timestamp_us, received, "byte-identical round trip: timestamp must match what was pushed" ); assert_eq!( got.weights, [received as f32; BLENDSHAPE_COUNT], "byte-identical round trip: weights must match what was pushed" ); if received % 250 == 0 { println!( "[{:?}] reader popped frame {received} at {:?} since start", std::thread::current().name().unwrap_or("?"), start.elapsed() ); } received += 1; } None => std::thread::yield_now(), } } println!( "[{:?}] reader done: {N} frames in {:?}", std::thread::current().name().unwrap_or("?"), start.elapsed() ); received }) .expect("spawn reader thread"); producer.join().expect("writer thread panicked"); let received = consumer.join().expect("reader thread panicked"); assert_eq!(received, N, "reader must observe every frame the writer sent"); } /// WO-1 REQ-1 / performance target: measures real mean round-trip /// latency over 10,000 push+pop cycles and asserts it stays within the /// ≤50 μs budget. If this fails, the most likely cause is false sharing /// between the head/tail atomics and slot data — verify the /// `#[repr(align(64))]`-equivalent cache-line separation in `CTRL_BYTES` /// is actually being honoured (see `layout_is_internally_consistent`). /// /// Deliberately native-only: this measures wall-clock performance, which /// MIRI's interpreter cannot represent (it reports emulated, not real, /// timing) — this test is meaningless under `cargo miri test` and is not /// run there. #[test] #[cfg(not(miri))] fn test_round_trip_latency() { let bus = MirandaBus::in_memory(); let payload = frame(0); // Warm up: first-touch page faults / cache effects should not count // against the measured budget. for _ in 0..1_000 { bus.push_blendshape(payload).unwrap(); let _ = bus.pop_blendshape(); } const ITERS: u32 = 10_000; let start = std::time::Instant::now(); for _ in 0..ITERS { bus.push_blendshape(payload).unwrap(); let _ = bus.pop_blendshape(); } let elapsed = start.elapsed(); let elapsed_us = elapsed.as_micros() as u64 / ITERS as u64; println!( "Mean round-trip latency over {ITERS} iterations: {elapsed_us} μs \ (total {elapsed:?}, target ≤50 μs)" ); assert!( elapsed_us <= 50, "Round-trip latency {elapsed_us} μs exceeds ≤50 μs target — \ check for false sharing on head/tail atomics" ); } /// Verifies the real mmap-backed path at `/dev/shm`. Skipped under MIRI, /// which cannot execute the `mmap` syscall — the logic itself is covered /// by the heap-backed tests above. #[test] #[cfg_attr(miri, ignore = "MIRI cannot execute the mmap syscall")] fn mmap_backed_round_trip() { let path = format!("/dev/shm/miranda_bus_test_mmap_{}", std::process::id()); let _ = std::fs::remove_file(&path); { let bus = MirandaBus::open_or_create_at(&path).expect("open bus"); bus.push_blendshape(frame(123)).unwrap(); let got = bus.pop_blendshape().expect("frame present"); assert_eq!(got.timestamp_us, 123); assert_eq!(got.weights, [123.0f32; BLENDSHAPE_COUNT]); } // A second opener sees the same region (interprocess semantics, // exercised here across two mappings in one process). { let bus = MirandaBus::open_or_create_at(&path).expect("reopen bus"); bus.push_audio(AudioChunk { timestamp_us: 55, sample_rate: miranda_core::AUDIO_SAMPLE_RATE_HZ, frame_count: 160, samples: [1.5; miranda_core::AUDIO_CHUNK_FRAMES], }) .unwrap(); assert_eq!(bus.pop_audio().unwrap().timestamp_us, 55); } std::fs::remove_file(&path).expect("cleanup"); } /// The mapping must be large enough for every region — a short file would /// mean slot writes past the end fault with SIGBUS at runtime. #[test] fn layout_is_internally_consistent() { assert_eq!(BUS_TOTAL_BYTES % CACHE_LINE, 0); assert!(SH_DATA_OFF + SH_DATA_BYTES <= BUS_TOTAL_BYTES); // head and tail really are on different cache lines. let bus = MirandaBus::in_memory(); let h = bus.head(BLEND_CTRL_OFF) as *const AtomicUsize as usize; let t = bus.tail(BLEND_CTRL_OFF) as *const AtomicUsize as usize; assert_eq!(t - h, CACHE_LINE, "false sharing guard"); assert_eq!(h % std::mem::align_of::(), 0); assert_eq!(t % std::mem::align_of::(), 0); } }