sync: new internal semaphore based on intrusive lists

benches/Cargo.toml

@@ -22,3 +22,14 @@ harness = false
 name = "scheduler"
 path = "scheduler.rs"
 harness = false
+
+
+[[bench]]
+name = "sync_rwlock"
+path = "sync_rwlock.rs"
+harness = false
+
+[[bench]]
+name = "sync_semaphore"
+path = "sync_semaphore.rs"
+harness = false

benches/sync_rwlock.rs

@@ -0,0 +1,152 @@
+use bencher::{black_box, Bencher};
+use tokio::{task, sync::RwLock};
+use std::sync::Arc;
+
+fn read_uncontended(b: &mut Bencher) {
+    let mut rt = tokio::runtime::Builder::new()
+        .core_threads(6)
+        .threaded_scheduler()
+        .build()
+        .unwrap();
+
+    let lock = Arc::new(RwLock::new(()));
+    b.iter(|| {
+
+        let lock = lock.clone();
+        rt.block_on(async move {
+            for _ in 0..6 {
+                let read = lock.read().await;
+                black_box(read);
+            }
+        })
+    });
+}
+
+fn read_concurrent_uncontended_multi(b: &mut Bencher) {
+    let mut rt = tokio::runtime::Builder::new()
+        .core_threads(6)
+        .threaded_scheduler()
+        .build()
+        .unwrap();
+
+    async fn task(lock: Arc<RwLock<()>>) {
+        let read = lock.read().await;
+        black_box(read);
+    }
+
+    let lock = Arc::new(RwLock::new(()));
+    b.iter(|| {
+        let lock = lock.clone();
+        rt.block_on(async move {
+            let j = tokio::try_join!{
+                task::spawn(task(lock.clone())),
+                task::spawn(task(lock.clone())),
+                task::spawn(task(lock.clone())),
+                task::spawn(task(lock.clone())),
+                task::spawn(task(lock.clone())),
+                task::spawn(task(lock.clone()))
+            };
+            j.unwrap();
+
+        })
+    });
+}
+
+fn read_concurrent_uncontended(b: &mut Bencher) {
+    let mut rt = tokio::runtime::Builder::new()
+        .basic_scheduler()
+        .build()
+        .unwrap();
+
+    async fn task(lock: Arc<RwLock<()>>) {
+        let read = lock.read().await;
+        black_box(read);
+    }
+
+    let lock = Arc::new(RwLock::new(()));
+    b.iter(|| {
+        let lock = lock.clone();
+        rt.block_on(async move {
+            tokio::join!{
+                task(lock.clone()),
+                task(lock.clone()),
+                task(lock.clone()),
+                task(lock.clone()),
+                task(lock.clone()),
+                task(lock.clone())
+            };
+
+        })
+    });
+}
+
+fn read_concurrent_contended_multi(b: &mut Bencher) {
+    let mut rt = tokio::runtime::Builder::new()
+        .core_threads(6)
+        .threaded_scheduler()
+        .build()
+        .unwrap();
+
+    async fn task(lock: Arc<RwLock<()>>) {
+        let read = lock.read().await;
+        black_box(read);
+    }
+
+    let lock = Arc::new(RwLock::new(()));
+    b.iter(|| {
+        let lock = lock.clone();
+        rt.block_on(async move {
+            let write = lock.write().await;
+            let j = tokio::try_join!{
+                async move { drop(write); Ok(()) },
+                task::spawn(task(lock.clone())),
+                task::spawn(task(lock.clone())),
+                task::spawn(task(lock.clone())),
+                task::spawn(task(lock.clone())),
+                task::spawn(task(lock.clone())),
+            };
+            j.unwrap();
+
+        })
+    });
+}
+
+fn read_concurrent_contended(b: &mut Bencher) {
+    let mut rt = tokio::runtime::Builder::new()
+        .basic_scheduler()
+        .build()
+        .unwrap();
+
+    async fn task(lock: Arc<RwLock<()>>) {
+        let read = lock.read().await;
+        black_box(read);
+    }
+
+    let lock = Arc::new(RwLock::new(()));
+    b.iter(|| {
+        let lock = lock.clone();
+        rt.block_on(async move {
+            let write = lock.write().await;
+            tokio::join!{
+                async move { drop(write) },
+                task(lock.clone()),
+                task(lock.clone()),
+                task(lock.clone()),
+                task(lock.clone()),
+                task(lock.clone()),
+            };
+        })
+    });
+}
+
+
+bencher::benchmark_group!(
+    sync_rwlock,
+    read_uncontended,
+    read_concurrent_uncontended,
+    read_concurrent_uncontended_multi,
+    read_concurrent_contended,
+    read_concurrent_contended_multi
+);
+
+bencher::benchmark_main!(sync_rwlock);

benches/sync_semaphore.rs

@@ -0,0 +1,136 @@
+use bencher::{black_box, Bencher};
+use tokio::{task, sync::Semaphore};
+use std::sync::Arc;
+
+fn uncontended(b: &mut Bencher) {
+    let mut rt = tokio::runtime::Builder::new()
+        .core_threads(6)
+        .threaded_scheduler()
+        .build()
+        .unwrap();
+
+    let s = Arc::new(Semaphore::new(10));
+    b.iter(|| {
+
+        let s = s.clone();
+        rt.block_on(async move {
+            for _ in 0..6 {
+                let permit = s.acquire().await;
+                drop(permit);
+            }
+        })
+    });
+}
+
+async fn task(s: Arc<Semaphore>) {
+    let permit = s.acquire().await;
+    drop(permit);
+}
+
+fn uncontended_concurrent_multi(b: &mut Bencher) {
+    let mut rt = tokio::runtime::Builder::new()
+        .core_threads(6)
+        .threaded_scheduler()
+        .build()
+        .unwrap();
+
+    let s = Arc::new(Semaphore::new(10));
+    b.iter(|| {
+        let s = s.clone();
+        rt.block_on(async move {
+            let j = tokio::try_join!{
+                task::spawn(task(s.clone())),
+                task::spawn(task(s.clone())),
+                task::spawn(task(s.clone())),
+                task::spawn(task(s.clone())),
+                task::spawn(task(s.clone())),
+                task::spawn(task(s.clone()))
+            };
+            j.unwrap();
+
+        })
+    });
+}
+
+fn uncontended_concurrent_single(b: &mut Bencher) {
+    let mut rt = tokio::runtime::Builder::new()
+        .basic_scheduler()
+        .build()
+        .unwrap();
+
+    let s = Arc::new(Semaphore::new(10));
+    b.iter(|| {
+        let s = s.clone();
+        rt.block_on(async move {
+            tokio::join!{
+                task(s.clone()),
+                task(s.clone()),
+                task(s.clone()),
+                task(s.clone()),
+                task(s.clone()),
+                task(s.clone())
+            };
+
+        })
+    });
+}
+
+
+fn contended_concurrent_multi(b: &mut Bencher) {
+    let mut rt = tokio::runtime::Builder::new()
+        .core_threads(6)
+        .threaded_scheduler()
+        .build()
+        .unwrap();
+
+    let s = Arc::new(Semaphore::new(5));
+    b.iter(|| {
+        let s = s.clone();
+        rt.block_on(async move {
+            let j = tokio::try_join!{
+                task::spawn(task(s.clone())),
+                task::spawn(task(s.clone())),
+                task::spawn(task(s.clone())),
+                task::spawn(task(s.clone())),
+                task::spawn(task(s.clone())),
+                task::spawn(task(s.clone()))
+            };
+            j.unwrap();
+
+        })
+    });
+}
+
+fn contended_concurrent_single(b: &mut Bencher) {
+    let mut rt = tokio::runtime::Builder::new()
+        .basic_scheduler()
+        .build()
+        .unwrap();
+
+    let s = Arc::new(Semaphore::new(5));
+    b.iter(|| {
+        let s = s.clone();
+        rt.block_on(async move {
+            tokio::join!{
+                task(s.clone()),
+                task(s.clone()),
+                task(s.clone()),
+                task(s.clone()),
+                task(s.clone()),
+                task(s.clone())
+            };
+
+        })
+    });
+}
+
+bencher::benchmark_group!(
+    sync_semaphore,
+    uncontended,
+    uncontended_concurrent_multi,
+    uncontended_concurrent_single,
+    contended_concurrent_multi,
+    contended_concurrent_single
+);
+
+bencher::benchmark_main!(sync_semaphore);