Prevent basic scheduler panic on task release (fixes tokio-rs#3662)

tokio/src/runtime/basic_scheduler.rs

@@ -373,12 +373,14 @@ impl Schedule for Arc<Shared> {
         use std::ptr::NonNull;
 
         CURRENT.with(|maybe_cx| {
-            let cx = maybe_cx.expect("scheduler context missing");
-
-            // safety: the task is inserted in the list in `bind`.
-            unsafe {
-                let ptr = NonNull::from(task.header());
-                cx.tasks.borrow_mut().owned.remove(ptr)
+            if let Some(cx) = maybe_cx {
+                // safety: the task is inserted in the list in `bind`.
+                unsafe {
+                    let ptr = NonNull::from(task.header());
+                    cx.tasks.borrow_mut().owned.remove(ptr)
+                }
+            } else {
+                None
             }
         })
     }

tokio/tests/task_abort.rs

@@ -24,3 +24,51 @@ fn test_abort_without_panic_3157() {
         let _ = handle.await;
     });
 }
+
+/// Checks that a suspended task can be aborted inside of a current_thread
+/// executor without panicking as reported in issue #3662:
+/// <https://github.com/tokio-rs/tokio/issues/3662>.
+#[test]
+fn test_abort_without_panic_3662() {
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Arc;
+
+    struct DropCheck(Arc<AtomicBool>);
+
+    impl Drop for DropCheck {
+        fn drop(&mut self) {
+            self.0.store(true, Ordering::SeqCst);
+        }
+    }
+
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .worker_threads(1)
+        .build()
+        .unwrap();
+
+    rt.block_on(async move {
+        let drop_flag = Arc::new(AtomicBool::new(false));
+        let drop_flag2 = drop_flag.clone();
+
+        let j = tokio::spawn(async move {
+            let drop_check = DropCheck(drop_flag2);
+            futures::future::pending::<()>().await;
+            drop(drop_check);
+        });
+
+        let task = tokio::task::spawn_blocking(move || {
+            // This runs in a separate thread so it doesn't have immediate
+            // thread-local access to the executor. It does however transition
+            // the underlying task to be completed, which will cause it to be
+            // dropped (in this thread no less).
+            j.abort();
+            j
+        })
+        .await
+        .unwrap();
+
+        assert!(drop_flag.load(Ordering::SeqCst));
+        let result = task.await;
+        assert!(result.unwrap_err().is_cancelled());
+    });
+}