fixes memory leak, bump to v0.2.2

.github/workflows/rust.yml

@@ -43,6 +43,4 @@ jobs:
     steps:
       - uses: actions/checkout@v3
       - uses: dtolnay/rust-toolchain@miri
-      - run: cargo miri test
-        env:
-          MIRIFLAGS: -Zmiri-ignore-leaks
+      - run: cargo miri test

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "async-unsync"
 description = "asynchronous, unsynchronized (thread-local) channels and semaphores"
-version = "0.2.0"
+version = "0.2.2"
 authors = ["Oliver Giersch"]
 license = "MIT/Apache-2.0"
 repository = "https://github.com/oliver-giersch/async-unsync.git"

README.md

@@ -20,7 +20,7 @@ To use this crate, add the following to your `Cargo.toml`:
 
 ```toml
 [dependencies]
-async-unsync = "0.2.0"
+async-unsync = "0.2.2"
 ```
 
 [1]: https://docs.rs/tokio/latest/tokio/sync/index.html

RELEASES.md

@@ -31,4 +31,9 @@
 
 ## Release `0.2.1`
 
-- Fixes a bug `Semaphore` that would not wake waiters on close.
+- Fixes a bug `Semaphore` that would not wake waiters on close.
+
+## Release `0.2.2`
+
+- Fixes a bug where a `Waker` registered by a `Semaphore` would not be dropped,
+  causing memory leaks.

src/queue.rs

@@ -282,7 +282,6 @@ impl<T, B> Queue<T, B> {
                 // when the queue first becomes empty, try to shrink it once.
                 if self.pop_count > 0 {
                     self.try_shrink_queue();
-                    self.pop_count = 0;
                 }
 
                 None

src/semaphore.rs

@@ -6,7 +6,7 @@ use core::{
     fmt,
     future::Future,
     marker::PhantomPinned,
-    mem::{self, MaybeUninit},
+    mem,
     pin::Pin,
     ptr::{self, NonNull},
     task::{Context, Poll, Waker},
@@ -655,22 +655,26 @@ struct Waiter {
 ///
 /// `get` is only called during traversal of that list, so it is guaranteed to
 /// have been initialized
-struct LateInitWaker(UnsafeCell<MaybeUninit<Waker>>);
+struct LateInitWaker(UnsafeCell<Option<Waker>>);
 
 impl LateInitWaker {
     const fn new() -> Self {
-        Self(UnsafeCell::new(MaybeUninit::uninit()))
+        Self(UnsafeCell::new(None))
     }
 
     fn set(&self, waker: Waker) {
         // SAFETY: no mutable or aliased access to waker possible, writing the
         // waker is unproblematic due to the required liveness of the pointer.
-        unsafe { (*self.0.get()).as_mut_ptr().write(waker) };
+        // this is never called when there already is a waker
+        unsafe { self.0.get().write(Some(waker)) };
     }
 
     unsafe fn get(&self) -> &Waker {
-        // SAFETY: initness required as function invariant, no al
-        unsafe { (*self.0.get()).assume_init_ref() }
+        // SAFETY: initness required as function invariant
+        match &*self.0.get() {
+            Some(waker) => waker,
+            None => core::hint::unreachable_unchecked(),
+        }
     }
 }
 
@@ -714,6 +718,41 @@ mod tests {
         assert_eq!(sem.outstanding_permits(), 0);
     }
 
+    #[test]
+    fn acquire_never() {
+        future::block_on(async {
+            let sem = super::Semaphore::new(0);
+            let mut fut = core::pin::pin!(sem.acquire());
+
+            core::future::poll_fn(|cx| {
+                assert!(fut.as_mut().poll(cx).is_pending());
+                Poll::Ready(())
+            })
+            .await;
+
+            assert_eq!(sem.available_permits(), 0);
+            assert_eq!(sem.outstanding_permits(), 0);
+        });
+    }
+
+    #[test]
+    fn acquire() {
+        future::block_on(async {
+            let sem = super::Semaphore::new(0);
+            let mut fut = core::pin::pin!(sem.acquire());
+            core::future::poll_fn(|cx| {
+                assert!(fut.as_mut().poll(cx).is_pending());
+                Poll::Ready(())
+            })
+            .await;
+
+            sem.add_permits(1);
+            let permit = fut.await.unwrap();
+            drop(permit);
+            assert_eq!(sem.available_permits(), 1);
+        });
+    }
+
     #[test]
     fn acquire_one() {
         future::block_on(async {
@@ -742,6 +781,30 @@ mod tests {
         });
     }
 
+    #[test]
+    fn poll_acquire_after_completion() {
+        future::block_on(async {
+            let sem = super::Semaphore::new(0);
+            let mut fut = core::pin::pin!(sem.acquire());
+            core::future::poll_fn(|cx| {
+                assert!(fut.as_mut().poll(cx).is_pending());
+                Poll::Ready(())
+            })
+            .await;
+
+            sem.add_permits(1);
+
+            core::future::poll_fn(|cx| {
+                assert!(fut.as_mut().poll(cx).is_ready());
+                assert!(fut.as_mut().poll(cx).is_pending());
+                Poll::Ready(())
+            })
+            .await;
+
+            assert_eq!(sem.available_permits(), 1);
+        });
+    }
+
     #[test]
     fn poll_future() {
         static RAW_VTABLE: RawWakerVTable = RawWakerVTable::new(