ph_eventing/event_buf.rs
1//! Bounded SPSC event buffer with backpressure — no heap, no alloc.
2//!
3//! [`EventBuf`] is a fixed-size, lock-free, single-producer single-consumer
4//! ring buffer that **rejects** pushes when full instead of overwriting.
5//! This gives the producer explicit backpressure so no events are silently
6//! lost.
7//!
8//! # When to use
9//! Use `EventBuf` when every event matters and the producer can afford to
10//! handle a "buffer full" signal (retry, log, or apply its own policy).
11//! If losing old events is acceptable, prefer [`crate::SeqRing`].
12//! If you only need a single-owner ring, see [`crate::RingBuf`].
13//!
14//! # Memory ordering
15//! This is a classic Lamport SPSC queue:
16//! - The producer owns `head` (Relaxed load, Release store) and reads
17//! `tail` with Acquire to see consumer progress.
18//! - The consumer owns `tail` (Relaxed load, Release store) and reads
19//! `head` with Acquire to see producer progress.
20//! - A slot is written before `head` is advanced and read before `tail` is
21//! advanced, so the Release/Acquire pairs on the cursors act as the
22//! publication fence.
23//! - The producer and consumer never touch the same slot: `push` writes at
24//! `head` only while `head - tail < N`, so there is no data race on the
25//! slots themselves, only on the cursors.
26//! - [`EventBuf::len`] is the one observer that reads both cursors, so it
27//! brackets its `head` load between two `tail` samples to get a consistent
28//! pair.
29//!
30//! # Example
31//! ```
32//! use ph_eventing::EventBuf;
33//!
34//! let buf = EventBuf::<u32, 4>::new();
35//! let producer = buf.try_producer().expect("producer");
36//! let consumer = buf.try_consumer().expect("consumer");
37//!
38//! assert!(producer.push(1).is_ok());
39//! assert!(producer.push(2).is_ok());
40//! assert_eq!(consumer.peek(), Some(1));
41//! assert_eq!(consumer.pop(), Some(1));
42//! assert_eq!(consumer.pop(), Some(2));
43//! assert_eq!(consumer.pop(), None); // empty
44//! ```
45
46use crate::sync::{AtomicBool, AtomicU32, Ordering, TrackedCell, fence};
47use core::cell::Cell;
48use core::marker::PhantomData;
49use core::mem::MaybeUninit;
50
51// Const on the host path so `EventBuf::new` can be const. Loom's cell is not
52// const-constructible, so the Loom build keeps a non-const helper.
53//
54// Prefer `[const { … }; N]` over `array::from_fn`: the latter is not
55// const-callable with these constructors on the MSRV toolchain.
56#[cfg(not(loom))]
57const fn slot_array<T, const N: usize>() -> [TrackedCell<MaybeUninit<T>>; N] {
58 [const { TrackedCell::new(MaybeUninit::uninit()) }; N]
59}
60
61#[cfg(loom)]
62fn slot_array<T, const N: usize>() -> [TrackedCell<MaybeUninit<T>>; N] {
63 core::array::from_fn(|_| TrackedCell::new(MaybeUninit::uninit()))
64}
65
66/// How many times [`EventBuf::len`] retries its snapshot before falling back
67/// to a clamped estimate. Bounded so `len` is always wait-free.
68const RETRY_LIMIT: usize = 2;
69
70/// Bounded SPSC event buffer with backpressure.
71///
72/// When the buffer is full, [`Producer::push`] returns `Err(val)` instead
73/// of overwriting, giving the producer a chance to retry, drop, or log.
74/// The consumer drains items with [`Consumer::pop`] or [`Consumer::drain`],
75/// and can inspect the oldest item with [`Consumer::peek`] without consuming it.
76///
77/// # Panics
78/// - `EventBuf::new()` fails to compile (const assertion) when `N == 0` on the
79/// host path; under Loom it panics at runtime.
80/// - `producer()` / `consumer()` panic if called while another handle of
81/// the same kind is already active. Use [`EventBuf::try_producer`] /
82/// [`EventBuf::try_consumer`] for a fallible alternative.
83pub struct EventBuf<T: Copy, const N: usize> {
84 head: AtomicU32,
85 tail: AtomicU32,
86 slots: [TrackedCell<MaybeUninit<T>>; N],
87 producer_taken: AtomicBool,
88 consumer_taken: AtomicBool,
89}
90
91// SAFETY: EventBuf is Sync because the producer/consumer handles enforce
92// SPSC usage, and the head/tail cursors are accessed via atomics with
93// Release/Acquire ordering that guarantees slot visibility. T: Send ensures
94// values can be transferred across threads safely.
95unsafe impl<T: Copy + Send, const N: usize> Sync for EventBuf<T, N> {}
96
97impl<T: Copy, const N: usize> EventBuf<T, N> {
98 /// Create a new, empty event buffer.
99 ///
100 /// On the normal (non-Loom) build this is a `const fn`, so the buffer can
101 /// be placed in a `static`:
102 /// `static BUF: EventBuf<u32, 64> = EventBuf::new();`.
103 /// Under `--cfg loom` it is deliberately non-const — Loom's atomics are
104 /// not const-constructible.
105 ///
106 /// # Capacity `0` is a build failure
107 /// The `N > 0` check is a *const* assertion, so a zero-capacity buffer
108 /// cannot be constructed at all -- there is no runtime panic left to
109 /// catch, and therefore no way to write the negative case as a `#[test]`.
110 /// This `compile_fail` doctest is that coverage, and pinning the error code
111 /// keeps it honest: without it the test would also pass on a typo.
112 ///
113 /// ```compile_fail,E0080
114 /// let _ = ph_eventing::EventBuf::<u32, 0>::new();
115 /// ```
116 ///
117 /// # Panics
118 /// Does not panic on the host path. Under Loom, where `new` is non-const,
119 /// `N == 0` is a runtime assertion instead.
120 #[cfg(not(loom))]
121 pub const fn new() -> Self {
122 const {
123 assert!(N > 0, "EventBuf capacity N must be > 0");
124 }
125 Self {
126 head: AtomicU32::new(0),
127 tail: AtomicU32::new(0),
128 slots: slot_array::<T, N>(),
129 producer_taken: AtomicBool::new(false),
130 consumer_taken: AtomicBool::new(false),
131 }
132 }
133
134 /// Create a new, empty event buffer (Loom build — non-const).
135 ///
136 /// # Panics
137 /// Panics if `N == 0`.
138 #[cfg(loom)]
139 pub fn new() -> Self {
140 assert!(N > 0, "EventBuf capacity N must be > 0");
141 Self {
142 head: AtomicU32::new(0),
143 tail: AtomicU32::new(0),
144 slots: slot_array::<T, N>(),
145 producer_taken: AtomicBool::new(false),
146 consumer_taken: AtomicBool::new(false),
147 }
148 }
149
150 #[inline(always)]
151 const fn slot_index(pos: u32) -> usize {
152 (pos as usize) % N
153 }
154
155 /// Maximum number of items the buffer can hold.
156 #[inline]
157 pub const fn capacity(&self) -> usize {
158 N
159 }
160
161 /// Approximate number of items currently buffered.
162 ///
163 /// Returns a consistent `(tail, head)` snapshot. The value may still be
164 /// stale by the time the caller acts on it, but it will never spuriously
165 /// exceed [`capacity`](Self::capacity).
166 ///
167 /// This never blocks: it makes a bounded number of attempts and then falls
168 /// back to a clamped estimate, so a busy consumer cannot stall the caller.
169 #[inline]
170 pub fn len(&self) -> usize {
171 // Seqlock-style read. Sampling `tail` on both sides of the `head` load
172 // and requiring the samples to match means `head` was observed while
173 // `tail` held still, so `head.wrapping_sub(tail)` cannot appear as a
174 // huge unsigned value after a concurrent consumer advance.
175 //
176 // The two barriers pin the `head` load between the samples. They also
177 // make equality a sound bound: if `h` observes a producer publication
178 // that reused consumer-freed space, the following Acquire fence
179 // synchronizes through that Relaxed load. The producer acquired the
180 // newer `tail` before publishing `h`, so `t2` cannot then observe an
181 // older tail. Thus `t1 == t2` implies `h - t1 <= N`.
182 //
183 // This depends on EventBuf's backpressure protocol: the producer reads
184 // `tail` with Acquire before advancing `head`. It is not a generic
185 // double-sampling property. See AGENTS.md for the full happens-before
186 // chain.
187 for _ in 0..RETRY_LIMIT {
188 let t1 = self.tail.load(Ordering::Acquire);
189 let h = self.head.load(Ordering::Relaxed);
190 fence(Ordering::Acquire);
191 let t2 = self.tail.load(Ordering::Relaxed);
192
193 if t1 == t2 {
194 return h.wrapping_sub(t1) as usize;
195 }
196 }
197
198 // The consumer moved during every attempt. Read `tail` first so a
199 // further advance can only make the count an over-estimate rather
200 // than an underflow, then clamp to preserve the capacity bound.
201 let t = self.tail.load(Ordering::Acquire);
202 let h = self.head.load(Ordering::Relaxed);
203 (h.wrapping_sub(t) as usize).min(N)
204 }
205
206 /// Returns `true` if the buffer contains no items (approximate).
207 #[inline]
208 pub fn is_empty(&self) -> bool {
209 self.len() == 0
210 }
211
212 /// Returns `true` if the buffer is at capacity (approximate).
213 #[inline]
214 pub fn is_full(&self) -> bool {
215 self.len() >= N
216 }
217
218 /// Try to create the producer handle.
219 ///
220 /// Returns `None` if a producer is already active. Prefer this over
221 /// [`producer`](Self::producer) when fallible bring-up is needed.
222 #[inline]
223 pub fn try_producer(&self) -> Option<Producer<'_, T, N>> {
224 if self.producer_taken.swap(true, Ordering::AcqRel) {
225 None
226 } else {
227 Some(Producer {
228 buf: self,
229 _not_sync: PhantomData,
230 })
231 }
232 }
233
234 /// Create the producer handle. Only one producer may be active.
235 ///
236 /// # Deprecated
237 /// Prefer [`try_producer`](Self::try_producer). This crate targets firmware,
238 /// where a panic is a reset and the panic machinery itself costs flash — a
239 /// code-size probe shows no panic strings reach the binary when only the
240 /// `try_*` constructors are used. The shorter, more discoverable name being
241 /// the hazardous one is the inversion this deprecation exists to correct.
242 ///
243 /// Still sound, still tested, and convenient on a host where a panic is just
244 /// a failed test. Scheduled for removal in 0.3.0.
245 ///
246 /// # Panics
247 /// Panics if a producer handle is already active.
248 #[deprecated(
249 since = "0.2.0",
250 note = "on an embedded target a panic is a reset, and the panic machinery costs flash; use try_producer() and handle None"
251 )]
252 #[inline]
253 pub fn producer(&self) -> Producer<'_, T, N> {
254 self.try_producer()
255 .expect("EventBuf: only one Producer may be active at a time")
256 }
257
258 /// Try to create the consumer handle.
259 ///
260 /// Returns `None` if a consumer is already active. Prefer this over
261 /// [`consumer`](Self::consumer) when fallible bring-up is needed.
262 #[inline]
263 pub fn try_consumer(&self) -> Option<Consumer<'_, T, N>> {
264 if self.consumer_taken.swap(true, Ordering::AcqRel) {
265 None
266 } else {
267 Some(Consumer {
268 buf: self,
269 _not_sync: PhantomData,
270 })
271 }
272 }
273
274 /// Create the consumer handle. Only one consumer may be active.
275 ///
276 /// # Deprecated
277 /// Prefer [`try_consumer`](Self::try_consumer). This crate targets firmware,
278 /// where a panic is a reset and the panic machinery itself costs flash — a
279 /// code-size probe shows no panic strings reach the binary when only the
280 /// `try_*` constructors are used. The shorter, more discoverable name being
281 /// the hazardous one is the inversion this deprecation exists to correct.
282 ///
283 /// Still sound, still tested, and convenient on a host where a panic is just
284 /// a failed test. Scheduled for removal in 0.3.0.
285 ///
286 /// # Panics
287 /// Panics if a consumer handle is already active.
288 #[deprecated(
289 since = "0.2.0",
290 note = "on an embedded target a panic is a reset, and the panic machinery costs flash; use try_consumer() and handle None"
291 )]
292 #[inline]
293 pub fn consumer(&self) -> Consumer<'_, T, N> {
294 self.try_consumer()
295 .expect("EventBuf: only one Consumer may be active at a time")
296 }
297}
298
299impl<T: Copy, const N: usize> Default for EventBuf<T, N> {
300 fn default() -> Self {
301 Self::new()
302 }
303}
304
305impl<T: Copy, const N: usize> core::fmt::Debug for EventBuf<T, N> {
306 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
307 f.debug_struct("EventBuf")
308 .field("len", &self.len())
309 .field("capacity", &N)
310 .finish()
311 }
312}
313
314/// Write handle for an [`EventBuf`].
315///
316/// Dropping the producer releases the slot so a new one can be created.
317pub struct Producer<'a, T: Copy, const N: usize> {
318 buf: &'a EventBuf<T, N>,
319 _not_sync: PhantomData<Cell<()>>,
320}
321
322impl<T: Copy, const N: usize> Producer<'_, T, N> {
323 /// Try to push a value into the buffer.
324 ///
325 /// Returns `Ok(())` on success, or `Err(val)` if the buffer is full
326 /// (the value is returned to the caller so nothing is lost).
327 #[inline]
328 pub fn push(&self, val: T) -> Result<(), T> {
329 let head = self.buf.head.load(Ordering::Relaxed);
330 let tail = self.buf.tail.load(Ordering::Acquire);
331 if head.wrapping_sub(tail) as usize >= N {
332 return Err(val);
333 }
334 let idx = EventBuf::<T, N>::slot_index(head);
335 // SAFETY: producer is the only writer to this slot; the consumer
336 // will not read it until head is advanced (Release below).
337 self.buf.slots[idx].with_mut(|slot| unsafe { (*slot).write(val) });
338 self.buf.head.store(head.wrapping_add(1), Ordering::Release);
339 Ok(())
340 }
341}
342
343impl<T: Copy, const N: usize> Drop for Producer<'_, T, N> {
344 fn drop(&mut self) {
345 self.buf.producer_taken.store(false, Ordering::Release);
346 }
347}
348
349impl<T: Copy, const N: usize> core::fmt::Debug for Producer<'_, T, N> {
350 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
351 f.debug_struct("event_buf::Producer")
352 .field("capacity", &N)
353 .finish()
354 }
355}
356
357/// Read handle for an [`EventBuf`].
358///
359/// Dropping the consumer releases the slot so a new one can be created.
360pub struct Consumer<'a, T: Copy, const N: usize> {
361 buf: &'a EventBuf<T, N>,
362 _not_sync: PhantomData<Cell<()>>,
363}
364
365impl<T: Copy, const N: usize> Consumer<'_, T, N> {
366 /// Pop the oldest item from the buffer.
367 ///
368 /// Returns `None` if the buffer is empty.
369 #[inline]
370 pub fn pop(&self) -> Option<T> {
371 let tail = self.buf.tail.load(Ordering::Relaxed);
372 let head = self.buf.head.load(Ordering::Acquire);
373 if tail == head {
374 return None;
375 }
376 let idx = EventBuf::<T, N>::slot_index(tail);
377 // SAFETY: consumer is the only reader of this slot; the producer
378 // will not overwrite it until tail is advanced (Release below).
379 let val = self.buf.slots[idx].with(|slot| unsafe { (*slot).assume_init_read() });
380 self.buf.tail.store(tail.wrapping_add(1), Ordering::Release);
381 Some(val)
382 }
383
384 /// Copy the oldest item without removing it.
385 ///
386 /// Returns `None` if the buffer is empty. The consumer cursor is not
387 /// advanced, so a following [`pop`](Self::pop) returns the same value.
388 #[inline]
389 pub fn peek(&self) -> Option<T> {
390 let tail = self.buf.tail.load(Ordering::Relaxed);
391 let head = self.buf.head.load(Ordering::Acquire);
392 if tail == head {
393 return None;
394 }
395 let idx = EventBuf::<T, N>::slot_index(tail);
396 // SAFETY: same slot exclusivity as `pop` — the producer will not
397 // overwrite this slot until `tail` advances. `T: Copy`, so reading
398 // without advancing leaves a valid value for a later `pop`.
399 Some(self.buf.slots[idx].with(|slot| unsafe { (*slot).assume_init_read() }))
400 }
401
402 /// Drain up to `max` items, passing each to `hook`.
403 ///
404 /// Returns the number of items consumed.
405 #[inline]
406 pub fn drain(&self, max: usize, mut hook: impl FnMut(T)) -> usize {
407 let mut count = 0;
408 while count < max {
409 match self.pop() {
410 Some(val) => {
411 hook(val);
412 count += 1;
413 }
414 None => break,
415 }
416 }
417 count
418 }
419}
420
421impl<T: Copy, const N: usize> Drop for Consumer<'_, T, N> {
422 fn drop(&mut self) {
423 self.buf.consumer_taken.store(false, Ordering::Release);
424 }
425}
426
427impl<T: Copy, const N: usize> core::fmt::Debug for Consumer<'_, T, N> {
428 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
429 f.debug_struct("event_buf::Consumer")
430 .field("capacity", &N)
431 .finish()
432 }
433}
434
435impl<T: Copy, const N: usize> crate::traits::Sink<T> for Producer<'_, T, N> {
436 type Error = T;
437
438 #[inline]
439 fn try_push(&mut self, val: T) -> Result<(), T> {
440 self.push(val)
441 }
442}
443
444impl<T: Copy, const N: usize> crate::traits::Source<T> for Consumer<'_, T, N> {
445 #[inline]
446 fn try_pop(&mut self) -> Option<T> {
447 self.pop()
448 }
449}
450
451#[cfg(test)]
452mod tests {
453 // The deprecated `producer()` / `consumer()` remain public API until 0.3.0,
454 // so these tests are their coverage -- including the two that assert the
455 // panic message. Allowing the lint here rather than at the crate root keeps
456 // the warning live for library code, which is where it should bite.
457 #![allow(deprecated)]
458
459 use super::*;
460
461 #[test]
462 fn new_buf_is_empty() {
463 let buf = EventBuf::<u32, 4>::new();
464 assert!(buf.is_empty());
465 assert!(!buf.is_full());
466 assert_eq!(buf.len(), 0);
467 assert_eq!(buf.capacity(), 4);
468 }
469
470 #[test]
471 fn push_and_pop_fifo() {
472 let buf = EventBuf::<u32, 4>::new();
473 let p = buf.producer();
474 let c = buf.consumer();
475
476 assert!(p.push(10).is_ok());
477 assert!(p.push(20).is_ok());
478 assert!(p.push(30).is_ok());
479
480 assert_eq!(c.pop(), Some(10));
481 assert_eq!(c.pop(), Some(20));
482 assert_eq!(c.pop(), Some(30));
483 assert_eq!(c.pop(), None);
484 }
485
486 #[test]
487 fn push_rejects_when_full() {
488 let buf = EventBuf::<u32, 2>::new();
489 let p = buf.producer();
490 let c = buf.consumer();
491
492 assert!(p.push(1).is_ok());
493 assert!(p.push(2).is_ok());
494 assert_eq!(p.push(3), Err(3)); // full — value returned
495
496 // drain one, then push succeeds
497 assert_eq!(c.pop(), Some(1));
498 assert!(p.push(3).is_ok());
499 }
500
501 #[test]
502 fn drain_returns_count() {
503 let buf = EventBuf::<u32, 8>::new();
504 let p = buf.producer();
505 let c = buf.consumer();
506
507 for i in 0..5 {
508 p.push(i).unwrap();
509 }
510
511 let mut out = std::vec::Vec::new();
512 let n = c.drain(3, |v| out.push(v));
513 assert_eq!(n, 3);
514 assert_eq!(out, [0, 1, 2]);
515
516 // remaining
517 let n = c.drain(100, |v| out.push(v));
518 assert_eq!(n, 2);
519 assert_eq!(out, [0, 1, 2, 3, 4]);
520 }
521
522 #[test]
523 fn drain_on_empty_returns_zero() {
524 let buf = EventBuf::<u32, 4>::new();
525 let _p = buf.producer();
526 let c = buf.consumer();
527
528 let n = c.drain(10, |_| panic!("should not be called"));
529 assert_eq!(n, 0);
530 }
531
532 #[test]
533 fn producer_consumer_can_be_recreated() {
534 let buf = EventBuf::<u32, 4>::new();
535 {
536 let p = buf.producer();
537 p.push(1).unwrap();
538 }
539 // producer dropped — can create a new one
540 let p = buf.producer();
541 p.push(2).unwrap();
542
543 {
544 let c = buf.consumer();
545 assert_eq!(c.pop(), Some(1));
546 }
547 // consumer dropped — can create a new one
548 let c = buf.consumer();
549 assert_eq!(c.pop(), Some(2));
550 assert_eq!(c.pop(), None);
551 }
552
553 #[test]
554 #[should_panic(expected = "only one Producer")]
555 fn double_producer_panics() {
556 let buf = EventBuf::<u32, 4>::new();
557 let _p1 = buf.producer();
558 let _p2 = buf.producer();
559 }
560
561 #[test]
562 #[should_panic(expected = "only one Consumer")]
563 fn double_consumer_panics() {
564 let buf = EventBuf::<u32, 4>::new();
565 let _c1 = buf.consumer();
566 let _c2 = buf.consumer();
567 }
568
569 #[test]
570 fn wraps_around_correctly() {
571 let buf = EventBuf::<u32, 3>::new();
572 let p = buf.producer();
573 let c = buf.consumer();
574
575 // fill, drain, fill again — exercises the wrap
576 for round in 0u32..4 {
577 let base = round * 3;
578 for i in 0..3 {
579 assert!(p.push(base + i).is_ok());
580 }
581 assert_eq!(p.push(99), Err(99)); // full
582 for i in 0..3 {
583 assert_eq!(c.pop(), Some(base + i));
584 }
585 assert_eq!(c.pop(), None); // empty
586 }
587 }
588
589 #[test]
590 fn default_is_new() {
591 let buf: EventBuf<u8, 4> = EventBuf::default();
592 assert!(buf.is_empty());
593 }
594
595 #[test]
596 fn len_and_full_track_state() {
597 let buf = EventBuf::<u32, 3>::new();
598 let p = buf.producer();
599 let c = buf.consumer();
600
601 assert_eq!(buf.len(), 0);
602 assert!(buf.is_empty());
603
604 p.push(1).unwrap();
605 assert_eq!(buf.len(), 1);
606
607 p.push(2).unwrap();
608 p.push(3).unwrap();
609 assert_eq!(buf.len(), 3);
610 assert!(buf.is_full());
611
612 c.pop();
613 assert_eq!(buf.len(), 2);
614 assert!(!buf.is_full());
615 }
616
617 #[test]
618 fn len_stays_within_capacity_while_consumer_drains() {
619 let buf = EventBuf::<u32, 8>::new();
620 let done = AtomicBool::new(false);
621 let pushes = crate::test_support::iterations(200_000);
622
623 std::thread::scope(|scope| {
624 scope.spawn(|| {
625 let p = buf.producer();
626 for i in 0..pushes {
627 let _ = p.push(i);
628 }
629 done.store(true, Ordering::Release);
630 });
631
632 scope.spawn(|| {
633 let c = buf.consumer();
634 while !done.load(Ordering::Acquire) {
635 c.pop();
636 }
637 });
638
639 // `len` races both handles; it may be stale, but it must never
640 // report more than the buffer can hold.
641 while !done.load(Ordering::Acquire) {
642 let observed = buf.len();
643 assert!(
644 observed <= buf.capacity(),
645 "len() reported {observed} for a capacity-{} buffer",
646 buf.capacity()
647 );
648 }
649 });
650 }
651
652 #[test]
653 fn concurrent_spsc_preserves_fifo_and_loses_nothing() {
654 let buf = EventBuf::<u32, 4>::new();
655 let total = crate::test_support::iterations(50_000);
656
657 let received = std::thread::scope(|scope| {
658 scope.spawn(|| {
659 let p = buf.producer();
660 // Backpressure means push can fail; retry so the stream is
661 // complete and any gap in the consumer's view is a real bug.
662 for i in 0..total {
663 let mut val = i;
664 while let Err(rejected) = p.push(val) {
665 val = rejected;
666 std::thread::yield_now();
667 }
668 }
669 });
670
671 let consumer = scope.spawn(|| {
672 let c = buf.consumer();
673 let mut seen = 0u32;
674 while seen < total {
675 match c.pop() {
676 // Strict FIFO: the nth item popped must be n.
677 Some(val) => {
678 assert_eq!(val, seen, "out-of-order pop at index {seen}");
679 seen += 1;
680 }
681 None => std::thread::yield_now(),
682 }
683 }
684 seen
685 });
686
687 consumer.join().unwrap()
688 });
689
690 assert_eq!(received, total);
691 assert_eq!(buf.len(), 0);
692 }
693
694 #[test]
695 fn handles_are_send() {
696 fn assert_send<T: Send>() {}
697 assert_send::<super::Producer<'_, u32, 4>>();
698 assert_send::<super::Consumer<'_, u32, 4>>();
699 }
700
701 #[test]
702 fn try_producer_and_try_consumer() {
703 let buf = EventBuf::<u32, 4>::new();
704 let p = buf.try_producer().expect("first producer");
705 assert!(buf.try_producer().is_none());
706 let c = buf.try_consumer().expect("first consumer");
707 assert!(buf.try_consumer().is_none());
708 p.push(1).unwrap();
709 assert_eq!(c.pop(), Some(1));
710 drop(p);
711 drop(c);
712 assert!(buf.try_producer().is_some());
713 assert!(buf.try_consumer().is_some());
714 }
715
716 #[test]
717 fn peek_copies_without_advancing() {
718 let buf = EventBuf::<u32, 4>::new();
719 let p = buf.producer();
720 let c = buf.consumer();
721
722 assert_eq!(c.peek(), None);
723 p.push(10).unwrap();
724 p.push(20).unwrap();
725 assert_eq!(c.peek(), Some(10));
726 assert_eq!(c.peek(), Some(10));
727 assert_eq!(buf.len(), 2);
728 assert_eq!(c.pop(), Some(10));
729 assert_eq!(c.peek(), Some(20));
730 assert_eq!(c.pop(), Some(20));
731 assert_eq!(c.peek(), None);
732 }
733
734 // Loom's `new` is deliberately non-const, so a `static` init only exists
735 // on the host path.
736 #[cfg(not(loom))]
737 #[test]
738 fn const_new_works_in_const_context() {
739 static BUF: EventBuf<u32, 4> = EventBuf::new();
740 assert!(BUF.is_empty());
741 assert_eq!(BUF.capacity(), 4);
742 }
743
744 // The point of the const `new` is not that a `static` compiles -- it is
745 // that handles borrowed from one are `'static` and `Send`, which is what
746 // lets the producer move into an ISR while the consumer stays in a task
747 // loop. A test that only builds the `static` would still pass if the
748 // lifetime were tied to a local, so pin the signature explicitly.
749 #[cfg(not(loom))]
750 #[test]
751 fn static_buf_yields_static_sendable_handles() {
752 static BUF: EventBuf<u32, 4> = EventBuf::new();
753
754 fn producer_for_isr() -> super::Producer<'static, u32, 4> {
755 BUF.producer()
756 }
757 fn consumer_for_task() -> super::Consumer<'static, u32, 4> {
758 BUF.consumer()
759 }
760 fn assert_send<T: Send>(_: &T) {}
761
762 let p = producer_for_isr();
763 let c = consumer_for_task();
764 assert_send(&p);
765 assert_send(&c);
766
767 p.push(7).unwrap();
768 assert_eq!(c.pop(), Some(7));
769 }
770}