From 494f0dc176ac938ffd0fa020c30cb8f9665366ed Mon Sep 17 00:00:00 2001 From: Roman Date: Wed, 21 Mar 2018 21:23:36 +0300 Subject: [PATCH] Runtime builder (#234) * Split runtime module into files * Add runtime::Builder to set up thread pool. --- src/runtime/builder.rs | 107 +++++++++++++++++ src/{runtime.rs => runtime/mod.rs} | 179 +++-------------------------- src/runtime/shutdown.rs | 46 ++++++++ src/runtime/task_executor.rs | 98 ++++++++++++++++ 4 files changed, 265 insertions(+), 165 deletions(-) create mode 100644 src/runtime/builder.rs rename src/{runtime.rs => runtime/mod.rs} (69%) create mode 100644 src/runtime/shutdown.rs create mode 100644 src/runtime/task_executor.rs diff --git a/src/runtime/builder.rs b/src/runtime/builder.rs new file mode 100644 index 000000000..0b9e759fa --- /dev/null +++ b/src/runtime/builder.rs @@ -0,0 +1,107 @@ +use runtime::{Inner, Runtime}; + +use reactor::Reactor; + +use std::io; + +use tokio_threadpool::Builder as ThreadPoolBuilder; + + + +/// Builds Tokio Runtime with custom configuration values. +/// +/// Methods can be chanined in order to set the configuration values. The +/// Runtime is constructed by calling [`build`]. +/// +/// New instances of `Builder` are obtained via [`Builder::new`]. +/// +/// See function level documentation for details on the various configuration +/// settings. +/// +/// [`build`]: #method.build +/// [`Builder::new`]: #method.new +/// +/// # Examples +/// +/// ``` +/// # extern crate tokio; +/// # extern crate tokio_threadpool; +/// # use tokio::runtime::Builder; +/// +/// # pub fn main() { +/// // create and configure ThreadPool +/// let mut threadpool_builder = tokio_threadpool::Builder::new(); +/// threadpool_builder +/// .name_prefix("my-runtime-worker-") +/// .pool_size(4); +/// +/// // build Runtime +/// let runtime = Builder::new() +/// .threadpool_builder(threadpool_builder) +/// .build(); +/// // ... call runtime.run(...) +/// # let _ = runtime; +/// # } +/// ``` +#[derive(Debug)] +pub struct Builder { + /// Thread pool specific builder + threadpool_builder: ThreadPoolBuilder, +} + +impl Builder { + /// Returns a new runtime builder initialized with default configuration + /// values. + /// + /// Configuration methods can be chained on the return value. + pub fn new() -> Builder { + let mut threadpool_builder = ThreadPoolBuilder::new(); + threadpool_builder.name_prefix("tokio-runtime-worker-"); + + Builder { threadpool_builder } + } + + /// Set builder to set up the thread pool instance. + pub fn threadpool_builder(&mut self, val: ThreadPoolBuilder) -> &mut Self { + self.threadpool_builder = val; + self + } + + /// Create the configured `Runtime`. + /// + /// The returned `ThreadPool` instance is ready to spawn tasks. + /// + /// # Examples + /// + /// ``` + /// # extern crate tokio; + /// # use tokio::runtime::Builder; + /// # pub fn main() { + /// let runtime = Builder::new().build(); + /// // ... call runtime.run(...) + /// # let _ = runtime; + /// # } + /// ``` + pub fn build(&mut self) -> io::Result { + // Spawn a reactor on a background thread. + let reactor = Reactor::new()?.background()?; + + // Get a handle to the reactor. + let handle = reactor.handle().clone(); + + let pool = self.threadpool_builder + .around_worker(move |w, enter| { + ::tokio_reactor::with_default(&handle, enter, |_| { + w.run(); + }); + }) + .build(); + + Ok(Runtime { + inner: Some(Inner { + reactor, + pool, + }), + }) + } +} diff --git a/src/runtime.rs b/src/runtime/mod.rs similarity index 69% rename from src/runtime.rs rename to src/runtime/mod.rs index dcc25fc2e..037473579 100644 --- a/src/runtime.rs +++ b/src/runtime/mod.rs @@ -104,14 +104,21 @@ //! [idle]: struct.Runtime.html#method.shutdown_on_idle //! [`tokio::spawn`]: ../executor/fn.spawn.html -use reactor::{Reactor, Handle, Background}; +mod builder; +mod shutdown; +mod task_executor; -use tokio_threadpool::{self as threadpool, ThreadPool, Sender}; -use futures::Poll; -use futures::future::{self, Future}; +pub use self::builder::Builder; +pub use self::shutdown::Shutdown; +pub use self::task_executor::TaskExecutor; -use std::{fmt, io}; +use reactor::{Background, Handle}; +use std::io; + +use tokio_threadpool as threadpool; + +use futures::future::Future; #[cfg(feature = "unstable-futures")] use futures2; @@ -128,29 +135,13 @@ pub struct Runtime { inner: Option, } -/// Executes futures on the runtime -/// -/// All futures spawned using this executor will be submitted to the associated -/// Runtime's executor. This executor is usually a thread pool. -/// -/// For more details, see the [module level](index.html) documentation. -#[derive(Debug, Clone)] -pub struct TaskExecutor { - inner: Sender, -} - -/// A future that resolves when the Tokio `Runtime` is shut down. -pub struct Shutdown { - inner: Box + Send>, -} - #[derive(Debug)] struct Inner { /// Reactor running on a background thread. reactor: Background, /// Task execution pool. - pool: ThreadPool, + pool: threadpool::ThreadPool, } // ===== impl Runtime ===== @@ -227,27 +218,7 @@ impl Runtime { /// /// [mod]: index.html pub fn new() -> io::Result { - // Spawn a reactor on a background thread. - let reactor = Reactor::new()?.background()?; - - // Get a handle to the reactor. - let handle = reactor.handle().clone(); - - let pool = threadpool::Builder::new() - .name_prefix("tokio-runtime-worker-") - .around_worker(move |w, enter| { - ::tokio_reactor::with_default(&handle, enter, |_| { - w.run(); - }); - }) - .build(); - - Ok(Runtime { - inner: Some(Inner { - reactor, - pool, - }), - }) + Builder::new().build() } /// Return a reference to the reactor handle for this runtime instance. @@ -388,125 +359,3 @@ impl Drop for Runtime { } } } - -// ===== impl TaskExecutor ===== - -impl TaskExecutor { - /// Spawn a future onto the Tokio runtime. - /// - /// This spawns the given future onto the runtime's executor, usually a - /// thread pool. The thread pool is then responsible for polling the future - /// until it completes. - /// - /// See [module level][mod] documentation for more details. - /// - /// [mod]: index.html - /// - /// # Examples - /// - /// ```rust - /// # extern crate tokio; - /// # extern crate futures; - /// # use futures::{future, Future, Stream}; - /// use tokio::runtime::Runtime; - /// - /// # fn dox() { - /// // Create the runtime - /// let mut rt = Runtime::new().unwrap(); - /// let executor = rt.executor(); - /// - /// // Spawn a future onto the runtime - /// executor.spawn(future::lazy(|| { - /// println!("now running on a worker thread"); - /// Ok(()) - /// })); - /// # } - /// # pub fn main() {} - /// ``` - /// - /// # Panics - /// - /// This function panics if the spawn fails. Failure occurs if the executor - /// is currently at capacity and is unable to spawn a new future. - pub fn spawn(&self, future: F) - where F: Future + Send + 'static, - { - self.inner.spawn(future).unwrap(); - } -} - -impl future::Executor for TaskExecutor -where T: Future + Send + 'static, -{ - fn execute(&self, future: T) -> Result<(), future::ExecuteError> { - self.inner.execute(future) - } -} - -impl ::executor::Executor for TaskExecutor { - fn spawn(&mut self, future: Box + Send>) - -> Result<(), ::executor::SpawnError> - { - self.inner.spawn(future) - } - - #[cfg(feature = "unstable-futures")] - fn spawn2(&mut self, future: Box + Send>) - -> Result<(), futures2::executor::SpawnError> - { - self.inner.spawn2(future) - } -} - -#[cfg(feature = "unstable-futures")] -type Task2 = Box + Send>; - -#[cfg(feature = "unstable-futures")] -impl futures2::executor::Executor for TaskExecutor { - fn spawn(&mut self, f: Task2) -> Result<(), futures2::executor::SpawnError> { - futures2::executor::Executor::spawn(&mut self.inner, f) - } - - fn status(&self) -> Result<(), futures2::executor::SpawnError> { - futures2::executor::Executor::status(&self.inner) - } -} - - -// ===== impl Shutdown ===== - -impl Shutdown { - 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(()) - }) - }) - }); - - Shutdown { inner } - } -} - -impl Future for Shutdown { - type Item = (); - type Error = (); - - fn poll(&mut self) -> Poll<(), ()> { - try_ready!(self.inner.poll()); - Ok(().into()) - } -} - -impl fmt::Debug for Shutdown { - fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { - fmt.debug_struct("Shutdown") - .field("inner", &"Box>") - .finish() - } -} diff --git a/src/runtime/shutdown.rs b/src/runtime/shutdown.rs new file mode 100644 index 000000000..1aca55727 --- /dev/null +++ b/src/runtime/shutdown.rs @@ -0,0 +1,46 @@ +use runtime::Inner; + +use std::fmt; + +use futures::{Future, Poll}; + +/// A future that resolves when the Tokio `Runtime` is shut down. +pub struct Shutdown { + pub(super) inner: Box + Send>, +} + +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(()) + }) + }) + }); + + Shutdown { inner } + } +} + +impl Future for Shutdown { + type Item = (); + type Error = (); + + fn poll(&mut self) -> Poll<(), ()> { + try_ready!(self.inner.poll()); + Ok(().into()) + } +} + +impl fmt::Debug for Shutdown { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("Shutdown") + .field("inner", &"Box>") + .finish() + } +} diff --git a/src/runtime/task_executor.rs b/src/runtime/task_executor.rs new file mode 100644 index 000000000..ed918be5f --- /dev/null +++ b/src/runtime/task_executor.rs @@ -0,0 +1,98 @@ + +use tokio_threadpool::Sender; + +use futures::future::{self, Future}; +#[cfg(feature = "unstable-futures")] +use futures2; + +/// Executes futures on the runtime +/// +/// All futures spawned using this executor will be submitted to the associated +/// Runtime's executor. This executor is usually a thread pool. +/// +/// For more details, see the [module level](index.html) documentation. +#[derive(Debug, Clone)] +pub struct TaskExecutor { + pub(super) inner: Sender, +} + +impl TaskExecutor { + /// Spawn a future onto the Tokio runtime. + /// + /// This spawns the given future onto the runtime's executor, usually a + /// thread pool. The thread pool is then responsible for polling the future + /// until it completes. + /// + /// See [module level][mod] documentation for more details. + /// + /// [mod]: index.html + /// + /// # Examples + /// + /// ```rust + /// # extern crate tokio; + /// # extern crate futures; + /// # use futures::{future, Future, Stream}; + /// use tokio::runtime::Runtime; + /// + /// # fn dox() { + /// // Create the runtime + /// let mut rt = Runtime::new().unwrap(); + /// let executor = rt.executor(); + /// + /// // Spawn a future onto the runtime + /// executor.spawn(future::lazy(|| { + /// println!("now running on a worker thread"); + /// Ok(()) + /// })); + /// # } + /// # pub fn main() {} + /// ``` + /// + /// # Panics + /// + /// This function panics if the spawn fails. Failure occurs if the executor + /// is currently at capacity and is unable to spawn a new future. + pub fn spawn(&self, future: F) + where F: Future + Send + 'static, + { + self.inner.spawn(future).unwrap(); + } +} + +impl future::Executor for TaskExecutor +where T: Future + Send + 'static, +{ + fn execute(&self, future: T) -> Result<(), future::ExecuteError> { + self.inner.execute(future) + } +} + +impl ::executor::Executor for TaskExecutor { + fn spawn(&mut self, future: Box + Send>) + -> Result<(), ::executor::SpawnError> + { + self.inner.spawn(future) + } + + #[cfg(feature = "unstable-futures")] + fn spawn2(&mut self, future: Box + Send>) + -> Result<(), futures2::executor::SpawnError> + { + self.inner.spawn2(future) + } +} + +#[cfg(feature = "unstable-futures")] +type Task2 = Box + Send>; + +#[cfg(feature = "unstable-futures")] +impl futures2::executor::Executor for TaskExecutor { + fn spawn(&mut self, f: Task2) -> Result<(), futures2::executor::SpawnError> { + futures2::executor::Executor::spawn(&mut self.inner, f) + } + + fn status(&self) -> Result<(), futures2::executor::SpawnError> { + futures2::executor::Executor::status(&self.inner) + } +}