1//! Notify async tasks or threads.
2//!
3//! This is a synchronization primitive similar to [eventcounts] invented by Dmitry Vyukov.
4//!
5//! You can use this crate to turn non-blocking data structures into async or blocking data
6//! structures. See a [simple mutex] implementation that exposes an async and a blocking interface
7//! for acquiring locks.
8//!
9//! [eventcounts]: https://www.1024cores.net/home/lock-free-algorithms/eventcounts
10//! [simple mutex]: https://github.com/smol-rs/event-listener/blob/master/examples/mutex.rs
11//!
12//! # Examples
13//!
14//! Wait until another thread sets a boolean flag:
15//!
16//! ```
17//! use std::sync::atomic::{AtomicBool, Ordering};
18//! use std::sync::Arc;
19//! use std::thread;
20//! use std::time::Duration;
21//! use std::usize;
22//! use event_listener::{Event, Listener};
23//!
24//! let flag = Arc::new(AtomicBool::new(false));
25//! let event = Arc::new(Event::new());
26//!
27//! // Spawn a thread that will set the flag after 1 second.
28//! thread::spawn({
29//! let flag = flag.clone();
30//! let event = event.clone();
31//! move || {
32//! // Wait for a second.
33//! thread::sleep(Duration::from_secs(1));
34//!
35//! // Set the flag.
36//! flag.store(true, Ordering::SeqCst);
37//!
38//! // Notify all listeners that the flag has been set.
39//! event.notify(usize::MAX);
40//! }
41//! });
42//!
43//! // Wait until the flag is set.
44//! loop {
45//! // Check the flag.
46//! if flag.load(Ordering::SeqCst) {
47//! break;
48//! }
49//!
50//! // Start listening for events.
51//! let mut listener = event.listen();
52//!
53//! // Check the flag again after creating the listener.
54//! if flag.load(Ordering::SeqCst) {
55//! break;
56//! }
57//!
58//! // Wait for a notification and continue the loop.
59//! listener.wait();
60//! }
61//! ```
62//!
63//! # Features
64//!
65//! - The `std` feature (enabled by default) enables the use of the Rust standard library. Disable it for `no_std`
66//! support.
67//!
68//! - The `critical-section` feature enables usage of the [`critical-section`] crate to enable a
69//! more efficient implementation of `event-listener` for `no_std` platforms.
70//!
71//! - The `portable-atomic` feature enables the use of the [`portable-atomic`] crate to provide
72//! atomic operations on platforms that don't support them.
73//!
74//! [`critical-section`]: https://crates.io/crates/critical-section
75//! [`portable-atomic`]: https://crates.io/crates/portable-atomic
76
77#![cfg_attr(not(feature = "std"), no_std)]
78#![allow(clippy::multiple_bound_locations)] // This is a WONTFIX issue with pin-project-lite
79#![warn(missing_docs, missing_debug_implementations, rust_2018_idioms)]
80#![doc(
81 html_favicon_url = "https://raw.githubusercontent.com/smol-rs/smol/master/assets/images/logo_fullsize_transparent.png"
82)]
83#![doc(
84 html_logo_url = "https://raw.githubusercontent.com/smol-rs/smol/master/assets/images/logo_fullsize_transparent.png"
85)]
86
87#[cfg(not(feature = "std"))]
88extern crate alloc;
89#[cfg(feature = "std")]
90extern crate std as alloc;
91
92#[cfg_attr(
93 any(feature = "std", feature = "critical-section"),
94 path = "intrusive.rs"
95)]
96#[cfg_attr(
97 not(any(feature = "std", feature = "critical-section")),
98 path = "slab.rs"
99)]
100mod sys;
101
102mod notify;
103
104#[cfg(not(feature = "std"))]
105use alloc::boxed::Box;
106
107use core::borrow::Borrow;
108use core::fmt;
109use core::future::Future;
110use core::mem::ManuallyDrop;
111use core::pin::Pin;
112use core::ptr;
113use core::task::{Context, Poll, Waker};
114
115#[cfg(all(feature = "std", not(target_family = "wasm")))]
116use {
117 parking::{Parker, Unparker},
118 std::time::{Duration, Instant},
119};
120
121use sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
122use sync::Arc;
123
124#[cfg(not(loom))]
125use sync::WithMut;
126
127use notify::NotificationPrivate;
128pub use notify::{IntoNotification, Notification};
129
130/// Inner state of [`Event`].
131struct Inner<T> {
132 /// The number of notified entries, or `usize::MAX` if all of them have been notified.
133 ///
134 /// If there are no entries, this value is set to `usize::MAX`.
135 notified: AtomicUsize,
136
137 /// Inner queue of event listeners.
138 ///
139 /// On `std` platforms, this is an intrusive linked list. On `no_std` platforms, this is a
140 /// more traditional `Vec` of listeners, with an atomic queue used as a backup for high
141 /// contention.
142 list: sys::List<T>,
143}
144
145impl<T> Inner<T> {
146 fn new() -> Self {
147 Self {
148 notified: AtomicUsize::new(usize::MAX),
149 list: sys::List::new(),
150 }
151 }
152}
153
154/// A synchronization primitive for notifying async tasks and threads.
155///
156/// Listeners can be registered using [`Event::listen()`]. There are two ways to notify listeners:
157///
158/// 1. [`Event::notify()`] notifies a number of listeners.
159/// 2. [`Event::notify_additional()`] notifies a number of previously unnotified listeners.
160///
161/// If there are no active listeners at the time a notification is sent, it simply gets lost.
162///
163/// There are two ways for a listener to wait for a notification:
164///
165/// 1. In an asynchronous manner using `.await`.
166/// 2. In a blocking manner by calling [`EventListener::wait()`] on it.
167///
168/// If a notified listener is dropped without receiving a notification, dropping will notify
169/// another active listener. Whether one *additional* listener will be notified depends on what
170/// kind of notification was delivered.
171///
172/// Listeners are registered and notified in the first-in first-out fashion, ensuring fairness.
173pub struct Event<T = ()> {
174 /// A pointer to heap-allocated inner state.
175 ///
176 /// This pointer is initially null and gets lazily initialized on first use. Semantically, it
177 /// is an `Arc<Inner>` so it's important to keep in mind that it contributes to the [`Arc`]'s
178 /// reference count.
179 inner: AtomicPtr<Inner<T>>,
180}
181
182unsafe impl<T: Send> Send for Event<T> {}
183unsafe impl<T: Send> Sync for Event<T> {}
184
185impl<T> core::panic::UnwindSafe for Event<T> {}
186impl<T> core::panic::RefUnwindSafe for Event<T> {}
187
188impl<T> fmt::Debug for Event<T> {
189 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
190 match self.try_inner() {
191 Some(inner) => {
192 let notified_count = inner.notified.load(Ordering::Relaxed);
193 let total_count = match inner.list.try_total_listeners() {
194 Some(total_count) => total_count,
195 None => {
196 return f
197 .debug_tuple("Event")
198 .field(&format_args!("<locked>"))
199 .finish()
200 }
201 };
202
203 f.debug_struct("Event")
204 .field("listeners_notified", &notified_count)
205 .field("listeners_total", &total_count)
206 .finish()
207 }
208 None => f
209 .debug_tuple("Event")
210 .field(&format_args!("<uninitialized>"))
211 .finish(),
212 }
213 }
214}
215
216impl Default for Event {
217 #[inline]
218 fn default() -> Self {
219 Self::new()
220 }
221}
222
223impl<T> Event<T> {
224 /// Creates a new `Event` with a tag type.
225 ///
226 /// Tagging cannot be implemented efficiently on `no_std`, so this is only available when the
227 /// `std` feature is enabled.
228 ///
229 /// # Examples
230 ///
231 /// ```
232 /// use event_listener::Event;
233 ///
234 /// let event = Event::<usize>::with_tag();
235 /// ```
236 #[cfg(all(feature = "std", not(loom)))]
237 #[inline]
238 pub const fn with_tag() -> Self {
239 Self {
240 inner: AtomicPtr::new(ptr::null_mut()),
241 }
242 }
243 #[cfg(all(feature = "std", loom))]
244 #[inline]
245 pub fn with_tag() -> Self {
246 Self {
247 inner: AtomicPtr::new(ptr::null_mut()),
248 }
249 }
250
251 /// Tell whether any listeners are currently notified.
252 ///
253 /// # Examples
254 ///
255 /// ```
256 /// use event_listener::{Event, Listener};
257 ///
258 /// let event = Event::new();
259 /// let listener = event.listen();
260 /// assert!(!event.is_notified());
261 ///
262 /// event.notify(1);
263 /// assert!(event.is_notified());
264 /// ```
265 #[inline]
266 pub fn is_notified(&self) -> bool {
267 self.try_inner()
268 .map_or(false, |inner| inner.notified.load(Ordering::Acquire) > 0)
269 }
270
271 /// Returns a guard listening for a notification.
272 ///
273 /// This method emits a `SeqCst` fence after registering a listener. For now, this method
274 /// is an alias for calling [`EventListener::new()`], pinning it to the heap, and then
275 /// inserting it into a list.
276 ///
277 /// # Examples
278 ///
279 /// ```
280 /// use event_listener::Event;
281 ///
282 /// let event = Event::new();
283 /// let listener = event.listen();
284 /// ```
285 ///
286 /// # Caveats
287 ///
288 /// The above example is equivalent to this code:
289 ///
290 /// ```no_compile
291 /// use event_listener::{Event, EventListener};
292 ///
293 /// let event = Event::new();
294 /// let mut listener = Box::pin(EventListener::new());
295 /// listener.listen(&event);
296 /// ```
297 ///
298 /// It creates a new listener, pins it to the heap, and inserts it into the linked list
299 /// of listeners. While this type of usage is simple, it may be desired to eliminate this
300 /// heap allocation. In this case, consider using the [`EventListener::new`] constructor
301 /// directly, which allows for greater control over where the [`EventListener`] is
302 /// allocated. However, users of this `new` method must be careful to ensure that the
303 /// [`EventListener`] is `listen`ing before waiting on it; panics may occur otherwise.
304 #[cold]
305 pub fn listen(&self) -> EventListener<T> {
306 let inner = ManuallyDrop::new(unsafe { Arc::from_raw(self.inner()) });
307
308 // Allocate the listener on the heap and insert it.
309 let mut listener = Box::pin(InnerListener {
310 event: Arc::clone(&inner),
311 listener: None,
312 });
313 listener.as_mut().listen();
314
315 // Return the listener.
316 EventListener { listener }
317 }
318
319 /// Notifies a number of active listeners.
320 ///
321 /// The number is allowed to be zero or exceed the current number of listeners.
322 ///
323 /// The [`Notification`] trait is used to define what kind of notification is delivered.
324 /// The default implementation (implemented on `usize`) is a notification that only notifies
325 /// *at least* the specified number of listeners.
326 ///
327 /// In certain cases, this function emits a `SeqCst` fence before notifying listeners.
328 ///
329 /// This function returns the number of [`EventListener`]s that were notified by this call.
330 ///
331 /// # Caveats
332 ///
333 /// If the `std` feature is disabled, the notification will be delayed under high contention,
334 /// such as when another thread is taking a while to `notify` the event. In this circumstance,
335 /// this function will return `0` instead of the number of listeners actually notified. Therefore
336 /// if the `std` feature is disabled the return value of this function should not be relied upon
337 /// for soundness and should be used only as a hint.
338 ///
339 /// If the `std` feature is enabled, no spurious returns are possible, since the `std`
340 /// implementation uses system locking primitives to ensure there is no unavoidable
341 /// contention.
342 ///
343 /// # Examples
344 ///
345 /// Use the default notification strategy:
346 ///
347 /// ```
348 /// use event_listener::Event;
349 ///
350 /// let event = Event::new();
351 ///
352 /// // This notification gets lost because there are no listeners.
353 /// event.notify(1);
354 ///
355 /// let listener1 = event.listen();
356 /// let listener2 = event.listen();
357 /// let listener3 = event.listen();
358 ///
359 /// // Notifies two listeners.
360 /// //
361 /// // Listener queueing is fair, which means `listener1` and `listener2`
362 /// // get notified here since they start listening before `listener3`.
363 /// event.notify(2);
364 /// ```
365 ///
366 /// Notify without emitting a `SeqCst` fence. This uses the [`relaxed`] notification strategy.
367 /// This is equivalent to calling [`Event::notify_relaxed()`].
368 ///
369 /// [`relaxed`]: IntoNotification::relaxed
370 ///
371 /// ```
372 /// use event_listener::{IntoNotification, Event};
373 /// use std::sync::atomic::{self, Ordering};
374 ///
375 /// let event = Event::new();
376 ///
377 /// // This notification gets lost because there are no listeners.
378 /// event.notify(1.relaxed());
379 ///
380 /// let listener1 = event.listen();
381 /// let listener2 = event.listen();
382 /// let listener3 = event.listen();
383 ///
384 /// // We should emit a fence manually when using relaxed notifications.
385 /// atomic::fence(Ordering::SeqCst);
386 ///
387 /// // Notifies two listeners.
388 /// //
389 /// // Listener queueing is fair, which means `listener1` and `listener2`
390 /// // get notified here since they start listening before `listener3`.
391 /// event.notify(2.relaxed());
392 /// ```
393 ///
394 /// Notify additional listeners. In contrast to [`Event::notify()`], this method will notify `n`
395 /// *additional* listeners that were previously unnotified. This uses the [`additional`]
396 /// notification strategy. This is equivalent to calling [`Event::notify_additional()`].
397 ///
398 /// [`additional`]: IntoNotification::additional
399 ///
400 /// ```
401 /// use event_listener::{IntoNotification, Event};
402 ///
403 /// let event = Event::new();
404 ///
405 /// // This notification gets lost because there are no listeners.
406 /// event.notify(1.additional());
407 ///
408 /// let listener1 = event.listen();
409 /// let listener2 = event.listen();
410 /// let listener3 = event.listen();
411 ///
412 /// // Notifies two listeners.
413 /// //
414 /// // Listener queueing is fair, which means `listener1` and `listener2`
415 /// // get notified here since they start listening before `listener3`.
416 /// event.notify(1.additional());
417 /// event.notify(1.additional());
418 /// ```
419 ///
420 /// Notifies with the [`additional`] and [`relaxed`] strategies at the same time. This is
421 /// equivalent to calling [`Event::notify_additional_relaxed()`].
422 ///
423 /// ```
424 /// use event_listener::{IntoNotification, Event};
425 /// use std::sync::atomic::{self, Ordering};
426 ///
427 /// let event = Event::new();
428 ///
429 /// // This notification gets lost because there are no listeners.
430 /// event.notify(1.additional().relaxed());
431 ///
432 /// let listener1 = event.listen();
433 /// let listener2 = event.listen();
434 /// let listener3 = event.listen();
435 ///
436 /// // We should emit a fence manually when using relaxed notifications.
437 /// atomic::fence(Ordering::SeqCst);
438 ///
439 /// // Notifies two listeners.
440 /// //
441 /// // Listener queueing is fair, which means `listener1` and `listener2`
442 /// // get notified here since they start listening before `listener3`.
443 /// event.notify(1.additional().relaxed());
444 /// event.notify(1.additional().relaxed());
445 /// ```
446 #[inline]
447 pub fn notify(&self, notify: impl IntoNotification<Tag = T>) -> usize {
448 let notify = notify.into_notification();
449
450 // Make sure the notification comes after whatever triggered it.
451 notify.fence(notify::Internal::new());
452
453 let inner = unsafe { &*self.inner() };
454 inner.notify(notify)
455 }
456
457 /// Return a reference to the inner state if it has been initialized.
458 #[inline]
459 fn try_inner(&self) -> Option<&Inner<T>> {
460 let inner = self.inner.load(Ordering::Acquire);
461 unsafe { inner.as_ref() }
462 }
463
464 /// Returns a raw, initialized pointer to the inner state.
465 ///
466 /// This returns a raw pointer instead of reference because `from_raw`
467 /// requires raw/mut provenance: <https://github.com/rust-lang/rust/pull/67339>.
468 fn inner(&self) -> *const Inner<T> {
469 let mut inner = self.inner.load(Ordering::Acquire);
470
471 // If this is the first use, initialize the state.
472 if inner.is_null() {
473 // Allocate the state on the heap.
474 let new = Arc::new(Inner::<T>::new());
475
476 // Convert the state to a raw pointer.
477 let new = Arc::into_raw(new) as *mut Inner<T>;
478
479 // Replace the null pointer with the new state pointer.
480 inner = self
481 .inner
482 .compare_exchange(inner, new, Ordering::AcqRel, Ordering::Acquire)
483 .unwrap_or_else(|x| x);
484
485 // Check if the old pointer value was indeed null.
486 if inner.is_null() {
487 // If yes, then use the new state pointer.
488 inner = new;
489 } else {
490 // If not, that means a concurrent operation has initialized the state.
491 // In that case, use the old pointer and deallocate the new one.
492 unsafe {
493 drop(Arc::from_raw(new));
494 }
495 }
496 }
497
498 inner
499 }
500
501 /// Get the number of listeners currently listening to this [`Event`].
502 ///
503 /// This call returns the number of [`EventListener`]s that are currently listening to
504 /// this event. It does this by acquiring the internal event lock and reading the listener
505 /// count. Therefore it is only available for `std`-enabled platforms.
506 ///
507 /// # Caveats
508 ///
509 /// This function returns just a snapshot of the number of listeners at this point in time.
510 /// Due to the nature of multi-threaded CPUs, it is possible that this number will be
511 /// inaccurate by the time that this function returns.
512 ///
513 /// It is possible for the actual number to change at any point. Therefore, the number should
514 /// only ever be used as a hint.
515 ///
516 /// # Examples
517 ///
518 /// ```
519 /// use event_listener::Event;
520 ///
521 /// let event = Event::new();
522 ///
523 /// assert_eq!(event.total_listeners(), 0);
524 ///
525 /// let listener1 = event.listen();
526 /// assert_eq!(event.total_listeners(), 1);
527 ///
528 /// let listener2 = event.listen();
529 /// assert_eq!(event.total_listeners(), 2);
530 ///
531 /// drop(listener1);
532 /// drop(listener2);
533 /// assert_eq!(event.total_listeners(), 0);
534 /// ```
535 #[cfg(feature = "std")]
536 #[inline]
537 pub fn total_listeners(&self) -> usize {
538 if let Some(inner) = self.try_inner() {
539 inner.list.total_listeners()
540 } else {
541 0
542 }
543 }
544}
545
546impl Event<()> {
547 /// Creates a new [`Event`].
548 ///
549 /// # Examples
550 ///
551 /// ```
552 /// use event_listener::Event;
553 ///
554 /// let event = Event::new();
555 /// ```
556 #[inline]
557 #[cfg(not(loom))]
558 pub const fn new() -> Self {
559 Self {
560 inner: AtomicPtr::new(ptr::null_mut()),
561 }
562 }
563
564 #[inline]
565 #[cfg(loom)]
566 pub fn new() -> Self {
567 Self {
568 inner: AtomicPtr::new(ptr::null_mut()),
569 }
570 }
571
572 /// Notifies a number of active listeners without emitting a `SeqCst` fence.
573 ///
574 /// The number is allowed to be zero or exceed the current number of listeners.
575 ///
576 /// In contrast to [`Event::notify_additional()`], this method only makes sure *at least* `n`
577 /// listeners among the active ones are notified.
578 ///
579 /// Unlike [`Event::notify()`], this method does not emit a `SeqCst` fence.
580 ///
581 /// This method only works for untagged events. In other cases, it is recommended to instead
582 /// use [`Event::notify()`] like so:
583 ///
584 /// ```
585 /// use event_listener::{IntoNotification, Event};
586 /// let event = Event::new();
587 ///
588 /// // Old way:
589 /// event.notify_relaxed(1);
590 ///
591 /// // New way:
592 /// event.notify(1.relaxed());
593 /// ```
594 ///
595 /// # Examples
596 ///
597 /// ```
598 /// use event_listener::{Event, IntoNotification};
599 /// use std::sync::atomic::{self, Ordering};
600 ///
601 /// let event = Event::new();
602 ///
603 /// // This notification gets lost because there are no listeners.
604 /// event.notify_relaxed(1);
605 ///
606 /// let listener1 = event.listen();
607 /// let listener2 = event.listen();
608 /// let listener3 = event.listen();
609 ///
610 /// // We should emit a fence manually when using relaxed notifications.
611 /// atomic::fence(Ordering::SeqCst);
612 ///
613 /// // Notifies two listeners.
614 /// //
615 /// // Listener queueing is fair, which means `listener1` and `listener2`
616 /// // get notified here since they start listening before `listener3`.
617 /// event.notify_relaxed(2);
618 /// ```
619 #[inline]
620 pub fn notify_relaxed(&self, n: usize) -> usize {
621 self.notify(n.relaxed())
622 }
623
624 /// Notifies a number of active and still unnotified listeners.
625 ///
626 /// The number is allowed to be zero or exceed the current number of listeners.
627 ///
628 /// In contrast to [`Event::notify()`], this method will notify `n` *additional* listeners that
629 /// were previously unnotified.
630 ///
631 /// This method emits a `SeqCst` fence before notifying listeners.
632 ///
633 /// This method only works for untagged events. In other cases, it is recommended to instead
634 /// use [`Event::notify()`] like so:
635 ///
636 /// ```
637 /// use event_listener::{IntoNotification, Event};
638 /// let event = Event::new();
639 ///
640 /// // Old way:
641 /// event.notify_additional(1);
642 ///
643 /// // New way:
644 /// event.notify(1.additional());
645 /// ```
646 ///
647 /// # Examples
648 ///
649 /// ```
650 /// use event_listener::Event;
651 ///
652 /// let event = Event::new();
653 ///
654 /// // This notification gets lost because there are no listeners.
655 /// event.notify_additional(1);
656 ///
657 /// let listener1 = event.listen();
658 /// let listener2 = event.listen();
659 /// let listener3 = event.listen();
660 ///
661 /// // Notifies two listeners.
662 /// //
663 /// // Listener queueing is fair, which means `listener1` and `listener2`
664 /// // get notified here since they start listening before `listener3`.
665 /// event.notify_additional(1);
666 /// event.notify_additional(1);
667 /// ```
668 #[inline]
669 pub fn notify_additional(&self, n: usize) -> usize {
670 self.notify(n.additional())
671 }
672
673 /// Notifies a number of active and still unnotified listeners without emitting a `SeqCst`
674 /// fence.
675 ///
676 /// The number is allowed to be zero or exceed the current number of listeners.
677 ///
678 /// In contrast to [`Event::notify()`], this method will notify `n` *additional* listeners that
679 /// were previously unnotified.
680 ///
681 /// Unlike [`Event::notify_additional()`], this method does not emit a `SeqCst` fence.
682 ///
683 /// This method only works for untagged events. In other cases, it is recommended to instead
684 /// use [`Event::notify()`] like so:
685 ///
686 /// ```
687 /// use event_listener::{IntoNotification, Event};
688 /// let event = Event::new();
689 ///
690 /// // Old way:
691 /// event.notify_additional_relaxed(1);
692 ///
693 /// // New way:
694 /// event.notify(1.additional().relaxed());
695 /// ```
696 ///
697 /// # Examples
698 ///
699 /// ```
700 /// use event_listener::Event;
701 /// use std::sync::atomic::{self, Ordering};
702 ///
703 /// let event = Event::new();
704 ///
705 /// // This notification gets lost because there are no listeners.
706 /// event.notify(1);
707 ///
708 /// let listener1 = event.listen();
709 /// let listener2 = event.listen();
710 /// let listener3 = event.listen();
711 ///
712 /// // We should emit a fence manually when using relaxed notifications.
713 /// atomic::fence(Ordering::SeqCst);
714 ///
715 /// // Notifies two listeners.
716 /// //
717 /// // Listener queueing is fair, which means `listener1` and `listener2`
718 /// // get notified here since they start listening before `listener3`.
719 /// event.notify_additional_relaxed(1);
720 /// event.notify_additional_relaxed(1);
721 /// ```
722 #[inline]
723 pub fn notify_additional_relaxed(&self, n: usize) -> usize {
724 self.notify(n.additional().relaxed())
725 }
726}
727
728impl<T> Drop for Event<T> {
729 #[inline]
730 fn drop(&mut self) {
731 self.inner.with_mut(|&mut inner: *mut Inner| {
732 // If the state pointer has been initialized, drop it.
733 if !inner.is_null() {
734 unsafe {
735 drop(Arc::from_raw(ptr:inner));
736 }
737 }
738 })
739 }
740}
741
742/// A handle that is listening to an [`Event`].
743///
744/// This trait represents a type waiting for a notification from an [`Event`]. See the
745/// [`EventListener`] type for more documentation on this trait's usage.
746pub trait Listener<T = ()>: Future<Output = T> + __sealed::Sealed {
747 /// Blocks until a notification is received.
748 ///
749 /// # Examples
750 ///
751 /// ```
752 /// use event_listener::{Event, Listener};
753 ///
754 /// let event = Event::new();
755 /// let mut listener = event.listen();
756 ///
757 /// // Notify `listener`.
758 /// event.notify(1);
759 ///
760 /// // Receive the notification.
761 /// listener.wait();
762 /// ```
763 #[cfg(all(feature = "std", not(target_family = "wasm")))]
764 fn wait(self) -> T;
765
766 /// Blocks until a notification is received or a timeout is reached.
767 ///
768 /// Returns `true` if a notification was received.
769 ///
770 /// # Examples
771 ///
772 /// ```
773 /// use std::time::Duration;
774 /// use event_listener::{Event, Listener};
775 ///
776 /// let event = Event::new();
777 /// let mut listener = event.listen();
778 ///
779 /// // There are no notification so this times out.
780 /// assert!(listener.wait_timeout(Duration::from_secs(1)).is_none());
781 /// ```
782 #[cfg(all(feature = "std", not(target_family = "wasm")))]
783 fn wait_timeout(self, timeout: Duration) -> Option<T>;
784
785 /// Blocks until a notification is received or a deadline is reached.
786 ///
787 /// Returns `true` if a notification was received.
788 ///
789 /// # Examples
790 ///
791 /// ```
792 /// use std::time::{Duration, Instant};
793 /// use event_listener::{Event, Listener};
794 ///
795 /// let event = Event::new();
796 /// let mut listener = event.listen();
797 ///
798 /// // There are no notification so this times out.
799 /// assert!(listener.wait_deadline(Instant::now() + Duration::from_secs(1)).is_none());
800 /// ```
801 #[cfg(all(feature = "std", not(target_family = "wasm")))]
802 fn wait_deadline(self, deadline: Instant) -> Option<T>;
803
804 /// Drops this listener and discards its notification (if any) without notifying another
805 /// active listener.
806 ///
807 /// Returns `true` if a notification was discarded.
808 ///
809 /// # Examples
810 ///
811 /// ```
812 /// use event_listener::{Event, Listener};
813 ///
814 /// let event = Event::new();
815 /// let mut listener1 = event.listen();
816 /// let mut listener2 = event.listen();
817 ///
818 /// event.notify(1);
819 ///
820 /// assert!(listener1.discard());
821 /// assert!(!listener2.discard());
822 /// ```
823 fn discard(self) -> bool;
824
825 /// Returns `true` if this listener listens to the given `Event`.
826 ///
827 /// # Examples
828 ///
829 /// ```
830 /// use event_listener::{Event, Listener};
831 ///
832 /// let event = Event::new();
833 /// let listener = event.listen();
834 ///
835 /// assert!(listener.listens_to(&event));
836 /// ```
837 fn listens_to(&self, event: &Event<T>) -> bool;
838
839 /// Returns `true` if both listeners listen to the same `Event`.
840 ///
841 /// # Examples
842 ///
843 /// ```
844 /// use event_listener::{Event, Listener};
845 ///
846 /// let event = Event::new();
847 /// let listener1 = event.listen();
848 /// let listener2 = event.listen();
849 ///
850 /// assert!(listener1.same_event(&listener2));
851 /// ```
852 fn same_event(&self, other: &Self) -> bool;
853}
854
855/// Implement the `Listener` trait using the underlying `InnerListener`.
856macro_rules! forward_impl_to_listener {
857 ($gen:ident => $ty:ty) => {
858 impl<$gen> crate::Listener<$gen> for $ty {
859 #[cfg(all(feature = "std", not(target_family = "wasm")))]
860 fn wait(mut self) -> $gen {
861 self.listener_mut().wait_internal(None).unwrap()
862 }
863
864 #[cfg(all(feature = "std", not(target_family = "wasm")))]
865 fn wait_timeout(mut self, timeout: std::time::Duration) -> Option<$gen> {
866 self.listener_mut()
867 .wait_internal(std::time::Instant::now().checked_add(timeout))
868 }
869
870 #[cfg(all(feature = "std", not(target_family = "wasm")))]
871 fn wait_deadline(mut self, deadline: std::time::Instant) -> Option<$gen> {
872 self.listener_mut().wait_internal(Some(deadline))
873 }
874
875 fn discard(mut self) -> bool {
876 self.listener_mut().discard()
877 }
878
879 #[inline]
880 fn listens_to(&self, event: &Event<$gen>) -> bool {
881 core::ptr::eq::<Inner<$gen>>(
882 &*self.listener().event,
883 event.inner.load(core::sync::atomic::Ordering::Acquire),
884 )
885 }
886
887 #[inline]
888 fn same_event(&self, other: &$ty) -> bool {
889 core::ptr::eq::<Inner<$gen>>(&*self.listener().event, &*other.listener().event)
890 }
891 }
892
893 impl<$gen> Future for $ty {
894 type Output = $gen;
895
896 #[inline]
897 fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<$gen> {
898 self.listener_mut().poll_internal(cx)
899 }
900 }
901 };
902}
903
904/// A guard waiting for a notification from an [`Event`].
905///
906/// There are two ways for a listener to wait for a notification:
907///
908/// 1. In an asynchronous manner using `.await`.
909/// 2. In a blocking manner by calling [`EventListener::wait()`] on it.
910///
911/// If a notified listener is dropped without receiving a notification, dropping will notify
912/// another active listener. Whether one *additional* listener will be notified depends on what
913/// kind of notification was delivered.
914///
915/// See the [`Listener`] trait for the functionality exposed by this type.
916///
917/// This structure allocates the listener on the heap.
918pub struct EventListener<T = ()> {
919 listener: Pin<Box<InnerListener<T, Arc<Inner<T>>>>>,
920}
921
922unsafe impl<T: Send> Send for EventListener<T> {}
923unsafe impl<T: Send> Sync for EventListener<T> {}
924
925impl<T> core::panic::UnwindSafe for EventListener<T> {}
926impl<T> core::panic::RefUnwindSafe for EventListener<T> {}
927impl<T> Unpin for EventListener<T> {}
928
929impl<T> fmt::Debug for EventListener<T> {
930 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
931 f.debug_struct(name:"EventListener").finish_non_exhaustive()
932 }
933}
934
935impl<T> EventListener<T> {
936 #[inline]
937 fn listener(&self) -> &InnerListener<T, Arc<Inner<T>>> {
938 &self.listener
939 }
940
941 #[inline]
942 fn listener_mut(&mut self) -> Pin<&mut InnerListener<T, Arc<Inner<T>>>> {
943 self.listener.as_mut()
944 }
945}
946
947forward_impl_to_listener! { T => EventListener<T> }
948
949/// Create a stack-based event listener for an [`Event`].
950///
951/// [`EventListener`] allocates the listener on the heap. While this works for most use cases, in
952/// practice this heap allocation can be expensive for repeated uses. This method allows for
953/// allocating the listener on the stack instead.
954///
955/// There are limitations to using this macro instead of the [`EventListener`] type, however.
956/// Firstly, it is significantly less flexible. The listener is locked to the current stack
957/// frame, meaning that it can't be returned or put into a place where it would go out of
958/// scope. For instance, this will not work:
959///
960/// ```compile_fail
961/// use event_listener::{Event, Listener, listener};
962///
963/// fn get_listener(event: &Event) -> impl Listener {
964/// listener!(event => cant_return_this);
965/// cant_return_this
966/// }
967/// ```
968///
969/// In addition, the types involved in creating this listener are not able to be named. Therefore
970/// it cannot be used in hand-rolled futures or similar structures.
971///
972/// The type created by this macro implements [`Listener`], allowing it to be used in cases where
973/// [`EventListener`] would normally be used.
974///
975/// ## Example
976///
977/// To use this macro, replace cases where you would normally use this...
978///
979/// ```no_compile
980/// let listener = event.listen();
981/// ```
982///
983/// ...with this:
984///
985/// ```no_compile
986/// listener!(event => listener);
987/// ```
988///
989/// Here is the top level example from this crate's documentation, but using [`listener`] instead
990/// of [`EventListener`].
991///
992/// ```
993/// use std::sync::atomic::{AtomicBool, Ordering};
994/// use std::sync::Arc;
995/// use std::thread;
996/// use std::time::Duration;
997/// use std::usize;
998/// use event_listener::{Event, listener, IntoNotification, Listener};
999///
1000/// let flag = Arc::new(AtomicBool::new(false));
1001/// let event = Arc::new(Event::new());
1002///
1003/// // Spawn a thread that will set the flag after 1 second.
1004/// thread::spawn({
1005/// let flag = flag.clone();
1006/// let event = event.clone();
1007/// move || {
1008/// // Wait for a second.
1009/// thread::sleep(Duration::from_secs(1));
1010///
1011/// // Set the flag.
1012/// flag.store(true, Ordering::SeqCst);
1013///
1014/// // Notify all listeners that the flag has been set.
1015/// event.notify(usize::MAX);
1016/// }
1017/// });
1018///
1019/// // Wait until the flag is set.
1020/// loop {
1021/// // Check the flag.
1022/// if flag.load(Ordering::SeqCst) {
1023/// break;
1024/// }
1025///
1026/// // Start listening for events.
1027/// // NEW: Changed to a stack-based listener.
1028/// listener!(event => listener);
1029///
1030/// // Check the flag again after creating the listener.
1031/// if flag.load(Ordering::SeqCst) {
1032/// break;
1033/// }
1034///
1035/// // Wait for a notification and continue the loop.
1036/// listener.wait();
1037/// }
1038/// ```
1039#[macro_export]
1040macro_rules! listener {
1041 ($event:expr => $listener:ident) => {
1042 let mut $listener = $crate::__private::StackSlot::new(&$event);
1043 // SAFETY: We shadow $listener so it can't be moved after.
1044 let mut $listener = unsafe { $crate::__private::Pin::new_unchecked(&mut $listener) };
1045 #[allow(unused_mut)]
1046 let mut $listener = $listener.listen();
1047 };
1048}
1049
1050pin_project_lite::pin_project! {
1051 #[project(!Unpin)]
1052 #[project = ListenerProject]
1053 struct InnerListener<T, B: Borrow<Inner<T>>>
1054 where
1055 B: Unpin,
1056 {
1057 // The reference to the original event.
1058 event: B,
1059
1060 // The inner state of the listener.
1061 //
1062 // This is only ever `None` during initialization. After `listen()` has completed, this
1063 // should be `Some`.
1064 #[pin]
1065 listener: Option<sys::Listener<T>>,
1066 }
1067
1068 impl<T, B: Borrow<Inner<T>>> PinnedDrop for InnerListener<T, B>
1069 where
1070 B: Unpin,
1071 {
1072 fn drop(mut this: Pin<&mut Self>) {
1073 // If we're being dropped, we need to remove ourself from the list.
1074 let this = this.project();
1075 (*this.event).borrow().remove(this.listener, true);
1076 }
1077 }
1078}
1079
1080unsafe impl<T: Send, B: Borrow<Inner<T>> + Unpin + Send> Send for InnerListener<T, B> {}
1081unsafe impl<T: Send, B: Borrow<Inner<T>> + Unpin + Sync> Sync for InnerListener<T, B> {}
1082
1083impl<T, B: Borrow<Inner<T>> + Unpin> InnerListener<T, B> {
1084 /// Insert this listener into the linked list.
1085 #[inline]
1086 fn listen(self: Pin<&mut Self>) {
1087 let this = self.project();
1088 (*this.event).borrow().insert(this.listener);
1089 }
1090
1091 /// Wait until the provided deadline.
1092 #[cfg(all(feature = "std", not(target_family = "wasm")))]
1093 fn wait_internal(mut self: Pin<&mut Self>, deadline: Option<Instant>) -> Option<T> {
1094 fn parker_and_task() -> (Parker, Task) {
1095 let parker = Parker::new();
1096 let unparker = parker.unparker();
1097 (parker, Task::Unparker(unparker))
1098 }
1099
1100 crate::sync::thread_local! {
1101 /// Cached thread-local parker/unparker pair.
1102 static PARKER: (Parker, Task) = parker_and_task();
1103 }
1104
1105 // Try to borrow the thread-local parker/unparker pair.
1106 PARKER
1107 .try_with({
1108 let this = self.as_mut();
1109 |(parker, unparker)| this.wait_with_parker(deadline, parker, unparker.as_task_ref())
1110 })
1111 .unwrap_or_else(|_| {
1112 // If the pair isn't accessible, we may be being called in a destructor.
1113 // Just create a new pair.
1114 let (parker, unparker) = parking::pair();
1115 self.as_mut()
1116 .wait_with_parker(deadline, &parker, TaskRef::Unparker(&unparker))
1117 })
1118 }
1119
1120 /// Wait until the provided deadline using the specified parker/unparker pair.
1121 #[cfg(all(feature = "std", not(target_family = "wasm")))]
1122 fn wait_with_parker(
1123 self: Pin<&mut Self>,
1124 deadline: Option<Instant>,
1125 parker: &Parker,
1126 unparker: TaskRef<'_>,
1127 ) -> Option<T> {
1128 let mut this = self.project();
1129 let inner = (*this.event).borrow();
1130
1131 // Set the listener's state to `Task`.
1132 if let Some(tag) = inner.register(this.listener.as_mut(), unparker).notified() {
1133 // We were already notified, so we don't need to park.
1134 return Some(tag);
1135 }
1136
1137 // Wait until a notification is received or the timeout is reached.
1138 loop {
1139 match deadline {
1140 None => parker.park(),
1141
1142 #[cfg(loom)]
1143 Some(_deadline) => {
1144 panic!("parking does not support timeouts under loom");
1145 }
1146
1147 #[cfg(not(loom))]
1148 Some(deadline) => {
1149 // Make sure we're not timed out already.
1150 let now = Instant::now();
1151 if now >= deadline {
1152 // Remove our entry and check if we were notified.
1153 return inner
1154 .remove(this.listener.as_mut(), false)
1155 .expect("We never removed ourself from the list")
1156 .notified();
1157 }
1158 parker.park_deadline(deadline);
1159 }
1160 }
1161
1162 // See if we were notified.
1163 if let Some(tag) = inner.register(this.listener.as_mut(), unparker).notified() {
1164 return Some(tag);
1165 }
1166 }
1167 }
1168
1169 /// Drops this listener and discards its notification (if any) without notifying another
1170 /// active listener.
1171 fn discard(self: Pin<&mut Self>) -> bool {
1172 let this = self.project();
1173 (*this.event)
1174 .borrow()
1175 .remove(this.listener, false)
1176 .map_or(false, |state| state.is_notified())
1177 }
1178
1179 /// Poll this listener for a notification.
1180 fn poll_internal(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
1181 let this = self.project();
1182 let inner = (*this.event).borrow();
1183
1184 // Try to register the listener.
1185 match inner
1186 .register(this.listener, TaskRef::Waker(cx.waker()))
1187 .notified()
1188 {
1189 Some(tag) => {
1190 // We were already notified, so we don't need to park.
1191 Poll::Ready(tag)
1192 }
1193
1194 None => {
1195 // We're now waiting for a notification.
1196 Poll::Pending
1197 }
1198 }
1199 }
1200}
1201
1202/// The state of a listener.
1203#[derive(PartialEq)]
1204enum State<T> {
1205 /// The listener was just created.
1206 Created,
1207
1208 /// The listener has received a notification.
1209 ///
1210 /// The `bool` is `true` if this was an "additional" notification.
1211 Notified {
1212 /// Whether or not this is an "additional" notification.
1213 additional: bool,
1214
1215 /// The tag associated with the notification.
1216 tag: T,
1217 },
1218
1219 /// A task is waiting for a notification.
1220 Task(Task),
1221
1222 /// Empty hole used to replace a notified listener.
1223 NotifiedTaken,
1224}
1225
1226impl<T> fmt::Debug for State<T> {
1227 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1228 match self {
1229 Self::Created => f.write_str(data:"Created"),
1230 Self::Notified { additional: &bool, .. } => f&mut DebugStruct<'_, '_>
1231 .debug_struct("Notified")
1232 .field(name:"additional", value:additional)
1233 .finish(),
1234 Self::Task(_) => f.write_str(data:"Task(_)"),
1235 Self::NotifiedTaken => f.write_str(data:"NotifiedTaken"),
1236 }
1237 }
1238}
1239
1240impl<T> State<T> {
1241 fn is_notified(&self) -> bool {
1242 matches!(self, Self::Notified { .. } | Self::NotifiedTaken)
1243 }
1244
1245 /// If this state was notified, return the tag associated with the notification.
1246 #[allow(unused)]
1247 fn notified(self) -> Option<T> {
1248 match self {
1249 Self::Notified { tag: T, .. } => Some(tag),
1250 Self::NotifiedTaken => panic!("listener was already notified but taken"),
1251 _ => None,
1252 }
1253 }
1254}
1255
1256/// The result of registering a listener.
1257#[derive(Debug, PartialEq)]
1258enum RegisterResult<T> {
1259 /// The listener was already notified.
1260 Notified(T),
1261
1262 /// The listener has been registered.
1263 Registered,
1264
1265 /// The listener was never inserted into the list.
1266 NeverInserted,
1267}
1268
1269impl<T> RegisterResult<T> {
1270 /// Whether or not the listener was notified.
1271 ///
1272 /// Panics if the listener was never inserted into the list.
1273 fn notified(self) -> Option<T> {
1274 match self {
1275 Self::Notified(tag: T) => Some(tag),
1276 Self::Registered => None,
1277 Self::NeverInserted => panic!("{}", NEVER_INSERTED_PANIC),
1278 }
1279 }
1280}
1281
1282/// A task that can be woken up.
1283#[derive(Debug, Clone)]
1284enum Task {
1285 /// A waker that wakes up a future.
1286 Waker(Waker),
1287
1288 /// An unparker that wakes up a thread.
1289 #[cfg(all(feature = "std", not(target_family = "wasm")))]
1290 Unparker(Unparker),
1291}
1292
1293impl Task {
1294 fn as_task_ref(&self) -> TaskRef<'_> {
1295 match self {
1296 Self::Waker(waker: &Waker) => TaskRef::Waker(waker),
1297 #[cfg(all(feature = "std", not(target_family = "wasm")))]
1298 Self::Unparker(unparker: &Unparker) => TaskRef::Unparker(unparker),
1299 }
1300 }
1301
1302 fn wake(self) {
1303 match self {
1304 Self::Waker(waker: Waker) => waker.wake(),
1305 #[cfg(all(feature = "std", not(target_family = "wasm")))]
1306 Self::Unparker(unparker: Unparker) => {
1307 unparker.unpark();
1308 }
1309 }
1310 }
1311}
1312
1313impl PartialEq for Task {
1314 fn eq(&self, other: &Self) -> bool {
1315 self.as_task_ref().will_wake(other.as_task_ref())
1316 }
1317}
1318
1319/// A reference to a task.
1320#[derive(Clone, Copy)]
1321enum TaskRef<'a> {
1322 /// A waker that wakes up a future.
1323 Waker(&'a Waker),
1324
1325 /// An unparker that wakes up a thread.
1326 #[cfg(all(feature = "std", not(target_family = "wasm")))]
1327 Unparker(&'a Unparker),
1328}
1329
1330impl TaskRef<'_> {
1331 /// Tells if this task will wake up the other task.
1332 #[allow(unreachable_patterns)]
1333 fn will_wake(self, other: Self) -> bool {
1334 match (self, other) {
1335 (Self::Waker(a: &Waker), Self::Waker(b: &Waker)) => a.will_wake(b),
1336 #[cfg(all(feature = "std", not(target_family = "wasm")))]
1337 (Self::Unparker(_), Self::Unparker(_)) => {
1338 // TODO: Use unreleased will_unpark API.
1339 false
1340 }
1341 _ => false,
1342 }
1343 }
1344
1345 /// Converts this task reference to a task by cloning.
1346 fn into_task(self) -> Task {
1347 match self {
1348 Self::Waker(waker: &Waker) => Task::Waker(waker.clone()),
1349 #[cfg(all(feature = "std", not(target_family = "wasm")))]
1350 Self::Unparker(unparker: &Unparker) => Task::Unparker(unparker.clone()),
1351 }
1352 }
1353}
1354
1355const NEVER_INSERTED_PANIC: &str = "\
1356EventListener was not inserted into the linked list, make sure you're not polling \
1357EventListener/listener! after it has finished";
1358
1359#[cfg(not(loom))]
1360/// Synchronization primitive implementation.
1361mod sync {
1362 #[cfg(not(feature = "portable-atomic"))]
1363 pub(super) use alloc::sync::Arc;
1364 #[cfg(not(feature = "portable-atomic"))]
1365 pub(super) use core::sync::atomic;
1366
1367 #[cfg(feature = "portable-atomic")]
1368 pub(super) use portable_atomic_crate as atomic;
1369 #[cfg(feature = "portable-atomic")]
1370 pub(super) use portable_atomic_util::Arc;
1371
1372 #[allow(unused)]
1373 #[cfg(all(feature = "std", not(feature = "critical-section"), not(loom)))]
1374 pub(super) use std::sync::{Mutex, MutexGuard};
1375 #[cfg(all(feature = "std", not(target_family = "wasm"), not(loom)))]
1376 pub(super) use std::thread_local;
1377
1378 pub(super) trait WithMut {
1379 type Output;
1380
1381 fn with_mut<F, R>(&mut self, f: F) -> R
1382 where
1383 F: FnOnce(&mut Self::Output) -> R;
1384 }
1385
1386 impl<T> WithMut for atomic::AtomicPtr<T> {
1387 type Output = *mut T;
1388
1389 #[inline]
1390 fn with_mut<F, R>(&mut self, f: F) -> R
1391 where
1392 F: FnOnce(&mut Self::Output) -> R,
1393 {
1394 f(self.get_mut())
1395 }
1396 }
1397
1398 pub(crate) mod cell {
1399 pub(crate) use core::cell::Cell;
1400
1401 /// This newtype around *mut T exists for interoperability with loom::cell::ConstPtr,
1402 /// which works as a guard and performs additional logic to track access scope.
1403 pub(crate) struct ConstPtr<T>(*mut T);
1404 impl<T> ConstPtr<T> {
1405 pub(crate) unsafe fn deref(&self) -> &T {
1406 &*self.0
1407 }
1408
1409 #[allow(unused)] // std code does not need this
1410 pub(crate) unsafe fn deref_mut(&mut self) -> &mut T {
1411 &mut *self.0
1412 }
1413 }
1414
1415 /// This UnsafeCell wrapper exists for interoperability with loom::cell::UnsafeCell, and
1416 /// only contains the interface that is needed for this crate.
1417 #[derive(Debug, Default)]
1418 pub(crate) struct UnsafeCell<T>(core::cell::UnsafeCell<T>);
1419
1420 impl<T> UnsafeCell<T> {
1421 pub(crate) fn new(data: T) -> UnsafeCell<T> {
1422 UnsafeCell(core::cell::UnsafeCell::new(data))
1423 }
1424
1425 pub(crate) fn get(&self) -> ConstPtr<T> {
1426 ConstPtr(self.0.get())
1427 }
1428
1429 #[allow(dead_code)] // no_std does not need this
1430 pub(crate) fn into_inner(self) -> T {
1431 self.0.into_inner()
1432 }
1433 }
1434 }
1435}
1436
1437#[cfg(loom)]
1438/// Synchronization primitive implementation.
1439mod sync {
1440 pub(super) use loom::sync::{atomic, Arc, Mutex, MutexGuard};
1441 pub(super) use loom::{cell, thread_local};
1442}
1443
1444fn __test_send_and_sync() {
1445 fn _assert_send<T: Send>() {}
1446 fn _assert_sync<T: Sync>() {}
1447
1448 _assert_send::<crate::__private::StackSlot<'_, ()>>();
1449 _assert_sync::<crate::__private::StackSlot<'_, ()>>();
1450 _assert_send::<crate::__private::StackListener<'_, '_, ()>>();
1451 _assert_sync::<crate::__private::StackListener<'_, '_, ()>>();
1452 _assert_send::<Event<()>>();
1453 _assert_sync::<Event<()>>();
1454 _assert_send::<EventListener<()>>();
1455 _assert_sync::<EventListener<()>>();
1456}
1457
1458#[doc(hidden)]
1459mod __sealed {
1460 use super::{EventListener, __private::StackListener};
1461
1462 pub trait Sealed {}
1463 impl<T> Sealed for EventListener<T> {}
1464 impl<T> Sealed for StackListener<'_, '_, T> {}
1465}
1466
1467/// Semver exempt module.
1468#[doc(hidden)]
1469pub mod __private {
1470 pub use core::pin::Pin;
1471
1472 use super::{Event, Inner, InnerListener};
1473 use core::fmt;
1474 use core::future::Future;
1475 use core::task::{Context, Poll};
1476
1477 pin_project_lite::pin_project! {
1478 /// Space on the stack where a stack-based listener can be allocated.
1479 #[doc(hidden)]
1480 #[project(!Unpin)]
1481 pub struct StackSlot<'ev, T> {
1482 #[pin]
1483 listener: InnerListener<T, &'ev Inner<T>>
1484 }
1485 }
1486
1487 impl<T> fmt::Debug for StackSlot<'_, T> {
1488 #[inline]
1489 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1490 f.debug_struct("StackSlot").finish_non_exhaustive()
1491 }
1492 }
1493
1494 impl<T> core::panic::UnwindSafe for StackSlot<'_, T> {}
1495 impl<T> core::panic::RefUnwindSafe for StackSlot<'_, T> {}
1496 unsafe impl<T> Send for StackSlot<'_, T> {}
1497 unsafe impl<T> Sync for StackSlot<'_, T> {}
1498
1499 impl<'ev, T> StackSlot<'ev, T> {
1500 /// Create a new `StackSlot` on the stack.
1501 #[inline]
1502 #[doc(hidden)]
1503 pub fn new(event: &'ev Event<T>) -> Self {
1504 let inner = unsafe { &*event.inner() };
1505 Self {
1506 listener: InnerListener {
1507 event: inner,
1508 listener: None,
1509 },
1510 }
1511 }
1512
1513 /// Start listening on this `StackSlot`.
1514 #[inline]
1515 #[doc(hidden)]
1516 pub fn listen(mut self: Pin<&mut Self>) -> StackListener<'ev, '_, T> {
1517 // Insert ourselves into the list.
1518 self.as_mut().project().listener.listen();
1519
1520 // We are now listening.
1521 StackListener { slot: self }
1522 }
1523 }
1524
1525 /// A stack-based `EventListener`.
1526 #[doc(hidden)]
1527 pub struct StackListener<'ev, 'stack, T> {
1528 slot: Pin<&'stack mut StackSlot<'ev, T>>,
1529 }
1530
1531 impl<T> core::panic::UnwindSafe for StackListener<'_, '_, T> {}
1532 impl<T> core::panic::RefUnwindSafe for StackListener<'_, '_, T> {}
1533 impl<T> Unpin for StackListener<'_, '_, T> {}
1534
1535 impl<T> fmt::Debug for StackListener<'_, '_, T> {
1536 #[inline]
1537 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1538 f.debug_struct("StackListener").finish_non_exhaustive()
1539 }
1540 }
1541
1542 impl<'ev, T> StackListener<'ev, '_, T> {
1543 #[inline]
1544 fn listener(&self) -> &InnerListener<T, &'ev Inner<T>> {
1545 &self.slot.listener
1546 }
1547
1548 #[inline]
1549 fn listener_mut(&mut self) -> Pin<&mut InnerListener<T, &'ev Inner<T>>> {
1550 self.slot.as_mut().project().listener
1551 }
1552 }
1553
1554 forward_impl_to_listener! { T => StackListener<'_, '_, T> }
1555}
1556