subetha_cxc/
shared_atomic.rs

1//! `SharedAtomic<T>` - cross-process atomic counter / flag.
2//!
3//! Backed by an MMF cell whose payload is interpreted directly as
4//! `AtomicU8 / AtomicU16 / AtomicU32 / AtomicU64`. The atomic ops
5//! are cross-process safe on every modern CPU because hardware
6//! cache coherence guarantees the atomic semantics across address
7//! spaces; the only requirement is that both processes map the
8//! same physical page (which the OS guarantees when they open the
9//! same MMF file).
10//!
11//! Three concrete types:
12//! - `SharedAtomicU32`
13//! - `SharedAtomicU64`
14//! - `SharedAtomicBool` (one byte, but enforced bool semantics)
15//!
16//! Type-erased layout: header + payload region the size of the
17//! native atomic, aligned naturally.
18
19use std::fs::{File, OpenOptions};
20use std::path::Path;
21use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
22
23use memmap2::{MmapMut, MmapOptions};
24
25pub const ATOMIC_MAGIC: u32 = 0x4150_5443;
26
27#[repr(C, align(64))]
28struct AtomicHeader {
29    magic: u32,
30    width: u32,  // 1, 4, or 8 bytes
31    payload_u64: AtomicU64,  // also covers u32, u8 via punning
32}
33
34const ATOMIC_FILE_SIZE: usize = std::mem::size_of::<AtomicHeader>();
35
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub enum SharedAtomicError {
38    LayoutMismatch,
39    IoError(std::io::ErrorKind),
40}
41
42impl From<std::io::Error> for SharedAtomicError {
43    fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
44}
45
46macro_rules! shared_atomic_impl {
47    ($name:ident, $atomic:ty, $native:ty, $width:expr) => {
48        pub struct $name {
49            _file: File,
50            mmap: MmapMut,
51            header_sidecar: subetha_core::HandshakeHeader,
52            ring_sidecar: Box<subetha_core::ObservationRing>,
53        }
54
55        unsafe impl Send for $name {}
56        unsafe impl Sync for $name {}
57
58        impl subetha_sidecar::AdaptiveInstance for $name {
59            fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
60            fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
61            fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
62                Box::new(subetha_sidecar::NoMigrationPolicy)
63            }
64        }
65
66        impl $name {
67            pub fn create(path: impl AsRef<Path>, init: $native) -> Result<Self, SharedAtomicError> {
68                let file = OpenOptions::new()
69                    .read(true).write(true).create(true).truncate(true)
70                    .open(path.as_ref())?;
71                file.set_len(ATOMIC_FILE_SIZE as u64)?;
72                let mut mmap = unsafe { MmapOptions::new().len(ATOMIC_FILE_SIZE).map_mut(&file)? };
73                let hdr = mmap.as_mut_ptr() as *mut AtomicHeader;
74                unsafe {
75                    std::ptr::write(hdr, AtomicHeader {
76                        magic: ATOMIC_MAGIC,
77                        width: $width,
78                        payload_u64: AtomicU64::new(0),
79                    });
80                }
81                let s = Self {
82                    _file: file, mmap,
83                    header_sidecar: subetha_core::HandshakeHeader::new(),
84                    ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
85                };
86                s.atomic().store(init, Ordering::Release);
87                Ok(s)
88            }
89
90            pub fn open(path: impl AsRef<Path>) -> Result<Self, SharedAtomicError> {
91                let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
92                if file.metadata()?.len() < ATOMIC_FILE_SIZE as u64 {
93                    return Err(SharedAtomicError::LayoutMismatch);
94                }
95                let mmap = unsafe { MmapOptions::new().len(ATOMIC_FILE_SIZE).map_mut(&file)? };
96                let hdr = unsafe { &*(mmap.as_ptr() as *const AtomicHeader) };
97                if hdr.magic != ATOMIC_MAGIC || hdr.width != $width {
98                    return Err(SharedAtomicError::LayoutMismatch);
99                }
100                Ok(Self {
101                    _file: file, mmap,
102                    header_sidecar: subetha_core::HandshakeHeader::new(),
103                    ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
104                })
105            }
106
107            #[inline]
108            fn atomic(&self) -> &$atomic {
109                let base = unsafe {
110                    self.mmap.as_ptr()
111                        .add(std::mem::offset_of!(AtomicHeader, payload_u64))
112                };
113                unsafe { &*(base as *const $atomic) }
114            }
115
116            #[inline]
117            pub fn load(&self, ord: Ordering) -> $native {
118                let v = self.atomic().load(ord);
119                self.ring_sidecar
120                    .push_op(crate::sidecar_ops::atomic::OP_LOAD, 0);
121                v
122            }
123
124            #[inline]
125            pub fn store(&self, v: $native, ord: Ordering) {
126                self.atomic().store(v, ord);
127                self.ring_sidecar
128                    .push_op(crate::sidecar_ops::atomic::OP_STORE, 0);
129            }
130
131            #[inline]
132            pub fn fetch_add(&self, v: $native, ord: Ordering) -> $native {
133                let prev = self.atomic().fetch_add(v, ord);
134                self.ring_sidecar
135                    .push_op(crate::sidecar_ops::atomic::OP_FETCH_ADD, 0);
136                prev
137            }
138
139            #[inline]
140            pub fn fetch_sub(&self, v: $native, ord: Ordering) -> $native {
141                let prev = self.atomic().fetch_sub(v, ord);
142                self.ring_sidecar
143                    .push_op(crate::sidecar_ops::atomic::OP_FETCH_ADD, 0);
144                prev
145            }
146
147            #[inline]
148            pub fn fetch_or(&self, v: $native, ord: Ordering) -> $native {
149                let prev = self.atomic().fetch_or(v, ord);
150                self.ring_sidecar
151                    .push_op(crate::sidecar_ops::atomic::OP_FETCH_ADD, 0);
152                prev
153            }
154
155            #[inline]
156            pub fn fetch_and(&self, v: $native, ord: Ordering) -> $native {
157                let prev = self.atomic().fetch_and(v, ord);
158                self.ring_sidecar
159                    .push_op(crate::sidecar_ops::atomic::OP_FETCH_ADD, 0);
160                prev
161            }
162
163            #[inline]
164            pub fn fetch_xor(&self, v: $native, ord: Ordering) -> $native {
165                let prev = self.atomic().fetch_xor(v, ord);
166                self.ring_sidecar
167                    .push_op(crate::sidecar_ops::atomic::OP_FETCH_ADD, 0);
168                prev
169            }
170
171            #[inline]
172            pub fn swap(&self, v: $native, ord: Ordering) -> $native {
173                let prev = self.atomic().swap(v, ord);
174                self.ring_sidecar
175                    .push_op(crate::sidecar_ops::atomic::OP_CAS, 0);
176                prev
177            }
178
179            #[inline]
180            pub fn compare_exchange(
181                &self, current: $native, new: $native,
182                success: Ordering, failure: Ordering,
183            ) -> Result<$native, $native> {
184                let r = self.atomic().compare_exchange(current, new, success, failure);
185                self.ring_sidecar
186                    .push_op(crate::sidecar_ops::atomic::OP_CAS, if r.is_err() { 1 } else { 0 });
187                r
188            }
189
190            pub fn flush(&self) -> Result<(), SharedAtomicError> {
191                self.mmap.flush()?;
192                Ok(())
193            }
194
195            /// Non-blocking flush: schedules a writeback via the OS
196            /// (msync(MS_ASYNC) on Linux; FlushViewOfFile without
197            /// FlushFileBuffers on Windows). Note: Windows is only
198            /// partially async (sync to page cache, not to disk).
199            pub fn flush_async(&self) -> Result<(), SharedAtomicError> {
200                self.mmap.flush_async()?;
201                Ok(())
202            }
203        }
204    };
205}
206
207shared_atomic_impl!(SharedAtomicU32, AtomicU32, u32, 4);
208shared_atomic_impl!(SharedAtomicU64, AtomicU64, u64, 8);
209
210pub struct SharedAtomicBool {
211    _file: File,
212    mmap: MmapMut,
213    header_sidecar: subetha_core::HandshakeHeader,
214    ring_sidecar: Box<subetha_core::ObservationRing>,
215}
216
217unsafe impl Send for SharedAtomicBool {}
218unsafe impl Sync for SharedAtomicBool {}
219
220impl subetha_sidecar::AdaptiveInstance for SharedAtomicBool {
221    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
222    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
223    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
224        Box::new(subetha_sidecar::NoMigrationPolicy)
225    }
226}
227
228impl SharedAtomicBool {
229    pub fn create(path: impl AsRef<Path>, init: bool) -> Result<Self, SharedAtomicError> {
230        let file = OpenOptions::new()
231            .read(true).write(true).create(true).truncate(true)
232            .open(path.as_ref())?;
233        file.set_len(ATOMIC_FILE_SIZE as u64)?;
234        let mut mmap = unsafe { MmapOptions::new().len(ATOMIC_FILE_SIZE).map_mut(&file)? };
235        let hdr = mmap.as_mut_ptr() as *mut AtomicHeader;
236        unsafe {
237            std::ptr::write(hdr, AtomicHeader {
238                magic: ATOMIC_MAGIC,
239                width: 1,
240                payload_u64: AtomicU64::new(0),
241            });
242        }
243        let s = Self {
244            _file: file, mmap,
245            header_sidecar: subetha_core::HandshakeHeader::new(),
246            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
247        };
248        s.atomic().store(init, Ordering::Release);
249        Ok(s)
250    }
251
252    pub fn open(path: impl AsRef<Path>) -> Result<Self, SharedAtomicError> {
253        let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
254        if file.metadata()?.len() < ATOMIC_FILE_SIZE as u64 {
255            return Err(SharedAtomicError::LayoutMismatch);
256        }
257        let mmap = unsafe { MmapOptions::new().len(ATOMIC_FILE_SIZE).map_mut(&file)? };
258        let hdr = unsafe { &*(mmap.as_ptr() as *const AtomicHeader) };
259        if hdr.magic != ATOMIC_MAGIC || hdr.width != 1 {
260            return Err(SharedAtomicError::LayoutMismatch);
261        }
262        Ok(Self {
263            _file: file, mmap,
264            header_sidecar: subetha_core::HandshakeHeader::new(),
265            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
266        })
267    }
268
269    fn atomic(&self) -> &AtomicBool {
270        let base = unsafe {
271            self.mmap.as_ptr().add(std::mem::offset_of!(AtomicHeader, payload_u64))
272        };
273        unsafe { &*(base as *const AtomicBool) }
274    }
275
276    pub fn load(&self, ord: Ordering) -> bool {
277        let v = self.atomic().load(ord);
278        self.ring_sidecar
279            .push_op(crate::sidecar_ops::atomic::OP_LOAD, 0);
280        v
281    }
282    pub fn store(&self, v: bool, ord: Ordering) {
283        self.atomic().store(v, ord);
284        self.ring_sidecar
285            .push_op(crate::sidecar_ops::atomic::OP_STORE, 0);
286    }
287    pub fn swap(&self, v: bool, ord: Ordering) -> bool {
288        let prev = self.atomic().swap(v, ord);
289        self.ring_sidecar
290            .push_op(crate::sidecar_ops::atomic::OP_CAS, 0);
291        prev
292    }
293
294    pub fn flush(&self) -> Result<(), SharedAtomicError> {
295        self.mmap.flush()?;
296        Ok(())
297    }
298
299    /// Non-blocking flush: schedules a writeback via the OS.
300    /// Note: Windows is only partially async (sync to page cache,
301    /// not to disk).
302    pub fn flush_async(&self) -> Result<(), SharedAtomicError> {
303        self.mmap.flush_async()?;
304        Ok(())
305    }
306}
307
308#[cfg(test)]
309mod tests {
310    use super::*;
311
312    fn tmp(name: &str) -> std::path::PathBuf {
313        let mut p = std::env::temp_dir();
314        let pid = std::process::id();
315        p.push(format!("subetha-atomic-{name}-{pid}.bin"));
316        p
317    }
318
319    #[test]
320    fn u64_load_store_round_trip() {
321        let p = tmp("u64-rt");
322        let a = SharedAtomicU64::create(&p, 42).unwrap();
323        assert_eq!(a.load(Ordering::Acquire), 42);
324        a.store(99, Ordering::Release);
325        assert_eq!(a.load(Ordering::Acquire), 99);
326        std::fs::remove_file(&p).ok();
327    }
328
329    #[test]
330    fn u32_fetch_add_increments() {
331        let p = tmp("u32-add");
332        let a = SharedAtomicU32::create(&p, 0).unwrap();
333        for _ in 0..100 { a.fetch_add(1, Ordering::AcqRel); }
334        assert_eq!(a.load(Ordering::Acquire), 100);
335        std::fs::remove_file(&p).ok();
336    }
337
338    #[test]
339    fn cross_handle_visibility() {
340        let p = tmp("cross-handle");
341        let writer = SharedAtomicU64::create(&p, 0).unwrap();
342        let reader = SharedAtomicU64::open(&p).unwrap();
343        writer.store(7777, Ordering::Release);
344        assert_eq!(reader.load(Ordering::Acquire), 7777);
345        std::fs::remove_file(&p).ok();
346    }
347
348    #[test]
349    fn concurrent_fetch_add_sums_correctly() {
350        use std::sync::Arc;
351        use std::thread;
352        let p = tmp("concurrent");
353        let a = Arc::new(SharedAtomicU64::create(&p, 0).unwrap());
354        let mut handles = vec![];
355        for _ in 0..8 {
356            let a = a.clone();
357            handles.push(thread::spawn(move || {
358                for _ in 0..1000 { a.fetch_add(1, Ordering::AcqRel); }
359            }));
360        }
361        for h in handles { h.join().unwrap(); }
362        assert_eq!(a.load(Ordering::Acquire), 8000);
363        std::fs::remove_file(&p).ok();
364    }
365
366    #[test]
367    fn compare_exchange_wins_once() {
368        let p = tmp("cas");
369        let a = SharedAtomicU64::create(&p, 5).unwrap();
370        let r1 = a.compare_exchange(5, 10, Ordering::AcqRel, Ordering::Acquire);
371        let r2 = a.compare_exchange(5, 20, Ordering::AcqRel, Ordering::Acquire);
372        assert_eq!(r1, Ok(5));
373        assert_eq!(r2, Err(10));
374        assert_eq!(a.load(Ordering::Acquire), 10);
375        std::fs::remove_file(&p).ok();
376    }
377
378    #[test]
379    fn bool_load_store_swap() {
380        let p = tmp("bool");
381        let b = SharedAtomicBool::create(&p, false).unwrap();
382        assert!(!b.load(Ordering::Acquire));
383        b.store(true, Ordering::Release);
384        assert!(b.load(Ordering::Acquire));
385        let prev = b.swap(false, Ordering::AcqRel);
386        assert!(prev);
387        assert!(!b.load(Ordering::Acquire));
388        std::fs::remove_file(&p).ok();
389    }
390
391    #[test]
392    fn disk_persistence_survives_reopen() {
393        let p = tmp("disk-persist");
394        {
395            let a = SharedAtomicU64::create(&p, 12345).unwrap();
396            a.flush().unwrap();
397        }
398        let a2 = SharedAtomicU64::open(&p).unwrap();
399        assert_eq!(a2.load(Ordering::Acquire), 12345);
400        std::fs::remove_file(&p).ok();
401    }
402
403    #[test]
404    fn open_rejects_wrong_width() {
405        let p = tmp("wrong-width");
406        let _a = SharedAtomicU64::create(&p, 0).unwrap();
407        match SharedAtomicU32::open(&p) {
408            Err(SharedAtomicError::LayoutMismatch) => {}
409            other => panic!("expected LayoutMismatch, got {:?}", other.as_ref().err()),
410        }
411        std::fs::remove_file(&p).ok();
412    }
413}
subetha_cxc/shared_atomic.rs

subetha_cxc/
shared_atomic.rs
