diff --git a/Cargo.toml b/Cargo.toml index f3f490315..bea9bf457 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,6 +24,7 @@ keywords = ["io", "async", "non-blocking", "futures"] members = [ "./", "tokio-codec", + "tokio-current-thread", "tokio-executor", "tokio-fs", "tokio-io", @@ -40,6 +41,7 @@ travis-ci = { repository = "tokio-rs/tokio" } appveyor = { repository = "carllerche/tokio", id = "s83yxhy9qeb58va7" } [dependencies] +tokio-current-thread = { version = "0.1.0", path = "tokio-current-thread" } tokio-io = { version = "0.1.6", path = "tokio-io" } tokio-executor = { version = "0.1.2", path = "tokio-executor" } tokio-reactor = { version = "0.1.1", path = "tokio-reactor" } diff --git a/src/executor/current_thread/mod.rs b/src/executor/current_thread/mod.rs index 4b25d7eb3..6036aa997 100644 --- a/src/executor/current_thread/mod.rs +++ b/src/executor/current_thread/mod.rs @@ -1,3 +1,5 @@ +#![allow(deprecated)] + //! Execute many tasks concurrently on the current thread. //! //! [`CurrentThread`] is an executor that keeps tasks on the same thread that @@ -102,76 +104,24 @@ //! [`CurrentThread`]: struct.CurrentThread.html //! [`Future::poll`]: https://docs.rs/futures/0.1/futures/future/trait.Future.html#tymethod.poll -#![allow(deprecated)] +pub use tokio_current_thread::{ + BlockError, + CurrentThread, + Entered, + Handle, + RunError, + RunTimeoutError, + TaskExecutor, + Turn, + TurnError, + block_on_all, + spawn, +}; -mod scheduler; -use self::scheduler::Scheduler; - -use tokio_executor::{self, Enter, SpawnError}; -use tokio_executor::park::{Park, Unpark, ParkThread}; - -use futures::{executor, Async, Future}; -use futures::future::{self, Executor, ExecuteError, ExecuteErrorKind}; - -use std::fmt; use std::cell::Cell; use std::marker::PhantomData; -use std::rc::Rc; -use std::time::{Duration, Instant}; -use std::sync::mpsc; -#[cfg(feature = "unstable-futures")] -use futures2; - -/// Executes tasks on the current thread -pub struct CurrentThread { - /// Execute futures and receive unpark notifications. - scheduler: Scheduler, - - /// Current number of futures being executed - num_futures: usize, - - /// Thread park handle - park: P, - - /// Handle for spawning new futures from other threads - spawn_handle: Handle, - - /// Receiver for futures spawned from other threads - spawn_receiver: mpsc::Receiver + Send + 'static>>, -} - -/// Executes futures on the current thread. -/// -/// All futures executed using this executor will be executed on the current -/// thread. As such, `run` will wait for these futures to complete before -/// returning. -/// -/// For more details, see the [module level](index.html) documentation. -#[derive(Debug, Clone)] -pub struct TaskExecutor { - // Prevent the handle from moving across threads. - _p: ::std::marker::PhantomData>, -} - -/// Returned by the `turn` function. -#[derive(Debug)] -pub struct Turn { - polled: bool -} - -impl Turn { - /// `true` if any futures were polled at all and `false` otherwise. - pub fn has_polled(&self) -> bool { - self.polled - } -} - -/// A `CurrentThread` instance bound to a supplied execution context. -pub struct Entered<'a, P: Park + 'a> { - executor: &'a mut CurrentThread

, - enter: &'a mut Enter, -} +use futures::future::{self}; #[deprecated(since = "0.1.2", note = "use block_on_all instead")] #[doc(hidden)] @@ -181,54 +131,17 @@ pub struct Context<'a> { _p: PhantomData<&'a ()>, } -/// Error returned by the `run` function. -#[derive(Debug)] -pub struct RunError { - _p: (), +impl<'a> Context<'a> { + /// Cancels *all* executing futures. + pub fn cancel_all_spawned(&self) { + self.cancel.set(true); + } } -/// Error returned by the `run_timeout` function. -#[derive(Debug)] -pub struct RunTimeoutError { - timeout: bool, -} - -/// Error returned by the `turn` function. -#[derive(Debug)] -pub struct TurnError { - _p: (), -} - -/// Error returned by the `block_on` function. -#[derive(Debug)] -pub struct BlockError { - inner: Option, -} - -/// This is mostly split out to make the borrow checker happy. -struct Borrow<'a, U: 'a> { - scheduler: &'a mut Scheduler, - num_futures: &'a mut usize, -} - -trait SpawnLocal { - fn spawn_local(&mut self, future: Box>); -} - -struct CurrentRunner { - spawn: Cell>, -} - -/// Current thread's task runner. This is set in `TaskRunner::with` -thread_local!(static CURRENT: CurrentRunner = CurrentRunner { - spawn: Cell::new(None), -}); - #[deprecated(since = "0.1.2", note = "use block_on_all instead")] #[doc(hidden)] -#[allow(deprecated)] pub fn run(f: F) -> R -where F: FnOnce(&mut Context) -> R + where F: FnOnce(&mut Context) -> R { let mut context = Context { cancel: Cell::new(false), @@ -249,587 +162,9 @@ where F: FnOnce(&mut Context) -> R ret } -/// Run the executor bootstrapping the execution with the provided future. -/// -/// This creates a new [`CurrentThread`] executor, spawns the provided future, -/// and blocks the current thread until the provided future and **all** -/// subsequently spawned futures complete. In other words: -/// -/// * If the provided bootstrap future does **not** spawn any additional tasks, -/// `block_on_all` returns once `future` completes. -/// * If the provided bootstrap future **does** spawn additional tasks, then -/// `block_on_all` returns once **all** spawned futures complete. -/// -/// See [module level][mod] documentation for more details. -/// -/// [`CurrentThread`]: struct.CurrentThread.html -/// [mod]: index.html -pub fn block_on_all(future: F) -> Result -where F: Future, -{ - let mut current_thread = CurrentThread::new(); - - let ret = current_thread.block_on(future); - current_thread.run().unwrap(); - - ret.map_err(|e| e.into_inner().expect("unexpected execution error")) -} - -/// Executes a future on the current thread. -/// -/// The provided future must complete or be canceled before `run` will return. -/// -/// Unlike [`tokio::spawn`], this function will always spawn on a -/// `CurrentThread` executor and is able to spawn futures that are not `Send`. -/// -/// # Panics -/// -/// This function can only be invoked from the context of a `run` call; any -/// other use will result in a panic. -/// -/// [`tokio::spawn`]: ../fn.spawn.html -pub fn spawn(future: F) -where F: Future + 'static -{ - TaskExecutor::current() - .spawn_local(Box::new(future)) - .unwrap(); -} - -// ===== impl CurrentThread ===== - -impl CurrentThread { - /// Create a new instance of `CurrentThread`. - pub fn new() -> Self { - CurrentThread::new_with_park(ParkThread::new()) - } -} - -impl CurrentThread

{ - /// Create a new instance of `CurrentThread` backed by the given park - /// handle. - pub fn new_with_park(park: P) -> Self { - let unpark = park.unpark(); - - let (spawn_sender, spawn_receiver) = mpsc::channel(); - - let scheduler = Scheduler::new(unpark); - let notify = scheduler.notify(); - - CurrentThread { - scheduler: scheduler, - num_futures: 0, - park, - spawn_handle: Handle { sender: spawn_sender, notify: notify }, - spawn_receiver: spawn_receiver, - } - } - - /// Returns `true` if the executor is currently idle. - /// - /// An idle executor is defined by not currently having any spawned tasks. - pub fn is_idle(&self) -> bool { - self.num_futures == 0 - } - - /// Spawn the future on the executor. - /// - /// This internally queues the future to be executed once `run` is called. - pub fn spawn(&mut self, future: F) -> &mut Self - where F: Future + 'static, - { - self.borrow().spawn_local(Box::new(future)); - self - } - - /// Synchronously waits for the provided `future` to complete. - /// - /// 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. - /// - /// The caller is responsible for ensuring that other spawned futures - /// complete execution. - pub fn block_on(&mut self, future: F) - -> Result> - where F: Future - { - let mut enter = tokio_executor::enter().unwrap(); - self.enter(&mut enter).block_on(future) - } - - /// Run the executor to completion, blocking the thread until **all** - /// spawned futures have completed. - pub fn run(&mut self) -> Result<(), RunError> { - let mut enter = tokio_executor::enter().unwrap(); - self.enter(&mut enter).run() - } - - /// Run the executor to completion, blocking the thread until all - /// spawned futures have completed **or** `duration` time has elapsed. - pub fn run_timeout(&mut self, duration: Duration) - -> Result<(), RunTimeoutError> - { - let mut enter = tokio_executor::enter().unwrap(); - self.enter(&mut enter).run_timeout(duration) - } - - /// Perform a single iteration of the event loop. - /// - /// This function blocks the current thread even if the executor is idle. - pub fn turn(&mut self, duration: Option) - -> Result - { - let mut enter = tokio_executor::enter().unwrap(); - self.enter(&mut enter).turn(duration) - } - - /// Bind `CurrentThread` instance with an execution context. - pub fn enter<'a>(&'a mut self, enter: &'a mut Enter) -> Entered<'a, P> { - Entered { - executor: self, - enter, - } - } - - /// Returns a reference to the underlying `Park` instance. - pub fn get_park(&self) -> &P { - &self.park - } - - /// Returns a mutable reference to the underlying `Park` instance. - pub fn get_park_mut(&mut self) -> &mut P { - &mut self.park - } - - fn borrow(&mut self) -> Borrow { - Borrow { - scheduler: &mut self.scheduler, - num_futures: &mut self.num_futures, - } - } - - /// Get a new handle to spawn futures on the executor - /// - /// Different to the executor itself, the handle can be sent to different - /// threads and can be used to spawn futures on the executor. - pub fn handle(&self) -> Handle { - self.spawn_handle.clone() - } -} - -impl tokio_executor::Executor for CurrentThread { - fn spawn(&mut self, future: Box + Send>) - -> Result<(), SpawnError> - { - self.borrow().spawn_local(future); - Ok(()) - } - - #[cfg(feature = "unstable-futures")] - fn spawn2(&mut self, _future: Box + Send>) - -> Result<(), futures2::executor::SpawnError> - { - panic!("Futures 0.2 integration is not available for current_thread"); - } -} - -impl fmt::Debug for CurrentThread

{ - fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { - fmt.debug_struct("CurrentThread") - .field("scheduler", &self.scheduler) - .field("num_futures", &self.num_futures) - .finish() - } -} - -// ===== impl Entered ===== - -impl<'a, P: Park> Entered<'a, P> { - /// Spawn the future on the executor. - /// - /// This internally queues the future to be executed once `run` is called. - pub fn spawn(&mut self, future: F) -> &mut Self - where F: Future + 'static, - { - self.executor.borrow().spawn_local(Box::new(future)); - self - } - - /// Synchronously waits for the provided `future` to complete. - /// - /// 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. - /// - /// The caller is responsible for ensuring that other spawned futures - /// complete execution. - pub fn block_on(&mut self, future: F) - -> Result> - where F: Future - { - let mut future = executor::spawn(future); - let notify = self.executor.scheduler.notify(); - - loop { - let res = self.executor.borrow().enter(self.enter, || { - future.poll_future_notify(¬ify, 0) - }); - - match res { - Ok(Async::Ready(e)) => return Ok(e), - Err(e) => return Err(BlockError { inner: Some(e) }), - Ok(Async::NotReady) => {} - } - - self.tick(); - - if let Err(_) = self.executor.park.park() { - return Err(BlockError { inner: None }); - } - } - } - - /// Run the executor to completion, blocking the thread until **all** - /// spawned futures have completed. - pub fn run(&mut self) -> Result<(), RunError> { - self.run_timeout2(None) - .map_err(|_| RunError { _p: () }) - } - - /// Run the executor to completion, blocking the thread until all - /// spawned futures have completed **or** `duration` time has elapsed. - pub fn run_timeout(&mut self, duration: Duration) - -> Result<(), RunTimeoutError> - { - self.run_timeout2(Some(duration)) - } - - /// Perform a single iteration of the event loop. - /// - /// This function blocks the current thread even if the executor is idle. - pub fn turn(&mut self, duration: Option) - -> Result - { - let res = if self.executor.scheduler.has_pending_futures() { - self.executor.park.park_timeout(Duration::from_millis(0)) - } else { - match duration { - Some(duration) => self.executor.park.park_timeout(duration), - None => self.executor.park.park(), - } - }; - - if res.is_err() { - return Err(TurnError { _p: () }); - } - - let polled = self.tick(); - - Ok(Turn { polled }) - } - - /// Returns a reference to the underlying `Park` instance. - pub fn get_park(&self) -> &P { - &self.executor.park - } - - /// Returns a mutable reference to the underlying `Park` instance. - pub fn get_park_mut(&mut self) -> &mut P { - &mut self.executor.park - } - - fn run_timeout2(&mut self, dur: Option) - -> Result<(), RunTimeoutError> - { - if self.executor.is_idle() { - // Nothing to do - return Ok(()); - } - - let mut time = dur.map(|dur| (Instant::now() + dur, dur)); - - loop { - self.tick(); - - if self.executor.is_idle() { - return Ok(()); - } - - match time { - Some((until, rem)) => { - if let Err(_) = self.executor.park.park_timeout(rem) { - return Err(RunTimeoutError::new(false)); - } - - let now = Instant::now(); - - if now >= until { - return Err(RunTimeoutError::new(true)); - } - - time = Some((until, until - now)); - } - None => { - if let Err(_) = self.executor.park.park() { - return Err(RunTimeoutError::new(false)); - } - } - } - } - } - - /// Returns `true` if any futures were processed - fn tick(&mut self) -> bool { - // Spawn any futures that were spawned from other threads by manually - // looping over the receiver stream - - // FIXME: Slightly ugly but needed to make the borrow checker happy - let (mut borrow, spawn_receiver) = ( - Borrow { - scheduler: &mut self.executor.scheduler, - num_futures: &mut self.executor.num_futures, - }, - &mut self.executor.spawn_receiver, - ); - - while let Ok(future) = spawn_receiver.try_recv() { - borrow.spawn_local(future); - } - - // After any pending futures were scheduled, do the actual tick - borrow.scheduler.tick( - &mut *self.enter, - borrow.num_futures) - } -} - -impl<'a, P: Park> fmt::Debug for Entered<'a, P> { - fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { - fmt.debug_struct("Entered") - .field("executor", &self.executor) - .field("enter", &self.enter) - .finish() - } -} - -// ===== impl Handle ===== - -/// Handle to spawn a future on the corresponding `CurrentThread` instance -#[derive(Clone)] -pub struct Handle { - sender: mpsc::Sender + Send + 'static>>, - notify: executor::NotifyHandle, -} - -// Manual implementation because the Sender does not implement Debug -impl fmt::Debug for Handle { - fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { - fmt.debug_struct("Handle") - .finish() - } -} - -impl Handle { - /// Spawn a future onto the `CurrentThread` instance corresponding to this handle - /// - /// # Panics - /// - /// This function panics if the spawn fails. Failure occurs if the `CurrentThread` - /// instance of the `Handle` does not exist anymore. - pub fn spawn(&self, future: F) -> Result<(), SpawnError> - where F: Future + Send + 'static { - self.sender.send(Box::new(future)) - .expect("CurrentThread does not exist anymore"); - // use 0 for the id, CurrentThread does not make use of it - self.notify.notify(0); - - Ok(()) - } -} - -// ===== impl TaskExecutor ===== - #[deprecated(since = "0.1.2", note = "use TaskExecutor::current instead")] #[doc(hidden)] pub fn task_executor() -> TaskExecutor { - TaskExecutor { - _p: ::std::marker::PhantomData, - } + TaskExecutor::current() } -impl TaskExecutor { - /// Returns an executor that executes futures on the current thread. - /// - /// The user of `TaskExecutor` must ensure that when a future is submitted, - /// that it is done within the context of a call to `run`. - /// - /// For more details, see the [module level](index.html) documentation. - pub fn current() -> TaskExecutor { - TaskExecutor { - _p: ::std::marker::PhantomData, - } - } - - /// Spawn a future onto the current `CurrentThread` instance. - pub fn spawn_local(&mut self, future: Box>) - -> Result<(), SpawnError> - { - CURRENT.with(|current| { - match current.spawn.get() { - Some(spawn) => { - unsafe { (*spawn).spawn_local(future) }; - Ok(()) - } - None => { - Err(SpawnError::shutdown()) - } - } - }) - } -} - -impl tokio_executor::Executor for TaskExecutor { - fn spawn(&mut self, future: Box + Send>) - -> Result<(), SpawnError> - { - self.spawn_local(future) - } - - #[cfg(feature = "unstable-futures")] - fn spawn2(&mut self, _future: Box + Send>) - -> Result<(), futures2::executor::SpawnError> - { - panic!("Futures 0.2 integration is not available for current_thread"); - } - - fn status(&self) -> Result<(), SpawnError> { - CURRENT.with(|current| { - if current.spawn.get().is_some() { - Ok(()) - } else { - Err(SpawnError::shutdown()) - } - }) - } -} - -impl Executor for TaskExecutor -where F: Future + 'static -{ - fn execute(&self, future: F) -> Result<(), ExecuteError> { - CURRENT.with(|current| { - match current.spawn.get() { - Some(spawn) => { - unsafe { (*spawn).spawn_local(Box::new(future)) }; - Ok(()) - } - None => { - Err(ExecuteError::new(ExecuteErrorKind::Shutdown, future)) - } - } - }) - } -} - -// ===== impl Context ===== - -impl<'a> Context<'a> { - /// Cancels *all* executing futures. - pub fn cancel_all_spawned(&self) { - self.cancel.set(true); - } -} - -// ===== impl Borrow ===== - -impl<'a, U: Unpark> Borrow<'a, U> { - fn enter(&mut self, _: &mut Enter, f: F) -> R - where F: FnOnce() -> R, - { - CURRENT.with(|current| { - current.set_spawn(self, || { - f() - }) - }) - } -} - -impl<'a, U: Unpark> SpawnLocal for Borrow<'a, U> { - fn spawn_local(&mut self, future: Box>) { - *self.num_futures += 1; - self.scheduler.schedule(future); - } -} - -// ===== impl CurrentRunner ===== - -impl CurrentRunner { - fn set_spawn(&self, spawn: &mut SpawnLocal, f: F) -> R - where F: FnOnce() -> R - { - struct Reset<'a>(&'a CurrentRunner); - - impl<'a> Drop for Reset<'a> { - fn drop(&mut self) { - self.0.spawn.set(None); - } - } - - let _reset = Reset(self); - - let spawn = unsafe { hide_lt(spawn as *mut SpawnLocal) }; - self.spawn.set(Some(spawn)); - - f() - } -} - -unsafe fn hide_lt<'a>(p: *mut (SpawnLocal + 'a)) -> *mut (SpawnLocal + 'static) { - use std::mem; - mem::transmute(p) -} - -// ===== impl RunTimeoutError ===== - -impl RunTimeoutError { - fn new(timeout: bool) -> Self { - RunTimeoutError { timeout } - } - - /// Returns `true` if the error was caused by the operation timing out. - pub fn is_timeout(&self) -> bool { - self.timeout - } -} - -impl From for RunTimeoutError { - fn from(_: tokio_executor::EnterError) -> Self { - RunTimeoutError::new(false) - } -} - -// ===== impl BlockError ===== - -impl BlockError { - /// Returns the error yielded by the future being blocked on - pub fn into_inner(self) -> Option { - self.inner - } -} - -impl From for BlockError { - fn from(_: tokio_executor::EnterError) -> Self { - BlockError { inner: None } - } -} diff --git a/src/lib.rs b/src/lib.rs index 7ff0b2fd1..b0d0b80a6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -70,6 +70,7 @@ #[macro_use] extern crate futures; extern crate mio; +extern crate tokio_current_thread; extern crate tokio_io; extern crate tokio_executor; extern crate tokio_fs; diff --git a/tokio-current-thread/CHANGELOG.md b/tokio-current-thread/CHANGELOG.md new file mode 100644 index 000000000..066575d45 --- /dev/null +++ b/tokio-current-thread/CHANGELOG.md @@ -0,0 +1,3 @@ +# Unreleased + +* Extract `tokio::executor::current_thread` to a tokio-current-thread crate (#356) diff --git a/tokio-current-thread/Cargo.toml b/tokio-current-thread/Cargo.toml new file mode 100644 index 000000000..ce227d747 --- /dev/null +++ b/tokio-current-thread/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "tokio-current-thread" + +# When releasing to crates.io: +# - Update html_root_url. +# - Update CHANGELOG.md. +# - Create "v0.1.x" git tag. +version = "0.1.0" +documentation = "https://docs.rs/tokio-current-thread" +repository = "https://github.com/tokio-rs/tokio" +homepage = "https://github.com/tokio-rs/tokio" +license = "MIT" +authors = ["Carl Lerche "] +description = """ +Single threaded executor which manage many tasks concurrently on the current thread. +""" +keywords = ["futures", "tokio"] +categories = ["concurrency", "asynchronous"] + +[dependencies] +tokio-executor = { version = "0.1.2", path = "../tokio-executor" } +futures = "0.1.19" diff --git a/tokio-current-thread/LICENSE b/tokio-current-thread/LICENSE new file mode 100644 index 000000000..38c1e27b8 --- /dev/null +++ b/tokio-current-thread/LICENSE @@ -0,0 +1,25 @@ +Copyright (c) 2018 Tokio Contributors + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. diff --git a/tokio-current-thread/README.md b/tokio-current-thread/README.md new file mode 100644 index 000000000..eb60e7d82 --- /dev/null +++ b/tokio-current-thread/README.md @@ -0,0 +1,19 @@ +# tokio-current-thread + +Single threaded executor for Tokio. + +[Documentation](https://tokio-rs.github.io/tokio/tokio_current_thread/) + +## Overview + +This crate provides the single threaded executor which execute many tasks concurrently. + +## License + +This project is licensed under the [MIT license](LICENSE). + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, shall be licensed as MIT, without any additional +terms or conditions. diff --git a/tokio-current-thread/src/lib.rs b/tokio-current-thread/src/lib.rs new file mode 100644 index 000000000..81c742976 --- /dev/null +++ b/tokio-current-thread/src/lib.rs @@ -0,0 +1,709 @@ +//! A single-threaded executor which executes tasks on the same thread from which +//! they are spawned. +//! +//! +//! The crate provides: +//! +//! * [`CurrentThread`] is the main type of this crate. It executes tasks on the current thread. +//! The easiest way to start a new [`CurrentThread`] executor is to call +//! [`block_on_all`] with an initial task to seed the executor. +//! All tasks that are being managed by a [`CurrentThread`] executor are able to +//! spawn additional tasks by calling [`spawn`]. +//! +//! +//! Application authors will not use this crate directly. Instead, they will use the +//! `tokio` crate. Library authors should only depend on `tokio-current-thread` if they +//! are building a custom task executor. +//! +//! For more details, see [executor module] documentation in the Tokio crate. +//! +//! [`CurrentThread`]: struct.CurrentThread.html +//! [`spawn`]: fn.spawn.html +//! [`block_on_all`]: fn.block_on_all.html +//! [executor module]: https://docs.rs/tokio/0.1/tokio/executor/index.html + +#![doc(html_root_url = "https://docs.rs/tokio-current-thread/0.1.0")] +#![deny(warnings, missing_docs, missing_debug_implementations)] + +extern crate futures; +extern crate tokio_executor; + +mod scheduler; + +use self::scheduler::Scheduler; + +use tokio_executor::{Enter, SpawnError}; +use tokio_executor::park::{Park, Unpark, ParkThread}; + +use futures::{executor, Async, Future}; +use futures::future::{Executor, ExecuteError, ExecuteErrorKind}; + +use std::fmt; +use std::cell::Cell; +use std::rc::Rc; +use std::time::{Duration, Instant}; +use std::sync::mpsc; + +#[cfg(feature = "unstable-futures")] +use futures2; + +/// Executes tasks on the current thread +pub struct CurrentThread { + /// Execute futures and receive unpark notifications. + scheduler: Scheduler, + + /// Current number of futures being executed + num_futures: usize, + + /// Thread park handle + park: P, + + /// Handle for spawning new futures from other threads + spawn_handle: Handle, + + /// Receiver for futures spawned from other threads + spawn_receiver: mpsc::Receiver + Send + 'static>>, +} + +/// Executes futures on the current thread. +/// +/// All futures executed using this executor will be executed on the current +/// thread. As such, `run` will wait for these futures to complete before +/// returning. +/// +/// For more details, see the [module level](index.html) documentation. +#[derive(Debug, Clone)] +pub struct TaskExecutor { + // Prevent the handle from moving across threads. + _p: ::std::marker::PhantomData>, +} + +/// Returned by the `turn` function. +#[derive(Debug)] +pub struct Turn { + polled: bool +} + +impl Turn { + /// `true` if any futures were polled at all and `false` otherwise. + pub fn has_polled(&self) -> bool { + self.polled + } +} + +/// A `CurrentThread` instance bound to a supplied execution context. +pub struct Entered<'a, P: Park + 'a> { + executor: &'a mut CurrentThread

, + enter: &'a mut Enter, +} + +/// Error returned by the `run` function. +#[derive(Debug)] +pub struct RunError { + _p: (), +} + +/// Error returned by the `run_timeout` function. +#[derive(Debug)] +pub struct RunTimeoutError { + timeout: bool, +} + +/// Error returned by the `turn` function. +#[derive(Debug)] +pub struct TurnError { + _p: (), +} + +/// Error returned by the `block_on` function. +#[derive(Debug)] +pub struct BlockError { + inner: Option, +} + +/// This is mostly split out to make the borrow checker happy. +struct Borrow<'a, U: 'a> { + scheduler: &'a mut Scheduler, + num_futures: &'a mut usize, +} + +trait SpawnLocal { + fn spawn_local(&mut self, future: Box>); +} + +struct CurrentRunner { + spawn: Cell>, +} + +/// Current thread's task runner. This is set in `TaskRunner::with` +thread_local!(static CURRENT: CurrentRunner = CurrentRunner { + spawn: Cell::new(None), +}); + +/// Run the executor bootstrapping the execution with the provided future. +/// +/// This creates a new [`CurrentThread`] executor, spawns the provided future, +/// and blocks the current thread until the provided future and **all** +/// subsequently spawned futures complete. In other words: +/// +/// * If the provided bootstrap future does **not** spawn any additional tasks, +/// `block_on_all` returns once `future` completes. +/// * If the provided bootstrap future **does** spawn additional tasks, then +/// `block_on_all` returns once **all** spawned futures complete. +/// +/// See [module level][mod] documentation for more details. +/// +/// [`CurrentThread`]: struct.CurrentThread.html +/// [mod]: index.html +pub fn block_on_all(future: F) -> Result +where F: Future, +{ + let mut current_thread = CurrentThread::new(); + + let ret = current_thread.block_on(future); + current_thread.run().unwrap(); + + ret.map_err(|e| e.into_inner().expect("unexpected execution error")) +} + +/// Executes a future on the current thread. +/// +/// The provided future must complete or be canceled before `run` will return. +/// +/// Unlike [`tokio::spawn`], this function will always spawn on a +/// `CurrentThread` executor and is able to spawn futures that are not `Send`. +/// +/// # Panics +/// +/// This function can only be invoked from the context of a `run` call; any +/// other use will result in a panic. +/// +/// [`tokio::spawn`]: ../fn.spawn.html +pub fn spawn(future: F) +where F: Future + 'static +{ + TaskExecutor::current() + .spawn_local(Box::new(future)) + .unwrap(); +} + +// ===== impl CurrentThread ===== + +impl CurrentThread { + /// Create a new instance of `CurrentThread`. + pub fn new() -> Self { + CurrentThread::new_with_park(ParkThread::new()) + } +} + +impl CurrentThread

{ + /// Create a new instance of `CurrentThread` backed by the given park + /// handle. + pub fn new_with_park(park: P) -> Self { + let unpark = park.unpark(); + + let (spawn_sender, spawn_receiver) = mpsc::channel(); + + let scheduler = Scheduler::new(unpark); + let notify = scheduler.notify(); + + CurrentThread { + scheduler: scheduler, + num_futures: 0, + park, + spawn_handle: Handle { sender: spawn_sender, notify: notify }, + spawn_receiver: spawn_receiver, + } + } + + /// Returns `true` if the executor is currently idle. + /// + /// An idle executor is defined by not currently having any spawned tasks. + pub fn is_idle(&self) -> bool { + self.num_futures == 0 + } + + /// Spawn the future on the executor. + /// + /// This internally queues the future to be executed once `run` is called. + pub fn spawn(&mut self, future: F) -> &mut Self + where F: Future + 'static, + { + self.borrow().spawn_local(Box::new(future)); + self + } + + /// Synchronously waits for the provided `future` to complete. + /// + /// 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. + /// + /// The caller is responsible for ensuring that other spawned futures + /// complete execution. + pub fn block_on(&mut self, future: F) + -> Result> + where F: Future + { + let mut enter = tokio_executor::enter().unwrap(); + self.enter(&mut enter).block_on(future) + } + + /// Run the executor to completion, blocking the thread until **all** + /// spawned futures have completed. + pub fn run(&mut self) -> Result<(), RunError> { + let mut enter = tokio_executor::enter().unwrap(); + self.enter(&mut enter).run() + } + + /// Run the executor to completion, blocking the thread until all + /// spawned futures have completed **or** `duration` time has elapsed. + pub fn run_timeout(&mut self, duration: Duration) + -> Result<(), RunTimeoutError> + { + let mut enter = tokio_executor::enter().unwrap(); + self.enter(&mut enter).run_timeout(duration) + } + + /// Perform a single iteration of the event loop. + /// + /// This function blocks the current thread even if the executor is idle. + pub fn turn(&mut self, duration: Option) + -> Result + { + let mut enter = tokio_executor::enter().unwrap(); + self.enter(&mut enter).turn(duration) + } + + /// Bind `CurrentThread` instance with an execution context. + pub fn enter<'a>(&'a mut self, enter: &'a mut Enter) -> Entered<'a, P> { + Entered { + executor: self, + enter, + } + } + + /// Returns a reference to the underlying `Park` instance. + pub fn get_park(&self) -> &P { + &self.park + } + + /// Returns a mutable reference to the underlying `Park` instance. + pub fn get_park_mut(&mut self) -> &mut P { + &mut self.park + } + + fn borrow(&mut self) -> Borrow { + Borrow { + scheduler: &mut self.scheduler, + num_futures: &mut self.num_futures, + } + } + + /// Get a new handle to spawn futures on the executor + /// + /// Different to the executor itself, the handle can be sent to different + /// threads and can be used to spawn futures on the executor. + pub fn handle(&self) -> Handle { + self.spawn_handle.clone() + } +} + +impl tokio_executor::Executor for CurrentThread { + fn spawn(&mut self, future: Box + Send>) + -> Result<(), SpawnError> + { + self.borrow().spawn_local(future); + Ok(()) + } + + #[cfg(feature = "unstable-futures")] + fn spawn2(&mut self, _future: Box + Send>) + -> Result<(), futures2::executor::SpawnError> + { + panic!("Futures 0.2 integration is not available for current_thread"); + } +} + +impl fmt::Debug for CurrentThread

{ + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("CurrentThread") + .field("scheduler", &self.scheduler) + .field("num_futures", &self.num_futures) + .finish() + } +} + +// ===== impl Entered ===== + +impl<'a, P: Park> Entered<'a, P> { + /// Spawn the future on the executor. + /// + /// This internally queues the future to be executed once `run` is called. + pub fn spawn(&mut self, future: F) -> &mut Self + where F: Future + 'static, + { + self.executor.borrow().spawn_local(Box::new(future)); + self + } + + /// Synchronously waits for the provided `future` to complete. + /// + /// 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. + /// + /// The caller is responsible for ensuring that other spawned futures + /// complete execution. + pub fn block_on(&mut self, future: F) + -> Result> + where F: Future + { + let mut future = executor::spawn(future); + let notify = self.executor.scheduler.notify(); + + loop { + let res = self.executor.borrow().enter(self.enter, || { + future.poll_future_notify(¬ify, 0) + }); + + match res { + Ok(Async::Ready(e)) => return Ok(e), + Err(e) => return Err(BlockError { inner: Some(e) }), + Ok(Async::NotReady) => {} + } + + self.tick(); + + if let Err(_) = self.executor.park.park() { + return Err(BlockError { inner: None }); + } + } + } + + /// Run the executor to completion, blocking the thread until **all** + /// spawned futures have completed. + pub fn run(&mut self) -> Result<(), RunError> { + self.run_timeout2(None) + .map_err(|_| RunError { _p: () }) + } + + /// Run the executor to completion, blocking the thread until all + /// spawned futures have completed **or** `duration` time has elapsed. + pub fn run_timeout(&mut self, duration: Duration) + -> Result<(), RunTimeoutError> + { + self.run_timeout2(Some(duration)) + } + + /// Perform a single iteration of the event loop. + /// + /// This function blocks the current thread even if the executor is idle. + pub fn turn(&mut self, duration: Option) + -> Result + { + let res = if self.executor.scheduler.has_pending_futures() { + self.executor.park.park_timeout(Duration::from_millis(0)) + } else { + match duration { + Some(duration) => self.executor.park.park_timeout(duration), + None => self.executor.park.park(), + } + }; + + if res.is_err() { + return Err(TurnError { _p: () }); + } + + let polled = self.tick(); + + Ok(Turn { polled }) + } + + /// Returns a reference to the underlying `Park` instance. + pub fn get_park(&self) -> &P { + &self.executor.park + } + + /// Returns a mutable reference to the underlying `Park` instance. + pub fn get_park_mut(&mut self) -> &mut P { + &mut self.executor.park + } + + fn run_timeout2(&mut self, dur: Option) + -> Result<(), RunTimeoutError> + { + if self.executor.is_idle() { + // Nothing to do + return Ok(()); + } + + let mut time = dur.map(|dur| (Instant::now() + dur, dur)); + + loop { + self.tick(); + + if self.executor.is_idle() { + return Ok(()); + } + + match time { + Some((until, rem)) => { + if let Err(_) = self.executor.park.park_timeout(rem) { + return Err(RunTimeoutError::new(false)); + } + + let now = Instant::now(); + + if now >= until { + return Err(RunTimeoutError::new(true)); + } + + time = Some((until, until - now)); + } + None => { + if let Err(_) = self.executor.park.park() { + return Err(RunTimeoutError::new(false)); + } + } + } + } + } + + /// Returns `true` if any futures were processed + fn tick(&mut self) -> bool { + // Spawn any futures that were spawned from other threads by manually + // looping over the receiver stream + + // FIXME: Slightly ugly but needed to make the borrow checker happy + let (mut borrow, spawn_receiver) = ( + Borrow { + scheduler: &mut self.executor.scheduler, + num_futures: &mut self.executor.num_futures, + }, + &mut self.executor.spawn_receiver, + ); + + while let Ok(future) = spawn_receiver.try_recv() { + borrow.spawn_local(future); + } + + // After any pending futures were scheduled, do the actual tick + borrow.scheduler.tick( + &mut *self.enter, + borrow.num_futures) + } +} + +impl<'a, P: Park> fmt::Debug for Entered<'a, P> { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("Entered") + .field("executor", &self.executor) + .field("enter", &self.enter) + .finish() + } +} + +// ===== impl Handle ===== + +/// Handle to spawn a future on the corresponding `CurrentThread` instance +#[derive(Clone)] +pub struct Handle { + sender: mpsc::Sender + Send + 'static>>, + notify: executor::NotifyHandle, +} + +// Manual implementation because the Sender does not implement Debug +impl fmt::Debug for Handle { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("Handle") + .finish() + } +} + +impl Handle { + /// Spawn a future onto the `CurrentThread` instance corresponding to this handle + /// + /// # Panics + /// + /// This function panics if the spawn fails. Failure occurs if the `CurrentThread` + /// instance of the `Handle` does not exist anymore. + pub fn spawn(&self, future: F) -> Result<(), SpawnError> + where F: Future + Send + 'static { + self.sender.send(Box::new(future)) + .expect("CurrentThread does not exist anymore"); + // use 0 for the id, CurrentThread does not make use of it + self.notify.notify(0); + + Ok(()) + } +} + +// ===== impl TaskExecutor ===== + +impl TaskExecutor { + /// Returns an executor that executes futures on the current thread. + /// + /// The user of `TaskExecutor` must ensure that when a future is submitted, + /// that it is done within the context of a call to `run`. + /// + /// For more details, see the [module level](index.html) documentation. + pub fn current() -> TaskExecutor { + TaskExecutor { + _p: ::std::marker::PhantomData, + } + } + + /// Spawn a future onto the current `CurrentThread` instance. + pub fn spawn_local(&mut self, future: Box>) + -> Result<(), SpawnError> + { + CURRENT.with(|current| { + match current.spawn.get() { + Some(spawn) => { + unsafe { (*spawn).spawn_local(future) }; + Ok(()) + } + None => { + Err(SpawnError::shutdown()) + } + } + }) + } +} + +impl tokio_executor::Executor for TaskExecutor { + fn spawn(&mut self, future: Box + Send>) + -> Result<(), SpawnError> + { + self.spawn_local(future) + } + + #[cfg(feature = "unstable-futures")] + fn spawn2(&mut self, _future: Box + Send>) + -> Result<(), futures2::executor::SpawnError> + { + panic!("Futures 0.2 integration is not available for current_thread"); + } + + fn status(&self) -> Result<(), SpawnError> { + CURRENT.with(|current| { + if current.spawn.get().is_some() { + Ok(()) + } else { + Err(SpawnError::shutdown()) + } + }) + } +} + +impl Executor for TaskExecutor +where F: Future + 'static +{ + fn execute(&self, future: F) -> Result<(), ExecuteError> { + CURRENT.with(|current| { + match current.spawn.get() { + Some(spawn) => { + unsafe { (*spawn).spawn_local(Box::new(future)) }; + Ok(()) + } + None => { + Err(ExecuteError::new(ExecuteErrorKind::Shutdown, future)) + } + } + }) + } +} + +// ===== impl Borrow ===== + +impl<'a, U: Unpark> Borrow<'a, U> { + fn enter(&mut self, _: &mut Enter, f: F) -> R + where F: FnOnce() -> R, + { + CURRENT.with(|current| { + current.set_spawn(self, || { + f() + }) + }) + } +} + +impl<'a, U: Unpark> SpawnLocal for Borrow<'a, U> { + fn spawn_local(&mut self, future: Box>) { + *self.num_futures += 1; + self.scheduler.schedule(future); + } +} + +// ===== impl CurrentRunner ===== + +impl CurrentRunner { + fn set_spawn(&self, spawn: &mut SpawnLocal, f: F) -> R + where F: FnOnce() -> R + { + struct Reset<'a>(&'a CurrentRunner); + + impl<'a> Drop for Reset<'a> { + fn drop(&mut self) { + self.0.spawn.set(None); + } + } + + let _reset = Reset(self); + + let spawn = unsafe { hide_lt(spawn as *mut SpawnLocal) }; + self.spawn.set(Some(spawn)); + + f() + } +} + +unsafe fn hide_lt<'a>(p: *mut (SpawnLocal + 'a)) -> *mut (SpawnLocal + 'static) { + use std::mem; + mem::transmute(p) +} + +// ===== impl RunTimeoutError ===== + +impl RunTimeoutError { + fn new(timeout: bool) -> Self { + RunTimeoutError { timeout } + } + + /// Returns `true` if the error was caused by the operation timing out. + pub fn is_timeout(&self) -> bool { + self.timeout + } +} + +impl From for RunTimeoutError { + fn from(_: tokio_executor::EnterError) -> Self { + RunTimeoutError::new(false) + } +} + +// ===== impl BlockError ===== + +impl BlockError { + /// Returns the error yielded by the future being blocked on + pub fn into_inner(self) -> Option { + self.inner + } +} + +impl From for BlockError { + fn from(_: tokio_executor::EnterError) -> Self { + BlockError { inner: None } + } +} diff --git a/src/executor/current_thread/scheduler.rs b/tokio-current-thread/src/scheduler.rs similarity index 100% rename from src/executor/current_thread/scheduler.rs rename to tokio-current-thread/src/scheduler.rs diff --git a/tests/current_thread.rs b/tokio-current-thread/tests/current_thread.rs similarity index 81% rename from tests/current_thread.rs rename to tokio-current-thread/tests/current_thread.rs index 79e6785e9..5d4d8124e 100644 --- a/tests/current_thread.rs +++ b/tokio-current-thread/tests/current_thread.rs @@ -1,10 +1,10 @@ #![cfg(not(feature = "unstable-futures"))] -extern crate tokio; +extern crate tokio_current_thread; extern crate tokio_executor; extern crate futures; -use tokio::executor::current_thread::{self, block_on_all, CurrentThread}; +use tokio_current_thread::{block_on_all, CurrentThread}; use std::any::Any; use std::cell::{Cell, RefCell}; @@ -22,11 +22,11 @@ fn spawn_from_block_on_all() { let cnt = Rc::new(Cell::new(0)); let c = cnt.clone(); - let msg = current_thread::block_on_all(lazy(move || { + let msg = tokio_current_thread::block_on_all(lazy(move || { c.set(1 + c.get()); // Spawn! - current_thread::spawn(lazy(move || { + tokio_current_thread::spawn(lazy(move || { c.set(1 + c.get()); Ok::<(), ()>(()) })); @@ -63,17 +63,17 @@ fn spawn_many() { const ITER: usize = 200; let cnt = Rc::new(Cell::new(0)); - let mut current_thread = CurrentThread::new(); + let mut tokio_current_thread = CurrentThread::new(); for _ in 0..ITER { let cnt = cnt.clone(); - current_thread.spawn(lazy(move || { + tokio_current_thread.spawn(lazy(move || { cnt.set(1 + cnt.get()); Ok::<(), ()>(()) })); } - current_thread.run().unwrap(); + tokio_current_thread.run().unwrap(); assert_eq!(cnt.get(), ITER); } @@ -95,12 +95,12 @@ fn does_not_set_global_executor_by_default() { fn spawn_from_block_on_future() { let cnt = Rc::new(Cell::new(0)); - let mut current_thread = CurrentThread::new(); + let mut tokio_current_thread = CurrentThread::new(); - current_thread.block_on(lazy(|| { + tokio_current_thread.block_on(lazy(|| { let cnt = cnt.clone(); - current_thread::spawn(lazy(move || { + tokio_current_thread::spawn(lazy(move || { cnt.set(1 + cnt.get()); Ok(()) })); @@ -108,7 +108,7 @@ fn spawn_from_block_on_future() { Ok::<_, ()>(()) })).unwrap(); - current_thread.run().unwrap(); + tokio_current_thread.run().unwrap(); assert_eq!(1, cnt.get()); } @@ -128,10 +128,10 @@ impl Future for Never { fn outstanding_tasks_are_dropped_when_executor_is_dropped() { let mut rc = Rc::new(()); - let mut current_thread = CurrentThread::new(); - current_thread.spawn(Never(rc.clone())); + let mut tokio_current_thread = CurrentThread::new(); + tokio_current_thread.spawn(Never(rc.clone())); - drop(current_thread); + drop(tokio_current_thread); // Ensure the daemon is dropped assert!(Rc::get_mut(&mut rc).is_some()); @@ -140,14 +140,14 @@ fn outstanding_tasks_are_dropped_when_executor_is_dropped() { let mut rc = Rc::new(()); - let mut current_thread = CurrentThread::new(); + let mut tokio_current_thread = CurrentThread::new(); - current_thread.block_on(lazy(|| { - current_thread::spawn(Never(rc.clone())); + tokio_current_thread.block_on(lazy(|| { + tokio_current_thread::spawn(Never(rc.clone())); Ok::<_, ()>(()) })).unwrap(); - drop(current_thread); + drop(tokio_current_thread); // Ensure the daemon is dropped assert!(Rc::get_mut(&mut rc).is_some()); @@ -169,7 +169,7 @@ fn nesting_run() { #[should_panic] fn run_in_future() { block_on_all(lazy(|| { - current_thread::spawn(lazy(|| { + tokio_current_thread::spawn(lazy(|| { block_on_all(lazy(|| { ok() })).unwrap(); @@ -246,12 +246,12 @@ fn tasks_are_scheduled_fairly() { } block_on_all(lazy(|| { - current_thread::spawn(Spin { + tokio_current_thread::spawn(Spin { state: state.clone(), idx: 0, }); - current_thread::spawn(Spin { + tokio_current_thread::spawn(Spin { state: state, idx: 1, }); @@ -265,21 +265,21 @@ fn spawn_and_turn() { let cnt = Rc::new(Cell::new(0)); let c = cnt.clone(); - let mut current_thread = CurrentThread::new(); + let mut tokio_current_thread = CurrentThread::new(); // Spawn a basic task to get the executor to turn - current_thread.spawn(lazy(move || { + tokio_current_thread.spawn(lazy(move || { Ok(()) })); // Turn once... - current_thread.turn(None).unwrap(); + tokio_current_thread.turn(None).unwrap(); - current_thread.spawn(lazy(move || { + tokio_current_thread.spawn(lazy(move || { c.set(1 + c.get()); // Spawn! - current_thread::spawn(lazy(move || { + tokio_current_thread::spawn(lazy(move || { c.set(1 + c.get()); Ok::<(), ()>(()) })); @@ -288,21 +288,21 @@ fn spawn_and_turn() { })); // This does not run the newly spawned thread - current_thread.turn(None).unwrap(); + tokio_current_thread.turn(None).unwrap(); assert_eq!(1, cnt.get()); // This runs the newly spawned thread - current_thread.turn(None).unwrap(); + tokio_current_thread.turn(None).unwrap(); assert_eq!(2, cnt.get()); } #[test] fn spawn_in_drop() { - let mut current_thread = CurrentThread::new(); + let mut tokio_current_thread = CurrentThread::new(); let (tx, rx) = oneshot::channel(); - current_thread.spawn({ + tokio_current_thread.spawn({ struct OnDrop(Option); impl Drop for OnDrop { @@ -326,7 +326,7 @@ fn spawn_in_drop() { MyFuture { _data: Box::new(OnDrop(Some(move || { - current_thread::spawn(lazy(move || { + tokio_current_thread::spawn(lazy(move || { tx.send(()).unwrap(); Ok(()) })); @@ -334,8 +334,8 @@ fn spawn_in_drop() { } }); - current_thread.block_on(rx).unwrap(); - current_thread.run().unwrap(); + tokio_current_thread.block_on(rx).unwrap(); + tokio_current_thread.run().unwrap(); } #[test] @@ -352,11 +352,11 @@ fn hammer_turn() { // Add some jitter for _ in 0..THREADS { let th = thread::spawn(|| { - let mut current_thread = CurrentThread::new(); + let mut tokio_current_thread = CurrentThread::new(); let (tx, rx) = mpsc::unbounded(); - current_thread.spawn({ + tokio_current_thread.spawn({ let cnt = Rc::new(Cell::new(0)); let c = cnt.clone(); @@ -378,8 +378,8 @@ fn hammer_turn() { } }); - while !current_thread.is_idle() { - current_thread.turn(None).unwrap(); + while !tokio_current_thread.is_idle() { + tokio_current_thread.turn(None).unwrap(); } }); @@ -394,20 +394,20 @@ fn hammer_turn() { #[test] fn turn_has_polled() { - let mut current_thread = CurrentThread::new(); + let mut tokio_current_thread = CurrentThread::new(); // Spawn oneshot receiver let (sender, receiver) = oneshot::channel::<()>(); - current_thread.spawn(receiver.then(|_| Ok(()))); + tokio_current_thread.spawn(receiver.then(|_| Ok(()))); // Turn once... - let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + let res = tokio_current_thread.turn(Some(Duration::from_millis(0))).unwrap(); // Should've polled the receiver once, but considered it not ready assert!(res.has_polled()); // Turn another time - let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + let res = tokio_current_thread.turn(Some(Duration::from_millis(0))).unwrap(); // Should've polled nothing, the receiver is not ready yet assert!(!res.has_polled()); @@ -416,14 +416,14 @@ fn turn_has_polled() { sender.send(()).unwrap(); // Turn another time - let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + let res = tokio_current_thread.turn(Some(Duration::from_millis(0))).unwrap(); // Should've polled the receiver, it's ready now assert!(res.has_polled()); // Now the executor should be empty - assert!(current_thread.is_idle()); - let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + assert!(tokio_current_thread.is_idle()); + let res = tokio_current_thread.turn(Some(Duration::from_millis(0))).unwrap(); // So should've polled nothing assert!(!res.has_polled()); @@ -478,14 +478,14 @@ fn turn_fair() { send_now: send_now.clone(), }; - let mut current_thread = CurrentThread::new_with_park(my_park); + let mut tokio_current_thread = CurrentThread::new_with_park(my_park); let receiver_1_done = Rc::new(Cell::new(false)); let receiver_1_done_clone = receiver_1_done.clone(); // Once an item is received on the oneshot channel, it will immediately // immediately make the second oneshot channel ready - current_thread.spawn(receiver + tokio_current_thread.spawn(receiver .map_err(|_| unreachable!()) .and_then(move |_| { sender_2.send(()).unwrap(); @@ -498,7 +498,7 @@ fn turn_fair() { let receiver_2_done = Rc::new(Cell::new(false)); let receiver_2_done_clone = receiver_2_done.clone(); - current_thread.spawn(receiver_2 + tokio_current_thread.spawn(receiver_2 .map_err(|_| unreachable!()) .and_then(move |_| { receiver_2_done_clone.set(true); @@ -511,7 +511,7 @@ fn turn_fair() { let receiver_3_done = Rc::new(Cell::new(false)); let receiver_3_done_clone = receiver_3_done.clone(); - current_thread.spawn(receiver_3 + tokio_current_thread.spawn(receiver_3 .map_err(|_| unreachable!()) .and_then(move |_| { receiver_3_done_clone.set(true); @@ -520,11 +520,11 @@ fn turn_fair() { ); // First turn should've polled both and considered them not ready - let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + let res = tokio_current_thread.turn(Some(Duration::from_millis(0))).unwrap(); assert!(res.has_polled()); // Next turn should've polled nothing - let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + let res = tokio_current_thread.turn(Some(Duration::from_millis(0))).unwrap(); assert!(!res.has_polled()); assert!(!receiver_1_done.get()); @@ -537,7 +537,7 @@ fn turn_fair() { // Now the first receiver should be done, the second receiver should be ready // to be polled again and the socket not yet - let res = current_thread.turn(None).unwrap(); + let res = tokio_current_thread.turn(None).unwrap(); assert!(res.has_polled()); assert!(receiver_1_done.get()); @@ -551,7 +551,7 @@ fn turn_fair() { // and read the packet from it. If it didn't do both here, we would handle // futures that are woken up from the reactor and directly unfairly and would // favour the ones that are woken up directly. - let res = current_thread.turn(None).unwrap(); + let res = tokio_current_thread.turn(None).unwrap(); assert!(res.has_polled()); assert!(receiver_1_done.get()); @@ -562,8 +562,8 @@ fn turn_fair() { send_now.set(false); // Now we should be idle and turning should not poll anything - assert!(current_thread.is_idle()); - let res = current_thread.turn(None).unwrap(); + assert!(tokio_current_thread.is_idle()); + let res = tokio_current_thread.turn(None).unwrap(); assert!(!res.has_polled()); }