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,147 @@
+use bencher::{black_box, Bencher};
+use std::sync::Arc;
+use tokio::{sync::RwLock, task};
+
+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,130 @@
+use bencher::Bencher;
+use std::sync::Arc;
+use tokio::{sync::Semaphore, task};
+
+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);

tokio/src/coop.rs

@@ -46,6 +46,8 @@
 // NOTE: The doctests in this module are ignored since the whole module is (currently) private.
 
 use std::cell::Cell;
+use std::future::Future;
+use std::pin::Pin;
 use std::task::{Context, Poll};
 
 /// Constant used to determine how much "work" a task is allowed to do without yielding.
@@ -250,6 +252,74 @@ pub async fn proceed() {
     poll_fn(|cx| poll_proceed(cx)).await;
 }
 
+pin_project_lite::pin_project! {
+    /// A future that cooperatively yields to the task scheduler when polling,
+    /// if the task's budget is exhausted.
+    ///
+    /// Internally, this is simply a future combinator which calls
+    /// [`poll_proceed`] in its `poll` implementation before polling the wrapped
+    /// future.
+    ///
+    /// # Examples
+    ///
+    /// ```rust,ignore
+    /// # #[tokio::main]
+    /// # async fn main() {
+    /// use tokio::coop::CoopFutureExt;
+    ///
+    /// async {  /* ... */ }
+    ///     .cooperate()
+    ///     .await;
+    /// # }
+    /// ```
+    ///
+    /// [`poll_proceed`]: fn.poll_proceed.html
+    #[derive(Debug)]
+    #[allow(unreachable_pub, dead_code)]
+    pub struct CoopFuture<F> {
+        #[pin]
+        future: F,
+    }
+}
+
+impl<F: Future> Future for CoopFuture<F> {
+    type Output = F::Output;
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        ready!(poll_proceed(cx));
+        self.project().future.poll(cx)
+    }
+}
+
+impl<F: Future> CoopFuture<F> {
+    /// Returns a new `CoopFuture` wrapping the given future.
+    ///
+    #[allow(unreachable_pub, dead_code)]
+    pub fn new(future: F) -> Self {
+        Self { future }
+    }
+}
+
+// Currently only used by `tokio::sync`; and if we make this combinator public,
+// it should probably be on the `FutureExt` trait instead.
+cfg_sync! {
+    /// Extension trait providing `Future::cooperate` extension method.
+    ///
+    /// Note: if/when the co-op API becomes public, this method should probably be
+    /// provided by `FutureExt`, instead.
+    pub(crate) trait CoopFutureExt: Future {
+        /// Wrap `self` to cooperatively yield to the scheduler when polling, if the
+        /// task's budget is exhausted.
+        fn cooperate(self) -> CoopFuture<Self>
+        where
+            Self: Sized,
+        {
+            CoopFuture::new(self)
+        }
+    }
+
+    impl<F> CoopFutureExt for F where F: Future {}
+}
+
 #[cfg(all(test, not(loom)))]
 mod test {
     use super::*;