add singleflight

Author: smallnest

special/Cargo.toml

@@ -8,6 +8,8 @@ edition = "2021"
 [dependencies]
 async-lock = "2.5.0"
 async-oneshot = "0.5.0"
+async-weighted-semaphore = "0.2.1"
+async_singleflight = "0.5.0"
 atomic_float = "0.1.0"
 atomicbox = "0.4.0"
 atomig = "0.4.0"
@@ -19,14 +21,17 @@ dashmap = "5.4.0"
 event-listener = "2.5.3"
 evmap = "10.0.2"
 flurry = "0.4.0"
+futures = "0.3.25"
 oneshot = "0.1.5"
 portable-atomic = { version = "0.3", features=["float"] }
 process_lock = "0.1.0"
 scc = "0.11.1"
 sharded-slab = "0.1.4"
 simple-mutex = "1.1.5"
+singleflight-async = "0.1.1"
 slab = "0.4.7"
 smol = "1.2.5"
+tokio = { version = "1.21.2", features = ["full"] }
 triggered = "0.1.2"
 triple_buffer = "6.2.0"
 try-lock = "0.2.3"

special/src/lib.rs

@@ -5,11 +5,15 @@ mod primitive;
 mod notify;
 mod queue;
 mod scc_examples;
+mod sema_examples;
+mod singleflight_example;
 
 pub use oslock::*;
 pub use oneshots::*;
 pub use map::*;
 pub use primitive::*;
 pub use notify::*;
 pub use queue::*;
-pub use scc_examples::*;
+pub use scc_examples::*;
+pub use sema_examples::*;
+pub use singleflight_example::*;

special/src/main.rs

@@ -26,7 +26,6 @@ fn main() {
     async_lock_mutex();
     async_lock_rwlock();
     async_lock_barrier();
-    async_lock_semaphore();
 
     portable_atomic_i128();
     portable_atomic_u128();
@@ -51,6 +50,14 @@ fn main() {
     scc_hashset();
     scc_queue();
 
+
+    async_lock_semaphore();
+    async_weighted_semaphore_example();
+    tokio_semaphore_example();
+
+    singleflight_example();
+    async_singleflight_example();
+
 }
 
 

special/src/primitive/async_lock_examples.rs

@@ -50,14 +50,3 @@ pub fn async_lock_barrier() {
         }
     });
 }
-
-pub fn async_lock_semaphore() {
-    let s = Arc::new(Semaphore::new(2));
-
-    let _g1 = s.try_acquire_arc().unwrap();
-    let g2 = s.try_acquire_arc().unwrap();
-
-    assert!(s.try_acquire_arc().is_none());
-    drop(g2);
-    assert!(s.try_acquire_arc().is_some());
-}

special/src/sema_examples.rs

@@ -0,0 +1,52 @@
+use futures::pin_mut;
+use futures::poll;
+use std::sync::Arc;
+
+pub fn tokio_semaphore_example() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    rt.block_on(async {
+        let semaphore = Arc::new(tokio::sync::Semaphore::new(3));
+        let mut join_handles = Vec::new();
+
+        for _ in 0..5 {
+            let permit = semaphore.clone().acquire_owned().await.unwrap();
+            join_handles.push(tokio::spawn(async move {
+                // perform task...
+                // explicitly own `permit` in the task
+                drop(permit);
+            }));
+        }
+
+        for handle in join_handles {
+            handle.await.unwrap();
+        }
+    });
+}
+
+pub fn async_weighted_semaphore_example() {
+    smol::block_on(async {
+        let sem = async_weighted_semaphore::Semaphore::new(1);
+        let a = sem.acquire(2);
+        let b = sem.acquire(1);
+        pin_mut!(a);
+        pin_mut!(b);
+        assert!(poll!(&mut a).is_pending());
+        assert!(poll!(&mut b).is_pending());
+
+        sem.release(1);
+        assert!(poll!(&mut a).is_ready());
+        assert!(poll!(&mut b).is_ready());
+    });
+}
+
+pub fn async_lock_semaphore() {
+    let s = Arc::new(async_lock::Semaphore::new(2));
+
+    let _g1 = s.try_acquire_arc().unwrap();
+    let g2 = s.try_acquire_arc().unwrap();
+
+    assert!(s.try_acquire_arc().is_none());
+    drop(g2);
+    assert!(s.try_acquire_arc().is_some());
+}

special/src/singleflight_example.rs

@@ -0,0 +1,50 @@
+use futures::future::join_all;
+use singleflight_async::SingleFlight;
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_singleflight::Group;
+
+pub fn singleflight_example() {
+    smol::block_on(async {
+        let group = SingleFlight::new();
+        let mut futures = Vec::new();
+        for _ in 0..10 {
+            futures.push(group.work("key", || async {
+                println!("will sleep to simulate async task");
+                smol::Timer::after(std::time::Duration::from_millis(100)).await;
+                println!("real task done");
+                "my-result"
+            }));
+        }
+
+        for fut in futures.into_iter() {
+            assert_eq!(fut.await, "my-result");
+            println!("task finished");
+        }
+    });
+}
+
+const RES: usize = 7;
+
+async fn expensive_fn() -> Result<usize, ()> {
+    smol::Timer::after(std::time::Duration::from_millis(100)).await;
+    Ok(RES)
+}
+
+pub fn async_singleflight_example() {
+    smol::block_on(async {
+        let g = Arc::new(Group::<_, ()>::new());
+        let mut handlers = Vec::new();
+        for _ in 0..10 {
+            let g = g.clone();
+            handlers.push(smol::spawn(async move {
+                let res = g.work("key", expensive_fn()).await.0;
+                let r = res.unwrap();
+                println!("{}", r);
+            }));
+        }
+
+        join_all(handlers).await;
+    });
+}