sync: new internal semaphore based on intrusive lists

tokio/src/sync/mod.rs

@@ -435,6 +435,7 @@ cfg_sync! {
 
     pub mod oneshot;
 
+    pub(crate) mod batch_semaphore;
     pub(crate) mod semaphore_ll;
     mod semaphore;
     pub use semaphore::{Semaphore, SemaphorePermit};

tokio/src/sync/mutex.rs

@@ -78,9 +78,7 @@
 //!
 //! [`Mutex`]: struct.Mutex.html
 //! [`MutexGuard`]: struct.MutexGuard.html
-
-use crate::future::poll_fn;
-use crate::sync::semaphore_ll as semaphore;
+use crate::sync::batch_semaphore as semaphore;
 
 use std::cell::UnsafeCell;
 use std::error::Error;
@@ -152,18 +150,13 @@ impl<T> Mutex<T> {
 
     /// A future that resolves on acquiring the lock and returns the `MutexGuard`.
     pub async fn lock(&self) -> MutexGuard<'_, T> {
-        let mut guard = MutexGuard {
-            lock: self,
-            permit: semaphore::Permit::new(),
-        };
-        poll_fn(|cx| guard.permit.poll_acquire(cx, 1, &self.s))
-            .await
-            .unwrap_or_else(|_| {
-                // The semaphore was closed. but, we never explicitly close it, and we have a
-                // handle to it through the Arc, which means that this can never happen.
-                unreachable!()
-            });
-        guard
+        let mut permit = semaphore::Permit::new();
+        permit.acquire(1, &self.s).await.unwrap_or_else(|_| {
+            // The semaphore was closed. but, we never explicitly close it, and we have a
+            // handle to it through the Arc, which means that this can never happen.
+            unreachable!()
+        });
+        MutexGuard { lock: self, permit }
     }
 
     /// Tries to acquire the lock

tokio/src/sync/rwlock.rs

@@ -1,8 +1,6 @@
-use crate::future::poll_fn;
-use crate::sync::semaphore_ll::{AcquireError, Permit, Semaphore};
+use crate::sync::batch_semaphore::{AcquireError, Permit, Semaphore};
 use std::cell::UnsafeCell;
 use std::ops;
-use std::task::{Context, Poll};
 
 #[cfg(not(loom))]
 const MAX_READS: usize = 32;
@@ -114,12 +112,8 @@ struct ReleasingPermit<'a, T> {
 }
 
 impl<'a, T> ReleasingPermit<'a, T> {
-    fn poll_acquire(
-        &mut self,
-        cx: &mut Context<'_>,
-        s: &Semaphore,
-    ) -> Poll<Result<(), AcquireError>> {
-        self.permit.poll_acquire(cx, self.num_permits, s)
+    async fn acquire(&mut self) -> Result<(), AcquireError> {
+        self.permit.acquire(self.num_permits, &self.lock.s).await
     }
 }
 
@@ -191,14 +185,11 @@ impl<T> RwLock<T> {
             permit: Permit::new(),
             lock: self,
         };
-
-        poll_fn(|cx| permit.poll_acquire(cx, &self.s))
-            .await
-            .unwrap_or_else(|_| {
-                // The semaphore was closed. but, we never explicitly close it, and we have a
-                // handle to it through the Arc, which means that this can never happen.
-                unreachable!()
-            });
+        permit.acquire().await.unwrap_or_else(|_| {
+            // The semaphore was closed. but, we never explicitly close it, and we have a
+            // handle to it through the Arc, which means that this can never happen.
+            unreachable!()
+        });
         RwLockReadGuard { lock: self, permit }
     }
 
@@ -231,13 +222,11 @@ impl<T> RwLock<T> {
             lock: self,
         };
 
-        poll_fn(|cx| permit.poll_acquire(cx, &self.s))
-            .await
-            .unwrap_or_else(|_| {
-                // The semaphore was closed. but, we never explicitly close it, and we have a
-                // handle to it through the Arc, which means that this can never happen.
-                unreachable!()
-            });
+        permit.acquire().await.unwrap_or_else(|_| {
+            // The semaphore was closed. but, we never explicitly close it, and we have a
+            // handle to it through the Arc, which means that this can never happen.
+            unreachable!()
+        });
 
         RwLockWriteGuard { lock: self, permit }
     }

tokio/src/sync/semaphore.rs

@@ -1,5 +1,4 @@
-use super::semaphore_ll as ll; // low level implementation
-use crate::future::poll_fn;
+use super::batch_semaphore as ll; // low level implementation
 
 /// Counting semaphore performing asynchronous permit aquisition.
 ///
@@ -56,14 +55,12 @@ impl Semaphore {
 
     /// Acquires permit from the semaphore
     pub async fn acquire(&self) -> SemaphorePermit<'_> {
-        let mut permit = SemaphorePermit {
+        let mut ll_permit = ll::Permit::new();
+        ll_permit.acquire(1, &self.ll_sem).await.unwrap();
+        SemaphorePermit {
             sem: &self,
-            ll_permit: ll::Permit::new(),
-        };
-        poll_fn(|cx| permit.ll_permit.poll_acquire(cx, 1, &self.ll_sem))
-            .await
-            .unwrap();
-        permit
+            ll_permit,
+        }
     }
 
     /// Tries to acquire a permit form the semaphore

tokio/src/sync/semaphore_ll.rs

@@ -610,6 +610,7 @@ impl Permit {
     }
 
     /// Returns `true` if the permit has been acquired
+    #[allow(dead_code)] // may be used later
     pub(crate) fn is_acquired(&self) -> bool {
         match self.state {
             PermitState::Acquired(num) if num > 0 => true,

tokio/src/sync/tests/loom_semaphore_batch.rs

@@ -0,0 +1,192 @@
+use crate::sync::batch_semaphore::*;
+
+use futures::future::poll_fn;
+use loom::future::block_on;
+use loom::thread;
+use std::future::Future;
+use std::pin::Pin;
+use loom::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering::SeqCst;
+use std::sync::Arc;
+use std::task::Poll::Ready;
+use std::task::{Context, Poll};
+
+#[test]
+fn basic_usage() {
+    const NUM: usize = 2;
+
+    struct Shared {
+        semaphore: Semaphore,
+        active: AtomicUsize,
+    }
+
+    async fn actor(shared: Arc<Shared>) {
+        let mut permit = Permit::new();
+        permit.acquire(1, &shared.semaphore).await.unwrap();
+        println!("acquired!");
+        let actual = shared.active.fetch_add(1, SeqCst);
+        assert!(actual <= NUM - 1);
+
+        let actual = shared.active.fetch_sub(1, SeqCst);
+        assert!(actual <= NUM);
+
+        permit.release(1, &shared.semaphore);
+    }
+
+    loom::model(|| {
+
+        let shared = Arc::new(Shared {
+            semaphore: Semaphore::new(NUM),
+            active: AtomicUsize::new(0),
+        });
+
+        for _ in 0..NUM {
+            let shared = shared.clone();
+
+            thread::spawn(move || {
+                block_on(actor(shared));
+            });
+        }
+
+        block_on(actor(shared));
+    });
+}
+
+#[test]
+fn release() {
+    loom::model(|| {
+        let semaphore = Arc::new(Semaphore::new(1));
+
+        {
+            let semaphore = semaphore.clone();
+            thread::spawn(move || {
+                let mut permit = Permit::new();
+
+                block_on(permit.acquire(1, &semaphore)).unwrap();
+
+                permit.release(1, &semaphore);
+            });
+        }
+
+        let mut permit = Permit::new();
+
+        block_on(permit.acquire(1, &semaphore)).unwrap();
+
+        permit.release(1, &semaphore);
+    });
+}
+
+#[test]
+fn basic_closing() {
+    const NUM: usize = 2;
+
+    loom::model(|| {
+        println!("-- iter --");
+        let semaphore = Arc::new(Semaphore::new(1));
+
+        for _ in 0..NUM {
+            let semaphore = semaphore.clone();
+
+            thread::spawn(move || {
+                let mut permit = Permit::new();
+
+                for _ in 0..2 {
+                    block_on(permit.acquire(1, &semaphore)).map_err(|_|())?;
+
+                    permit.release(1, &semaphore);
+                }
+
+                Ok::<(), ()>(())
+            });
+        }
+
+        semaphore.close();
+    });
+}
+
+#[test]
+fn concurrent_close() {
+    const NUM: usize = 3;
+
+    loom::model(|| {
+        let semaphore = Arc::new(Semaphore::new(1));
+
+        for _ in 0..NUM {
+            let semaphore = semaphore.clone();
+
+            thread::spawn(move || {
+                let mut permit = Permit::new();
+
+                block_on(permit.acquire(1, &semaphore)).map_err(|_|())?;
+
+                permit.release(1, &semaphore);
+
+                semaphore.close();
+
+                Ok::<(), ()>(())
+            });
+        }
+    });
+}
+
+#[test]
+fn batch() {
+    let mut b = loom::model::Builder::new();
+    b.preemption_bound = Some(1);
+
+    b.check(|| {
+        let semaphore = Arc::new(Semaphore::new(10));
+        let active = Arc::new(AtomicUsize::new(0));
+        let mut ths = vec![];
+
+        for _ in 0..2 {
+            let semaphore = semaphore.clone();
+            let active = active.clone();
+
+            ths.push(thread::spawn(move || {
+                let mut permit = Permit::new();
+
+                for n in &[4, 10, 8] {
+                    block_on(permit.acquire(*n, &semaphore)).unwrap();
+
+                    active.fetch_add(*n as usize, SeqCst);
+
+                    let num_active = active.load(SeqCst);
+                    assert!(num_active <= 10);
+
+                    thread::yield_now();
+
+                    active.fetch_sub(*n as usize, SeqCst);
+
+                    permit.release(*n, &semaphore);
+                }
+            }));
+        }
+
+        for th in ths.into_iter() {
+            th.join().unwrap();
+        }
+
+        assert_eq!(10, semaphore.available_permits());
+    });
+}
+
+#[test]
+fn release_during_acquire() {
+    loom::model(|| {
+        let semaphore = Arc::new(Semaphore::new(10));
+        let mut permit1 = Permit::new();
+        permit1.try_acquire(8, &semaphore).expect("try_acquire should succeed; semaphore uncontended");
+        let semaphore2 = semaphore.clone();
+        let thread = thread::spawn(move || {
+            let mut permit = Permit::new();
+            block_on(permit.acquire(4, &semaphore2)).unwrap();
+            permit
+        });
+
+        permit1.release(8, &semaphore);
+        let mut permit2 = thread.join().unwrap();
+        permit2.release(4, &semaphore);
+        assert_eq!(10, semaphore.available_permits());
+    })
+}

tokio/src/sync/tests/mod.rs

@@ -1,6 +1,7 @@
 cfg_not_loom! {
     mod atomic_waker;
     mod semaphore_ll;
+    mod semaphore_batch;
 }
 
 cfg_loom! {
@@ -10,5 +11,6 @@ cfg_loom! {
     mod loom_mpsc;
     mod loom_notify;
     mod loom_oneshot;
+    mod loom_semaphore_batch;
     mod loom_semaphore_ll;
 }