mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-09-09 00:00:08 +02:00
executor: add TypedExecutor (#993)
Adds a `TypedExecutor` trait that describes how to spawn futures of a specific type. This is useful for implementing functions that are generic over an executor and wish to support both `Send` and `!Send` cases.
This commit is contained in:
@@ -21,3 +21,6 @@ members = [
|
|||||||
"tokio-udp",
|
"tokio-udp",
|
||||||
"tokio-uds",
|
"tokio-uds",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[patch.crates-io]
|
||||||
|
tokio-executor = { path = "tokio-executor" }
|
||||||
|
|||||||
@@ -0,0 +1,16 @@
|
|||||||
|
steps:
|
||||||
|
- script: |
|
||||||
|
set -e
|
||||||
|
|
||||||
|
# Remove any existing patch statements
|
||||||
|
mv Cargo.toml Cargo.toml.bck
|
||||||
|
sed -n '/\[patch.crates-io\]/q;p' Cargo.toml.bck > Cargo.toml
|
||||||
|
|
||||||
|
# Patch all crates
|
||||||
|
cat ci/patch.toml >> Cargo.toml
|
||||||
|
|
||||||
|
# Print `Cargo.toml` for debugging
|
||||||
|
echo "~~~~ Cargo.toml ~~~~"
|
||||||
|
cat Cargo.toml
|
||||||
|
echo "~~~~~~~~~~~~~~~~~~~~"
|
||||||
|
displayName: Patch Cargo.toml
|
||||||
@@ -27,13 +27,7 @@ jobs:
|
|||||||
displayName: cargo test -p ${{ crate }}
|
displayName: cargo test -p ${{ crate }}
|
||||||
workingDirectory: $(Build.SourcesDirectory)/${{ crate }}
|
workingDirectory: $(Build.SourcesDirectory)/${{ crate }}
|
||||||
|
|
||||||
- script: |
|
- template: azure-patch-crates.yml
|
||||||
set -e
|
|
||||||
cat ci/patch.toml >> Cargo.toml
|
|
||||||
echo "~~~~ Cargo.toml ~~~~"
|
|
||||||
cat Cargo.toml
|
|
||||||
echo "~~~~~~~~~~~~~~~~~~~~"
|
|
||||||
displayName: Patch Cargo.toml
|
|
||||||
|
|
||||||
- ${{ each crate in parameters.crates }}:
|
- ${{ each crate in parameters.crates }}:
|
||||||
- script: cargo test
|
- script: cargo test
|
||||||
|
|||||||
+1
-5
@@ -14,14 +14,10 @@ jobs:
|
|||||||
parameters:
|
parameters:
|
||||||
rust_version: nightly-2018-11-18
|
rust_version: nightly-2018-11-18
|
||||||
|
|
||||||
|
- template: azure-patch-crates.yml
|
||||||
- script: |
|
- script: |
|
||||||
set -e
|
set -e
|
||||||
|
|
||||||
cat ci/patch.toml >> Cargo.toml
|
|
||||||
echo "~~~~ Cargo.toml ~~~~"
|
|
||||||
cat Cargo.toml
|
|
||||||
echo "~~~~~~~~~~~~~~~~~~~~"
|
|
||||||
|
|
||||||
# Make sure the benchmarks compile
|
# Make sure the benchmarks compile
|
||||||
export ASAN_OPTIONS="detect_odr_violation=0 detect_leaks=0"
|
export ASAN_OPTIONS="detect_odr_violation=0 detect_leaks=0"
|
||||||
export TSAN_OPTIONS="suppressions=`pwd`/ci/tsan"
|
export TSAN_OPTIONS="suppressions=`pwd`/ci/tsan"
|
||||||
|
|||||||
@@ -21,5 +21,5 @@ keywords = ["futures", "tokio"]
|
|||||||
categories = ["concurrency", "asynchronous"]
|
categories = ["concurrency", "asynchronous"]
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
tokio-executor = "0.1.5"
|
tokio-executor = { version = "0.1.5", path = "../tokio-executor" }
|
||||||
futures = "0.1.19"
|
futures = "0.1.19"
|
||||||
|
|||||||
@@ -431,6 +431,16 @@ impl tokio_executor::Executor for CurrentThread {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<T> tokio_executor::TypedExecutor<T> for CurrentThread
|
||||||
|
where
|
||||||
|
T: Future<Item = (), Error = ()> + 'static,
|
||||||
|
{
|
||||||
|
fn spawn(&mut self, future: T) -> Result<(), SpawnError> {
|
||||||
|
self.borrow().spawn_local(Box::new(future), false);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl<P: Park> fmt::Debug for CurrentThread<P> {
|
impl<P: Park> fmt::Debug for CurrentThread<P> {
|
||||||
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
|
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
|
||||||
fmt.debug_struct("CurrentThread")
|
fmt.debug_struct("CurrentThread")
|
||||||
@@ -742,6 +752,15 @@ impl tokio_executor::Executor for TaskExecutor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<F> tokio_executor::TypedExecutor<F> for TaskExecutor
|
||||||
|
where
|
||||||
|
F: Future<Item = (), Error = ()> + 'static,
|
||||||
|
{
|
||||||
|
fn spawn(&mut self, future: F) -> Result<(), SpawnError> {
|
||||||
|
self.spawn_local(Box::new(future))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl<F> Executor<F> for TaskExecutor
|
impl<F> Executor<F> for TaskExecutor
|
||||||
where
|
where
|
||||||
F: Future<Item = (), Error = ()> + 'static,
|
F: Future<Item = (), Error = ()> + 'static,
|
||||||
|
|||||||
@@ -23,3 +23,6 @@ categories = ["concurrency", "asynchronous"]
|
|||||||
[dependencies]
|
[dependencies]
|
||||||
crossbeam-utils = "0.6.2"
|
crossbeam-utils = "0.6.2"
|
||||||
futures = "0.1.19"
|
futures = "0.1.19"
|
||||||
|
|
||||||
|
[dev-dependencies]
|
||||||
|
tokio = { version = "0.1.17", path = "../tokio" }
|
||||||
|
|||||||
@@ -0,0 +1,50 @@
|
|||||||
|
use std::error::Error;
|
||||||
|
use std::fmt;
|
||||||
|
|
||||||
|
/// Errors returned by `Executor::spawn`.
|
||||||
|
///
|
||||||
|
/// Spawn errors should represent relatively rare scenarios. Currently, the two
|
||||||
|
/// scenarios represented by `SpawnError` are:
|
||||||
|
///
|
||||||
|
/// * An executor being at capacity or full. As such, the executor is not able
|
||||||
|
/// to accept a new future. This error state is expected to be transient.
|
||||||
|
/// * An executor has been shutdown and can no longer accept new futures. This
|
||||||
|
/// error state is expected to be permanent.
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub struct SpawnError {
|
||||||
|
is_shutdown: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SpawnError {
|
||||||
|
/// Return a new `SpawnError` reflecting a shutdown executor failure.
|
||||||
|
pub fn shutdown() -> Self {
|
||||||
|
SpawnError { is_shutdown: true }
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Return a new `SpawnError` reflecting an executor at capacity failure.
|
||||||
|
pub fn at_capacity() -> Self {
|
||||||
|
SpawnError { is_shutdown: false }
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns `true` if the error reflects a shutdown executor failure.
|
||||||
|
pub fn is_shutdown(&self) -> bool {
|
||||||
|
self.is_shutdown
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns `true` if the error reflects an executor at capacity failure.
|
||||||
|
pub fn is_at_capacity(&self) -> bool {
|
||||||
|
!self.is_shutdown
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl fmt::Display for SpawnError {
|
||||||
|
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
|
||||||
|
write!(fmt, "{}", self.description())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Error for SpawnError {
|
||||||
|
fn description(&self) -> &str {
|
||||||
|
"attempted to spawn task while the executor is at capacity or shut down"
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,151 @@
|
|||||||
|
use futures::Future;
|
||||||
|
use SpawnError;
|
||||||
|
|
||||||
|
/// A value that executes futures.
|
||||||
|
///
|
||||||
|
/// The [`spawn`] function is used to submit a future to an executor. Once
|
||||||
|
/// submitted, the executor takes ownership of the future and becomes
|
||||||
|
/// responsible for driving the future to completion.
|
||||||
|
///
|
||||||
|
/// The strategy employed by the executor to handle the future is less defined
|
||||||
|
/// and is left up to the `Executor` implementation. The `Executor` instance is
|
||||||
|
/// expected to call [`poll`] on the future once it has been notified, however
|
||||||
|
/// the "when" and "how" can vary greatly.
|
||||||
|
///
|
||||||
|
/// For example, the executor might be a thread pool, in which case a set of
|
||||||
|
/// threads have already been spawned up and the future is inserted into a
|
||||||
|
/// queue. A thread will acquire the future and poll it.
|
||||||
|
///
|
||||||
|
/// The `Executor` trait is only for futures that **are** `Send`. These are most
|
||||||
|
/// common. There currently is no trait that describes executors that operate
|
||||||
|
/// entirely on the current thread (i.e., are able to spawn futures that are not
|
||||||
|
/// `Send`). Note that single threaded executors can still implement `Executor`,
|
||||||
|
/// but only futures that are `Send` can be spawned via the trait.
|
||||||
|
///
|
||||||
|
/// This trait is primarily intended to implemented by executors and used to
|
||||||
|
/// back `tokio::spawn`. Libraries and applications **may** use this trait to
|
||||||
|
/// bound generics, but doing so will limit usage to futures that implement
|
||||||
|
/// `Send`. Instead, libraries and applications are recommended to use
|
||||||
|
/// [`TypedExecutor`] as a bound.
|
||||||
|
///
|
||||||
|
/// # Errors
|
||||||
|
///
|
||||||
|
/// The [`spawn`] function returns `Result` with an error type of `SpawnError`.
|
||||||
|
/// This error type represents the reason that the executor was unable to spawn
|
||||||
|
/// the future. The two current represented scenarios are:
|
||||||
|
///
|
||||||
|
/// * An executor being at capacity or full. As such, the executor is not able
|
||||||
|
/// to accept a new future. This error state is expected to be transient.
|
||||||
|
/// * An executor has been shutdown and can no longer accept new futures. This
|
||||||
|
/// error state is expected to be permanent.
|
||||||
|
///
|
||||||
|
/// If a caller encounters an at capacity error, the caller should try to shed
|
||||||
|
/// load. This can be as simple as dropping the future that was spawned.
|
||||||
|
///
|
||||||
|
/// If the caller encounters a shutdown error, the caller should attempt to
|
||||||
|
/// gracefully shutdown.
|
||||||
|
///
|
||||||
|
/// # Examples
|
||||||
|
///
|
||||||
|
/// ```rust
|
||||||
|
/// # extern crate futures;
|
||||||
|
/// # extern crate tokio_executor;
|
||||||
|
/// # use tokio_executor::Executor;
|
||||||
|
/// # fn docs(my_executor: &mut Executor) {
|
||||||
|
/// use futures::future::lazy;
|
||||||
|
/// my_executor.spawn(Box::new(lazy(|| {
|
||||||
|
/// println!("running on the executor");
|
||||||
|
/// Ok(())
|
||||||
|
/// }))).unwrap();
|
||||||
|
/// # }
|
||||||
|
/// # fn main() {}
|
||||||
|
/// ```
|
||||||
|
///
|
||||||
|
/// [`spawn`]: #tymethod.spawn
|
||||||
|
/// [`poll`]: https://docs.rs/futures/0.1/futures/future/trait.Future.html#tymethod.poll
|
||||||
|
/// [`TypedExecutor`]: ../trait.TypedExecutor.html
|
||||||
|
pub trait Executor {
|
||||||
|
/// Spawns a future object to run on this executor.
|
||||||
|
///
|
||||||
|
/// `future` is passed to the executor, which will begin running it. The
|
||||||
|
/// future may run on the current thread or another thread at the discretion
|
||||||
|
/// of the `Executor` implementation.
|
||||||
|
///
|
||||||
|
/// # Panics
|
||||||
|
///
|
||||||
|
/// Implementations are encouraged to avoid panics. However, panics are
|
||||||
|
/// permitted and the caller should check the implementation specific
|
||||||
|
/// documentation for more details on possible panics.
|
||||||
|
///
|
||||||
|
/// # Examples
|
||||||
|
///
|
||||||
|
/// ```rust
|
||||||
|
/// # extern crate futures;
|
||||||
|
/// # extern crate tokio_executor;
|
||||||
|
/// # use tokio_executor::Executor;
|
||||||
|
/// # fn docs(my_executor: &mut Executor) {
|
||||||
|
/// use futures::future::lazy;
|
||||||
|
/// my_executor.spawn(Box::new(lazy(|| {
|
||||||
|
/// println!("running on the executor");
|
||||||
|
/// Ok(())
|
||||||
|
/// }))).unwrap();
|
||||||
|
/// # }
|
||||||
|
/// # fn main() {}
|
||||||
|
/// ```
|
||||||
|
fn spawn(
|
||||||
|
&mut self,
|
||||||
|
future: Box<Future<Item = (), Error = ()> + Send>,
|
||||||
|
) -> Result<(), SpawnError>;
|
||||||
|
|
||||||
|
/// 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.
|
||||||
|
///
|
||||||
|
/// # Panics
|
||||||
|
///
|
||||||
|
/// This function must not panic. Implementers must ensure that panics do
|
||||||
|
/// not happen.
|
||||||
|
///
|
||||||
|
/// # Examples
|
||||||
|
///
|
||||||
|
/// ```rust
|
||||||
|
/// # extern crate futures;
|
||||||
|
/// # extern crate tokio_executor;
|
||||||
|
/// # use tokio_executor::Executor;
|
||||||
|
/// # fn docs(my_executor: &mut Executor) {
|
||||||
|
/// use futures::future::lazy;
|
||||||
|
///
|
||||||
|
/// if my_executor.status().is_ok() {
|
||||||
|
/// my_executor.spawn(Box::new(lazy(|| {
|
||||||
|
/// println!("running on the executor");
|
||||||
|
/// Ok(())
|
||||||
|
/// }))).unwrap();
|
||||||
|
/// } else {
|
||||||
|
/// println!("the executor is not in a good state");
|
||||||
|
/// }
|
||||||
|
/// # }
|
||||||
|
/// # fn main() {}
|
||||||
|
/// ```
|
||||||
|
fn status(&self) -> Result<(), SpawnError> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<E: Executor + ?Sized> Executor for Box<E> {
|
||||||
|
fn spawn(
|
||||||
|
&mut self,
|
||||||
|
future: Box<Future<Item = (), Error = ()> + Send>,
|
||||||
|
) -> Result<(), SpawnError> {
|
||||||
|
(**self).spawn(future)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn status(&self) -> Result<(), SpawnError> {
|
||||||
|
(**self).status()
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -84,6 +84,19 @@ impl super::Executor for DefaultExecutor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<T> super::TypedExecutor<T> for DefaultExecutor
|
||||||
|
where
|
||||||
|
T: Future<Item = (), Error = ()> + Send + 'static,
|
||||||
|
{
|
||||||
|
fn spawn(&mut self, future: T) -> Result<(), SpawnError> {
|
||||||
|
super::Executor::spawn(self, Box::new(future))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn status(&self) -> Result<(), SpawnError> {
|
||||||
|
super::Executor::status(self)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl<T> future::Executor<T> for DefaultExecutor
|
impl<T> future::Executor<T> for DefaultExecutor
|
||||||
where
|
where
|
||||||
T: Future<Item = (), Error = ()> + Send + 'static,
|
T: Future<Item = (), Error = ()> + Send + 'static,
|
||||||
|
|||||||
+24
-197
@@ -17,8 +17,12 @@
|
|||||||
//! This crate provides traits and utilities that are necessary for building an
|
//! This crate provides traits and utilities that are necessary for building an
|
||||||
//! executor, including:
|
//! executor, including:
|
||||||
//!
|
//!
|
||||||
//! * The [`Executor`] trait describes the API for spawning a future onto an
|
//! * The [`Executor`] trait spawns future object onto an executor.
|
||||||
//! executor.
|
//!
|
||||||
|
//! * The [`TypedExecutor`] trait spawns futures of a specific type onto an
|
||||||
|
//! executor. This is used to be generic over executors that spawn futures
|
||||||
|
//! that are either `Send` or `!Send` or implement executors that apply to
|
||||||
|
//! specific futures.
|
||||||
//!
|
//!
|
||||||
//! * [`enter`] marks that the current thread is entering an execution
|
//! * [`enter`] marks that the current thread is entering an execution
|
||||||
//! context. This prevents a second executor from accidentally starting from
|
//! context. This prevents a second executor from accidentally starting from
|
||||||
@@ -29,7 +33,19 @@
|
|||||||
//!
|
//!
|
||||||
//! * [`Park`] abstracts over blocking and unblocking the current thread.
|
//! * [`Park`] abstracts over blocking and unblocking the current thread.
|
||||||
//!
|
//!
|
||||||
|
//! # Implementing an executor
|
||||||
|
//!
|
||||||
|
//! Executors should always implement `TypedExecutor`. This usually is the bound
|
||||||
|
//! that applications and libraries will use when generic over an executor. See
|
||||||
|
//! the [trait documentation][`TypedExecutor`] for more details.
|
||||||
|
//!
|
||||||
|
//! If the executor is able to spawn all futures that are `Send`, then the
|
||||||
|
//! executor should also implement the `Executor` trait. This trait is rarely
|
||||||
|
//! used directly by applications and libraries. Instead, `tokio::spawn` is
|
||||||
|
//! configured to dispatch to type that implements `Executor`.
|
||||||
|
//!
|
||||||
//! [`Executor`]: trait.Executor.html
|
//! [`Executor`]: trait.Executor.html
|
||||||
|
//! [`TypedExecutor`]: trait.TypedExecutor.html
|
||||||
//! [`enter`]: fn.enter.html
|
//! [`enter`]: fn.enter.html
|
||||||
//! [`DefaultExecutor`]: struct.DefaultExecutor.html
|
//! [`DefaultExecutor`]: struct.DefaultExecutor.html
|
||||||
//! [`Park`]: park/index.html
|
//! [`Park`]: park/index.html
|
||||||
@@ -39,203 +55,14 @@ extern crate crossbeam_utils;
|
|||||||
extern crate futures;
|
extern crate futures;
|
||||||
|
|
||||||
mod enter;
|
mod enter;
|
||||||
|
mod error;
|
||||||
|
mod executor;
|
||||||
mod global;
|
mod global;
|
||||||
pub mod park;
|
pub mod park;
|
||||||
|
mod typed;
|
||||||
|
|
||||||
pub use enter::{enter, Enter, EnterError};
|
pub use enter::{enter, Enter, EnterError};
|
||||||
|
pub use error::SpawnError;
|
||||||
|
pub use executor::Executor;
|
||||||
pub use global::{spawn, with_default, DefaultExecutor};
|
pub use global::{spawn, with_default, DefaultExecutor};
|
||||||
|
pub use typed::TypedExecutor;
|
||||||
use futures::Future;
|
|
||||||
|
|
||||||
use std::error::Error;
|
|
||||||
use std::fmt;
|
|
||||||
|
|
||||||
/// A value that executes futures.
|
|
||||||
///
|
|
||||||
/// The [`spawn`] function is used to submit a future to an executor. Once
|
|
||||||
/// submitted, the executor takes ownership of the future and becomes
|
|
||||||
/// responsible for driving the future to completion.
|
|
||||||
///
|
|
||||||
/// The strategy employed by the executor to handle the future is less defined
|
|
||||||
/// and is left up to the `Executor` implementation. The `Executor` instance is
|
|
||||||
/// expected to call [`poll`] on the future once it has been notified, however
|
|
||||||
/// the "when" and "how" can vary greatly.
|
|
||||||
///
|
|
||||||
/// For example, the executor might be a thread pool, in which case a set of
|
|
||||||
/// threads have already been spawned up and the future is inserted into a
|
|
||||||
/// queue. A thread will acquire the future and poll it.
|
|
||||||
///
|
|
||||||
/// The `Executor` trait is only for futures that **are** `Send`. These are most
|
|
||||||
/// common. There currently is no trait that describes executors that operate
|
|
||||||
/// entirely on the current thread (i.e., are able to spawn futures that are not
|
|
||||||
/// `Send`). Note that single threaded executors can still implement `Executor`,
|
|
||||||
/// but only futures that are `Send` can be spawned via the trait.
|
|
||||||
///
|
|
||||||
/// # Errors
|
|
||||||
///
|
|
||||||
/// The [`spawn`] function returns `Result` with an error type of `SpawnError`.
|
|
||||||
/// This error type represents the reason that the executor was unable to spawn
|
|
||||||
/// the future. The two current represented scenarios are:
|
|
||||||
///
|
|
||||||
/// * An executor being at capacity or full. As such, the executor is not able
|
|
||||||
/// to accept a new future. This error state is expected to be transient.
|
|
||||||
/// * An executor has been shutdown and can no longer accept new futures. This
|
|
||||||
/// error state is expected to be permanent.
|
|
||||||
///
|
|
||||||
/// If a caller encounters an at capacity error, the caller should try to shed
|
|
||||||
/// load. This can be as simple as dropping the future that was spawned.
|
|
||||||
///
|
|
||||||
/// If the caller encounters a shutdown error, the caller should attempt to
|
|
||||||
/// gracefully shutdown.
|
|
||||||
///
|
|
||||||
/// # Examples
|
|
||||||
///
|
|
||||||
/// ```rust
|
|
||||||
/// # extern crate futures;
|
|
||||||
/// # extern crate tokio_executor;
|
|
||||||
/// # use tokio_executor::Executor;
|
|
||||||
/// # fn docs(my_executor: &mut Executor) {
|
|
||||||
/// use futures::future::lazy;
|
|
||||||
/// my_executor.spawn(Box::new(lazy(|| {
|
|
||||||
/// println!("running on the executor");
|
|
||||||
/// Ok(())
|
|
||||||
/// }))).unwrap();
|
|
||||||
/// # }
|
|
||||||
/// # fn main() {}
|
|
||||||
/// ```
|
|
||||||
///
|
|
||||||
/// [`spawn`]: #tymethod.spawn
|
|
||||||
/// [`poll`]: https://docs.rs/futures/0.1/futures/future/trait.Future.html#tymethod.poll
|
|
||||||
pub trait Executor {
|
|
||||||
/// Spawns a future object to run on this executor.
|
|
||||||
///
|
|
||||||
/// `future` is passed to the executor, which will begin running it. The
|
|
||||||
/// future may run on the current thread or another thread at the discretion
|
|
||||||
/// of the `Executor` implementation.
|
|
||||||
///
|
|
||||||
/// # Panics
|
|
||||||
///
|
|
||||||
/// Implementers are encouraged to avoid panics. However, a panic is
|
|
||||||
/// permitted and the caller should check the implementation specific
|
|
||||||
/// documentation for more details on possible panics.
|
|
||||||
///
|
|
||||||
/// # Examples
|
|
||||||
///
|
|
||||||
/// ```rust
|
|
||||||
/// # extern crate futures;
|
|
||||||
/// # extern crate tokio_executor;
|
|
||||||
/// # use tokio_executor::Executor;
|
|
||||||
/// # fn docs(my_executor: &mut Executor) {
|
|
||||||
/// use futures::future::lazy;
|
|
||||||
/// my_executor.spawn(Box::new(lazy(|| {
|
|
||||||
/// println!("running on the executor");
|
|
||||||
/// Ok(())
|
|
||||||
/// }))).unwrap();
|
|
||||||
/// # }
|
|
||||||
/// # fn main() {}
|
|
||||||
/// ```
|
|
||||||
fn spawn(
|
|
||||||
&mut self,
|
|
||||||
future: Box<Future<Item = (), Error = ()> + Send>,
|
|
||||||
) -> Result<(), SpawnError>;
|
|
||||||
|
|
||||||
/// 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.
|
|
||||||
///
|
|
||||||
/// # Panics
|
|
||||||
///
|
|
||||||
/// This function must not panic. Implementers must ensure that panics do
|
|
||||||
/// not happen.
|
|
||||||
///
|
|
||||||
/// # Examples
|
|
||||||
///
|
|
||||||
/// ```rust
|
|
||||||
/// # extern crate futures;
|
|
||||||
/// # extern crate tokio_executor;
|
|
||||||
/// # use tokio_executor::Executor;
|
|
||||||
/// # fn docs(my_executor: &mut Executor) {
|
|
||||||
/// use futures::future::lazy;
|
|
||||||
///
|
|
||||||
/// if my_executor.status().is_ok() {
|
|
||||||
/// my_executor.spawn(Box::new(lazy(|| {
|
|
||||||
/// println!("running on the executor");
|
|
||||||
/// Ok(())
|
|
||||||
/// }))).unwrap();
|
|
||||||
/// } else {
|
|
||||||
/// println!("the executor is not in a good state");
|
|
||||||
/// }
|
|
||||||
/// # }
|
|
||||||
/// # fn main() {}
|
|
||||||
/// ```
|
|
||||||
fn status(&self) -> Result<(), SpawnError> {
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<E: Executor + ?Sized> Executor for Box<E> {
|
|
||||||
fn spawn(
|
|
||||||
&mut self,
|
|
||||||
future: Box<Future<Item = (), Error = ()> + Send>,
|
|
||||||
) -> Result<(), SpawnError> {
|
|
||||||
(**self).spawn(future)
|
|
||||||
}
|
|
||||||
|
|
||||||
fn status(&self) -> Result<(), SpawnError> {
|
|
||||||
(**self).status()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Errors returned by `Executor::spawn`.
|
|
||||||
///
|
|
||||||
/// Spawn errors should represent relatively rare scenarios. Currently, the two
|
|
||||||
/// scenarios represented by `SpawnError` are:
|
|
||||||
///
|
|
||||||
/// * An executor being at capacity or full. As such, the executor is not able
|
|
||||||
/// to accept a new future. This error state is expected to be transient.
|
|
||||||
/// * An executor has been shutdown and can no longer accept new futures. This
|
|
||||||
/// error state is expected to be permanent.
|
|
||||||
#[derive(Debug)]
|
|
||||||
pub struct SpawnError {
|
|
||||||
is_shutdown: bool,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl SpawnError {
|
|
||||||
/// Return a new `SpawnError` reflecting a shutdown executor failure.
|
|
||||||
pub fn shutdown() -> Self {
|
|
||||||
SpawnError { is_shutdown: true }
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Return a new `SpawnError` reflecting an executor at capacity failure.
|
|
||||||
pub fn at_capacity() -> Self {
|
|
||||||
SpawnError { is_shutdown: false }
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Returns `true` if the error reflects a shutdown executor failure.
|
|
||||||
pub fn is_shutdown(&self) -> bool {
|
|
||||||
self.is_shutdown
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Returns `true` if the error reflects an executor at capacity failure.
|
|
||||||
pub fn is_at_capacity(&self) -> bool {
|
|
||||||
!self.is_shutdown
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl fmt::Display for SpawnError {
|
|
||||||
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
|
|
||||||
write!(fmt, "{}", self.description())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl Error for SpawnError {
|
|
||||||
fn description(&self) -> &str {
|
|
||||||
"attempted to spawn task while the executor is at capacity or shut down"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -0,0 +1,181 @@
|
|||||||
|
use SpawnError;
|
||||||
|
|
||||||
|
/// A value that spawns futures of a specific type.
|
||||||
|
///
|
||||||
|
/// The trait is generic over `T`: the type of future that can be spawened. This
|
||||||
|
/// is useful for implementing an executor that is only able to spawn a specific
|
||||||
|
/// type of future.
|
||||||
|
///
|
||||||
|
/// The [`spawn`] function is used to submit the future to the executor. Once
|
||||||
|
/// submitted, the executor takes ownership of the future and becomes
|
||||||
|
/// responsible for driving the future to completion.
|
||||||
|
///
|
||||||
|
/// This trait is useful as a bound for applications and libraries in order to
|
||||||
|
/// be generic over futures that are `Send` vs. `!Send`.
|
||||||
|
///
|
||||||
|
/// # Examples
|
||||||
|
///
|
||||||
|
/// Consider a function that provides an API for draining a `Stream` in the
|
||||||
|
/// background. To do this, a task must be spawned to perform the draining. As
|
||||||
|
/// such, the function takes a stream and an executor on which the background
|
||||||
|
/// task is spawned.
|
||||||
|
///
|
||||||
|
/// ```rust
|
||||||
|
/// #[macro_use]
|
||||||
|
/// extern crate futures;
|
||||||
|
/// extern crate tokio;
|
||||||
|
///
|
||||||
|
/// use futures::{Future, Stream, Poll};
|
||||||
|
/// use tokio::executor::TypedExecutor;
|
||||||
|
/// use tokio::sync::oneshot;
|
||||||
|
///
|
||||||
|
/// pub fn drain<T, E>(stream: T, executor: &mut E)
|
||||||
|
/// -> impl Future<Item = (), Error = ()>
|
||||||
|
/// where
|
||||||
|
/// T: Stream,
|
||||||
|
/// E: TypedExecutor<Drain<T>>
|
||||||
|
/// {
|
||||||
|
/// let (tx, rx) = oneshot::channel();
|
||||||
|
///
|
||||||
|
/// executor.spawn(Drain {
|
||||||
|
/// stream,
|
||||||
|
/// tx: Some(tx),
|
||||||
|
/// }).unwrap();
|
||||||
|
///
|
||||||
|
/// rx.map_err(|_| ())
|
||||||
|
/// }
|
||||||
|
///
|
||||||
|
/// // The background task
|
||||||
|
/// pub struct Drain<T: Stream> {
|
||||||
|
/// stream: T,
|
||||||
|
/// tx: Option<oneshot::Sender<()>>,
|
||||||
|
/// }
|
||||||
|
///
|
||||||
|
/// impl<T: Stream> Future for Drain<T> {
|
||||||
|
/// type Item = ();
|
||||||
|
/// type Error = ();
|
||||||
|
///
|
||||||
|
/// fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||||
|
/// loop {
|
||||||
|
/// let item = try_ready!(
|
||||||
|
/// self.stream.poll()
|
||||||
|
/// .map_err(|_| ())
|
||||||
|
/// );
|
||||||
|
///
|
||||||
|
/// if item.is_none() { break; }
|
||||||
|
/// }
|
||||||
|
///
|
||||||
|
/// self.tx.take().unwrap().send(()).map_err(|_| ());
|
||||||
|
/// Ok(().into())
|
||||||
|
/// }
|
||||||
|
/// }
|
||||||
|
/// # pub fn main() {}
|
||||||
|
/// ```
|
||||||
|
///
|
||||||
|
/// By doing this, the `drain` fn can accept a stream that is `!Send` as long as
|
||||||
|
/// the supplied executor is able to spawn `!Send` types.
|
||||||
|
pub trait TypedExecutor<T> {
|
||||||
|
/// Spawns a future to run on this executor.
|
||||||
|
///
|
||||||
|
/// `future` is passed to the executor, which will begin running it. The
|
||||||
|
/// executor takes ownership of the future and becomes responsible for
|
||||||
|
/// driving the future to completion.
|
||||||
|
///
|
||||||
|
/// # Panics
|
||||||
|
///
|
||||||
|
/// Implementations are encouraged to avoid panics. However, panics are
|
||||||
|
/// permitted and the caller should check the implementation specific
|
||||||
|
/// documentation for more details on possible panics.
|
||||||
|
///
|
||||||
|
/// # Examples
|
||||||
|
///
|
||||||
|
/// ```rust
|
||||||
|
/// # extern crate futures;
|
||||||
|
/// # extern crate tokio_executor;
|
||||||
|
/// # use tokio_executor::TypedExecutor;
|
||||||
|
/// # use futures::{Future, Poll};
|
||||||
|
/// fn example<T>(my_executor: &mut T)
|
||||||
|
/// where
|
||||||
|
/// T: TypedExecutor<MyFuture>,
|
||||||
|
/// {
|
||||||
|
/// my_executor.spawn(MyFuture).unwrap();
|
||||||
|
/// }
|
||||||
|
///
|
||||||
|
/// struct MyFuture;
|
||||||
|
///
|
||||||
|
/// impl Future for MyFuture {
|
||||||
|
/// type Item = ();
|
||||||
|
/// type Error = ();
|
||||||
|
///
|
||||||
|
/// fn poll(&mut self) -> Poll<(), ()> {
|
||||||
|
/// println!("running on the executor");
|
||||||
|
/// Ok(().into())
|
||||||
|
/// }
|
||||||
|
/// }
|
||||||
|
/// # fn main() {}
|
||||||
|
/// ```
|
||||||
|
fn spawn(&mut self, future: T) -> Result<(), SpawnError>;
|
||||||
|
|
||||||
|
/// 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.
|
||||||
|
///
|
||||||
|
/// # Panics
|
||||||
|
///
|
||||||
|
/// This function must not panic. Implementers must ensure that panics do
|
||||||
|
/// not happen.
|
||||||
|
///
|
||||||
|
/// # Examples
|
||||||
|
///
|
||||||
|
/// ```rust
|
||||||
|
/// # extern crate futures;
|
||||||
|
/// # extern crate tokio_executor;
|
||||||
|
/// # use tokio_executor::TypedExecutor;
|
||||||
|
/// # use futures::{Future, Poll};
|
||||||
|
/// fn example<T>(my_executor: &mut T)
|
||||||
|
/// where
|
||||||
|
/// T: TypedExecutor<MyFuture>,
|
||||||
|
/// {
|
||||||
|
/// if my_executor.status().is_ok() {
|
||||||
|
/// my_executor.spawn(MyFuture).unwrap();
|
||||||
|
/// } else {
|
||||||
|
/// println!("the executor is not in a good state");
|
||||||
|
/// }
|
||||||
|
/// }
|
||||||
|
///
|
||||||
|
/// struct MyFuture;
|
||||||
|
///
|
||||||
|
/// impl Future for MyFuture {
|
||||||
|
/// type Item = ();
|
||||||
|
/// type Error = ();
|
||||||
|
///
|
||||||
|
/// fn poll(&mut self) -> Poll<(), ()> {
|
||||||
|
/// println!("running on the executor");
|
||||||
|
/// Ok(().into())
|
||||||
|
/// }
|
||||||
|
/// }
|
||||||
|
/// # fn main() {}
|
||||||
|
/// ```
|
||||||
|
fn status(&self) -> Result<(), SpawnError> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<E, T> TypedExecutor<T> for Box<E>
|
||||||
|
where
|
||||||
|
E: TypedExecutor<T>,
|
||||||
|
{
|
||||||
|
fn spawn(&mut self, future: T) -> Result<(), SpawnError> {
|
||||||
|
(**self).spawn(future)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn status(&self) -> Result<(), SpawnError> {
|
||||||
|
(**self).status()
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -2,10 +2,11 @@ extern crate futures;
|
|||||||
extern crate tokio_executor;
|
extern crate tokio_executor;
|
||||||
|
|
||||||
use futures::{future::lazy, Future};
|
use futures::{future::lazy, Future};
|
||||||
use tokio_executor::*;
|
use tokio_executor::DefaultExecutor;
|
||||||
|
|
||||||
mod out_of_executor_context {
|
mod out_of_executor_context {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use tokio_executor::Executor;
|
||||||
|
|
||||||
fn test<F, E>(spawn: F)
|
fn test<F, E>(spawn: F)
|
||||||
where
|
where
|
||||||
|
|||||||
@@ -20,7 +20,7 @@ keywords = ["futures", "tokio"]
|
|||||||
categories = ["concurrency", "asynchronous"]
|
categories = ["concurrency", "asynchronous"]
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
tokio-executor = "0.1.2"
|
tokio-executor = { version = "0.1.5", path = "../tokio-executor" }
|
||||||
futures = "0.1.19"
|
futures = "0.1.19"
|
||||||
crossbeam-deque = "0.7.0"
|
crossbeam-deque = "0.7.0"
|
||||||
crossbeam-queue = "0.1.0"
|
crossbeam-queue = "0.1.0"
|
||||||
|
|||||||
@@ -176,6 +176,19 @@ impl<'a> tokio_executor::Executor for &'a Sender {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<T> tokio_executor::TypedExecutor<T> for Sender
|
||||||
|
where
|
||||||
|
T: Future<Item = (), Error = ()> + Send + 'static,
|
||||||
|
{
|
||||||
|
fn status(&self) -> Result<(), tokio_executor::SpawnError> {
|
||||||
|
tokio_executor::Executor::status(self)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn spawn(&mut self, future: T) -> Result<(), SpawnError> {
|
||||||
|
tokio_executor::Executor::spawn(self, Box::new(future))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl<T> future::Executor<T> for Sender
|
impl<T> future::Executor<T> for Sender
|
||||||
where
|
where
|
||||||
T: Future<Item = (), Error = ()> + Send + 'static,
|
T: Future<Item = (), Error = ()> + Send + 'static,
|
||||||
|
|||||||
+1
-1
@@ -71,7 +71,7 @@ tokio-codec = { version = "0.1.0", optional = true }
|
|||||||
tokio-current-thread = { version = "0.1.3", optional = true }
|
tokio-current-thread = { version = "0.1.3", optional = true }
|
||||||
tokio-fs = { version = "0.1.6", optional = true }
|
tokio-fs = { version = "0.1.6", optional = true }
|
||||||
tokio-io = { version = "0.1.6", optional = true }
|
tokio-io = { version = "0.1.6", optional = true }
|
||||||
tokio-executor = { version = "0.1.5", optional = true }
|
tokio-executor = { version = "0.1.5", optional = true, path = "../tokio-executor" }
|
||||||
tokio-reactor = { version = "0.1.1", optional = true }
|
tokio-reactor = { version = "0.1.1", optional = true }
|
||||||
tokio-sync = { version = "0.1.3", optional = true }
|
tokio-sync = { version = "0.1.3", optional = true }
|
||||||
tokio-threadpool = { version = "0.1.8", optional = true }
|
tokio-threadpool = { version = "0.1.8", optional = true }
|
||||||
|
|||||||
@@ -60,7 +60,7 @@ pub mod thread_pool {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
pub use tokio_executor::{Executor, DefaultExecutor, SpawnError};
|
pub use tokio_executor::{Executor, TypedExecutor, DefaultExecutor, SpawnError};
|
||||||
|
|
||||||
use futures::{Future, IntoFuture};
|
use futures::{Future, IntoFuture};
|
||||||
use futures::future::{self, FutureResult};
|
use futures::future::{self, FutureResult};
|
||||||
|
|||||||
@@ -76,6 +76,15 @@ where T: Future<Item = (), Error = ()> + Send + 'static,
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<T> ::executor::TypedExecutor<T> for Handle
|
||||||
|
where
|
||||||
|
T: Future<Item = (), Error = ()> + Send + 'static,
|
||||||
|
{
|
||||||
|
fn spawn(&mut self, future: T) -> Result<(), ::executor::SpawnError> {
|
||||||
|
Handle::spawn(self, future)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Error returned by the `run` function.
|
/// Error returned by the `run` function.
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct RunError {
|
pub struct RunError {
|
||||||
|
|||||||
@@ -73,3 +73,12 @@ impl ::executor::Executor for TaskExecutor {
|
|||||||
self.inner.spawn(future)
|
self.inner.spawn(future)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<T> ::executor::TypedExecutor<T> for TaskExecutor
|
||||||
|
where
|
||||||
|
T: Future<Item = (), Error = ()> + Send + 'static,
|
||||||
|
{
|
||||||
|
fn spawn(&mut self, future: T) -> Result<(), ::executor::SpawnError> {
|
||||||
|
::executor::Executor::spawn(self, Box::new(future))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user