tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-16 00:00:12 +02:00
Code Issues Packages Projects Releases Wiki Activity

Runtime builder (#234)

Browse Source 
* Split runtime module into files
* Add runtime::Builder to set up thread pool.
This commit is contained in:
Roman
2018-03-21 11:23:36 -07:00
committed by Carl Lerche
parent df9025594c
commit 494f0dc176
4 changed files with 265 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/runtime/builder.rs
+107
View File
@@ -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<Runtime> {
// 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,
}),
})
}
}
src/runtime.rs → src/runtime/mod.rs
+14 -165
View File
@@ -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<Inner>,
}
/// 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<Future<Item = (), Error = ()> + 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<Self> {
// 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<F>(&self, future: F)
where F: Future<Item = (), Error = ()> + Send + 'static,
{
self.inner.spawn(future).unwrap();
}
}
impl<T> future::Executor<T> for TaskExecutor
where T: Future<Item = (), Error = ()> + Send + 'static,
{
fn execute(&self, future: T) -> Result<(), future::ExecuteError<T>> {
self.inner.execute(future)
}
}
impl ::executor::Executor for TaskExecutor {
fn spawn(&mut self, future: Box<Future<Item = (), Error = ()> + Send>)
-> Result<(), ::executor::SpawnError>
{
self.inner.spawn(future)
}
#[cfg(feature = "unstable-futures")]
fn spawn2(&mut self, future: Box<futures2::Future<Item = (), Error = futures2::Never> + Send>)
-> Result<(), futures2::executor::SpawnError>
{
self.inner.spawn2(future)
}
}
#[cfg(feature = "unstable-futures")]
type Task2 = Box<futures2::Future<Item = (), Error = futures2::Never> + 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<Future<Item = (), Error = ()>>")
.finish()
}
}
src/runtime/shutdown.rs
+46
View File
@@ -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<Future<Item = (), Error = ()> + 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<Future<Item = (), Error = ()>>")
.finish()
}
}
src/runtime/task_executor.rs
+98
View File
@@ -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<F>(&self, future: F)
where F: Future<Item = (), Error = ()> + Send + 'static,
{
self.inner.spawn(future).unwrap();
}
}
impl<T> future::Executor<T> for TaskExecutor
where T: Future<Item = (), Error = ()> + Send + 'static,
{
fn execute(&self, future: T) -> Result<(), future::ExecuteError<T>> {
self.inner.execute(future)
}
}
impl ::executor::Executor for TaskExecutor {
fn spawn(&mut self, future: Box<Future<Item = (), Error = ()> + Send>)
-> Result<(), ::executor::SpawnError>
{
self.inner.spawn(future)
}
#[cfg(feature = "unstable-futures")]
fn spawn2(&mut self, future: Box<futures2::Future<Item = (), Error = futures2::Never> + Send>)
-> Result<(), futures2::executor::SpawnError>
{
self.inner.spawn2(future)
}
}
#[cfg(feature = "unstable-futures")]
type Task2 = Box<futures2::Future<Item = (), Error = futures2::Never> + 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)
}
}