strat9_kernel/ipc/
lockfree_ring.rs

1//! Lock-free bounded message queue backed by [`crossbeam_queue::ArrayQueue`].
2//!
3//! Replaces the previous manual atomic-ring implementation with a safe,
4//! well-tested SPSC/MPSC queue.  Keeps the same external API
5//! (`write`/`read`/`try_write`/`try_read`/`has_data`/`has_space`) for
6//! backward compatibility.
7//!
8//! Physical frames are still allocated for `frame_phys_addrs()` (IPC ring
9//! syscall compat).  DMA is not supported : the NIC data plane uses
10//! dedicated hardware ring buffers.
11
12use alloc::{boxed::Box, sync::Arc, vec::Vec};
13use crossbeam_queue::ArrayQueue;
14
15use crate::{
16    memory::{allocate_frame, free_frame, PhysFrame},
17    sync::with_irqs_disabled,
18};
19
20// ---------------------------------------------------------------------------
21// RingError
22// ---------------------------------------------------------------------------
23
24/// Errors returned by ring operations.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum RingError {
27    /// Ring has no free slots.
28    Full,
29    /// Ring contains no messages.
30    Empty,
31    /// Message exceeds slot capacity.
32    MessageTooLarge,
33    /// Provided buffer is too small for the pending message.
34    BufferTooSmall,
35    /// Memory allocation failed during ring creation.
36    AllocFailed,
37    /// Invalid ring parameters (size, capacity, etc.).
38    InvalidParameters,
39}
40
41// ---------------------------------------------------------------------------
42// LockFreeRing
43// ---------------------------------------------------------------------------
44
45/// A bounded lock-free message queue.
46///
47/// Internally uses [`crossbeam_queue::ArrayQueue<Box<[u8]>>`] for safe,
48/// tested message passing.  Also allocates minimal physical frames for
49/// `frame_phys_addrs()` used by the shared-ring IPC syscall.
50#[derive(Debug)]
51pub struct LockFreeRing {
52    queue: ArrayQueue<Box<[u8]>>,
53    slot_limit: usize,
54    frames: Vec<PhysFrame>,
55}
56
57unsafe impl Send for LockFreeRing {}
58unsafe impl Sync for LockFreeRing {}
59
60impl LockFreeRing {
61    /// Default slot count.
62    const DEFAULT_CAPACITY: u32 = 256;
63
64    /// Create a new ring with `slot_count` slots for messages up to
65    /// `slot_size` bytes.
66    pub fn new(slot_count: u32, slot_size: usize) -> Result<Arc<Self>, RingError> {
67        if slot_count == 0 || slot_size == 0 {
68            return Err(RingError::InvalidParameters);
69        }
70        let cap = slot_count.next_power_of_two() as usize;
71        let queue = ArrayQueue::new(cap);
72
73        // Allocate a minimal set of physical frames for IPC ring compat.
74        let mut frames = Vec::new();
75        if let Ok(frame) = with_irqs_disabled(|token| allocate_frame(token)) {
76            frames.push(frame);
77        }
78
79        Ok(Arc::new(LockFreeRing {
80            queue,
81            slot_limit: slot_size,
82            frames,
83        }))
84    }
85
86    /// The number of slots in this ring.
87    #[inline]
88    pub fn capacity(&self) -> u32 {
89        self.queue.capacity() as u32
90    }
91
92    /// The maximum message size in bytes.
93    #[inline]
94    pub fn slot_size(&self) -> u32 {
95        self.slot_limit as u32
96    }
97
98    /// Physical addresses of the backing frames (for IPC ring compat).
99    pub fn frame_phys_addrs(&self) -> Vec<u64> {
100        self.frames
101            .iter()
102            .map(|f| f.start_address.as_u64())
103            .collect()
104    }
105
106    /// Write `data` into the ring.  Returns `Full` if no slot is available,
107    /// or `MessageTooLarge` if the data exceeds the slot capacity.
108    #[inline]
109    pub fn write(&self, data: &[u8]) -> Result<(), RingError> {
110        if data.len() > self.slot_limit {
111            return Err(RingError::MessageTooLarge);
112        }
113        let buf: Box<[u8]> = data.into();
114        self.queue.push(buf).map_err(|_| RingError::Full)
115    }
116
117    /// Read one message from the ring into `buf`.
118    #[inline]
119    pub fn read(&self, buf: &mut [u8]) -> Result<usize, RingError> {
120        let msg = self.queue.pop().ok_or(RingError::Empty)?;
121        if msg.len() > buf.len() {
122            return Err(RingError::BufferTooSmall);
123        }
124        buf[..msg.len()].copy_from_slice(&msg);
125        Ok(msg.len())
126    }
127
128    /// Non-blocking write.
129    #[inline]
130    pub fn try_write(&self, data: &[u8]) -> Result<(), RingError> {
131        self.write(data)
132    }
133
134    /// Write multiple buffers in a single slot (scatter-gather).
135    pub fn write_vectored(&self, bufs: &[&[u8]]) -> Result<(), RingError> {
136        let total: usize = bufs.iter().map(|b| b.len()).sum();
137        if total > self.slot_limit {
138            return Err(RingError::MessageTooLarge);
139        }
140        let mut buf = alloc::vec![0u8; total];
141        let mut offset = 0;
142        for b in bufs {
143            buf[offset..offset + b.len()].copy_from_slice(b);
144            offset += b.len();
145        }
146        let boxed: Box<[u8]> = buf.into_boxed_slice();
147        self.queue.push(boxed).map_err(|_| RingError::Full)
148    }
149
150    /// Non-blocking read.  Returns `Ok(None)` when the ring is empty.
151    #[inline]
152    pub fn try_read(&self, buf: &mut [u8]) -> Result<Option<usize>, RingError> {
153        match self.queue.pop() {
154            Some(msg) => {
155                if msg.len() > buf.len() {
156                    return Err(RingError::BufferTooSmall);
157                }
158                buf[..msg.len()].copy_from_slice(&msg);
159                Ok(Some(msg.len()))
160            }
161            None => Ok(None),
162        }
163    }
164
165    /// Notify the consumer that new data is available.
166    /// No-op under ArrayQueue (notification is handled by the caller).
167    pub fn notify_consumer_raw(&self) {}
168
169    /// Notify the producer that space has been freed.
170    /// No-op under ArrayQueue.
171    pub fn notify_producer_raw(&self) {}
172
173    /// Returns true if the ring contains at least one message.
174    pub fn has_data(&self) -> bool {
175        !self.queue.is_empty()
176    }
177
178    /// Returns true if at least one slot is free.
179    pub fn has_space(&self) -> bool {
180        self.queue.len() < self.queue.capacity()
181    }
182
183    /// DMA buffer descriptor.
184    ///
185    /// ArrayQueue does not expose physical addresses; this method returns
186    /// `None`.  The NIC data plane uses dedicated hardware ring buffers.
187    pub fn dma_buffer(&self, _slot_index: u32) -> Option<DmaBuffer> {
188        None
189    }
190}
191
192impl Drop for LockFreeRing {
193    fn drop(&mut self) {
194        for frame in self.frames.drain(..) {
195            with_irqs_disabled(|token| free_frame(token, frame));
196        }
197    }
198}
199
200// ---------------------------------------------------------------------------
201// DmaBuffer
202// ---------------------------------------------------------------------------
203
204/// Descriptor for a DMA-accessible buffer region.
205///
206/// Note: the ArrayQueue-backed ring does not support DMA; this type is
207/// kept for backward compatibility only.
208#[derive(Debug, Clone, Copy)]
209pub struct DmaBuffer {
210    pub phys_addr: u64,
211    pub virt_addr: *const u8,
212}
213
214unsafe impl Send for DmaBuffer {}
215unsafe impl Sync for DmaBuffer {}
216
217// ---------------------------------------------------------------------------
218// IpcNotification impl
219// ---------------------------------------------------------------------------
220
221use super::transport::{IpcError, IpcNotification};
222
223impl IpcNotification for LockFreeRing {
224    fn notify_consumer(&self) {
225        self.notify_consumer_raw();
226    }
227
228    fn notify_producer(&self) {
229        self.notify_producer_raw();
230    }
231
232    fn wait_notification(&self) -> Result<(), IpcError> {
233        for _ in 0..64 {
234            if self.has_data() {
235                return Ok(());
236            }
237            core::hint::spin_loop();
238        }
239        loop {
240            if self.has_data() {
241                return Ok(());
242            }
243            crate::process::block_current_task();
244        }
245    }
246}
247
248// ---------------------------------------------------------------------------
249// Tests
250// ---------------------------------------------------------------------------
251
252#[cfg(test)]
253mod tests {
254    use super::*;
255
256    #[test]
257    fn ping_pong_64b() {
258        let ring = LockFreeRing::new(64, 256).unwrap();
259        let msg = [0xABu8; 64];
260        ring.write(&msg).unwrap();
261        let mut buf = [0u8; 64];
262        let n = ring.read(&mut buf).unwrap();
263        assert_eq!(n, 64);
264        assert_eq!(buf, [0xAB; 64]);
265    }
266
267    #[test]
268    fn full_ring() {
269        // F9 fix (L2 host harness): capacity is slot_count.next_power_of_two(),
270        // so a 4-slot ring accepts 4 messages before returning Full.
271        let ring = LockFreeRing::new(4, 64).unwrap();
272        for _ in 0..4 {
273            ring.write(b"hello").unwrap();
274        }
275        assert_eq!(ring.write(b"world"), Err(RingError::Full));
276        let mut buf = [0u8; 64];
277        ring.read(&mut buf).unwrap();
278        ring.write(b"world").unwrap();
279    }
280
281    #[test]
282    fn message_too_large() {
283        let ring = LockFreeRing::new(8, 64).unwrap();
284        let oversized = [0u8; 128];
285        assert_eq!(ring.write(&oversized), Err(RingError::MessageTooLarge));
286    }
287
288    #[test]
289    fn buffer_too_small() {
290        let ring = LockFreeRing::new(8, 256).unwrap();
291        ring.write(b"hello world").unwrap();
292        let mut tiny = [0u8; 4];
293        assert_eq!(ring.read(&mut tiny), Err(RingError::BufferTooSmall));
294    }
295
296    #[test]
297    fn empty_ring() {
298        let ring = LockFreeRing::new(8, 64).unwrap();
299        let mut buf = [0u8; 64];
300        assert_eq!(ring.read(&mut buf), Err(RingError::Empty));
301    }
302
303    #[test]
304    fn try_read_empty() {
305        let ring = LockFreeRing::new(8, 64).unwrap();
306        let mut buf = [0u8; 64];
307        assert_eq!(ring.try_read(&mut buf), Ok(None));
308    }
309
310    #[test]
311    fn write_vectored() {
312        let ring = LockFreeRing::new(8, 256).unwrap();
313        ring.write_vectored(&[b"hello", b" ", b"world"]).unwrap();
314        let mut buf = [0u8; 256];
315        let n = ring.read(&mut buf).unwrap();
316        assert_eq!(&buf[..n], b"hello world");
317    }
318}
strat9_kernel/ipc/lockfree_ring.rs

strat9_kernel/ipc/
lockfree_ring.rs
