low_latency_data_structures/spmc/
producer.rs

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

low_latency_data_structures/spmc/
producer.rs
