subetha_cxc/
frame_region.rs

1//! `FrameRegion` - concurrent fixed-block payload region for the
2//! self-describing offset path shared by every `AdaptiveRing` shape.
3//!
4//! Records too large to inline in a ring slot spill here: the producer
5//! allocates a block, copies the payload in, and writes the block index
6//! into the ring descriptor; the consumer reads the block and frees it.
7//! Because the offset payloads of every shape (SPSC / MPSC / MPMC /
8//! Vyukov) land in one region, the allocator must be safe for many
9//! producers allocating and many consumers freeing at once, in any
10//! order. That is a Treiber-stack free list with an ABA counter plus a
11//! bump high-water mark - the same allocator
12//! [`SharedRegion`](crate::shared_region::SharedRegion) ships, here with
13//! a runtime block size instead of a const-generic `T` so the
14//! `AdaptiveRing` can size its blocks to the workload.
15//!
16//! Reclaim order does not matter: a freed block returns to the stack
17//! and is handed to the next allocation regardless of which consumer
18//! freed it, so no FIFO bookkeeping is needed across consumers.
19
20use std::fs::{File, OpenOptions};
21use std::path::Path;
22use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
23
24use memmap2::{MmapMut, MmapOptions};
25
26use crate::shared_ring::RingError;
27
28/// Magic identifying a `FrameRegion` layout. ASCII "FRGN" + version.
29pub const FRAME_REGION_MAGIC: u64 = 0x4652_474e_0000_0001;
30
31/// Free-list sentinel: "no next block".
32const NIL: u32 = u32::MAX;
33
34/// Smallest block: must hold the 4-byte free-list link.
35pub const MIN_BLOCK_SIZE: usize = 8;
36
37#[inline]
38fn pack(counter: u32, index: u32) -> u64 {
39    ((counter as u64) << 32) | (index as u64)
40}
41#[inline]
42fn unpack(v: u64) -> (u32, u32) {
43    ((v >> 32) as u32, v as u32)
44}
45
46/// Header: metadata line, then the bump cursor and the free-list head
47/// each on their own cache line so allocators and freers do not
48/// false-share.
49#[repr(C, align(64))]
50struct FrameRegionHeader {
51    magic: u64,
52    block_size: u64,
53    block_count: u64,
54    _pad_meta: [u8; 64 - 24],
55    /// Bump high-water mark (next never-yet-allocated block).
56    bump_next: AtomicU32,
57    _pad_bump: [u8; 64 - 4],
58    /// Treiber-stack free-list head, ABA-tagged (`counter << 32 | idx`).
59    free_head: AtomicU64,
60    _pad_free: [u8; 64 - 8],
61}
62
63/// Total mapped bytes for `block_count` blocks of `block_size`.
64pub const fn frame_region_file_size(block_size: usize, block_count: usize) -> usize {
65    std::mem::size_of::<FrameRegionHeader>() + block_size * block_count
66}
67
68#[allow(dead_code)]
69enum RegionBacking {
70    Anon(MmapMut),
71    File(File, MmapMut),
72    Shm(crate::shm_file::ShmFile),
73}
74
75/// Concurrent fixed-block region. Multi-producer `alloc`,
76/// multi-consumer `free`, any-order reclaim.
77pub struct FrameRegion {
78    _backing: RegionBacking,
79    raw_ptr: *mut u8,
80    block_size: usize,
81    block_count: usize,
82    blocks_base: usize,
83}
84
85unsafe impl Send for FrameRegion {}
86unsafe impl Sync for FrameRegion {}
87
88fn validate(block_size: usize, block_count: usize) -> Result<(), RingError> {
89    if block_size < MIN_BLOCK_SIZE || !block_size.is_multiple_of(8) {
90        return Err(RingError::LayoutMismatch);
91    }
92    if block_count < 1 || block_count >= NIL as usize {
93        return Err(RingError::LayoutMismatch);
94    }
95    Ok(())
96}
97
98unsafe fn init_region(ptr: *mut u8, block_size: usize, block_count: usize) {
99    unsafe {
100        std::ptr::write(ptr as *mut FrameRegionHeader, FrameRegionHeader {
101            magic: FRAME_REGION_MAGIC,
102            block_size: block_size as u64,
103            block_count: block_count as u64,
104            _pad_meta: [0; 64 - 24],
105            bump_next: AtomicU32::new(0),
106            _pad_bump: [0; 64 - 4],
107            free_head: AtomicU64::new(pack(0, NIL)),
108            _pad_free: [0; 64 - 8],
109        });
110    }
111}
112
113impl FrameRegion {
114    /// Anonymous in-process region.
115    pub fn create_anon(block_size: usize, block_count: usize) -> Result<Self, RingError> {
116        validate(block_size, block_count)?;
117        let total = frame_region_file_size(block_size, block_count);
118        let mut mmap = MmapOptions::new().len(total).map_anon()?;
119        unsafe { init_region(mmap.as_mut_ptr(), block_size, block_count) };
120        let raw_ptr = mmap.as_mut_ptr();
121        Ok(Self::from_parts(RegionBacking::Anon(mmap), raw_ptr, block_size, block_count))
122    }
123
124    /// File-backed region; cross-process via the page cache.
125    pub fn create(
126        path: impl AsRef<Path>, block_size: usize, block_count: usize,
127    ) -> Result<Self, RingError> {
128        validate(block_size, block_count)?;
129        let total = frame_region_file_size(block_size, block_count);
130        let file = OpenOptions::new()
131            .read(true).write(true).create(true).truncate(true)
132            .open(path.as_ref())?;
133        file.set_len(total as u64)?;
134        let mut mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
135        unsafe { init_region(mmap.as_mut_ptr(), block_size, block_count) };
136        let raw_ptr = mmap.as_mut_ptr();
137        Ok(Self::from_parts(RegionBacking::File(file, mmap), raw_ptr, block_size, block_count))
138    }
139
140    /// Open an existing file-backed region. Validates the header.
141    pub fn open(
142        path: impl AsRef<Path>, block_size: usize, block_count: usize,
143    ) -> Result<Self, RingError> {
144        validate(block_size, block_count)?;
145        let total = frame_region_file_size(block_size, block_count);
146        let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
147        if (file.metadata()?.len() as usize) < total {
148            return Err(RingError::LayoutMismatch);
149        }
150        let mut mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
151        Self::check_header(mmap.as_ptr(), block_size, block_count)?;
152        let raw_ptr = mmap.as_mut_ptr();
153        Ok(Self::from_parts(RegionBacking::File(file, mmap), raw_ptr, block_size, block_count))
154    }
155
156    /// Build a region on a named RAM-resident shared-memory backing.
157    pub fn create_from_shm(
158        mut shm: crate::shm_file::ShmFile, block_size: usize, block_count: usize,
159    ) -> Result<Self, RingError> {
160        validate(block_size, block_count)?;
161        if shm.len() < frame_region_file_size(block_size, block_count) {
162            return Err(RingError::LayoutMismatch);
163        }
164        let raw_ptr = shm.as_mut_slice().as_mut_ptr();
165        unsafe { init_region(raw_ptr, block_size, block_count) };
166        Ok(Self::from_parts(RegionBacking::Shm(shm), raw_ptr, block_size, block_count))
167    }
168
169    /// Open an existing named ShmFs-backed region (no re-init).
170    pub fn open_from_shm(
171        mut shm: crate::shm_file::ShmFile, block_size: usize, block_count: usize,
172    ) -> Result<Self, RingError> {
173        validate(block_size, block_count)?;
174        if shm.len() < frame_region_file_size(block_size, block_count) {
175            return Err(RingError::LayoutMismatch);
176        }
177        let raw_ptr = shm.as_mut_slice().as_mut_ptr();
178        Self::check_header(raw_ptr, block_size, block_count)?;
179        Ok(Self::from_parts(RegionBacking::Shm(shm), raw_ptr, block_size, block_count))
180    }
181
182    fn from_parts(
183        backing: RegionBacking, raw_ptr: *mut u8, block_size: usize, block_count: usize,
184    ) -> Self {
185        Self {
186            _backing: backing, raw_ptr, block_size, block_count,
187            blocks_base: std::mem::size_of::<FrameRegionHeader>(),
188        }
189    }
190
191    fn check_header(ptr: *const u8, block_size: usize, block_count: usize) -> Result<(), RingError> {
192        let h = unsafe { &*(ptr as *const FrameRegionHeader) };
193        if h.magic != FRAME_REGION_MAGIC
194            || h.block_size != block_size as u64
195            || h.block_count != block_count as u64
196        {
197            return Err(RingError::LayoutMismatch);
198        }
199        Ok(())
200    }
201
202    /// Largest payload a block holds.
203    pub fn block_size(&self) -> usize { self.block_size }
204    /// Number of blocks.
205    pub fn block_count(&self) -> usize { self.block_count }
206
207    fn header(&self) -> &FrameRegionHeader {
208        unsafe { &*(self.raw_ptr as *const FrameRegionHeader) }
209    }
210
211    fn block_ptr(&self, idx: u32) -> *mut u8 {
212        unsafe { self.raw_ptr.add(self.blocks_base + idx as usize * self.block_size) }
213    }
214
215    /// The block's first 4 bytes reinterpreted as the free-list link
216    /// (only meaningful while the block is free).
217    fn next_link(&self, idx: u32) -> &AtomicU32 {
218        unsafe { &*(self.block_ptr(idx) as *const AtomicU32) }
219    }
220
221    /// Allocate a block. Free list first, then bump. `None` when full.
222    pub fn alloc(&self) -> Option<u32> {
223        loop {
224            let head = self.header().free_head.load(Ordering::Acquire);
225            let (counter, idx) = unpack(head);
226            if idx == NIL {
227                break;
228            }
229            let next = self.next_link(idx).load(Ordering::Acquire);
230            let new_head = pack(counter.wrapping_add(1), next);
231            if self.header().free_head.compare_exchange(
232                head, new_head, Ordering::AcqRel, Ordering::Acquire,
233            ).is_ok() {
234                return Some(idx);
235            }
236        }
237        let idx = self.header().bump_next.fetch_add(1, Ordering::AcqRel);
238        if idx >= self.block_count as u32 {
239            self.header().bump_next.fetch_sub(1, Ordering::AcqRel);
240            return None;
241        }
242        Some(idx)
243    }
244
245    /// Return a block to the free list. Any consumer may free any block.
246    pub fn free(&self, idx: u32) {
247        if idx as usize >= self.block_count {
248            return;
249        }
250        loop {
251            let head = self.header().free_head.load(Ordering::Acquire);
252            let (counter, old_top) = unpack(head);
253            self.next_link(idx).store(old_top, Ordering::Release);
254            let new_head = pack(counter.wrapping_add(1), idx);
255            if self.header().free_head.compare_exchange(
256                head, new_head, Ordering::AcqRel, Ordering::Acquire,
257            ).is_ok() {
258                return;
259            }
260        }
261    }
262
263    /// Copy `payload` into block `idx`. Caller guarantees
264    /// `payload.len() <= block_size`.
265    pub fn write_block(&self, idx: u32, payload: &[u8]) {
266        debug_assert!(payload.len() <= self.block_size);
267        unsafe {
268            std::ptr::copy_nonoverlapping(
269                payload.as_ptr(), self.block_ptr(idx), payload.len(),
270            );
271        }
272    }
273
274    /// Copy `len` bytes out of block `idx` into `out` (appended).
275    pub fn read_block_into(&self, idx: u32, len: usize, out: &mut Vec<u8>) {
276        debug_assert!(len <= self.block_size);
277        out.reserve(len);
278        unsafe {
279            std::ptr::copy_nonoverlapping(
280                self.block_ptr(idx),
281                out.spare_capacity_mut().as_mut_ptr() as *mut u8,
282                len,
283            );
284            let new_len = out.len() + len;
285            out.set_len(new_len);
286        }
287    }
288}
289
290#[cfg(test)]
291mod tests {
292    use super::*;
293    use std::sync::Arc;
294    use std::sync::atomic::AtomicUsize;
295    use std::thread;
296
297    #[test]
298    fn alloc_write_read_free_cycle() {
299        let r = FrameRegion::create_anon(256, 8).unwrap();
300        let idx = r.alloc().unwrap();
301        let payload = vec![0xABu8; 200];
302        r.write_block(idx, &payload);
303        let mut out = Vec::new();
304        r.read_block_into(idx, 200, &mut out);
305        assert_eq!(out, payload);
306        r.free(idx);
307        // Freed block is reused by the next alloc.
308        let idx2 = r.alloc().unwrap();
309        assert_eq!(idx2, idx, "freed block returns to the stack");
310    }
311
312    #[test]
313    fn exhausts_then_full() {
314        let r = FrameRegion::create_anon(64, 4).unwrap();
315        let a: Vec<u32> = (0..4).map(|_| r.alloc().unwrap()).collect();
316        assert_eq!(a.len(), 4);
317        assert!(r.alloc().is_none(), "region full");
318        r.free(a[1]);
319        assert!(r.alloc().is_some(), "freeing reopens a block");
320    }
321
322    #[test]
323    fn concurrent_alloc_free_no_double_issue() {
324        // Many threads alloc + free in a loop; assert no index is ever
325        // held by two threads at once (a double-issue would corrupt).
326        let r = Arc::new(FrameRegion::create_anon(64, 64).unwrap());
327        let held: Arc<Vec<AtomicUsize>> =
328            Arc::new((0..64).map(|_| AtomicUsize::new(0)).collect());
329        let mut handles = Vec::new();
330        for _ in 0..8 {
331            let r = r.clone();
332            let held = held.clone();
333            handles.push(thread::spawn(move || {
334                for _ in 0..20_000 {
335                    if let Some(idx) = r.alloc() {
336                        let prev = held[idx as usize].fetch_add(1, Ordering::AcqRel);
337                        assert_eq!(prev, 0, "block {idx} double-issued");
338                        held[idx as usize].fetch_sub(1, Ordering::AcqRel);
339                        r.free(idx);
340                    }
341                }
342            }));
343        }
344        for h in handles {
345            h.join().unwrap();
346        }
347    }
348
349    #[test]
350    fn shm_cross_handle() {
351        use crate::shm_file::ShmFile;
352        let nonce = std::time::SystemTime::now()
353            .duration_since(std::time::UNIX_EPOCH)
354            .map(|d| d.as_nanos())
355            .unwrap_or(0);
356        let name = format!("frame_region_{}_{}", std::process::id(), nonce);
357        let (bs, bc) = (256usize, 8usize);
358        let size = frame_region_file_size(bs, bc);
359        let a = FrameRegion::create_from_shm(
360            ShmFile::create_or_open_named(&name, size).unwrap(), bs, bc).unwrap();
361        let b = FrameRegion::open_from_shm(
362            ShmFile::create_or_open_named(&name, size).unwrap(), bs, bc).unwrap();
363        let idx = a.alloc().unwrap();
364        a.write_block(idx, b"shared across handles");
365        let mut out = Vec::new();
366        b.read_block_into(idx, 21, &mut out);
367        assert_eq!(out, b"shared across handles");
368    }
369
370    #[test]
371    fn rejects_bad_params() {
372        assert!(matches!(FrameRegion::create_anon(7, 8), Err(RingError::LayoutMismatch)));
373        assert!(matches!(FrameRegion::create_anon(64, 0), Err(RingError::LayoutMismatch)));
374    }
375}
subetha_cxc/frame_region.rs

subetha_cxc/
frame_region.rs
