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

low_latency_data_structures/spsc/
consumer.rs
