low_latency_data_structures/spmc/
producer.rs

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

low_latency_data_structures/spmc/
producer.rs
