revert to mutex

Author: al8n

.gitignore

@@ -1,2 +1,3 @@
 /target
 Cargo.lock
+.DS_Store

Cargo.toml

@@ -6,27 +6,25 @@ homepage = "https://github.com/al8n/wg"
 repository = "https://github.com/al8n/wg.git"
 documentation = "https://docs.rs/wg/"
 readme = "README.md"
-version = "0.4.2"
+version = "0.5.0"
 license = "MIT OR Apache-2.0"
 keywords = ["waitgroup", "async", "sync", "notify", "wake"]
 categories = ["asynchronous", "concurrency", "data-structures"]
 edition = "2021"
 
 [features]
 default = []
-full = ["async", "triomphe", "parking_lot"]
-async = ["atomic-waker"]
+full = ["triomphe", "parking_lot"]
 triomphe = ["dep:triomphe"]
 parking_lot = ["dep:parking_lot"]
 
 [dependencies]
-atomic-waker = { version = "1", optional = true }
 parking_lot = {version = "0.12", optional = true }
 triomphe = { version = "0.1", optional = true }
 
 [dev-dependencies]
-tokio = { version = "1.32", features = ["full"] }
-async-std = { version = "1.12.0", features = ["attributes"]  }
+tokio = { version = "1", features = ["full"] }
+async-std = { version = "1.12", features = ["attributes"]  }
 
 [package.metadata.docs.rs]
 all-features = true

README.md

@@ -19,12 +19,13 @@ Golang like WaitGroup implementation for sync/async Rust.
 ## Installation
 ```toml
 [dependencies]
-wg = "0.4"
+wg = "0.5"
 ```
 
 ## Example
 
 ### Sync
+
 ```rust
 use wg::WaitGroup;
 use std::sync::Arc;
@@ -86,8 +87,8 @@ async fn main() {
 ```
 
 ## Acknowledgements
-- Inspired by Golang sync.WaitGroup, [ibraheemdev's `AwaitGroup`] and [`crossbeam_utils::WaitGroup`]. 
 
+- Inspired by Golang sync.WaitGroup, [ibraheemdev's `AwaitGroup`] and [`crossbeam_utils::WaitGroup`]. 
 
 ## License
 
@@ -96,8 +97,6 @@ Licensed under either of <a href="https://opensource.org/licenses/Apache-2.0">Ap
 2.0</a> or <a href="https://opensource.org/licenses/MIT">MIT license</a> at your option.
 </sup>
 
-<br>
-
 <sub>
 Unless you explicitly state otherwise, any contribution intentionally submitted
 for inclusion in this project by you, as defined in the Apache-2.0 license,
@@ -111,8 +110,3 @@ shall be dual licensed as above, without any additional terms or conditions.
 [doc-url]: https://docs.rs/wg
 [crates-url]: https://crates.io/crates/wg
 [codecov-url]: https://app.codecov.io/gh/al8n/wg/
-[license-url]: https://opensource.org/licenses/Apache-2.0
-[rustc-url]: https://github.com/rust-lang/rust/blob/master/RELEASES.md
-[license-apache-url]: https://opensource.org/licenses/Apache-2.0
-[license-mit-url]: https://opensource.org/licenses/MIT
-[rustc-image]: https://img.shields.io/badge/rustc-1.56.0%2B-orange.svg?style=for-the-badge&logo=Rust

src/lib.rs

@@ -371,23 +371,20 @@ impl WaitGroup {
     }
 }
 
-#[cfg(feature = "atomic-waker")]
 pub use r#async::*;
 
-#[cfg(feature = "atomic-waker")]
 mod r#async {
     use super::*;
-    use atomic_waker::AtomicWaker;
     use std::sync::atomic::{AtomicUsize, Ordering};
 
     use std::{
         future::Future,
         pin::Pin,
-        task::{Context, Poll},
+        task::{Context, Poll, Waker},
     };
 
     struct AsyncInner {
-        waker: AtomicWaker,
+        waker: Mutex<Option<Waker>>,
         count: AtomicUsize,
     }
 
@@ -442,7 +439,7 @@ mod r#async {
             Self {
                 inner: Arc::new(AsyncInner {
                     count: AtomicUsize::new(0),
-                    waker: AtomicWaker::new(),
+                    waker: Mutex::new(None),
                 }),
             }
         }
@@ -453,7 +450,7 @@ mod r#async {
             Self {
                 inner: Arc::new(AsyncInner {
                     count: AtomicUsize::new(count),
-                    waker: AtomicWaker::new(),
+                    waker: Mutex::new(None),
                 }),
             }
         }
@@ -542,22 +539,11 @@ mod r#async {
         /// }
         /// ```
         pub fn done(&self) {
-            let res = self
-                .inner
-                .count
-                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |val| {
-                    // We are the last worker
-                    if val == 1 {
-                        Some(0)
-                    } else if val == 0 {
-                        None
-                    } else {
-                        Some(val - 1)
-                    }
-                });
-            if let Ok(count) = res {
-                if count == 1 {
-                    self.inner.waker.wake();
+            let count = self.inner.count.fetch_sub(1, Ordering::Relaxed);
+            // We are the last worker
+            if count == 1 {
+                if let Some(waker) = self.inner.waker.lock_me().take() {
+                    waker.wake();
                 }
             }
         }
@@ -643,12 +629,16 @@ mod r#async {
         type Output = ();
 
         fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-            match self.inner.count.load(Ordering::Acquire) {
+            if self.inner.count.load(Ordering::Relaxed) == 0 {
+                return Poll::Ready(());
+            }
+
+            let waker = cx.waker().clone();
+            *self.inner.waker.lock_me() = Some(waker);
+
+            match self.inner.count.load(Ordering::Relaxed) {
                 0 => Poll::Ready(()),
-                _ => {
-                    self.inner.waker.register(cx.waker());
-                    Poll::Pending
-                }
+                _ => Poll::Pending,
             }
         }
     }
@@ -772,9 +762,7 @@ mod r#async {
         #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
         async fn test_async_block_wait() {
             let wg = AsyncWaitGroup::new();
-            wg.add(1);
-            let t_wg = wg.clone();
-
+            let t_wg = wg.add(1);
             tokio::spawn(async move {
                 // do some time consuming task
                 t_wg.done()