From d35d0518f5b49ea2b301a44d31933a3b64a193b8 Mon Sep 17 00:00:00 2001 From: Stjepan Glavina Date: Wed, 3 Oct 2018 03:19:27 +0200 Subject: [PATCH] runtime: create reactor per worker (#660) --- Cargo.toml | 1 + src/lib.rs | 1 + src/runtime/builder.rs | 65 ++++++++++++++++++------------ src/runtime/mod.rs | 21 ++++------ src/runtime/shutdown.rs | 16 ++------ tokio-threadpool/src/worker/mod.rs | 8 ++++ 6 files changed, 60 insertions(+), 52 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 39f69b496..dfc8ddb4f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -54,6 +54,7 @@ appveyor = { repository = "carllerche/tokio", id = "s83yxhy9qeb58va7" } [dependencies] bytes = "0.4" +num_cpus = "1.8.0" tokio-codec = { version = "0.1.0", path = "tokio-codec" } tokio-current-thread = { version = "0.1.3", path = "tokio-current-thread" } tokio-io = { version = "0.1.6", path = "tokio-io" } diff --git a/src/lib.rs b/src/lib.rs index f652e53a9..950651dd7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -76,6 +76,7 @@ extern crate bytes; #[macro_use] extern crate futures; extern crate mio; +extern crate num_cpus; extern crate tokio_current_thread; extern crate tokio_io; extern crate tokio_executor; diff --git a/src/runtime/builder.rs b/src/runtime/builder.rs index d09f08bf6..b4b0346da 100644 --- a/src/runtime/builder.rs +++ b/src/runtime/builder.rs @@ -3,11 +3,12 @@ use runtime::{Inner, Runtime}; use reactor::Reactor; use std::io; +use std::sync::Mutex; use std::time::Duration; +use num_cpus; use tokio_reactor; use tokio_threadpool::Builder as ThreadPoolBuilder; -use tokio_threadpool::park::DefaultPark; use tokio_timer::clock::{self, Clock}; use tokio_timer::timer::{self, Timer}; @@ -51,6 +52,9 @@ pub struct Builder { /// Thread pool specific builder threadpool_builder: ThreadPoolBuilder, + /// The number of worker threads + core_threads: usize, + /// The clock to use clock: Clock, } @@ -61,11 +65,15 @@ impl Builder { /// /// Configuration methods can be chained on the return value. pub fn new() -> Builder { + let core_threads = num_cpus::get().max(1); + let mut threadpool_builder = ThreadPoolBuilder::new(); threadpool_builder.name_prefix("tokio-runtime-worker-"); + threadpool_builder.pool_size(core_threads); Builder { threadpool_builder, + core_threads, clock: Clock::new(), } } @@ -110,6 +118,7 @@ impl Builder { /// # } /// ``` pub fn core_threads(&mut self, val: usize) -> &mut Self { + self.core_threads = val; self.threadpool_builder.pool_size(val); self } @@ -243,44 +252,50 @@ impl Builder { /// # } /// ``` pub fn build(&mut self) -> io::Result { - use std::collections::HashMap; - use std::sync::{Arc, Mutex}; + // TODO(stjepang): Once we remove the `threadpool_builder` method, remove this line too. + self.threadpool_builder.pool_size(self.core_threads); + + let mut reactor_handles = Vec::new(); + let mut timer_handles = Vec::new(); + let mut timers = Vec::new(); + + for _ in 0..self.core_threads { + // Create a new reactor. + let reactor = Reactor::new()?; + reactor_handles.push(reactor.handle()); + + // Create a new timer. + let timer = Timer::new_with_now(reactor, self.clock.clone()); + timer_handles.push(timer.handle()); + timers.push(Mutex::new(Some(timer))); + } // Get a handle to the clock for the runtime. - let clock1 = self.clock.clone(); - let clock2 = clock1.clone(); + let clock = self.clock.clone(); - let timers = Arc::new(Mutex::new(HashMap::<_, timer::Handle>::new())); - let t1 = timers.clone(); - - // Spawn a reactor on a background thread. - let reactor = Reactor::new()?.background()?; - - // Get a handle to the reactor. - let reactor_handle = reactor.handle().clone(); + // Get a handle to the first reactor. + let reactor = reactor_handles[0].clone(); let pool = self.threadpool_builder .around_worker(move |w, enter| { - let timer_handle = t1.lock().unwrap() - .get(w.id()).unwrap() - .clone(); + let index = w.id().to_usize(); - tokio_reactor::with_default(&reactor_handle, enter, |enter| { - clock::with_default(&clock1, enter, |enter| { - timer::with_default(&timer_handle, enter, |_| { + tokio_reactor::with_default(&reactor_handles[index], enter, |enter| { + clock::with_default(&clock, enter, |enter| { + timer::with_default(&timer_handles[index], enter, |_| { w.run(); }); }) }); }) .custom_park(move |worker_id| { - // Create a new timer - let timer = Timer::new_with_now(DefaultPark::new(), clock2.clone()); + let index = worker_id.to_usize(); - timers.lock().unwrap() - .insert(worker_id.clone(), timer.handle()); - - timer + timers[index] + .lock() + .unwrap() + .take() + .unwrap() }) .build(); diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 9ff0cc4c2..0c9be1639 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -121,7 +121,7 @@ pub use self::builder::Builder; pub use self::shutdown::Shutdown; pub use self::task_executor::TaskExecutor; -use reactor::{Background, Handle}; +use reactor::Handle; use std::io; @@ -152,8 +152,8 @@ pub struct Runtime { #[derive(Debug)] struct Inner { - /// Reactor running on a background thread. - reactor: Background, + /// A handle to one of the per-worker reactors. + reactor: Handle, /// Task execution pool. pool: threadpool::ThreadPool, @@ -254,6 +254,7 @@ impl Runtime { #[deprecated(since = "0.1.5", note = "use `reactor` instead")] #[doc(hidden)] pub fn handle(&self) -> &Handle { + #[allow(deprecated)] self.reactor() } @@ -275,8 +276,9 @@ impl Runtime { /// /// // use `reactor_handle` /// ``` + #[deprecated(since = "0.1.11", note = "there is now a reactor per worker thread")] pub fn reactor(&self) -> &Handle { - self.inner().reactor.handle() + &self.inner().reactor } /// Return a handle to the runtime's executor. @@ -424,16 +426,7 @@ impl Runtime { /// [mod]: index.html pub fn shutdown_on_idle(mut self) -> Shutdown { let inner = self.inner.take().unwrap(); - - let inner = Box::new({ - let pool = inner.pool; - let reactor = inner.reactor; - - pool.shutdown_on_idle().and_then(|_| { - reactor.shutdown_on_idle() - }) - }); - + let inner = inner.pool.shutdown_on_idle(); Shutdown { inner } } diff --git a/src/runtime/shutdown.rs b/src/runtime/shutdown.rs index 1aca55727..1b6ba4656 100644 --- a/src/runtime/shutdown.rs +++ b/src/runtime/shutdown.rs @@ -1,4 +1,5 @@ use runtime::Inner; +use tokio_threadpool as threadpool; use std::fmt; @@ -6,23 +7,12 @@ use futures::{Future, Poll}; /// A future that resolves when the Tokio `Runtime` is shut down. pub struct Shutdown { - pub(super) inner: Box + Send>, + pub(super) inner: threadpool::Shutdown, } impl Shutdown { pub(super) fn shutdown_now(inner: Inner) -> Self { - let inner = Box::new({ - let pool = inner.pool; - let reactor = inner.reactor; - - pool.shutdown_now().and_then(|_| { - reactor.shutdown_now() - .then(|_| { - Ok(()) - }) - }) - }); - + let inner = inner.pool.shutdown_now(); Shutdown { inner } } } diff --git a/tokio-threadpool/src/worker/mod.rs b/tokio-threadpool/src/worker/mod.rs index 3ac7f039b..729316499 100644 --- a/tokio-threadpool/src/worker/mod.rs +++ b/tokio-threadpool/src/worker/mod.rs @@ -892,4 +892,12 @@ impl WorkerId { pub(crate) fn new(idx: usize) -> WorkerId { WorkerId(idx) } + + /// Returns this identifier represented as an integer. + /// + /// Worker identifiers in a single thread pool are guaranteed to correspond to integers in the + /// range `0..pool_size`. + pub fn to_usize(&self) -> usize { + self.0 + } }