disrupt_rs/consumer/event_poller.rs
1use crate::{
2 barrier::Barrier,
3 cursor::Cursor,
4 ringbuffer::RingBuffer,
5 wait_strategies::{WaitOutcome, Waiter, WakeupNotifier},
6 Sequence,
7};
8use crossbeam_utils::CachePadded;
9use std::sync::{
10 atomic::{fence, AtomicI64, Ordering},
11 Arc,
12};
13use thiserror::Error;
14
15/// Represents an EventPoller that can be used to poll events from the ring buffer.
16/// Use the EventPoller when you want to control your own thread
17/// (instead of letting the Disruptor manage it).
18/// The EventPoller supports batch reading and can be used to process events in a loop.
19///
20/// When the [`EventGuard`] is dropped, it advances the consumer cursor and wakes any
21/// downstream consumers that may be blocked (when using blocking wait strategies).
22///
23/// ```
24///# use disrupt_rs::*;
25///#
26///# #[derive(Debug)]
27///# struct Event {
28///# price: f64
29///# }
30///# let factory = || Event { price: 0.0 };
31///# let builder = build_single_producer(8, factory, BusySpin);
32///# let (mut event_poller, builder) = builder.event_poller();
33///# let mut producer = builder.build();
34///# producer.publish(|e| { e.price = 42.0; });
35///# drop(producer);
36/// loop {
37/// // 1. Either poll all available events:
38/// match event_poller.poll() {
39/// Ok(mut events) => {
40/// // Batch process events if efficient in your use case.
41/// let batch_size = (&mut events).len();
42/// // The guard named `events` is an `Iterator` so you can read events by iterating.
43/// for (sequence, event) in &mut events {
44/// println!("Processing event: {:?}", event);
45/// let _ = sequence;
46/// }
47/// },// When the guard (here named `events`) goes out of scope,
48/// // it signals to the Disruptor that reading is done.
49/// Err(Polling::NoEvents) => { /* Do other work or try again. */ },
50/// Err(Polling::Shutdown) => { break; }, // Exit the loop if the Disruptor is shut down.
51/// }
52/// // 2. Or limit the number of events per poll:
53/// match event_poller.poll_take(64) {
54/// Ok(mut events) => {
55/// // Process events same as above but yielding at most 64 events.
56/// for (sequence, event) in &mut events {
57/// println!("Processing event: {:?}", event);
58/// let _ = sequence;
59/// }
60/// },
61/// Err(Polling::NoEvents) => {},
62/// Err(Polling::Shutdown) => { break; },
63/// }
64/// }
65/// ```
66pub struct EventPoller<E, B, N: WakeupNotifier = crate::wait_strategies::NoopWakeupNotifier> {
67 ring_buffer: Arc<RingBuffer<E>>,
68 dependent_barrier: Arc<B>,
69 shutdown_at_sequence: Arc<CachePadded<AtomicI64>>,
70 cursor: Arc<Cursor>,
71 notifier: N,
72}
73
74/// Error types that can occur if polling is unsuccessful.
75#[derive(Debug, Error, PartialEq)]
76pub enum Polling {
77 /// Indicates that there are no events available to process.
78 #[error("No available events.")]
79 NoEvents,
80 /// Indicates that the Disruptor has been shut down.
81 #[error("Disruptor is shut down.")]
82 Shutdown,
83}
84
85/// Guards the available events that can be processed when using the EventPoller API.
86/// It can be used as an iterator to read the published events and the dropping of the `EventGuard`
87/// will signal to the `Disruptor` that the reading is completed. This will allow other consumers or
88/// producers to advance.
89///
90/// On drop, the guard advances the consumer cursor and wakes any downstream consumers
91/// that may be blocked on this poller's progress.
92pub struct EventGuard<'p, E, B, N: WakeupNotifier = crate::wait_strategies::NoopWakeupNotifier> {
93 parent: &'p mut EventPoller<E, B, N>,
94 sequence: Sequence,
95 available: Sequence,
96}
97
98impl<E, B, N: WakeupNotifier> EventGuard<'_, E, B, N> {
99 /// Returns the number of events available to read.
100 ///
101 /// This is identical to the `ExactSizeIterator::len` implementation, but provided as an
102 /// inherent method so callers don't need to pass `&mut` purely to compute length.
103 pub fn len(&self) -> usize {
104 if self.sequence > self.available {
105 0
106 } else {
107 (self.available - self.sequence + 1) as usize
108 }
109 }
110
111 /// Returns true if no events are available to read.
112 pub fn is_empty(&self) -> bool {
113 self.len() == 0
114 }
115}
116
117/// The Iterator is implemented for the `&mut EventGuard` to bind the returned
118/// events' lifetime to the lifetime (`'g`) of the `EventGuard`.
119/// (And not the lifetime of the `EventPoller`. The latter would be catastrophic because a client
120/// could hold on to a reference to a en Event after the drop method was run for the
121/// `EventGuard` and a publisher could write to the immutable reference - UB.)
122impl<'g, E, B, N: WakeupNotifier> Iterator for &'g mut EventGuard<'_, E, B, N> {
123 type Item = (Sequence, &'g E);
124
125 fn next(&mut self) -> Option<Self::Item> {
126 if self.sequence > self.available {
127 return None;
128 }
129
130 let sequence = self.sequence;
131 // SAFETY: The Guard is authorized to read up to and including `available` sequence.
132 let event_ptr = unsafe { self.parent.ring_buffer.get(sequence).as_ptr() };
133 let event = unsafe { &*event_ptr };
134 self.sequence += 1;
135 Some((sequence, event))
136 }
137}
138
139impl<E, B, N: WakeupNotifier> ExactSizeIterator for &mut EventGuard<'_, E, B, N> {
140 /// Returns the number of events available to read.
141 fn len(&self) -> usize {
142 (**self).len()
143 }
144}
145
146impl<E, B, N: WakeupNotifier> Drop for EventGuard<'_, E, B, N> {
147 fn drop(&mut self) {
148 // Signal to producers or later consumers that we're done processing `available` sequence.
149 // Note, not all events in the range have to have been read.
150 // (I.e. client code can skip reading any number events.)
151 self.parent.cursor.store(self.available);
152 // Wake any blocking waiters that might be gated on this consumer's progress.
153 self.parent.notifier.wake();
154 }
155}
156
157impl<E, B, N> EventPoller<E, B, N>
158where
159 B: Barrier,
160 N: WakeupNotifier,
161{
162 pub(crate) fn new(
163 ring_buffer: Arc<RingBuffer<E>>,
164 dependent_barrier: Arc<B>,
165 shutdown_at_sequence: Arc<CachePadded<AtomicI64>>,
166 cursor: Arc<Cursor>,
167 notifier: N,
168 ) -> Self {
169 Self {
170 ring_buffer,
171 dependent_barrier,
172 shutdown_at_sequence,
173 cursor,
174 notifier,
175 }
176 }
177
178 /// Polls the ring buffer and returns an [`EventGuard`] if any events are available.
179 /// The guard can be used like an iterator and yields all available events at the time of polling.
180 /// Dropping the guard will signal to the Disruptor that the events have been processed.
181 ///
182 /// This method does not block; it will return immediately.
183 ///
184 /// This method is equivalent to calling [`EventPoller::poll_take`] with `u64::MAX` as the limit.
185 ///
186 /// # Errors
187 ///
188 /// This method returns an error if:
189 /// 1. No events are available: [`Polling::NoEvents`]
190 /// 2. The Disruptor is shut down: [`Polling::Shutdown`]
191 ///
192 /// # Examples
193 ///
194 /// ```
195 ///# use disrupt_rs::*;
196 ///#
197 ///# #[derive(Debug)]
198 ///# struct Event {
199 ///# price: f64
200 ///# }
201 ///# let factory = || Event { price: 0.0 };
202 ///# let builder = build_single_producer(8, factory, BusySpin);
203 ///# let (mut event_poller, builder) = builder.event_poller();
204 ///# let mut producer = builder.build();
205 ///# producer.publish(|e| { e.price = 42.0; });
206 ///# drop(producer);
207 /// match event_poller.poll() {
208 /// Ok(mut events) => {
209 /// for (sequence, event) in &mut events {
210 /// let _ = sequence;
211 /// let _ = event;
212 /// }
213 /// },
214 /// Err(Polling::NoEvents) => { /* ... */ },
215 /// Err(Polling::Shutdown) => { /* ... */ },
216 /// };
217 /// ```
218 pub fn poll(&mut self) -> Result<EventGuard<'_, E, B, N>, Polling> {
219 self.poll_take(u64::MAX)
220 }
221
222 /// Polls for available events, yielding at most `limit` events.
223 ///
224 /// This method behaves like [`EventPoller::poll`], but caps the number of events yielded by the returned
225 /// [`EventGuard`]. Fewer events may be yielded if less are available at the time of polling.
226 ///
227 /// Note: A `limit` of zero returns an [`EventGuard`] that yields no events (and not a [Polling::NoEvents] error).
228 ///
229 /// # Examples
230 ///
231 /// ```
232 ///# use disrupt_rs::*;
233 ///#
234 ///# #[derive(Debug)]
235 ///# struct Event {
236 ///# price: f64
237 ///# }
238 ///# let factory = || Event { price: 0.0 };
239 ///# let builder = build_single_producer(8, factory, BusySpin);
240 ///# let (mut event_poller, builder) = builder.event_poller();
241 ///# let mut producer = builder.build();
242 ///# producer.publish(|e| { e.price = 42.0; });
243 ///# drop(producer);
244 /// match event_poller.poll_take(64) {
245 /// Ok(mut events) => {
246 /// // Process events same as above but yielding at most 64 events.
247 /// for (sequence, event) in &mut events {
248 /// let _ = sequence;
249 /// let _ = event;
250 /// }
251 /// },
252 /// Err(Polling::NoEvents) => { /* ... */ },
253 /// Err(Polling::Shutdown) => { /* ... */ },
254 /// };
255 /// ```
256 pub fn poll_take(&mut self, limit: u64) -> Result<EventGuard<'_, E, B, N>, Polling> {
257 let cursor_at = self.cursor.relaxed_value();
258 let sequence = cursor_at + 1;
259
260 if sequence == self.shutdown_at_sequence.load(Ordering::Relaxed) {
261 return Err(Polling::Shutdown);
262 }
263
264 let available = self.dependent_barrier.get_after(sequence);
265 if available < sequence {
266 return Err(Polling::NoEvents);
267 }
268 fence(Ordering::Acquire);
269
270 let max_sequence = cursor_at.saturating_add_unsigned(limit);
271 let available = std::cmp::min(available, max_sequence);
272
273 Ok(EventGuard {
274 parent: self,
275 sequence,
276 available,
277 })
278 }
279
280 /// Wait for events using the provided [`Waiter`] and then return an [`EventGuard`].
281 ///
282 /// This is useful when you want to drive an [`EventPoller`] from your own thread while still
283 /// reusing the library's wait strategies (e.g. [`BusySpin`](crate::BusySpin),
284 /// [`YieldingWaitStrategy`](crate::YieldingWaitStrategy)).
285 ///
286 /// If the waiter returns [`WaitOutcome::Timeout`], this returns [`Polling::NoEvents`] so the
287 /// caller can do other work.
288 pub fn poll_wait(
289 &mut self,
290 waiter: &mut impl Waiter,
291 ) -> Result<EventGuard<'_, E, B, N>, Polling> {
292 let sequence = self.cursor.relaxed_value() + 1;
293
294 let available = match waiter.wait_for(
295 sequence,
296 self.dependent_barrier.as_ref(),
297 self.shutdown_at_sequence.as_ref(),
298 ) {
299 WaitOutcome::Available { upper } => upper,
300 WaitOutcome::Shutdown => return Err(Polling::Shutdown),
301 WaitOutcome::Timeout => return Err(Polling::NoEvents),
302 };
303
304 Ok(EventGuard {
305 parent: self,
306 sequence,
307 available,
308 })
309 }
310}
311
312#[cfg(test)]
313mod tests {
314 use super::*;
315 use crate::consumer::SingleConsumerBarrier;
316 use crate::ringbuffer::RingBuffer;
317 use crate::wait_strategies::{NoopWakeupNotifier, WaitStrategy};
318 use crate::Producer;
319
320 #[test]
321 fn poller_observes_manual_shutdown_without_producer_drop() {
322 #[derive(Debug)]
323 struct Event;
324
325 let ring_buffer = Arc::new(RingBuffer::new(2, || Event));
326 let cursor = Arc::new(Cursor::new(-1));
327 let shutdown = Arc::new(CachePadded::new(AtomicI64::new(-1)));
328 let barrier = Arc::new(SingleConsumerBarrier::new(Arc::clone(&cursor)));
329
330 let mut poller = EventPoller::new(
331 ring_buffer,
332 barrier,
333 Arc::clone(&shutdown),
334 Arc::clone(&cursor),
335 NoopWakeupNotifier,
336 );
337
338 // Signal shutdown as if producer had advanced to sequence 0.
339 shutdown.store(0, Ordering::Relaxed);
340 assert_eq!(poller.poll().err(), Some(Polling::Shutdown));
341 }
342
343 #[test]
344 fn poller_exits_on_shutdown_race() {
345 #[derive(Debug)]
346 struct Event;
347
348 let ring_buffer = Arc::new(RingBuffer::new(2, || Event));
349 let cursor = Arc::new(Cursor::new(-1));
350 let shutdown = Arc::new(CachePadded::new(AtomicI64::new(-1)));
351 let barrier = Arc::new(SingleConsumerBarrier::new(Arc::clone(&cursor)));
352
353 let mut poller = EventPoller::new(
354 ring_buffer,
355 barrier,
356 Arc::clone(&shutdown),
357 Arc::clone(&cursor),
358 NoopWakeupNotifier,
359 );
360
361 let shutdown_ref = shutdown.clone();
362 let handle = std::thread::spawn(move || {
363 std::thread::sleep(std::time::Duration::from_millis(20));
364 shutdown_ref.store(0, Ordering::Relaxed);
365 });
366
367 // Spin poll until shutdown is observed.
368 let mut saw_shutdown = false;
369 for _ in 0..10 {
370 if matches!(poller.poll(), Err(Polling::Shutdown)) {
371 saw_shutdown = true;
372 break;
373 }
374 std::thread::sleep(std::time::Duration::from_millis(5));
375 }
376 handle.join().unwrap();
377 assert!(saw_shutdown, "poller should see shutdown promptly");
378 }
379
380 #[test]
381 fn poll_wait_can_drive_poller_with_library_waiter() {
382 #[derive(Debug)]
383 struct Event(i64);
384
385 let factory = || Event(0);
386 let builder = crate::build_single_producer(8, factory, crate::BusySpin);
387 let (mut poller, builder) = builder.event_poller();
388 let mut producer = builder.build();
389
390 producer.publish(|e| e.0 = 42);
391 drop(producer);
392
393 let mut waiter = crate::BusySpin.new_waiter();
394
395 let mut guard = poller
396 .poll_wait(&mut waiter)
397 .expect("should see published event");
398 let (_seq, got) = (&mut guard).next().unwrap();
399 let got = got.0;
400 assert_eq!(got, 42);
401 drop(guard);
402
403 assert_eq!(poller.poll_wait(&mut waiter).err(), Some(Polling::Shutdown));
404 }
405}