subetha_cxc/
shared_vec.rs

1//! `SharedVec<T>` - cross-process bounded indexable sequence.
2//!
3//! Distinct from [`SharedRing`](crate::SharedRing) (FIFO; drain
4//! semantics): SharedVec is RANDOM-ACCESS, accumulates monotonically
5//! up to capacity, and supports `get(i)` for any prior index.
6//!
7//! # Layout
8//!
9//! Single MMF file:
10//!
11//! ```text
12//! +---------------------------+
13//! | VecHeader  (64B aligned)  |  magic, capacity, len, slot_size
14//! +---------------------------+
15//! | Slot[0]    (64B = cache)  |  version + payload[VEC_PAYLOAD_BYTES]
16//! | Slot[1]                   |
17//! | ...                       |
18//! | Slot[capacity - 1]        |
19//! +---------------------------+
20//! ```
21//!
22//! Each slot is its own SeqLock cell (same shape as SharedCell);
23//! per-slot writes never false-share because each is its own cache
24//! line.
25//!
26//! # Concurrency
27//!
28//! - `push_back`: atomic `len.fetch_add(1)` claims a slot index; if
29//!   it exceeds capacity, rollback with `fetch_sub(1)` and return
30//!   `Full`. On success, write the payload under the slot's SeqLock
31//!   (version bump odd → write → bump even).
32//! - `get(i)`: load `len` (Acquire). If `i >= len`, return None.
33//!   Otherwise SeqLock-read `slot[i]`: spin if version is odd
34//!   (writer in progress), reread on version change.
35//! - `pop_back`: `compare_exchange` on `len` to decrement; if
36//!   successful, read the now-popped slot's payload at the old
37//!   index. The slot bytes remain in place but are no longer
38//!   addressable via `len`-bounded access.
39//! - `set(i, v)`: bounds-check against `len`, then SeqLock-write.
40//! - `clear`: store `len = 0` (Release). Previously-pushed slot
41//!   payloads remain on disk but become unreachable through the
42//!   bounded indexing.
43//!
44//! # Capacity
45//!
46//! Fixed at create time. The MMF is pre-allocated to the full size;
47//! no resize-on-grow protocol. The unbounded variant (with
48//! coordinator-mediated MMF resize) is a separate primitive.
49
50use std::fs::{File, OpenOptions};
51use std::marker::PhantomData;
52use std::mem::size_of;
53use std::path::Path;
54use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
55
56use memmap2::{Mmap, MmapMut, MmapOptions};
57
58pub const VEC_MAGIC: u32 = 0x4150_5656;
59pub const VEC_PAYLOAD_BYTES: usize = 52;
60
61/// How this process mapped the file. `MmapMut` demands a read+write
62/// file handle, which a consumer holding read access alone cannot get.
63enum Mapping {
64    Writable(MmapMut),
65    ReadOnly(Mmap),
66}
67
68impl Mapping {
69    #[inline]
70    fn as_ptr(&self) -> *const u8 {
71        match self {
72            Mapping::Writable(m) => m.as_ptr(),
73            Mapping::ReadOnly(m) => m.as_ptr(),
74        }
75    }
76
77    #[inline]
78    fn is_writable(&self) -> bool {
79        matches!(self, Mapping::Writable(_))
80    }
81
82    /// A no-op on a read-only mapping, so a caller flushing on a timer
83    /// need not know which kind it holds.
84    fn flush(&self) -> Result<(), std::io::Error> {
85        match self {
86            Mapping::Writable(m) => m.flush(),
87            Mapping::ReadOnly(_) => Ok(()),
88        }
89    }
90
91    fn flush_async(&self) -> Result<(), std::io::Error> {
92        match self {
93            Mapping::Writable(m) => m.flush_async(),
94            Mapping::ReadOnly(_) => Ok(()),
95        }
96    }
97}
98
99#[repr(C, align(64))]
100pub struct VecHeader {
101    pub magic: u32,
102    pub slot_payload_size: u32,
103    pub capacity: u64,
104    pub len: AtomicU64,
105    _pad: [u8; 40],
106}
107
108#[repr(C, align(64))]
109pub struct VecSlot {
110    pub version: AtomicU32,
111    _pad: [u8; 4],
112    pub payload: [u8; VEC_PAYLOAD_BYTES],
113}
114
115const _: () = {
116    assert!(size_of::<VecHeader>() == 64);
117    assert!(size_of::<VecSlot>() == 64);
118};
119
120pub const fn vec_file_size(capacity: usize) -> usize {
121    size_of::<VecHeader>() + capacity * size_of::<VecSlot>()
122}
123
124#[derive(Debug, Clone, Copy, PartialEq, Eq)]
125pub enum VecError {
126    Full,
127    OutOfBounds,
128    LayoutMismatch,
129    PayloadTooLarge,
130    /// The vec was opened read-only and something tried to write it.
131    ReadOnly,
132    IoError(std::io::ErrorKind),
133}
134
135impl From<std::io::Error> for VecError {
136    fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
137}
138
139pub struct SharedVec<T: Copy + 'static> {
140    _file: File,
141    mmap: Mapping,
142    capacity: usize,
143    _phantom: PhantomData<T>,
144    header_sidecar: subetha_core::HandshakeHeader,
145    ring_sidecar: Box<subetha_core::ObservationRing>,
146}
147
148unsafe impl<T: Copy + Send + 'static> Send for SharedVec<T> {}
149unsafe impl<T: Copy + Sync + 'static> Sync for SharedVec<T> {}
150
151impl<T: Copy + Send + Sync + 'static> subetha_sidecar::AdaptiveInstance for SharedVec<T> {
152    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
153    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
154    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
155        Box::new(subetha_sidecar::NoMigrationPolicy)
156    }
157}
158
159impl<T: Copy + 'static> SharedVec<T> {
160    /// Obtain the vec at `path`, initializing an empty one if the path
161    /// does not yet exist and attaching to it if it does. Attaching
162    /// leaves live elements and `len` in place; a region built with a
163    /// different capacity is a `LayoutMismatch`.
164    /// [`reset`](Self::reset) reinitializes.
165    pub fn create(
166        path: impl AsRef<Path>, capacity: usize,
167    ) -> Result<Self, VecError> {
168        if size_of::<T>() > VEC_PAYLOAD_BYTES {
169            return Err(VecError::PayloadTooLarge);
170        }
171        assert!(capacity >= 1);
172        let (file, mmap) = crate::mmf_attach::create_or_attach(
173            path.as_ref(),
174            vec_file_size(capacity),
175            |ptr| unsafe { Self::init_region(ptr, capacity) },
176            |ptr| unsafe { (*(ptr as *const VecHeader)).magic == VEC_MAGIC },
177        )?;
178        let this = Self {
179            _file: file, mmap: Mapping::Writable(mmap), capacity, _phantom: PhantomData,
180            header_sidecar: subetha_core::HandshakeHeader::new(),
181            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
182        };
183        this.validate(capacity)?;
184        Ok(this)
185    }
186
187    /// Truncate the vec at `path` and initialize an empty one,
188    /// discarding every element live peers share. For a caller that
189    /// knows it owns the path.
190    pub fn reset(
191        path: impl AsRef<Path>, capacity: usize,
192    ) -> Result<Self, VecError> {
193        if size_of::<T>() > VEC_PAYLOAD_BYTES {
194            return Err(VecError::PayloadTooLarge);
195        }
196        assert!(capacity >= 1);
197        let (file, mmap) = crate::mmf_attach::reset(path.as_ref(), vec_file_size(capacity), |ptr| unsafe {
198            Self::init_region(ptr, capacity)
199        })?;
200        Ok(Self {
201            _file: file, mmap: Mapping::Writable(mmap), capacity, _phantom: PhantomData,
202            header_sidecar: subetha_core::HandshakeHeader::new(),
203            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
204        })
205    }
206
207    /// Lay out an empty vec: sizes first, magic last, because attachers
208    /// spin on it. The zeroed region is already the empty slot array
209    /// (version 0) and `len` 0.
210    ///
211    /// # Safety
212    /// `ptr` addresses at least `vec_file_size(capacity)` writable
213    /// zeroed bytes.
214    unsafe fn init_region(ptr: *mut u8, capacity: usize) {
215        let hdr = ptr as *mut VecHeader;
216        unsafe {
217            (*hdr).slot_payload_size = VEC_PAYLOAD_BYTES as u32;
218            (*hdr).capacity = capacity as u64;
219            std::ptr::write_volatile(&raw mut (*hdr).magic, VEC_MAGIC);
220        }
221    }
222
223    pub fn open(
224        path: impl AsRef<Path>, expected_capacity: usize,
225    ) -> Result<Self, VecError> {
226        if size_of::<T>() > VEC_PAYLOAD_BYTES {
227            return Err(VecError::PayloadTooLarge);
228        }
229        let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
230        let total = vec_file_size(expected_capacity);
231        if file.metadata()?.len() < total as u64 {
232            return Err(VecError::LayoutMismatch);
233        }
234        let mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
235        let this = Self {
236            _file: file, mmap: Mapping::Writable(mmap), capacity: expected_capacity,
237            _phantom: PhantomData,
238            header_sidecar: subetha_core::HandshakeHeader::new(),
239            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
240        };
241        this.validate(expected_capacity)?;
242        Ok(this)
243    }
244
245    /// Open a vec this process may only read.
246    ///
247    /// [`open`](Self::open) needs a read+write file handle, which a
248    /// consumer of a privileged producer's vec does not have - granting
249    /// it would let any reader corrupt the state for all of them.
250    /// Reads behave identically, under the same per-slot SeqLock;
251    /// writes return [`VecError::ReadOnly`].
252    pub fn open_read_only(
253        path: impl AsRef<Path>, expected_capacity: usize,
254    ) -> Result<Self, VecError> {
255        if size_of::<T>() > VEC_PAYLOAD_BYTES {
256            return Err(VecError::PayloadTooLarge);
257        }
258        let file = OpenOptions::new().read(true).open(path.as_ref())?;
259        let total = vec_file_size(expected_capacity);
260        if file.metadata()?.len() < total as u64 {
261            return Err(VecError::LayoutMismatch);
262        }
263        let mmap = unsafe { MmapOptions::new().len(total).map(&file)? };
264        let this = Self {
265            _file: file, mmap: Mapping::ReadOnly(mmap), capacity: expected_capacity,
266            _phantom: PhantomData,
267            header_sidecar: subetha_core::HandshakeHeader::new(),
268            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
269        };
270        this.validate(expected_capacity)?;
271        Ok(this)
272    }
273
274    /// Whether the header on disk is the one this mapping expects.
275    fn validate(&self, expected_capacity: usize) -> Result<(), VecError> {
276        let hdr = self.header();
277        if hdr.magic != VEC_MAGIC || hdr.capacity != expected_capacity as u64 {
278            return Err(VecError::LayoutMismatch);
279        }
280        Ok(())
281    }
282
283    /// Whether this mapping may be written.
284    #[inline]
285    pub fn is_writable(&self) -> bool {
286        self.mmap.is_writable()
287    }
288
289    #[inline]
290    pub fn capacity(&self) -> usize { self.capacity }
291
292    #[inline]
293    pub fn len(&self) -> usize {
294        self.header().len.load(Ordering::Acquire) as usize
295    }
296
297    #[inline]
298    pub fn is_empty(&self) -> bool { self.len() == 0 }
299
300    fn header(&self) -> &VecHeader {
301        unsafe { &*(self.mmap.as_ptr() as *const VecHeader) }
302    }
303
304    fn slot(&self, i: usize) -> &VecSlot {
305        assert!(i < self.capacity, "slot index {i} out of bounds for cap {}", self.capacity);
306        let base = unsafe { self.mmap.as_ptr().add(size_of::<VecHeader>()) };
307        unsafe { &*(base.add(i * size_of::<VecSlot>()) as *const VecSlot) }
308    }
309
310    /// SeqLock write of a payload into a slot. Caller is responsible
311    /// for ensuring `i < capacity`.
312    fn write_slot(&self, i: usize, v: T) {
313        let slot = self.slot(i);
314        // Bump version odd; subsequent readers will spin.
315        slot.version.fetch_add(1, Ordering::AcqRel);
316        // Memcpy the value.
317        let dst = unsafe {
318            let base = self.mmap.as_ptr().add(size_of::<VecHeader>())
319                .add(i * size_of::<VecSlot>())
320                .add(std::mem::offset_of!(VecSlot, payload));
321            base as *mut u8
322        };
323        unsafe {
324            std::ptr::copy_nonoverlapping(
325                &v as *const T as *const u8,
326                dst,
327                size_of::<T>(),
328            );
329        }
330        // Bump version even; readers resume.
331        slot.version.fetch_add(1, Ordering::AcqRel);
332    }
333
334    /// SeqLock read of a slot. Spins if version is odd; rereads on
335    /// version change.
336    fn read_slot(&self, i: usize) -> T {
337        let slot = self.slot(i);
338        loop {
339            let v1 = slot.version.load(Ordering::Acquire);
340            if v1 & 1 != 0 {
341                std::hint::spin_loop();
342                continue;
343            }
344            let mut out = std::mem::MaybeUninit::<T>::uninit();
345            let src = unsafe {
346                self.mmap.as_ptr().add(size_of::<VecHeader>())
347                    .add(i * size_of::<VecSlot>())
348                    .add(std::mem::offset_of!(VecSlot, payload))
349            };
350            unsafe {
351                std::ptr::copy_nonoverlapping(
352                    src, out.as_mut_ptr() as *mut u8, size_of::<T>(),
353                );
354            }
355            let v2 = slot.version.load(Ordering::Acquire);
356            if v1 == v2 {
357                return unsafe { out.assume_init() };
358            }
359        }
360    }
361
362    /// Append a value. Returns the index it landed at.
363    /// Returns `Err(Full)` when the vec is at capacity.
364    pub fn push_back(&self, v: T) -> Result<usize, VecError> {
365        if !self.mmap.is_writable() {
366            return Err(VecError::ReadOnly);
367        }
368        let idx = self.header().len.fetch_add(1, Ordering::AcqRel) as usize;
369        if idx >= self.capacity {
370            self.header().len.fetch_sub(1, Ordering::AcqRel);
371            self.ring_sidecar
372                .push_op(crate::sidecar_ops::ordered::OP_INSERT, 1); // full
373            return Err(VecError::Full);
374        }
375        self.write_slot(idx, v);
376        self.ring_sidecar
377            .push_op(crate::sidecar_ops::ordered::OP_INSERT, 0);
378        Ok(idx)
379    }
380
381    /// Remove and return the last element. Returns None when empty.
382    pub fn pop_back(&self) -> Option<T> {
383        if !self.mmap.is_writable() {
384            return None;
385        }
386        loop {
387            let cur = self.header().len.load(Ordering::Acquire);
388            if cur == 0 {
389                self.ring_sidecar
390                    .push_op(crate::sidecar_ops::ordered::OP_POP, 2); // empty
391                return None;
392            }
393            let new = cur - 1;
394            if self.header().len.compare_exchange(
395                cur, new, Ordering::AcqRel, Ordering::Acquire,
396            ).is_ok() {
397                let v = self.read_slot(new as usize);
398                self.ring_sidecar
399                    .push_op(crate::sidecar_ops::ordered::OP_POP, 0);
400                return Some(v);
401            }
402        }
403    }
404
405    /// Read the value at index `i`. Returns None when `i >= len`.
406    pub fn get(&self, i: usize) -> Option<T> {
407        if i >= self.len() {
408            self.ring_sidecar
409                .push_op(crate::sidecar_ops::ordered::OP_GET, 2); // out of bounds / absent
410            return None;
411        }
412        let v = self.read_slot(i);
413        self.ring_sidecar
414            .push_op(crate::sidecar_ops::ordered::OP_GET, 0);
415        Some(v)
416    }
417
418    /// Overwrite the value at index `i`. Returns `Err(OutOfBounds)`
419    /// when `i >= len`.
420    pub fn set(&self, i: usize, v: T) -> Result<(), VecError> {
421        if !self.mmap.is_writable() {
422            return Err(VecError::ReadOnly);
423        }
424        if i >= self.len() {
425            self.ring_sidecar
426                .push_op(crate::sidecar_ops::ordered::OP_INSERT, 1); // out of bounds (positional write rejected)
427            return Err(VecError::OutOfBounds);
428        }
429        self.write_slot(i, v);
430        self.ring_sidecar
431            .push_op(crate::sidecar_ops::ordered::OP_INSERT, 0);
432        Ok(())
433    }
434
435    /// Clear the vec by resetting len to 0. Slot payloads are not
436    /// zeroed; they become unreachable through bounded indexing.
437    pub fn clear(&self) {
438        if !self.mmap.is_writable() {
439            return;
440        }
441        self.header().len.store(0, Ordering::Release);
442        self.ring_sidecar
443            .push_op(crate::sidecar_ops::ordered::OP_REMOVE, 0);
444    }
445
446    /// Snapshot all current values into a Vec. Best-effort: under
447    /// concurrent writers, the snapshot is a consistent prefix at
448    /// the moment of the `len` load, with each slot read under its
449    /// own SeqLock.
450    pub fn snapshot(&self) -> Vec<T> {
451        let n = self.len();
452        let mut out = Vec::with_capacity(n);
453        for i in 0..n {
454            out.push(self.read_slot(i));
455        }
456        self.ring_sidecar
457            .push_op(crate::sidecar_ops::ordered::OP_ITER, 0);
458        out
459    }
460
461    /// Read every live slot in order and hand each to `f`, without
462    /// building a `Vec`.
463    ///
464    /// For a consumer that looks at every element rather than keeping
465    /// them. [`snapshot`](Self::snapshot) costs an allocation the size
466    /// of the data plus a pass to fill it; this costs neither.
467    ///
468    /// Each slot is read through its own SeqLock into a local, as `get`
469    /// does, so `f` never sees a torn value. A reference into the
470    /// mapping would let a writer change the bytes under the reader.
471    ///
472    /// `f` is called in index order, over `len` as of entry.
473    pub fn for_each<F: FnMut(usize, &T)>(&self, f: F) {
474        self.for_each_range(0, self.len(), f);
475    }
476
477    /// [`for_each`](Self::for_each) over `start..end`, clamped to the
478    /// live length. The range form lets workers take disjoint spans
479    /// without materializing their share first.
480    pub fn for_each_range<F: FnMut(usize, &T)>(&self, start: usize, end: usize, mut f: F) {
481        let end = end.min(self.len());
482        for i in start..end {
483            let v = self.read_slot(i);
484            f(i, &v);
485        }
486        self.ring_sidecar
487            .push_op(crate::sidecar_ops::ordered::OP_ITER, 0);
488    }
489
490    pub fn flush(&self) -> Result<(), VecError> {
491        self.mmap.flush()?;
492        Ok(())
493    }
494
495    /// Non-blocking flush: schedules a writeback via the OS.
496    /// Note: Windows is only partially async (sync to page cache,
497    /// not to disk).
498    pub fn flush_async(&self) -> Result<(), VecError> {
499        self.mmap.flush_async()?;
500        Ok(())
501    }
502}
503
504#[cfg(test)]
505mod tests {
506    use super::*;
507    use std::sync::Arc;
508    use std::thread;
509
510    /// A second create attaches with live elements in place; reset is
511    /// what strips them.
512    #[test]
513    fn second_create_attaches_and_keeps_elements() {
514        let p = tmp("attach");
515        std::fs::remove_file(&p).ok();
516        let v = SharedVec::<u64>::create(&p, 16).unwrap();
517        v.push_back(777).unwrap();
518
519        let v2 = SharedVec::<u64>::create(&p, 16).unwrap();
520        assert_eq!(v2.len(), 1, "attach emptied a live vec");
521        assert_eq!(v2.get(0), Some(777));
522        assert!(matches!(
523            SharedVec::<u64>::create(&p, 8),
524            Err(VecError::LayoutMismatch),
525        ));
526
527        // Windows refuses to truncate a mapped file, so every handle goes
528        // before the reset.
529        drop(v);
530        drop(v2);
531        let fresh = SharedVec::<u64>::reset(&p, 16).unwrap();
532        assert_eq!(fresh.len(), 0, "reset kept an element");
533        drop(fresh);
534        std::fs::remove_file(&p).ok();
535    }
536
537    #[test]
538    fn for_each_visits_every_live_slot_in_order() {
539        let p = tmp("foreach");
540        let v = SharedVec::<u64>::create(&p, 256).unwrap();
541        for i in 0..100u64 {
542            v.push_back(i * 3).unwrap();
543        }
544        let mut seen = Vec::new();
545        v.for_each(|i, x| seen.push((i, *x)));
546        assert_eq!(seen.len(), 100, "capacity beyond len is not visited");
547        assert!(seen.iter().enumerate().all(|(n, (i, x))| *i == n && *x == n as u64 * 3));
548        std::fs::remove_file(&p).ok();
549    }
550
551    #[test]
552    fn for_each_range_is_clamped_and_disjoint() {
553        let p = tmp("foreach-range");
554        let v = SharedVec::<u64>::create(&p, 64).unwrap();
555        for i in 0..40u64 {
556            v.push_back(i).unwrap();
557        }
558        let mut first = Vec::new();
559        let mut second = Vec::new();
560        v.for_each_range(0, 20, |_, x| first.push(*x));
561        v.for_each_range(20, 40, |_, x| second.push(*x));
562        // Together the spans are the whole vec, once each.
563        first.extend_from_slice(&second);
564        assert_eq!(first, (0..40u64).collect::<Vec<_>>());
565        // An end past len is clamped, not out of bounds.
566        let mut over = Vec::new();
567        v.for_each_range(30, 9999, |_, x| over.push(*x));
568        assert_eq!(over.len(), 10);
569        std::fs::remove_file(&p).ok();
570    }
571
572    #[test]
573    fn for_each_agrees_with_snapshot() {
574        let p = tmp("foreach-vs-snapshot");
575        let v = SharedVec::<u64>::create(&p, 512).unwrap();
576        for i in 0..300u64 {
577            v.push_back(i ^ 0xa5a5).unwrap();
578        }
579        let mut walked = Vec::new();
580        v.for_each(|_, x| walked.push(*x));
581        assert_eq!(walked, v.snapshot());
582        std::fs::remove_file(&p).ok();
583    }
584
585    #[test]
586    fn a_read_only_vec_reads_the_same_and_refuses_writes() {
587        let p = tmp("readonly");
588        {
589            let w = SharedVec::<u64>::create(&p, 64).unwrap();
590            for i in 0..10u64 {
591                w.push_back(i * 7).unwrap();
592            }
593            w.flush().unwrap();
594        }
595        let r = SharedVec::<u64>::open_read_only(&p, 64).unwrap();
596        assert!(!r.is_writable());
597        assert_eq!(r.len(), 10);
598        assert_eq!(r.get(3), Some(21));
599        assert_eq!(r.snapshot(), (0..10u64).map(|i| i * 7).collect::<Vec<_>>());
600        assert_eq!(r.push_back(1), Err(VecError::ReadOnly));
601        assert_eq!(r.set(0, 1), Err(VecError::ReadOnly));
602        assert_eq!(r.pop_back(), None);
603        r.clear();
604        assert_eq!(r.len(), 10, "clear on a read-only vec changes nothing");
605        r.flush().unwrap();
606        std::fs::remove_file(&p).ok();
607    }
608
609    #[test]
610    fn a_read_only_open_still_validates_the_header() {
611        let p = tmp("readonly-mismatch");
612        {
613            let w = SharedVec::<u64>::create(&p, 64).unwrap();
614            w.push_back(1).unwrap();
615            w.flush().unwrap();
616        }
617        assert_eq!(
618            SharedVec::<u64>::open_read_only(&p, 32).err(),
619            Some(VecError::LayoutMismatch)
620        );
621        std::fs::remove_file(&p).ok();
622    }
623
624    fn tmp(name: &str) -> std::path::PathBuf {
625        let mut p = std::env::temp_dir();
626        let pid = std::process::id();
627        p.push(format!("subetha-vec-{name}-{pid}.bin"));
628        p
629    }
630
631    #[test]
632    fn create_initial_state_is_empty() {
633        let p = tmp("init");
634        let v: SharedVec<u32> = SharedVec::create(&p, 16).unwrap();
635        assert_eq!(v.capacity(), 16);
636        assert_eq!(v.len(), 0);
637        assert!(v.is_empty());
638        assert_eq!(v.get(0), None);
639        std::fs::remove_file(&p).ok();
640    }
641
642    #[test]
643    fn push_back_advances_len_and_get_round_trip() {
644        let p = tmp("push");
645        let v: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
646        for i in 0..5u32 {
647            let idx = v.push_back(i * 10).unwrap();
648            assert_eq!(idx, i as usize);
649        }
650        assert_eq!(v.len(), 5);
651        for i in 0..5 {
652            assert_eq!(v.get(i), Some((i as u32) * 10));
653        }
654        assert_eq!(v.get(5), None);
655        std::fs::remove_file(&p).ok();
656    }
657
658    #[test]
659    fn full_capacity_returns_error() {
660        let p = tmp("full");
661        let v: SharedVec<u32> = SharedVec::create(&p, 4).unwrap();
662        for i in 0..4u32 { v.push_back(i).unwrap(); }
663        assert_eq!(v.push_back(99).err(), Some(VecError::Full));
664        assert_eq!(v.len(), 4);  // rolled back
665        std::fs::remove_file(&p).ok();
666    }
667
668    #[test]
669    fn pop_back_returns_last_then_none() {
670        let p = tmp("pop");
671        let v: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
672        v.push_back(10).unwrap();
673        v.push_back(20).unwrap();
674        v.push_back(30).unwrap();
675        assert_eq!(v.pop_back(), Some(30));
676        assert_eq!(v.pop_back(), Some(20));
677        assert_eq!(v.len(), 1);
678        assert_eq!(v.pop_back(), Some(10));
679        assert_eq!(v.pop_back(), None);
680        std::fs::remove_file(&p).ok();
681    }
682
683    #[test]
684    fn set_replaces_value_at_index() {
685        let p = tmp("set");
686        let v: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
687        v.push_back(1).unwrap();
688        v.push_back(2).unwrap();
689        v.set(0, 100).unwrap();
690        assert_eq!(v.get(0), Some(100));
691        assert_eq!(v.get(1), Some(2));
692        assert_eq!(v.set(2, 200).err(), Some(VecError::OutOfBounds));
693        std::fs::remove_file(&p).ok();
694    }
695
696    #[test]
697    fn clear_resets_len_to_zero() {
698        let p = tmp("clear");
699        let v: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
700        for i in 0..5u32 { v.push_back(i).unwrap(); }
701        assert_eq!(v.len(), 5);
702        v.clear();
703        assert_eq!(v.len(), 0);
704        assert_eq!(v.get(0), None);
705        // After clear, push works fresh.
706        v.push_back(42).unwrap();
707        assert_eq!(v.get(0), Some(42));
708        std::fs::remove_file(&p).ok();
709    }
710
711    #[test]
712    fn snapshot_returns_consistent_prefix() {
713        let p = tmp("snapshot");
714        let v: SharedVec<u32> = SharedVec::create(&p, 16).unwrap();
715        for i in 0..7u32 { v.push_back(i + 100).unwrap(); }
716        let snap = v.snapshot();
717        assert_eq!(snap, vec![100, 101, 102, 103, 104, 105, 106]);
718        std::fs::remove_file(&p).ok();
719    }
720
721    #[test]
722    fn cross_handle_visibility() {
723        let p = tmp("cross-handle");
724        let writer: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
725        let reader: SharedVec<u32> = SharedVec::open(&p, 8).unwrap();
726        writer.push_back(777).unwrap();
727        assert_eq!(reader.get(0), Some(777));
728        reader.push_back(888).unwrap();
729        assert_eq!(writer.get(1), Some(888));
730        assert_eq!(writer.len(), 2);
731        std::fs::remove_file(&p).ok();
732    }
733
734    #[test]
735    fn concurrent_pushers_all_land_at_distinct_indices() {
736        let p = tmp("concurrent");
737        let v: Arc<SharedVec<u32>> = Arc::new(SharedVec::create(&p, 1024).unwrap());
738        let n_threads = 4;
739        let per_thread = 50u32;
740        let mut handles = vec![];
741        for t in 0..n_threads {
742            let v = v.clone();
743            handles.push(thread::spawn(move || {
744                let mut indices = vec![];
745                for i in 0..per_thread {
746                    let val = (t as u32) * per_thread + i;
747                    let idx = v.push_back(val).unwrap();
748                    indices.push(idx);
749                }
750                indices
751            }));
752        }
753        let mut all_indices: Vec<usize> = handles.into_iter()
754            .flat_map(|h| h.join().unwrap())
755            .collect();
756        all_indices.sort();
757        for (expected, actual) in all_indices.iter().enumerate() {
758            assert_eq!(*actual, expected,
759                "indices must form a contiguous 0..N sequence");
760        }
761        assert_eq!(v.len(), (n_threads * per_thread as usize));
762        std::fs::remove_file(&p).ok();
763    }
764
765    #[test]
766    fn payload_too_large_at_create() {
767        #[allow(dead_code)] // size_of<Big> is the test signal, not the field
768        struct Big([u8; VEC_PAYLOAD_BYTES + 1]);
769        impl Copy for Big {}
770        impl Clone for Big { fn clone(&self) -> Self { *self } }
771        let p = tmp("too-large");
772        let r = SharedVec::<Big>::create(&p, 4);
773        assert_eq!(r.err(), Some(VecError::PayloadTooLarge));
774        std::fs::remove_file(&p).ok();
775    }
776
777    #[test]
778    fn struct_payload_round_trip() {
779        #[derive(Clone, Copy, Debug, PartialEq)]
780        #[repr(C)]
781        struct Point { x: f64, y: f64, z: f64 }
782        let p = tmp("struct");
783        let v: SharedVec<Point> = SharedVec::create(&p, 8).unwrap();
784        v.push_back(Point { x: 1.0, y: 2.0, z: 3.0 }).unwrap();
785        v.push_back(Point { x: -1.5, y: 0.0, z: 7.25 }).unwrap();
786        assert_eq!(v.get(0), Some(Point { x: 1.0, y: 2.0, z: 3.0 }));
787        assert_eq!(v.get(1), Some(Point { x: -1.5, y: 0.0, z: 7.25 }));
788        std::fs::remove_file(&p).ok();
789    }
790
791    #[test]
792    fn disk_persistence_data_survives_reopen() {
793        let p = tmp("disk");
794        {
795            let v: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
796            for i in 0..4u32 { v.push_back(i * 100).unwrap(); }
797            v.flush().unwrap();
798        }
799        let v2: SharedVec<u32> = SharedVec::open(&p, 8).unwrap();
800        assert_eq!(v2.len(), 4);
801        for i in 0..4 {
802            assert_eq!(v2.get(i), Some((i as u32) * 100));
803        }
804        std::fs::remove_file(&p).ok();
805    }
806
807    #[test]
808    fn concurrent_reader_during_writes_sees_consistent_data() {
809        let p = tmp("read-during-write");
810        let v: Arc<SharedVec<u32>> = Arc::new(SharedVec::create(&p, 256).unwrap());
811        let v_w = v.clone();
812        let writer = thread::spawn(move || {
813            for i in 0..100u32 {
814                v_w.push_back(i).unwrap();
815            }
816        });
817        let v_r = v.clone();
818        let reader = thread::spawn(move || {
819            let mut last_len = 0;
820            loop {
821                let n = v_r.len();
822                if n == 100 { break; }
823                // Read every visible slot; values must equal index.
824                for i in last_len..n {
825                    let got = v_r.get(i);
826                    assert_eq!(got, Some(i as u32),
827                        "slot {i} should hold {i}, got {got:?}");
828                }
829                last_len = n;
830                std::thread::yield_now();
831            }
832        });
833        writer.join().unwrap();
834        reader.join().unwrap();
835        std::fs::remove_file(&p).ok();
836    }
837}
subetha_cxc/shared_vec.rs

subetha_cxc/
shared_vec.rs
