rt: Refactor Runtime::block_on to take &self

tokio-rs · Aug 20, 2020 · 013e8ef · 013e8ef
1 parent c393236
commit 013e8ef
Show file tree

Hide file tree

Showing 24 changed files with 149 additions and 141 deletions.
diff --git a/benches/mpsc.rs b/benches/mpsc.rs
@@ -43,7 +43,7 @@ fn send_large(b: &mut Bencher) {
 }
 
 fn contention_bounded(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()
@@ -70,7 +70,7 @@ fn contention_bounded(b: &mut Bencher) {
 }
 
 fn contention_bounded_full(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()
@@ -97,7 +97,7 @@ fn contention_bounded_full(b: &mut Bencher) {
 }
 
 fn contention_unbounded(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()
@@ -124,7 +124,7 @@ fn contention_unbounded(b: &mut Bencher) {
 }
 
 fn uncontented_bounded(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()
@@ -146,7 +146,7 @@ fn uncontented_bounded(b: &mut Bencher) {
 }
 
 fn uncontented_unbounded(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()

diff --git a/benches/scheduler.rs b/benches/scheduler.rs
@@ -13,7 +13,7 @@ use std::sync::{mpsc, Arc};
 fn spawn_many(b: &mut Bencher) {
     const NUM_SPAWN: usize = 10_000;
 
-    let mut rt = rt();
+    let rt = rt();
 
     let (tx, rx) = mpsc::sync_channel(1000);
     let rem = Arc::new(AtomicUsize::new(0));
@@ -68,7 +68,7 @@ fn yield_many(b: &mut Bencher) {
 fn ping_pong(b: &mut Bencher) {
     const NUM_PINGS: usize = 1_000;
 
-    let mut rt = rt();
+    let rt = rt();
 
     let (done_tx, done_rx) = mpsc::sync_channel(1000);
     let rem = Arc::new(AtomicUsize::new(0));
@@ -111,7 +111,7 @@ fn ping_pong(b: &mut Bencher) {
 fn chained_spawn(b: &mut Bencher) {
     const ITER: usize = 1_000;
 
-    let mut rt = rt();
+    let rt = rt();
 
     fn iter(done_tx: mpsc::SyncSender<()>, n: usize) {
         if n == 0 {

diff --git a/benches/spawn.rs b/benches/spawn.rs
@@ -10,7 +10,7 @@ async fn work() -> usize {
 }
 
 fn basic_scheduler_local_spawn(bench: &mut Bencher) {
-    let mut runtime = tokio::runtime::Builder::new()
+    let runtime = tokio::runtime::Builder::new()
         .basic_scheduler()
         .build()
         .unwrap();
@@ -23,7 +23,7 @@ fn basic_scheduler_local_spawn(bench: &mut Bencher) {
 }
 
 fn threaded_scheduler_local_spawn(bench: &mut Bencher) {
-    let mut runtime = tokio::runtime::Builder::new()
+    let runtime = tokio::runtime::Builder::new()
         .threaded_scheduler()
         .build()
         .unwrap();

diff --git a/benches/sync_rwlock.rs b/benches/sync_rwlock.rs
@@ -3,7 +3,7 @@ use std::sync::Arc;
 use tokio::{sync::RwLock, task};
 
 fn read_uncontended(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()
@@ -22,7 +22,7 @@ fn read_uncontended(b: &mut Bencher) {
 }
 
 fn read_concurrent_uncontended_multi(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()
@@ -51,7 +51,7 @@ fn read_concurrent_uncontended_multi(b: &mut Bencher) {
 }
 
 fn read_concurrent_uncontended(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .basic_scheduler()
         .build()
         .unwrap();
@@ -78,7 +78,7 @@ fn read_concurrent_uncontended(b: &mut Bencher) {
 }
 
 fn read_concurrent_contended_multi(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()
@@ -108,7 +108,7 @@ fn read_concurrent_contended_multi(b: &mut Bencher) {
 }
 
 fn read_concurrent_contended(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .basic_scheduler()
         .build()
         .unwrap();

diff --git a/benches/sync_semaphore.rs b/benches/sync_semaphore.rs
@@ -3,7 +3,7 @@ use std::sync::Arc;
 use tokio::{sync::Semaphore, task};
 
 fn uncontended(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()
@@ -27,7 +27,7 @@ async fn task(s: Arc<Semaphore>) {
 }
 
 fn uncontended_concurrent_multi(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()
@@ -51,7 +51,7 @@ fn uncontended_concurrent_multi(b: &mut Bencher) {
 }
 
 fn uncontended_concurrent_single(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .basic_scheduler()
         .build()
         .unwrap();
@@ -73,7 +73,7 @@ fn uncontended_concurrent_single(b: &mut Bencher) {
 }
 
 fn contended_concurrent_multi(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .core_threads(6)
         .threaded_scheduler()
         .build()
@@ -97,7 +97,7 @@ fn contended_concurrent_multi(b: &mut Bencher) {
 }
 
 fn contended_concurrent_single(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new()
         .basic_scheduler()
         .build()
         .unwrap();

diff --git a/tests-integration/tests/rt_shell.rs b/tests-integration/tests/rt_shell.rs
@@ -18,7 +18,7 @@ fn basic_shell_rt() {
     });
 
     for _ in 0..1_000 {
-        let mut rt = runtime::Builder::new().build().unwrap();
+        let rt = runtime::Builder::new().build().unwrap();
 
         let (tx, rx) = oneshot::channel();
 

diff --git a/tokio-test/src/lib.rs b/tokio-test/src/lib.rs
@@ -28,7 +28,7 @@ pub mod task;
 pub fn block_on<F: std::future::Future>(future: F) -> F::Output {
     use tokio::runtime;
 
-    let mut rt = runtime::Builder::new()
+    let rt = runtime::Builder::new()
         .basic_scheduler()
         .enable_all()
         .build()

diff --git a/tokio/src/runtime/blocking/mod.rs b/tokio/src/runtime/blocking/mod.rs
@@ -12,24 +12,25 @@ cfg_blocking_impl! {
     pub(crate) mod task;
 
     use crate::runtime::Builder;
+    use crate::loom::sync::Arc;
 
-    pub(crate) fn create_blocking_pool(builder: &Builder, thread_cap: usize) -> BlockingPool {
-        BlockingPool::new(builder, thread_cap)
-
+    pub(crate) fn create_blocking_pool(builder: &Builder, thread_cap: usize) -> Arc<BlockingPool> {
+        Arc::new(BlockingPool::new(builder, thread_cap))
     }
 }
 
 cfg_not_blocking_impl! {
     use crate::runtime::Builder;
     use std::time::Duration;
+    use crate::loom::sync::Arc;
 
     #[derive(Debug, Clone)]
     pub(crate) struct BlockingPool {}
 
     pub(crate) use BlockingPool as Spawner;
 
-    pub(crate) fn create_blocking_pool(_builder: &Builder, _thread_cap: usize) -> BlockingPool {
-        BlockingPool {}
+    pub(crate) fn create_blocking_pool(_builder: &Builder, _thread_cap: usize) -> Arc<BlockingPool> {
+        Arc::new(BlockingPool {})
     }
 
     impl BlockingPool {

diff --git a/tokio/src/runtime/builder.rs b/tokio/src/runtime/builder.rs
@@ -1,10 +1,9 @@
+use crate::loom::sync::{Arc, Mutex};
 use crate::runtime::handle::Handle;
 use crate::runtime::shell::Shell;
 use crate::runtime::{blocking, io, time, Callback, Runtime, Spawner};
 
 use std::fmt;
-#[cfg(not(loom))]
-use std::sync::Arc;
 
 /// Builds Tokio Runtime with custom configuration values.
 ///
@@ -263,7 +262,7 @@ impl Builder {
     where
         F: Fn() + Send + Sync + 'static,
     {
-        self.after_start = Some(Arc::new(f));
+        self.after_start = Some(std::sync::Arc::new(f));
         self
     }
 
@@ -303,7 +302,7 @@ impl Builder {
     /// ```
     /// use tokio::runtime::Builder;
     ///
-    /// let mut rt = Builder::new().build().unwrap();
+    /// let rt  = Builder::new().build().unwrap();
     ///
     /// rt.block_on(async {
     ///     println!("Hello from the Tokio runtime");
@@ -334,7 +333,7 @@ impl Builder {
         let blocking_spawner = blocking_pool.spawner().clone();
 
         Ok(Runtime {
-            kind: Kind::Shell(Shell::new(driver)),
+            kind: Kind::Shell(Arc::new(Mutex::new(Shell::new(driver)))),
             handle: Handle {
                 spawner,
                 io_handle,
@@ -433,7 +432,7 @@ cfg_rt_core! {
             let blocking_spawner = blocking_pool.spawner().clone();
 
             Ok(Runtime {
-                kind: Kind::Basic(scheduler),
+                kind: Kind::Basic(Arc::new(Mutex::new(scheduler))),
                 handle: Handle {
                     spawner,
                     io_handle,
@@ -493,7 +492,7 @@ cfg_rt_threaded! {
             handle.enter(|| launch.launch());
 
             Ok(Runtime {
-                kind: Kind::ThreadPool(scheduler),
+                kind: Kind::ThreadPool(Arc::new(scheduler)),
                 handle,
                 blocking_pool,
             })

diff --git a/tokio/src/runtime/handle.rs b/tokio/src/runtime/handle.rs
@@ -244,7 +244,7 @@ cfg_rt_core! {
         /// use std::thread;
         ///
         /// // Create the runtime.
-        /// let mut rt = Builder::new()
+        /// let rt  = Builder::new()
         ///     .enable_all()
         ///     .basic_scheduler()
         ///     .build()