subetha_cxc/
shared_time_point.rs

1//! `SharedTimePointTile<T>` - cross-process BSPA + Versioned tile
2//! with AVX2 SIMD snapshot-isolation scan.
3//!
4//! Direct lift of the in-process TimePointTile to MMF. The SIMD
5//! code is unchanged because AVX2 instructions operate on memory
6//! addresses identically whether the address is stack-local or
7//! memory-mapped. Cross-process safety comes from atomic insert
8//! (CAS on the occupied bitmap) and atomic version writes; the
9//! SIMD scan is a pure read (no synchronization needed since
10//! version writes are AtomicU64 with Release semantics).
11//!
12//! # Layout
13//!
14//! ```text
15//! +-----------------------------+
16//! | TileHeader (64B)            |
17//! |   - magic                   |
18//! |   - capacity (always 16)    |
19//! |   - payload_size            |
20//! |   - occupied: AtomicU32     |
21//! +-----------------------------+
22//! | VersionedSlot[0] (64B)      |
23//! |   - version: AtomicU64      |
24//! |   - payload: [u8; 56]       |
25//! +-----------------------------+
26//! | ... 15 more slots           |
27//! +-----------------------------+
28//! ```
29
30use std::fs::{File, OpenOptions};
31use std::marker::PhantomData;
32use std::mem::{align_of, size_of};
33use std::path::Path;
34use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
35
36use memmap2::{MmapMut, MmapOptions};
37
38pub const TIME_POINT_MAGIC: u64 = 0x4150_4D46_5450_5054;
39pub const TILE_CAP: usize = 16;
40pub const SLOT_PAYLOAD: usize = 56;
41
42#[repr(C, align(64))]
43pub struct TileHeader {
44    pub magic: u64,
45    pub capacity: u32,
46    pub payload_size: u32,
47    pub occupied: AtomicU32,
48    _pad: [u8; 44],
49}
50
51#[repr(C, align(64))]
52pub struct VersionedSlot {
53    pub version: AtomicU64,
54    pub payload: [u8; SLOT_PAYLOAD],
55}
56
57pub const fn tile_file_size() -> usize {
58    size_of::<TileHeader>() + TILE_CAP * size_of::<VersionedSlot>()
59}
60
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub enum TileError {
63    LayoutMismatch,
64    PayloadTooLarge,
65    Full,
66    IoError(std::io::ErrorKind),
67}
68
69impl From<std::io::Error> for TileError {
70    fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
71}
72
73pub struct SharedTimePointTile<T: Copy + 'static> {
74    _file: File,
75    mmap: MmapMut,
76    _phantom: PhantomData<T>,
77    header_sidecar: subetha_core::HandshakeHeader,
78    ring_sidecar: Box<subetha_core::ObservationRing>,
79}
80
81unsafe impl<T: Copy + Send + 'static> Send for SharedTimePointTile<T> {}
82unsafe impl<T: Copy + Sync + 'static> Sync for SharedTimePointTile<T> {}
83
84impl<T: Copy + Send + Sync + 'static> subetha_sidecar::AdaptiveInstance for SharedTimePointTile<T> {
85    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
86    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
87    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
88        Box::new(subetha_sidecar::NoMigrationPolicy)
89    }
90}
91
92impl<T: Copy + 'static> SharedTimePointTile<T> {
93    pub fn create(path: impl AsRef<Path>) -> Result<Self, TileError> {
94        Self::check_layout()?;
95        let total = tile_file_size();
96        let file = OpenOptions::new()
97            .read(true).write(true).create(true).truncate(true)
98            .open(path.as_ref())?;
99        file.set_len(total as u64)?;
100        let mut mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
101        let hdr = mmap.as_mut_ptr() as *mut TileHeader;
102        unsafe {
103            std::ptr::write(hdr, TileHeader {
104                magic: TIME_POINT_MAGIC,
105                capacity: TILE_CAP as u32,
106                payload_size: size_of::<T>() as u32,
107                occupied: AtomicU32::new(0),
108                _pad: [0; 44],
109            });
110        }
111        let slots_base = unsafe { mmap.as_mut_ptr().add(size_of::<TileHeader>()) };
112        for i in 0..TILE_CAP {
113            let slot_ptr = unsafe {
114                slots_base.add(i * size_of::<VersionedSlot>()) as *mut VersionedSlot
115            };
116            unsafe {
117                std::ptr::write(slot_ptr, VersionedSlot {
118                    version: AtomicU64::new(0),
119                    payload: [0; SLOT_PAYLOAD],
120                });
121            }
122        }
123        Ok(Self {
124            _file: file, mmap, _phantom: PhantomData,
125            header_sidecar: subetha_core::HandshakeHeader::new(),
126            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
127        })
128    }
129
130    pub fn open(path: impl AsRef<Path>) -> Result<Self, TileError> {
131        Self::check_layout()?;
132        let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
133        if file.metadata()?.len() < tile_file_size() as u64 {
134            return Err(TileError::LayoutMismatch);
135        }
136        let mmap = unsafe { MmapOptions::new().len(tile_file_size()).map_mut(&file)? };
137        let hdr = unsafe { &*(mmap.as_ptr() as *const TileHeader) };
138        if hdr.magic != TIME_POINT_MAGIC
139            || hdr.capacity != TILE_CAP as u32
140            || hdr.payload_size as usize != size_of::<T>()
141        {
142            return Err(TileError::LayoutMismatch);
143        }
144        Ok(Self {
145            _file: file, mmap, _phantom: PhantomData,
146            header_sidecar: subetha_core::HandshakeHeader::new(),
147            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
148        })
149    }
150
151    fn check_layout() -> Result<(), TileError> {
152        if size_of::<T>() > SLOT_PAYLOAD {
153            return Err(TileError::PayloadTooLarge);
154        }
155        if align_of::<T>() > 8 {
156            return Err(TileError::PayloadTooLarge);
157        }
158        Ok(())
159    }
160
161    pub fn header(&self) -> &TileHeader {
162        unsafe { &*(self.mmap.as_ptr() as *const TileHeader) }
163    }
164
165    fn slot(&self, idx: usize) -> &VersionedSlot {
166        let base = unsafe { self.mmap.as_ptr().add(size_of::<TileHeader>()) };
167        unsafe {
168            &*(base.add(idx * size_of::<VersionedSlot>()) as *const VersionedSlot)
169        }
170    }
171
172    /// Atomic insert via CAS on the occupied bitmap. Returns the
173    /// claimed lane index, or `Err(Full)`.
174    pub fn insert(&self, version: u64, value: T) -> Result<usize, TileError> {
175        let header = self.header();
176        loop {
177            let cur = header.occupied.load(Ordering::Acquire);
178            let free = !cur & ((1u32 << TILE_CAP) - 1);
179            if free == 0 {
180                self.ring_sidecar
181                    .push_op(crate::sidecar_ops::versioned::OP_PUSH, 1);
182                return Err(TileError::Full);
183            }
184            let lane = free.trailing_zeros() as usize;
185            let new_occupied = cur | (1u32 << lane);
186            if header.occupied.compare_exchange_weak(
187                cur, new_occupied, Ordering::AcqRel, Ordering::Acquire,
188            ).is_ok() {
189                let slot = self.slot(lane);
190                // SAFETY: lane is now exclusively ours (CAS won).
191                unsafe {
192                    let dst = slot.payload.as_ptr() as *mut T;
193                    std::ptr::write_unaligned(dst, value);
194                }
195                slot.version.store(version, Ordering::Release);
196                self.ring_sidecar
197                    .push_op(crate::sidecar_ops::versioned::OP_PUSH, 0);
198                return Ok(lane);
199            }
200            std::hint::spin_loop();
201        }
202    }
203
204    pub fn remove(&self, lane: usize) {
205        if lane < TILE_CAP {
206            self.header().occupied.fetch_and(!(1u32 << lane), Ordering::AcqRel);
207        }
208    }
209
210    pub fn len(&self) -> usize {
211        self.header().occupied.load(Ordering::Acquire).count_ones() as usize
212    }
213
214    pub fn is_empty(&self) -> bool { self.len() == 0 }
215    pub fn is_full(&self) -> bool {
216        self.header().occupied.load(Ordering::Acquire) == ((1u32 << TILE_CAP) - 1)
217    }
218
219    /// SIMD scan: return a 16-bit lane mask of entries with version
220    /// <= snapshot AND currently occupied. AVX2 uses the unsigned-
221    /// compare-via-sign-XOR trick because cmpgt_epi64 is signed.
222    #[inline]
223    pub fn visible_mask(&self, snapshot: u64) -> u16 {
224        let header = self.header();
225        let occupied = header.occupied.load(Ordering::Acquire) as u16;
226        self.ring_sidecar.push_op(
227            crate::sidecar_ops::versioned::OP_VISIBLE_MASK,
228            if occupied == 0 { 2 } else { 0 },
229        );
230        if occupied == 0 { return 0; }
231        let versions_base = unsafe {
232            self.mmap.as_ptr().add(size_of::<TileHeader>())
233        };
234        // The versions are at offset 0 of each VersionedSlot. We
235        // need a contiguous u64 array of versions for the SIMD load;
236        // since slots are 64-byte aligned and versions are at slot
237        // offset 0, a naive gather is needed. For simplicity we
238        // copy into a stack buffer; the bench shows this is still
239        // very fast for 16 entries.
240        let mut versions = [0u64; TILE_CAP];
241        for (i, v) in versions.iter_mut().enumerate() {
242            let slot = unsafe {
243                &*(versions_base.add(i * size_of::<VersionedSlot>()) as *const VersionedSlot)
244            };
245            *v = slot.version.load(Ordering::Acquire);
246        }
247        Self::simd_visible_mask(&versions, snapshot) & occupied
248    }
249
250    /// SIMD visibility scan dispatcher. Picks AVX-512F (one ZMM
251    /// per 8-lane half + mask-producing `_mm512_cmple_epu64_mask`)
252    /// when present, AVX2 (4 YMM compares with sign-bit XOR trick)
253    /// otherwise, scalar on non-x86 or feature-stripped builds.
254    #[inline]
255    pub fn simd_visible_mask(versions: &[u64; TILE_CAP], snapshot: u64) -> u16 {
256        #[cfg(target_arch = "x86_64")]
257        {
258            if std::is_x86_feature_detected!("avx512f") {
259                // SAFETY: AVX-512F runtime-detected.
260                return unsafe { Self::simd_visible_mask_avx512(versions, snapshot) };
261            }
262            if std::is_x86_feature_detected!("avx2") {
263                // SAFETY: AVX2 runtime-detected.
264                return unsafe { Self::simd_visible_mask_avx2(versions, snapshot) };
265            }
266        }
267        Self::simd_visible_mask_scalar(versions, snapshot)
268    }
269
270    /// AVX-512F path: TILE_CAP=16 covered by two 8-u64 chunks. Each
271    /// chunk uses one `_mm512_loadu_si512` and one
272    /// `_mm512_cmple_epu64_mask` (returns `__mmask8` directly - no
273    /// sign-bit XOR trick needed because the instruction is unsigned
274    /// natively). Two 8-bit masks pack into the 16-bit result via
275    /// `low | (high << 8)`.
276    ///
277    /// # Safety
278    /// Caller must ensure AVX-512F is available.
279    #[cfg(target_arch = "x86_64")]
280    #[target_feature(enable = "avx512f")]
281    pub unsafe fn simd_visible_mask_avx512(
282        versions: &[u64; TILE_CAP],
283        snapshot: u64,
284    ) -> u16 {
285        use std::arch::x86_64::*;
286        let snap = _mm512_set1_epi64(snapshot as i64);
287        // SAFETY: versions is 16 contiguous u64s; two 8-u64 loads at
288        // offsets 0 and 8 cover the full tile.
289        let v_lo = unsafe {
290            _mm512_loadu_si512(versions.as_ptr() as *const __m512i)
291        };
292        let v_hi = unsafe {
293            _mm512_loadu_si512(versions.as_ptr().add(8) as *const __m512i)
294        };
295        let mask_lo: u8 = _mm512_cmple_epu64_mask(v_lo, snap);
296        let mask_hi: u8 = _mm512_cmple_epu64_mask(v_hi, snap);
297        (mask_lo as u16) | ((mask_hi as u16) << 8)
298    }
299
300    /// AVX2 path: 4 chunks of 4 u64. Signed `cmpgt_epi64` plus
301    /// sign-bit XOR delivers the unsigned `<=` predicate; `cmpeq`
302    /// handles the equality boundary.
303    ///
304    /// # Safety
305    /// Caller must ensure AVX2 is available.
306    #[cfg(target_arch = "x86_64")]
307    #[target_feature(enable = "avx2")]
308    pub unsafe fn simd_visible_mask_avx2(
309        versions: &[u64; TILE_CAP],
310        snapshot: u64,
311    ) -> u16 {
312        use std::arch::x86_64::*;
313        let sign_bit = _mm256_set1_epi64x(i64::MIN);
314        let snap_raw = _mm256_set1_epi64x(snapshot as i64);
315        let snap_s = _mm256_xor_si256(snap_raw, sign_bit);
316        // SAFETY: versions is 16 contiguous u64s; four 4-u64 loads at
317        // offsets 0, 4, 8, 12 cover the full tile.
318        let load_xord = |off: usize| -> __m256i {
319            let raw = unsafe {
320                _mm256_loadu_si256(versions.as_ptr().add(off) as *const __m256i)
321            };
322            _mm256_xor_si256(raw, sign_bit)
323        };
324        let load_raw = |off: usize| -> __m256i {
325            unsafe {
326                _mm256_loadu_si256(versions.as_ptr().add(off) as *const __m256i)
327            }
328        };
329        let v0 = load_xord(0);
330        let v1 = load_xord(4);
331        let v2 = load_xord(8);
332        let v3 = load_xord(12);
333        let raw0 = load_raw(0);
334        let raw1 = load_raw(4);
335        let raw2 = load_raw(8);
336        let raw3 = load_raw(12);
337        let gt0 = _mm256_cmpgt_epi64(snap_s, v0);
338        let gt1 = _mm256_cmpgt_epi64(snap_s, v1);
339        let gt2 = _mm256_cmpgt_epi64(snap_s, v2);
340        let gt3 = _mm256_cmpgt_epi64(snap_s, v3);
341        let eq0 = _mm256_cmpeq_epi64(snap_raw, raw0);
342        let eq1 = _mm256_cmpeq_epi64(snap_raw, raw1);
343        let eq2 = _mm256_cmpeq_epi64(snap_raw, raw2);
344        let eq3 = _mm256_cmpeq_epi64(snap_raw, raw3);
345        let m0 = _mm256_or_si256(gt0, eq0);
346        let m1 = _mm256_or_si256(gt1, eq1);
347        let m2 = _mm256_or_si256(gt2, eq2);
348        let m3 = _mm256_or_si256(gt3, eq3);
349        let bits0 = _mm256_movemask_pd(_mm256_castsi256_pd(m0)) as u16;
350        let bits1 = _mm256_movemask_pd(_mm256_castsi256_pd(m1)) as u16;
351        let bits2 = _mm256_movemask_pd(_mm256_castsi256_pd(m2)) as u16;
352        let bits3 = _mm256_movemask_pd(_mm256_castsi256_pd(m3)) as u16;
353        bits0 | (bits1 << 4) | (bits2 << 8) | (bits3 << 12)
354    }
355
356    /// Scalar reference: always available, used as the fallback for
357    /// non-x86 builds and as the ground-truth oracle in tests.
358    #[inline]
359    pub fn simd_visible_mask_scalar(versions: &[u64; TILE_CAP], snapshot: u64) -> u16 {
360        let mut mask = 0u16;
361        for (i, v) in versions.iter().enumerate() {
362            if *v <= snapshot {
363                mask |= 1u16 << i;
364            }
365        }
366        mask
367    }
368
369    /// Read the payload at `lane` if occupied.
370    pub fn at(&self, lane: usize) -> Option<(u64, T)> {
371        if lane >= TILE_CAP {
372            self.ring_sidecar
373                .push_op(crate::sidecar_ops::versioned::OP_READ_AT, 2);
374            return None;
375        }
376        let occ = self.header().occupied.load(Ordering::Acquire);
377        if (occ >> lane) & 1 == 0 {
378            self.ring_sidecar
379                .push_op(crate::sidecar_ops::versioned::OP_READ_AT, 2);
380            return None;
381        }
382        let slot = self.slot(lane);
383        let v = slot.version.load(Ordering::Acquire);
384        let value: T = unsafe {
385            let src = slot.payload.as_ptr() as *const T;
386            std::ptr::read_unaligned(src)
387        };
388        self.ring_sidecar
389            .push_op(crate::sidecar_ops::versioned::OP_READ_AT, 0);
390        Some((v, value))
391    }
392
393    /// Count visible at `snapshot`.
394    pub fn visible_count(&self, snapshot: u64) -> u32 {
395        self.visible_mask(snapshot).count_ones()
396    }
397
398    pub fn flush(&self) -> Result<(), TileError> {
399        self.mmap.flush()?;
400        Ok(())
401    }
402
403    /// Non-blocking flush: schedules a writeback via the OS.
404    /// Note: Windows is only partially async (sync to page cache,
405    /// not to disk).
406    pub fn flush_async(&self) -> Result<(), TileError> {
407        self.mmap.flush_async()?;
408        Ok(())
409    }
410}
411
412#[cfg(test)]
413mod tests {
414    use super::*;
415
416    fn tmp(name: &str) -> std::path::PathBuf {
417        let mut p = std::env::temp_dir();
418        let pid = std::process::id();
419        p.push(format!("subetha-tile-{name}-{pid}.bin"));
420        p
421    }
422
423    #[test]
424    fn empty_tile_visible_mask_is_zero() {
425        let p = tmp("empty");
426        let t: SharedTimePointTile<u64> = SharedTimePointTile::create(&p).unwrap();
427        assert_eq!(t.visible_mask(u64::MAX), 0);
428        assert!(t.is_empty());
429        std::fs::remove_file(&p).ok();
430    }
431
432    #[test]
433    fn insert_then_visible_at_snapshot() {
434        let p = tmp("visible");
435        let t: SharedTimePointTile<u64> = SharedTimePointTile::create(&p).unwrap();
436        t.insert(10, 100).unwrap();
437        t.insert(20, 200).unwrap();
438        t.insert(30, 300).unwrap();
439        // Snapshot 25 sees lanes 0+1 (versions 10, 20).
440        let m = t.visible_mask(25);
441        assert_eq!(m, 0b011);
442        assert_eq!(t.visible_count(25), 2);
443        // Snapshot u64::MAX sees all three.
444        let m_all = t.visible_mask(u64::MAX);
445        assert_eq!(m_all.count_ones(), 3);
446        std::fs::remove_file(&p).ok();
447    }
448
449    #[test]
450    fn fill_to_capacity_then_overflow() {
451        let p = tmp("fill");
452        let t: SharedTimePointTile<u64> = SharedTimePointTile::create(&p).unwrap();
453        for i in 0..TILE_CAP as u64 {
454            t.insert(i, i * 10).unwrap();
455        }
456        assert!(t.is_full());
457        assert_eq!(t.insert(99, 999).unwrap_err(), TileError::Full);
458        std::fs::remove_file(&p).ok();
459    }
460
461    #[test]
462    fn remove_frees_lane_for_reinsert() {
463        let p = tmp("remove");
464        let t: SharedTimePointTile<u64> = SharedTimePointTile::create(&p).unwrap();
465        let l0 = t.insert(1, 100).unwrap();
466        let l1 = t.insert(2, 200).unwrap();
467        t.remove(l0);
468        assert_eq!(t.len(), 1);
469        let l2 = t.insert(3, 300).unwrap();
470        assert_eq!(l2, l0, "freed lane reused");
471        assert_eq!(t.at(l1), Some((2, 200)));
472        assert_eq!(t.at(l2), Some((3, 300)));
473        std::fs::remove_file(&p).ok();
474    }
475
476    #[test]
477    fn empty_lanes_dont_match_zero_snapshot() {
478        let p = tmp("zero-snap");
479        let t: SharedTimePointTile<u64> = SharedTimePointTile::create(&p).unwrap();
480        t.insert(0, 100).unwrap();
481        // Snapshot 0: only the occupied lane with version 0 matches.
482        let m = t.visible_mask(0);
483        assert_eq!(m, 0b1);
484        assert_eq!(m.count_ones(), 1);
485        std::fs::remove_file(&p).ok();
486    }
487
488    #[test]
489    fn simd_matches_scalar_for_boundary_snapshots() {
490        let p = tmp("simd-vs-scalar");
491        let t: SharedTimePointTile<u64> = SharedTimePointTile::create(&p).unwrap();
492        let versions = [5u64, 10, 15, 20, 25, 30, 35, 40,
493                        45, 50, 55, 60, 65, 70, 75, 80];
494        for &v in versions.iter() {
495            t.insert(v, v * 100).unwrap();
496        }
497        for snap in [0u64, 10, 35, 80, 100, u64::MAX] {
498            let simd = t.visible_mask(snap);
499            let mut scalar = 0u16;
500            for (i, &v) in versions.iter().enumerate() {
501                if v <= snap { scalar |= 1 << i; }
502            }
503            assert_eq!(simd, scalar, "snap={snap}: simd={simd:#b} scalar={scalar:#b}");
504        }
505        std::fs::remove_file(&p).ok();
506    }
507
508    #[test]
509    fn cross_handle_inserts_visible() {
510        let p = tmp("cross-handle");
511        let writer: SharedTimePointTile<u64> = SharedTimePointTile::create(&p).unwrap();
512        let reader: SharedTimePointTile<u64> = SharedTimePointTile::open(&p).unwrap();
513        writer.insert(10, 100).unwrap();
514        writer.insert(20, 200).unwrap();
515        assert_eq!(reader.visible_count(u64::MAX), 2);
516        let m = reader.visible_mask(15);
517        assert_eq!(m.count_ones(), 1);
518        std::fs::remove_file(&p).ok();
519    }
520
521    #[test]
522    fn disk_persistence_survives_reopen() {
523        let p = tmp("disk-persist");
524        {
525            let t: SharedTimePointTile<u64> = SharedTimePointTile::create(&p).unwrap();
526            t.insert(7, 70).unwrap();
527            t.insert(8, 80).unwrap();
528            t.flush().unwrap();
529        }
530        let t2: SharedTimePointTile<u64> = SharedTimePointTile::open(&p).unwrap();
531        assert_eq!(t2.len(), 2);
532        assert_eq!(t2.visible_count(u64::MAX), 2);
533        assert_eq!(t2.at(0), Some((7, 70)));
534        assert_eq!(t2.at(1), Some((8, 80)));
535        std::fs::remove_file(&p).ok();
536    }
537}
subetha_cxc/shared_time_point.rs

subetha_cxc/
shared_time_point.rs
