m: Port to event-listener v5.0.0

cc smol-rs/event-listener#105

Signed-off-by: John Nunley <dev@notgull.net>

.github/workflows/ci.yml

@@ -58,6 +58,11 @@ jobs:
 
   msrv:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        # When updating this, the reminder to update the minimum supported
+        # Rust version in Cargo.toml.
+        rust: ['1.61']
     steps:
       - uses: actions/checkout@v4
       - name: Install cargo-hack

Cargo.toml

@@ -15,16 +15,16 @@ categories = ["asynchronous", "concurrency"]
 exclude = ["/.*"]
 
 [dependencies]
-event-listener = { version = "4.0.0", default-features = false }
-event-listener-strategy = { version = "0.4.0", default-features = false }
+event-listener = { version = "5.0.0", default-features = false }
+event-listener-strategy = { version = "0.5.0", default-features = false }
 pin-project-lite = "0.2.11"
 
 [features]
 default = ["std"]
 std = ["event-listener/std", "event-listener-strategy/std"]
 
 [dev-dependencies]
-async-channel = "2.0.0"
+async-channel = "2.2.0"
 fastrand = "2.0.0"
 futures-lite = "2.0.0"
 waker-fn = "1.1.0"

src/barrier.rs

@@ -82,7 +82,7 @@ impl Barrier {
         BarrierWait::_new(BarrierWaitInner {
             barrier: self,
             lock: Some(self.state.lock()),
-            evl: EventListener::new(),
+            evl: None,
             state: WaitState::Initial,
         })
     }
@@ -148,8 +148,7 @@ pin_project_lite::pin_project! {
         lock: Option<Lock<'a, State>>,
 
         // An event listener for the `barrier.event` event.
-        #[pin]
-        evl: EventListener,
+        evl: Option<EventListener>,
 
         // The current state of the future.
         state: WaitState,
@@ -200,7 +199,7 @@ impl EventListenerFuture for BarrierWaitInner<'_> {
 
                     if state.count < this.barrier.n {
                         // We need to wait for the event.
-                        this.evl.as_mut().listen(&this.barrier.event);
+                        *this.evl = Some(this.barrier.event.listen());
                         *this.state = WaitState::Waiting { local_gen };
                     } else {
                         // We are the last one.
@@ -212,7 +211,7 @@ impl EventListenerFuture for BarrierWaitInner<'_> {
                 }
 
                 WaitState::Waiting { local_gen } => {
-                    ready!(strategy.poll(this.evl.as_mut(), cx));
+                    ready!(strategy.poll(this.evl, cx));
 
                     // We are now re-acquiring the mutex.
                     this.lock.as_mut().set(Some(this.barrier.state.lock()));
@@ -233,7 +232,7 @@ impl EventListenerFuture for BarrierWaitInner<'_> {
 
                     if *local_gen == state.generation_id && state.count < this.barrier.n {
                         // We need to wait for the event again.
-                        this.evl.as_mut().listen(&this.barrier.event);
+                        *this.evl = Some(this.barrier.event.listen());
                         *this.state = WaitState::Waiting {
                             local_gen: *local_gen,
                         };

src/lib.rs

@@ -58,6 +58,7 @@ macro_rules! ready {
 /// Pins a variable on the stack.
 ///
 /// TODO: Drop in favor of `core::pin::pin`, once MSRV is bumped to 1.68.
+#[cfg(all(feature = "std", not(target_family = "wasm")))]
 macro_rules! pin {
     ($($x:ident),* $(,)?) => {
         $(

src/mutex.rs

@@ -1,7 +1,7 @@
 use core::borrow::Borrow;
 use core::cell::UnsafeCell;
 use core::fmt;
-use core::marker::PhantomData;
+use core::marker::{PhantomData, PhantomPinned};
 use core::ops::{Deref, DerefMut};
 use core::pin::Pin;
 use core::sync::atomic::{AtomicUsize, Ordering};
@@ -445,8 +445,7 @@ pin_project_lite::pin_project! {
         mutex: Option<B>,
 
         // The event listener waiting on the mutex.
-        #[pin]
-        listener: EventListener,
+        listener: Option<EventListener>,
 
         // The point at which the mutex lock was started.
         start: Start,
@@ -457,6 +456,10 @@ pin_project_lite::pin_project! {
         // Capture the `T` lifetime.
         #[pin]
         _marker: PhantomData<T>,
+
+        // Keeping this type `!Unpin` enables future optimizations.
+        #[pin]
+        _pin: PhantomPinned
     }
 
     impl<T: ?Sized, B: Borrow<Mutex<T>>> PinnedDrop for AcquireSlow<B, T> {
@@ -477,18 +480,16 @@ impl<T: ?Sized, B: Borrow<Mutex<T>>> AcquireSlow<B, T> {
     /// Create a new `AcquireSlow` future.
     #[cold]
     fn new(mutex: B) -> Self {
-        // Create a new instance of the listener.
-        let listener = { EventListener::new() };
-
         AcquireSlow {
             mutex: Some(mutex),
-            listener,
+            listener: None,
             start: Start {
                 #[cfg(all(feature = "std", not(target_family = "wasm")))]
                 start: None,
             },
             starved: false,
             _marker: PhantomData,
+            _pin: PhantomPinned,
         }
     }
 
@@ -517,7 +518,7 @@ impl<T: ?Sized, B: Unpin + Borrow<Mutex<T>>> EventListenerFuture for AcquireSlow
         strategy: &mut S,
         context: &mut S::Context,
     ) -> Poll<Self::Output> {
-        let mut this = self.as_mut().project();
+        let this = self.as_mut().project();
         #[cfg(all(feature = "std", not(target_family = "wasm")))]
         let start = *this.start.start.get_or_insert_with(Instant::now);
         let mutex = Borrow::<Mutex<T>>::borrow(
@@ -528,8 +529,8 @@ impl<T: ?Sized, B: Unpin + Borrow<Mutex<T>>> EventListenerFuture for AcquireSlow
         if !*this.starved {
             loop {
                 // Start listening for events.
-                if !this.listener.is_listening() {
-                    this.listener.as_mut().listen(&mutex.lock_ops);
+                if this.listener.is_none() {
+                    *this.listener = Some(mutex.lock_ops.listen());
 
                     // Try locking if nobody is being starved.
                     match mutex
@@ -547,7 +548,7 @@ impl<T: ?Sized, B: Unpin + Borrow<Mutex<T>>> EventListenerFuture for AcquireSlow
                         _ => break,
                     }
                 } else {
-                    ready!(strategy.poll(this.listener.as_mut(), context));
+                    ready!(strategy.poll(this.listener, context));
 
                     // Try locking if nobody is being starved.
                     match mutex
@@ -591,9 +592,9 @@ impl<T: ?Sized, B: Unpin + Borrow<Mutex<T>>> EventListenerFuture for AcquireSlow
 
         // Fairer locking loop.
         loop {
-            if !this.listener.is_listening() {
+            if this.listener.is_none() {
                 // Start listening for events.
-                this.listener.as_mut().listen(&mutex.lock_ops);
+                *this.listener = Some(mutex.lock_ops.listen());
 
                 // Try locking if nobody else is being starved.
                 match mutex
@@ -615,7 +616,7 @@ impl<T: ?Sized, B: Unpin + Borrow<Mutex<T>>> EventListenerFuture for AcquireSlow
                 }
             } else {
                 // Wait for a notification.
-                ready!(strategy.poll(this.listener.as_mut(), context));
+                ready!(strategy.poll(this.listener, context));
 
                 // Try acquiring the lock without waiting for others.
                 if mutex.state.fetch_or(1, Ordering::Acquire) % 2 == 0 {

src/once_cell.rs

@@ -9,9 +9,12 @@ use core::sync::atomic::{AtomicUsize, Ordering};
 #[cfg(all(feature = "std", not(target_family = "wasm")))]
 use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
 
-use event_listener::{Event, EventListener};
+use event_listener::Event;
 use event_listener_strategy::{NonBlocking, Strategy};
 
+#[cfg(all(feature = "std", not(target_family = "wasm")))]
+use event_listener::Listener;
+
 /// The current state of the `OnceCell`.
 #[derive(Copy, Clone, PartialEq, Eq)]
 #[repr(usize)]
@@ -274,9 +277,7 @@ impl<T> OnceCell<T> {
         }
 
         // Slow path: wait for the value to be initialized.
-        let listener = EventListener::new();
-        pin!(listener);
-        listener.as_mut().listen(&self.passive_waiters);
+        event_listener::listener!(self.passive_waiters => listener);
 
         // Try again.
         if let Some(value) = self.get() {
@@ -329,9 +330,7 @@ impl<T> OnceCell<T> {
         }
 
         // Slow path: wait for the value to be initialized.
-        let listener = EventListener::new();
-        pin!(listener);
-        listener.as_mut().listen(&self.passive_waiters);
+        event_listener::listener!(self.passive_waiters => listener);
 
         // Try again.
         if let Some(value) = self.get() {
@@ -503,10 +502,11 @@ impl<T> OnceCell<T> {
     /// ```
     #[cfg(all(feature = "std", not(target_family = "wasm")))]
     pub fn get_or_init_blocking(&self, closure: impl FnOnce() -> T + Unpin) -> &T {
-        match self.get_or_try_init_blocking(move || {
+        let result = self.get_or_try_init_blocking(move || {
             let result: Result<T, Infallible> = Ok(closure());
             result
-        }) {
+        });
+        match result {
             Ok(value) => value,
             Err(infallible) => match infallible {},
         }
@@ -591,8 +591,7 @@ impl<T> OnceCell<T> {
         strategy: &mut impl for<'a> Strategy<'a>,
     ) -> Result<(), E> {
         // The event listener we're currently waiting on.
-        let event_listener = EventListener::new();
-        pin!(event_listener);
+        let mut event_listener = None;
 
         let mut closure = Some(closure);
 
@@ -611,10 +610,10 @@ impl<T> OnceCell<T> {
                     // but we do not have the ability to initialize it.
                     //
                     // We need to wait the initialization to complete.
-                    if event_listener.is_listening() {
-                        strategy.wait(event_listener.as_mut()).await;
+                    if let Some(listener) = event_listener.take() {
+                        strategy.wait(listener).await;
                     } else {
-                        event_listener.as_mut().listen(&self.active_initializers);
+                        event_listener = Some(self.active_initializers.listen());
                     }
                 }
                 State::Uninitialized => {

src/rwlock/futures.rs

@@ -426,13 +426,18 @@ pin_project_lite::pin_project! {
     impl<T: ?Sized> PinnedDrop for UpgradeArcInner<T> {
         fn drop(this: Pin<&mut Self>) {
             let this = this.project();
-            if !this.raw.is_ready() {
+            let is_ready = this.raw.is_ready();
+
+            // SAFETY: The drop impl for raw assumes that it is pinned.
+            unsafe {
+                ManuallyDrop::drop(this.raw.get_unchecked_mut());
+            }
+
+            if !is_ready {
                 // SAFETY: we drop the `Arc` (decrementing the reference count)
                 // only if this future was cancelled before returning an
                 // upgraded lock.
                 unsafe {
-                    // SAFETY: The drop impl for raw assumes that it is pinned.
-                    ManuallyDrop::drop(this.raw.get_unchecked_mut());
                     ManuallyDrop::drop(this.lock);
                 };
             }