From 3e7d0be51d04fa6562071ec7d726bec73fff6314 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Fri, 1 Nov 2019 13:50:17 -0700 Subject: [PATCH] executor: remove Executor & TypedExecutor traits (#1724) The `Executor` trait is sub-optimal as it forces a `Box` to spawn. Instead, `tokio::spawn` delegates to the specific runtime implementation set for the current execution context. `TypedExecutor`, while useful, has seen limited adoption. As such, it is removed from `tokio` proper. Moving it to `tokio-util` is a possibility that can be explored as follow up work. --- tokio/src/executor/current_thread/mod.rs | 17 -- tokio/src/executor/error.rs | 49 ----- tokio/src/executor/executor.rs | 181 ------------------ tokio/src/executor/global.rs | 144 ++------------ tokio/src/executor/mod.rs | 12 +- tokio/src/executor/thread_pool/mod.rs | 8 - tokio/src/executor/thread_pool/pool.rs | 21 +- tokio/src/executor/thread_pool/set.rs | 19 +- tokio/src/executor/thread_pool/spawner.rs | 2 +- .../src/executor/thread_pool/tests/worker.rs | 2 +- tokio/src/executor/thread_pool/worker.rs | 14 +- tokio/src/executor/typed.rs | 178 ----------------- tokio/tests/executor.rs | 24 --- tokio/tests/executor_global.rs | 17 -- 14 files changed, 26 insertions(+), 662 deletions(-) delete mode 100644 tokio/src/executor/error.rs delete mode 100644 tokio/src/executor/executor.rs delete mode 100644 tokio/src/executor/typed.rs delete mode 100644 tokio/tests/executor.rs delete mode 100644 tokio/tests/executor_global.rs diff --git a/tokio/src/executor/current_thread/mod.rs b/tokio/src/executor/current_thread/mod.rs index 8ee4da829..24fee3a87 100644 --- a/tokio/src/executor/current_thread/mod.rs +++ b/tokio/src/executor/current_thread/mod.rs @@ -1,6 +1,5 @@ use crate::executor::park::{Park, Unpark}; use crate::executor::task::{self, JoinHandle, Schedule, Task}; -use crate::executor::Executor; use std::cell::UnsafeCell; use std::collections::VecDeque; @@ -294,22 +293,6 @@ impl Schedule for Scheduler { } } -impl Executor for &Scheduler { - fn spawn( - &mut self, - future: std::pin::Pin + Send>>, - ) -> Result<(), crate::executor::SpawnError> { - // Safety: This implementation should only be called by `global.rs` from - // the thread local. - // - // TODO: Delete this implementation. - unsafe { - Scheduler::spawn_background(self, future); - } - Ok(()) - } -} - impl

Drop for CurrentThread

where P: Park, diff --git a/tokio/src/executor/error.rs b/tokio/src/executor/error.rs deleted file mode 100644 index e9a5d9110..000000000 --- a/tokio/src/executor/error.rs +++ /dev/null @@ -1,49 +0,0 @@ -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, - "attempted to spawn task while the executor is at capacity or shut down" - ) - } -} - -impl Error for SpawnError {} diff --git a/tokio/src/executor/executor.rs b/tokio/src/executor/executor.rs deleted file mode 100644 index 5eeb43a22..000000000 --- a/tokio/src/executor/executor.rs +++ /dev/null @@ -1,181 +0,0 @@ -use crate::executor::SpawnError; - -use futures_util::future::{FutureExt, RemoteHandle}; -use std::future::Future; -use std::pin::Pin; - -/// 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 -/// -/// ``` -/// use tokio::executor::Executor; -/// -/// # fn docs(my_executor: &mut dyn Executor) { -/// my_executor.spawn(Box::pin(async { -/// println!("running on the executor"); -/// })).unwrap(); -/// # } -/// ``` -/// -/// [`spawn`]: #tymethod.spawn -/// [`poll`]: https://doc.rust-lang.org/std/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 - /// - /// ``` - /// use tokio::executor::Executor; - /// - /// # fn docs(my_executor: &mut dyn Executor) { - /// my_executor.spawn(Box::pin(async { - /// println!("running on the executor"); - /// })).unwrap(); - /// # } - /// ``` - fn spawn(&mut self, future: Pin + 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 - /// - /// ``` - /// use tokio::executor::Executor; - /// - /// # fn docs(my_executor: &mut dyn Executor) { - /// if my_executor.status().is_ok() { - /// my_executor.spawn(Box::pin(async { - /// println!("running on the executor"); - /// })).unwrap(); - /// } else { - /// println!("the executor is not in a good state"); - /// } - /// # } - /// ``` - fn status(&self) -> Result<(), SpawnError> { - Ok(()) - } -} - -impl dyn Executor { - /// Spawns a future object to run on this executor, returning a result of - /// its `RemoteHandle`. - /// - /// `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 - /// - /// ``` - /// use tokio::executor::Executor; - /// use futures_util::future::FutureExt; - /// - /// # fn docs(my_executor: &'static mut (dyn Executor + 'static)) { - /// let handle = my_executor.spawn_with_handle(Box::pin(async { - /// println!("running on the executor"); - /// })).unwrap(); - /// - /// let handle = handle.map(|_| println!("the future has completed")); - /// # } - /// ``` - pub fn spawn_with_handle( - &mut self, - future: Fut, - ) -> Result, SpawnError> - where - Fut: Future + Send + 'static, - Fut::Output: Send, - { - let (future, handle) = future.remote_handle(); - self.spawn(Box::pin(future))?; - Ok(handle) - } -} - -impl Executor for Box { - fn spawn( - &mut self, - future: Pin + Send>>, - ) -> Result<(), SpawnError> { - (**self).spawn(future) - } - - fn status(&self) -> Result<(), SpawnError> { - (**self).status() - } -} diff --git a/tokio/src/executor/global.rs b/tokio/src/executor/global.rs index ca9fdaf3a..ddb9b1ab0 100644 --- a/tokio/src/executor/global.rs +++ b/tokio/src/executor/global.rs @@ -2,66 +2,10 @@ use crate::executor::current_thread; #[cfg(feature = "rt-full")] -use crate::executor::thread_pool::ThreadPool; -use crate::executor::{Executor, SpawnError}; +use crate::executor::thread_pool; use std::cell::Cell; use std::future::Future; -use std::pin::Pin; - -/// Executes futures on the default executor for the current execution context. -/// -/// `DefaultExecutor` implements `Executor` and can be used to spawn futures -/// without referencing a specific executor. -/// -/// When an executor starts, it sets the `DefaultExecutor` handle to point to an -/// executor (usually itself) that is used to spawn new tasks. -/// -/// The current `DefaultExecutor` reference is tracked using a thread-local -/// variable and is set using `tokio::executor::with_default` -#[derive(Debug, Clone)] -pub struct DefaultExecutor { - _dummy: (), -} - -impl DefaultExecutor { - /// Returns a handle to the default executor for the current context. - /// - /// Futures may be spawned onto the default executor using this handle. - /// - /// The returned handle will reference whichever executor is configured as - /// the default **at the time `spawn` is called**. This enables - /// `DefaultExecutor::current()` to be called before an execution context is - /// setup, then passed **into** an execution context before it is used. - /// - /// This is also true for sending the handle across threads, so calling - /// `DefaultExecutor::current()` on thread A and then sending the result to - /// thread B will _not_ reference the default executor that was set on thread A. - pub fn current() -> DefaultExecutor { - DefaultExecutor { _dummy: () } - } - - #[inline] - fn with_current R, R>(f: F) -> Option { - EXECUTOR.with(|current_executor| match current_executor.get() { - State::Ready(executor_ptr) => { - let executor = unsafe { &mut *executor_ptr }; - Some(f(executor)) - } - #[cfg(feature = "rt-full")] - State::ThreadPool(threadpool_ptr) => { - let mut thread_pool = unsafe { &*threadpool_ptr }; - Some(f(&mut thread_pool)) - } - #[cfg(feature = "rt-current-thread")] - State::CurrentThread(current_thread_ptr) => { - let mut current_thread = unsafe { &*current_thread_ptr }; - Some(f(&mut current_thread)) - } - State::Empty => None, - }) - } -} #[derive(Clone, Copy)] enum State { @@ -70,14 +14,11 @@ enum State { // default executor is a thread pool instance. #[cfg(feature = "rt-full")] - ThreadPool(*const ThreadPool), + ThreadPool(*const thread_pool::Spawner), // Current-thread executor #[cfg(feature = "rt-current-thread")] CurrentThread(*const current_thread::Scheduler), - - // default executor is set to a custom executor. - Ready(*mut dyn Executor), } thread_local! { @@ -85,36 +26,6 @@ thread_local! { static EXECUTOR: Cell = Cell::new(State::Empty) } -// ===== impl DefaultExecutor ===== - -impl super::Executor for DefaultExecutor { - fn spawn( - &mut self, - future: Pin + Send>>, - ) -> Result<(), SpawnError> { - DefaultExecutor::with_current(|executor| executor.spawn(future)) - .unwrap_or_else(|| Err(SpawnError::shutdown())) - } - - fn status(&self) -> Result<(), SpawnError> { - DefaultExecutor::with_current(|executor| executor.status()) - .unwrap_or_else(|| Err(SpawnError::shutdown())) - } -} - -impl super::TypedExecutor for DefaultExecutor -where - T: Future + Send + 'static, -{ - fn spawn(&mut self, future: T) -> Result<(), SpawnError> { - super::Executor::spawn(self, Box::pin(future)) - } - - fn status(&self) -> Result<(), SpawnError> { - super::Executor::status(self) - } -} - // ===== global spawn fns ===== /// Spawns a future on the default executor. @@ -163,10 +74,6 @@ where T: Future + Send + 'static, { EXECUTOR.with(|current_executor| match current_executor.get() { - State::Ready(executor_ptr) => { - let executor = unsafe { &mut *executor_ptr }; - executor.spawn(Box::pin(future)).unwrap(); - } #[cfg(feature = "rt-full")] State::ThreadPool(threadpool_ptr) => { let thread_pool = unsafe { &*threadpool_ptr }; @@ -182,7 +89,12 @@ where current_thread.spawn_background(future); } } - State::Empty => panic!("must be called from the context of Tokio runtime"), + State::Empty => { + // Explicit drop of `future` silences the warning that `future` is + // not used when neither rt-* feature flags are enabled. + drop(future); + panic!("must be called from the context of Tokio runtime"); + } }) } @@ -206,33 +118,14 @@ pub(super) fn current_thread_is_current(current_thread: ¤t_thread::Schedul } #[cfg(feature = "rt-full")] -pub(super) fn with_threadpool(thread_pool: &ThreadPool, f: F) -> R +pub(super) fn with_thread_pool(thread_pool: &thread_pool::Spawner, f: F) -> R where F: FnOnce() -> R, { - with_state(State::ThreadPool(thread_pool as *const ThreadPool), f) -} - -/// Set the default executor for the duration of the closure -/// -/// If a default executor is already set, it will be restored when the closure returns or if it -/// panics. -pub fn with_default(executor: &mut T, f: F) -> R -where - T: Executor, - F: FnOnce() -> R, -{ - // While scary, this is safe. The function takes a - // `&mut Executor`, which guarantees that the reference lives for the - // duration of `with_default`. - // - // Because we are always clearing the TLS value at the end of the - // function, we can cast the reference to 'static which thread-local - // cells require. - let executor = unsafe { hide_lt(executor as &mut _ as *mut _) }; - with_state(State::Ready(executor), f) + with_state(State::ThreadPool(thread_pool as *const _), f) } +#[cfg(feature = "rt-current-thread")] fn with_state(state: State, f: F) -> R where F: FnOnce() -> R, @@ -252,23 +145,8 @@ where let _reset = Reset(cell, was); - if let State::Ready(executor) = state { - let executor = unsafe { &mut *executor }; - - if executor.status().is_err() { - panic!("executor not active; is this because `with_default` is called with `DefaultExecutor`?"); - } - } - cell.set(state); f() }) } - -unsafe fn hide_lt<'a>(p: *mut (dyn Executor + 'a)) -> *mut (dyn Executor + 'static) { - use std::mem; - // false positive: https://github.com/rust-lang/rust-clippy/issues/2906 - #[allow(clippy::transmute_ptr_to_ptr)] - mem::transmute(p) -} diff --git a/tokio/src/executor/mod.rs b/tokio/src/executor/mod.rs index fad957818..a888c61b4 100644 --- a/tokio/src/executor/mod.rs +++ b/tokio/src/executor/mod.rs @@ -47,15 +47,8 @@ mod tests; mod enter; pub use self::enter::{enter, exit, Enter, EnterError}; -mod error; -pub use self::error::SpawnError; - -#[allow(clippy::module_inception)] -mod executor; -pub use self::executor::Executor; - mod global; -pub use self::global::{spawn, with_default, DefaultExecutor}; +pub use self::global::spawn; pub(crate) mod loom; @@ -66,9 +59,6 @@ mod task; #[cfg(feature = "rt-current-thread")] pub use self::task::{JoinError, JoinHandle}; -mod typed; -pub use self::typed::TypedExecutor; - #[cfg(feature = "rt-full")] mod util; diff --git a/tokio/src/executor/thread_pool/mod.rs b/tokio/src/executor/thread_pool/mod.rs index 76dc78e2e..c18d87661 100644 --- a/tokio/src/executor/thread_pool/mod.rs +++ b/tokio/src/executor/thread_pool/mod.rs @@ -37,14 +37,6 @@ mod tests; #[cfg(feature = "blocking")] pub use worker::blocking; -// These exports are used in tests -#[cfg(test)] -#[allow(warnings)] -pub(crate) use self::worker::create_set as create_pool; - -pub(crate) type BoxFuture = - std::pin::Pin + Send + 'static>>; - #[cfg(not(loom))] const LOCAL_QUEUE_CAPACITY: usize = 256; diff --git a/tokio/src/executor/thread_pool/pool.rs b/tokio/src/executor/thread_pool/pool.rs index 6ffcbba45..67235eae2 100644 --- a/tokio/src/executor/thread_pool/pool.rs +++ b/tokio/src/executor/thread_pool/pool.rs @@ -1,7 +1,6 @@ use crate::executor::blocking::PoolWaiter; use crate::executor::task::JoinHandle; use crate::executor::thread_pool::{shutdown, Builder, Spawner}; -use crate::executor::Executor; use std::fmt; use std::future::Future; @@ -52,14 +51,6 @@ impl ThreadPool { self.spawner.spawn(future) } - /// Spawn a task in the background - pub(crate) fn spawn_background(&self, future: F) - where - F: Future + Send + 'static, - { - self.spawner.spawn_background(future); - } - /// Block the current thread waiting for the future to complete. /// /// The future will execute on the current thread, but all spawned tasks @@ -68,7 +59,7 @@ impl ThreadPool { where F: Future, { - crate::executor::global::with_threadpool(self, || { + crate::executor::global::with_thread_pool(self.spawner(), || { let mut enter = crate::executor::enter().expect("attempting to block while on a Tokio executor"); crate::executor::blocking::with_pool(self.spawner.blocking_pool(), || { @@ -92,16 +83,6 @@ impl Default for ThreadPool { } } -impl Executor for &ThreadPool { - fn spawn( - &mut self, - future: std::pin::Pin + Send>>, - ) -> Result<(), crate::executor::SpawnError> { - ThreadPool::spawn_background(self, future); - Ok(()) - } -} - impl fmt::Debug for ThreadPool { fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { fmt.debug_struct("ThreadPool").finish() diff --git a/tokio/src/executor/thread_pool/set.rs b/tokio/src/executor/thread_pool/set.rs index a158c0d3e..e878cad5f 100644 --- a/tokio/src/executor/thread_pool/set.rs +++ b/tokio/src/executor/thread_pool/set.rs @@ -6,14 +6,13 @@ use crate::executor::loom::rand::seed; use crate::executor::loom::sync::Arc; use crate::executor::park::Unpark; use crate::executor::task::{self, JoinHandle, Task}; -use crate::executor::thread_pool::{current, queue, BoxFuture, Idle, Owned, Shared}; +use crate::executor::thread_pool::{current, queue, Idle, Owned, Shared}; use crate::executor::util::{CachePadded, FastRand}; -use crate::executor::{Executor, SpawnError}; use std::cell::UnsafeCell; use std::future::Future; -pub(crate) struct Set

+pub(super) struct Set

where P: 'static, { @@ -206,17 +205,3 @@ impl Set> { handle } } - -impl

Executor for &Set

-where - P: Unpark, -{ - fn spawn(&mut self, future: BoxFuture) -> Result<(), SpawnError> { - self.spawn_background(future); - Ok(()) - } - - fn status(&self) -> Result<(), SpawnError> { - Ok(()) - } -} diff --git a/tokio/src/executor/thread_pool/spawner.rs b/tokio/src/executor/thread_pool/spawner.rs index 301ac5aae..b33c7cad1 100644 --- a/tokio/src/executor/thread_pool/spawner.rs +++ b/tokio/src/executor/thread_pool/spawner.rs @@ -38,7 +38,7 @@ impl Spawner { } /// Spawn a task in the background - pub(super) fn spawn_background(&self, future: F) + pub(crate) fn spawn_background(&self, future: F) where F: Future + Send + 'static, { diff --git a/tokio/src/executor/thread_pool/tests/worker.rs b/tokio/src/executor/thread_pool/tests/worker.rs index f5f9bace0..20b640c86 100644 --- a/tokio/src/executor/thread_pool/tests/worker.rs +++ b/tokio/src/executor/thread_pool/tests/worker.rs @@ -13,7 +13,7 @@ macro_rules! pool { (! $n:expr) => {{ let mut mock_park = crate::executor::tests::mock_park::MockPark::new(); let blocking = std::sync::Arc::new(crate::executor::blocking::Pool::default()); - let (pool, workers) = thread_pool::create_pool( + let (pool, workers) = thread_pool::worker::create_set( $n, |index| Box::new(mock_park.mk_park(index)), Arc::new(Box::new(|_| { diff --git a/tokio/src/executor/thread_pool/worker.rs b/tokio/src/executor/thread_pool/worker.rs index b78c214ab..130dad0a4 100644 --- a/tokio/src/executor/thread_pool/worker.rs +++ b/tokio/src/executor/thread_pool/worker.rs @@ -1,7 +1,7 @@ use crate::executor::loom::sync::Arc; use crate::executor::park::{Park, Unpark}; use crate::executor::task::Task; -use crate::executor::thread_pool::{current, Owned, Shared}; +use crate::executor::thread_pool::{current, Owned, Shared, Spawner}; use std::cell::Cell; use std::ops::{Deref, DerefMut}; @@ -71,7 +71,7 @@ pub(crate) struct Worker { gone: Cell, } -pub(crate) fn create_set( +pub(super) fn create_set( pool_size: usize, mk_park: F, launch_worker: LaunchWorker

, @@ -128,12 +128,16 @@ where } } - pub(super) fn run(mut self) { + pub(super) fn run(mut self) + where + P: Park>, + { let pool = Arc::clone(&self.entry.pool); let pool = &pool; let index = self.entry.index; - let mut executor = &**pool; + let executor = &**pool; + let spawner = Spawner::new(pool.clone()); let entry = &mut self.entry; let launch_worker = &self.launch_worker; @@ -146,7 +150,7 @@ where current::set(&pool, index, || { let _enter = crate::executor::enter().expect("executor already running on thread"); - crate::executor::with_default(&mut executor, || { + crate::executor::global::with_thread_pool(&spawner, || { crate::executor::blocking::with_pool(blocking, || { ON_BLOCK.with(|ob| { // Ensure that the ON_BLOCK is removed from the thread-local context diff --git a/tokio/src/executor/typed.rs b/tokio/src/executor/typed.rs deleted file mode 100644 index e7cbe373b..000000000 --- a/tokio/src/executor/typed.rs +++ /dev/null @@ -1,178 +0,0 @@ -use crate::executor::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. -/// -/// [`spawn`]: TypedExecutor::spawn -/// ``` -/// use tokio::executor::TypedExecutor; -/// use tokio::sync::oneshot; -/// -/// use futures_core::{ready, Stream}; -/// use std::future::Future; -/// use std::pin::Pin; -/// use std::task::{Context, Poll}; -/// -/// async fn drain(stream: T, executor: &mut E) -/// where -/// T: Stream + Unpin, -/// E: TypedExecutor> -/// { -/// let (tx, rx) = oneshot::channel(); -/// -/// executor.spawn(Drain { -/// stream, -/// tx: Some(tx), -/// }).unwrap(); -/// -/// rx.await.unwrap() -/// } -/// -/// // The background task -/// pub struct Drain { -/// stream: T, -/// tx: Option>, -/// } -/// -/// impl Future for Drain { -/// type Output = (); -/// -/// fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { -/// loop { -/// let item = ready!( -/// Pin::new(&mut self.stream).poll_next(cx) -/// ); -/// -/// if item.is_none() { break; } -/// } -/// -/// let _ = self.tx.take().unwrap().send(()).map_err(|_| ()); -/// Poll::Ready(()) -/// } -/// } -/// ``` -/// -/// 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 { - /// 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 - /// use tokio::executor::TypedExecutor; - /// - /// use std::future::Future; - /// use std::pin::Pin; - /// use std::task::{Context, Poll}; - /// - /// fn example(my_executor: &mut T) - /// where - /// T: TypedExecutor, - /// { - /// my_executor.spawn(MyFuture).unwrap(); - /// } - /// - /// struct MyFuture; - /// - /// impl Future for MyFuture { - /// type Output = (); - /// - /// fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { - /// println!("running on the executor"); - /// Poll::Ready(()) - /// } - /// } - /// ``` - 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 - /// use tokio::executor::TypedExecutor; - /// - /// use std::future::Future; - /// use std::pin::Pin; - /// use std::task::{Context, Poll}; - /// - /// fn example(my_executor: &mut T) - /// where - /// T: TypedExecutor, - /// { - /// 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 Output = (); - /// - /// fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { - /// println!("running on the executor"); - /// Poll::Ready(()) - /// } - /// } - /// ``` - fn status(&self) -> Result<(), SpawnError> { - Ok(()) - } -} - -impl TypedExecutor for Box -where - E: TypedExecutor, -{ - fn spawn(&mut self, future: T) -> Result<(), SpawnError> { - (**self).spawn(future) - } - - fn status(&self) -> Result<(), SpawnError> { - (**self).status() - } -} diff --git a/tokio/tests/executor.rs b/tokio/tests/executor.rs deleted file mode 100644 index e3fd6a5e4..000000000 --- a/tokio/tests/executor.rs +++ /dev/null @@ -1,24 +0,0 @@ -#![warn(rust_2018_idioms)] - -use tokio::executor::DefaultExecutor; - -use std::future::Future; -use std::pin::Pin; - -mod out_of_executor_context { - use super::*; - use tokio::executor::Executor; - - fn test(spawn: F) - where - F: Fn(Pin + Send>>) -> Result<(), E>, - { - let res = spawn(Box::pin(async {})); - assert!(res.is_err()); - } - - #[test] - fn spawn() { - test(|f| DefaultExecutor::current().spawn(f)); - } -} diff --git a/tokio/tests/executor_global.rs b/tokio/tests/executor_global.rs deleted file mode 100644 index 6c14f065b..000000000 --- a/tokio/tests/executor_global.rs +++ /dev/null @@ -1,17 +0,0 @@ -use tokio::executor::{with_default, DefaultExecutor}; - -#[test] -fn default_executor_is_send_and_sync() { - fn assert_send_sync() {} - - assert_send_sync::(); -} - -#[test] -#[should_panic] -fn nested_default_executor_status() { - let _enter = tokio::executor::enter().unwrap(); - let mut executor = DefaultExecutor::current(); - - let _result = with_default(&mut executor, || ()); -}