runtime: don't sleep in current_thread if before_park populated deferred queue

This ensures that if a before_park hook calls yield_now(), the runtime
doesn't enter a blocking sleep, which would delay or deadlock the
deferred task.

Author: Darksonn

tokio/tests/rt_basic.rs

@@ -456,3 +456,43 @@ fn rt() -> Runtime {
         .build()
         .unwrap()
 }
+
+#[test]
+fn before_park_yields() {
+    use futures::task::ArcWake;
+    use std::sync::Arc;
+    use tokio::runtime::Builder;
+    use tokio::sync::Notify;
+
+    struct MyWaker(Notify);
+
+    impl ArcWake for MyWaker {
+        fn wake_by_ref(arc_self: &Arc<Self>) {
+            arc_self.0.notify_one();
+        }
+    }
+
+    let notify = Arc::new(MyWaker(Notify::new()));
+    let notify2 = notify.clone();
+    let waker = futures::task::waker(notify2);
+    let woken = Arc::new(AtomicBool::new(false));
+    let woken2 = woken.clone();
+
+    let rt = Builder::new_current_thread()
+        .enable_all()
+        .on_thread_park(move || {
+            if !woken2.swap(true, Ordering::SeqCst) {
+                let mut cx = Context::from_waker(&waker);
+                let mut fut = std::pin::pin!(tokio::task::yield_now());
+                let _ = Pin::new(&mut fut).poll(&mut cx);
+            }
+        })
+        .build()
+        .unwrap();
+
+    rt.block_on(async {
+        notify.0.notified().await;
+    });
+
+    assert!(woken.load(Ordering::SeqCst));
+}