low_latency_data_structures/spmc/
producer.rs

1use std::marker::PhantomData;
2use std::sync::Arc;
3use std::sync::atomic::AtomicUsize;
4
5use crossbeam_utils::CachePadded;
6
7use crate::mem::Allocator;
8use crate::spmc::queue::{Queue, Slot};
9
10pub(super) type ProducerState = CachePadded<ProducerStateInner>;
11
12pub(super) struct ProducerStateInner {
13    pub(super) write_cursor: AtomicUsize,
14}
15
16/// The publishing handle of an SPMC broadcast queue.
17///
18/// Created together with its [`Consumer`](crate::spmc::Consumer) by
19/// [`new`](crate::spmc::new). A `Producer` is [`Send`] but not [`Sync`]:
20/// at most one thread may publish at a time. To enforce that, the type is
21/// neither [`Clone`] nor [`Copy`].
22///
23/// Publishing never blocks and never allocates: when the queue is full the
24/// oldest slot is overwritten, and any consumer that hasn't read past that
25/// slot will observe a [`ReadResult::Lapped`](crate::spmc::ReadResult::Lapped)
26/// on its next read.
27pub struct Producer<T, const CAPACITY: usize, A>
28where
29    T: bytemuck::AnyBitPattern,
30    A: Allocator,
31{
32    inner: Arc<Queue<T, CAPACITY, A>>,
33    _not_sync: PhantomData<*const ()>,
34}
35
36impl<T, const CAPACITY: usize, A> std::fmt::Debug for Producer<T, CAPACITY, A>
37where
38    T: bytemuck::AnyBitPattern,
39    A: Allocator,
40{
41    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
42        f.debug_struct("Producer")
43            .field("capacity", &CAPACITY)
44            .finish_non_exhaustive()
45    }
46}
47
48// SAFETY: it is Send on its own but we need to restrict only Sync.
49unsafe impl<T, const CAPACITY: usize, A> Send for Producer<T, CAPACITY, A>
50where
51    A: Allocator,
52    T: bytemuck::AnyBitPattern + Send,
53    A::Allocation<Slot<T>>: Send,
54{
55}
56
57#[cfg(test)]
58mod tests {
59    use super::*;
60    use crate::mem::test_util::NeverAllocator;
61
62    static_assertions::assert_impl_all!(Producer<u32, 1, NeverAllocator>: Send);
63    static_assertions::assert_not_impl_any!(Producer<u32, 1, NeverAllocator>: Sync, Clone, Copy);
64}
65
66impl<T, const CAPACITY: usize, A> Producer<T, CAPACITY, A>
67where
68    T: bytemuck::AnyBitPattern,
69    A: Allocator,
70{
71    pub(super) fn new(queue: Arc<Queue<T, CAPACITY, A>>) -> Self {
72        Self {
73            inner: queue,
74            _not_sync: PhantomData,
75        }
76    }
77
78    /// Publishes `value` into the next slot.
79    ///
80    /// Wait-free. Never blocks, never allocates. If the queue is full, the
81    /// oldest slot is silently overwritten; consumers that haven't read past
82    /// that slot will observe a [`ReadResult::Lapped`](crate::spmc::ReadResult::Lapped)
83    /// on their next [`try_read`](crate::spmc::Consumer::try_read).
84    ///
85    /// # Examples
86    ///
87    /// ```
88    /// use low_latency_data_structures::spmc::{self, ReadResult, new};
89    /// use low_latency_data_structures::mem::global::GlobalAllocator;
90    ///
91    /// let (producer, mut consumer) = new::<u64, 16, GlobalAllocator>(
92    ///     spmc::Options::global_mlocked(),
93    /// );
94    /// producer.publish(42);
95    /// assert_eq!(consumer.try_read(), ReadResult::Value(42));
96    /// ```
97    #[inline]
98    pub fn publish(&self, value: T) {
99        self.inner.publish(value)
100    }
101}
low_latency_data_structures/spmc/producer.rs

low_latency_data_structures/spmc/
producer.rs
