runtime: check Enter in more places when blocking (#708)

- `tokio::run` checks Enter before creating a new threadpool and
  spawning the main future.
- `Runtime::block_on` now checks Enter
- `Runtime::block_on_all` now checks Enter
This commit is contained in:
Sean McArthur
2018-10-17 15:25:40 -07:00
committed by GitHub
parent 753336de8e
commit 7b5ef61aeb
2 changed files with 89 additions and 15 deletions
+17 -7
View File
@@ -209,11 +209,13 @@ struct Inner {
pub fn run<F>(future: F) pub fn run<F>(future: F)
where F: Future<Item = (), Error = ()> + Send + 'static, where F: Future<Item = (), Error = ()> + Send + 'static,
{ {
let mut runtime = Runtime::new().unwrap(); // Check enter before creating a new Runtime...
let mut entered = enter().expect("nested tokio::run");
let mut runtime = Runtime::new().expect("failed to start new Runtime");
runtime.spawn(future); runtime.spawn(future);
enter().expect("nested tokio::run") entered
.block_on(runtime.shutdown_on_idle()) .block_on(runtime.shutdown_on_idle())
.unwrap(); .expect("shutdown cannot error")
} }
impl Runtime { impl Runtime {
@@ -362,9 +364,10 @@ impl Runtime {
R: Send + 'static, R: Send + 'static,
E: Send + 'static, E: Send + 'static,
{ {
let mut entered = enter().expect("nested block_on");
let (tx, rx) = futures::sync::oneshot::channel(); let (tx, rx) = futures::sync::oneshot::channel();
self.spawn(future.then(move |r| tx.send(r).map_err(|_| unreachable!()))); self.spawn(future.then(move |r| tx.send(r).map_err(|_| unreachable!())));
rx.wait().unwrap() entered.block_on(rx).unwrap()
} }
/// Run a future to completion on the Tokio runtime, then wait for all /// Run a future to completion on the Tokio runtime, then wait for all
@@ -387,9 +390,16 @@ impl Runtime {
R: Send + 'static, R: Send + 'static,
E: Send + 'static, E: Send + 'static,
{ {
let res = self.block_on(future); let mut entered = enter().expect("nested block_on_all");
self.shutdown_on_idle().wait().unwrap(); let (tx, rx) = futures::sync::oneshot::channel();
res self.spawn(future.then(move |r| tx.send(r).map_err(|_| unreachable!())));
let block = rx
.map_err(|_| unreachable!())
.and_then(move |r| {
self.shutdown_on_idle()
.map(move |()| r)
});
entered.block_on(block).unwrap()
} }
/// Signals the runtime to shutdown once it becomes idle. /// Signals the runtime to shutdown once it becomes idle.
+72 -8
View File
@@ -391,14 +391,78 @@ mod from_block_on_all {
} }
} }
#[test] mod nested_enter {
fn run_in_run() { use super::*;
use tokio::runtime::current_thread;
use std::panic; use std::panic;
tokio::run(lazy(|| { fn test<F1, F2>(first: F1, nested: F2)
panic::catch_unwind(|| { where
tokio::run(lazy(|| { Ok::<(), ()>(()) })) F1: Fn(Box<Future<Item=(), Error=()> + Send>) + Send + 'static,
}).unwrap_err(); F2: Fn(Box<Future<Item=(), Error=()> + Send>) + panic::UnwindSafe + Send + 'static,
Ok::<(), ()>(()) {
})); let panicked = Arc::new(Mutex::new(false));
let panicked2 = panicked.clone();
// Since this is testing panics in other threads, printing about panics
// is noisy and can give the impression that the test is ignoring panics.
//
// It *is* ignoring them, but on purpose.
let prev_hook = panic::take_hook();
panic::set_hook(Box::new(|info| {
let s = info.to_string();
if s.starts_with("panicked at 'nested ")
|| s.starts_with("panicked at 'Multiple executors at once")
{
// expected, noop
} else {
println!("{}", s);
}
}));
first(Box::new(lazy(move || {
panic::catch_unwind(move || {
nested(Box::new(lazy(|| { Ok::<(), ()>(()) })))
}).expect_err("nested should panic");
*panicked2.lock().unwrap() = true;
Ok::<(), ()>(())
})));
panic::set_hook(prev_hook);
assert!(*panicked.lock().unwrap(), "nested call should have panicked");
}
fn threadpool_new() -> Runtime {
Runtime::new().expect("rt new")
}
#[test]
fn run_in_run() {
test(tokio::run, tokio::run);
}
#[test]
fn threadpool_block_on_in_run() {
test(tokio::run, |fut| {
let mut rt = threadpool_new();
rt.block_on(fut).unwrap();
});
}
#[test]
fn threadpool_block_on_all_in_run() {
test(tokio::run, |fut| {
let rt = threadpool_new();
rt.block_on_all(fut).unwrap();
});
}
#[test]
fn current_thread_block_on_all_in_run() {
test(tokio::run, |fut| {
current_thread::block_on_all(fut).unwrap();
});
}
} }