Skip to main content

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}