subetha_cxc/
shared_versioned_chain.rs

1//! `SharedVersionedChain<T>` - cross-process MVCC linked list.
2//!
3//! Each node holds `(version: u64, value: T)`. Nodes are linked
4//! newest-first via an `AtomicU32` head and per-node `next` offsets.
5//! A reader walking from head sees nodes in descending version
6//! order and can `read_at(snapshot)` to find the newest version
7//! that's <= the snapshot.
8//!
9//! # Layout
10//!
11//! ```text
12//! +-----------------------------+
13//! | ChainHeader (64B)           |
14//! |   - magic                   |
15//! |   - capacity                |
16//! |   - payload_size            |
17//! |   - head: AtomicU32 (idx)   |
18//! |   - free_list_head: u64     |  (counter, idx) packed
19//! |   - live_count: u64         |
20//! +-----------------------------+
21//! | VersionNode[0] (64B)        |
22//! |   - version: AtomicU64      |
23//! |   - next:   AtomicU32       |
24//! |   - next_free: AtomicU32    |
25//! |   - payload: [u8; 48]       |
26//! +-----------------------------+
27//! | VersionNode[1] ...          |
28//! +-----------------------------+
29//! ```
30//!
31//! Same slot-allocator pattern as `SharedHandleTable`: ABA-free
32//! Treiber stack for the free list, atomic CAS for head updates.
33
34use std::fs::{File, OpenOptions};
35use std::marker::PhantomData;
36use std::mem::{align_of, size_of};
37use std::path::Path;
38use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
39
40use memmap2::{MmapMut, MmapOptions};
41
42pub const VERSIONED_CHAIN_MAGIC: u64 = 0x4150_4D46_5643_4E48;
43pub const NODE_PAYLOAD_BYTES: usize = 48;
44pub const NIL_NODE: u32 = u32::MAX;
45
46#[repr(C, align(64))]
47pub struct ChainHeader {
48    pub magic: u64,
49    pub capacity: u32,
50    pub payload_size: u32,
51    pub head: AtomicU32,
52    pub free_list_head: AtomicU64,
53    pub live_count: AtomicU64,
54    _pad: [u8; 32],
55}
56
57#[repr(C, align(64))]
58pub struct VersionNode {
59    pub version: AtomicU64,
60    pub next: AtomicU32,
61    pub next_free: AtomicU32,
62    pub payload: [u8; NODE_PAYLOAD_BYTES],
63}
64
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66pub enum ChainError {
67    LayoutMismatch,
68    PayloadTooLarge,
69    Full,
70    NonMonotonicVersion,
71    IoError(std::io::ErrorKind),
72}
73
74impl From<std::io::Error> for ChainError {
75    fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
76}
77
78pub const fn versioned_chain_file_size(capacity: usize) -> usize {
79    size_of::<ChainHeader>() + capacity * size_of::<VersionNode>()
80}
81
82pub struct SharedVersionedChain<T: Copy + 'static> {
83    _file: File,
84    mmap: MmapMut,
85    capacity: usize,
86    _phantom: PhantomData<T>,
87    header_sidecar: subetha_core::HandshakeHeader,
88    ring_sidecar: Box<subetha_core::ObservationRing>,
89}
90
91unsafe impl<T: Copy + Send + 'static> Send for SharedVersionedChain<T> {}
92unsafe impl<T: Copy + Sync + 'static> Sync for SharedVersionedChain<T> {}
93
94impl<T: Copy + Send + Sync + 'static> subetha_sidecar::AdaptiveInstance for SharedVersionedChain<T> {
95    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
96    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
97    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
98        Box::new(subetha_sidecar::NoMigrationPolicy)
99    }
100}
101
102#[inline]
103fn pack_head(counter: u32, idx: u32) -> u64 {
104    ((counter as u64) << 32) | (idx as u64)
105}
106
107#[inline]
108fn unpack_head(v: u64) -> (u32, u32) {
109    ((v >> 32) as u32, v as u32)
110}
111
112impl<T: Copy + 'static> SharedVersionedChain<T> {
113    pub fn create(path: impl AsRef<Path>, capacity: usize) -> Result<Self, ChainError> {
114        Self::check_layout()?;
115        assert!(capacity >= 1 && capacity < (u32::MAX - 1) as usize);
116        let total = versioned_chain_file_size(capacity);
117        let file = OpenOptions::new()
118            .read(true).write(true).create(true).truncate(true)
119            .open(path.as_ref())?;
120        file.set_len(total as u64)?;
121        let mut mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
122        let hdr = mmap.as_mut_ptr() as *mut ChainHeader;
123        unsafe {
124            std::ptr::write(hdr, ChainHeader {
125                magic: VERSIONED_CHAIN_MAGIC,
126                capacity: capacity as u32,
127                payload_size: size_of::<T>() as u32,
128                head: AtomicU32::new(NIL_NODE),
129                free_list_head: AtomicU64::new(pack_head(0, 0)),
130                live_count: AtomicU64::new(0),
131                _pad: [0; 32],
132            });
133        }
134        let nodes_base = unsafe { mmap.as_mut_ptr().add(size_of::<ChainHeader>()) };
135        for i in 0..capacity {
136            let node_ptr = unsafe {
137                nodes_base.add(i * size_of::<VersionNode>()) as *mut VersionNode
138            };
139            let next_free = if i + 1 < capacity { (i + 1) as u32 } else { NIL_NODE };
140            unsafe {
141                std::ptr::write(node_ptr, VersionNode {
142                    version: AtomicU64::new(0),
143                    next: AtomicU32::new(NIL_NODE),
144                    next_free: AtomicU32::new(next_free),
145                    payload: [0; NODE_PAYLOAD_BYTES],
146                });
147            }
148        }
149        Ok(Self {
150            _file: file, mmap, capacity, _phantom: PhantomData,
151            header_sidecar: subetha_core::HandshakeHeader::new(),
152            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
153        })
154    }
155
156    pub fn open(path: impl AsRef<Path>, expected_capacity: usize) -> Result<Self, ChainError> {
157        Self::check_layout()?;
158        let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
159        let total = versioned_chain_file_size(expected_capacity);
160        if file.metadata()?.len() < total as u64 {
161            return Err(ChainError::LayoutMismatch);
162        }
163        let mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
164        let hdr = unsafe { &*(mmap.as_ptr() as *const ChainHeader) };
165        if hdr.magic != VERSIONED_CHAIN_MAGIC
166            || hdr.capacity != expected_capacity as u32
167            || hdr.payload_size as usize != size_of::<T>()
168        {
169            return Err(ChainError::LayoutMismatch);
170        }
171        Ok(Self {
172            _file: file, mmap, capacity: expected_capacity, _phantom: PhantomData,
173            header_sidecar: subetha_core::HandshakeHeader::new(),
174            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
175        })
176    }
177
178    fn check_layout() -> Result<(), ChainError> {
179        if size_of::<T>() > NODE_PAYLOAD_BYTES {
180            return Err(ChainError::PayloadTooLarge);
181        }
182        if align_of::<T>() > 8 {
183            return Err(ChainError::PayloadTooLarge);
184        }
185        Ok(())
186    }
187
188    pub fn capacity(&self) -> usize { self.capacity }
189
190    /// Reset to empty: head -> NIL, live_count -> 0, free list rebuilt
191    /// to contain all slots. Stale `version` values referring to old
192    /// snapshots become invalid. Not thread-safe with concurrent
193    /// push/read from other threads.
194    pub fn clear(&self) {
195        let header = self.header();
196        header.head.store(NIL_NODE, Ordering::Release);
197        header.live_count.store(0, Ordering::Release);
198        // Rebuild free list: slot 0 is head, slot i links to i+1, last links to NIL.
199        for i in 0..self.capacity {
200            let next_free = if i + 1 < self.capacity { (i + 1) as u32 } else { NIL_NODE };
201            self.node(i as u32).next_free.store(next_free, Ordering::Release);
202            self.node(i as u32).next.store(NIL_NODE, Ordering::Release);
203            self.node(i as u32).version.store(0, Ordering::Release);
204        }
205        header.free_list_head.store(pack_head(0, 0), Ordering::Release);
206    }
207
208    pub fn header(&self) -> &ChainHeader {
209        unsafe { &*(self.mmap.as_ptr() as *const ChainHeader) }
210    }
211
212    fn node(&self, idx: u32) -> &VersionNode {
213        let base = unsafe { self.mmap.as_ptr().add(size_of::<ChainHeader>()) };
214        unsafe { &*(base.add((idx as usize) * size_of::<VersionNode>()) as *const VersionNode) }
215    }
216
217    fn pop_free(&self) -> Option<u32> {
218        let header = self.header();
219        loop {
220            let head = header.free_list_head.load(Ordering::Acquire);
221            let (cnt, idx) = unpack_head(head);
222            if idx == NIL_NODE { return None; }
223            let next = self.node(idx).next_free.load(Ordering::Acquire);
224            let new_head = pack_head(cnt.wrapping_add(1), next);
225            if header.free_list_head.compare_exchange_weak(
226                head, new_head, Ordering::AcqRel, Ordering::Acquire,
227            ).is_ok() {
228                return Some(idx);
229            }
230            std::hint::spin_loop();
231        }
232    }
233
234    /// Push a new version at the head. `version` must be strictly
235    /// greater than the current head's version (MVCC invariant).
236    pub fn push(&self, version: u64, value: T) -> Result<(), ChainError> {
237        let r = self.push_inner(version, value);
238        self.ring_sidecar.push_op(
239            crate::sidecar_ops::versioned::OP_PUSH,
240            if r.is_err() { 1 } else { 0 },
241        );
242        r
243    }
244
245    fn push_inner(&self, version: u64, value: T) -> Result<(), ChainError> {
246        let header = self.header();
247        // Optimistic CAS loop on the head pointer; verify version
248        // monotonicity under contention.
249        loop {
250            let cur_head = header.head.load(Ordering::Acquire);
251            if cur_head != NIL_NODE {
252                let cur_version = self.node(cur_head).version.load(Ordering::Acquire);
253                if version <= cur_version {
254                    return Err(ChainError::NonMonotonicVersion);
255                }
256            }
257            let new_idx = self.pop_free().ok_or(ChainError::Full)?;
258            let new_node = self.node(new_idx);
259            new_node.version.store(version, Ordering::Release);
260            new_node.next.store(cur_head, Ordering::Release);
261            // SAFETY: we just allocated new_idx from the free list,
262            // so we own its payload exclusively until the head CAS.
263            unsafe {
264                let dst = new_node.payload.as_ptr() as *mut T;
265                std::ptr::write_unaligned(dst, value);
266            }
267            // CAS head from cur_head -> new_idx.
268            if header.head.compare_exchange_weak(
269                cur_head, new_idx, Ordering::AcqRel, Ordering::Acquire,
270            ).is_ok() {
271                header.live_count.fetch_add(1, Ordering::AcqRel);
272                return Ok(());
273            }
274            // CAS failed; return the node to the free list and retry.
275            self.push_free(new_idx);
276            std::hint::spin_loop();
277        }
278    }
279
280    fn push_free(&self, idx: u32) {
281        let header = self.header();
282        loop {
283            let head = header.free_list_head.load(Ordering::Acquire);
284            let (cnt, head_idx) = unpack_head(head);
285            self.node(idx).next_free.store(head_idx, Ordering::Release);
286            let new_head = pack_head(cnt.wrapping_add(1), idx);
287            if header.free_list_head.compare_exchange_weak(
288                head, new_head, Ordering::AcqRel, Ordering::Acquire,
289            ).is_ok() {
290                return;
291            }
292            std::hint::spin_loop();
293        }
294    }
295
296    /// Read the value visible at `snapshot_version`. Walks back from
297    /// head through the chain until a node with version <= snapshot
298    /// is found. Returns `None` if no such version exists.
299    pub fn read_at(&self, snapshot_version: u64) -> Option<T> {
300        let header = self.header();
301        let mut cur = header.head.load(Ordering::Acquire);
302        while cur != NIL_NODE {
303            let node = self.node(cur);
304            let v = node.version.load(Ordering::Acquire);
305            if v <= snapshot_version {
306                let value: T = unsafe {
307                    let src = node.payload.as_ptr() as *const T;
308                    std::ptr::read_unaligned(src)
309                };
310                self.ring_sidecar
311                    .push_op(crate::sidecar_ops::versioned::OP_READ_AT, 0);
312                return Some(value);
313            }
314            cur = node.next.load(Ordering::Acquire);
315        }
316        self.ring_sidecar
317            .push_op(crate::sidecar_ops::versioned::OP_READ_AT, 2);
318        None
319    }
320
321    /// Latest (head) version + value, or `None` if empty.
322    pub fn current(&self) -> Option<(u64, T)> {
323        let head = self.header().head.load(Ordering::Acquire);
324        if head == NIL_NODE {
325            self.ring_sidecar
326                .push_op(crate::sidecar_ops::versioned::OP_CURRENT, 2);
327            return None;
328        }
329        let node = self.node(head);
330        let v = node.version.load(Ordering::Acquire);
331        let value: T = unsafe {
332            let src = node.payload.as_ptr() as *const T;
333            std::ptr::read_unaligned(src)
334        };
335        self.ring_sidecar
336            .push_op(crate::sidecar_ops::versioned::OP_CURRENT, 0);
337        Some((v, value))
338    }
339
340    pub fn len(&self) -> usize {
341        self.header().live_count.load(Ordering::Acquire) as usize
342    }
343
344    pub fn is_empty(&self) -> bool { self.len() == 0 }
345
346    pub fn flush(&self) -> Result<(), ChainError> {
347        self.mmap.flush()?;
348        Ok(())
349    }
350
351    /// Non-blocking flush: schedules a writeback via the OS.
352    /// Note: Windows is only partially async (sync to page cache,
353    /// not to disk).
354    pub fn flush_async(&self) -> Result<(), ChainError> {
355        self.mmap.flush_async()?;
356        Ok(())
357    }
358}
359
360#[cfg(test)]
361mod tests {
362    use super::*;
363
364    fn tmp(name: &str) -> std::path::PathBuf {
365        let mut p = std::env::temp_dir();
366        let pid = std::process::id();
367        p.push(format!("subetha-chain-{name}-{pid}.bin"));
368        p
369    }
370
371    #[test]
372    fn push_then_read_at_returns_correct_version() {
373        let p = tmp("push-read");
374        let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 8).unwrap();
375        c.push(1, 10).unwrap();
376        c.push(2, 20).unwrap();
377        c.push(3, 30).unwrap();
378        // Time-travel reads.
379        assert_eq!(c.read_at(0), None);
380        assert_eq!(c.read_at(1), Some(10));
381        assert_eq!(c.read_at(2), Some(20));
382        assert_eq!(c.read_at(3), Some(30));
383        assert_eq!(c.read_at(100), Some(30));
384        assert_eq!(c.current(), Some((3, 30)));
385        assert_eq!(c.len(), 3);
386        std::fs::remove_file(&p).ok();
387    }
388
389    #[test]
390    fn push_rejects_non_monotonic_version() {
391        let p = tmp("non-mono");
392        let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 4).unwrap();
393        c.push(10, 100).unwrap();
394        assert_eq!(c.push(5, 50).unwrap_err(), ChainError::NonMonotonicVersion);
395        assert_eq!(c.push(10, 100).unwrap_err(), ChainError::NonMonotonicVersion);
396        std::fs::remove_file(&p).ok();
397    }
398
399    #[test]
400    fn full_chain_returns_error() {
401        let p = tmp("full");
402        let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 2).unwrap();
403        c.push(1, 10).unwrap();
404        c.push(2, 20).unwrap();
405        assert_eq!(c.push(3, 30).unwrap_err(), ChainError::Full);
406        std::fs::remove_file(&p).ok();
407    }
408
409    #[test]
410    fn cross_handle_visibility() {
411        let p = tmp("cross-handle");
412        let writer: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 8).unwrap();
413        let reader: SharedVersionedChain<u64> = SharedVersionedChain::open(&p, 8).unwrap();
414        writer.push(1, 100).unwrap();
415        writer.push(2, 200).unwrap();
416        assert_eq!(reader.read_at(2), Some(200));
417        assert_eq!(reader.current(), Some((2, 200)));
418        std::fs::remove_file(&p).ok();
419    }
420
421    #[test]
422    fn disk_persistence_survives_reopen() {
423        let p = tmp("disk-persist");
424        {
425            let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 8).unwrap();
426            c.push(10, 1000).unwrap();
427            c.push(20, 2000).unwrap();
428            c.flush().unwrap();
429        }
430        let c2: SharedVersionedChain<u64> = SharedVersionedChain::open(&p, 8).unwrap();
431        assert_eq!(c2.read_at(20), Some(2000));
432        assert_eq!(c2.read_at(10), Some(1000));
433        assert_eq!(c2.len(), 2);
434        std::fs::remove_file(&p).ok();
435    }
436
437    #[test]
438    fn empty_chain_reads_none() {
439        let p = tmp("empty");
440        let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 4).unwrap();
441        assert!(c.is_empty());
442        assert_eq!(c.read_at(0), None);
443        assert_eq!(c.read_at(u64::MAX), None);
444        assert_eq!(c.current(), None);
445        std::fs::remove_file(&p).ok();
446    }
447}
subetha_cxc/shared_versioned_chain.rs

subetha_cxc/
shared_versioned_chain.rs
