subetha_cxc/
shared_handle_table.rs

1//! `SharedHandleTable<T>` - cross-process ECS-style slotmap.
2//!
3//! Same architectural shape as the in-process `AdaptiveHandle` /
4//! `Slotmap`, but the slot table lives in a memory-mapped file so
5//! handles are valid across processes. Handle = `(generation: u32,
6//! slot: u32)` packed into a `u64`; generation bumps on every
7//! insert and every remove so stale handles fail visibility check
8//! across the process boundary.
9//!
10//! # Layout
11//!
12//! ```text
13//! +-----------------------------+
14//! | HandleHeader (64B)          |
15//! |   - magic                   |
16//! |   - capacity                |
17//! |   - slot_size               |
18//! |   - free_list_head_packed   |  (counter:u32, slot:u32) packed u64
19//! |   - live_count              |
20//! +-----------------------------+
21//! | Slot[0] (64B cache line)    |
22//! |   - generation: AtomicU32   |
23//! |   - occupied:   AtomicU32   |
24//! |   - next_free:  AtomicU32   |
25//! |   - _pad:       u32         |
26//! |   - payload:    [u8; 48]    |
27//! +-----------------------------+
28//! | Slot[1] ...                 |
29//! +-----------------------------+
30//! ```
31//!
32//! # Generation parity
33//!
34//! Even generation = vacant, odd = occupied. Bumped on every
35//! insert (vacant -> occupied) and every remove (occupied ->
36//! vacant). A handle from generation N matches only when the slot
37//! is currently at generation N.
38//!
39//! # Free list
40//!
41//! ABA-free Treiber stack. Head is packed `(counter, slot_idx)` so
42//! every CAS bumps the counter, making the (head, next) sequence
43//! distinguishable from the (head, _, next-after-reinsert) sequence
44//! that would otherwise alias. Each vacant slot's `next_free` field
45//! is the linkage.
46
47use std::fs::{File, OpenOptions};
48use std::marker::PhantomData;
49use std::mem::{align_of, size_of};
50use std::path::Path;
51use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
52
53use memmap2::{MmapMut, MmapOptions};
54
55pub const HANDLE_TABLE_MAGIC: u64 = 0x4150_4D46_5354_424C;
56
57pub const SLOT_PAYLOAD_BYTES: usize = 48;
58
59/// Sentinel for "no slot" in the free list.
60pub const NIL_SLOT: u32 = u32::MAX;
61
62/// Generation 0 is reserved for `Handle::NULL`.
63const GEN_VACANT_BIT: u32 = 0;
64
65#[repr(C, align(64))]
66pub struct HandleHeader {
67    pub magic: u64,
68    pub capacity: u32,
69    pub slot_size: u32,
70    pub free_list_head: AtomicU64,
71    pub live_count: AtomicU64,
72    _pad: [u8; 32],
73}
74
75#[repr(C, align(64))]
76pub struct SharedSlot {
77    pub generation: AtomicU32,
78    pub occupied: AtomicU32,
79    pub next_free: AtomicU32,
80    _pad: u32,
81    pub payload: [u8; SLOT_PAYLOAD_BYTES],
82}
83
84/// Opaque cross-process handle. Packed `(generation: u32 high, slot: u32 low)`.
85#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
86pub struct Handle(u64);
87
88impl Handle {
89    pub const NULL: Self = Self(0);
90
91    #[inline]
92    pub const fn from_parts(generation: u32, slot: u32) -> Self {
93        Self(((generation as u64) << 32) | (slot as u64))
94    }
95
96    #[inline]
97    pub const fn generation(self) -> u32 { (self.0 >> 32) as u32 }
98
99    #[inline]
100    pub const fn slot(self) -> u32 { self.0 as u32 }
101
102    #[inline]
103    pub const fn is_null(self) -> bool { self.0 == 0 }
104
105    #[inline]
106    pub const fn raw(self) -> u64 { self.0 }
107}
108
109pub const fn slot_offset() -> usize { size_of::<HandleHeader>() }
110
111pub const fn handle_table_file_size(capacity: usize) -> usize {
112    slot_offset() + capacity * size_of::<SharedSlot>()
113}
114
115#[derive(Debug, Clone, Copy, PartialEq, Eq)]
116pub enum HandleTableError {
117    LayoutMismatch,
118    PayloadTooLarge,
119    Full,
120    IoError(std::io::ErrorKind),
121}
122
123impl From<std::io::Error> for HandleTableError {
124    fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
125}
126
127pub struct SharedHandleTable<T: Copy + 'static> {
128    _file: File,
129    mmap: MmapMut,
130    capacity: usize,
131    _phantom: PhantomData<T>,
132    header_sidecar: subetha_core::HandshakeHeader,
133    ring_sidecar: Box<subetha_core::ObservationRing>,
134}
135
136unsafe impl<T: Copy + Send + 'static> Send for SharedHandleTable<T> {}
137unsafe impl<T: Copy + Sync + 'static> Sync for SharedHandleTable<T> {}
138
139impl<T: Copy + Send + Sync + 'static> subetha_sidecar::AdaptiveInstance for SharedHandleTable<T> {
140    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
141    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
142    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
143        Box::new(subetha_sidecar::NoMigrationPolicy)
144    }
145}
146
147#[inline]
148fn pack_head(counter: u32, slot: u32) -> u64 {
149    ((counter as u64) << 32) | (slot as u64)
150}
151
152#[inline]
153fn unpack_head(v: u64) -> (u32, u32) {
154    ((v >> 32) as u32, v as u32)
155}
156
157impl<T: Copy + 'static> SharedHandleTable<T> {
158    pub fn create(path: impl AsRef<Path>, capacity: usize) -> Result<Self, HandleTableError> {
159        Self::check_layout()?;
160        assert!(capacity >= 1 && capacity <= (u32::MAX - 1) as usize);
161        let total = handle_table_file_size(capacity);
162        let file = OpenOptions::new()
163            .read(true).write(true).create(true).truncate(true)
164            .open(path.as_ref())?;
165        file.set_len(total as u64)?;
166        let mut mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
167        let hdr = mmap.as_mut_ptr() as *mut HandleHeader;
168        // Initial free list: link all slots in order [0 -> 1 -> 2 -> ... -> NIL].
169        unsafe {
170            std::ptr::write(hdr, HandleHeader {
171                magic: HANDLE_TABLE_MAGIC,
172                capacity: capacity as u32,
173                slot_size: size_of::<T>() as u32,
174                free_list_head: AtomicU64::new(pack_head(0, 0)),
175                live_count: AtomicU64::new(0),
176                _pad: [0; 32],
177            });
178        }
179        let slots_base = unsafe { mmap.as_mut_ptr().add(slot_offset()) };
180        for i in 0..capacity {
181            let slot_ptr = unsafe { slots_base.add(i * size_of::<SharedSlot>()) as *mut SharedSlot };
182            let next = if i + 1 < capacity { (i + 1) as u32 } else { NIL_SLOT };
183            unsafe {
184                std::ptr::write(slot_ptr, SharedSlot {
185                    generation: AtomicU32::new(GEN_VACANT_BIT),
186                    occupied: AtomicU32::new(0),
187                    next_free: AtomicU32::new(next),
188                    _pad: 0,
189                    payload: [0; SLOT_PAYLOAD_BYTES],
190                });
191            }
192        }
193        Ok(Self {
194            _file: file, mmap, capacity, _phantom: PhantomData,
195            header_sidecar: subetha_core::HandshakeHeader::new(),
196            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
197        })
198    }
199
200    pub fn open(path: impl AsRef<Path>, expected_capacity: usize) -> Result<Self, HandleTableError> {
201        Self::check_layout()?;
202        let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
203        let total = handle_table_file_size(expected_capacity);
204        if file.metadata()?.len() < total as u64 {
205            return Err(HandleTableError::LayoutMismatch);
206        }
207        let mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
208        let hdr = unsafe { &*(mmap.as_ptr() as *const HandleHeader) };
209        if hdr.magic != HANDLE_TABLE_MAGIC
210            || hdr.capacity != expected_capacity as u32
211            || hdr.slot_size as usize != size_of::<T>()
212        {
213            return Err(HandleTableError::LayoutMismatch);
214        }
215        Ok(Self {
216            _file: file, mmap, capacity: expected_capacity, _phantom: PhantomData,
217            header_sidecar: subetha_core::HandshakeHeader::new(),
218            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
219        })
220    }
221
222    fn check_layout() -> Result<(), HandleTableError> {
223        if size_of::<T>() > SLOT_PAYLOAD_BYTES {
224            return Err(HandleTableError::PayloadTooLarge);
225        }
226        if align_of::<T>() > 8 {
227            return Err(HandleTableError::PayloadTooLarge);
228        }
229        Ok(())
230    }
231
232    #[inline]
233    pub fn capacity(&self) -> usize { self.capacity }
234
235    #[inline]
236    pub fn header(&self) -> &HandleHeader {
237        unsafe { &*(self.mmap.as_ptr() as *const HandleHeader) }
238    }
239
240    #[inline]
241    fn slot(&self, idx: u32) -> &SharedSlot {
242        debug_assert!((idx as usize) < self.capacity);
243        let base = unsafe { self.mmap.as_ptr().add(slot_offset()) };
244        unsafe { &*(base.add((idx as usize) * size_of::<SharedSlot>()) as *const SharedSlot) }
245    }
246
247    /// Pop the head of the free list. ABA-free via the packed counter.
248    /// Returns `None` when the table is full.
249    fn pop_free(&self) -> Option<u32> {
250        let header = self.header();
251        loop {
252            let head = header.free_list_head.load(Ordering::Acquire);
253            let (cnt, idx) = unpack_head(head);
254            if idx == NIL_SLOT { return None; }
255            let next = self.slot(idx).next_free.load(Ordering::Acquire);
256            let new_head = pack_head(cnt.wrapping_add(1), next);
257            if header.free_list_head.compare_exchange_weak(
258                head, new_head, Ordering::AcqRel, Ordering::Acquire,
259            ).is_ok() {
260                return Some(idx);
261            }
262            std::hint::spin_loop();
263        }
264    }
265
266    /// Push `slot_idx` onto the free list. Updates its `next_free` first.
267    fn push_free(&self, slot_idx: u32) {
268        let header = self.header();
269        loop {
270            let head = header.free_list_head.load(Ordering::Acquire);
271            let (cnt, head_idx) = unpack_head(head);
272            self.slot(slot_idx).next_free.store(head_idx, Ordering::Release);
273            let new_head = pack_head(cnt.wrapping_add(1), slot_idx);
274            if header.free_list_head.compare_exchange_weak(
275                head, new_head, Ordering::AcqRel, Ordering::Acquire,
276            ).is_ok() {
277                return;
278            }
279            std::hint::spin_loop();
280        }
281    }
282
283    /// Insert `value`, returning a Handle. Errors with `Full` when
284    /// the slot table is exhausted.
285    pub fn insert(&self, value: T) -> Result<Handle, HandleTableError> {
286        let slot_idx = match self.pop_free() {
287            Some(i) => i,
288            None => {
289                self.ring_sidecar
290                    .push_op(crate::sidecar_ops::ownership::OP_ACQUIRE, 1);
291                return Err(HandleTableError::Full);
292            }
293        };
294        let slot = self.slot(slot_idx);
295        // Bump generation: even (vacant) -> odd (occupied).
296        let new_gen = slot.generation.fetch_add(1, Ordering::AcqRel)
297            .wrapping_add(1)
298            .max(1);  // Skip the reserved gen=0 sentinel.
299        // SAFETY: this slot is exclusively ours (we just popped it
300        // from the free list) until we set occupied=1 below.
301        unsafe {
302            let dst = slot.payload.as_ptr() as *mut T;
303            std::ptr::write_unaligned(dst, value);
304        }
305        slot.occupied.store(1, Ordering::Release);
306        self.header().live_count.fetch_add(1, Ordering::AcqRel);
307        self.ring_sidecar
308            .push_op(crate::sidecar_ops::ownership::OP_ACQUIRE, 0);
309        Ok(Handle::from_parts(new_gen, slot_idx))
310    }
311
312    /// Look up by handle. Returns `None` when the handle is stale
313    /// (generation mismatch) or the slot is currently vacant.
314    pub fn get(&self, h: Handle) -> Option<T> {
315        let r = self.get_inner(h);
316        self.ring_sidecar.push_op(
317            crate::sidecar_ops::ownership::OP_GET,
318            if r.is_none() { 2 } else { 0 },
319        );
320        r
321    }
322
323    fn get_inner(&self, h: Handle) -> Option<T> {
324        if h.is_null() { return None; }
325        let slot_idx = h.slot();
326        if (slot_idx as usize) >= self.capacity { return None; }
327        let slot = self.slot(slot_idx);
328        // Re-check generation AFTER reading payload to catch
329        // mid-read modification by a remover.
330        loop {
331            let gen1 = slot.generation.load(Ordering::Acquire);
332            if gen1 != h.generation() { return None; }
333            if slot.occupied.load(Ordering::Acquire) == 0 { return None; }
334            let value: T = unsafe {
335                let src = slot.payload.as_ptr() as *const T;
336                std::ptr::read_unaligned(src)
337            };
338            let gen2 = slot.generation.load(Ordering::Acquire);
339            if gen1 == gen2 {
340                return Some(value);
341            }
342            // Concurrent remove + reinsert; retry.
343            std::hint::spin_loop();
344        }
345    }
346
347    /// True when handle is currently live.
348    pub fn contains(&self, h: Handle) -> bool {
349        if h.is_null() { return false; }
350        let slot_idx = h.slot();
351        if (slot_idx as usize) >= self.capacity { return false; }
352        let slot = self.slot(slot_idx);
353        slot.generation.load(Ordering::Acquire) == h.generation()
354            && slot.occupied.load(Ordering::Acquire) == 1
355    }
356
357    /// Remove the value at handle. Returns the value if live,
358    /// `None` if stale or already removed.
359    pub fn remove(&self, h: Handle) -> Option<T> {
360        let r = self.remove_inner(h);
361        self.ring_sidecar.push_op(
362            crate::sidecar_ops::ownership::OP_RELEASE,
363            if r.is_none() { 2 } else { 0 },
364        );
365        r
366    }
367
368    fn remove_inner(&self, h: Handle) -> Option<T> {
369        if h.is_null() { return None; }
370        let slot_idx = h.slot();
371        if (slot_idx as usize) >= self.capacity { return None; }
372        let slot = self.slot(slot_idx);
373        // Atomic CAS occupied 1 -> 0; loses to concurrent remove.
374        if slot.occupied.compare_exchange(
375            1, 0, Ordering::AcqRel, Ordering::Acquire,
376        ).is_err() {
377            return None;
378        }
379        // Generation check: when generation already differs, our
380        // remove was on a slot that's already been reused. Restore
381        // occupied (the other thread should have set it).
382        let cur_gen = slot.generation.load(Ordering::Acquire);
383        if cur_gen != h.generation() {
384            // Roll back: somebody else owns the slot.
385            slot.occupied.store(1, Ordering::Release);
386            return None;
387        }
388        let value: T = unsafe {
389            let src = slot.payload.as_ptr() as *const T;
390            std::ptr::read_unaligned(src)
391        };
392        // Bump generation to mark slot vacant again (odd -> even).
393        slot.generation.fetch_add(1, Ordering::AcqRel);
394        self.header().live_count.fetch_sub(1, Ordering::AcqRel);
395        // Return to free list.
396        self.push_free(slot_idx);
397        Some(value)
398    }
399
400    pub fn len(&self) -> usize {
401        self.header().live_count.load(Ordering::Acquire) as usize
402    }
403
404    pub fn is_empty(&self) -> bool { self.len() == 0 }
405
406    /// Non-blocking flush: schedules a writeback via the OS.
407    /// Note: Windows is only partially async (sync to page cache,
408    /// not to disk).
409    pub fn flush_async(&self) -> Result<(), HandleTableError> {
410        self.mmap.flush_async()?;
411        Ok(())
412    }
413
414    pub fn flush(&self) -> Result<(), HandleTableError> {
415        self.mmap.flush()?;
416        Ok(())
417    }
418}
419
420#[cfg(test)]
421mod tests {
422    use super::*;
423    use std::thread;
424    use std::sync::Arc;
425
426    fn tmp(name: &str) -> std::path::PathBuf {
427        let mut p = std::env::temp_dir();
428        let pid = std::process::id();
429        p.push(format!("subetha-handle-{name}-{pid}.bin"));
430        p
431    }
432
433    #[test]
434    fn insert_get_remove_round_trip() {
435        let p = tmp("rt");
436        let t: SharedHandleTable<u64> = SharedHandleTable::create(&p, 16).unwrap();
437        let h = t.insert(42).unwrap();
438        assert!(t.contains(h));
439        assert_eq!(t.get(h), Some(42));
440        assert_eq!(t.len(), 1);
441        let v = t.remove(h);
442        assert_eq!(v, Some(42));
443        assert!(!t.contains(h));
444        assert_eq!(t.get(h), None);
445        assert_eq!(t.len(), 0);
446        std::fs::remove_file(&p).ok();
447    }
448
449    #[test]
450    fn stale_handle_after_remove_returns_none() {
451        let p = tmp("stale");
452        let t: SharedHandleTable<u64> = SharedHandleTable::create(&p, 4).unwrap();
453        let h1 = t.insert(100).unwrap();
454        assert_eq!(t.remove(h1), Some(100));
455        // h1 now stale; reinsert into the same slot.
456        let h2 = t.insert(200).unwrap();
457        assert_eq!(h2.slot(), h1.slot(), "free-list LIFO reused the slot");
458        assert_ne!(h2.generation(), h1.generation(), "generation differs");
459        assert_eq!(t.get(h1), None, "stale h1 must not see h2's value");
460        assert_eq!(t.get(h2), Some(200));
461        std::fs::remove_file(&p).ok();
462    }
463
464    #[test]
465    fn full_table_insert_returns_error() {
466        let p = tmp("full");
467        let t: SharedHandleTable<u64> = SharedHandleTable::create(&p, 2).unwrap();
468        let _val = t.insert(1).unwrap();
469        let _val = t.insert(2).unwrap();
470        assert_eq!(t.insert(3), Err(HandleTableError::Full));
471        std::fs::remove_file(&p).ok();
472    }
473
474    #[test]
475    fn cross_handle_visibility() {
476        let p = tmp("cross-handle");
477        let writer: SharedHandleTable<u64> = SharedHandleTable::create(&p, 16).unwrap();
478        let reader: SharedHandleTable<u64> = SharedHandleTable::open(&p, 16).unwrap();
479        let h = writer.insert(777).unwrap();
480        assert_eq!(reader.get(h), Some(777));
481        assert!(reader.contains(h));
482        // Confirm remove returns the just-inserted value.
483        assert_eq!(writer.remove(h), Some(777));
484        assert_eq!(reader.get(h), None);
485        std::fs::remove_file(&p).ok();
486    }
487
488    #[test]
489    fn concurrent_inserts_and_removes_preserve_count() {
490        let p = tmp("concurrent");
491        let t: Arc<SharedHandleTable<u64>>
492            = Arc::new(SharedHandleTable::create(&p, 256).unwrap());
493        let n_threads = 4usize;
494        let per_thread = 50usize;
495        let mut handles = vec![];
496        for tid in 0..n_threads {
497            let t = t.clone();
498            handles.push(thread::spawn(move || {
499                let mut owned = Vec::with_capacity(per_thread);
500                for i in 0..per_thread {
501                    let v = (tid * per_thread + i) as u64;
502                    loop {
503                        match t.insert(v) {
504                            Ok(h) => { owned.push((h, v)); break; }
505                            Err(HandleTableError::Full) => {
506                                std::thread::yield_now();
507                            }
508                            Err(e) => panic!("unexpected error: {e:?}"),
509                        }
510                    }
511                }
512                // Verify everything we own is still ours.
513                for &(h, v) in &owned {
514                    assert_eq!(t.get(h), Some(v));
515                }
516                // Remove everything.
517                for (h, v) in owned {
518                    assert_eq!(t.remove(h), Some(v));
519                }
520            }));
521        }
522        for h in handles { h.join().unwrap(); }
523        assert_eq!(t.len(), 0);
524        std::fs::remove_file(&p).ok();
525    }
526
527    #[test]
528    fn disk_persistence_survives_reopen() {
529        let p = tmp("disk-persist");
530        let h1;
531        let h2;
532        {
533            let t: SharedHandleTable<u64> = SharedHandleTable::create(&p, 8).unwrap();
534            h1 = t.insert(11).unwrap();
535            h2 = t.insert(22).unwrap();
536            t.flush().unwrap();
537        }
538        let t2: SharedHandleTable<u64> = SharedHandleTable::open(&p, 8).unwrap();
539        assert_eq!(t2.get(h1), Some(11));
540        assert_eq!(t2.get(h2), Some(22));
541        assert_eq!(t2.len(), 2);
542        std::fs::remove_file(&p).ok();
543    }
544
545    #[test]
546    fn null_handle_returns_none() {
547        let p = tmp("null-handle");
548        let t: SharedHandleTable<u64> = SharedHandleTable::create(&p, 4).unwrap();
549        assert_eq!(t.get(Handle::NULL), None);
550        assert!(!t.contains(Handle::NULL));
551        assert_eq!(t.remove(Handle::NULL), None);
552        std::fs::remove_file(&p).ok();
553    }
554
555    #[test]
556    fn handle_packing_round_trips() {
557        let h = Handle::from_parts(42, 1337);
558        assert_eq!(h.generation(), 42);
559        assert_eq!(h.slot(), 1337);
560        assert!(!h.is_null());
561        assert!(Handle::NULL.is_null());
562        assert_eq!(Handle::NULL.generation(), 0);
563        assert_eq!(Handle::NULL.slot(), 0);
564    }
565
566    #[test]
567    fn struct_payload_round_trip() {
568        #[derive(Clone, Copy, Debug, PartialEq)]
569        #[repr(C)]
570        struct Item { id: u32, weight: f32, flags: u32 }
571        let p = tmp("struct");
572        let t: SharedHandleTable<Item> = SharedHandleTable::create(&p, 8).unwrap();
573        let item = Item { id: 42, weight: 1.5, flags: 0xFF };
574        let h = t.insert(item).unwrap();
575        assert_eq!(t.get(h), Some(item));
576        std::fs::remove_file(&p).ok();
577    }
578
579    #[test]
580    fn payload_too_large_at_create() {
581        #[allow(dead_code)] // size_of<Big> is the test signal, not the field
582        struct Big([u8; SLOT_PAYLOAD_BYTES + 1]);
583        impl Copy for Big {}
584        impl Clone for Big { fn clone(&self) -> Self { *self } }
585        let p = tmp("too-large");
586        match SharedHandleTable::<Big>::create(&p, 4) {
587            Err(HandleTableError::PayloadTooLarge) => {}
588            other => panic!("expected PayloadTooLarge, got {:?}", other.as_ref().err()),
589        }
590        std::fs::remove_file(&p).ok();
591    }
592}
subetha_cxc/shared_handle_table.rs

subetha_cxc/
shared_handle_table.rs
