low_latency_data_structures/spsc/
consumer.rs

1use std::marker::PhantomData;
2
3use crate::mem::Allocation;
4use crate::shim::cell::Cell;
5use crate::shim::sync::Arc;
6use crate::shim::sync::atomic::AtomicUsize;
7use crate::spsc::queue::Queue;
8
9#[repr(C, align(128))]
10pub(super) struct ConsumerState {
11    pub head: AtomicUsize,
12    pub cached_tail: Cell<usize>,
13}
14
15/// The popping handle of an SPSC FIFO queue.
16///
17/// Created together with its paired [`Producer`](crate::spsc::Producer) by
18/// [`new`](crate::spsc::new) (or [`new_hugepage_backed`](crate::spsc::new_hugepage_backed)).
19/// `Consumer` is [`Send`] but not [`Sync`]: at most one thread may pop at a
20/// time.
21pub struct Consumer<T, const CAPACITY: usize, AllocT: Allocation<T>> {
22    inner: Arc<Queue<T, CAPACITY, AllocT>>,
23    _not_sync: PhantomData<*const ()>,
24}
25
26impl<T, const CAPACITY: usize, AllocT: Allocation<T>> std::fmt::Debug
27    for Consumer<T, CAPACITY, AllocT>
28{
29    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
30        f.debug_struct("Consumer")
31            .field("capacity", &CAPACITY)
32            .finish_non_exhaustive()
33    }
34}
35
36// SAFETY: Consumer is Send because the underlying Queue is Send when both T
37// and the allocation are Send; PhantomData<*const ()> blocks Sync.
38unsafe impl<T: Send, const CAPACITY: usize, AllocT: Allocation<T> + Send> Send
39    for Consumer<T, CAPACITY, AllocT>
40{
41}
42
43impl<T, const CAPACITY: usize, AllocT: Allocation<T>> Consumer<T, CAPACITY, AllocT> {
44    pub(super) fn new(queue: Arc<Queue<T, CAPACITY, AllocT>>) -> Self {
45        Self {
46            inner: queue,
47            _not_sync: PhantomData,
48        }
49    }
50
51    /// Pops the next item from the queue.
52    ///
53    /// Wait-free. Returns `Some(item)` on success, `None` if the queue is
54    /// currently empty.
55    ///
56    /// # Examples
57    ///
58    /// ```
59    /// use low_latency_data_structures::spsc::new;
60    ///
61    /// let (producer, consumer) = new::<u64, 4>();
62    /// assert_eq!(consumer.pop(), None);
63    /// let _ = producer.push(42);
64    /// assert_eq!(consumer.pop(), Some(42));
65    /// ```
66    #[inline]
67    #[must_use = "ignoring the popped item silently drops it"]
68    pub fn pop(&self) -> Option<T> {
69        self.inner.pop()
70    }
71}
72
73#[cfg(test)]
74mod tests {
75
76    use super::*;
77    use crate::shim::cell::UnsafeCell;
78
79    struct NeverAlloc;
80    impl<T> Allocation<T> for NeverAlloc {
81        fn ptr(&self) -> *mut UnsafeCell<std::mem::MaybeUninit<T>> {
82            unreachable!("it's just a stub")
83        }
84    }
85
86    // Shouldn't be possible to construct Arc<Consumer<T>> and then use it from different threads as it
87    // will break the requirement of *Single* producer *Single* consumer queue.
88    static_assertions::assert_not_impl_any!(Consumer<u32, 0, NeverAlloc>: Sync);
89}
low_latency_data_structures/spsc/consumer.rs

low_latency_data_structures/spsc/
consumer.rs
