@@ -48,14 +48,39 @@ pub struct Publisher<T: Copy> {
4848unsafe impl < T : Copy + Send > Send for Publisher < T > { }
4949
5050impl < T : Copy > Publisher < T > {
51- /// Publish a single value. Zero-allocation, O(1).
51+ /// Write a single value to the ring without any backpressure check.
52+ /// This is the raw publish path used by both `publish()` (lossy) and
53+ /// `try_publish()` (after backpressure check passes).
5254#[ inline]
53- pub fn publish ( & mut self , value : T ) {
55+ fn publish_unchecked ( & mut self , value : T ) {
5456 self . ring . slot ( self . seq ) . write ( self . seq , value) ;
5557 self . ring . cursor . 0 . store ( self . seq , Ordering :: Release ) ;
5658 self . seq += 1 ;
5759 }
5860
61+ /// Publish a single value. Zero-allocation, O(1).
62+ ///
63+ /// On a bounded channel (created with [`channel_bounded()`]), this method
64+ /// spin-waits until there is room in the ring, ensuring no message loss.
65+ /// On a regular (lossy) channel, this publishes immediately without any
66+ /// backpressure check.
67+ #[ inline]
68+ pub fn publish ( & mut self , value : T ) {
69+ if self . ring . backpressure . is_some ( ) {
70+ let mut v = value;
71+ loop {
72+ match self . try_publish ( v) {
73+ Ok ( ( ) ) => return ,
74+ Err ( PublishError :: Full ( returned) ) => {
75+ v = returned;
76+ core:: hint:: spin_loop ( ) ;
77+ }
78+ }
79+ }
80+ }
81+ self . publish_unchecked ( value) ;
82+ }
83+
5984 /// Try to publish a single value with backpressure awareness.
6085///
6186/// - On a regular (lossy) channel created with [`channel()`], this always
@@ -86,7 +111,7 @@ impl<T: Copy> Publisher<T> {
86111 }
87112 }
88113 }
89- self . publish ( value) ;
114+ self . publish_unchecked ( value) ;
90115 Ok ( ( ) )
91116 }
92117
@@ -95,11 +120,30 @@ impl<T: Copy> Publisher<T> {
95120/// Each slot is written atomically (seqlock), but the cursor advances only
96121/// once at the end — consumers see the entire batch appear at once, and
97122/// cache-line bouncing on the shared cursor is reduced to one store.
123+ ///
124+ /// On a bounded channel, this spin-waits for room before publishing each
125+ /// value, ensuring no message loss. Values are still committed with a
126+ /// single cursor update at the end.
98127#[ inline]
99128 pub fn publish_batch ( & mut self , values : & [ T ] ) {
100129 if values. is_empty ( ) {
101130 return ;
102131 }
132+ if self . ring . backpressure . is_some ( ) {
133+ for & v in values. iter ( ) {
134+ let mut val = v;
135+ loop {
136+ match self . try_publish ( val) {
137+ Ok ( ( ) ) => break ,
138+ Err ( PublishError :: Full ( returned) ) => {
139+ val = returned;
140+ core:: hint:: spin_loop ( ) ;
141+ }
142+ }
143+ }
144+ }
145+ return ;
146+ }
103147 for ( i, & v) in values. iter ( ) . enumerate ( ) {
104148 let seq = self . seq + i as u64 ;
105149 self . ring . slot ( seq) . write ( seq, v) ;
@@ -143,11 +187,18 @@ impl<T: Copy> Publisher<T> {
143187
144188 /// Pre-fault all ring buffer pages by writing a zero byte to each 4 KiB
145189/// page. Ensures the first publish does not trigger a page fault.
190+ ///
191+ /// # Safety
192+ ///
193+ /// Must be called before any publish/subscribe operations begin.
194+ /// Calling this while the ring is in active use is undefined behavior
195+ /// because it writes zero bytes to live ring memory via raw pointers,
196+ /// which can corrupt slot data and seqlock stamps.
146197#[ cfg( all( target_os = "linux" , feature = "hugepages" ) ) ]
147- pub fn prefault ( & self ) {
198+ pub unsafe fn prefault ( & self ) {
148199 let ptr = self . ring . slots_ptr ( ) as * mut u8 ;
149200 let len = self . ring . slots_byte_len ( ) ;
150- unsafe { crate :: mem:: prefault_pages ( ptr, len) }
201+ crate :: mem:: prefault_pages ( ptr, len)
151202 }
152203}
153204
@@ -195,14 +246,21 @@ impl<T: Copy> Subscribable<T> {
195246///
196247/// This is dramatically faster than `N` independent [`Subscriber`]s when
197248/// polled in a loop on the same thread.
249+ ///
250+ /// # Panics
251+ ///
252+ /// Panics if `N` is 0.
198253pub fn subscribe_group < const N : usize > ( & self ) -> SubscriberGroup < T , N > {
254+ assert ! ( N > 0 , "SubscriberGroup requires at least 1 subscriber" ) ;
199255 let head = self . ring . cursor . 0 . load ( Ordering :: Acquire ) ;
200256 let start = if head == u64:: MAX { 0 } else { head + 1 } ;
257+ let tracker = self . ring . register_tracker ( start) ;
201258 SubscriberGroup {
202259 ring : self . ring . clone ( ) ,
203260 cursors : [ start; N ] ,
204261 total_lagged : 0 ,
205262 total_received : 0 ,
263+ tracker,
206264 }
207265 }
208266
@@ -471,6 +529,17 @@ impl<T: Copy> Subscriber<T> {
471529 }
472530}
473531
532+ impl < T : Copy > Drop for Subscriber < T > {
533+ fn drop ( & mut self ) {
534+ if let Some ( ref tracker) = self . tracker {
535+ if let Some ( ref bp) = self . ring . backpressure {
536+ let mut trackers = bp. trackers . lock ( ) ;
537+ trackers. retain ( |t| !Arc :: ptr_eq ( t, tracker) ) ;
538+ }
539+ }
540+ }
541+ }
542+
474543// ---------------------------------------------------------------------------
475544// SubscriberGroup (batched multi-consumer read)
476545// ---------------------------------------------------------------------------
@@ -495,6 +564,9 @@ pub struct SubscriberGroup<T: Copy, const N: usize> {
495564total_lagged : u64 ,
496565 /// Cumulative messages successfully received.
497566total_received : u64 ,
567+ /// Per-group cursor tracker for backpressure. `None` on regular
568+ /// (lossy) channels — zero overhead.
569+ tracker : Option < Arc < Padded < AtomicU64 > > > ,
498570}
499571
500572unsafe impl < T : Copy + Send , const N : usize > Send for SubscriberGroup < T , N > { }
@@ -521,6 +593,7 @@ impl<T: Copy, const N: usize> SubscriberGroup<T, N> {
521593 }
522594 }
523595 self . total_received += 1 ;
596+ self . update_tracker ( ) ;
524597 Ok ( value)
525598 }
526599 Ok ( None ) => Err ( TryRecvError :: Empty ) ,
@@ -544,6 +617,7 @@ impl<T: Copy, const N: usize> SubscriberGroup<T, N> {
544617 }
545618 }
546619 self . total_lagged += skipped;
620+ self . update_tracker ( ) ;
547621 return Err ( TryRecvError :: Lagged { skipped } ) ;
548622 }
549623 }
@@ -635,6 +709,27 @@ impl<T: Copy, const N: usize> SubscriberGroup<T, N> {
635709 self . total_received as f64 / total as f64
636710 }
637711 }
712+
713+ /// Update the backpressure tracker to reflect the minimum cursor position.
714+ /// No-op on regular (lossy) channels.
715+ #[ inline]
716+ fn update_tracker ( & self ) {
717+ if let Some ( ref tracker) = self . tracker {
718+ let min = self . cursors . iter ( ) . copied ( ) . min ( ) . unwrap_or ( 0 ) ;
719+ tracker. 0 . store ( min, Ordering :: Release ) ;
720+ }
721+ }
722+ }
723+
724+ impl < T : Copy , const N : usize > Drop for SubscriberGroup < T , N > {
725+ fn drop ( & mut self ) {
726+ if let Some ( ref tracker) = self . tracker {
727+ if let Some ( ref bp) = self . ring . backpressure {
728+ let mut trackers = bp. trackers . lock ( ) ;
729+ trackers. retain ( |t| !Arc :: ptr_eq ( t, tracker) ) ;
730+ }
731+ }
732+ }
638733}
639734
640735// ---------------------------------------------------------------------------
0 commit comments