low_latency_data_structures/spsc/
producer.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 ProducerState {
11    pub tail: AtomicUsize,
12    pub cached_head: Cell<usize>,
13}
14
15/// The pushing handle of an SPSC FIFO queue.
16///
17/// Created together with its paired [`Consumer`](crate::spsc::Consumer) by
18/// [`new`](crate::spsc::new) (or [`new_hugepage_backed`](crate::spsc::new_hugepage_backed)).
19/// `Producer` is [`Send`] but not [`Sync`]: at most one thread may push at a
20/// time.
21pub struct Producer<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 Producer<T, CAPACITY, AllocT>
28{
29    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
30        f.debug_struct("Producer")
31            .field("capacity", &CAPACITY)
32            .finish_non_exhaustive()
33    }
34}
35
36// SAFETY: Producer 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 Producer<T, CAPACITY, AllocT>
40{
41}
42
43impl<T, const CAPACITY: usize, AllocT: Allocation<T>> Producer<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    /// Pushes `item` onto the queue.
52    ///
53    /// Wait-free. Never blocks, never allocates. Returns `None` on success.
54    /// If the queue is full the item is returned unchanged as `Some(item)`,
55    /// so the caller can retry or back off without losing data.
56    ///
57    /// # Examples
58    ///
59    /// ```
60    /// use low_latency_data_structures::spsc::new;
61    ///
62    /// let (producer, consumer) = new::<u64, 2>();
63    /// assert_eq!(producer.push(1), None);
64    /// assert_eq!(producer.push(2), None);
65    /// // Queue is full; item is handed back so we can retry later.
66    /// assert_eq!(producer.push(3), Some(3));
67    /// # let _ = consumer;
68    /// ```
69    #[inline]
70    #[must_use = "if the queue is full, the returned item must be handled (e.g. retried) or it is silently dropped"]
71    pub fn push(&self, item: T) -> Option<T> {
72        self.inner.push(item)
73    }
74}
75
76#[cfg(test)]
77mod tests {
78
79    use super::*;
80    use crate::shim::cell::UnsafeCell;
81
82    struct NeverAlloc;
83    impl<T> Allocation<T> for NeverAlloc {
84        fn ptr(&self) -> *mut UnsafeCell<std::mem::MaybeUninit<T>> {
85            unreachable!("it's just a stub")
86        }
87    }
88
89    // Shouldn't be possible to construct Arc<Producer<T>> and then use it from different threads as it
90    // will break the requirement of *Single* producer *Single* consumer queue.
91    static_assertions::assert_not_impl_any!(Producer<u32, 0, NeverAlloc>: Sync);
92}
low_latency_data_structures/spsc/producer.rs

low_latency_data_structures/spsc/
producer.rs
