rt: fix basic_scheduler notification bug (#1861)

The "global executor" thread-local is to track where to spawn new tasks,
**not** which scheduler is active on the current thread. This fixes a
bug with scheduling tasks on the basic_scheduler by tracking the
currently active basic_scheduler with a dedicated thread-local variable.

Fixes: #1851

tokio/tests/fs.rs

@@ -0,0 +1,20 @@
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+
+use tokio::fs;
+use tokio_test::assert_ok;
+
+#[tokio::test]
+async fn path_read_write() {
+    let temp = tempdir();
+    let dir = temp.path();
+
+    assert_ok!(fs::write(dir.join("bar"), b"bytes").await);
+    let out = assert_ok!(fs::read(dir.join("bar")).await);
+
+    assert_eq!(out, b"bytes");
+}
+
+fn tempdir() -> tempfile::TempDir {
+    tempfile::tempdir().unwrap()
+}

tokio/tests/rt_common.rs

@@ -198,6 +198,21 @@ rt_test! {
         }
     }
 
+    #[test]
+    fn spawn_await_chain() {
+        let mut rt = rt();
+
+        let out = rt.block_on(async {
+            assert_ok!(tokio::spawn(async {
+                assert_ok!(tokio::spawn(async {
+                    "hello"
+                }).await)
+            }).await)
+        });
+
+        assert_eq!(out, "hello");
+    }
+
     #[test]
     fn outstanding_tasks_dropped() {
         let mut rt = rt();

tokio/tests/task_blocking.rs

@@ -0,0 +1,29 @@
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+
+use tokio::task;
+use tokio_test::assert_ok;
+
+use std::thread;
+use std::time::Duration;
+
+#[tokio::test]
+async fn basic_blocking() {
+    // Run a few times
+    for _ in 0..100 {
+        let out = assert_ok!(
+            tokio::spawn(async {
+                assert_ok!(
+                    task::spawn_blocking(|| {
+                        thread::sleep(Duration::from_millis(5));
+                        "hello"
+                    })
+                    .await
+                )
+            })
+            .await
+        );
+
+        assert_eq!(out, "hello");
+    }
+}