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

tokio-current-thread crate (#370)

Browse Source 
Extract `tokio::executor::current_thread` to a tokio-current-thread
crate. Deprecated fns stay in the old location. The new crate only
contains thee most recent API.
This commit is contained in:
jpbriquet
2018-06-12 10:26:03 -07:00
committed by Carl Lerche
parent ba05c39d65
commit 2cd854c2c7
10 changed files with 858 additions and 742 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+2
View File
@@ -24,6 +24,7 @@ keywords = ["io", "async", "non-blocking", "futures"]
members = [ members = [
"./", "./",
"tokio-codec", "tokio-codec",
"tokio-current-thread",
"tokio-executor", "tokio-executor",
"tokio-fs", "tokio-fs",
"tokio-io", "tokio-io",
@@ -40,6 +41,7 @@ travis-ci = { repository = "tokio-rs/tokio" }
appveyor = { repository = "carllerche/tokio", id = "s83yxhy9qeb58va7" } appveyor = { repository = "carllerche/tokio", id = "s83yxhy9qeb58va7" }
[dependencies] [dependencies]
tokio-current-thread = { version = "0.1.0", path = "tokio-current-thread" }
tokio-io = { version = "0.1.6", path = "tokio-io" } tokio-io = { version = "0.1.6", path = "tokio-io" }
tokio-executor = { version = "0.1.2", path = "tokio-executor" } tokio-executor = { version = "0.1.2", path = "tokio-executor" }
tokio-reactor = { version = "0.1.1", path = "tokio-reactor" } tokio-reactor = { version = "0.1.1", path = "tokio-reactor" }
src/executor/current_thread/mod.rs
+23 -688
View File
@@ -1,3 +1,5 @@
#![allow(deprecated)]
//! Execute many tasks concurrently on the current thread. //! Execute many tasks concurrently on the current thread.
//! //!
//! [`CurrentThread`] is an executor that keeps tasks on the same thread that //! [`CurrentThread`] is an executor that keeps tasks on the same thread that
@@ -102,76 +104,24 @@
//! [`CurrentThread`]: struct.CurrentThread.html //! [`CurrentThread`]: struct.CurrentThread.html
//! [`Future::poll`]: https://docs.rs/futures/0.1/futures/future/trait.Future.html#tymethod.poll //! [`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::cell::Cell;
use std::marker::PhantomData; use std::marker::PhantomData;
use std::rc::Rc;
use std::time::{Duration, Instant};
use std::sync::mpsc;
#[cfg(feature = "unstable-futures")] use futures::future::{self};
use futures2;
/// Executes tasks on the current thread
pub struct CurrentThread<P: Park = ParkThread> {
/// Execute futures and receive unpark notifications.
scheduler: Scheduler<P::Unpark>,
/// 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<Box<Future<Item = (), Error = ()> + 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<Rc<()>>,
}
/// 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<P>,
enter: &'a mut Enter,
}
#[deprecated(since = "0.1.2", note = "use block_on_all instead")] #[deprecated(since = "0.1.2", note = "use block_on_all instead")]
#[doc(hidden)] #[doc(hidden)]
@@ -181,54 +131,17 @@ pub struct Context<'a> {
_p: PhantomData<&'a ()>, _p: PhantomData<&'a ()>,
} }
/// Error returned by the `run` function. impl<'a> Context<'a> {
#[derive(Debug)] /// Cancels *all* executing futures.
pub struct RunError { pub fn cancel_all_spawned(&self) {
_p: (), 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<T> {
inner: Option<T>,
}
/// This is mostly split out to make the borrow checker happy.
struct Borrow<'a, U: 'a> {
scheduler: &'a mut Scheduler<U>,
num_futures: &'a mut usize,
}
trait SpawnLocal {
fn spawn_local(&mut self, future: Box<Future<Item = (), Error = ()>>);
}
struct CurrentRunner {
spawn: Cell<Option<*mut SpawnLocal>>,
}
/// 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")] #[deprecated(since = "0.1.2", note = "use block_on_all instead")]
#[doc(hidden)] #[doc(hidden)]
#[allow(deprecated)]
pub fn run<F, R>(f: F) -> R pub fn run<F, R>(f: F) -> R
where F: FnOnce(&mut Context) -> R where F: FnOnce(&mut Context) -> R
{ {
let mut context = Context { let mut context = Context {
cancel: Cell::new(false), cancel: Cell::new(false),
@@ -249,587 +162,9 @@ where F: FnOnce(&mut Context) -> R
ret 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<F>(future: F) -> Result<F::Item, F::Error>
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<F>(future: F)
where F: Future<Item = (), Error = ()> + 'static
{
TaskExecutor::current()
.spawn_local(Box::new(future))
.unwrap();
}
// ===== impl CurrentThread =====
impl CurrentThread<ParkThread> {
/// Create a new instance of `CurrentThread`.
pub fn new() -> Self {
CurrentThread::new_with_park(ParkThread::new())
}
}
impl<P: Park> CurrentThread<P> {
/// 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<F>(&mut self, future: F) -> &mut Self
where F: Future<Item = (), Error = ()> + '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<F>(&mut self, future: F)
-> Result<F::Item, BlockError<F::Error>>
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<Duration>)
-> Result<Turn, TurnError>
{
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<P::Unpark> {
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<Future<Item = (), Error = ()> + Send>)
-> Result<(), SpawnError>
{
self.borrow().spawn_local(future);
Ok(())
}
#[cfg(feature = "unstable-futures")]
fn spawn2(&mut self, _future: Box<futures2::Future<Item = (), Error = futures2::Never> + Send>)
-> Result<(), futures2::executor::SpawnError>
{
panic!("Futures 0.2 integration is not available for current_thread");
}
}
impl<P: Park> fmt::Debug for CurrentThread<P> {
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<F>(&mut self, future: F) -> &mut Self
where F: Future<Item = (), Error = ()> + '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<F>(&mut self, future: F)
-> Result<F::Item, BlockError<F::Error>>
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(&notify, 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<Duration>)
-> Result<Turn, TurnError>
{
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<Duration>)
-> 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<Box<Future<Item = (), Error = ()> + 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<F>(&self, future: F) -> Result<(), SpawnError>
where F: Future<Item = (), Error = ()> + 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")] #[deprecated(since = "0.1.2", note = "use TaskExecutor::current instead")]
#[doc(hidden)] #[doc(hidden)]
pub fn task_executor() -> TaskExecutor { pub fn task_executor() -> TaskExecutor {
TaskExecutor { TaskExecutor::current()
_p: ::std::marker::PhantomData,
}
} }
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<Future<Item = (), Error = ()>>)
-> 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<Future<Item = (), Error = ()> + Send>)
-> Result<(), SpawnError>
{
self.spawn_local(future)
}
#[cfg(feature = "unstable-futures")]
fn spawn2(&mut self, _future: Box<futures2::Future<Item = (), Error = futures2::Never> + 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<F> Executor<F> for TaskExecutor
where F: Future<Item = (), Error = ()> + 'static
{
fn execute(&self, future: F) -> Result<(), ExecuteError<F>> {
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<F, R>(&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<Future<Item = (), Error = ()>>) {
*self.num_futures += 1;
self.scheduler.schedule(future);
}
}
// ===== impl CurrentRunner =====
impl CurrentRunner {
fn set_spawn<F, R>(&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<tokio_executor::EnterError> for RunTimeoutError {
fn from(_: tokio_executor::EnterError) -> Self {
RunTimeoutError::new(false)
}
}
// ===== impl BlockError =====
impl<T> BlockError<T> {
/// Returns the error yielded by the future being blocked on
pub fn into_inner(self) -> Option<T> {
self.inner
}
}
impl<T> From<tokio_executor::EnterError> for BlockError<T> {
fn from(_: tokio_executor::EnterError) -> Self {
BlockError { inner: None }
}
}
src/lib.rs
+1
View File
@@ -70,6 +70,7 @@
#[macro_use] #[macro_use]
extern crate futures; extern crate futures;
extern crate mio; extern crate mio;
extern crate tokio_current_thread;
extern crate tokio_io; extern crate tokio_io;
extern crate tokio_executor; extern crate tokio_executor;
extern crate tokio_fs; extern crate tokio_fs;
tokio-current-thread/CHANGELOG.md
+3
View File
@@ -0,0 +1,3 @@
# Unreleased
* Extract `tokio::executor::current_thread` to a tokio-current-thread crate (#356)
tokio-current-thread/Cargo.toml
+22
View File
@@ -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 <[email protected]>"]
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"
tokio-current-thread/LICENSE
+25
View File
@@ -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.
tokio-current-thread/README.md
+19
View File
@@ -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.
tokio-current-thread/src/lib.rs
+709
View File
@@ -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<P: Park = ParkThread> {
/// Execute futures and receive unpark notifications.
scheduler: Scheduler<P::Unpark>,
/// 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<Box<Future<Item = (), Error = ()> + 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<Rc<()>>,
}
/// 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<P>,
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<T> {
inner: Option<T>,
}
/// This is mostly split out to make the borrow checker happy.
struct Borrow<'a, U: 'a> {
scheduler: &'a mut Scheduler<U>,
num_futures: &'a mut usize,
}
trait SpawnLocal {
fn spawn_local(&mut self, future: Box<Future<Item = (), Error = ()>>);
}
struct CurrentRunner {
spawn: Cell<Option<*mut SpawnLocal>>,
}
/// 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<F>(future: F) -> Result<F::Item, F::Error>
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<F>(future: F)
where F: Future<Item = (), Error = ()> + 'static
{
TaskExecutor::current()
.spawn_local(Box::new(future))
.unwrap();
}
// ===== impl CurrentThread =====
impl CurrentThread<ParkThread> {
/// Create a new instance of `CurrentThread`.
pub fn new() -> Self {
CurrentThread::new_with_park(ParkThread::new())
}
}
impl<P: Park> CurrentThread<P> {
/// 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<F>(&mut self, future: F) -> &mut Self
where F: Future<Item = (), Error = ()> + '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<F>(&mut self, future: F)
-> Result<F::Item, BlockError<F::Error>>
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<Duration>)
-> Result<Turn, TurnError>
{
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<P::Unpark> {
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<Future<Item = (), Error = ()> + Send>)
-> Result<(), SpawnError>
{
self.borrow().spawn_local(future);
Ok(())
}
#[cfg(feature = "unstable-futures")]
fn spawn2(&mut self, _future: Box<futures2::Future<Item = (), Error = futures2::Never> + Send>)
-> Result<(), futures2::executor::SpawnError>
{
panic!("Futures 0.2 integration is not available for current_thread");
}
}
impl<P: Park> fmt::Debug for CurrentThread<P> {
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<F>(&mut self, future: F) -> &mut Self
where F: Future<Item = (), Error = ()> + '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<F>(&mut self, future: F)
-> Result<F::Item, BlockError<F::Error>>
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(&notify, 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<Duration>)
-> Result<Turn, TurnError>
{
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<Duration>)
-> 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<Box<Future<Item = (), Error = ()> + 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<F>(&self, future: F) -> Result<(), SpawnError>
where F: Future<Item = (), Error = ()> + 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<Future<Item = (), Error = ()>>)
-> 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<Future<Item = (), Error = ()> + Send>)
-> Result<(), SpawnError>
{
self.spawn_local(future)
}
#[cfg(feature = "unstable-futures")]
fn spawn2(&mut self, _future: Box<futures2::Future<Item = (), Error = futures2::Never> + 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<F> Executor<F> for TaskExecutor
where F: Future<Item = (), Error = ()> + 'static
{
fn execute(&self, future: F) -> Result<(), ExecuteError<F>> {
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<F, R>(&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<Future<Item = (), Error = ()>>) {
*self.num_futures += 1;
self.scheduler.schedule(future);
}
}
// ===== impl CurrentRunner =====
impl CurrentRunner {
fn set_spawn<F, R>(&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<tokio_executor::EnterError> for RunTimeoutError {
fn from(_: tokio_executor::EnterError) -> Self {
RunTimeoutError::new(false)
}
}
// ===== impl BlockError =====
impl<T> BlockError<T> {
/// Returns the error yielded by the future being blocked on
pub fn into_inner(self) -> Option<T> {
self.inner
}
}
impl<T> From<tokio_executor::EnterError> for BlockError<T> {
fn from(_: tokio_executor::EnterError) -> Self {
BlockError { inner: None }
}
}
src/executor/current_thread/scheduler.rs → tokio-current-thread/src/scheduler.rs
View File
tests/current_thread.rs → tokio-current-thread/tests/current_thread.rs
+54 -54
View File
@@ -1,10 +1,10 @@
#![cfg(not(feature = "unstable-futures"))] #![cfg(not(feature = "unstable-futures"))]
extern crate tokio; extern crate tokio_current_thread;
extern crate tokio_executor; extern crate tokio_executor;
extern crate futures; 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::any::Any;
use std::cell::{Cell, RefCell}; use std::cell::{Cell, RefCell};
@@ -22,11 +22,11 @@ fn spawn_from_block_on_all() {
let cnt = Rc::new(Cell::new(0)); let cnt = Rc::new(Cell::new(0));
let c = cnt.clone(); 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()); c.set(1 + c.get());
// Spawn! // Spawn!
current_thread::spawn(lazy(move || { tokio_current_thread::spawn(lazy(move || {
c.set(1 + c.get()); c.set(1 + c.get());
Ok::<(), ()>(()) Ok::<(), ()>(())
})); }));
@@ -63,17 +63,17 @@ fn spawn_many() {
const ITER: usize = 200; const ITER: usize = 200;
let cnt = Rc::new(Cell::new(0)); let cnt = Rc::new(Cell::new(0));
let mut current_thread = CurrentThread::new(); let mut tokio_current_thread = CurrentThread::new();
for _ in 0..ITER { for _ in 0..ITER {
let cnt = cnt.clone(); let cnt = cnt.clone();
current_thread.spawn(lazy(move || { tokio_current_thread.spawn(lazy(move || {
cnt.set(1 + cnt.get()); cnt.set(1 + cnt.get());
Ok::<(), ()>(()) Ok::<(), ()>(())
})); }));
} }
current_thread.run().unwrap(); tokio_current_thread.run().unwrap();
assert_eq!(cnt.get(), ITER); assert_eq!(cnt.get(), ITER);
} }
@@ -95,12 +95,12 @@ fn does_not_set_global_executor_by_default() {
fn spawn_from_block_on_future() { fn spawn_from_block_on_future() {
let cnt = Rc::new(Cell::new(0)); 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(); let cnt = cnt.clone();
current_thread::spawn(lazy(move || { tokio_current_thread::spawn(lazy(move || {
cnt.set(1 + cnt.get()); cnt.set(1 + cnt.get());
Ok(()) Ok(())
})); }));
@@ -108,7 +108,7 @@ fn spawn_from_block_on_future() {
Ok::<_, ()>(()) Ok::<_, ()>(())
})).unwrap(); })).unwrap();
current_thread.run().unwrap(); tokio_current_thread.run().unwrap();
assert_eq!(1, cnt.get()); assert_eq!(1, cnt.get());
} }
@@ -128,10 +128,10 @@ impl Future for Never {
fn outstanding_tasks_are_dropped_when_executor_is_dropped() { fn outstanding_tasks_are_dropped_when_executor_is_dropped() {
let mut rc = Rc::new(()); let mut rc = Rc::new(());
let mut current_thread = CurrentThread::new(); let mut tokio_current_thread = CurrentThread::new();
current_thread.spawn(Never(rc.clone())); tokio_current_thread.spawn(Never(rc.clone()));
drop(current_thread); drop(tokio_current_thread);
// Ensure the daemon is dropped // Ensure the daemon is dropped
assert!(Rc::get_mut(&mut rc).is_some()); 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 rc = Rc::new(());
let mut current_thread = CurrentThread::new(); let mut tokio_current_thread = CurrentThread::new();
current_thread.block_on(lazy(|| { tokio_current_thread.block_on(lazy(|| {
current_thread::spawn(Never(rc.clone())); tokio_current_thread::spawn(Never(rc.clone()));
Ok::<_, ()>(()) Ok::<_, ()>(())
})).unwrap(); })).unwrap();
drop(current_thread); drop(tokio_current_thread);
// Ensure the daemon is dropped // Ensure the daemon is dropped
assert!(Rc::get_mut(&mut rc).is_some()); assert!(Rc::get_mut(&mut rc).is_some());
@@ -169,7 +169,7 @@ fn nesting_run() {
#[should_panic] #[should_panic]
fn run_in_future() { fn run_in_future() {
block_on_all(lazy(|| { block_on_all(lazy(|| {
current_thread::spawn(lazy(|| { tokio_current_thread::spawn(lazy(|| {
block_on_all(lazy(|| { block_on_all(lazy(|| {
ok() ok()
})).unwrap(); })).unwrap();
@@ -246,12 +246,12 @@ fn tasks_are_scheduled_fairly() {
} }
block_on_all(lazy(|| { block_on_all(lazy(|| {
current_thread::spawn(Spin { tokio_current_thread::spawn(Spin {
state: state.clone(), state: state.clone(),
idx: 0, idx: 0,
}); });
current_thread::spawn(Spin { tokio_current_thread::spawn(Spin {
state: state, state: state,
idx: 1, idx: 1,
}); });
@@ -265,21 +265,21 @@ fn spawn_and_turn() {
let cnt = Rc::new(Cell::new(0)); let cnt = Rc::new(Cell::new(0));
let c = cnt.clone(); 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 // Spawn a basic task to get the executor to turn
current_thread.spawn(lazy(move || { tokio_current_thread.spawn(lazy(move || {
Ok(()) Ok(())
})); }));
// Turn once... // 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()); c.set(1 + c.get());
// Spawn! // Spawn!
current_thread::spawn(lazy(move || { tokio_current_thread::spawn(lazy(move || {
c.set(1 + c.get()); c.set(1 + c.get());
Ok::<(), ()>(()) Ok::<(), ()>(())
})); }));
@@ -288,21 +288,21 @@ fn spawn_and_turn() {
})); }));
// This does not run the newly spawned thread // This does not run the newly spawned thread
current_thread.turn(None).unwrap(); tokio_current_thread.turn(None).unwrap();
assert_eq!(1, cnt.get()); assert_eq!(1, cnt.get());
// This runs the newly spawned thread // This runs the newly spawned thread
current_thread.turn(None).unwrap(); tokio_current_thread.turn(None).unwrap();
assert_eq!(2, cnt.get()); assert_eq!(2, cnt.get());
} }
#[test] #[test]
fn spawn_in_drop() { fn spawn_in_drop() {
let mut current_thread = CurrentThread::new(); let mut tokio_current_thread = CurrentThread::new();
let (tx, rx) = oneshot::channel(); let (tx, rx) = oneshot::channel();
current_thread.spawn({ tokio_current_thread.spawn({
struct OnDrop<F: FnOnce()>(Option<F>); struct OnDrop<F: FnOnce()>(Option<F>);
impl<F: FnOnce()> Drop for OnDrop<F> { impl<F: FnOnce()> Drop for OnDrop<F> {
@@ -326,7 +326,7 @@ fn spawn_in_drop() {
MyFuture { MyFuture {
_data: Box::new(OnDrop(Some(move || { _data: Box::new(OnDrop(Some(move || {
current_thread::spawn(lazy(move || { tokio_current_thread::spawn(lazy(move || {
tx.send(()).unwrap(); tx.send(()).unwrap();
Ok(()) Ok(())
})); }));
@@ -334,8 +334,8 @@ fn spawn_in_drop() {
} }
}); });
current_thread.block_on(rx).unwrap(); tokio_current_thread.block_on(rx).unwrap();
current_thread.run().unwrap(); tokio_current_thread.run().unwrap();
} }
#[test] #[test]
@@ -352,11 +352,11 @@ fn hammer_turn() {
// Add some jitter // Add some jitter
for _ in 0..THREADS { for _ in 0..THREADS {
let th = thread::spawn(|| { let th = thread::spawn(|| {
let mut current_thread = CurrentThread::new(); let mut tokio_current_thread = CurrentThread::new();
let (tx, rx) = mpsc::unbounded(); let (tx, rx) = mpsc::unbounded();
current_thread.spawn({ tokio_current_thread.spawn({
let cnt = Rc::new(Cell::new(0)); let cnt = Rc::new(Cell::new(0));
let c = cnt.clone(); let c = cnt.clone();
@@ -378,8 +378,8 @@ fn hammer_turn() {
} }
}); });
while !current_thread.is_idle() { while !tokio_current_thread.is_idle() {
current_thread.turn(None).unwrap(); tokio_current_thread.turn(None).unwrap();
} }
}); });
@@ -394,20 +394,20 @@ fn hammer_turn() {
#[test] #[test]
fn turn_has_polled() { fn turn_has_polled() {
let mut current_thread = CurrentThread::new(); let mut tokio_current_thread = CurrentThread::new();
// Spawn oneshot receiver // Spawn oneshot receiver
let (sender, receiver) = oneshot::channel::<()>(); let (sender, receiver) = oneshot::channel::<()>();
current_thread.spawn(receiver.then(|_| Ok(()))); tokio_current_thread.spawn(receiver.then(|_| Ok(())));
// Turn once... // 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 // Should've polled the receiver once, but considered it not ready
assert!(res.has_polled()); assert!(res.has_polled());
// Turn another time // 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 // Should've polled nothing, the receiver is not ready yet
assert!(!res.has_polled()); assert!(!res.has_polled());
@@ -416,14 +416,14 @@ fn turn_has_polled() {
sender.send(()).unwrap(); sender.send(()).unwrap();
// Turn another time // 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 // Should've polled the receiver, it's ready now
assert!(res.has_polled()); assert!(res.has_polled());
// Now the executor should be empty // Now the executor should be empty
assert!(current_thread.is_idle()); assert!(tokio_current_thread.is_idle());
let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); let res = tokio_current_thread.turn(Some(Duration::from_millis(0))).unwrap();
// So should've polled nothing // So should've polled nothing
assert!(!res.has_polled()); assert!(!res.has_polled());
@@ -478,14 +478,14 @@ fn turn_fair() {
send_now: send_now.clone(), 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 = Rc::new(Cell::new(false));
let receiver_1_done_clone = receiver_1_done.clone(); let receiver_1_done_clone = receiver_1_done.clone();
// Once an item is received on the oneshot channel, it will immediately // Once an item is received on the oneshot channel, it will immediately
// immediately make the second oneshot channel ready // immediately make the second oneshot channel ready
current_thread.spawn(receiver tokio_current_thread.spawn(receiver
.map_err(|_| unreachable!()) .map_err(|_| unreachable!())
.and_then(move |_| { .and_then(move |_| {
sender_2.send(()).unwrap(); sender_2.send(()).unwrap();
@@ -498,7 +498,7 @@ fn turn_fair() {
let receiver_2_done = Rc::new(Cell::new(false)); let receiver_2_done = Rc::new(Cell::new(false));
let receiver_2_done_clone = receiver_2_done.clone(); let receiver_2_done_clone = receiver_2_done.clone();
current_thread.spawn(receiver_2 tokio_current_thread.spawn(receiver_2
.map_err(|_| unreachable!()) .map_err(|_| unreachable!())
.and_then(move |_| { .and_then(move |_| {
receiver_2_done_clone.set(true); 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 = Rc::new(Cell::new(false));
let receiver_3_done_clone = receiver_3_done.clone(); let receiver_3_done_clone = receiver_3_done.clone();
current_thread.spawn(receiver_3 tokio_current_thread.spawn(receiver_3
.map_err(|_| unreachable!()) .map_err(|_| unreachable!())
.and_then(move |_| { .and_then(move |_| {
receiver_3_done_clone.set(true); receiver_3_done_clone.set(true);
@@ -520,11 +520,11 @@ fn turn_fair() {
); );
// First turn should've polled both and considered them not ready // 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()); assert!(res.has_polled());
// Next turn should've polled nothing // 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!(!res.has_polled());
assert!(!receiver_1_done.get()); 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 // Now the first receiver should be done, the second receiver should be ready
// to be polled again and the socket not yet // 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!(res.has_polled());
assert!(receiver_1_done.get()); 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 // 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 // futures that are woken up from the reactor and directly unfairly and would
// favour the ones that are woken up directly. // 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!(res.has_polled());
assert!(receiver_1_done.get()); assert!(receiver_1_done.get());
@@ -562,8 +562,8 @@ fn turn_fair() {
send_now.set(false); send_now.set(false);
// Now we should be idle and turning should not poll anything // Now we should be idle and turning should not poll anything
assert!(current_thread.is_idle()); assert!(tokio_current_thread.is_idle());
let res = current_thread.turn(None).unwrap(); let res = tokio_current_thread.turn(None).unwrap();
assert!(!res.has_polled()); assert!(!res.has_polled());
} }