rt: avoid dropping a task in calls to wake() (#1972)

Calls to tasks should not be nested. Currently, while a task is being
executed and the runtime is shutting down, a call to wake() can result
in the wake target to be dropped. This, in turn, results in the drop
handler being called.

If the user holds a ref cell borrow, a mutex guard, or any such value,
dropping the task inline can result in a deadlock.

The fix is to permit tasks to be scheduled during the shutdown process
and dropping the tasks once they are popped from the queue.

Fixes #1929, #1886

1 files changed, 50 insertions, 0 deletions

diff --git a/tokio/tests/task_local_set.rs b/tokio/tests/task_local_set.rs
new file mode 100644
index 00000000..f5014275
--- /dev/null
+++ b/tokio/tests/task_local_set.rs
@@ -0,0 +1,50 @@
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+
+use tokio::runtime::Runtime;
+use tokio::sync::oneshot;
+use tokio::task::{self, LocalSet};
+
+#[test]
+fn acquire_mutex_in_drop() {
+    use futures::future::pending;
+
+    let (tx1, rx1) = oneshot::channel();
+    let (tx2, rx2) = oneshot::channel();
+
+    let mut rt = rt();
+    let local = LocalSet::new();
+
+    local.spawn_local(async move {
+        let _ = rx2.await;
+        unreachable!();
+    });
+
+    local.spawn_local(async move {
+        let _ = rx1.await;
+        let _ = tx2.send(()).unwrap();
+        unreachable!();
+    });
+
+    // Spawn a task that will never notify
+    local.spawn_local(async move {
+        pending::<()>().await;
+        tx1.send(()).unwrap();
+    });
+
+    // Tick the loop
+    local.block_on(&mut rt, async {
+        task::yield_now().await;
+    });
+
+    // Drop the LocalSet
+    drop(local);
+}
+
+fn rt() -> Runtime {
+    tokio::runtime::Builder::new()
+        .basic_scheduler()
+        .enable_all()
+        .build()
+        .unwrap()
+}