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    /// Obtain the chain at `path`, initializing an empty one if the
114    /// path does not yet exist and attaching to it if it does.
115    /// Attaching leaves live nodes, the head and the free list in
116    /// place; a chain built with a different capacity or payload type
117    /// is a `LayoutMismatch`. [`reset`](Self::reset) reinitializes.
118    pub fn create(path: impl AsRef<Path>, capacity: usize) -> Result<Self, ChainError> {
119        Self::check_layout()?;
120        assert!(capacity >= 1 && capacity < (u32::MAX - 1) as usize);
121        let (file, mmap) = crate::mmf_attach::create_or_attach(
122            path.as_ref(),
123            versioned_chain_file_size(capacity),
124            |ptr| unsafe { Self::init_region(ptr, capacity) },
125            |ptr| unsafe { (*(ptr as *const ChainHeader)).magic == VERSIONED_CHAIN_MAGIC },
126        )?;
127        Self::from_region(file, mmap, capacity)
128    }
129
130    /// Truncate the chain at `path` and initialize an empty one,
131    /// discarding every node live peers hold. For a caller that knows
132    /// it owns the path.
133    pub fn reset(path: impl AsRef<Path>, capacity: usize) -> Result<Self, ChainError> {
134        Self::check_layout()?;
135        assert!(capacity >= 1 && capacity < (u32::MAX - 1) as usize);
136        let (file, mmap) = crate::mmf_attach::reset(
137            path.as_ref(),
138            versioned_chain_file_size(capacity),
139            |ptr| unsafe { Self::init_region(ptr, capacity) },
140        )?;
141        Self::from_region(file, mmap, capacity)
142    }
143
144    /// Lay out an empty chain: config, the NIL chain head, the free
145    /// chain threading every node, then the magic, last, because
146    /// attachers spin on it. The zeroed free-list head is already
147    /// (counter=0, idx=0) and zeroed node versions and payloads are
148    /// the empty state.
149    ///
150    /// # Safety
151    /// `ptr` addresses at least `versioned_chain_file_size(capacity)`
152    /// writable zeroed bytes.
153    unsafe fn init_region(ptr: *mut u8, capacity: usize) {
154        unsafe {
155            let hdr = ptr as *mut ChainHeader;
156            (*hdr).capacity = capacity as u32;
157            (*hdr).payload_size = size_of::<T>() as u32;
158            std::ptr::write(&raw mut (*hdr).head, AtomicU32::new(NIL_NODE));
159            let nodes_base = ptr.add(size_of::<ChainHeader>());
160            for i in 0..capacity {
161                let node_ptr = nodes_base.add(i * size_of::<VersionNode>()) as *mut VersionNode;
162                let next_free = if i + 1 < capacity { (i + 1) as u32 } else { NIL_NODE };
163                std::ptr::write(&raw mut (*node_ptr).next, AtomicU32::new(NIL_NODE));
164                std::ptr::write(&raw mut (*node_ptr).next_free, AtomicU32::new(next_free));
165            }
166            std::ptr::write_volatile(&raw mut (*hdr).magic, VERSIONED_CHAIN_MAGIC);
167        }
168    }
169
170    /// Wrap an initialized chain, refusing one built with a different
171    /// capacity or payload type.
172    fn from_region(file: File, mmap: MmapMut, capacity: usize) -> Result<Self, ChainError> {
173        let hdr = unsafe { &*(mmap.as_ptr() as *const ChainHeader) };
174        if hdr.magic != VERSIONED_CHAIN_MAGIC
175            || hdr.capacity != capacity as u32
176            || hdr.payload_size as usize != size_of::<T>()
177        {
178            return Err(ChainError::LayoutMismatch);
179        }
180        Ok(Self {
181            _file: file, mmap, capacity, _phantom: PhantomData,
182            header_sidecar: subetha_core::HandshakeHeader::new(),
183            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
184        })
185    }
186
187    pub fn open(path: impl AsRef<Path>, expected_capacity: usize) -> Result<Self, ChainError> {
188        Self::check_layout()?;
189        let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
190        let total = versioned_chain_file_size(expected_capacity);
191        if file.metadata()?.len() < total as u64 {
192            return Err(ChainError::LayoutMismatch);
193        }
194        let mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
195        Self::from_region(file, mmap, expected_capacity)
196    }
197
198    fn check_layout() -> Result<(), ChainError> {
199        if size_of::<T>() > NODE_PAYLOAD_BYTES {
200            return Err(ChainError::PayloadTooLarge);
201        }
202        if align_of::<T>() > 8 {
203            return Err(ChainError::PayloadTooLarge);
204        }
205        Ok(())
206    }
207
208    pub fn capacity(&self) -> usize { self.capacity }
209
210    /// Reset to empty: head -> NIL, live_count -> 0, free list rebuilt
211    /// to contain all slots. Stale `version` values referring to old
212    /// snapshots become invalid. Not thread-safe with concurrent
213    /// push/read from other threads.
214    pub fn clear(&self) {
215        let header = self.header();
216        header.head.store(NIL_NODE, Ordering::Release);
217        header.live_count.store(0, Ordering::Release);
218        // Rebuild free list: slot 0 is head, slot i links to i+1, last links to NIL.
219        for i in 0..self.capacity {
220            let next_free = if i + 1 < self.capacity { (i + 1) as u32 } else { NIL_NODE };
221            self.node(i as u32).next_free.store(next_free, Ordering::Release);
222            self.node(i as u32).next.store(NIL_NODE, Ordering::Release);
223            self.node(i as u32).version.store(0, Ordering::Release);
224        }
225        header.free_list_head.store(pack_head(0, 0), Ordering::Release);
226    }
227
228    pub fn header(&self) -> &ChainHeader {
229        unsafe { &*(self.mmap.as_ptr() as *const ChainHeader) }
230    }
231
232    fn node(&self, idx: u32) -> &VersionNode {
233        let base = unsafe { self.mmap.as_ptr().add(size_of::<ChainHeader>()) };
234        unsafe { &*(base.add((idx as usize) * size_of::<VersionNode>()) as *const VersionNode) }
235    }
236
237    fn pop_free(&self) -> Option<u32> {
238        let header = self.header();
239        loop {
240            let head = header.free_list_head.load(Ordering::Acquire);
241            let (cnt, idx) = unpack_head(head);
242            if idx == NIL_NODE { return None; }
243            let next = self.node(idx).next_free.load(Ordering::Acquire);
244            let new_head = pack_head(cnt.wrapping_add(1), next);
245            if header.free_list_head.compare_exchange_weak(
246                head, new_head, Ordering::AcqRel, Ordering::Acquire,
247            ).is_ok() {
248                return Some(idx);
249            }
250            std::hint::spin_loop();
251        }
252    }
253
254    /// Push a new version at the head. `version` must be strictly
255    /// greater than the current head's version (MVCC invariant).
256    pub fn push(&self, version: u64, value: T) -> Result<(), ChainError> {
257        let r = self.push_inner(version, value);
258        self.ring_sidecar.push_op(
259            crate::sidecar_ops::versioned::OP_PUSH,
260            if r.is_err() { 1 } else { 0 },
261        );
262        r
263    }
264
265    fn push_inner(&self, version: u64, value: T) -> Result<(), ChainError> {
266        let header = self.header();
267        // Optimistic CAS loop on the head pointer; verify version
268        // monotonicity under contention.
269        loop {
270            let cur_head = header.head.load(Ordering::Acquire);
271            if cur_head != NIL_NODE {
272                let cur_version = self.node(cur_head).version.load(Ordering::Acquire);
273                if version <= cur_version {
274                    return Err(ChainError::NonMonotonicVersion);
275                }
276            }
277            let new_idx = self.pop_free().ok_or(ChainError::Full)?;
278            let new_node = self.node(new_idx);
279            new_node.version.store(version, Ordering::Release);
280            new_node.next.store(cur_head, Ordering::Release);
281            // SAFETY: we just allocated new_idx from the free list,
282            // so we own its payload exclusively until the head CAS.
283            unsafe {
284                let dst = new_node.payload.as_ptr() as *mut T;
285                std::ptr::write_unaligned(dst, value);
286            }
287            // CAS head from cur_head -> new_idx.
288            if header.head.compare_exchange_weak(
289                cur_head, new_idx, Ordering::AcqRel, Ordering::Acquire,
290            ).is_ok() {
291                header.live_count.fetch_add(1, Ordering::AcqRel);
292                return Ok(());
293            }
294            // CAS failed; return the node to the free list and retry.
295            self.push_free(new_idx);
296            std::hint::spin_loop();
297        }
298    }
299
300    fn push_free(&self, idx: u32) {
301        let header = self.header();
302        loop {
303            let head = header.free_list_head.load(Ordering::Acquire);
304            let (cnt, head_idx) = unpack_head(head);
305            self.node(idx).next_free.store(head_idx, Ordering::Release);
306            let new_head = pack_head(cnt.wrapping_add(1), idx);
307            if header.free_list_head.compare_exchange_weak(
308                head, new_head, Ordering::AcqRel, Ordering::Acquire,
309            ).is_ok() {
310                return;
311            }
312            std::hint::spin_loop();
313        }
314    }
315
316    /// Read the value visible at `snapshot_version`. Walks back from
317    /// head through the chain until a node with version <= snapshot
318    /// is found. Returns `None` if no such version exists.
319    pub fn read_at(&self, snapshot_version: u64) -> Option<T> {
320        let header = self.header();
321        let mut cur = header.head.load(Ordering::Acquire);
322        while cur != NIL_NODE {
323            let node = self.node(cur);
324            let v = node.version.load(Ordering::Acquire);
325            if v <= snapshot_version {
326                let value: T = unsafe {
327                    let src = node.payload.as_ptr() as *const T;
328                    std::ptr::read_unaligned(src)
329                };
330                self.ring_sidecar
331                    .push_op(crate::sidecar_ops::versioned::OP_READ_AT, 0);
332                return Some(value);
333            }
334            cur = node.next.load(Ordering::Acquire);
335        }
336        self.ring_sidecar
337            .push_op(crate::sidecar_ops::versioned::OP_READ_AT, 2);
338        None
339    }
340
341    /// Latest (head) version + value, or `None` if empty.
342    pub fn current(&self) -> Option<(u64, T)> {
343        let head = self.header().head.load(Ordering::Acquire);
344        if head == NIL_NODE {
345            self.ring_sidecar
346                .push_op(crate::sidecar_ops::versioned::OP_CURRENT, 2);
347            return None;
348        }
349        let node = self.node(head);
350        let v = node.version.load(Ordering::Acquire);
351        let value: T = unsafe {
352            let src = node.payload.as_ptr() as *const T;
353            std::ptr::read_unaligned(src)
354        };
355        self.ring_sidecar
356            .push_op(crate::sidecar_ops::versioned::OP_CURRENT, 0);
357        Some((v, value))
358    }
359
360    pub fn len(&self) -> usize {
361        self.header().live_count.load(Ordering::Acquire) as usize
362    }
363
364    pub fn is_empty(&self) -> bool { self.len() == 0 }
365
366    pub fn flush(&self) -> Result<(), ChainError> {
367        self.mmap.flush()?;
368        Ok(())
369    }
370
371    /// Non-blocking flush: schedules a writeback via the OS.
372    /// Note: Windows is only partially async (sync to page cache,
373    /// not to disk).
374    pub fn flush_async(&self) -> Result<(), ChainError> {
375        self.mmap.flush_async()?;
376        Ok(())
377    }
378}
379
380#[cfg(test)]
381mod tests {
382    use super::*;
383
384    fn tmp(name: &str) -> std::path::PathBuf {
385        let mut p = std::env::temp_dir();
386        let pid = std::process::id();
387        p.push(format!("subetha-chain-{name}-{pid}.bin"));
388        p
389    }
390
391    #[test]
392    fn push_then_read_at_returns_correct_version() {
393        let p = tmp("push-read");
394        let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 8).unwrap();
395        c.push(1, 10).unwrap();
396        c.push(2, 20).unwrap();
397        c.push(3, 30).unwrap();
398        // Time-travel reads.
399        assert_eq!(c.read_at(0), None);
400        assert_eq!(c.read_at(1), Some(10));
401        assert_eq!(c.read_at(2), Some(20));
402        assert_eq!(c.read_at(3), Some(30));
403        assert_eq!(c.read_at(100), Some(30));
404        assert_eq!(c.current(), Some((3, 30)));
405        assert_eq!(c.len(), 3);
406        std::fs::remove_file(&p).ok();
407    }
408
409    /// A second create attaches with live nodes in place; reset is
410    /// what strips them.
411    #[test]
412    fn second_create_attaches_and_keeps_nodes() {
413        let p = tmp("attach");
414        std::fs::remove_file(&p).ok();
415        let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 8).unwrap();
416        c.push(1, 10).unwrap();
417        c.push(2, 20).unwrap();
418
419        let c2: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 8).unwrap();
420        assert_eq!(c2.current(), Some((2, 20)), "attach lost the chain head");
421        assert_eq!(c2.read_at(1), Some(10), "attach lost a chained node");
422        assert_eq!(c2.len(), 2);
423        assert!(matches!(
424            SharedVersionedChain::<u64>::create(&p, 4),
425            Err(ChainError::LayoutMismatch),
426        ));
427        assert!(matches!(
428            SharedVersionedChain::<u32>::create(&p, 8),
429            Err(ChainError::LayoutMismatch),
430        ));
431
432        // Windows refuses to truncate a mapped file, so every handle
433        // goes before the reset.
434        drop(c);
435        drop(c2);
436        let fresh: SharedVersionedChain<u64> = SharedVersionedChain::reset(&p, 8).unwrap();
437        assert_eq!(fresh.current(), None, "reset kept the chain head");
438        assert_eq!(fresh.len(), 0, "reset kept nodes");
439        fresh.push(1, 11).unwrap();
440        assert_eq!(fresh.current(), Some((1, 11)));
441        drop(fresh);
442        std::fs::remove_file(&p).ok();
443    }
444
445    #[test]
446    fn push_rejects_non_monotonic_version() {
447        let p = tmp("non-mono");
448        let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 4).unwrap();
449        c.push(10, 100).unwrap();
450        assert_eq!(c.push(5, 50).unwrap_err(), ChainError::NonMonotonicVersion);
451        assert_eq!(c.push(10, 100).unwrap_err(), ChainError::NonMonotonicVersion);
452        std::fs::remove_file(&p).ok();
453    }
454
455    #[test]
456    fn full_chain_returns_error() {
457        let p = tmp("full");
458        let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 2).unwrap();
459        c.push(1, 10).unwrap();
460        c.push(2, 20).unwrap();
461        assert_eq!(c.push(3, 30).unwrap_err(), ChainError::Full);
462        std::fs::remove_file(&p).ok();
463    }
464
465    #[test]
466    fn cross_handle_visibility() {
467        let p = tmp("cross-handle");
468        let writer: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 8).unwrap();
469        let reader: SharedVersionedChain<u64> = SharedVersionedChain::open(&p, 8).unwrap();
470        writer.push(1, 100).unwrap();
471        writer.push(2, 200).unwrap();
472        assert_eq!(reader.read_at(2), Some(200));
473        assert_eq!(reader.current(), Some((2, 200)));
474        std::fs::remove_file(&p).ok();
475    }
476
477    #[test]
478    fn disk_persistence_survives_reopen() {
479        let p = tmp("disk-persist");
480        {
481            let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 8).unwrap();
482            c.push(10, 1000).unwrap();
483            c.push(20, 2000).unwrap();
484            c.flush().unwrap();
485        }
486        let c2: SharedVersionedChain<u64> = SharedVersionedChain::open(&p, 8).unwrap();
487        assert_eq!(c2.read_at(20), Some(2000));
488        assert_eq!(c2.read_at(10), Some(1000));
489        assert_eq!(c2.len(), 2);
490        std::fs::remove_file(&p).ok();
491    }
492
493    #[test]
494    fn empty_chain_reads_none() {
495        let p = tmp("empty");
496        let c: SharedVersionedChain<u64> = SharedVersionedChain::create(&p, 4).unwrap();
497        assert!(c.is_empty());
498        assert_eq!(c.read_at(0), None);
499        assert_eq!(c.read_at(u64::MAX), None);
500        assert_eq!(c.current(), None);
501        std::fs::remove_file(&p).ok();
502    }
503}
subetha_cxc/shared_versioned_chain.rs

subetha_cxc/
shared_versioned_chain.rs
