diff --git a/src/runtime/builder.rs b/src/runtime/builder.rs index e65fa7634..2c2bbf02f 100644 --- a/src/runtime/builder.rs +++ b/src/runtime/builder.rs @@ -78,7 +78,7 @@ impl Builder { /// # extern crate tokio; /// # use tokio::runtime::Builder; /// # pub fn main() { - /// let runtime = Builder::new().build(); + /// let runtime = Builder::new().build().unwrap(); /// // ... call runtime.run(...) /// # let _ = runtime; /// # } diff --git a/src/runtime/current_thread/mod.rs b/src/runtime/current_thread/mod.rs new file mode 100644 index 000000000..191533eee --- /dev/null +++ b/src/runtime/current_thread/mod.rs @@ -0,0 +1,72 @@ +//! A runtime implementation that runs everything on the current thread. +//! +//! [`current_thread::Runtime`][rt] is similar to the primary +//! [`Runtime`][concurrent-rt] except that it runs all components on the current +//! thread instead of using a thread pool. This means that it is able to spawn +//! futures that do not implement `Send`. +//! +//! Same as the default [`Runtime`][concurrent-rt], the +//! [`current_thread::Runtime`][rt] includes: +//! +//! * A [reactor] to drive I/O resources. +//! * An [executor] to execute tasks that use these I/O resources. +//! * A [timer] for scheduling work to run after a set period of time. +//! +//! Note that [`current_thread::Runtime`][rt] does not implement `Send` itself +//! and cannot be safely moved to other threads. +//! +//! # Spawning from other threads +//! +//! By default, [`current_thread::Runtime`][rt] does not provide a way to spawn +//! tasks from other threads. However, this can be accomplished by using a +//! [`mpsc::channel`][chan]. To do so, create a channel to send the task, then +//! spawn a task on [`current_thread::Runtime`][rt] that consumes the channel +//! messages and spawns new tasks for them. +//! +//! For example: +//! +//! ``` +//! # extern crate tokio; +//! # extern crate futures; +//! use tokio::runtime::current_thread::Runtime; +//! use tokio::prelude::*; +//! use futures::sync::mpsc; +//! +//! # fn main() { +//! let mut runtime = Runtime::new().unwrap(); +//! let (tx, rx) = mpsc::channel(128); +//! # tx.send(future::ok(())); +//! +//! runtime.spawn(rx.for_each(|task| { +//! tokio::spawn(task); +//! Ok(()) +//! }).map_err(|e| panic!("channel error"))); +//! +//! # /* +//! runtime.run().unwrap(); +//! # */ +//! # } +//! ``` +//! +//! # Examples +//! +//! Creating a new `Runtime` and running a future `f` until its completion and +//! returning its result. +//! +//! ``` +//! use tokio::runtime::current_thread::Runtime; +//! use tokio::prelude::*; +//! +//! let mut runtime = Runtime::new().unwrap(); +//! +//! // Use the runtime... +//! // runtime.block_on(f); // where f is a future +//! ``` +//! +//! [rt]: struct.Runtime.html +//! [concurrent-rt]: ../struct.Runtime.html +//! [chan]: https://docs.rs/futures/0.1/futures/sync/mpsc/fn.channel.html + +mod runtime; + +pub use self::runtime::Runtime; diff --git a/src/runtime/current_thread/runtime.rs b/src/runtime/current_thread/runtime.rs new file mode 100644 index 000000000..0d0ff2106 --- /dev/null +++ b/src/runtime/current_thread/runtime.rs @@ -0,0 +1,149 @@ +use executor::current_thread::{self, CurrentThread}; + +use tokio_reactor::{self, Reactor}; +use tokio_timer::timer::{self, Timer}; +use tokio_executor; + +use futures::Future; + +use std::io; + +/// Single-threaded runtime provides a way to start reactor +/// and executor on the current thread. +/// +/// See [module level][mod] documentation for more details. +/// +/// [mod]: index.html +#[derive(Debug)] +pub struct Runtime { + reactor_handle: tokio_reactor::Handle, + timer_handle: timer::Handle, + executor: CurrentThread>, +} + +/// Error returned by the `run` function. +#[derive(Debug)] +pub struct RunError { + inner: current_thread::RunError, +} + +impl Runtime { + /// Returns a new runtime initialized with default configuration values. + pub fn new() -> io::Result { + // We need a reactor to receive events about IO objects from kernel + let reactor = Reactor::new()?; + let reactor_handle = reactor.handle(); + + // Place a timer wheel on top of the reactor. If there are no timeouts to fire, it'll let the + // reactor pick up some new external events. + let timer = Timer::new(reactor); + let timer_handle = timer.handle(); + + // And now put a single-threaded executor on top of the timer. When there are no futures ready + // to do something, it'll let the timer or the reactor to generate some new stimuli for the + // futures to continue in their life. + let executor = CurrentThread::new_with_park(timer); + + let runtime = Runtime { reactor_handle, timer_handle, executor }; + Ok(runtime) + } + + /// Spawn a future onto the single-threaded Tokio runtime. + /// + /// 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::current_thread::Runtime; + /// + /// # fn dox() { + /// // Create the runtime + /// let mut rt = Runtime::new().unwrap(); + /// + /// // Spawn a future onto the runtime + /// rt.spawn(future::lazy(|| { + /// println!("running on the runtime"); + /// 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(&mut self, future: F) -> &mut Self + where F: Future + 'static, + { + self.executor.spawn(future); + self + } + + /// Runs the provided future, blocking the current thread until the future + /// completes. + /// + /// This function can be used to synchronously block the current thread + /// until the provided `future` has resolved either successfully or with an + /// error. The result of the future is then returned from this function + /// call. + /// + /// Note that this function will **also** execute any spawned futures on the + /// current thread, but will **not** block until these other spawned futures + /// have completed. Once the function returns, any uncompleted futures + /// remain pending in the `Runtime` instance. These futures will not run + /// until `block_on` or `run` is called again. + /// + /// The caller is responsible for ensuring that other spawned futures + /// complete execution by calling `block_on` or `run`. + pub fn block_on(&mut self, f: F) -> Result + where F: Future + { + self.enter(|executor| { + // Run the provided future + let ret = executor.block_on(f); + ret.map_err(|e| e.into_inner().expect("unexpected execution error")) + }) + } + + /// Run the executor to completion, blocking the thread until **all** + /// spawned futures have completed. + pub fn run(&mut self) -> Result<(), RunError> { + self.enter(|executor| executor.run()) + .map_err(|e| RunError { + inner: e, + }) + } + + fn enter(&mut self, f: F) -> R + where F: FnOnce(&mut current_thread::Entered>) -> R + { + let Runtime { ref reactor_handle, ref timer_handle, ref mut executor } = *self; + + // Binds an executor to this thread + let mut enter = tokio_executor::enter().expect("Multiple executors at once"); + + // This will set the default handle and timer to use inside the closure + // and run the future. + tokio_reactor::with_default(&reactor_handle, &mut enter, |enter| { + timer::with_default(&timer_handle, enter, |enter| { + // The TaskExecutor is a fake executor that looks into the + // current single-threaded executor when used. This is a trick, + // because we need two mutable references to the executor (one + // to run the provided future, another to install as the default + // one). We use the fake one here as the default one. + let mut default_executor = current_thread::TaskExecutor::current(); + tokio_executor::with_default(&mut default_executor, enter, |enter| { + let mut executor = executor.enter(enter); + f(&mut executor) + }) + }) + }) + } +} diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 27024a5a4..6584f9402 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -113,6 +113,7 @@ //! [`Timer`]: https://docs.rs/tokio-timer/0.2/tokio_timer/timer/struct.Timer.html mod builder; +pub mod current_thread; mod shutdown; mod task_executor; diff --git a/tests/runtime.rs b/tests/runtime.rs index 309ad0d4d..9c76b08e8 100644 --- a/tests/runtime.rs +++ b/tests/runtime.rs @@ -12,36 +12,60 @@ macro_rules! t { }) } +fn create_client_server_future() -> Box + Send> { + let server = t!(TcpListener::bind(&"127.0.0.1:0".parse().unwrap())); + let addr = t!(server.local_addr()); + let client = TcpStream::connect(&addr); + + let server = server.incoming().take(1) + .map_err(|e| panic!("accept err = {:?}", e)) + .for_each(|socket| { + tokio::spawn({ + io::write_all(socket, b"hello") + .map(|_| ()) + .map_err(|e| panic!("write err = {:?}", e)) + }) + }) + .map(|_| ()); + + let client = client + .map_err(|e| panic!("connect err = {:?}", e)) + .and_then(|client| { + // Read all + io::read_to_end(client, vec![]) + .map(|_| ()) + .map_err(|e| panic!("read err = {:?}", e)) + }); + + let future = server.join(client) + .map(|_| ()); + Box::new(future) +} + #[test] -fn basic_runtime_usage() { +fn runtime_tokio_run() { let _ = env_logger::init(); - tokio::run({ - let server = t!(TcpListener::bind(&"127.0.0.1:0".parse().unwrap())); - let addr = t!(server.local_addr()); - let client = TcpStream::connect(&addr); - - let server = server.incoming().take(1) - .map_err(|e| panic!("accept err = {:?}", e)) - .for_each(|socket| { - tokio::spawn({ - io::write_all(socket, b"hello") - .map(|_| ()) - .map_err(|e| panic!("write err = {:?}", e)) - }) - }) - .map(|_| ()); - - let client = client - .map_err(|e| panic!("connect err = {:?}", e)) - .and_then(|client| { - // Read all - io::read_to_end(client, vec![]) - .map(|_| ()) - .map_err(|e| panic!("read err = {:?}", e)) - }); - - server.join(client) - .map(|_| ()) - }); + tokio::run(create_client_server_future()); +} + +#[test] +fn runtime_single_threaded() { + let _ = env_logger::init(); + + let mut runtime = tokio::runtime::current_thread::Runtime::new() + .unwrap(); + runtime.block_on(create_client_server_future()).unwrap(); + runtime.run().unwrap(); +} + +#[test] +fn runtime_multi_threaded() { + let _ = env_logger::init(); + + let mut runtime = tokio::runtime::Builder::new() + .build() + .unwrap(); + runtime.spawn(create_client_server_future()); + runtime.shutdown_on_idle().wait().unwrap(); }