rill_core/queues/
spsc.rs

1use std::fmt;
2use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
3
4use super::{OverflowPolicy, QueueError, QueueResult, QueueStatsSnapshot, RtQueueBase};
5use crate::buffer::AtomicCell;
6
7// =============================================================================
8// Main structure
9// =============================================================================
10
11/// A lock-free single-producer single-consumer queue.
12///
13/// Uses atomic operations for real-time safe push/pop without blocking.
14/// The capacity must be a power of two for efficient mask-based indexing.
15#[repr(C, align(64))]
16pub struct SpscQueue<T: Copy, const CAP: usize> {
17    /// Ring buffer of atomic cells holding queue elements.
18    buffer: [AtomicCell<T>; CAP],
19    /// Producer index (written by producer, read by consumer).
20    head: AtomicUsize,
21    /// Consumer index (written by consumer, read by producer).
22    tail: AtomicUsize,
23    /// Flag indicating whether the queue is full.
24    full: AtomicBool,
25    /// Bitmask for wrapping (CAP - 1, requires CAP to be a power of two).
26    mask: usize,
27    /// Behaviour when a push would overflow the queue.
28    overflow_policy: OverflowPolicy,
29    /// Default value returned when popping from an empty queue.
30    default_value: Option<T>,
31}
32
33impl<T: Copy + Default, const CAP: usize> SpscQueue<T, CAP> {
34    /// Create a new SPSC queue with default policies.
35    ///
36    /// The overflow policy defaults to [`OverflowPolicy::OverwriteOldest`].
37    ///
38    /// # Panics
39    /// Panics if `CAP` is not a power of two.
40    pub fn new() -> Self {
41        assert!(CAP.is_power_of_two(), "CAP must be a power of two");
42
43        let buffer = std::array::from_fn(|_| AtomicCell::new(T::default()));
44
45        Self {
46            buffer,
47            head: AtomicUsize::new(0),
48            tail: AtomicUsize::new(0),
49            full: AtomicBool::new(false),
50            mask: CAP - 1,
51            overflow_policy: OverflowPolicy::OverwriteOldest,
52            default_value: None,
53        }
54    }
55
56    /// Create a queue with custom overflow policy and default value.
57    pub fn with_policies(overflow_policy: OverflowPolicy, default_value: Option<T>) -> Self {
58        let mut queue = Self::new();
59        queue.overflow_policy = overflow_policy;
60        queue.default_value = default_value;
61        queue
62    }
63
64    /// Push a value into the queue.
65    ///
66    /// If the queue is full, behaviour depends on [`OverflowPolicy`].
67    ///
68    /// # Errors
69    /// Returns `QueueFull` when the policy is [`OverflowPolicy::DropNewest`]
70    /// or [`OverflowPolicy::Block`] and the queue is full.
71    ///
72    /// # Panics
73    /// Panics when the policy is [`OverflowPolicy::Panic`] and the queue is full.
74    pub fn push(&self, value: T) -> QueueResult<()> {
75        let head = self.head.load(Ordering::Relaxed);
76        let next_head = (head + 1) & self.mask;
77
78        if self.full.load(Ordering::Acquire) {
79            match self.overflow_policy {
80                OverflowPolicy::OverwriteOldest => {
81                    self.tail.store(
82                        (self.tail.load(Ordering::Relaxed) + 1) & self.mask,
83                        Ordering::Release,
84                    );
85                    self.full.store(false, Ordering::Release);
86                }
87
88                OverflowPolicy::DropNewest => {
89                    return Err(QueueError::QueueFull);
90                }
91
92                OverflowPolicy::Panic => {
93                    panic!("SpscQueue overflow (capacity: {})", CAP);
94                }
95
96                OverflowPolicy::Block => {
97                    return Err(QueueError::QueueFull);
98                }
99            }
100        }
101
102        self.buffer[head].store(value);
103
104        self.head.store(next_head, Ordering::Release);
105
106        if next_head == self.tail.load(Ordering::Acquire) {
107            self.full.store(true, Ordering::Release);
108        }
109
110        Ok(())
111    }
112
113    /// Pop a value from the queue, or return the default value if empty.
114    pub fn pop(&self) -> Option<T> {
115        if self.is_empty() {
116            return self.default_value;
117        }
118
119        let tail = self.tail.load(Ordering::Relaxed) & self.mask;
120        let value = self.buffer[tail].load();
121
122        let next_tail = (tail + 1) & self.mask;
123        self.tail.store(next_tail, Ordering::Release);
124
125        self.full.store(false, Ordering::Release);
126
127        Some(value)
128    }
129
130    /// Peek at the front value without removing it.
131    pub fn peek(&self) -> Option<T> {
132        if self.is_empty() {
133            None
134        } else {
135            let tail = self.tail.load(Ordering::Acquire);
136            Some(self.buffer[tail].load())
137        }
138    }
139
140    /// Return the current number of elements in the queue.
141    pub fn len(&self) -> usize {
142        if self.full.load(Ordering::Acquire) {
143            CAP
144        } else {
145            let head = self.head.load(Ordering::Acquire);
146            let tail = self.tail.load(Ordering::Acquire);
147
148            if head >= tail {
149                head - tail
150            } else {
151                CAP - tail + head
152            }
153        }
154    }
155
156    /// Return the fixed capacity of the queue.
157    pub const fn capacity(&self) -> usize {
158        CAP
159    }
160
161    /// Return true if the queue is empty.
162    pub fn is_empty(&self) -> bool {
163        !self.full.load(Ordering::Acquire)
164            && self.head.load(Ordering::Acquire) == self.tail.load(Ordering::Acquire)
165    }
166
167    /// Return true if the queue is full.
168    pub fn is_full(&self) -> bool {
169        self.full.load(Ordering::Acquire)
170    }
171
172    /// Clear the queue, resetting both head and tail pointers.
173    pub fn clear(&self) {
174        self.head.store(0, Ordering::Relaxed);
175        self.tail.store(0, Ordering::Relaxed);
176        self.full.store(false, Ordering::Relaxed);
177    }
178
179    /// Return a statistics snapshot (currently always empty).
180    pub fn stats(&self) -> QueueStatsSnapshot {
181        QueueStatsSnapshot::default()
182    }
183
184    /// Set the default value returned when popping from an empty queue.
185    pub fn set_default(&mut self, value: T) {
186        self.default_value = Some(value);
187    }
188
189    /// Return the current overflow policy.
190    pub fn overflow_policy(&self) -> OverflowPolicy {
191        self.overflow_policy
192    }
193
194    /// Set the overflow policy.
195    pub fn set_overflow_policy(&mut self, policy: OverflowPolicy) {
196        self.overflow_policy = policy;
197    }
198}
199
200impl<T: Copy + Default + Send + Sync, const CAP: usize> RtQueueBase<T> for SpscQueue<T, CAP> {
201    fn push(&self, value: T) -> QueueResult<()> {
202        self.push(value)
203    }
204
205    fn pop(&self) -> Option<T> {
206        self.pop()
207    }
208
209    fn len(&self) -> usize {
210        self.len()
211    }
212
213    fn capacity(&self) -> usize {
214        CAP
215    }
216
217    fn clear(&self) {
218        self.clear();
219    }
220}
221
222impl<T: Copy + Default + fmt::Debug, const CAP: usize> fmt::Debug for SpscQueue<T, CAP> {
223    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
224        f.debug_struct("SpscQueue")
225            .field("head", &self.head.load(Ordering::Relaxed))
226            .field("tail", &self.tail.load(Ordering::Relaxed))
227            .field("capacity", &CAP)
228            .field("len", &self.len())
229            .field("overflow_policy", &self.overflow_policy)
230            .field("default_value", &self.default_value)
231            .finish()
232    }
233}
234
235impl<T: Copy + Default, const CAP: usize> Default for SpscQueue<T, CAP> {
236    fn default() -> Self {
237        Self::new()
238    }
239}
240#[allow(unsafe_code)]
241unsafe impl<T: Copy + Send, const CAP: usize> Send for SpscQueue<T, CAP> {}
242#[allow(unsafe_code)]
243unsafe impl<T: Copy + Sync, const CAP: usize> Sync for SpscQueue<T, CAP> {}
244
245// =============================================================================
246// Tests
247// =============================================================================
248
249#[cfg(test)]
250mod tests {
251    use super::*;
252
253    #[test]
254    fn test_spsc_basic() {
255        let queue = SpscQueue::<i32, 4>::new();
256
257        assert!(queue.is_empty());
258        assert_eq!(queue.capacity(), 4);
259        assert_eq!(queue.len(), 0);
260
261        queue.push(1).unwrap();
262        assert_eq!(queue.len(), 1);
263        assert!(!queue.is_empty());
264        assert!(!queue.is_full()); // Not full after 1 element
265
266        queue.push(2).unwrap();
267        queue.push(3).unwrap();
268        queue.push(4).unwrap();
269
270        assert!(queue.is_full()); // Full after 4 elements
271        assert_eq!(queue.len(), 4);
272
273        assert_eq!(queue.pop(), Some(1));
274        assert_eq!(queue.pop(), Some(2));
275        assert_eq!(queue.pop(), Some(3));
276        assert_eq!(queue.pop(), Some(4));
277        assert_eq!(queue.pop(), None);
278        assert!(queue.is_empty());
279    }
280
281    #[test]
282    fn test_spsc_overwrite_policy() {
283        let queue = SpscQueue::<i32, 2>::new(); // default policy is OverwriteOldest
284
285        queue.push(1).unwrap();
286        queue.push(2).unwrap();
287        assert!(queue.is_full());
288
289        // Overwrite the oldest (1)
290        queue.push(3).unwrap();
291        assert_eq!(queue.len(), 2);
292
293        // Now the queue is [2, 3] (2 became the oldest)
294        assert_eq!(queue.pop(), Some(2));
295        assert_eq!(queue.pop(), Some(3));
296        assert_eq!(queue.pop(), None);
297    }
298
299    #[test]
300    fn test_spsc_drop_newest_policy() {
301        let queue = SpscQueue::<i32, 2>::with_policies(OverflowPolicy::DropNewest, None);
302
303        queue.push(1).unwrap();
304        queue.push(2).unwrap();
305        assert!(queue.is_full());
306
307        // Should return an error, element is not added
308        assert!(queue.push(3).is_err());
309
310        // Queue should contain [1, 2] in the same order
311        assert_eq!(queue.pop(), Some(1));
312        assert_eq!(queue.pop(), Some(2));
313        assert_eq!(queue.pop(), None);
314    }
315
316    #[test]
317    fn test_spsc_wraparound() {
318        let queue = SpscQueue::<i32, 4>::new();
319
320        // Fill
321        queue.push(0).unwrap();
322        queue.push(1).unwrap();
323        queue.push(2).unwrap();
324        queue.push(3).unwrap();
325        assert!(queue.is_full());
326
327        // Pop two
328        assert_eq!(queue.pop(), Some(0));
329        assert_eq!(queue.pop(), Some(1));
330
331        // Push two new ones
332        queue.push(4).unwrap();
333        queue.push(5).unwrap();
334        assert!(queue.is_full());
335
336        // Verify order
337        assert_eq!(queue.pop(), Some(2));
338        assert_eq!(queue.pop(), Some(3));
339        assert_eq!(queue.pop(), Some(4));
340        assert_eq!(queue.pop(), Some(5));
341        assert_eq!(queue.pop(), None);
342    }
343
344    #[test]
345    fn test_spsc_peek() {
346        let queue = SpscQueue::<i32, 4>::new();
347
348        assert_eq!(queue.peek(), None);
349
350        queue.push(42).unwrap();
351        assert_eq!(queue.peek(), Some(42));
352        assert_eq!(queue.len(), 1);
353        assert_eq!(queue.pop(), Some(42));
354        assert_eq!(queue.peek(), None);
355    }
356
357    #[test]
358    fn test_spsc_clear() {
359        let queue = SpscQueue::<i32, 4>::new();
360
361        queue.push(1).unwrap();
362        queue.push(2).unwrap();
363        queue.push(3).unwrap();
364
365        assert_eq!(queue.len(), 3);
366
367        queue.clear();
368        assert_eq!(queue.len(), 0);
369        assert!(queue.is_empty());
370    }
371
372    #[test]
373    fn test_spsc_default_value() {
374        let queue = SpscQueue::<i32, 4>::with_policies(OverflowPolicy::OverwriteOldest, Some(-1));
375
376        assert_eq!(queue.pop(), Some(-1));
377
378        queue.push(42).unwrap();
379        assert_eq!(queue.pop(), Some(42));
380        assert_eq!(queue.pop(), Some(-1));
381    }
382
383    #[test]
384    fn test_spsc_policy_change() {
385        let mut queue = SpscQueue::<i32, 2>::new();
386        assert_eq!(queue.overflow_policy(), OverflowPolicy::OverwriteOldest);
387
388        queue.set_overflow_policy(OverflowPolicy::DropNewest);
389        assert_eq!(queue.overflow_policy(), OverflowPolicy::DropNewest);
390    }
391
392    #[test]
393    #[should_panic(expected = "CAP must be a power of two")]
394    fn test_spsc_invalid_capacity() {
395        let _ = SpscQueue::<i32, 3>::new();
396    }
397}
rill_core/queues/spsc.rs

rill_core/queues/
spsc.rs
