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    pub fn create(
161        path: impl AsRef<Path>, capacity: usize,
162    ) -> Result<Self, VecError> {
163        if size_of::<T>() > VEC_PAYLOAD_BYTES {
164            return Err(VecError::PayloadTooLarge);
165        }
166        assert!(capacity >= 1);
167        let total = vec_file_size(capacity);
168        let file = OpenOptions::new()
169            .read(true).write(true).create(true).truncate(true)
170            .open(path.as_ref())?;
171        file.set_len(total as u64)?;
172        let mut mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
173        let hdr = mmap.as_mut_ptr() as *mut VecHeader;
174        unsafe {
175            std::ptr::write(hdr, VecHeader {
176                magic: VEC_MAGIC,
177                slot_payload_size: VEC_PAYLOAD_BYTES as u32,
178                capacity: capacity as u64,
179                len: AtomicU64::new(0),
180                _pad: [0; 40],
181            });
182        }
183        for i in 0..capacity {
184            let slot_ptr = unsafe {
185                mmap.as_mut_ptr()
186                    .add(size_of::<VecHeader>())
187                    .add(i * size_of::<VecSlot>())
188            } as *mut VecSlot;
189            unsafe {
190                std::ptr::write(slot_ptr, VecSlot {
191                    version: AtomicU32::new(0),
192                    _pad: [0; 4],
193                    payload: [0u8; VEC_PAYLOAD_BYTES],
194                });
195            }
196        }
197        Ok(Self {
198            _file: file, mmap: Mapping::Writable(mmap), capacity, _phantom: PhantomData,
199            header_sidecar: subetha_core::HandshakeHeader::new(),
200            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
201        })
202    }
203
204    pub fn open(
205        path: impl AsRef<Path>, expected_capacity: usize,
206    ) -> Result<Self, VecError> {
207        if size_of::<T>() > VEC_PAYLOAD_BYTES {
208            return Err(VecError::PayloadTooLarge);
209        }
210        let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
211        let total = vec_file_size(expected_capacity);
212        if file.metadata()?.len() < total as u64 {
213            return Err(VecError::LayoutMismatch);
214        }
215        let mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
216        let this = Self {
217            _file: file, mmap: Mapping::Writable(mmap), capacity: expected_capacity,
218            _phantom: PhantomData,
219            header_sidecar: subetha_core::HandshakeHeader::new(),
220            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
221        };
222        this.validate(expected_capacity)?;
223        Ok(this)
224    }
225
226    /// Open a vec this process may only read.
227    ///
228    /// [`open`](Self::open) needs a read+write file handle, which a
229    /// consumer of a privileged producer's vec does not have - granting
230    /// it would let any reader corrupt the state for all of them.
231    /// Reads behave identically, under the same per-slot SeqLock;
232    /// writes return [`VecError::ReadOnly`].
233    pub fn open_read_only(
234        path: impl AsRef<Path>, expected_capacity: usize,
235    ) -> Result<Self, VecError> {
236        if size_of::<T>() > VEC_PAYLOAD_BYTES {
237            return Err(VecError::PayloadTooLarge);
238        }
239        let file = OpenOptions::new().read(true).open(path.as_ref())?;
240        let total = vec_file_size(expected_capacity);
241        if file.metadata()?.len() < total as u64 {
242            return Err(VecError::LayoutMismatch);
243        }
244        let mmap = unsafe { MmapOptions::new().len(total).map(&file)? };
245        let this = Self {
246            _file: file, mmap: Mapping::ReadOnly(mmap), capacity: expected_capacity,
247            _phantom: PhantomData,
248            header_sidecar: subetha_core::HandshakeHeader::new(),
249            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
250        };
251        this.validate(expected_capacity)?;
252        Ok(this)
253    }
254
255    /// Whether the header on disk is the one this mapping expects.
256    fn validate(&self, expected_capacity: usize) -> Result<(), VecError> {
257        let hdr = self.header();
258        if hdr.magic != VEC_MAGIC || hdr.capacity != expected_capacity as u64 {
259            return Err(VecError::LayoutMismatch);
260        }
261        Ok(())
262    }
263
264    /// Whether this mapping may be written.
265    #[inline]
266    pub fn is_writable(&self) -> bool {
267        self.mmap.is_writable()
268    }
269
270    #[inline]
271    pub fn capacity(&self) -> usize { self.capacity }
272
273    #[inline]
274    pub fn len(&self) -> usize {
275        self.header().len.load(Ordering::Acquire) as usize
276    }
277
278    #[inline]
279    pub fn is_empty(&self) -> bool { self.len() == 0 }
280
281    fn header(&self) -> &VecHeader {
282        unsafe { &*(self.mmap.as_ptr() as *const VecHeader) }
283    }
284
285    fn slot(&self, i: usize) -> &VecSlot {
286        assert!(i < self.capacity, "slot index {i} out of bounds for cap {}", self.capacity);
287        let base = unsafe { self.mmap.as_ptr().add(size_of::<VecHeader>()) };
288        unsafe { &*(base.add(i * size_of::<VecSlot>()) as *const VecSlot) }
289    }
290
291    /// SeqLock write of a payload into a slot. Caller is responsible
292    /// for ensuring `i < capacity`.
293    fn write_slot(&self, i: usize, v: T) {
294        let slot = self.slot(i);
295        // Bump version odd; subsequent readers will spin.
296        slot.version.fetch_add(1, Ordering::AcqRel);
297        // Memcpy the value.
298        let dst = unsafe {
299            let base = self.mmap.as_ptr().add(size_of::<VecHeader>())
300                .add(i * size_of::<VecSlot>())
301                .add(std::mem::offset_of!(VecSlot, payload));
302            base as *mut u8
303        };
304        unsafe {
305            std::ptr::copy_nonoverlapping(
306                &v as *const T as *const u8,
307                dst,
308                size_of::<T>(),
309            );
310        }
311        // Bump version even; readers resume.
312        slot.version.fetch_add(1, Ordering::AcqRel);
313    }
314
315    /// SeqLock read of a slot. Spins if version is odd; rereads on
316    /// version change.
317    fn read_slot(&self, i: usize) -> T {
318        let slot = self.slot(i);
319        loop {
320            let v1 = slot.version.load(Ordering::Acquire);
321            if v1 & 1 != 0 {
322                std::hint::spin_loop();
323                continue;
324            }
325            let mut out = std::mem::MaybeUninit::<T>::uninit();
326            let src = unsafe {
327                self.mmap.as_ptr().add(size_of::<VecHeader>())
328                    .add(i * size_of::<VecSlot>())
329                    .add(std::mem::offset_of!(VecSlot, payload))
330            };
331            unsafe {
332                std::ptr::copy_nonoverlapping(
333                    src, out.as_mut_ptr() as *mut u8, size_of::<T>(),
334                );
335            }
336            let v2 = slot.version.load(Ordering::Acquire);
337            if v1 == v2 {
338                return unsafe { out.assume_init() };
339            }
340        }
341    }
342
343    /// Append a value. Returns the index it landed at.
344    /// Returns `Err(Full)` when the vec is at capacity.
345    pub fn push_back(&self, v: T) -> Result<usize, VecError> {
346        if !self.mmap.is_writable() {
347            return Err(VecError::ReadOnly);
348        }
349        let idx = self.header().len.fetch_add(1, Ordering::AcqRel) as usize;
350        if idx >= self.capacity {
351            self.header().len.fetch_sub(1, Ordering::AcqRel);
352            self.ring_sidecar
353                .push_op(crate::sidecar_ops::ordered::OP_INSERT, 1); // full
354            return Err(VecError::Full);
355        }
356        self.write_slot(idx, v);
357        self.ring_sidecar
358            .push_op(crate::sidecar_ops::ordered::OP_INSERT, 0);
359        Ok(idx)
360    }
361
362    /// Remove and return the last element. Returns None when empty.
363    pub fn pop_back(&self) -> Option<T> {
364        if !self.mmap.is_writable() {
365            return None;
366        }
367        loop {
368            let cur = self.header().len.load(Ordering::Acquire);
369            if cur == 0 {
370                self.ring_sidecar
371                    .push_op(crate::sidecar_ops::ordered::OP_POP, 2); // empty
372                return None;
373            }
374            let new = cur - 1;
375            if self.header().len.compare_exchange(
376                cur, new, Ordering::AcqRel, Ordering::Acquire,
377            ).is_ok() {
378                let v = self.read_slot(new as usize);
379                self.ring_sidecar
380                    .push_op(crate::sidecar_ops::ordered::OP_POP, 0);
381                return Some(v);
382            }
383        }
384    }
385
386    /// Read the value at index `i`. Returns None when `i >= len`.
387    pub fn get(&self, i: usize) -> Option<T> {
388        if i >= self.len() {
389            self.ring_sidecar
390                .push_op(crate::sidecar_ops::ordered::OP_GET, 2); // out of bounds / absent
391            return None;
392        }
393        let v = self.read_slot(i);
394        self.ring_sidecar
395            .push_op(crate::sidecar_ops::ordered::OP_GET, 0);
396        Some(v)
397    }
398
399    /// Overwrite the value at index `i`. Returns `Err(OutOfBounds)`
400    /// when `i >= len`.
401    pub fn set(&self, i: usize, v: T) -> Result<(), VecError> {
402        if !self.mmap.is_writable() {
403            return Err(VecError::ReadOnly);
404        }
405        if i >= self.len() {
406            self.ring_sidecar
407                .push_op(crate::sidecar_ops::ordered::OP_INSERT, 1); // out of bounds (positional write rejected)
408            return Err(VecError::OutOfBounds);
409        }
410        self.write_slot(i, v);
411        self.ring_sidecar
412            .push_op(crate::sidecar_ops::ordered::OP_INSERT, 0);
413        Ok(())
414    }
415
416    /// Clear the vec by resetting len to 0. Slot payloads are not
417    /// zeroed; they become unreachable through bounded indexing.
418    pub fn clear(&self) {
419        if !self.mmap.is_writable() {
420            return;
421        }
422        self.header().len.store(0, Ordering::Release);
423        self.ring_sidecar
424            .push_op(crate::sidecar_ops::ordered::OP_REMOVE, 0);
425    }
426
427    /// Snapshot all current values into a Vec. Best-effort: under
428    /// concurrent writers, the snapshot is a consistent prefix at
429    /// the moment of the `len` load, with each slot read under its
430    /// own SeqLock.
431    pub fn snapshot(&self) -> Vec<T> {
432        let n = self.len();
433        let mut out = Vec::with_capacity(n);
434        for i in 0..n {
435            out.push(self.read_slot(i));
436        }
437        self.ring_sidecar
438            .push_op(crate::sidecar_ops::ordered::OP_ITER, 0);
439        out
440    }
441
442    /// Read every live slot in order and hand each to `f`, without
443    /// building a `Vec`.
444    ///
445    /// For a consumer that looks at every element rather than keeping
446    /// them. [`snapshot`](Self::snapshot) costs an allocation the size
447    /// of the data plus a pass to fill it; this costs neither.
448    ///
449    /// Each slot is read through its own SeqLock into a local, as `get`
450    /// does, so `f` never sees a torn value. A reference into the
451    /// mapping would let a writer change the bytes under the reader.
452    ///
453    /// `f` is called in index order, over `len` as of entry.
454    pub fn for_each<F: FnMut(usize, &T)>(&self, f: F) {
455        self.for_each_range(0, self.len(), f);
456    }
457
458    /// [`for_each`](Self::for_each) over `start..end`, clamped to the
459    /// live length. The range form lets workers take disjoint spans
460    /// without materializing their share first.
461    pub fn for_each_range<F: FnMut(usize, &T)>(&self, start: usize, end: usize, mut f: F) {
462        let end = end.min(self.len());
463        for i in start..end {
464            let v = self.read_slot(i);
465            f(i, &v);
466        }
467        self.ring_sidecar
468            .push_op(crate::sidecar_ops::ordered::OP_ITER, 0);
469    }
470
471    pub fn flush(&self) -> Result<(), VecError> {
472        self.mmap.flush()?;
473        Ok(())
474    }
475
476    /// Non-blocking flush: schedules a writeback via the OS.
477    /// Note: Windows is only partially async (sync to page cache,
478    /// not to disk).
479    pub fn flush_async(&self) -> Result<(), VecError> {
480        self.mmap.flush_async()?;
481        Ok(())
482    }
483}
484
485#[cfg(test)]
486mod tests {
487    use super::*;
488    use std::sync::Arc;
489    use std::thread;
490
491    #[test]
492    fn for_each_visits_every_live_slot_in_order() {
493        let p = tmp("foreach");
494        let v = SharedVec::<u64>::create(&p, 256).unwrap();
495        for i in 0..100u64 {
496            v.push_back(i * 3).unwrap();
497        }
498        let mut seen = Vec::new();
499        v.for_each(|i, x| seen.push((i, *x)));
500        assert_eq!(seen.len(), 100, "capacity beyond len is not visited");
501        assert!(seen.iter().enumerate().all(|(n, (i, x))| *i == n && *x == n as u64 * 3));
502        std::fs::remove_file(&p).ok();
503    }
504
505    #[test]
506    fn for_each_range_is_clamped_and_disjoint() {
507        let p = tmp("foreach-range");
508        let v = SharedVec::<u64>::create(&p, 64).unwrap();
509        for i in 0..40u64 {
510            v.push_back(i).unwrap();
511        }
512        let mut first = Vec::new();
513        let mut second = Vec::new();
514        v.for_each_range(0, 20, |_, x| first.push(*x));
515        v.for_each_range(20, 40, |_, x| second.push(*x));
516        // Together the spans are the whole vec, once each.
517        first.extend_from_slice(&second);
518        assert_eq!(first, (0..40u64).collect::<Vec<_>>());
519        // An end past len is clamped, not out of bounds.
520        let mut over = Vec::new();
521        v.for_each_range(30, 9999, |_, x| over.push(*x));
522        assert_eq!(over.len(), 10);
523        std::fs::remove_file(&p).ok();
524    }
525
526    #[test]
527    fn for_each_agrees_with_snapshot() {
528        let p = tmp("foreach-vs-snapshot");
529        let v = SharedVec::<u64>::create(&p, 512).unwrap();
530        for i in 0..300u64 {
531            v.push_back(i ^ 0xa5a5).unwrap();
532        }
533        let mut walked = Vec::new();
534        v.for_each(|_, x| walked.push(*x));
535        assert_eq!(walked, v.snapshot());
536        std::fs::remove_file(&p).ok();
537    }
538
539    #[test]
540    fn a_read_only_vec_reads_the_same_and_refuses_writes() {
541        let p = tmp("readonly");
542        {
543            let w = SharedVec::<u64>::create(&p, 64).unwrap();
544            for i in 0..10u64 {
545                w.push_back(i * 7).unwrap();
546            }
547            w.flush().unwrap();
548        }
549        let r = SharedVec::<u64>::open_read_only(&p, 64).unwrap();
550        assert!(!r.is_writable());
551        assert_eq!(r.len(), 10);
552        assert_eq!(r.get(3), Some(21));
553        assert_eq!(r.snapshot(), (0..10u64).map(|i| i * 7).collect::<Vec<_>>());
554        assert_eq!(r.push_back(1), Err(VecError::ReadOnly));
555        assert_eq!(r.set(0, 1), Err(VecError::ReadOnly));
556        assert_eq!(r.pop_back(), None);
557        r.clear();
558        assert_eq!(r.len(), 10, "clear on a read-only vec changes nothing");
559        r.flush().unwrap();
560        std::fs::remove_file(&p).ok();
561    }
562
563    #[test]
564    fn a_read_only_open_still_validates_the_header() {
565        let p = tmp("readonly-mismatch");
566        {
567            let w = SharedVec::<u64>::create(&p, 64).unwrap();
568            w.push_back(1).unwrap();
569            w.flush().unwrap();
570        }
571        assert_eq!(
572            SharedVec::<u64>::open_read_only(&p, 32).err(),
573            Some(VecError::LayoutMismatch)
574        );
575        std::fs::remove_file(&p).ok();
576    }
577
578    fn tmp(name: &str) -> std::path::PathBuf {
579        let mut p = std::env::temp_dir();
580        let pid = std::process::id();
581        p.push(format!("subetha-vec-{name}-{pid}.bin"));
582        p
583    }
584
585    #[test]
586    fn create_initial_state_is_empty() {
587        let p = tmp("init");
588        let v: SharedVec<u32> = SharedVec::create(&p, 16).unwrap();
589        assert_eq!(v.capacity(), 16);
590        assert_eq!(v.len(), 0);
591        assert!(v.is_empty());
592        assert_eq!(v.get(0), None);
593        std::fs::remove_file(&p).ok();
594    }
595
596    #[test]
597    fn push_back_advances_len_and_get_round_trip() {
598        let p = tmp("push");
599        let v: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
600        for i in 0..5u32 {
601            let idx = v.push_back(i * 10).unwrap();
602            assert_eq!(idx, i as usize);
603        }
604        assert_eq!(v.len(), 5);
605        for i in 0..5 {
606            assert_eq!(v.get(i), Some((i as u32) * 10));
607        }
608        assert_eq!(v.get(5), None);
609        std::fs::remove_file(&p).ok();
610    }
611
612    #[test]
613    fn full_capacity_returns_error() {
614        let p = tmp("full");
615        let v: SharedVec<u32> = SharedVec::create(&p, 4).unwrap();
616        for i in 0..4u32 { v.push_back(i).unwrap(); }
617        assert_eq!(v.push_back(99).err(), Some(VecError::Full));
618        assert_eq!(v.len(), 4);  // rolled back
619        std::fs::remove_file(&p).ok();
620    }
621
622    #[test]
623    fn pop_back_returns_last_then_none() {
624        let p = tmp("pop");
625        let v: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
626        v.push_back(10).unwrap();
627        v.push_back(20).unwrap();
628        v.push_back(30).unwrap();
629        assert_eq!(v.pop_back(), Some(30));
630        assert_eq!(v.pop_back(), Some(20));
631        assert_eq!(v.len(), 1);
632        assert_eq!(v.pop_back(), Some(10));
633        assert_eq!(v.pop_back(), None);
634        std::fs::remove_file(&p).ok();
635    }
636
637    #[test]
638    fn set_replaces_value_at_index() {
639        let p = tmp("set");
640        let v: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
641        v.push_back(1).unwrap();
642        v.push_back(2).unwrap();
643        v.set(0, 100).unwrap();
644        assert_eq!(v.get(0), Some(100));
645        assert_eq!(v.get(1), Some(2));
646        assert_eq!(v.set(2, 200).err(), Some(VecError::OutOfBounds));
647        std::fs::remove_file(&p).ok();
648    }
649
650    #[test]
651    fn clear_resets_len_to_zero() {
652        let p = tmp("clear");
653        let v: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
654        for i in 0..5u32 { v.push_back(i).unwrap(); }
655        assert_eq!(v.len(), 5);
656        v.clear();
657        assert_eq!(v.len(), 0);
658        assert_eq!(v.get(0), None);
659        // After clear, push works fresh.
660        v.push_back(42).unwrap();
661        assert_eq!(v.get(0), Some(42));
662        std::fs::remove_file(&p).ok();
663    }
664
665    #[test]
666    fn snapshot_returns_consistent_prefix() {
667        let p = tmp("snapshot");
668        let v: SharedVec<u32> = SharedVec::create(&p, 16).unwrap();
669        for i in 0..7u32 { v.push_back(i + 100).unwrap(); }
670        let snap = v.snapshot();
671        assert_eq!(snap, vec![100, 101, 102, 103, 104, 105, 106]);
672        std::fs::remove_file(&p).ok();
673    }
674
675    #[test]
676    fn cross_handle_visibility() {
677        let p = tmp("cross-handle");
678        let writer: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
679        let reader: SharedVec<u32> = SharedVec::open(&p, 8).unwrap();
680        writer.push_back(777).unwrap();
681        assert_eq!(reader.get(0), Some(777));
682        reader.push_back(888).unwrap();
683        assert_eq!(writer.get(1), Some(888));
684        assert_eq!(writer.len(), 2);
685        std::fs::remove_file(&p).ok();
686    }
687
688    #[test]
689    fn concurrent_pushers_all_land_at_distinct_indices() {
690        let p = tmp("concurrent");
691        let v: Arc<SharedVec<u32>> = Arc::new(SharedVec::create(&p, 1024).unwrap());
692        let n_threads = 4;
693        let per_thread = 50u32;
694        let mut handles = vec![];
695        for t in 0..n_threads {
696            let v = v.clone();
697            handles.push(thread::spawn(move || {
698                let mut indices = vec![];
699                for i in 0..per_thread {
700                    let val = (t as u32) * per_thread + i;
701                    let idx = v.push_back(val).unwrap();
702                    indices.push(idx);
703                }
704                indices
705            }));
706        }
707        let mut all_indices: Vec<usize> = handles.into_iter()
708            .flat_map(|h| h.join().unwrap())
709            .collect();
710        all_indices.sort();
711        for (expected, actual) in all_indices.iter().enumerate() {
712            assert_eq!(*actual, expected,
713                "indices must form a contiguous 0..N sequence");
714        }
715        assert_eq!(v.len(), (n_threads * per_thread as usize));
716        std::fs::remove_file(&p).ok();
717    }
718
719    #[test]
720    fn payload_too_large_at_create() {
721        #[allow(dead_code)] // size_of<Big> is the test signal, not the field
722        struct Big([u8; VEC_PAYLOAD_BYTES + 1]);
723        impl Copy for Big {}
724        impl Clone for Big { fn clone(&self) -> Self { *self } }
725        let p = tmp("too-large");
726        let r = SharedVec::<Big>::create(&p, 4);
727        assert_eq!(r.err(), Some(VecError::PayloadTooLarge));
728        std::fs::remove_file(&p).ok();
729    }
730
731    #[test]
732    fn struct_payload_round_trip() {
733        #[derive(Clone, Copy, Debug, PartialEq)]
734        #[repr(C)]
735        struct Point { x: f64, y: f64, z: f64 }
736        let p = tmp("struct");
737        let v: SharedVec<Point> = SharedVec::create(&p, 8).unwrap();
738        v.push_back(Point { x: 1.0, y: 2.0, z: 3.0 }).unwrap();
739        v.push_back(Point { x: -1.5, y: 0.0, z: 7.25 }).unwrap();
740        assert_eq!(v.get(0), Some(Point { x: 1.0, y: 2.0, z: 3.0 }));
741        assert_eq!(v.get(1), Some(Point { x: -1.5, y: 0.0, z: 7.25 }));
742        std::fs::remove_file(&p).ok();
743    }
744
745    #[test]
746    fn disk_persistence_data_survives_reopen() {
747        let p = tmp("disk");
748        {
749            let v: SharedVec<u32> = SharedVec::create(&p, 8).unwrap();
750            for i in 0..4u32 { v.push_back(i * 100).unwrap(); }
751            v.flush().unwrap();
752        }
753        let v2: SharedVec<u32> = SharedVec::open(&p, 8).unwrap();
754        assert_eq!(v2.len(), 4);
755        for i in 0..4 {
756            assert_eq!(v2.get(i), Some((i as u32) * 100));
757        }
758        std::fs::remove_file(&p).ok();
759    }
760
761    #[test]
762    fn concurrent_reader_during_writes_sees_consistent_data() {
763        let p = tmp("read-during-write");
764        let v: Arc<SharedVec<u32>> = Arc::new(SharedVec::create(&p, 256).unwrap());
765        let v_w = v.clone();
766        let writer = thread::spawn(move || {
767            for i in 0..100u32 {
768                v_w.push_back(i).unwrap();
769            }
770        });
771        let v_r = v.clone();
772        let reader = thread::spawn(move || {
773            let mut last_len = 0;
774            loop {
775                let n = v_r.len();
776                if n == 100 { break; }
777                // Read every visible slot; values must equal index.
778                for i in last_len..n {
779                    let got = v_r.get(i);
780                    assert_eq!(got, Some(i as u32),
781                        "slot {i} should hold {i}, got {got:?}");
782                }
783                last_len = n;
784                std::thread::yield_now();
785            }
786        });
787        writer.join().unwrap();
788        reader.join().unwrap();
789        std::fs::remove_file(&p).ok();
790    }
791}
subetha_cxc/shared_vec.rs

subetha_cxc/
shared_vec.rs
