subetha_cxc/
shared_region.rs

1//! `SharedRegion<T>` - cross-process typed arena with position-
2//! independent `OffsetPtr<T>` references.
3//!
4//! The foundational building block for cross-process pointer-bearing
5//! data structures (BTree nodes, trie nodes, linked lists, any
6//! pointer-graph). Pointers are `OffsetPtr<T> { index: u32 }` which
7//! resolve via `mmap_base + index * size_of::<T>()` in any process.
8//!
9//! # Layout
10//!
11//! ```text
12//! +----------------------------+
13//! | RegionHeader (64B aligned) |
14//! |   magic, capacity          |
15//! |   bump_next: AtomicU64     |
16//! |   free_head: AtomicU64     |  (counter << 32 | index)
17//! +----------------------------+
18//! | next[capacity]             |  (free-list links, when slot free)
19//! +----------------------------+
20//! | slots[capacity]            |  (T payloads, size_of<T> each)
21//! +----------------------------+
22//! ```
23//!
24//! Free-list links live in a separate `next[capacity]` array (not
25//! union'd into T storage) because T need not be 4-byte aligned. The
26//! cost is `4 * capacity` extra bytes; the gain is layout simplicity
27//! and zero overhead on the T storage.
28//!
29//! # Concurrent allocation protocol
30//!
31//! Allocate:
32//! 1. Try to pop from `free_head` (lock-free Treiber stack):
33//!    - Load packed `(counter, index)`.
34//!    - If `index == NIL_INDEX`, free list is empty; fall through.
35//!    - Read `next[index]`; CAS `free_head` to `(counter+1, next)`.
36//!    - On success, return `OffsetPtr { index }`.
37//! 2. Bump alloc: `bump_next.fetch_add(1, AcqRel)`.
38//!    - If the returned index >= capacity, rollback and return Full.
39//!    - Write `value` into `slots[index]`; return `OffsetPtr { index }`.
40//!
41//! Free:
42//! 1. Read current `(counter, head)` from `free_head`.
43//! 2. Write `head` into `next[ptr.index]`.
44//! 3. CAS `free_head` to `(counter+1, ptr.index)`.
45//! 4. On CAS lose, retry from step 1.
46//!
47//! # ABA safety
48//!
49//! The 32-bit counter prevents ABA: every push bumps the counter, so
50//! the packed word is different even when an index repeats. 32 bits
51//! of counter span 4B operations, which is far beyond any realistic
52//! concrete race window.
53//!
54//! # No drop semantics
55//!
56//! T: Copy + Sized. Allocated T values are NOT dropped on `free`
57//! (Copy types don't need drop, and we can't run drop glue on bytes
58//! living in shared memory anyway). `free` returns the value as a
59//! by-copy.
60
61use std::fs::{File, OpenOptions};
62use std::marker::PhantomData;
63use std::mem::size_of;
64use std::path::Path;
65use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
66
67use memmap2::{MmapMut, MmapOptions};
68
69pub const REGION_MAGIC: u32 = 0x4150_5247;
70pub const NIL_INDEX: u32 = u32::MAX;
71
72#[repr(C, align(64))]
73pub struct RegionHeader {
74    pub magic: u32,
75    pub capacity: u32,
76    pub slot_size: u32,
77    _pad1: u32,
78    pub bump_next: AtomicU64,
79    pub free_head: AtomicU64,
80    _pad2: [u8; 32],
81}
82
83const _: () = {
84    assert!(size_of::<RegionHeader>() == 64);
85};
86
87pub const fn region_file_size(capacity: usize, slot_size: usize) -> usize {
88    size_of::<RegionHeader>()
89        + capacity * size_of::<AtomicU32>()    // next[] array
90        + capacity * slot_size                 // slots[]
91}
92
93#[derive(Debug, Clone, Copy, PartialEq, Eq)]
94pub enum RegionError {
95    Full,
96    InvalidPtr,
97    PayloadTooLarge,
98    LayoutMismatch,
99    IoError(std::io::ErrorKind),
100}
101
102impl From<std::io::Error> for RegionError {
103    fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
104}
105
106/// Position-independent pointer to a slot in a SharedRegion. Stable
107/// across processes because the underlying MMF is byte-identical;
108/// adding `mmap_base + ptr.index * size_of::<T>()` resolves in any
109/// process.
110#[derive(Debug)]
111#[repr(C)]
112pub struct OffsetPtr<T> {
113    pub index: u32,
114    _phantom: PhantomData<T>,
115}
116
117impl<T> Clone for OffsetPtr<T> {
118    fn clone(&self) -> Self { *self }
119}
120impl<T> Copy for OffsetPtr<T> {}
121impl<T> PartialEq for OffsetPtr<T> {
122    fn eq(&self, other: &Self) -> bool { self.index == other.index }
123}
124impl<T> Eq for OffsetPtr<T> {}
125impl<T> std::hash::Hash for OffsetPtr<T> {
126    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
127        self.index.hash(state);
128    }
129}
130
131impl<T> OffsetPtr<T> {
132    pub const NIL: Self = Self { index: NIL_INDEX, _phantom: PhantomData };
133
134    #[inline]
135    pub fn new(index: u32) -> Self {
136        Self { index, _phantom: PhantomData }
137    }
138
139    #[inline]
140    pub fn is_nil(self) -> bool { self.index == NIL_INDEX }
141}
142
143#[inline]
144fn pack(counter: u32, index: u32) -> u64 {
145    ((counter as u64) << 32) | (index as u64)
146}
147#[inline]
148fn unpack(v: u64) -> (u32, u32) {
149    ((v >> 32) as u32, v as u32)
150}
151
152pub struct SharedRegion<T: Copy + 'static> {
153    _file: File,
154    mmap: MmapMut,
155    capacity: usize,
156    next_offset: usize,
157    slots_offset: usize,
158    _phantom: PhantomData<T>,
159    header_sidecar: subetha_core::HandshakeHeader,
160    ring_sidecar: Box<subetha_core::ObservationRing>,
161}
162
163unsafe impl<T: Copy + Send + 'static> Send for SharedRegion<T> {}
164unsafe impl<T: Copy + Sync + 'static> Sync for SharedRegion<T> {}
165
166impl<T: Copy + Send + Sync + 'static> subetha_sidecar::AdaptiveInstance for SharedRegion<T> {
167    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
168    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
169    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
170        Box::new(subetha_sidecar::NoMigrationPolicy)
171    }
172}
173
174impl<T: Copy + 'static> SharedRegion<T> {
175    /// Obtain the region at `path`, initializing an empty one if the
176    /// path does not yet exist and attaching to it if it does.
177    /// Attaching leaves allocated slots and the free list in place, so
178    /// outstanding [`OffsetPtr`]s stay resolvable; a region built with
179    /// a different capacity or payload type is a `LayoutMismatch`.
180    /// [`reset`](Self::reset) reinitializes.
181    pub fn create(
182        path: impl AsRef<Path>, capacity: usize,
183    ) -> Result<Self, RegionError> {
184        assert!(capacity >= 1);
185        assert!(capacity < NIL_INDEX as usize, "capacity must be < u32::MAX");
186        let (file, mut mmap) = crate::mmf_attach::create_or_attach(
187            path.as_ref(),
188            region_file_size(capacity, size_of::<T>()),
189            |ptr| unsafe { Self::init_region(ptr, capacity) },
190            |ptr| unsafe { (*(ptr as *const RegionHeader)).magic == REGION_MAGIC },
191        )?;
192        crate::mmf_warm::warm_mmap(&mut mmap);
193        Self::from_region(file, mmap, capacity)
194    }
195
196    /// Truncate the region at `path` and initialize an empty one,
197    /// invalidating every OffsetPtr live peers hold. For a caller that
198    /// knows it owns the path.
199    pub fn reset(
200        path: impl AsRef<Path>, capacity: usize,
201    ) -> Result<Self, RegionError> {
202        assert!(capacity >= 1);
203        assert!(capacity < NIL_INDEX as usize, "capacity must be < u32::MAX");
204        let (file, mut mmap) = crate::mmf_attach::reset(
205            path.as_ref(),
206            region_file_size(capacity, size_of::<T>()),
207            |ptr| unsafe { Self::init_region(ptr, capacity) },
208        )?;
209        crate::mmf_warm::warm_mmap(&mut mmap);
210        Self::from_region(file, mmap, capacity)
211    }
212
213    /// Lay out an empty region: config, the zero bump cursor and the
214    /// NIL free head first, magic last, because attachers spin on it.
215    /// The zeroed next[] array is the not-on-a-free-chain state.
216    ///
217    /// # Safety
218    /// `ptr` addresses at least `region_file_size(capacity,
219    /// size_of::<T>())` writable zeroed bytes.
220    unsafe fn init_region(ptr: *mut u8, capacity: usize) {
221        let hdr = ptr as *mut RegionHeader;
222        unsafe {
223            (*hdr).capacity = capacity as u32;
224            (*hdr).slot_size = size_of::<T>() as u32;
225            std::ptr::write(&raw mut (*hdr).free_head, AtomicU64::new(pack(0, NIL_INDEX)));
226            std::ptr::write_volatile(&raw mut (*hdr).magic, REGION_MAGIC);
227        }
228    }
229
230    /// Wrap an initialized region, refusing one built with a different
231    /// capacity or payload type.
232    fn from_region(
233        file: File,
234        mmap: MmapMut,
235        capacity: usize,
236    ) -> Result<Self, RegionError> {
237        let hdr = unsafe { &*(mmap.as_ptr() as *const RegionHeader) };
238        if hdr.magic != REGION_MAGIC
239            || hdr.capacity != capacity as u32
240            || hdr.slot_size != size_of::<T>() as u32
241        {
242            return Err(RegionError::LayoutMismatch);
243        }
244        let next_offset = size_of::<RegionHeader>();
245        let slots_offset = next_offset + capacity * size_of::<AtomicU32>();
246        Ok(Self {
247            _file: file, mmap, capacity, next_offset, slots_offset,
248            _phantom: PhantomData,
249            header_sidecar: subetha_core::HandshakeHeader::new(),
250            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
251        })
252    }
253
254    pub fn open(
255        path: impl AsRef<Path>, expected_capacity: usize,
256    ) -> Result<Self, RegionError> {
257        let total = region_file_size(expected_capacity, size_of::<T>());
258        let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
259        if file.metadata()?.len() < total as u64 {
260            return Err(RegionError::LayoutMismatch);
261        }
262        let mut mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
263        crate::mmf_warm::warm_mmap(&mut mmap);
264        Self::from_region(file, mmap, expected_capacity)
265    }
266
267    #[inline]
268    pub fn capacity(&self) -> usize { self.capacity }
269
270    /// Raw pointer to the start of the MMF. Useful for primitives
271    /// built on top of SharedRegion that need direct memory access
272    /// (e.g., SharedLinkedList reads node next/prev/value fields directly
273    /// at computed offsets instead of copying the whole node).
274    #[inline]
275    pub fn mmap_ptr(&self) -> *const u8 { self.mmap.as_ptr() }
276
277    fn header(&self) -> &RegionHeader {
278        unsafe { &*(self.mmap.as_ptr() as *const RegionHeader) }
279    }
280
281    fn next_link(&self, idx: usize) -> &AtomicU32 {
282        let base = unsafe { self.mmap.as_ptr().add(self.next_offset) };
283        unsafe { &*(base.add(idx * size_of::<AtomicU32>()) as *const AtomicU32) }
284    }
285
286    fn slot_ptr(&self, idx: usize) -> *mut T {
287        let base = unsafe { self.mmap.as_ptr().add(self.slots_offset) };
288        unsafe { base.add(idx * size_of::<T>()) as *mut T }
289    }
290
291    /// Number of slots currently allocated (bump high-water minus
292    /// free-list length). This is a snapshot; concurrent
293    /// alloc/free may race.
294    pub fn len(&self) -> usize {
295        let bump = self.header().bump_next.load(Ordering::Acquire) as usize;
296        bump.saturating_sub(self.free_count())
297    }
298
299    pub fn is_empty(&self) -> bool { self.len() == 0 }
300
301    /// Walk the free list (best-effort under concurrent activity)
302    /// and return its length. O(N_free).
303    pub fn free_count(&self) -> usize {
304        let mut count = 0usize;
305        let (_, mut idx) = unpack(self.header().free_head.load(Ordering::Acquire));
306        let mut visited = 0;
307        while idx != NIL_INDEX && (visited as usize) < self.capacity {
308            count += 1;
309            visited += 1;
310            idx = self.next_link(idx as usize).load(Ordering::Acquire);
311        }
312        count
313    }
314
315    /// Allocate a slot. Tries the free list first, then bump
316    /// allocates. Returns `Err(Full)` when both are exhausted.
317    pub fn allocate(&self, value: T) -> Result<OffsetPtr<T>, RegionError> {
318        // 1. Try Treiber-stack pop from the free list.
319        loop {
320            let head = self.header().free_head.load(Ordering::Acquire);
321            let (counter, idx) = unpack(head);
322            if idx == NIL_INDEX { break; } // free list empty; bump path
323            let next_idx = self.next_link(idx as usize).load(Ordering::Acquire);
324            let new_head = pack(counter.wrapping_add(1), next_idx);
325            if self.header().free_head.compare_exchange(
326                head, new_head, Ordering::AcqRel, Ordering::Acquire,
327            ).is_ok() {
328                unsafe { std::ptr::write(self.slot_ptr(idx as usize), value); }
329                self.ring_sidecar
330                    .push_op(crate::sidecar_ops::region::OP_ALLOCATE, 0);
331                return Ok(OffsetPtr::new(idx));
332            }
333            // CAS lost; retry.
334        }
335        // 2. Bump allocation.
336        let idx = self.header().bump_next.fetch_add(1, Ordering::AcqRel);
337        if idx >= self.capacity as u64 {
338            self.header().bump_next.fetch_sub(1, Ordering::AcqRel);
339            self.ring_sidecar
340                .push_op(crate::sidecar_ops::region::OP_ALLOCATE, 1); // full
341            return Err(RegionError::Full);
342        }
343        let idx = idx as u32;
344        unsafe { std::ptr::write(self.slot_ptr(idx as usize), value); }
345        self.ring_sidecar
346            .push_op(crate::sidecar_ops::region::OP_ALLOCATE, 0);
347        Ok(OffsetPtr::new(idx))
348    }
349
350    /// Free a slot, returning the T it held. Pushes onto the
351    /// Treiber-stack free list.
352    pub fn free(&self, ptr: OffsetPtr<T>) -> Result<T, RegionError> {
353        if ptr.is_nil() || (ptr.index as usize) >= self.capacity {
354            self.ring_sidecar
355                .push_op(crate::sidecar_ops::region::OP_FREE, 1); // invalid ptr
356            return Err(RegionError::InvalidPtr);
357        }
358        let value = unsafe { std::ptr::read(self.slot_ptr(ptr.index as usize)) };
359        loop {
360            let head = self.header().free_head.load(Ordering::Acquire);
361            let (counter, old_top) = unpack(head);
362            self.next_link(ptr.index as usize).store(old_top, Ordering::Release);
363            let new_head = pack(counter.wrapping_add(1), ptr.index);
364            if self.header().free_head.compare_exchange(
365                head, new_head, Ordering::AcqRel, Ordering::Acquire,
366            ).is_ok() {
367                self.ring_sidecar
368                    .push_op(crate::sidecar_ops::region::OP_FREE, 0);
369                return Ok(value);
370            }
371            // CAS lost; retry.
372        }
373    }
374
375    /// Read the value at `ptr`. Returns `Err(InvalidPtr)` for nil or
376    /// out-of-bounds. Caller is responsible for ensuring the ptr
377    /// refers to a still-allocated slot (free'd slots may be
378    /// reallocated and contain a different value).
379    pub fn get(&self, ptr: OffsetPtr<T>) -> Result<T, RegionError> {
380        if ptr.is_nil() || (ptr.index as usize) >= self.capacity {
381            self.ring_sidecar
382                .push_op(crate::sidecar_ops::region::OP_GET, 1); // invalid ptr
383            return Err(RegionError::InvalidPtr);
384        }
385        let v = unsafe { std::ptr::read(self.slot_ptr(ptr.index as usize)) };
386        self.ring_sidecar
387            .push_op(crate::sidecar_ops::region::OP_GET, 0);
388        Ok(v)
389    }
390
391    /// Overwrite the value at `ptr`. Same caveats as `get`: caller
392    /// must hold a still-valid pointer.
393    pub fn set(&self, ptr: OffsetPtr<T>, value: T) -> Result<(), RegionError> {
394        if ptr.is_nil() || (ptr.index as usize) >= self.capacity {
395            self.ring_sidecar
396                .push_op(crate::sidecar_ops::region::OP_SET, 1); // invalid ptr
397            return Err(RegionError::InvalidPtr);
398        }
399        unsafe { std::ptr::write(self.slot_ptr(ptr.index as usize), value); }
400        self.ring_sidecar
401            .push_op(crate::sidecar_ops::region::OP_SET, 0);
402        Ok(())
403    }
404
405    /// Reset the region to empty: bump pointer back to 0 and free
406    /// list back to NIL. Existing `OffsetPtr` values become stale;
407    /// callers must drop them. Useful for steady-state benches that
408    /// need to reset accumulated state between iterations.
409    ///
410    /// Not thread-safe: do not call concurrently with `allocate` /
411    /// `free` from other threads. Intended for single-threaded reset
412    /// (e.g. test setup, bench iter setup).
413    pub fn clear(&self) {
414        self.header().bump_next.store(0, Ordering::Release);
415        self.header().free_head.store(pack(0, NIL_INDEX), Ordering::Release);
416    }
417
418    pub fn flush(&self) -> Result<(), RegionError> {
419        self.mmap.flush()?;
420        Ok(())
421    }
422
423    /// Non-blocking flush: schedules a writeback via the OS.
424    /// Note: Windows is only partially async (sync to page cache,
425    /// not to disk).
426    pub fn flush_async(&self) -> Result<(), RegionError> {
427        self.mmap.flush_async()?;
428        Ok(())
429    }
430}
431
432#[cfg(test)]
433mod tests {
434    use super::*;
435    use std::sync::Arc;
436    use std::thread;
437
438    fn tmp(name: &str) -> std::path::PathBuf {
439        let mut p = std::env::temp_dir();
440        let pid = std::process::id();
441        p.push(format!("subetha-region-{name}-{pid}.bin"));
442        p
443    }
444
445    #[test]
446    fn create_initial_state_is_empty() {
447        let p = tmp("init");
448        let r: SharedRegion<u64> = SharedRegion::create(&p, 16).unwrap();
449        assert_eq!(r.capacity(), 16);
450        assert_eq!(r.len(), 0);
451        assert!(r.is_empty());
452        assert_eq!(r.free_count(), 0);
453        std::fs::remove_file(&p).ok();
454    }
455
456    /// A second create attaches with allocated slots in place; reset
457    /// is what strips them.
458    #[test]
459    fn second_create_attaches_and_keeps_allocations() {
460        let p = tmp("attach");
461        std::fs::remove_file(&p).ok();
462        let r: SharedRegion<u64> = SharedRegion::create(&p, 16).unwrap();
463        let ptr = r.allocate(777).unwrap();
464
465        let r2: SharedRegion<u64> = SharedRegion::create(&p, 16).unwrap();
466        assert_eq!(r2.get(ptr), Ok(777), "attach lost an allocation");
467        assert!(matches!(
468            SharedRegion::<u64>::create(&p, 8),
469            Err(RegionError::LayoutMismatch),
470        ));
471
472        // Windows refuses to truncate a mapped file, so every handle goes
473        // before the reset.
474        drop(r);
475        drop(r2);
476        let fresh: SharedRegion<u64> = SharedRegion::reset(&p, 16).unwrap();
477        assert!(fresh.is_empty(), "reset kept an allocation");
478        drop(fresh);
479        std::fs::remove_file(&p).ok();
480    }
481
482    #[test]
483    fn allocate_and_get_round_trip() {
484        let p = tmp("rt");
485        let r: SharedRegion<u64> = SharedRegion::create(&p, 8).unwrap();
486        let p1 = r.allocate(100).unwrap();
487        let p2 = r.allocate(200).unwrap();
488        let p3 = r.allocate(300).unwrap();
489        assert_eq!(p1.index, 0);
490        assert_eq!(p2.index, 1);
491        assert_eq!(p3.index, 2);
492        assert_eq!(r.get(p1).unwrap(), 100);
493        assert_eq!(r.get(p2).unwrap(), 200);
494        assert_eq!(r.get(p3).unwrap(), 300);
495        assert_eq!(r.len(), 3);
496        std::fs::remove_file(&p).ok();
497    }
498
499    #[test]
500    fn full_region_returns_error_on_bump() {
501        let p = tmp("full");
502        let r: SharedRegion<u32> = SharedRegion::create(&p, 4).unwrap();
503        for i in 0..4u32 { r.allocate(i).unwrap(); }
504        assert_eq!(r.allocate(99).err(), Some(RegionError::Full));
505        assert_eq!(r.len(), 4);  // rolled back
506        std::fs::remove_file(&p).ok();
507    }
508
509    #[test]
510    fn free_returns_value_and_decrements_len() {
511        let p = tmp("free");
512        let r: SharedRegion<u64> = SharedRegion::create(&p, 8).unwrap();
513        let ptr1 = r.allocate(111).unwrap();
514        let ptr2 = r.allocate(222).unwrap();
515        assert_eq!(r.len(), 2);
516        let v = r.free(ptr1).unwrap();
517        assert_eq!(v, 111);
518        assert_eq!(r.len(), 1);
519        assert_eq!(r.free_count(), 1);
520        // ptr2 still valid.
521        assert_eq!(r.get(ptr2).unwrap(), 222);
522        std::fs::remove_file(&p).ok();
523    }
524
525    #[test]
526    fn free_then_allocate_reuses_slot() {
527        let p = tmp("reuse");
528        let r: SharedRegion<u64> = SharedRegion::create(&p, 8).unwrap();
529        let p1 = r.allocate(100).unwrap();
530        let _p2 = r.allocate(200).unwrap();
531        r.free(p1).unwrap();
532        // Next allocate should reuse p1's slot (LIFO Treiber stack).
533        let p3 = r.allocate(999).unwrap();
534        assert_eq!(p3.index, p1.index, "free list should have returned slot 0");
535        assert_eq!(r.get(p3).unwrap(), 999);
536        std::fs::remove_file(&p).ok();
537    }
538
539    #[test]
540    fn free_invalid_ptr_returns_error() {
541        let p = tmp("invalid");
542        let r: SharedRegion<u64> = SharedRegion::create(&p, 4).unwrap();
543        assert_eq!(r.free(OffsetPtr::NIL).err(), Some(RegionError::InvalidPtr));
544        let oob = OffsetPtr::<u64>::new(100);
545        assert_eq!(r.free(oob).err(), Some(RegionError::InvalidPtr));
546        std::fs::remove_file(&p).ok();
547    }
548
549    #[test]
550    fn set_overwrites_value_in_place() {
551        let p = tmp("set");
552        let r: SharedRegion<u64> = SharedRegion::create(&p, 4).unwrap();
553        let ptr = r.allocate(42).unwrap();
554        r.set(ptr, 999).unwrap();
555        assert_eq!(r.get(ptr).unwrap(), 999);
556        std::fs::remove_file(&p).ok();
557    }
558
559    #[test]
560    fn cross_handle_visibility() {
561        let p = tmp("cross-handle");
562        let writer: SharedRegion<u64> = SharedRegion::create(&p, 16).unwrap();
563        let reader: SharedRegion<u64> = SharedRegion::open(&p, 16).unwrap();
564        let ptr = writer.allocate(7777).unwrap();
565        assert_eq!(reader.get(ptr).unwrap(), 7777);
566        writer.set(ptr, 8888).unwrap();
567        assert_eq!(reader.get(ptr).unwrap(), 8888);
568        std::fs::remove_file(&p).ok();
569    }
570
571    #[test]
572    fn concurrent_allocations_get_distinct_indices() {
573        let p = tmp("concurrent");
574        let r: Arc<SharedRegion<u64>> = Arc::new(SharedRegion::create(&p, 1024).unwrap());
575        let n_threads = 4;
576        let per_thread = 100;
577        let mut handles = vec![];
578        for t in 0..n_threads as u64 {
579            let r = r.clone();
580            handles.push(thread::spawn(move || {
581                let mut ptrs = vec![];
582                for i in 0..per_thread as u64 {
583                    let v = t * 1000 + i;
584                    let ptr = r.allocate(v).unwrap();
585                    ptrs.push((v, ptr));
586                }
587                ptrs
588            }));
589        }
590        let all: Vec<(u64, OffsetPtr<u64>)> = handles.into_iter()
591            .flat_map(|h| h.join().unwrap()).collect();
592
593        // All indices distinct.
594        let mut indices: Vec<u32> = all.iter().map(|(_, p)| p.index).collect();
595        indices.sort();
596        for w in indices.windows(2) {
597            assert_ne!(w[0], w[1], "two threads got the same slot index");
598        }
599        // Every allocation reads back its written value.
600        for (expected, ptr) in &all {
601            assert_eq!(r.get(*ptr).unwrap(), *expected);
602        }
603        std::fs::remove_file(&p).ok();
604    }
605
606    #[test]
607    fn concurrent_free_and_realloc_no_corruption() {
608        let p = tmp("free-realloc");
609        let r: Arc<SharedRegion<u64>> = Arc::new(SharedRegion::create(&p, 64).unwrap());
610        // Pre-populate.
611        let initial: Vec<OffsetPtr<u64>> = (0..64u64)
612            .map(|i| r.allocate(i * 10).unwrap()).collect();
613        let r_a = r.clone();
614        let _init = initial.clone();
615        // Worker A: free even-indexed slots.
616        let freer = thread::spawn(move || {
617            for (i, ptr) in _init.iter().enumerate() {
618                if i % 2 == 0 { r_a.free(*ptr).ok(); }
619            }
620        });
621        // Worker B: allocate new slots.
622        let r_b = r.clone();
623        let alloc = thread::spawn(move || {
624            let mut new_ptrs = vec![];
625            for i in 100u64..132 {
626                if let Ok(p) = r_b.allocate(i) { new_ptrs.push((i, p)); }
627            }
628            new_ptrs
629        });
630        freer.join().unwrap();
631        let new_ptrs = alloc.join().unwrap();
632        // Each new ptr resolves to its value.
633        for (val, ptr) in new_ptrs {
634            assert_eq!(r.get(ptr).unwrap(), val);
635        }
636        std::fs::remove_file(&p).ok();
637    }
638
639    #[test]
640    fn offset_ptr_is_position_independent() {
641        // The whole point: the same index resolves to the same value
642        // in two independently-mapped views of the same file.
643        let p = tmp("position-indep");
644        let writer: SharedRegion<u64> = SharedRegion::create(&p, 4).unwrap();
645        let ptr = writer.allocate(0xCAFE_BABE).unwrap();
646        let reader: SharedRegion<u64> = SharedRegion::open(&p, 4).unwrap();
647        // Verify the ptr's raw index field (u32) is the same in both.
648        let same_ptr = OffsetPtr::<u64>::new(ptr.index);
649        assert_eq!(reader.get(same_ptr).unwrap(), 0xCAFE_BABE);
650        std::fs::remove_file(&p).ok();
651    }
652
653    #[test]
654    fn struct_payload_round_trip() {
655        #[derive(Clone, Copy, Debug, PartialEq)]
656        #[repr(C)]
657        struct Node { left: u32, right: u32, key: u64 }
658        let p = tmp("struct");
659        let r: SharedRegion<Node> = SharedRegion::create(&p, 16).unwrap();
660        let n = Node { left: 1, right: 2, key: 42 };
661        let ptr = r.allocate(n).unwrap();
662        assert_eq!(r.get(ptr).unwrap(), n);
663        std::fs::remove_file(&p).ok();
664    }
665
666    #[test]
667    fn nil_ptr_round_trips() {
668        let p: OffsetPtr<u64> = OffsetPtr::NIL;
669        assert!(p.is_nil());
670        assert_eq!(p.index, NIL_INDEX);
671    }
672
673    #[test]
674    fn offset_ptr_equality_and_hash() {
675        use std::collections::HashSet;
676        let a: OffsetPtr<u64> = OffsetPtr::new(5);
677        let b: OffsetPtr<u64> = OffsetPtr::new(5);
678        let c: OffsetPtr<u64> = OffsetPtr::new(6);
679        assert_eq!(a, b);
680        assert_ne!(a, c);
681        let mut s = HashSet::new();
682        s.insert(a);
683        assert!(s.contains(&b));
684        assert!(!s.contains(&c));
685    }
686
687    #[test]
688    fn disk_persistence_survives_reopen() {
689        let p = tmp("disk");
690        let saved_ptr_index;
691        {
692            let r: SharedRegion<u64> = SharedRegion::create(&p, 16).unwrap();
693            let p1 = r.allocate(1111).unwrap();
694            let _p2 = r.allocate(2222).unwrap();
695            r.flush().unwrap();
696            saved_ptr_index = p1.index;
697        }
698        let r2: SharedRegion<u64> = SharedRegion::open(&p, 16).unwrap();
699        let restored = OffsetPtr::<u64>::new(saved_ptr_index);
700        assert_eq!(r2.get(restored).unwrap(), 1111);
701        // Continue allocating.
702        let p3 = r2.allocate(3333).unwrap();
703        assert_eq!(p3.index, 2);
704        assert_eq!(r2.get(p3).unwrap(), 3333);
705        std::fs::remove_file(&p).ok();
706    }
707}
subetha_cxc/shared_region.rs

subetha_cxc/
shared_region.rs
