Merge e9060bd79d into 2493d3c595

Co-authored-by: Jacob Rothstein <hi@jbr.me>
Signed-off-by: John Nunley <dev@notgull.net>

.github/workflows/ci.yml

@@ -94,3 +94,4 @@ jobs:
       - uses: rustsec/audit-check@master
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
+

CHANGELOG.md

@@ -1,3 +1,64 @@
+# Version 5.3.0
+
+- Add a `loom` implementation. This feature is unstable and is not semver-supported. (#126)
+- Make the panic message for polling the `EventListener` after it has completed more clear. (#125)
+
+# Version 5.2.0
+
+- Make `StackSlot` `Sync`. (#121)
+
+# Version 5.1.0
+
+- Make `StackSlot` `Send`. (#119)
+
+# Version 5.0.0
+
+- **Breaking:** Rework the API to afford better usage. (#105)
+  - The heap-based API of the v2.x line is back.
+  - However, there is a stack-based API as an alternative.
+- Add a way to get the total number of listeners. (#114)
+
+# Version 4.0.3
+
+- Relax MSRV to 1.60. (#110)
+
+# Version 4.0.2
+
+- Avoid spinning in `wait_deadline`. (#107)
+
+# Version 4.0.1
+
+- Fix a use-after-move error after an `EventListener` is assigned to listen to
+  another `Event`. (#101)
+
+# Version 4.0.0
+
+- **Breaking:** Fix a footgun in the `EventListener` type. `EventListener::new()`
+  now no longer takes an `&Event` as an argument, and `EventListener::listen()`
+  takes the  `&Event` as an argument. Hopefully this should prevent `.await`ing
+  on a listener without making sure it's listening first. (#94)
+
+# Version 3.1.0
+
+- Implement `UnwindSafe` and `RefUnwindSafe` for `EventListener`. This was unintentionally removed in version 3 (#96).
+
+# Version 3.0.1
+
+- Emphasize that `listen()` must be called on `EventListener` in documentation. (#90)
+- Write useful output in `fmt::Debug` implementations. (#86)
+
+# Version 3.0.0
+
+- Use the `parking` crate instead of threading APIs (#27)
+- Bump MSRV to 1.59 (#71)
+- **Breaking:** Make this crate `no_std`-compatible on `default-features = false`. (#34)
+- Create a new `event-listener-strategy` crate for abstracting over blocking/non-blocking operations. (#49)
+- **Breaking:** Change the `EventListener` API to be `!Unpin`. (#51)
+- Enable a feature for the `portable-atomic` crate. (#53)
+- **Breaking:** Add a `Notification` trait which is used to enable tagged events. (#52)
+- Add an `is_notified()` method to `Event`. (#48)
+- **Breaking:** Make it so `notify()` returns the number of listeners notified. (#57)
+
 # Version 2.5.3
 
 - Fix fence on x86 and miri.

Cargo.toml

@@ -3,7 +3,7 @@ name = "event-listener"
 # When publishing a new version:
 # - Update CHANGELOG.md
 # - Create "v2.x.y" git tag
-version = "2.5.3"
+version = "5.3.0"
 authors = ["Stjepan Glavina <stjepang@gmail.com>"]
 edition = "2021"
 rust-version = "1.56"
@@ -26,4 +26,13 @@ portable-atomic-util = { version = "0.1.5", optional = true }
 [dev-dependencies]
 futures = { version = "0.3", default-features = false, features = ["std"] }
 futures-lite = "2.3.0"
+criterion = { version = "0.3.4", default-features = false, features = ["cargo_bench_support"] }
 waker-fn = "1"
+
+[[bench]]
+name = "bench"
+harness = false
+required-features = ["std"]
+
+[lib]
+bench = false

README.md

@@ -1,6 +1,6 @@
 # event-listener
 
-[![Build](https://github.com/smol-rs/event-listener/workflows/Build%20and%20test/badge.svg)](
+[![Build](https://github.com/smol-rs/event-listener/workflows/CI/badge.svg)](
 https://github.com/smol-rs/event-listener/actions)
 [![License](https://img.shields.io/badge/license-Apache--2.0_OR_MIT-blue.svg)](
 https://github.com/smol-rs/event-listener)
@@ -17,7 +17,7 @@ You can use this crate to turn non-blocking data structures into async or blocki
 structures. See a [simple mutex] implementation that exposes an async and a blocking interface
 for acquiring locks.
 
-[eventcounts]: http://www.1024cores.net/home/lock-free-algorithms/eventcounts
+[eventcounts]: https://www.1024cores.net/home/lock-free-algorithms/eventcounts
 [simple mutex]: ./examples/mutex.rs
 
 ## Examples
@@ -74,8 +74,8 @@ loop {
 
 Licensed under either of
 
- * Apache License, Version 2.0 ([LICENSE-APACHE](LICENSE-APACHE) or http://www.apache.org/licenses/LICENSE-2.0)
- * MIT license ([LICENSE-MIT](LICENSE-MIT) or http://opensource.org/licenses/MIT)
+ * Apache License, Version 2.0 ([LICENSE-APACHE](LICENSE-APACHE) or https://www.apache.org/licenses/LICENSE-2.0)
+ * MIT license ([LICENSE-MIT](LICENSE-MIT) or https://opensource.org/licenses/MIT)
 
 at your option.
 

benches/bench.rs

@@ -0,0 +1,23 @@
+use criterion::{criterion_group, criterion_main, Criterion};
+use event_listener::{Event, Listener};
+use std::iter;
+
+const COUNT: usize = 8000;
+
+fn bench_events(c: &mut Criterion) {
+    c.bench_function("notify_and_wait", |b| {
+        let ev = Event::new();
+        let mut handles = Vec::with_capacity(COUNT);
+        b.iter(|| {
+            handles.extend(iter::repeat_with(|| ev.listen()).take(COUNT));
+            ev.notify(COUNT);
+
+            for handle in handles.drain(..) {
+                handle.wait();
+            }
+        });
+    });
+}
+
+criterion_group!(benches, bench_events);
+criterion_main!(benches);

examples/mutex.rs

@@ -90,9 +90,7 @@ impl<T> Mutex<T> {
                 }
                 Some(l) => {
                     // Wait until a notification is received.
-                    if l.wait_deadline(deadline).is_none() {
-                        return None;
-                    }
+                    l.wait_deadline(deadline)?;
                 }
             }
         }

src/lib.rs

@@ -6,7 +6,7 @@
 //! structures. See a [simple mutex] implementation that exposes an async and a blocking interface
 //! for acquiring locks.
 //!
-//! [eventcounts]: http://www.1024cores.net/home/lock-free-algorithms/eventcounts
+//! [eventcounts]: https://www.1024cores.net/home/lock-free-algorithms/eventcounts
 //! [simple mutex]: https://github.com/smol-rs/event-listener/blob/master/examples/mutex.rs
 //!
 //! # Examples
@@ -67,6 +67,12 @@
 //!
 //! [`portable-atomic`]: https://crates.io/crates/portable-atomic
 
+#![doc(
+    html_favicon_url = "https://raw.githubusercontent.com/smol-rs/smol/master/assets/images/logo_fullsize_transparent.png"
+)]
+#![doc(
+    html_logo_url = "https://raw.githubusercontent.com/smol-rs/smol/master/assets/images/logo_fullsize_transparent.png"
+)]
 #![warn(missing_docs, missing_debug_implementations, rust_2018_idioms)]
 
 use loom::atomic::{self, AtomicPtr, AtomicUsize, Ordering};
@@ -81,11 +87,10 @@ use std::ops::{Deref, DerefMut};
 use std::panic::{RefUnwindSafe, UnwindSafe};
 use std::pin::Pin;
 use std::ptr::{self, NonNull};
-use std::sync::{Mutex, MutexGuard};
+use std::sync::{Mutex, MutexGuard, TryLockError};
 use std::task::{Context, Poll, Waker};
 use std::thread::{self, Thread};
 use std::time::{Duration, Instant};
-use std::usize;
 
 mod notify;
 pub use notify::{IntoNotification, Notification};
@@ -379,6 +384,41 @@ impl Event {
     pub fn notify_additional_relaxed(&self, n: usize) -> usize {
         self.notify(n.additional().relaxed())
     }
+
+    /// Return the listener count by acquiring a lock.
+    ///
+    /// This is just a snapshot of the number of listeners at this point in time.
+    /// It is possible for the actual number to change at any point.
+    /// The number should only ever be used as a hint.
+    /// This is only available when `std` feature is enabled.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use event_listener::Event;
+    ///
+    /// let event = Event::new();
+    ///
+    /// assert_eq!(event.total_listeners(), 0);
+    ///
+    /// let listener1 = event.listen();
+    /// assert_eq!(event.total_listeners(), 1);    
+    ///
+    /// let listener2 = event.listen();
+    /// assert_eq!(event.total_listeners(), 2);        
+    ///
+    /// drop(listener1);
+    /// drop(listener2);
+    /// assert_eq!(event.total_listeners(), 0);        
+    /// ```
+    #[inline]
+    pub fn total_listeners(&self) -> usize {
+        if let Some(inner) = self.try_inner() {
+            inner.lock().len
+        } else {
+            0
+        }
+    }
 }
 
 impl<T> Event<T> {
@@ -462,7 +502,7 @@ impl<T> Event<T> {
         if let Some(inner) = self.try_inner() {
             let limit = if notify.is_additional(Internal::new()) {
                 // Notify if there is at least one unnotified listener.
-                std::usize::MAX
+                usize::MAX
             } else {
                 // Notify if there is at least one unnotified listener and the number of notified
                 // listeners is less than `n`.
@@ -477,6 +517,26 @@ impl<T> Event<T> {
         0
     }
 
+    /// Tell whether any listeners are currently notified.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use event_listener::Event;
+    ///
+    /// let event = Event::new();
+    /// let listener = event.listen();
+    /// assert!(!event.is_notified());
+    ///
+    /// event.notify(1);
+    /// assert!(event.is_notified());
+    /// ```
+    #[inline]
+    pub fn is_notified(&self) -> bool {
+        self.try_inner()
+            .map_or(false, |inner| inner.notified.load(Ordering::Acquire) > 0)
+    }
+
     /// Returns a reference to the inner state if it was initialized.
     #[inline]
     fn try_inner(&self) -> Option<&Inner<T>> {
@@ -552,7 +612,33 @@ impl<T> Drop for Event<T> {
 
 impl fmt::Debug for Event {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.pad("Event { .. }")
+        let inner = match self.try_inner() {
+            None => {
+                return f
+                    .debug_tuple("Event")
+                    .field(&format_args!("<uninitialized>"))
+                    .finish()
+            }
+            Some(inner) => inner,
+        };
+
+        let guard = match inner.list.try_lock() {
+            Err(TryLockError::WouldBlock) => {
+                return f
+                    .debug_tuple("Event")
+                    .field(&format_args!("<locked>"))
+                    .finish()
+            }
+
+            Err(TryLockError::Poisoned(err)) => err.into_inner(),
+
+            Ok(lock) => lock,
+        };
+
+        f.debug_struct("Event")
+            .field("listeners_notified", &guard.notified)
+            .field("listeners_total", &guard.len)
+            .finish()
     }
 }
 
@@ -583,7 +669,6 @@ pub trait Listener<T = ()>: Future<Output = T> + __sealed::Sealed {
     /// // Receive the notification.
     /// listener.wait();
     /// ```
-    #[cfg(all(feature = "std", not(target_family = "wasm")))]
     fn wait(self) -> T;
 
     /// Blocks until a notification is received or a timeout is reached.
@@ -1170,3 +1255,13 @@ mod __sealed {
     #[doc(hidden)]
     pub trait Sealed {}
 }
+
+fn __test_send_and_sync() {
+    fn _assert_send<T: Send>() {}
+    fn _assert_sync<T: Sync>() {}
+
+    _assert_send::<Event<()>>();
+    _assert_sync::<Event<()>>();
+    _assert_send::<EventListener<()>>();
+    _assert_sync::<EventListener<()>>();
+}

src/notify.rs

@@ -1,6 +1,5 @@
 //! The `Notification` trait for specifying notification.
 
-#[cfg(feature = "std")]
 use core::fmt;
 
 use crate::loom::atomic::{self, Ordering};
@@ -161,7 +160,6 @@ where
 }
 
 /// Use a tag to notify listeners.
-#[cfg(feature = "std")]
 #[derive(Debug, Clone)]
 #[doc(hidden)]
 pub struct Tag<N: ?Sized, T> {
@@ -169,7 +167,6 @@ pub struct Tag<N: ?Sized, T> {
     inner: N,
 }
 
-#[cfg(feature = "std")]
 impl<N: ?Sized, T> Tag<N, T> {
     /// Create a new `Tag` with the given tag and notification.
     fn new(tag: T, inner: N) -> Self
@@ -180,7 +177,6 @@ impl<N: ?Sized, T> Tag<N, T> {
     }
 }
 
-#[cfg(feature = "std")]
 impl<N, T> NotificationPrivate for Tag<N, T>
 where
     N: Notification + ?Sized,
@@ -206,14 +202,12 @@ where
 }
 
 /// Use a function to generate a tag to notify listeners.
-#[cfg(feature = "std")]
 #[doc(hidden)]
 pub struct TagWith<N: ?Sized, F> {
     tag: F,
     inner: N,
 }
 
-#[cfg(feature = "std")]
 impl<N: fmt::Debug, F> fmt::Debug for TagWith<N, F> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("TagWith")
@@ -223,7 +217,6 @@ impl<N: fmt::Debug, F> fmt::Debug for TagWith<N, F> {
     }
 }
 
-#[cfg(feature = "std")]
 impl<N, F> TagWith<N, F> {
     /// Create a new `TagFn` with the given tag function and notification.
     fn new(tag: F, inner: N) -> Self {
@@ -231,7 +224,6 @@ impl<N, F> TagWith<N, F> {
     }
 }
 
-#[cfg(feature = "std")]
 impl<N, F, T> NotificationPrivate for TagWith<N, F>
 where
     N: Notification + ?Sized,
@@ -490,7 +482,6 @@ pub trait IntoNotification: __private::Sealed {
     /// assert_eq!(listener1.wait(), true);
     /// assert_eq!(listener2.wait(), false);
     /// ```
-    #[cfg(feature = "std")]
     fn tag<T: Clone>(self, tag: T) -> Tag<Self::Notify, T>
     where
         Self: Sized + IntoNotification<Tag = ()>,
@@ -524,7 +515,6 @@ pub trait IntoNotification: __private::Sealed {
     /// assert_eq!(listener1.wait(), true);
     /// assert_eq!(listener2.wait(), false);
     /// ```
-    #[cfg(feature = "std")]
     fn tag_with<T, F>(self, tag: F) -> TagWith<Self::Notify, F>
     where
         Self: Sized + IntoNotification<Tag = ()>,
@@ -568,7 +558,7 @@ impl_for_numeric_types! { usize u8 u16 u32 u64 u128 isize i8 i16 i32 i64 i128 }
 /// Equivalent to `atomic::fence(Ordering::SeqCst)`, but in some cases faster.
 #[inline]
 pub(super) fn full_fence() {
-    #[cfg(all(any(target_arch = "x86", target_arch = "x86_64"), not(miri), not(loom)))]
+    #[cfg(all(any(target_arch = "x86", target_arch = "x86_64"), not(miri)))]
     {
         use core::{arch::asm, cell::UnsafeCell};
         // HACK(stjepang): On x86 architectures there are two different ways of executing

tests/notify.rs

@@ -2,9 +2,8 @@ use std::future::Future;
 use std::pin::Pin;
 use std::sync::{Arc, Mutex};
 use std::task::Context;
-use std::usize;
 
-use event_listener::{Event, EventListener};
+use event_listener::{Event, EventListener, Listener};
 use waker_fn::waker_fn;
 
 fn is_notified(listener: &mut EventListener) -> bool {
@@ -14,6 +13,20 @@ fn is_notified(listener: &mut EventListener) -> bool {
         .is_ready()
 }
 
+#[test]
+fn debug() {
+    let event = Event::new();
+    let fmt = format!("{:?}", &event);
+    assert!(fmt.contains("Event"));
+
+    let listener = event.listen();
+    let fmt = format!("{:?}", &listener);
+    assert!(fmt.contains("EventListener"));
+
+    let fmt = format!("{:?}", &event);
+    assert!(fmt.contains("Event"));
+}
+
 #[test]
 fn notify() {
     let event = Event::new();
@@ -22,12 +35,15 @@ fn notify() {
     let mut l2 = event.listen();
     let mut l3 = event.listen();
 
+    assert!(!event.is_notified());
     assert!(!is_notified(&mut l1));
     assert!(!is_notified(&mut l2));
     assert!(!is_notified(&mut l3));
 
     assert_eq!(event.notify(2), 2);
+    assert!(event.is_notified());
     assert_eq!(event.notify(1), 0);
+    assert!(event.is_notified());
     assert!(is_notified(&mut l1));
     assert!(is_notified(&mut l2));
     assert!(!is_notified(&mut l3));
@@ -41,9 +57,11 @@ fn notify_additional() {
     let mut l2 = event.listen();
     let mut l3 = event.listen();
 
+    assert!(!event.is_notified());
     assert_eq!(event.notify_additional(1), 1);
     assert_eq!(event.notify(1), 0);
     assert_eq!(event.notify_additional(1), 1);
+    assert!(event.is_notified());
 
     assert!(is_notified(&mut l1));
     assert!(is_notified(&mut l2));
@@ -54,6 +72,7 @@ fn notify_additional() {
 fn notify_zero() {
     let event = Event::new();
     assert_eq!(event.notify(1), 0);
+    assert!(!event.is_notified());
 }
 
 #[test]
@@ -119,8 +138,10 @@ fn notify_all() {
 
     assert!(!is_notified(&mut l1));
     assert!(!is_notified(&mut l2));
+    assert!(!event.is_notified());
 
     assert_eq!(event.notify(usize::MAX), 2);
+    assert!(event.is_notified());
     assert!(is_notified(&mut l1));
     assert!(is_notified(&mut l2));
 }
@@ -184,6 +205,55 @@ fn drop_non_notified() {
     assert!(!is_notified(&mut l2));
 }
 
+#[test]
+fn discard() {
+    let event = Event::default();
+
+    let l1 = event.listen();
+    assert!(!l1.discard());
+
+    let l1 = event.listen();
+    event.notify(1);
+    assert!(l1.discard());
+
+    let l1 = event.listen();
+    event.notify_additional(1);
+    assert!(l1.discard());
+
+    let mut l1 = event.listen();
+    event.notify(1);
+    assert!(is_notified(&mut l1));
+    assert!(!l1.discard());
+}
+
+#[test]
+fn same_event() {
+    let e1 = Event::new();
+    let e2 = Event::new();
+
+    let l1 = e1.listen();
+    let l2 = e1.listen();
+    let l3 = e2.listen();
+
+    assert!(l1.listens_to(&e1));
+    assert!(!l1.listens_to(&e2));
+    assert!(l1.same_event(&l2));
+    assert!(!l1.same_event(&l3));
+}
+
+#[test]
+#[should_panic = "cannot poll a completed `EventListener` future"]
+fn poll_twice() {
+    let event = Event::new();
+    let mut l1 = event.listen();
+    event.notify(1);
+
+    assert!(is_notified(&mut l1));
+
+    // Panic here.
+    is_notified(&mut l1);
+}
+
 #[test]
 fn notify_all_fair() {
     let event = Event::new();

tests/park.rs

@@ -1,8 +1,35 @@
 //! Test the wait() family of methods.
 
-use event_listener::{Event, IntoNotification, Listener};
+use event_listener::{Event, EventListener, IntoNotification, Listener};
+
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::Context;
+use std::thread;
 use std::time::{Duration, Instant};
 
+use waker_fn::waker_fn;
+
+fn is_notified(listener: &mut EventListener) -> bool {
+    let waker = waker_fn(|| ());
+    Pin::new(listener)
+        .poll(&mut Context::from_waker(&waker))
+        .is_ready()
+}
+
+#[test]
+fn total_listeners() {
+    let event = Event::new();
+    assert_eq!(event.total_listeners(), 0);
+
+    let listener = event.listen();
+    assert_eq!(event.total_listeners(), 1);
+
+    drop(listener);
+    assert_eq!(event.total_listeners(), 0);
+}
+
 #[test]
 fn wait() {
     let event = Event::new();
@@ -30,6 +57,18 @@ fn wait_timeout() {
     assert_eq!(listener.wait_timeout(Duration::from_millis(50)), Some(()));
 }
 
+#[test]
+fn wait_deadline() {
+    let event = Event::new();
+    let listener = event.listen();
+
+    assert_eq!(event.notify(1), 1);
+    assert_eq!(
+        listener.wait_deadline(Instant::now() + Duration::from_millis(50)),
+        Some(())
+    );
+}
+
 #[test]
 fn wait_timeout_expiry() {
     let event = Event::new();
@@ -39,3 +78,49 @@ fn wait_timeout_expiry() {
     assert_eq!(listener.wait_timeout(Duration::from_millis(200)), None);
     assert!(Instant::now().duration_since(start) >= Duration::from_millis(200));
 }
+
+#[test]
+fn unpark() {
+    let event = Arc::new(Event::new());
+    let listener = event.listen();
+
+    thread::spawn({
+        let event = event.clone();
+        move || {
+            thread::sleep(Duration::from_millis(100));
+            event.notify(1);
+        }
+    });
+
+    listener.wait();
+}
+
+#[test]
+fn unpark_timeout() {
+    let event = Arc::new(Event::new());
+    let listener = event.listen();
+
+    thread::spawn({
+        let event = event.clone();
+        move || {
+            thread::sleep(Duration::from_millis(100));
+            event.notify(1);
+        }
+    });
+
+    let x = listener.wait_timeout(Duration::from_millis(200));
+    assert!(x.is_some());
+}
+
+#[test]
+#[should_panic = "cannot wait twice on an `EventListener`"]
+fn wait_twice() {
+    let event = Event::new();
+    let mut listener = event.listen();
+    event.notify(1);
+
+    assert!(is_notified(&mut listener));
+
+    // Panic here.
+    listener.wait();
+}