low_latency_data_structures/spsc/
producer.rs

1use std::marker::PhantomData;
2
3use crossbeam_utils::CachePadded;
4
5use crate::mem::Allocator;
6use crate::shim::cell::Cell;
7use crate::shim::sync::Arc;
8use crate::shim::sync::atomic::AtomicUsize;
9use crate::spsc::queue::Queue;
10
11pub(super) type ProducerState = CachePadded<ProducerStateInner>;
12
13#[derive(Default)]
14pub(super) struct ProducerStateInner {
15    pub tail: AtomicUsize,
16    pub cached_head: Cell<usize>,
17}
18
19/// The pushing handle of an SPSC FIFO queue.
20///
21/// Created together with its paired [`Consumer`](crate::spsc::Consumer) by
22/// [`new`](crate::spsc::new).
23/// `Producer` is [`Send`] but not [`Sync`]: at most one thread may push at a
24/// time.
25pub struct Producer<T, const CAPACITY: usize, A>
26where
27    A: Allocator,
28{
29    inner: Arc<Queue<T, CAPACITY, A>>,
30    _not_sync: PhantomData<*const ()>,
31}
32
33impl<T, const CAPACITY: usize, A> std::fmt::Debug for Producer<T, CAPACITY, A>
34where
35    A: Allocator,
36{
37    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        f.debug_struct("Producer")
39            .field("capacity", &CAPACITY)
40            .finish_non_exhaustive()
41    }
42}
43
44// SAFETY: Producer is Send because the underlying Queue is Send when both T
45// and the allocation are Send; PhantomData<*const ()> blocks Sync.
46unsafe impl<T, const CAPACITY: usize, A> Send for Producer<T, CAPACITY, A>
47where
48    T: Send,
49    A: Allocator,
50    A::Allocation<T>: Send,
51{
52}
53
54impl<T, const CAPACITY: usize, A> Producer<T, CAPACITY, A>
55where
56    A: Allocator,
57{
58    pub(super) fn new(queue: Arc<Queue<T, CAPACITY, A>>) -> Self {
59        Self {
60            inner: queue,
61            _not_sync: PhantomData,
62        }
63    }
64
65    /// Pushes `item` onto the queue.
66    ///
67    /// Wait-free. Never blocks, never allocates. Returns `None` on success.
68    /// If the queue is full the item is returned unchanged as `Some(item)`,
69    /// so the caller can retry or back off without losing data.
70    ///
71    /// # Examples
72    ///
73    /// ```
74    /// use low_latency_data_structures::spsc::{self, new};
75    /// use low_latency_data_structures::mem::global::GlobalAllocator;
76    ///
77    /// let (producer, consumer) = new::<u64, 2, GlobalAllocator>(
78    ///     spsc::Options::global_mlocked(),
79    /// );
80    /// assert_eq!(producer.push(1), None);
81    /// assert_eq!(producer.push(2), None);
82    /// // Queue is full; item is handed back so we can retry later.
83    /// assert_eq!(producer.push(3), Some(3));
84    /// # let _ = consumer;
85    /// ```
86    #[inline]
87    #[must_use = "if the queue is full, the returned item must be handled (e.g. retried) or it is silently dropped"]
88    pub fn push(&self, item: T) -> Option<T> {
89        self.inner.push(item)
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96    use crate::mem::test_util::NeverAllocator;
97
98    // Shouldn't be possible to construct Arc<Producer<T>> and then use it from different threads as it
99    // will break the requirement of *Single* producer *Single* consumer queue.
100    static_assertions::assert_not_impl_any!(Producer<u32, 0, NeverAllocator>: Sync);
101}
low_latency_data_structures/spsc/producer.rs

low_latency_data_structures/spsc/
producer.rs
