executor: add executor::exit (#1155)

This allows blocking on executors from within a `threadpool::blocking` call.

tokio-executor/src/enter.rs

@@ -67,6 +67,42 @@ pub fn enter() -> Result<Enter, EnterError> {
     })
 }
 
+// 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<F: FnOnce() -> 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.

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};

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"

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.

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.
 //!

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;