diff --git a/tokio-executor/src/enter.rs b/tokio-executor/src/enter.rs index 64bfeb340..b005a1ada 100644 --- a/tokio-executor/src/enter.rs +++ b/tokio-executor/src/enter.rs @@ -67,6 +67,42 @@ pub fn enter() -> Result { }) } +// Forces the current "entered" state to be cleared while the closure +// is executed. +// +// # Warning +// +// This is hidden for a reason. Do not use without fully understanding +// executors. Misuing can easily cause your program to deadlock. +#[doc(hidden)] +pub fn exit R, R>(f: F) -> R { + // Reset in case the closure panics + struct Reset; + impl Drop for Reset { + fn drop(&mut self) { + ENTERED.with(|c| { + c.set(true); + }); + } + } + + ENTERED.with(|c| { + debug_assert!(c.get()); + c.set(false); + }); + + let reset = Reset; + let ret = f(); + ::std::mem::forget(reset); + + ENTERED.with(|c| { + assert!(!c.get(), "closure claimed permanent executor"); + c.set(true); + }); + + ret +} + impl Enter { /// Register a callback to be invoked if and when the thread /// ceased to act as an executor. diff --git a/tokio-executor/src/lib.rs b/tokio-executor/src/lib.rs index 5362d0fae..e7ae676a3 100644 --- a/tokio-executor/src/lib.rs +++ b/tokio-executor/src/lib.rs @@ -1,5 +1,7 @@ #![deny(missing_docs, missing_debug_implementations, warnings)] #![doc(html_root_url = "https://docs.rs/tokio-executor/0.1.7")] +// Our MSRV doesn't allow us to fix these warnings yet +#![allow(rust_2018_idioms)] //! Task execution related traits and utilities. //! @@ -61,7 +63,7 @@ mod global; pub mod park; mod typed; -pub use enter::{enter, Enter, EnterError}; +pub use enter::{enter, exit, Enter, EnterError}; pub use error::SpawnError; pub use executor::Executor; pub use global::{spawn, with_default, DefaultExecutor}; diff --git a/tokio-threadpool/Cargo.toml b/tokio-threadpool/Cargo.toml index 39c6e1947..1eecddbc9 100644 --- a/tokio-threadpool/Cargo.toml +++ b/tokio-threadpool/Cargo.toml @@ -21,7 +21,7 @@ keywords = ["futures", "tokio"] categories = ["concurrency", "asynchronous"] [dependencies] -tokio-executor = "0.1.7" +tokio-executor = { path = "../tokio-executor", version = "0.1.7" } futures = "0.1.19" crossbeam-deque = "0.7.0" crossbeam-queue = "0.1.0" diff --git a/tokio-threadpool/src/blocking.rs b/tokio-threadpool/src/blocking.rs index 9f91234b9..04f3356de 100644 --- a/tokio-threadpool/src/blocking.rs +++ b/tokio-threadpool/src/blocking.rs @@ -1,6 +1,7 @@ use worker::Worker; use futures::Poll; +use tokio_executor; use std::error::Error; use std::fmt; @@ -142,8 +143,11 @@ where // If the transition cannot happen, exit early try_ready!(res); - // Currently in blocking mode, so call the inner closure - let ret = f(); + // Currently in blocking mode, so call the inner closure. + // + // "Exit" the current executor in case the blocking function wants + // to call a different executor. + let ret = tokio_executor::exit(move || f()); // Try to transition out of blocking mode. This is a fast path that takes // back ownership of the worker if the worker handoff didn't complete yet. diff --git a/tokio-threadpool/src/lib.rs b/tokio-threadpool/src/lib.rs index 922a3f4d9..ecb2244c1 100644 --- a/tokio-threadpool/src/lib.rs +++ b/tokio-threadpool/src/lib.rs @@ -1,5 +1,7 @@ #![doc(html_root_url = "https://docs.rs/tokio-threadpool/0.1.14")] #![deny(warnings, missing_docs, missing_debug_implementations)] +// Our MSRV doesn't allow us to fix these warnings yet +#![allow(rust_2018_idioms)] //! A work-stealing based thread pool for executing futures. //! diff --git a/tokio-threadpool/tests/blocking.rs b/tokio-threadpool/tests/blocking.rs index 5fae2af27..74520e529 100644 --- a/tokio-threadpool/tests/blocking.rs +++ b/tokio-threadpool/tests/blocking.rs @@ -1,3 +1,4 @@ +extern crate tokio_executor; extern crate tokio_threadpool; extern crate env_logger; @@ -44,6 +45,28 @@ fn basic() { rx2.recv().unwrap(); } +#[test] +fn other_executors_can_run_inside_blocking() { + let _ = ::env_logger::try_init(); + + let pool = Builder::new().pool_size(1).max_blocking(1).build(); + + let (tx, rx) = mpsc::channel(); + + pool.spawn(lazy(move || { + let res = blocking(|| { + let _e = tokio_executor::enter().expect("nested blocking enter"); + tx.send(()).unwrap(); + }) + .unwrap(); + + assert!(res.is_ready()); + Ok(().into()) + })); + + rx.recv().unwrap(); +} + #[test] fn notify_task_on_capacity() { const BLOCKING: usize = 10;