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

low_latency_data_structures/spsc/
producer.rs
