photon_ring/channel/publisher.rs
1// Copyright 2026 Photon Ring Contributors
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4use super::errors::PublishError;
5use crate::pod::Pod;
6use crate::ring::{RingIndex, SharedRing};
7use crate::slot::Slot;
8use alloc::sync::Arc;
9use core::sync::atomic::{AtomicU64, Ordering};
10
11use super::prefetch_write_next;
12
13/// The write side of a Photon SPMC channel.
14///
15/// There is exactly one `Publisher` per channel. It is `Send` but not `Sync` —
16/// only one thread may publish at a time (single-producer guarantee enforced
17/// by `&mut self`).
18pub struct Publisher<T: Pod> {
19 pub(super) ring: Arc<SharedRing<T>>,
20 /// Cached raw pointer to the slot array. Avoids Arc + Box deref on the
21 /// hot path. Valid for the lifetime of `ring` (the Arc keeps it alive).
22 pub(super) slots_ptr: *const Slot<T>,
23 /// Precomputed slot indexing (capacity, mask, reciprocal, pow2 flag).
24 pub(super) index: RingIndex,
25 /// Cached raw pointer to `ring.cursor.0`. Avoids Arc deref on hot path.
26 pub(super) cursor_ptr: *const AtomicU64,
27 pub(super) seq: u64,
28 /// Cached minimum cursor from the last tracker scan. Used as a fast-path
29 /// check to avoid scanning on every `try_publish` call.
30 pub(super) cached_slowest: u64,
31 /// Cached backpressure flag. Avoids Arc deref + Option check on every
32 /// publish() for lossy channels. Immutable after construction.
33 pub(super) has_backpressure: bool,
34}
35
36unsafe impl<T: Pod> Send for Publisher<T> {}
37
38impl<T: Pod> Publisher<T> {
39 /// Spin-wait until backpressure allows publishing.
40 ///
41 /// On a bounded channel, this blocks until the slowest subscriber has
42 /// advanced far enough. On a lossy channel (no backpressure), this is
43 /// a no-op.
44 #[inline]
45 fn wait_for_backpressure(&mut self) {
46 if !self.has_backpressure {
47 return;
48 }
49 while !self.has_room() {
50 core::hint::spin_loop();
51 }
52 }
53
54 /// Whether the current `seq` can be published without lapping the slowest
55 /// subscriber (accounting for the watermark). Always `true` on a lossy
56 /// channel. Refreshes `cached_slowest` as a side effect when it rescans.
57 #[inline]
58 fn has_room(&mut self) -> bool {
59 let effective = match self.ring.backpressure.as_ref() {
60 Some(bp) => self.ring.capacity() - bp.watermark,
61 None => return true,
62 };
63 // Fast path: trust the cached slowest cursor until it says we're close.
64 if self.seq >= self.cached_slowest + effective {
65 // Slow path: rescan all trackers. `None` means no subscribers yet,
66 // so the ring is effectively unbounded — room available.
67 if let Some(slowest) = self.ring.slowest_cursor() {
68 self.cached_slowest = slowest;
69 if self.seq >= slowest + effective {
70 return false;
71 }
72 }
73 }
74 true
75 }
76
77 /// Write a single value to the ring without any backpressure check.
78 /// This is the raw publish path used by both `publish()` (lossy) and
79 /// `try_publish()` (after backpressure check passes).
80 #[inline]
81 fn publish_unchecked(&mut self, value: T) {
82 // SAFETY: slots_ptr is valid for the lifetime of self.ring (Arc-owned).
83 // Index stays within the allocated slot array via RingIndex::slot.
84 let slot = unsafe { &*self.slots_ptr.add(self.index.slot(self.seq)) };
85 prefetch_write_next(self.slots_ptr, self.index.slot(self.seq + 1) as u64);
86 slot.write(self.seq, value);
87 // SAFETY: cursor_ptr points to ring.cursor.0, kept alive by self.ring.
88 unsafe { &*self.cursor_ptr }.store(self.seq, Ordering::Release);
89 self.seq += 1;
90 }
91
92 /// Publish a value built by a closure.
93 ///
94 /// The closure returns the value, so it cannot leave the payload partly
95 /// initialised; it is built as a stack temporary and then written to the
96 /// slot.
97 ///
98 /// On a bounded channel (created with [`channel_bounded()`](super::constructors::channel_bounded)), this method
99 /// spin-waits until there is room in the ring, ensuring no message loss
100 /// (same backpressure semantics as [`publish()`](Self::publish)).
101 /// On a regular (lossy) channel, this publishes immediately.
102 ///
103 /// # Example
104 ///
105 /// ```
106 /// let (mut p, s) = photon_ring::channel::<u64>(64);
107 /// let mut sub = s.subscribe();
108 /// p.publish_with(|| 42u64);
109 /// assert_eq!(sub.try_recv(), Ok(42));
110 /// ```
111 #[inline]
112 pub fn publish_with(&mut self, f: impl FnOnce() -> T) {
113 self.wait_for_backpressure();
114 // SAFETY: see publish_unchecked.
115 let slot = unsafe { &*self.slots_ptr.add(self.index.slot(self.seq)) };
116 prefetch_write_next(self.slots_ptr, self.index.slot(self.seq + 1) as u64);
117 slot.write_with(self.seq, f);
118 unsafe { &*self.cursor_ptr }.store(self.seq, Ordering::Release);
119 self.seq += 1;
120 }
121
122 /// Publish a single value. Zero-allocation, O(1).
123 ///
124 /// On a bounded channel (created with [`channel_bounded()`](super::constructors::channel_bounded)), this method
125 /// spin-waits until there is room in the ring, ensuring no message loss.
126 /// On a regular (lossy) channel, this publishes immediately without any
127 /// backpressure check.
128 #[inline]
129 pub fn publish(&mut self, value: T) {
130 if self.has_backpressure {
131 let mut v = value;
132 loop {
133 match self.try_publish(v) {
134 Ok(()) => return,
135 Err(PublishError::Full(returned)) => {
136 v = returned;
137 core::hint::spin_loop();
138 }
139 }
140 }
141 }
142 self.publish_unchecked(value);
143 }
144
145 /// Try to publish a single value with backpressure awareness.
146 ///
147 /// - On a regular (lossy) channel created with
148 /// [`channel()`](super::constructors::channel), this always
149 /// succeeds — it publishes the value and returns `Ok(())`.
150 /// - On a bounded channel created with [`channel_bounded()`](super::constructors::channel_bounded), this checks
151 /// whether the slowest subscriber has fallen too far behind. If
152 /// `publisher_seq - slowest_cursor >= capacity - watermark`, it returns
153 /// `Err(PublishError::Full(value))` without writing.
154 #[inline]
155 pub fn try_publish(&mut self, value: T) -> Result<(), PublishError<T>> {
156 if !self.has_room() {
157 return Err(PublishError::Full(value));
158 }
159 self.publish_unchecked(value);
160 Ok(())
161 }
162
163 /// Publish a batch of values.
164 ///
165 /// Both lossy and bounded channels advance the cursor per-value, so
166 /// a `subscribe()` call concurrent with publication will only see
167 /// messages published after the subscribe point (future-only contract).
168 ///
169 /// On a **bounded** channel: spin-waits for room before each value,
170 /// ensuring no message loss.
171 #[inline]
172 pub fn publish_batch(&mut self, values: &[T]) {
173 if values.is_empty() {
174 return;
175 }
176 if self.has_backpressure {
177 for &v in values.iter() {
178 let mut val = v;
179 loop {
180 match self.try_publish(val) {
181 Ok(()) => break,
182 Err(PublishError::Full(returned)) => {
183 val = returned;
184 core::hint::spin_loop();
185 }
186 }
187 }
188 }
189 return;
190 }
191 // Write each slot and advance the cursor per-value to maintain the
192 // "future-only subscribe" invariant: subscribe() snapshots the cursor,
193 // so any slot written before the cursor update could be visible to a
194 // subscriber created mid-batch.
195 for &v in values.iter() {
196 self.publish_unchecked(v);
197 }
198 }
199
200 /// Number of messages published so far, which is also the next sequence
201 /// number to be written. Useful for computing lag:
202 /// `publisher.published() - subscriber.cursor`.
203 #[inline]
204 pub fn published(&self) -> u64 {
205 self.seq
206 }
207
208 /// Ring capacity.
209 #[inline]
210 pub fn capacity(&self) -> u64 {
211 self.ring.capacity()
212 }
213
214 /// Lock the ring buffer pages in RAM, preventing the OS from swapping
215 /// them to disk. Reduces worst-case latency by eliminating page-fault
216 /// stalls on the hot path.
217 ///
218 /// Returns `true` on success. Requires `CAP_IPC_LOCK` or sufficient
219 /// `RLIMIT_MEMLOCK` on Linux. No-op on other platforms.
220 #[cfg(all(target_os = "linux", feature = "hugepages"))]
221 pub fn mlock(&self) -> bool {
222 let ptr = self.ring.slots_ptr() as *const u8;
223 let len = self.ring.slots_byte_len();
224 unsafe { crate::mem::mlock_pages(ptr, len) }
225 }
226
227 /// Pre-fault all ring buffer pages by writing a zero byte to each 4 KiB
228 /// page. Ensures the first publish does not trigger a page fault.
229 ///
230 /// # Safety
231 ///
232 /// Must be called before any publish/subscribe operations begin.
233 /// Calling this while the ring is in active use is undefined behavior
234 /// because it writes zero bytes to live ring memory via raw pointers,
235 /// which can corrupt slot data and seqlock stamps.
236 #[cfg(all(target_os = "linux", feature = "hugepages"))]
237 pub unsafe fn prefault(&self) {
238 assert!(
239 self.seq == 0,
240 "prefault() must be called before any publish operations"
241 );
242 let ptr = self.ring.slots_ptr() as *mut u8;
243 let len = self.ring.slots_byte_len();
244 crate::mem::prefault_pages(ptr, len)
245 }
246}