mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-08-07 00:00:09 +02:00
compat: add a compat runtime (#1663)
## Motivation The `futures` crate's [`compat` module][futures-compat] provides interoperability between `futures` 0.1 and `std::future` _future types_ (e.g. implementing `std::future::Future` for a type that implements the `futures` 0.1 `Future` trait). However, this on its own is insufficient to run code written against `tokio` 0.1 on a `tokio` 0.2 runtime, if that code also relies on `tokio`'s runtime services. If legacy tasks are executed that rely on `tokio::timer`, perform IO using `tokio`'s reactor, or call `tokio::spawn`, those API calls will fail unless there is also a runtime compatibility layer. ## Solution As proposed in #1549, this branch introduces a new `tokio-compat` crate, with implementations of the thread pool and current-thread runtimes that are capable of running both tokio 0.1 and tokio 0.2 tasks. The compat runtime creates a background thread that runs a `tokio` 0.1 timer and reactor, and sets itself as the `tokio` 0.1 executor as well as the default 0.2 executor. This allows 0.1 futures that use 0.1 timer, reactor, and executor APIs may run alongside `std::future` tasks on the 0.2 runtime. ### Examples Spawning both `tokio` 0.1 and `tokio` 0.2 futures: ```rust use futures_01::future::lazy; tokio_compat::run(lazy(|| { // spawn a `futures` 0.1 future using the `spawn` function from the // `tokio` 0.1 crate: tokio_01::spawn(lazy(|| { println!("hello from tokio 0.1!"); Ok(()) })); // spawn an `async` block future on the same runtime using `tokio` // 0.2's `spawn`: tokio_02::spawn(async { println!("hello from tokio 0.2!"); }); Ok(()) })) ``` Futures on the compat runtime can use `timer` APIs from both 0.1 and 0.2 versions of `tokio`: ```rust use std::time::{Duration, Instant}; use futures_01::future::lazy; use tokio_compat::prelude::*; tokio_compat::run_03(async { // Wait for a `tokio` 0.1 `Delay`... let when = Instant::now() + Duration::from_millis(10); tokio_01::timer::Delay::new(when) // convert the delay future into a `std::future` that we can `await`. .compat() .await .expect("tokio 0.1 timer should work!"); println!("10 ms have elapsed"); // Wait for a `tokio` 0.2 `Delay`... let when = Instant::now() + Duration::from_millis(20); tokio_02::timer::delay(when).await; println!("20 ms have elapsed"); }); ``` ## Future Work This is just an initial implementation of a `tokio-compat` crate; there are more compatibility layers we'll want to provide before that crate is complete. For example, we should also provide compatibility between `tokio` 0.2's `AsyncRead` and `AsyncWrite` traits and the `futures` 0.1 and `futures` 0.3 versions of those traits. In #1549, @carllerche also suggests that the `compat` crate provide reimplementations of APIs that were removed from `tokio` 0.2 proper, such as the `tcp::Incoming` future. Additionally, there is likely extra work required to get the `tokio-threadpool` 0.1 `blocking` APIs to work on the compat runtime. This will be addressed in a follow-up PR. Fixes: #1605 Fixes: #1552 Refs: #1549 [futures-compat]: https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures/compat/index.html
This commit is contained in:
Eliza Weisman
@@ -2,6 +2,7 @@
|
||||
|
||||
members = [
|
||||
"tokio",
|
||||
"tokio-compat",
|
||||
"tokio-macros",
|
||||
"tokio-test",
|
||||
"tokio-tls",
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
[package]
|
||||
name = "tokio-compat"
|
||||
# When releasing to crates.io:
|
||||
# - Remove path dependencies
|
||||
# - Update html_root_url.
|
||||
# - Update doc url
|
||||
# - Cargo.toml
|
||||
# - README.md
|
||||
# - Update CHANGELOG.md.
|
||||
# - Create "v0.1.x" git tag.
|
||||
version = "0.1.0-alpha.1"
|
||||
edition = "2018"
|
||||
authors = ["Tokio Contributors <[email protected]>"]
|
||||
license = "MIT"
|
||||
readme = "README.md"
|
||||
documentation = "https://docs.rs/tokio-compat/0.1.0-alpha.1/tokio-compat/"
|
||||
repository = "https://github.com/tokio-rs/tokio"
|
||||
homepage = "https://tokio.rs"
|
||||
description = """
|
||||
Compatibility between `tokio` 0.2 and legacy versions.
|
||||
"""
|
||||
categories = ["asynchronous", "network-programming"]
|
||||
keywords = ["io", "async", "non-blocking", "futures"]
|
||||
|
||||
|
||||
[features]
|
||||
default = ["rt-full", "sink"]
|
||||
# enables the compat runtimes.
|
||||
rt-current-thread = [
|
||||
"tokio-timer-02",
|
||||
"tokio-reactor-01",
|
||||
"tokio-executor-01",
|
||||
"tokio-02/rt-current-thread",
|
||||
"tokio-02/timer",
|
||||
"tokio-02/sync",
|
||||
"tokio-02/net-driver",
|
||||
]
|
||||
rt-full = [
|
||||
"tracing-core",
|
||||
"num_cpus",
|
||||
"tokio-02/rt-full",
|
||||
"rt-current-thread",
|
||||
]
|
||||
sink = ["futures-util/sink"]
|
||||
|
||||
[dependencies]
|
||||
futures-01 = { package = "futures", version = "0.1" }
|
||||
futures-03-core = { package = "futures-core-preview", version = "0.3.0-alpha.19" }
|
||||
futures-util = { package = "futures-util-preview", version = "0.3.0-alpha.19", default-features = false, features = ["compat"] }
|
||||
tokio-02 = { package = "tokio", path = "../tokio", default_features = false }
|
||||
|
||||
# runtime-only
|
||||
tokio-timer-02 = { package = "tokio-timer", version = "0.2", optional = true}
|
||||
tokio-reactor-01 = { package = "tokio-reactor", version = "0.1", optional = true }
|
||||
tokio-executor-01 = { package = "tokio-executor", version = "0.1", optional = true }
|
||||
num_cpus = { version = "1", optional = true }
|
||||
tracing-core = { version = "0.1", optional = true }
|
||||
|
||||
[dev-dependencies]
|
||||
tokio-01 = { package = "tokio", version = "0.1" }
|
||||
@@ -0,0 +1,88 @@
|
||||
# Tokio Compat
|
||||
|
||||
Compatibility layers between `tokio` 0.2 and legacy versions.
|
||||
|
||||
[![Crates.io][crates-badge]][crates-url]
|
||||
[![MIT licensed][mit-badge]][mit-url]
|
||||
[![Build Status][azure-badge]][azure-url]
|
||||
[![Gitter chat][gitter-badge]][gitter-url]
|
||||
|
||||
[crates-badge]: https://img.shields.io/crates/v/tokio-compat.svg
|
||||
[crates-url]: https://crates.io/crates/tokio-compat
|
||||
[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
|
||||
[mit-url]: LICENSE
|
||||
[azure-badge]: https://dev.azure.com/tokio-rs/Tokio/_apis/build/status/tokio-rs.tokio?branchName=master
|
||||
[azure-url]: https://dev.azure.com/tokio-rs/Tokio/_build/latest?definitionId=1&branchName=master
|
||||
[gitter-badge]: https://img.shields.io/gitter/room/tokio-rs/tokio.svg
|
||||
[gitter-url]: https://gitter.im/tokio-rs/tokio
|
||||
|
||||
[Website](https://tokio.rs) |
|
||||
[Guides](https://tokio.rs/docs/) |
|
||||
[API Docs](https://docs.rs/tokio-compat/0.1.0-alpha.1/tokio-compat) |
|
||||
[Chat](https://gitter.im/tokio-rs/tokio)
|
||||
|
||||
## Overview
|
||||
|
||||
This crate provides compatibility runtimes that allow running both `futures` 0.1
|
||||
futures that use `tokio` 0.1 runtime services _and_ `std::future` futures that
|
||||
use `tokio` 0.2 runtime services.
|
||||
|
||||
### Examples
|
||||
|
||||
Spawning both `tokio` 0.1 and `tokio` 0.2 futures:
|
||||
|
||||
```rust
|
||||
use futures_01::future::lazy;
|
||||
|
||||
tokio_compat::run(lazy(|| {
|
||||
// spawn a `futures` 0.1 future using the `spawn` function from the
|
||||
// `tokio` 0.1 crate:
|
||||
tokio_01::spawn(lazy(|| {
|
||||
println!("hello from tokio 0.1!");
|
||||
Ok(())
|
||||
}));
|
||||
|
||||
// spawn an `async` block future on the same runtime using `tokio`
|
||||
// 0.2's `spawn`:
|
||||
tokio_02::spawn(async {
|
||||
println!("hello from tokio 0.2!");
|
||||
});
|
||||
|
||||
Ok(())
|
||||
}))
|
||||
```
|
||||
|
||||
Futures on the compat runtime can use `timer` APIs from both 0.1 and 0.2
|
||||
versions of `tokio`:
|
||||
|
||||
```rust
|
||||
use std::time::{Duration, Instant};
|
||||
use futures_01::future::lazy;
|
||||
use tokio_compat::prelude::*;
|
||||
|
||||
tokio_compat::run_std(async {
|
||||
// Wait for a `tokio` 0.1 `Delay`...
|
||||
let when = Instant::now() + Duration::from_millis(10);
|
||||
tokio_01::timer::Delay::new(when)
|
||||
// convert the delay future into a `std::future` that we can `await`.
|
||||
.compat()
|
||||
.await
|
||||
.expect("tokio 0.1 timer should work!");
|
||||
println!("10 ms have elapsed");
|
||||
|
||||
// Wait for a `tokio` 0.2 `Delay`...
|
||||
let when = Instant::now() + Duration::from_millis(20);
|
||||
tokio_02::timer::delay(when).await;
|
||||
println!("20 ms have elapsed");
|
||||
});
|
||||
```
|
||||
|
||||
## 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.
|
||||
@@ -0,0 +1,78 @@
|
||||
//! Compatibility between `tokio` 0.2 and legacy versions.
|
||||
//!
|
||||
//! ## Overview
|
||||
//!
|
||||
//! This crate provides compatibility runtimes that allow running both `futures` 0.1
|
||||
//! futures that use `tokio` 0.1 runtime services _and_ `std::future` futures that
|
||||
//! use `tokio` 0.2 runtime services.
|
||||
//!
|
||||
//! ### Examples
|
||||
//!
|
||||
//! Spawning both `tokio` 0.1 and `tokio` 0.2 futures:
|
||||
//!
|
||||
//! ```rust
|
||||
//! use futures_01::future::lazy;
|
||||
//!
|
||||
//! tokio_compat::run(lazy(|| {
|
||||
//! // spawn a `futures` 0.1 future using the `spawn` function from the
|
||||
//! // `tokio` 0.1 crate:
|
||||
//! tokio_01::spawn(lazy(|| {
|
||||
//! println!("hello from tokio 0.1!");
|
||||
//! Ok(())
|
||||
//! }));
|
||||
//!
|
||||
//! // spawn an `async` block future on the same runtime using `tokio`
|
||||
//! // 0.2's `spawn`:
|
||||
//! tokio_02::spawn(async {
|
||||
//! println!("hello from tokio 0.2!");
|
||||
//! });
|
||||
//!
|
||||
//! Ok(())
|
||||
//! }))
|
||||
//! ```
|
||||
//!
|
||||
//! Futures on the compat runtime can use `timer` APIs from both 0.1 and 0.2
|
||||
//! versions of `tokio`:
|
||||
//!
|
||||
//! ```rust
|
||||
//! use std::time::{Duration, Instant};
|
||||
//! use tokio_compat::prelude::*;
|
||||
//!
|
||||
//! tokio_compat::run_std(async {
|
||||
//! // Wait for a `tokio` 0.1 `Delay`...
|
||||
//! let when = Instant::now() + Duration::from_millis(10);
|
||||
//! tokio_01::timer::Delay::new(when)
|
||||
//! // convert the delay future into a `std::future` that we can `await`.
|
||||
//! .compat()
|
||||
//! .await
|
||||
//! .expect("tokio 0.1 timer should work!");
|
||||
//! println!("10 ms have elapsed");
|
||||
//!
|
||||
//! // Wait for a `tokio` 0.2 `Delay`...
|
||||
//! let when = Instant::now() + Duration::from_millis(20);
|
||||
//! tokio_02::timer::delay(when).await;
|
||||
//! println!("20 ms have elapsed");
|
||||
//! });
|
||||
//! ```
|
||||
//!
|
||||
//! ## Feature Flags
|
||||
//!
|
||||
//! - `rt-current-thread`: enables the `current_thread` compatibilty runtime
|
||||
//! - `rt-full`: enables the `current_thread` and threadpool compatibility
|
||||
//! runtimes (enabled by default)
|
||||
#![warn(
|
||||
missing_debug_implementations,
|
||||
missing_docs,
|
||||
rust_2018_idioms,
|
||||
unreachable_pub
|
||||
)]
|
||||
#![deny(intra_doc_link_resolution_failure)]
|
||||
#![doc(test(
|
||||
no_crate_inject,
|
||||
attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
|
||||
))]
|
||||
#[cfg(any(feature = "rt-current-thread", feature = "rt-full"))]
|
||||
pub mod runtime;
|
||||
#[cfg(feature = "rt-full")]
|
||||
pub use self::runtime::{run, run_std};
|
||||
pub mod prelude;
|
||||
@@ -0,0 +1,4 @@
|
||||
//! A prelude for `tokio` 0.1/0.2 compatibility.
|
||||
#[cfg(feature = "sink")]
|
||||
pub use futures_util::compat::Sink01CompatExt as _;
|
||||
pub use futures_util::compat::{Future01CompatExt as _, Stream01CompatExt as _};
|
||||
@@ -0,0 +1,113 @@
|
||||
use tokio_executor_01::{self as executor_01, park as park_01};
|
||||
use tokio_reactor_01 as reactor_01;
|
||||
use tokio_timer_02::{clock as clock_02, timer as timer_02};
|
||||
|
||||
use std::{
|
||||
io, thread,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
use tokio_02::executor::{current_thread::CurrentThread, park};
|
||||
use tokio_02::sync::oneshot;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(super) struct Background {
|
||||
reactor_handle: reactor_01::Handle,
|
||||
timer_handle: timer_02::Handle,
|
||||
shutdown_tx: Option<oneshot::Sender<()>>,
|
||||
thread: Option<thread::JoinHandle<()>>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(super) struct Now<N>(N);
|
||||
|
||||
#[derive(Debug)]
|
||||
struct CompatPark<P>(P);
|
||||
|
||||
impl Background {
|
||||
pub(super) fn spawn(clock: &tokio_02::timer::clock::Clock) -> io::Result<Self> {
|
||||
let clock = clock_02::Clock::new_with_now(Now(clock.clone()));
|
||||
|
||||
let reactor = reactor_01::Reactor::new()?;
|
||||
let reactor_handle = reactor.handle();
|
||||
|
||||
let timer = timer_02::Timer::new_with_now(reactor, clock);
|
||||
let timer_handle = timer.handle();
|
||||
|
||||
let (shutdown_tx, shutdown_rx) = oneshot::channel();
|
||||
let shutdown_tx = Some(shutdown_tx);
|
||||
|
||||
let thread = thread::spawn(move || {
|
||||
let mut rt = CurrentThread::new_with_park(CompatPark(timer));
|
||||
let _ = rt.block_on(shutdown_rx);
|
||||
});
|
||||
let thread = Some(thread);
|
||||
|
||||
Ok(Self {
|
||||
reactor_handle,
|
||||
timer_handle,
|
||||
thread,
|
||||
shutdown_tx,
|
||||
})
|
||||
}
|
||||
|
||||
pub(super) fn reactor(&self) -> &reactor_01::Handle {
|
||||
&self.reactor_handle
|
||||
}
|
||||
|
||||
pub(super) fn timer(&self) -> &timer_02::Handle {
|
||||
&self.timer_handle
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Background {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.shutdown_tx.take().unwrap().send(());
|
||||
let _ = self.thread.take().unwrap().join();
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn spawn_err(new: tokio_02::executor::SpawnError) -> executor_01::SpawnError {
|
||||
match new {
|
||||
_ if new.is_shutdown() => executor_01::SpawnError::shutdown(),
|
||||
_ if new.is_at_capacity() => executor_01::SpawnError::at_capacity(),
|
||||
e => unreachable!("weird spawn error {:?}", e),
|
||||
}
|
||||
}
|
||||
|
||||
impl<P> park::Park for CompatPark<P>
|
||||
where
|
||||
P: park_01::Park,
|
||||
{
|
||||
type Unpark = CompatPark<P::Unpark>;
|
||||
type Error = P::Error;
|
||||
|
||||
fn unpark(&self) -> Self::Unpark {
|
||||
CompatPark(self.0.unpark())
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn park(&mut self) -> Result<(), Self::Error> {
|
||||
self.0.park()
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> {
|
||||
self.0.park_timeout(duration)
|
||||
}
|
||||
}
|
||||
|
||||
impl<U> park::Unpark for CompatPark<U>
|
||||
where
|
||||
U: park_01::Unpark,
|
||||
{
|
||||
#[inline]
|
||||
fn unpark(&self) {
|
||||
self.0.unpark()
|
||||
}
|
||||
}
|
||||
|
||||
impl clock_02::Now for Now<tokio_02::timer::clock::Clock> {
|
||||
fn now(&self) -> Instant {
|
||||
self.0.now()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
use crate::runtime::current_thread::Runtime;
|
||||
|
||||
use tokio_02::executor::current_thread::CurrentThread;
|
||||
use tokio_02::net::driver::Reactor;
|
||||
use tokio_02::timer::clock::Clock;
|
||||
use tokio_02::timer::timer::Timer;
|
||||
|
||||
use std::io;
|
||||
|
||||
/// Builds a single-threaded compatibility runtime with custom configuration
|
||||
/// values.
|
||||
///
|
||||
/// Methods can be chained in order to set the configuration values. The
|
||||
/// Runtime is constructed by calling [`build`].
|
||||
///
|
||||
/// New instances of `Builder` are obtained via [`Builder::new`].
|
||||
///
|
||||
/// See function level documentation for details on the various configuration
|
||||
/// settings.
|
||||
///
|
||||
/// [`build`]: #method.build
|
||||
/// [`Builder::new`]: #method.new
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::current_thread::Builder;
|
||||
/// use tokio_02::timer::clock::Clock;
|
||||
///
|
||||
/// # pub fn main() {
|
||||
/// // build Runtime
|
||||
/// let runtime = Builder::new()
|
||||
/// .clock(Clock::new())
|
||||
/// .build();
|
||||
/// // ... call runtime.run(...)
|
||||
/// # let _ = runtime;
|
||||
/// # }
|
||||
/// ```
|
||||
#[derive(Debug)]
|
||||
pub struct Builder {
|
||||
/// The clock to use
|
||||
clock: Clock,
|
||||
}
|
||||
|
||||
impl Builder {
|
||||
/// Returns a new compatibility runtime builder initialized with default
|
||||
/// configuration values.
|
||||
///
|
||||
/// Configuration methods can be chained on the return value.
|
||||
pub fn new() -> Builder {
|
||||
Builder {
|
||||
clock: Clock::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Set the `Clock` instance that will be used by the runtime.
|
||||
pub fn clock(&mut self, clock: Clock) -> &mut Self {
|
||||
self.clock = clock;
|
||||
self
|
||||
}
|
||||
|
||||
/// Create the configured `Runtime`.
|
||||
pub fn build(&mut self) -> io::Result<Runtime> {
|
||||
// We need a reactor to receive events about IO objects from kernel
|
||||
let reactor = Reactor::new()?;
|
||||
let reactor_handle = reactor.handle();
|
||||
|
||||
// Place a timer wheel on top of the reactor. If there are no timeouts to fire, it'll let the
|
||||
// reactor pick up some new external events.
|
||||
let timer = Timer::new_with_clock(reactor, self.clock.clone());
|
||||
let timer_handle = timer.handle();
|
||||
|
||||
// And now put a single-threaded executor on top of the timer. When there are no futures ready
|
||||
// to do something, it'll let the timer or the reactor to generate some new stimuli for the
|
||||
// futures to continue in their life.
|
||||
let executor = CurrentThread::new_with_park(timer);
|
||||
|
||||
Runtime::new2(reactor_handle, timer_handle, self.clock.clone(), executor)
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for Builder {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,145 @@
|
||||
//! A compatibility implementation that runs everything on the current thread.
|
||||
//!
|
||||
//! [`current_thread::Runtime`][rt] is similar to the primary
|
||||
//! [compatibility `Runtime`][concurrent-rt] except that it runs all components
|
||||
//! on the current thread instead of using a thread pool. This means that it is
|
||||
//! able to spawn futures that do not implement `Send`.
|
||||
//!
|
||||
//! Same as the default [`current_thread::Runtime`][default-rt] in the main
|
||||
//! `tokio` crate, the [`tokio_compat::current_thread::Runtime`][rt] includes:
|
||||
//!
|
||||
//! * A [reactor] to drive I/O resources.
|
||||
//! * An [executor] to execute tasks that use these I/O resources.
|
||||
//! * A [timer] for scheduling work to run after a set period of time.
|
||||
//!
|
||||
//! Unlike the default `current_thread::Runtime`, however, the `tokio_compat`
|
||||
//! version must spawn an additional background thread to run the `tokio` 0.1
|
||||
//! [`Reactor`][reactor-01] and [`Timer`][timer-01]. This is necessary to
|
||||
//! support legacy tasks, as the main thread is already running a `tokio` 0.2
|
||||
//! `Reactor` and `Timer`.
|
||||
//!
|
||||
//! Note that [`current_thread::Runtime`][rt] does not implement `Send` itself
|
||||
//! and cannot be safely moved to other threads
|
||||
//!
|
||||
//! # Spawning from other threads
|
||||
//!
|
||||
//! While [`current_thread::Runtime`][rt] does not implement `Send` and cannot
|
||||
//! safely be moved to other threads, it provides a `Handle` that can be sent
|
||||
//! to other threads and allows to spawn new tasks from there.
|
||||
//!
|
||||
//! For example:
|
||||
//!
|
||||
//! ```
|
||||
//! use tokio_compat::runtime::current_thread::Runtime;
|
||||
//! use std::thread;
|
||||
//!
|
||||
//! let runtime = Runtime::new().unwrap();
|
||||
//! let handle = runtime.handle();
|
||||
//!
|
||||
//! thread::spawn(move || {
|
||||
//! // Spawn a `futures` 0.1 task on the other thread's runtime.
|
||||
//! let _ = handle.spawn(futures_01::future::lazy(|| {
|
||||
//! println!("hello from futures 0.1!");
|
||||
//! Ok(())
|
||||
//! }));
|
||||
//!
|
||||
//! // Spawn a `std::future` task on the other thread's runtime.
|
||||
//! let _ = handle.spawn_std(async {
|
||||
//! println!("hello from std::future!");
|
||||
//! });
|
||||
//! }).join().unwrap();
|
||||
//! ```
|
||||
//!
|
||||
//! # Examples
|
||||
//!
|
||||
//! Creating a new `Runtime` and running a future `f` until its completion and
|
||||
//! returning its result.
|
||||
//!
|
||||
//! ```
|
||||
//! use tokio_compat::runtime::current_thread::Runtime;
|
||||
//!
|
||||
//! let runtime = Runtime::new().unwrap();
|
||||
//!
|
||||
//! // Use the runtime...
|
||||
//! // runtime.block_on(f); // where f is a future
|
||||
//! ```
|
||||
//!
|
||||
//! [rt]: struct.Runtime.html
|
||||
//! [concurrent-rt]: ../struct.Runtime.html
|
||||
//! [default-rt]:
|
||||
//! https://docs.rs/tokio/0.2.0-alpha.6/tokio/runtime/current_thread/struct.Runtime.html
|
||||
//! [chan]: https://docs.rs/futures/0.1/futures/sync/mpsc/fn.channel.html
|
||||
//! [reactor]: ../../reactor/struct.Reactor.html
|
||||
//! [executor]: https://tokio.rs/docs/internals/runtime-model/#executors
|
||||
//! [timer]: ../../timer/index.html
|
||||
//! [timer-01]: https://docs.rs/tokio/0.1.22/tokio/timer/index.html
|
||||
//! [reactor-01]: https://docs.rs/tokio/0.1.22/tokio/reactor/struct.Reactor.html
|
||||
use super::compat;
|
||||
|
||||
mod builder;
|
||||
mod runtime;
|
||||
mod task_executor;
|
||||
|
||||
pub use self::builder::Builder;
|
||||
pub use self::runtime::{Handle, RunError, Runtime};
|
||||
pub use self::task_executor::TaskExecutor;
|
||||
|
||||
use futures_01::future::Future as Future01;
|
||||
use futures_util::{compat::Future01CompatExt, FutureExt};
|
||||
use std::future::Future;
|
||||
|
||||
/// Run the provided `futures` 0.1 future to completion using a runtime running on the current thread.
|
||||
///
|
||||
/// This first creates a new [`Runtime`], and calls [`Runtime::block_on`] with the provided future,
|
||||
/// which blocks the current thread until the provided future completes. It then calls
|
||||
/// [`Runtime::run`] to wait for any other spawned futures to resolve.
|
||||
pub fn block_on_all<F>(future: F) -> Result<F::Item, F::Error>
|
||||
where
|
||||
F: Future01,
|
||||
{
|
||||
block_on_all_std(future.compat())
|
||||
}
|
||||
|
||||
/// Run the provided `std::future` future to completion using a runtime running on the current thread.
|
||||
///
|
||||
/// This first creates a new [`Runtime`], and calls [`Runtime::block_on`] with the provided future,
|
||||
/// which blocks the current thread until the provided future completes. It then calls
|
||||
/// [`Runtime::run`] to wait for any other spawned futures to resolve.
|
||||
pub fn block_on_all_std<F>(future: F) -> F::Output
|
||||
where
|
||||
F: Future,
|
||||
{
|
||||
let mut r = Runtime::new().expect("failed to start runtime on current thread");
|
||||
let v = r.block_on_std(future);
|
||||
r.run().expect("failed to resolve remaining futures");
|
||||
v
|
||||
}
|
||||
|
||||
/// Start a current-thread runtime using the supplied `futures` 0.1 future to bootstrap execution.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if called from the context of an executor.
|
||||
pub fn run<F>(future: F)
|
||||
where
|
||||
F: Future01<Item = (), Error = ()> + 'static,
|
||||
{
|
||||
run_std(future.compat().map(|_| ()))
|
||||
}
|
||||
|
||||
/// Start a current-thread runtime using the supplied `std::future` ture to bootstrap execution.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if called from the context of an executor.
|
||||
pub fn run_std<F>(future: F)
|
||||
where
|
||||
F: Future<Output = ()> + 'static,
|
||||
{
|
||||
let mut r = Runtime::new().expect("failed to start runtime on current thread");
|
||||
r.spawn_std(future);
|
||||
r.run().expect("failed to resolve remaining futures");
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
@@ -0,0 +1,341 @@
|
||||
use super::{compat, Builder};
|
||||
|
||||
use tokio_02::executor::current_thread::Handle as ExecutorHandle;
|
||||
use tokio_02::executor::current_thread::{self, CurrentThread};
|
||||
use tokio_02::net::driver::{self, Reactor};
|
||||
use tokio_02::timer::clock::{self, Clock};
|
||||
use tokio_02::timer::timer::{self, Timer};
|
||||
use tokio_executor_01 as executor_01;
|
||||
use tokio_reactor_01 as reactor_01;
|
||||
use tokio_timer_02 as timer_02;
|
||||
|
||||
use futures_01::future::Future as Future01;
|
||||
use futures_util::{compat::Future01CompatExt, future::FutureExt};
|
||||
use std::error::Error;
|
||||
use std::fmt;
|
||||
use std::future::Future;
|
||||
use std::io;
|
||||
|
||||
/// Single-threaded runtime provides a way to start reactor
|
||||
/// and executor on the current thread.
|
||||
///
|
||||
/// See [module level][mod] documentation for more details.
|
||||
///
|
||||
/// [mod]: index.html
|
||||
#[derive(Debug)]
|
||||
pub struct Runtime {
|
||||
reactor_handle: driver::Handle,
|
||||
timer_handle: timer::Handle,
|
||||
clock: Clock,
|
||||
executor: CurrentThread<Parker>,
|
||||
|
||||
/// Compatibility background thread.
|
||||
///
|
||||
/// This maintains a `tokio` 0.1 timer and reactor to support running
|
||||
/// futures that use older tokio APIs.
|
||||
compat: compat::Background,
|
||||
}
|
||||
|
||||
pub(super) type Parker = Timer<Reactor>;
|
||||
|
||||
/// Handle to spawn a future on the corresponding `CurrentThread` runtime instance
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Handle(ExecutorHandle);
|
||||
|
||||
impl Handle {
|
||||
/// Spawn a `futures` 0.1 future onto the `CurrentThread` runtime 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<(), executor_01::SpawnError>
|
||||
where
|
||||
F: Future01<Item = (), Error = ()> + Send + 'static,
|
||||
{
|
||||
self.0
|
||||
.spawn(future.compat().map(|_| ()))
|
||||
.map_err(compat::spawn_err)
|
||||
}
|
||||
|
||||
/// Spawn a `std::future` future onto the `CurrentThread` runtime 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_std<F>(&self, future: F) -> Result<(), tokio_02::executor::SpawnError>
|
||||
where
|
||||
F: Future<Output = ()> + Send + 'static,
|
||||
{
|
||||
self.0.spawn(future)
|
||||
}
|
||||
/// Provides a best effort **hint** to whether or not `spawn` will succeed.
|
||||
///
|
||||
/// This function may return both false positives **and** false negatives.
|
||||
/// If `status` returns `Ok`, then a call to `spawn` will *probably*
|
||||
/// succeed, but may fail. If `status` returns `Err`, a call to `spawn` will
|
||||
/// *probably* fail, but may succeed.
|
||||
///
|
||||
/// This allows a caller to avoid creating the task if the call to `spawn`
|
||||
/// has a high likelihood of failing.
|
||||
pub fn status(&self) -> Result<(), executor_01::SpawnError> {
|
||||
self.0.status().map_err(compat::spawn_err)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> tokio_02::executor::TypedExecutor<T> for Handle
|
||||
where
|
||||
T: Future<Output = ()> + Send + 'static,
|
||||
{
|
||||
fn spawn(&mut self, future: T) -> Result<(), tokio_02::executor::SpawnError> {
|
||||
Handle::spawn_std(self, future)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> executor_01::TypedExecutor<T> for Handle
|
||||
where
|
||||
T: Future01<Item = (), Error = ()> + Send + 'static,
|
||||
{
|
||||
fn spawn(&mut self, future: T) -> Result<(), executor_01::SpawnError> {
|
||||
Handle::spawn(self, future)
|
||||
}
|
||||
}
|
||||
|
||||
/// Error returned by the `run` function.
|
||||
#[derive(Debug)]
|
||||
pub struct RunError {
|
||||
inner: current_thread::RunError,
|
||||
}
|
||||
|
||||
impl fmt::Display for RunError {
|
||||
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(fmt, "{}", self.inner)
|
||||
}
|
||||
}
|
||||
|
||||
impl Error for RunError {
|
||||
fn source(&self) -> Option<&(dyn Error + 'static)> {
|
||||
self.inner.source()
|
||||
}
|
||||
}
|
||||
|
||||
struct CompatExec {
|
||||
inner: current_thread::TaskExecutor,
|
||||
}
|
||||
|
||||
impl Runtime {
|
||||
/// Returns a new runtime initialized with default configuration values.
|
||||
pub fn new() -> io::Result<Runtime> {
|
||||
Builder::new().build()
|
||||
}
|
||||
|
||||
pub(super) fn new2(
|
||||
reactor_handle: driver::Handle,
|
||||
timer_handle: timer::Handle,
|
||||
clock: Clock,
|
||||
executor: CurrentThread<Parker>,
|
||||
) -> io::Result<Runtime> {
|
||||
let compat = compat::Background::spawn(&clock)?;
|
||||
Ok(Runtime {
|
||||
reactor_handle,
|
||||
timer_handle,
|
||||
clock,
|
||||
executor,
|
||||
compat,
|
||||
})
|
||||
}
|
||||
|
||||
/// Get a new handle to spawn futures on the single-threaded Tokio runtime
|
||||
///
|
||||
/// Different to the runtime itself, the handle can be sent to different
|
||||
/// threads.
|
||||
pub fn handle(&self) -> Handle {
|
||||
Handle(self.executor.handle().clone())
|
||||
}
|
||||
|
||||
/// Spawn a `futures` 0.1 future onto the single-threaded Tokio runtime.
|
||||
///
|
||||
/// See [module level][mod] documentation for more details.
|
||||
///
|
||||
/// [mod]: index.html
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::current_thread::Runtime;
|
||||
///
|
||||
/// # fn dox() {
|
||||
/// // Create the runtime
|
||||
/// let mut rt = Runtime::new().unwrap();
|
||||
///
|
||||
/// // Spawn a future onto the runtime
|
||||
/// rt.spawn(futures_01::future::lazy(|| {
|
||||
/// println!("now running on a worker thread");
|
||||
/// Ok(())
|
||||
/// }));
|
||||
/// # }
|
||||
/// ```
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if the spawn fails. Failure occurs if the executor
|
||||
/// is currently at capacity and is unable to spawn a new future.
|
||||
pub fn spawn<F>(&mut self, future: F) -> &mut Self
|
||||
where
|
||||
F: Future01<Item = (), Error = ()> + 'static,
|
||||
{
|
||||
self.executor.spawn(future.compat().map(|_| ()));
|
||||
self
|
||||
}
|
||||
|
||||
/// Spawn a `std::future` future onto the single-threaded Tokio runtime.
|
||||
///
|
||||
/// See [module level][mod] documentation for more details.
|
||||
///
|
||||
/// [mod]: index.html
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::current_thread::Runtime;
|
||||
///
|
||||
/// # fn dox() {
|
||||
/// // Create the runtime
|
||||
/// let mut rt = Runtime::new().unwrap();
|
||||
///
|
||||
/// // Spawn a future onto the runtime
|
||||
/// rt.spawn_std(async {
|
||||
/// println!("now running on a worker thread");
|
||||
/// });
|
||||
/// # }
|
||||
/// ```
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if the spawn fails. Failure occurs if the executor
|
||||
/// is currently at capacity and is unable to spawn a new future.
|
||||
pub fn spawn_std<F>(&mut self, future: F) -> &mut Self
|
||||
where
|
||||
F: Future<Output = ()> + 'static,
|
||||
{
|
||||
self.executor.spawn(future);
|
||||
self
|
||||
}
|
||||
|
||||
/// Runs the provided `futures` 0.1 future, blocking the current thread
|
||||
/// until the future completes.
|
||||
///
|
||||
/// This function can be used to synchronously block the current thread
|
||||
/// until the provided `future` has resolved either successfully or with an
|
||||
/// error. The result of the future is then returned from this function
|
||||
/// call.
|
||||
///
|
||||
/// Note that this function will **also** execute any spawned futures on the
|
||||
/// current thread, but will **not** block until these other spawned futures
|
||||
/// have completed. Once the function returns, any uncompleted futures
|
||||
/// remain pending in the `Runtime` instance. These futures will not run
|
||||
/// until `block_on` or `run` is called again.
|
||||
///
|
||||
/// The caller is responsible for ensuring that other spawned futures
|
||||
/// complete execution by calling `block_on` or `run`.
|
||||
pub fn block_on<F>(&mut self, f: F) -> Result<F::Item, F::Error>
|
||||
where
|
||||
F: Future01,
|
||||
{
|
||||
self.enter(|executor| {
|
||||
// Run the provided future
|
||||
executor.block_on(f.compat())
|
||||
})
|
||||
}
|
||||
|
||||
/// Runs the provided `std::future` future, blocking the current thread
|
||||
/// until the future completes.
|
||||
///
|
||||
/// This function can be used to synchronously block the current thread
|
||||
/// until the provided `future` has resolved either successfully or with an
|
||||
/// error. The result of the future is then returned from this function
|
||||
/// call.
|
||||
///
|
||||
/// Note that this function will **also** execute any spawned futures on the
|
||||
/// current thread, but will **not** block until these other spawned futures
|
||||
/// have completed. Once the function returns, any uncompleted futures
|
||||
/// remain pending in the `Runtime` instance. These futures will not run
|
||||
/// until `block_on` or `run` is called again.
|
||||
///
|
||||
/// The caller is responsible for ensuring that other spawned futures
|
||||
/// complete execution by calling `block_on` or `run`.
|
||||
pub fn block_on_std<F>(&mut self, f: F) -> F::Output
|
||||
where
|
||||
F: Future,
|
||||
{
|
||||
self.enter(|executor| {
|
||||
// Run the provided future
|
||||
executor.block_on(f)
|
||||
})
|
||||
}
|
||||
|
||||
/// Run the executor to completion, blocking the thread until **all**
|
||||
/// spawned futures have completed.
|
||||
pub fn run(&mut self) -> Result<(), RunError> {
|
||||
self.enter(|executor| executor.run())
|
||||
.map_err(|e| RunError { inner: e })
|
||||
}
|
||||
|
||||
fn enter<F, R>(&mut self, f: F) -> R
|
||||
where
|
||||
F: FnOnce(&mut current_thread::CurrentThread<Parker>) -> R,
|
||||
{
|
||||
let Runtime {
|
||||
ref reactor_handle,
|
||||
ref timer_handle,
|
||||
ref clock,
|
||||
ref mut executor,
|
||||
ref compat,
|
||||
} = *self;
|
||||
|
||||
let mut enter = executor_01::enter().unwrap();
|
||||
// Set the default tokio 0.1 reactor to the background compat reactor.
|
||||
reactor_01::with_default(compat.reactor(), &mut enter, |enter| {
|
||||
// This will set the default handle and timer to use inside the closure
|
||||
// and run the future.
|
||||
let _reactor = driver::set_default(&reactor_handle);
|
||||
clock::with_default(clock, || {
|
||||
// Set up a default timer for tokio 0.1 compat.
|
||||
timer_02::with_default(compat.timer(), enter, |enter| {
|
||||
let _timer = timer::set_default(&timer_handle);
|
||||
// Set default executor for tokio 0.1 futures.
|
||||
let mut compat_exec = CompatExec {
|
||||
inner: current_thread::TaskExecutor::current(),
|
||||
};
|
||||
executor_01::with_default(&mut compat_exec, enter, |_enter| {
|
||||
// The TaskExecutor is a fake executor that looks into the
|
||||
// current single-threaded executor when used. This is a trick,
|
||||
// because we need two mutable references to the executor (one
|
||||
// to run the provided future, another to install as the default
|
||||
// one). We use the fake one here as the default one.
|
||||
let mut default_executor = current_thread::TaskExecutor::current();
|
||||
tokio_02::executor::with_default(&mut default_executor, || f(executor))
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl executor_01::Executor for CompatExec {
|
||||
fn spawn(
|
||||
&mut self,
|
||||
future: Box<dyn futures_01::Future<Item = (), Error = ()> + Send>,
|
||||
) -> Result<(), executor_01::SpawnError> {
|
||||
let future = future.compat().map(|_| ());
|
||||
tokio_02::executor::Executor::spawn(&mut self.inner, Box::pin(future))
|
||||
.map_err(compat::spawn_err)
|
||||
}
|
||||
|
||||
fn status(&self) -> Result<(), executor_01::SpawnError> {
|
||||
tokio_02::executor::Executor::status(&self.inner).map_err(compat::spawn_err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
use futures_01::future::{self as future_01, Future as Future01};
|
||||
use futures_util::{compat::Future01CompatExt, FutureExt};
|
||||
use std::{future::Future, pin::Pin};
|
||||
use tokio_02::executor::{self as executor_02, current_thread::TaskExecutor as TaskExecutor02};
|
||||
use tokio_executor_01 as executor_01;
|
||||
|
||||
/// 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(Clone, Debug)]
|
||||
pub struct TaskExecutor {
|
||||
inner: TaskExecutor02,
|
||||
}
|
||||
|
||||
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 {
|
||||
inner: TaskExecutor02::current(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn a `futures` 0.1 future onto the current `CurrentThread` instance.
|
||||
pub fn spawn_local(
|
||||
&mut self,
|
||||
future: impl Future01<Item = (), Error = ()> + 'static,
|
||||
) -> Result<(), executor_01::SpawnError> {
|
||||
let future = Box::pin(future.compat().map(|_| ()));
|
||||
self.spawn_local_std(future).map_err(map_spawn_err)
|
||||
}
|
||||
|
||||
/// Spawn a `std::future` future onto the current `CurrentThread` instance.
|
||||
pub fn spawn_local_std(
|
||||
&mut self,
|
||||
future: Pin<Box<dyn Future<Output = ()>>>,
|
||||
) -> Result<(), executor_02::SpawnError> {
|
||||
self.inner.spawn_local(future)
|
||||
}
|
||||
}
|
||||
|
||||
fn map_spawn_err(new: executor_02::SpawnError) -> executor_01::SpawnError {
|
||||
match new {
|
||||
_ if new.is_shutdown() => executor_01::SpawnError::shutdown(),
|
||||
_ if new.is_at_capacity() => executor_01::SpawnError::at_capacity(),
|
||||
e => unreachable!("weird spawn error {:?}", e),
|
||||
}
|
||||
}
|
||||
|
||||
impl executor_01::Executor for TaskExecutor {
|
||||
fn spawn(
|
||||
&mut self,
|
||||
future: Box<dyn Future01<Item = (), Error = ()> + Send>,
|
||||
) -> Result<(), executor_01::SpawnError> {
|
||||
self.spawn_local(future)
|
||||
}
|
||||
|
||||
fn status(&self) -> Result<(), executor_01::SpawnError> {
|
||||
executor_02::Executor::status(&self.inner).map_err(map_spawn_err)
|
||||
}
|
||||
}
|
||||
|
||||
impl<F> executor_01::TypedExecutor<F> for TaskExecutor
|
||||
where
|
||||
F: Future01<Item = (), Error = ()> + 'static,
|
||||
{
|
||||
fn spawn(&mut self, future: F) -> Result<(), executor_01::SpawnError> {
|
||||
let future = Box::pin(future.compat().map(|_| ()));
|
||||
self.spawn_local_std(future).map_err(map_spawn_err)
|
||||
}
|
||||
|
||||
fn status(&self) -> Result<(), executor_01::SpawnError> {
|
||||
executor_02::Executor::status(&self.inner).map_err(map_spawn_err)
|
||||
}
|
||||
}
|
||||
|
||||
impl executor_02::Executor for TaskExecutor {
|
||||
fn spawn(
|
||||
&mut self,
|
||||
future: Pin<Box<dyn Future<Output = ()> + Send>>,
|
||||
) -> Result<(), executor_02::SpawnError> {
|
||||
self.spawn_local_std(future)
|
||||
}
|
||||
|
||||
fn status(&self) -> Result<(), executor_02::SpawnError> {
|
||||
executor_02::Executor::status(&self.inner)
|
||||
}
|
||||
}
|
||||
|
||||
impl<F> executor_02::TypedExecutor<F> for TaskExecutor
|
||||
where
|
||||
F: Future<Output = ()> + 'static,
|
||||
{
|
||||
fn spawn(&mut self, future: F) -> Result<(), executor_02::SpawnError> {
|
||||
self.spawn_local_std(Box::pin(future))
|
||||
}
|
||||
|
||||
fn status(&self) -> Result<(), executor_02::SpawnError> {
|
||||
executor_02::Executor::status(&self.inner)
|
||||
}
|
||||
}
|
||||
|
||||
impl<F> future_01::Executor<F> for TaskExecutor
|
||||
where
|
||||
F: Future01<Item = (), Error = ()> + 'static,
|
||||
{
|
||||
fn execute(&self, future: F) -> Result<(), future_01::ExecuteError<F>> {
|
||||
match executor_02::Executor::status(&self.inner) {
|
||||
Err(e) if e.is_shutdown() => Err(future_01::ExecuteError::new(
|
||||
future_01::ExecuteErrorKind::Shutdown,
|
||||
future,
|
||||
)),
|
||||
Err(e) if e.is_at_capacity() => Err(future_01::ExecuteError::new(
|
||||
future_01::ExecuteErrorKind::NoCapacity,
|
||||
future,
|
||||
)),
|
||||
Err(e) => panic!("unexpected spawn error {:?}", e),
|
||||
Ok(_) => {
|
||||
let mut this = self.clone();
|
||||
this.spawn_local(future).unwrap_or_else(|e| {
|
||||
debug_assert!(false, "status succeeded, but spawn failed: {}", e)
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
use std::sync::{
|
||||
atomic::{AtomicBool, Ordering},
|
||||
Arc,
|
||||
};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use futures_01::future::Future as Future01;
|
||||
use futures_util::compat::Future01CompatExt;
|
||||
|
||||
#[test]
|
||||
fn can_run_01_futures() {
|
||||
let future_ran = Arc::new(AtomicBool::new(false));
|
||||
let ran = future_ran.clone();
|
||||
|
||||
super::run(futures_01::future::lazy(move || {
|
||||
future_ran.store(true, Ordering::SeqCst);
|
||||
Ok::<(), ()>(())
|
||||
}));
|
||||
assert!(ran.load(Ordering::SeqCst));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn can_spawn_01_futures() {
|
||||
let future_ran = Arc::new(AtomicBool::new(false));
|
||||
let ran = future_ran.clone();
|
||||
super::run(futures_01::future::lazy(move || {
|
||||
tokio_01::spawn(futures_01::future::lazy(move || {
|
||||
future_ran.store(true, Ordering::SeqCst);
|
||||
Ok(())
|
||||
}));
|
||||
Ok(())
|
||||
}));
|
||||
assert!(ran.load(Ordering::SeqCst));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn can_spawn_std_futures() {
|
||||
let future_ran = Arc::new(AtomicBool::new(false));
|
||||
let ran = future_ran.clone();
|
||||
super::run(futures_01::future::lazy(move || {
|
||||
tokio_02::spawn(async move {
|
||||
future_ran.store(true, Ordering::SeqCst);
|
||||
});
|
||||
Ok(())
|
||||
}));
|
||||
assert!(ran.load(Ordering::SeqCst));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tokio_01_timers_work() {
|
||||
let future1_ran = Arc::new(AtomicBool::new(false));
|
||||
let ran = future1_ran.clone();
|
||||
let future1 = futures_01::future::lazy(|| {
|
||||
let when = Instant::now() + Duration::from_millis(15);
|
||||
tokio_01::timer::Delay::new(when).map(move |_| when)
|
||||
})
|
||||
.map(move |when| {
|
||||
ran.store(true, Ordering::SeqCst);
|
||||
assert!(Instant::now() >= when);
|
||||
})
|
||||
.map_err(|_| panic!("timer should work"));
|
||||
|
||||
let future2_ran = Arc::new(AtomicBool::new(false));
|
||||
let ran = future2_ran.clone();
|
||||
let future2 = async move {
|
||||
let when = Instant::now() + Duration::from_millis(10);
|
||||
tokio_01::timer::Delay::new(when).compat().await.unwrap();
|
||||
ran.store(true, Ordering::SeqCst);
|
||||
assert!(Instant::now() >= when);
|
||||
};
|
||||
|
||||
super::run(futures_01::future::lazy(move || {
|
||||
tokio_02::spawn(future2);
|
||||
tokio_01::spawn(future1);
|
||||
Ok(())
|
||||
}));
|
||||
assert!(future1_ran.load(Ordering::SeqCst));
|
||||
assert!(future2_ran.load(Ordering::SeqCst));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_01_timer() {
|
||||
let mut rt = super::Runtime::new().unwrap();
|
||||
let when = Instant::now() + Duration::from_millis(10);
|
||||
rt.block_on(tokio_01::timer::Delay::new(when)).unwrap();
|
||||
assert!(Instant::now() >= when);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_std_01_timer() {
|
||||
let mut rt = super::Runtime::new().unwrap();
|
||||
let when = Instant::now() + Duration::from_millis(10);
|
||||
rt.block_on_std(async move {
|
||||
tokio_01::timer::Delay::new(when).compat().await.unwrap();
|
||||
});
|
||||
assert!(Instant::now() >= when);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_01_spawn() {
|
||||
let mut rt = super::Runtime::new().unwrap();
|
||||
// other tests assert that spawned 0.1 tasks actually *run*, all we care
|
||||
// is that we're able to spawn it successfully.
|
||||
rt.block_on(futures_01::future::lazy(|| {
|
||||
tokio_01::spawn(futures_01::future::lazy(|| Ok(())))
|
||||
}))
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_std_01_spawn() {
|
||||
let mut rt = super::Runtime::new().unwrap();
|
||||
// other tests assert that spawned 0.1 tasks actually *run*, all we care
|
||||
// is that we're able to spawn it successfully.
|
||||
rt.block_on_std(async { tokio_01::spawn(futures_01::future::lazy(|| Ok(()))) });
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
//! Runtimes compatible with both `tokio` 0.1 and `tokio` 0.2 futures.
|
||||
//!
|
||||
//! This module is similar to the [`tokio::runtime`] module, with one
|
||||
//! key difference: the runtimes in this crate are capable of executing
|
||||
//! both `futures` 0.1 futures that use the `tokio` 0.1 runtime services
|
||||
//! (i.e. `timer`, `reactor`, and `executor`), **and** `std::future`
|
||||
//! futures that use the `tokio` 0.2 runtime services.
|
||||
//!
|
||||
//! The `futures` crate's [`compat` module][futures-compat] provides
|
||||
//! interoperability between `futures` 0.1 and `std::future` _future types_
|
||||
//! (e.g. implementing `std::future::Future` for a type that implements the
|
||||
//! `futures` 0.1 `Future` trait). However, this on its own is insufficient to
|
||||
//! run code written against `tokio` 0.1 on a `tokio` 0.2 runtime, if that code
|
||||
//! also relies on `tokio`'s runtime services. If legacy tasks are executed that
|
||||
//! rely on `tokio::timer`, perform IO using `tokio`'s reactor, or call
|
||||
//! `tokio::spawn`, those API calls will fail unless there is also a runtime
|
||||
//! compatibility layer.
|
||||
//!
|
||||
//! `tokio-compat`'s `runtime` module contains modified versions of the `tokio`
|
||||
//! 0.2 `Runtime` and `current_thread::Runtime` that are capable of providing
|
||||
//! `tokio` 0.1 and `tokio` 0.2-compatible runtime services.
|
||||
//!
|
||||
//! Creating a [`Runtime`] does the following:
|
||||
//!
|
||||
//! * Spawn a background thread running a [`Reactor`] instance.
|
||||
//! * Start a [`ThreadPool`] for executing futures.
|
||||
//! * Run an instance of [`Timer`] **per** thread pool worker thread.
|
||||
//! * Run a **single** `tokio` 0.1 [`Reactor`][reactor-01] and
|
||||
//! [`Timer`][timer-01] on a background thread, for legacy tasks.
|
||||
//!
|
||||
//! Legacy `futures` 0.1 tasks will be executed by the `tokio` 0.2 thread pool
|
||||
//! workers, alongside `std::future` tasks. However, they will use the timer and
|
||||
//! reactor provided by the compatibility background thread.
|
||||
//!
|
||||
//! ## Examples
|
||||
//!
|
||||
//! Spawning both `tokio` 0.1 and `tokio` 0.2 futures:
|
||||
//!
|
||||
//! ```rust
|
||||
//! use futures_01::future::lazy;
|
||||
//!
|
||||
//! tokio_compat::run(lazy(|| {
|
||||
//! // spawn a `futures` 0.1 future using the `spawn` function from the
|
||||
//! // `tokio` 0.1 crate:
|
||||
//! tokio_01::spawn(lazy(|| {
|
||||
//! println!("hello from tokio 0.1!");
|
||||
//! Ok(())
|
||||
//! }));
|
||||
//!
|
||||
//! // spawn an `async` block future on the same runtime using `tokio`
|
||||
//! // 0.2's `spawn`:
|
||||
//! tokio_02::spawn(async {
|
||||
//! println!("hello from tokio 0.2!");
|
||||
//! });
|
||||
//!
|
||||
//! Ok(())
|
||||
//! }))
|
||||
//! ```
|
||||
//!
|
||||
//! Futures on the compat runtime can use `timer` APIs from both 0.1 and 0.2
|
||||
//! versions of `tokio`:
|
||||
//!
|
||||
//! ```rust
|
||||
//! # use std::time::{Duration, Instant};
|
||||
//! use tokio_compat::prelude::*;
|
||||
//!
|
||||
//! tokio_compat::run_std(async {
|
||||
//! // Wait for a `tokio` 0.1 `Delay`...
|
||||
//! let when = Instant::now() + Duration::from_millis(10);
|
||||
//! tokio_01::timer::Delay::new(when)
|
||||
//! // convert the delay future into a `std::future` that we can `await`.
|
||||
//! .compat()
|
||||
//! .await
|
||||
//! .expect("tokio 0.1 timer should work!");
|
||||
//! println!("10 ms have elapsed");
|
||||
//!
|
||||
//! // Wait for a `tokio` 0.2 `Delay`...
|
||||
//! let when = Instant::now() + Duration::from_millis(20);
|
||||
//! tokio_02::timer::delay(when).await;
|
||||
//! println!("20 ms have elapsed");
|
||||
//! });
|
||||
//! ```
|
||||
//!
|
||||
//! ## Notes
|
||||
//!
|
||||
//! In order to allow drop-in compatibility for legacy codebases using
|
||||
//! `tokio` 0.1, the [`run`], [`spawn`], and [`block_on`] methods provided by the
|
||||
//! compatibility runtimes take `futures` 0.1 futures. This allows the
|
||||
//! compatibility runtimes to replace the `tokio` 0.1 runtimes in those codebases
|
||||
//! without requiring changes to unrelated code. The compatibility runtimes
|
||||
//! _also_ provide `std::future`-compatible versions of these methods, named
|
||||
//! [`run_std`], [`spawn_std`], and [`block_on_std`].
|
||||
//!
|
||||
//! Also, please note that the compatibility thread pool runtime does **not**
|
||||
//! currently support the `tokio` 0.1 [`tokio_threadpool::blocking][blocking]
|
||||
//! API. Calls to the legacy version of `blocking` made on the compatibility
|
||||
//! runtime will currently fail. In the future, `tokio-compat` will allow
|
||||
//! transparently replacing legacy `blocking` with the `tokio` 0.2 blocking
|
||||
//! APIs, but in the meantime, it will be necessary to convert this code to call
|
||||
//! into the `tokio` 0.2 version of `blocking`.
|
||||
//!
|
||||
//! [`tokio::runtime`]: https://docs.rs/tokio/0.2.0-alpha.6/tokio/runtime/index.html
|
||||
//! [futures-compat]: https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures/compat/index.html
|
||||
//! [`Timer`]: https://docs.rs/tokio/0.2.0-alpha.6/tokio/timer/index.html
|
||||
//! [`Runtime`]: struct.Runtime.html
|
||||
//! [`Reactor`]:https://docs.rs/tokio/0.2.0-alpha.6/tokio/reactor/struct.Reactor.html
|
||||
//! [timer-01]: https://docs.rs/tokio/0.1.22/tokio/timer/index.html
|
||||
//! [reactor-01]: https://docs.rs/tokio/0.1.22/tokio/reactor/struct.Reactor.html
|
||||
//! [`ThreadPool`]: https://docs.rs/tokio-executor/0.2.0-alpha.2/tokio_executor/threadpool/struct.ThreadPool.html
|
||||
//! [`run`]: struct.Runtime.html#method.run
|
||||
//! [`spawn`]: struct.Runtime.html#method.spawn
|
||||
//! [`block_on`]: struct.Runtime.html#method.block_on
|
||||
//! [`run_std`]: struct.Runtime.html#method.run_std
|
||||
//! [`spawn_std`]: struct.Runtime.html#method.spawn_std
|
||||
//! [`block_on_std`]: struct.Runtime.html#method.spawn_std
|
||||
//! [blocking]: https://docs.rs/tokio-threadpool/0.1.16/tokio_threadpool/fn.blocking.html
|
||||
mod compat;
|
||||
pub mod current_thread;
|
||||
#[cfg(feature = "rt-full")]
|
||||
mod threadpool;
|
||||
|
||||
#[cfg(feature = "rt-full")]
|
||||
pub use threadpool::{run, run_std, Builder, Runtime, TaskExecutor};
|
||||
@@ -0,0 +1,61 @@
|
||||
//! Temporary reactor + timer that runs on a background thread. This it to make
|
||||
//! `block_on` work.
|
||||
|
||||
use tokio_02::executor::current_thread::CurrentThread;
|
||||
use tokio_02::net::driver::{self, Reactor};
|
||||
use tokio_02::sync::oneshot;
|
||||
use tokio_02::timer::clock::Clock;
|
||||
use tokio_02::timer::timer::{self, Timer};
|
||||
|
||||
use std::{io, thread};
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct Background {
|
||||
reactor_handle: driver::Handle,
|
||||
timer_handle: timer::Handle,
|
||||
shutdown_tx: Option<oneshot::Sender<()>>,
|
||||
thread: Option<thread::JoinHandle<()>>,
|
||||
}
|
||||
|
||||
pub(crate) fn spawn(clock: &Clock) -> io::Result<Background> {
|
||||
let clock = clock.clone();
|
||||
|
||||
let reactor = Reactor::new()?;
|
||||
let reactor_handle = reactor.handle();
|
||||
|
||||
let timer = Timer::new_with_clock(reactor, clock);
|
||||
let timer_handle = timer.handle();
|
||||
|
||||
let (shutdown_tx, shutdown_rx) = oneshot::channel();
|
||||
let shutdown_tx = Some(shutdown_tx);
|
||||
|
||||
let thread = thread::spawn(move || {
|
||||
let mut rt = CurrentThread::new_with_park(timer);
|
||||
let _ = rt.block_on(shutdown_rx);
|
||||
});
|
||||
let thread = Some(thread);
|
||||
|
||||
Ok(Background {
|
||||
reactor_handle,
|
||||
timer_handle,
|
||||
shutdown_tx,
|
||||
thread,
|
||||
})
|
||||
}
|
||||
|
||||
impl Background {
|
||||
pub(super) fn reactor(&self) -> &driver::Handle {
|
||||
&self.reactor_handle
|
||||
}
|
||||
|
||||
pub(super) fn timer(&self) -> &timer::Handle {
|
||||
&self.timer_handle
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Background {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.shutdown_tx.take().unwrap().send(());
|
||||
let _ = self.thread.take().unwrap().join();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,285 @@
|
||||
use super::{background, compat, Inner, Runtime};
|
||||
|
||||
use tokio_02::executor::thread_pool;
|
||||
use tokio_02::net::driver::{self, Reactor};
|
||||
use tokio_02::timer::clock::{self, Clock};
|
||||
use tokio_02::timer::timer::{self, Timer};
|
||||
use tokio_executor_01 as executor_01;
|
||||
use tokio_reactor_01 as reactor_01;
|
||||
use tokio_timer_02 as timer_02;
|
||||
|
||||
use num_cpus;
|
||||
use std::io;
|
||||
use std::sync::{Arc, Barrier, Mutex, RwLock};
|
||||
|
||||
/// Builds a compatibility runtime with custom configuration values.
|
||||
///
|
||||
/// This runtime is compatible with code using both the current release version
|
||||
/// of `tokio` (0.1) and with legacy code using `tokio` 0.1.
|
||||
///
|
||||
/// Methods can be chained in order to set the configuration values. The
|
||||
/// Runtime is constructed by calling [`build`].
|
||||
///
|
||||
/// New instances of `Builder` are obtained via [`Builder::new`].
|
||||
///
|
||||
/// See function level documentation for details on the various configuration
|
||||
/// settings.
|
||||
///
|
||||
/// [`build`]: #method.build
|
||||
/// [`Builder::new`]: #method.new
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
///
|
||||
/// use tokio_compat::runtime::Builder;
|
||||
/// use tokio_02::timer::clock::Clock;
|
||||
///
|
||||
/// fn main() {
|
||||
/// // build Runtime
|
||||
/// let runtime = Builder::new()
|
||||
/// .clock(Clock::system())
|
||||
/// .core_threads(4)
|
||||
/// .name_prefix("my-custom-name-")
|
||||
/// .stack_size(3 * 1024 * 1024)
|
||||
/// .build()
|
||||
/// .unwrap();
|
||||
///
|
||||
/// // use runtime ...
|
||||
/// }
|
||||
/// ```
|
||||
#[derive(Debug)]
|
||||
pub struct Builder {
|
||||
/// Thread pool specific builder
|
||||
threadpool_builder: thread_pool::Builder,
|
||||
|
||||
/// The number of worker threads
|
||||
core_threads: usize,
|
||||
|
||||
/// The clock to use
|
||||
clock: Clock,
|
||||
}
|
||||
|
||||
impl Builder {
|
||||
/// Returns a new runtime builder initialized with default configuration
|
||||
/// values.
|
||||
///
|
||||
/// Configuration methods can be chained on the return value.
|
||||
pub fn new() -> Builder {
|
||||
let core_threads = num_cpus::get().max(1);
|
||||
|
||||
let mut threadpool_builder = thread_pool::Builder::new();
|
||||
threadpool_builder.name("tokio-runtime-worker");
|
||||
threadpool_builder.num_threads(core_threads);
|
||||
|
||||
Builder {
|
||||
threadpool_builder,
|
||||
core_threads,
|
||||
clock: Clock::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Set the `Clock` instance that will be used by the runtime.
|
||||
pub fn clock(&mut self, clock: Clock) -> &mut Self {
|
||||
self.clock = clock;
|
||||
self
|
||||
}
|
||||
|
||||
/// Set the maximum number of worker threads for the `Runtime`'s thread pool.
|
||||
///
|
||||
/// This must be a number between 1 and 32,768 though it is advised to keep
|
||||
/// this value on the smaller side.
|
||||
///
|
||||
/// The default value is the number of cores available to the system.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// # use tokio_compat::runtime;
|
||||
///
|
||||
/// # pub fn main() {
|
||||
/// let rt = runtime::Builder::new()
|
||||
/// .core_threads(4)
|
||||
/// .build()
|
||||
/// .unwrap();
|
||||
/// # }
|
||||
/// ```
|
||||
pub fn core_threads(&mut self, val: usize) -> &mut Self {
|
||||
self.core_threads = val;
|
||||
self.threadpool_builder.num_threads(val);
|
||||
self
|
||||
}
|
||||
|
||||
/// Set name prefix of threads spawned by the `Runtime`'s thread pool.
|
||||
///
|
||||
/// Thread name prefix is used for generating thread names. For example, if
|
||||
/// prefix is `my-pool-`, then threads in the pool will get names like
|
||||
/// `my-pool-1` etc.
|
||||
///
|
||||
/// The default prefix is "tokio-runtime-worker-".
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// # use tokio_compat::runtime;
|
||||
///
|
||||
/// # pub fn main() {
|
||||
/// let rt = runtime::Builder::new()
|
||||
/// .name_prefix("my-pool-")
|
||||
/// .build();
|
||||
/// # }
|
||||
/// ```
|
||||
pub fn name_prefix<S: Into<String>>(&mut self, val: S) -> &mut Self {
|
||||
self.threadpool_builder.name(val);
|
||||
self
|
||||
}
|
||||
|
||||
/// Set the stack size (in bytes) for worker threads.
|
||||
///
|
||||
/// The actual stack size may be greater than this value if the platform
|
||||
/// specifies minimal stack size.
|
||||
///
|
||||
/// The default stack size for spawned threads is 2 MiB, though this
|
||||
/// particular stack size is subject to change in the future.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// # use tokio_compat::runtime;
|
||||
///
|
||||
/// # pub fn main() {
|
||||
/// let rt = runtime::Builder::new()
|
||||
/// .stack_size(32 * 1024)
|
||||
/// .build();
|
||||
/// # }
|
||||
/// ```
|
||||
pub fn stack_size(&mut self, val: usize) -> &mut Self {
|
||||
self.threadpool_builder.stack_size(val);
|
||||
self
|
||||
}
|
||||
|
||||
/// Create the configured `Runtime`.
|
||||
///
|
||||
/// The returned `ThreadPool` instance is ready to spawn tasks.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// # use tokio_compat::runtime::Builder;
|
||||
/// # pub fn main() {
|
||||
/// let runtime = Builder::new().build().unwrap();
|
||||
/// // ... call runtime.run(...)
|
||||
/// # let _ = runtime;
|
||||
/// # }
|
||||
/// ```
|
||||
pub fn build(&mut self) -> io::Result<Runtime> {
|
||||
self.threadpool_builder.num_threads(self.core_threads);
|
||||
|
||||
let mut reactor_handles = Vec::new();
|
||||
let mut timer_handles = Vec::new();
|
||||
let mut timers = Vec::new();
|
||||
|
||||
for _ in 0..self.core_threads {
|
||||
// Create a new reactor.
|
||||
let reactor = Reactor::new()?;
|
||||
reactor_handles.push(reactor.handle());
|
||||
|
||||
// Create a new timer.
|
||||
let timer = Timer::new_with_clock(reactor, self.clock.clone());
|
||||
timer_handles.push(timer.handle());
|
||||
timers.push(Mutex::new(Some(timer)));
|
||||
}
|
||||
|
||||
// Get a handle to the clock for the runtime.
|
||||
let clock = self.clock.clone();
|
||||
|
||||
// Get the current trace dispatcher.
|
||||
let dispatch = tracing_core::dispatcher::get_default(tracing_core::Dispatch::clone);
|
||||
let trace = dispatch.clone();
|
||||
|
||||
let background = background::spawn(&clock)?;
|
||||
let compat_bg = compat::Background::spawn(&clock)?;
|
||||
let compat_reactor = compat_bg.reactor().clone();
|
||||
let compat_timer = compat_bg.timer().clone();
|
||||
|
||||
// The `tokio` 0.2 default executor for the worker threads will be set
|
||||
// by the threadpool itself, but in order to set a default executor for
|
||||
// `tokio-executor` 0.1 compatibility, we need a `Sender` for the pool
|
||||
// in the `around_worker` closure.
|
||||
//
|
||||
// Unfortunately, we can't get a sender until the pool is constructed,
|
||||
// which requires the `around_worker` closure. As a workaround, we can
|
||||
// an `Arc<RwLock>`, which can be moved into the closure, and then be
|
||||
// set once the pool is constructed. Since the closures won't _run_
|
||||
// until we actually try to run futures on the pool, it's okay to set
|
||||
// the sender after constructing the pool.
|
||||
//
|
||||
// The lock is only acquired in `around_worker`; once it is acquired the
|
||||
// sender is cloned, and the lock doesn't need to be acquired to spawn a
|
||||
// future. Since we only use it when creating the pool, there shouldn't
|
||||
// be much of a performance impact.
|
||||
let compat_sender = Arc::new((RwLock::new(None), Barrier::new(self.core_threads + 1)));
|
||||
let compat_sender2 = compat_sender.clone();
|
||||
|
||||
let pool = self
|
||||
.threadpool_builder
|
||||
.around_worker(move |index, work| {
|
||||
let mut enter = executor_01::enter().unwrap();
|
||||
// We need the threadpool's sender to set up the default tokio
|
||||
// 0.1 executor.
|
||||
let (compat_sender, sender_ready) = &*compat_sender2;
|
||||
// Wait for the sender to be set.
|
||||
sender_ready.wait();
|
||||
let mut compat_sender = compat_sender
|
||||
.read()
|
||||
.unwrap()
|
||||
.clone()
|
||||
.expect("compat executor needs to be set before the pool is run!");
|
||||
|
||||
// Set the default tokio 0.1 reactor to the background compat reactor.
|
||||
reactor_01::with_default(&compat_reactor, &mut enter, |enter| {
|
||||
// Set the default tokio 0.2 reactor to this worker thread's
|
||||
// reactor.
|
||||
let _reactor = driver::set_default(&reactor_handles[index]);
|
||||
clock::with_default(&clock, || {
|
||||
// Set up a default timer for tokio 0.1 compat.
|
||||
timer_02::with_default(&compat_timer, enter, |enter| {
|
||||
let _timer = timer::set_default(&timer_handles[index]);
|
||||
tracing_core::dispatcher::with_default(&dispatch, || {
|
||||
// Set the default executor for tokio 0.1 compat.
|
||||
executor_01::with_default(&mut compat_sender, enter, |_enter| {
|
||||
work();
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
});
|
||||
})
|
||||
.build_with_park(move |index| timers[index].lock().unwrap().take().unwrap());
|
||||
|
||||
let (idle, idle_rx) = super::idle::Idle::new();
|
||||
let runtime = Runtime {
|
||||
inner: Some(Inner {
|
||||
pool,
|
||||
background,
|
||||
compat_bg,
|
||||
trace,
|
||||
}),
|
||||
idle_rx,
|
||||
idle,
|
||||
};
|
||||
|
||||
// Set the tokio 0.1 executor to be used by the worker threads.
|
||||
let (compat_sender, sender_ready) = &*compat_sender;
|
||||
*compat_sender.write().unwrap() = Some(runtime.spawner());
|
||||
sender_ready.wait();
|
||||
|
||||
Ok(runtime)
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for Builder {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
use std::future::Future;
|
||||
use std::sync::{
|
||||
atomic::{fence, AtomicUsize, Ordering},
|
||||
Arc,
|
||||
};
|
||||
use tokio_02::sync::mpsc;
|
||||
|
||||
/// Tracks the number of tasks spawned on a runtime.
|
||||
///
|
||||
/// This is required to implement `shutdown_on_idle` and `tokio::run` APIs that
|
||||
/// exist in `tokio` 0.1, as the `tokio` 0.2 threadpool does not expose a
|
||||
/// `shutdown_on_idle` API.
|
||||
#[derive(Clone, Debug)]
|
||||
pub(super) struct Idle {
|
||||
tx: mpsc::Sender<()>,
|
||||
spawned: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
/// Wraps a future to decrement the spawned count when it completes.
|
||||
///
|
||||
/// This is obtained from `Idle::reserve`.
|
||||
pub(super) struct Track(Idle);
|
||||
|
||||
impl Idle {
|
||||
pub(super) fn new() -> (Self, mpsc::Receiver<()>) {
|
||||
let (tx, rx) = mpsc::channel(1);
|
||||
let this = Self {
|
||||
tx,
|
||||
spawned: Arc::new(AtomicUsize::new(0)),
|
||||
};
|
||||
(this, rx)
|
||||
}
|
||||
|
||||
/// Prepare to spawn a task on the runtime, incrementing the spawned count.
|
||||
pub(super) fn reserve(&self) -> Track {
|
||||
self.spawned.fetch_add(1, Ordering::Relaxed);
|
||||
Track(self.clone())
|
||||
}
|
||||
}
|
||||
|
||||
impl Track {
|
||||
/// Run a task, decrementing the spawn count when it completes.
|
||||
///
|
||||
/// If the spawned count is now 0, this sends a notification on the idle channel.
|
||||
pub(super) async fn with<T>(mut self, f: impl Future<Output = T>) -> T {
|
||||
let result = f.await;
|
||||
let spawned = self.0.spawned.fetch_sub(1, Ordering::Release);
|
||||
if spawned == 1 {
|
||||
fence(Ordering::Acquire);
|
||||
let _ = self.0.tx.send(()).await;
|
||||
}
|
||||
result
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,545 @@
|
||||
mod background;
|
||||
mod builder;
|
||||
mod idle;
|
||||
mod task_executor;
|
||||
|
||||
#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
|
||||
pub use builder::Builder;
|
||||
#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
|
||||
pub use task_executor::TaskExecutor;
|
||||
|
||||
use super::compat;
|
||||
use background::Background;
|
||||
|
||||
use futures_01::future::Future as Future01;
|
||||
use futures_util::{compat::Future01CompatExt, future::FutureExt};
|
||||
use std::{future::Future, io, pin::Pin};
|
||||
use tokio_02::executor::enter;
|
||||
use tokio_02::executor::thread_pool::{Spawner, ThreadPool};
|
||||
use tokio_02::net::driver;
|
||||
use tokio_02::timer::timer;
|
||||
use tokio_executor_01 as executor_01;
|
||||
use tokio_reactor_01 as reactor_01;
|
||||
use tokio_timer_02 as timer_02;
|
||||
|
||||
/// A thread pool runtime that can run tasks that use both `tokio` 0.1 and
|
||||
/// `tokio` 0.2 APIs.
|
||||
///
|
||||
/// This functions similarly to the [`tokio::runtime::Runtime`][rt] struct in the
|
||||
/// `tokio` crate. However, unlike that runtime, the `tokio-compat` runtime is
|
||||
/// capable of running both `std::future::Future` tasks that use `tokio` 0.2
|
||||
/// runtime services. and `futures` 0.1 tasks that use `tokio` 0.1 runtime
|
||||
/// services.
|
||||
///
|
||||
/// [rt]: https://docs.rs/tokio/0.2.0-alpha.6/tokio/runtime/struct.Runtime.html
|
||||
#[derive(Debug)]
|
||||
pub struct Runtime {
|
||||
inner: Option<Inner>,
|
||||
idle: idle::Idle,
|
||||
idle_rx: tokio_02::sync::mpsc::Receiver<()>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Inner {
|
||||
/// Task execution pool.
|
||||
pool: ThreadPool,
|
||||
|
||||
/// Tracing dispatcher
|
||||
trace: tracing_core::Dispatch,
|
||||
|
||||
/// Maintains a reactor and timer that are always running on a background
|
||||
/// thread. This is to support `runtime.block_on` w/o requiring the future
|
||||
/// to be `Send`.
|
||||
///
|
||||
/// A dedicated background thread is required as the threadpool threads
|
||||
/// might not be running. However, this is a temporary work around.
|
||||
background: Background,
|
||||
|
||||
/// Compatibility background thread.
|
||||
///
|
||||
/// This maintains a `tokio` 0.1 timer and reactor to support running
|
||||
/// futures that use older tokio APIs.
|
||||
compat_bg: compat::Background,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct CompatSpawner<S> {
|
||||
inner: S,
|
||||
idle: idle::Idle,
|
||||
}
|
||||
|
||||
// ===== impl Runtime =====
|
||||
|
||||
/// Start the Tokio runtime using the supplied `futures` 0.1 future to bootstrap
|
||||
/// execution.
|
||||
///
|
||||
/// This function is used to bootstrap the execution of a Tokio application. It
|
||||
/// does the following:
|
||||
///
|
||||
/// * Start the Tokio runtime using a default configuration.
|
||||
/// * Spawn the given future onto the thread pool.
|
||||
/// * Block the current thread until the runtime shuts down.
|
||||
///
|
||||
/// Note that the function will not return immediately once `future` has
|
||||
/// completed. Instead it waits for the entire runtime to become idle.
|
||||
///
|
||||
/// See the [module level][mod] documentation for more details.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```rust
|
||||
/// use futures_01::{Future as Future01, Stream as Stream01};
|
||||
/// use tokio_01::net::TcpListener;
|
||||
///
|
||||
/// # fn process<T>(_: T) -> Box<dyn Future01<Item = (), Error = ()> + Send> {
|
||||
/// # unimplemented!();
|
||||
/// # }
|
||||
/// # fn dox() {
|
||||
/// # let addr = "127.0.0.1:8080".parse().unwrap();
|
||||
/// let listener = TcpListener::bind(&addr).unwrap();
|
||||
///
|
||||
/// let server = listener.incoming()
|
||||
/// .map_err(|e| println!("error = {:?}", e))
|
||||
/// .for_each(|socket| {
|
||||
/// tokio_01::spawn(process(socket))
|
||||
/// });
|
||||
///
|
||||
/// tokio_compat::run(server);
|
||||
/// # }
|
||||
/// # pub fn main() {}
|
||||
/// ```
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if called from the context of an executor.
|
||||
///
|
||||
/// [mod]: ../index.html
|
||||
pub fn run<F>(future: F)
|
||||
where
|
||||
F: Future01<Item = (), Error = ()> + Send + 'static,
|
||||
{
|
||||
run_std(future.compat().map(|_| ()))
|
||||
}
|
||||
|
||||
/// Start the Tokio runtime using the supplied `std::future` future to bootstrap
|
||||
/// execution.
|
||||
///
|
||||
/// This function is used to bootstrap the execution of a Tokio application. It
|
||||
/// does the following:
|
||||
///
|
||||
/// * Start the Tokio runtime using a default configuration.
|
||||
/// * Spawn the given future onto the thread pool.
|
||||
/// * Block the current thread until the runtime shuts down.
|
||||
///
|
||||
/// Note that the function will not return immediately once `future` has
|
||||
/// completed. Instead it waits for the entire runtime to become idle.
|
||||
///
|
||||
/// See the [module level][mod] documentation for more details.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```rust
|
||||
/// use futures_01::{Future as Future01, Stream as Stream01};
|
||||
/// use tokio_01::net::TcpListener;
|
||||
///
|
||||
/// # fn process<T>(_: T) -> Box<dyn Future01<Item = (), Error = ()> + Send> {
|
||||
/// # unimplemented!();
|
||||
/// # }
|
||||
/// # fn dox() {
|
||||
/// # let addr = "127.0.0.1:8080".parse().unwrap();
|
||||
/// let listener = TcpListener::bind(&addr).unwrap();
|
||||
///
|
||||
/// let server = listener.incoming()
|
||||
/// .map_err(|e| println!("error = {:?}", e))
|
||||
/// .for_each(|socket| {
|
||||
/// tokio_01::spawn(process(socket))
|
||||
/// });
|
||||
///
|
||||
/// tokio_compat::run(server);
|
||||
/// # }
|
||||
/// # pub fn main() {}
|
||||
/// ```
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if called from the context of an executor.
|
||||
///
|
||||
/// [mod]: ../index.html
|
||||
pub fn run_std<F>(future: F)
|
||||
where
|
||||
F: Future<Output = ()> + Send + 'static,
|
||||
{
|
||||
// Check enter before creating a new Runtime...
|
||||
let runtime = Runtime::new().expect("failed to start new Runtime");
|
||||
runtime.spawn_std(future);
|
||||
runtime.shutdown_on_idle();
|
||||
}
|
||||
|
||||
impl Runtime {
|
||||
/// Create a new runtime instance with default configuration values.
|
||||
///
|
||||
/// This results in a reactor, thread pool, and timer being initialized. The
|
||||
/// thread pool will not spawn any worker threads until it needs to, i.e.
|
||||
/// tasks are scheduled to run.
|
||||
///
|
||||
/// Most users will not need to call this function directly, instead they
|
||||
/// will use [`tokio_compat::run`](fn.run.html).
|
||||
///
|
||||
/// See [module level][mod] documentation for more details.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// Creating a new `Runtime` with default configuration values.
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::Runtime;
|
||||
///
|
||||
/// let rt = Runtime::new()
|
||||
/// .unwrap();
|
||||
///
|
||||
/// // Use the runtime...
|
||||
/// ```
|
||||
///
|
||||
/// [mod]: index.html
|
||||
pub fn new() -> io::Result<Self> {
|
||||
Builder::new().build()
|
||||
}
|
||||
|
||||
/// Return a handle to the runtime's executor.
|
||||
///
|
||||
/// The returned handle can be used to spawn both `futures` 0.1 and
|
||||
/// `std::future` tasks that run on this runtime.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::Runtime;
|
||||
///
|
||||
/// let rt = Runtime::new()
|
||||
/// .unwrap();
|
||||
///
|
||||
/// let executor_handle = rt.executor();
|
||||
///
|
||||
/// // use `executor_handle`
|
||||
/// ```
|
||||
pub fn executor(&self) -> TaskExecutor {
|
||||
let inner = self.spawner();
|
||||
TaskExecutor { inner }
|
||||
}
|
||||
|
||||
/// Spawn a `futures` 0.1 future onto the Tokio runtime.
|
||||
///
|
||||
/// This spawns the given future onto the runtime's executor, usually a
|
||||
/// thread pool. The thread pool is then responsible for polling the future
|
||||
/// until it completes.
|
||||
///
|
||||
/// See [module level][mod] documentation for more details.
|
||||
///
|
||||
/// [mod]: index.html
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::Runtime;
|
||||
///
|
||||
/// fn main() {
|
||||
/// // Create the runtime
|
||||
/// let rt = Runtime::new().unwrap();
|
||||
///
|
||||
/// // Spawn a future onto the runtime
|
||||
/// rt.spawn(futures_01::future::lazy(|| {
|
||||
/// println!("now running on a worker thread");
|
||||
/// Ok(())
|
||||
/// }));
|
||||
///
|
||||
/// rt.shutdown_on_idle();
|
||||
/// }
|
||||
/// ```
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if the spawn fails. Failure occurs if the executor
|
||||
/// is currently at capacity and is unable to spawn a new future.
|
||||
pub fn spawn<F>(&self, future: F) -> &Self
|
||||
where
|
||||
F: Future01<Item = (), Error = ()> + Send + 'static,
|
||||
{
|
||||
self.spawn_std(future.compat().map(|_| ()))
|
||||
}
|
||||
|
||||
/// Spawn a `std::future` future onto the Tokio runtime.
|
||||
///
|
||||
/// This spawns the given future onto the runtime's executor, usually a
|
||||
/// thread pool. The thread pool is then responsible for polling the future
|
||||
/// until it completes.
|
||||
///
|
||||
/// See [module level][mod] documentation for more details.
|
||||
///
|
||||
/// [mod]: index.html
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::Runtime;
|
||||
///
|
||||
/// fn main() {
|
||||
/// // Create the runtime
|
||||
/// let rt = Runtime::new().unwrap();
|
||||
///
|
||||
/// // Spawn a future onto the runtime
|
||||
/// rt.spawn_std(async {
|
||||
/// println!("now running on a worker thread");
|
||||
/// });
|
||||
///
|
||||
/// rt.shutdown_on_idle();
|
||||
/// }
|
||||
/// ```
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if the spawn fails. Failure occurs if the executor
|
||||
/// is currently at capacity and is unable to spawn a new future.
|
||||
pub fn spawn_std<F>(&self, future: F) -> &Self
|
||||
where
|
||||
F: Future<Output = ()> + Send + 'static,
|
||||
{
|
||||
let idle = self.idle.reserve();
|
||||
self.inner().pool.spawn(idle.with(future));
|
||||
self
|
||||
}
|
||||
|
||||
/// Run a `futures` 0.1 future to completion on the Tokio runtime.
|
||||
///
|
||||
/// This runs the given future on the runtime, blocking until it is
|
||||
/// complete, and yielding its resolved result. Any tasks or timers which
|
||||
/// the future spawns internally will be executed on the runtime.
|
||||
///
|
||||
/// This method should not be called from an asynchronous context.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if the executor is at capacity, if the provided
|
||||
/// future panics, or if called within an asynchronous execution context.
|
||||
pub fn block_on<F>(&self, future: F) -> Result<F::Item, F::Error>
|
||||
where
|
||||
F: Future01,
|
||||
{
|
||||
self.block_on_std(future.compat())
|
||||
}
|
||||
|
||||
/// Run a `std::future` future to completion on the Tokio runtime.
|
||||
///
|
||||
/// This runs the given future on the runtime, blocking until it is
|
||||
/// complete, and yielding its resolved result. Any tasks or timers which
|
||||
/// the future spawns internally will be executed on the runtime.
|
||||
///
|
||||
/// This method should not be called from an asynchronous context.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if the executor is at capacity, if the provided
|
||||
/// future panics, or if called within an asynchronous execution context.
|
||||
pub fn block_on_std<F>(&self, future: F) -> F::Output
|
||||
where
|
||||
F: Future,
|
||||
{
|
||||
let mut entered = enter().expect("nested block_on");
|
||||
let mut old_entered = executor_01::enter().expect("nested block_on");
|
||||
let bg = &self.inner().background;
|
||||
let trace = &self.inner().trace;
|
||||
let compat = &self.inner().compat_bg;
|
||||
|
||||
tokio_02::executor::with_default(&mut self.spawner_ref(), || {
|
||||
executor_01::with_default(
|
||||
&mut self.spawner_ref(),
|
||||
&mut old_entered,
|
||||
|mut old_entered| {
|
||||
let _reactor = driver::set_default(bg.reactor());
|
||||
let _timer = timer::set_default(bg.timer());
|
||||
// Set up a default timer for tokio 0.1 compat.
|
||||
reactor_01::with_default(
|
||||
compat.reactor(),
|
||||
&mut old_entered,
|
||||
|mut old_entered| {
|
||||
// Set the default tokio 0.2 reactor to this worker thread's
|
||||
// reactor.
|
||||
timer_02::with_default(
|
||||
compat.timer(),
|
||||
&mut old_entered,
|
||||
|_old_entered| {
|
||||
tracing_core::dispatcher::with_default(trace, || {
|
||||
entered.block_on(future)
|
||||
})
|
||||
},
|
||||
)
|
||||
},
|
||||
)
|
||||
},
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
/// Signals the runtime to shutdown once it becomes idle.
|
||||
///
|
||||
/// Blocks the current thread until the shutdown operation has completed.
|
||||
/// This function can be used to perform a graceful shutdown of the runtime.
|
||||
///
|
||||
/// The runtime enters an idle state once **all** of the following occur.
|
||||
///
|
||||
/// * The thread pool has no tasks to execute, i.e., all tasks that were
|
||||
/// spawned have completed.
|
||||
/// * The reactor is not managing any I/O resources.
|
||||
///
|
||||
/// See [module level][mod] documentation for more details.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::Runtime;
|
||||
///
|
||||
/// let rt = Runtime::new()
|
||||
/// .unwrap();
|
||||
///
|
||||
/// // Use the runtime...
|
||||
/// # rt.spawn_std(async {});
|
||||
///
|
||||
/// // Shutdown the runtime
|
||||
/// rt.shutdown_on_idle();
|
||||
/// ```
|
||||
///
|
||||
/// [mod]: index.html
|
||||
pub fn shutdown_on_idle(mut self) {
|
||||
use futures_util::stream::StreamExt;
|
||||
let mut e = tokio_02::executor::enter().unwrap();
|
||||
|
||||
e.block_on(self.idle_rx.next());
|
||||
}
|
||||
|
||||
/// Signals the runtime to shutdown immediately.
|
||||
///
|
||||
/// Blocks the current thread until the shutdown operation has completed.
|
||||
/// This function will forcibly shutdown the runtime, causing any
|
||||
/// in-progress work to become canceled.
|
||||
///
|
||||
/// The shutdown steps are:
|
||||
///
|
||||
/// * Drain any scheduled work queues.
|
||||
/// * Drop any futures that have not yet completed.
|
||||
/// * Drop the reactor.
|
||||
///
|
||||
/// Once the reactor has dropped, any outstanding I/O resources bound to
|
||||
/// that reactor will no longer function. Calling any method on them will
|
||||
/// result in an error.
|
||||
///
|
||||
/// See [module level][mod] documentation for more details.
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::Runtime;
|
||||
///
|
||||
/// let rt = Runtime::new()
|
||||
/// .unwrap();
|
||||
///
|
||||
/// // Use the runtime...
|
||||
///
|
||||
/// // Shutdown the runtime
|
||||
/// rt.shutdown_now();
|
||||
/// ```
|
||||
///
|
||||
/// [mod]: index.html
|
||||
#[allow(warnings)]
|
||||
pub fn shutdown_now(mut self) {
|
||||
self.inner.take().unwrap().pool.shutdown_now();
|
||||
}
|
||||
|
||||
fn spawner(&self) -> CompatSpawner<Spawner> {
|
||||
CompatSpawner {
|
||||
inner: self.inner().pool.spawner().clone(),
|
||||
idle: self.idle.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
fn spawner_ref(&self) -> CompatSpawner<&'_ Spawner> {
|
||||
CompatSpawner {
|
||||
inner: self.inner().pool.spawner(),
|
||||
idle: self.idle.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
fn inner(&self) -> &Inner {
|
||||
self.inner.as_ref().unwrap()
|
||||
}
|
||||
}
|
||||
|
||||
impl tokio_02::executor::Executor for CompatSpawner<&'_ Spawner> {
|
||||
fn spawn(
|
||||
&mut self,
|
||||
future: Pin<Box<dyn Future<Output = ()> + Send>>,
|
||||
) -> Result<(), tokio_02::executor::SpawnError> {
|
||||
let idle = self.idle.reserve();
|
||||
self.inner.spawn(idle.with(future));
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl executor_01::Executor for CompatSpawner<&'_ Spawner> {
|
||||
fn spawn(
|
||||
&mut self,
|
||||
future: Box<dyn futures_01::Future<Item = (), Error = ()> + Send>,
|
||||
) -> Result<(), executor_01::SpawnError> {
|
||||
let future = future.compat().map(|_| ());
|
||||
let idle = self.idle.reserve();
|
||||
self.inner.spawn(idle.with(future));
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl tokio_02::executor::Executor for CompatSpawner<Spawner> {
|
||||
fn spawn(
|
||||
&mut self,
|
||||
future: Pin<Box<dyn Future<Output = ()> + Send>>,
|
||||
) -> Result<(), tokio_02::executor::SpawnError> {
|
||||
let idle = self.idle.reserve();
|
||||
self.inner.spawn(idle.with(future));
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl executor_01::Executor for CompatSpawner<Spawner> {
|
||||
fn spawn(
|
||||
&mut self,
|
||||
future: Box<dyn futures_01::Future<Item = (), Error = ()> + Send>,
|
||||
) -> Result<(), executor_01::SpawnError> {
|
||||
let future = future.compat().map(|_| ());
|
||||
let idle = self.idle.reserve();
|
||||
self.inner.spawn(idle.with(future));
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> executor_01::TypedExecutor<T> for CompatSpawner<Spawner>
|
||||
where
|
||||
T: Future01<Item = (), Error = ()> + Send + 'static,
|
||||
{
|
||||
fn spawn(&mut self, future: T) -> Result<(), executor_01::SpawnError> {
|
||||
let idle = self.idle.reserve();
|
||||
let future = Box::pin(idle.with(future.compat().map(|_| ())));
|
||||
// Use the `tokio` 0.2 `TypedExecutor` impl so we don't have to box the
|
||||
// future twice (once to spawn it using `Executor01::spawn` and a second
|
||||
// time to pin the compat future).
|
||||
self.inner.spawn(future);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Runtime {
|
||||
fn drop(&mut self) {
|
||||
if let Some(inner) = self.inner.take() {
|
||||
drop(inner);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
@@ -0,0 +1,133 @@
|
||||
use tokio_02::executor::{thread_pool::Spawner, Executor};
|
||||
use tokio_executor_01::{self as executor_01, Executor as Executor01};
|
||||
|
||||
use futures_01::future::Future as Future01;
|
||||
use futures_util::{compat::Future01CompatExt, future::FutureExt};
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
|
||||
/// Executes futures on the runtime
|
||||
///
|
||||
/// All futures spawned using this executor will be submitted to the associated
|
||||
/// Runtime's executor. This executor is usually a thread pool.
|
||||
///
|
||||
/// For more details, see the [module level](index.html) documentation.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TaskExecutor {
|
||||
pub(super) inner: super::CompatSpawner<Spawner>,
|
||||
}
|
||||
|
||||
impl TaskExecutor {
|
||||
/// Spawn a `futures` 0.1 future onto the Tokio runtime.
|
||||
///
|
||||
/// This spawns the given future onto the runtime's executor, usually a
|
||||
/// thread pool. The thread pool is then responsible for polling the future
|
||||
/// until it completes.
|
||||
///
|
||||
/// See [module level][mod] documentation for more details.
|
||||
///
|
||||
/// [mod]: index.html
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::Runtime;
|
||||
/// # fn dox() {
|
||||
/// // Create the runtime
|
||||
/// let rt = Runtime::new().unwrap();
|
||||
/// let executor = rt.executor();
|
||||
///
|
||||
/// // Spawn a `futures` 0.1 future onto the runtime
|
||||
/// executor.spawn(futures_01::future::lazy(|| {
|
||||
/// println!("now running on a worker thread");
|
||||
/// Ok(())
|
||||
/// }));
|
||||
/// # }
|
||||
/// ```
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if the spawn fails. Failure occurs if the executor
|
||||
/// is currently at capacity and is unable to spawn a new future.
|
||||
pub fn spawn<F>(&self, future: F)
|
||||
where
|
||||
F: Future01<Item = (), Error = ()> + Send + 'static,
|
||||
{
|
||||
self.spawn_std(Box::pin(future.compat().map(|_| ())));
|
||||
}
|
||||
|
||||
/// Spawn a `std::future` future onto the Tokio runtime.
|
||||
///
|
||||
/// This spawns the given future onto the runtime's executor, usually a
|
||||
/// thread pool. The thread pool is then responsible for polling the future
|
||||
/// until it completes.
|
||||
///
|
||||
/// See [module level][mod] documentation for more details.
|
||||
///
|
||||
/// [mod]: index.html
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
/// ```
|
||||
/// use tokio_compat::runtime::Runtime;
|
||||
///
|
||||
/// # fn dox() {
|
||||
/// // Create the runtime
|
||||
/// let rt = Runtime::new().unwrap();
|
||||
/// let executor = rt.executor();
|
||||
///
|
||||
/// // Spawn a `std::future` future onto the runtime
|
||||
/// executor.spawn_std(async {
|
||||
/// println!("now running on a worker thread");
|
||||
/// });
|
||||
/// # }
|
||||
/// ```
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if the spawn fails. Failure occurs if the executor
|
||||
/// is currently at capacity and is unable to spawn a new future.
|
||||
pub fn spawn_std<F>(&self, future: F)
|
||||
where
|
||||
F: Future<Output = ()> + Send + 'static,
|
||||
{
|
||||
let idle = self.inner.idle.reserve();
|
||||
self.inner.inner.spawn(idle.with(future));
|
||||
}
|
||||
}
|
||||
|
||||
impl Executor for TaskExecutor {
|
||||
fn spawn(
|
||||
&mut self,
|
||||
future: Pin<Box<dyn Future<Output = ()> + Send>>,
|
||||
) -> Result<(), tokio_02::executor::SpawnError> {
|
||||
Executor::spawn(&mut self.inner, future)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> tokio_02::executor::TypedExecutor<T> for TaskExecutor
|
||||
where
|
||||
T: Future<Output = ()> + Send + 'static,
|
||||
{
|
||||
fn spawn(&mut self, future: T) -> Result<(), tokio_02::executor::SpawnError> {
|
||||
Executor::spawn(&mut self.inner, Box::pin(future))
|
||||
}
|
||||
}
|
||||
|
||||
impl Executor01 for TaskExecutor {
|
||||
fn spawn(
|
||||
&mut self,
|
||||
future: Box<dyn Future01<Item = (), Error = ()> + Send>,
|
||||
) -> Result<(), executor_01::SpawnError> {
|
||||
Executor01::spawn(&mut self.inner, future)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> executor_01::TypedExecutor<T> for TaskExecutor
|
||||
where
|
||||
T: Future01<Item = (), Error = ()> + Send + 'static,
|
||||
{
|
||||
fn spawn(&mut self, future: T) -> Result<(), executor_01::SpawnError> {
|
||||
executor_01::TypedExecutor::spawn(&mut self.inner, future)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
use std::sync::{
|
||||
atomic::{AtomicBool, Ordering},
|
||||
Arc,
|
||||
};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use futures_01::future::Future as Future01;
|
||||
use futures_util::compat::Future01CompatExt;
|
||||
|
||||
#[test]
|
||||
fn can_run_01_futures() {
|
||||
let future_ran = Arc::new(AtomicBool::new(false));
|
||||
let ran = future_ran.clone();
|
||||
super::run(futures_01::future::lazy(move || {
|
||||
future_ran.store(true, Ordering::SeqCst);
|
||||
Ok(())
|
||||
}));
|
||||
assert!(ran.load(Ordering::SeqCst));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn can_spawn_01_futures() {
|
||||
let future_ran = Arc::new(AtomicBool::new(false));
|
||||
let ran = future_ran.clone();
|
||||
super::run(futures_01::future::lazy(move || {
|
||||
tokio_01::spawn(futures_01::future::lazy(move || {
|
||||
future_ran.store(true, Ordering::SeqCst);
|
||||
Ok(())
|
||||
}));
|
||||
Ok(())
|
||||
}));
|
||||
assert!(ran.load(Ordering::SeqCst));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn can_spawn_std_futures() {
|
||||
let future_ran = Arc::new(AtomicBool::new(false));
|
||||
let ran = future_ran.clone();
|
||||
super::run(futures_01::future::lazy(move || {
|
||||
tokio_02::spawn(async move {
|
||||
future_ran.store(true, Ordering::SeqCst);
|
||||
});
|
||||
Ok(())
|
||||
}));
|
||||
assert!(ran.load(Ordering::SeqCst));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tokio_01_timers_work() {
|
||||
let future1_ran = Arc::new(AtomicBool::new(false));
|
||||
let ran = future1_ran.clone();
|
||||
let future1 = futures_01::future::lazy(|| {
|
||||
let when = Instant::now() + Duration::from_millis(15);
|
||||
tokio_01::timer::Delay::new(when).map(move |_| when)
|
||||
})
|
||||
.map(move |when| {
|
||||
ran.store(true, Ordering::SeqCst);
|
||||
assert!(Instant::now() >= when);
|
||||
})
|
||||
.map_err(|_| panic!("timer should work"));
|
||||
|
||||
let future2_ran = Arc::new(AtomicBool::new(false));
|
||||
let ran = future2_ran.clone();
|
||||
let future2 = async move {
|
||||
let when = Instant::now() + Duration::from_millis(10);
|
||||
tokio_01::timer::Delay::new(when).compat().await.unwrap();
|
||||
ran.store(true, Ordering::SeqCst);
|
||||
assert!(Instant::now() >= when);
|
||||
};
|
||||
|
||||
super::run(futures_01::future::lazy(move || {
|
||||
tokio_02::spawn(future2);
|
||||
tokio_01::spawn(future1);
|
||||
Ok(())
|
||||
}));
|
||||
assert!(future1_ran.load(Ordering::SeqCst));
|
||||
assert!(future2_ran.load(Ordering::SeqCst));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_01_timer() {
|
||||
let rt = super::Runtime::new().unwrap();
|
||||
let when = Instant::now() + Duration::from_millis(10);
|
||||
rt.block_on(tokio_01::timer::Delay::new(when)).unwrap();
|
||||
assert!(Instant::now() >= when);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_std_01_timer() {
|
||||
let rt = super::Runtime::new().unwrap();
|
||||
let when = Instant::now() + Duration::from_millis(10);
|
||||
rt.block_on_std(async move {
|
||||
tokio_01::timer::Delay::new(when).compat().await.unwrap();
|
||||
});
|
||||
assert!(Instant::now() >= when);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_01_spawn() {
|
||||
let rt = super::Runtime::new().unwrap();
|
||||
// other tests assert that spawned 0.1 tasks actually *run*, all we care
|
||||
// is that we're able to spawn it successfully.
|
||||
rt.block_on(futures_01::future::lazy(|| {
|
||||
tokio_01::spawn(futures_01::future::lazy(|| Ok(())))
|
||||
}))
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_std_01_spawn() {
|
||||
let rt = super::Runtime::new().unwrap();
|
||||
// other tests assert that spawned 0.1 tasks actually *run*, all we care
|
||||
// is that we're able to spawn it successfully.
|
||||
rt.block_on_std(async { tokio_01::spawn(futures_01::future::lazy(|| Ok(()))) });
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
use super::super::{RESERVED_BITS, WIDTH};
|
||||
use super::ScheduledIo;
|
||||
use crate::sync::{
|
||||
atomic::{AtomicBool, Ordering},
|
||||
CausalCell,
|
||||
};
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct Slot {
|
||||
empty: AtomicBool,
|
||||
/// The offset of the next item on the free list.
|
||||
next: CausalCell<usize>,
|
||||
/// The data stored in the slot.
|
||||
item: ScheduledIo,
|
||||
}
|
||||
|
||||
#[repr(transparent)]
|
||||
#[derive(Copy, Clone, Debug, PartialEq, Eq, Ord, PartialOrd)]
|
||||
pub(crate) struct Generation {
|
||||
value: usize,
|
||||
}
|
||||
|
||||
impl Pack for Generation {
|
||||
/// Use all the remaining bits in the word for the generation counter, minus
|
||||
/// any bits reserved by the user.
|
||||
const LEN: usize = (WIDTH - RESERVED_BITS) - Self::SHIFT;
|
||||
|
||||
type Prev = Tid;
|
||||
|
||||
#[inline(always)]
|
||||
fn from_usize(u: usize) -> Self {
|
||||
debug_assert!(u <= Self::BITS);
|
||||
Self::new(u)
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
fn as_usize(&self) -> usize {
|
||||
self.value
|
||||
}
|
||||
}
|
||||
|
||||
impl Generation {
|
||||
fn new(value: usize) -> Self {
|
||||
Self { value }
|
||||
}
|
||||
}
|
||||
|
||||
impl Slot {
|
||||
pub(super) fn new(next: usize) -> Self {
|
||||
Self {
|
||||
empty: AtomicBool::new(true),
|
||||
item: ScheduledIo::default(),
|
||||
next: CausalCell::new(next),
|
||||
}
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
pub(super) fn get(&self, gen: Generation) -> Option<&T> {
|
||||
let current = self.gen.load(Ordering::Acquire);
|
||||
test_println!("-> get {:?}; current={:?}", gen, current);
|
||||
|
||||
// Is the index's generation the same as the current generation? If not,
|
||||
// the item that index referred to was removed, so return `None`.
|
||||
if gen.value != current {
|
||||
return None;
|
||||
}
|
||||
|
||||
Some(&self.item)
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub(super) fn insert(&self) -> Generation {
|
||||
Generation::from_usize(self.gen.load(Ordering::Acquire))
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
pub(super) fn next(&self) -> usize {
|
||||
self.next.with(|next| unsafe { *next })
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub(super) fn reset(&self, gen: Generation) -> bool {
|
||||
let next = (gen.value + 1) % Generation::BITS;
|
||||
let actual = self
|
||||
.generation
|
||||
.compare_and_swap(gen.value, next, Ordering::AcqRel);
|
||||
test_println!("-> remove {:?}; next={:?}; actual={:?}", gen, next, actual);
|
||||
if actual != gen {
|
||||
return false;
|
||||
};
|
||||
|
||||
self.item.reset();
|
||||
true
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
pub(super) fn set_next(&self, next: usize) {
|
||||
self.next.with_mut(|n| unsafe {
|
||||
(*n) = next;
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user