Make the initial Waker provided by block_on_future Sync

As it was pointed out in the review, the initial `Waker` for `block_on_future`
was not holding up the `Sync` guarantees. If the `Waker` reference had been
passed to another thread and cloned there, the cloned threadsafe `Waker` would
have captured the wrong `Thread` handle.

This change removes the optimization. A threadsafe `Waker` is now immediately
created.

Author: Matthias247

src/libstd/tests/block_on_future.rs

@@ -75,23 +75,6 @@ impl Future for Yield {
     }
 }
 
-struct NeverReady {
-}
-
-impl NeverReady {
-    fn new() -> Self {
-        NeverReady {}
-    }
-}
-
-impl Future for NeverReady {
-    type Output = ();
-
-    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
-        Poll::Pending
-    }
-}
-
 struct WakerStore {
     waker: Option<Waker>,
 }
@@ -203,14 +186,6 @@ fn returns_result_from_task() {
     assert_eq!(42, result);
 }
 
-#[test]
-#[should_panic]
-fn panics_if_waker_was_not_cloned_and_task_is_not_ready() {
-    block_on_future(async {
-        NeverReady::new().await;
-    });
-}
-
 #[test]
 fn does_not_panic_if_waker_is_cloned_and_used_a_lot_later() {
     let store = Arc::new(Mutex::new(WakerStore {
@@ -227,3 +202,73 @@ fn does_not_panic_if_waker_is_cloned_and_used_a_lot_later() {
         WakeFromPreviouslyStoredWakerFuture::new(store).await;
     });
 }
+
+struct WakeSynchronouslyFromOtherThreadFuture {
+    was_polled: bool,
+    use_clone: bool,
+}
+
+impl WakeSynchronouslyFromOtherThreadFuture {
+    fn new(use_clone: bool) -> Self {
+        WakeSynchronouslyFromOtherThreadFuture {
+            was_polled: false,
+            use_clone,
+        }
+    }
+}
+
+/// This is just a helper to transfer a waker by reference/pointer
+/// to another thread without the availability of scoped threads.
+struct WakerBox {
+    waker: *const Waker,
+}
+
+unsafe impl Send for WakerBox {}
+
+impl Future for WakeSynchronouslyFromOtherThreadFuture {
+    type Output = ();
+
+    fn poll(mut self: core::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if !self.was_polled {
+            self.was_polled = true;
+            // This captures the waker by pointer and passes it to the other thread,
+            // since we don't have a scoped thread API available here.
+            // The pointer is however guaranteed to be alive when we call it, due to
+            // joining the thread in this scope.
+            let waker_box = WakerBox {
+                waker: cx.waker() as *const Waker,
+            };
+            let use_clone = self.use_clone;
+            spawn(move ||{
+                let x = waker_box;
+                unsafe {
+                    if !use_clone {
+                        (*(x.waker as *mut Waker)).wake_by_ref();
+                    } else {
+                        let cloned_waker = (*(x.waker as *mut Waker)).clone();
+                        cloned_waker.wake_by_ref();
+                    }
+                }
+            }).join().unwrap();
+            Poll::Pending
+        } else {
+            Poll::Ready(())
+        }
+    }
+}
+
+#[test]
+fn wake_synchronously_by_ref_from_other_thread() {
+    block_on_future(async {
+        WakeSynchronouslyFromOtherThreadFuture::new(false).await;
+        Yield::new(10).await;
+    })
+}
+
+#[test]
+fn clone_and_wake_synchronously_from_other_thread() {
+    block_on_future(async {
+        WakeSynchronouslyFromOtherThreadFuture::new(true).await;
+        Yield::new(10).await;
+    })
+}

src/libstd/thread/block_on_future.rs

@@ -9,44 +9,21 @@ use crate::pin::Pin;
 use crate::sync::Arc;
 use super::{current, park, Inner, Thread};
 
-/// Carries a flag that is used to wakeup the executor.
-/// A pointer to this struct is passed to the thread-local waker.
-struct LocalWakeState {
-    is_woken: bool,
-    waker_was_cloned: bool,
-}
-
-/// Returns the vtable that is used for waking up the executor
-/// from another thread.
-fn threadsafe_waker_vtable() -> &'static RawWakerVTable {
-    &RawWakerVTable::new(
-        clone_threadsafe_waker,
-        wake_threadsafe_waker,
-        wake_threadsafe_waker_by_ref,
-        drop_threadsafe_waker,
-    )
-}
-
 /// Returns the vtable that is used for waking up the executor
-/// from inside it's execution on the current thread.
-fn current_thread_waker_vtable() -> &'static RawWakerVTable {
+/// from any thread.
+fn waker_vtable() -> &'static RawWakerVTable {
     &RawWakerVTable::new(
-        create_threadsafe_waker,
-        wake_current_thread,
-        wake_current_thread_by_ref,
-        |_| {},
+        clone_waker,
+        wake_waker,
+        wake_waker_by_ref,
+        drop_waker,
     )
 }
 
-/// This method will be called when the waker reference gets cloned,
-/// which makes it possible to transfer it to another thread. In this
-/// case we have to create a threadsafe `Waker`. In order to to this
-/// we retain the thread handle and store it in the new `RawWaker`s
-/// data pointer.
-unsafe fn create_threadsafe_waker(data: *const()) -> RawWaker {
-    let wake_state = data as *mut LocalWakeState;
-    (*wake_state).waker_was_cloned = true;
-
+/// Creates a [`RawWaker`] which captures the current thread handle
+/// and allows to wake up the [`block_on_future`] executor from any
+/// thread by calling [`Thread::unpark()`].
+fn create_threadsafe_raw_waker() -> RawWaker {
     // Get the `Arc<Inner>` of a current thread handle and store into in
     // the type erased pointer.
     //
@@ -61,32 +38,23 @@ unsafe fn create_threadsafe_waker(data: *const()) -> RawWaker {
     // `let arc_thread = Arc::new(current());`
     let arc_thread_inner = current().inner;
     let ptr = Arc::into_raw(arc_thread_inner) as *const ();
-    RawWaker::new(ptr, threadsafe_waker_vtable())
+    RawWaker::new(ptr, waker_vtable())
 }
 
-unsafe fn clone_threadsafe_waker(data: *const()) -> RawWaker {
+unsafe fn clone_waker(data: *const()) -> RawWaker {
     increase_refcount(data);
-    RawWaker::new(data, threadsafe_waker_vtable())
-}
-
-fn wake_current_thread(_data: *const()) {
-    unreachable!("A current thread waker can only be woken by reference");
+    RawWaker::new(data, waker_vtable())
 }
 
-unsafe fn wake_current_thread_by_ref(data: *const()) {
-    let wake_state = data as *mut LocalWakeState;
-    (*wake_state).is_woken = true;
-}
-
-unsafe fn wake_threadsafe_waker(data: *const ()) {
+unsafe fn wake_waker(data: *const ()) {
     let arc_thread_inner = Arc::from_raw(data as *const Inner);
     let thread = Thread {
         inner: arc_thread_inner,
     };
     thread.unpark();
 }
 
-unsafe fn wake_threadsafe_waker_by_ref(data: *const ()) {
+unsafe fn wake_waker_by_ref(data: *const ()) {
     // Retain `Arc`, but don't touch refcount by wrapping in `ManuallyDrop`
     let arc_thread_inner = Arc::from_raw(data as *const Inner);
     let thread = mem::ManuallyDrop::new(Thread {
@@ -95,7 +63,7 @@ unsafe fn wake_threadsafe_waker_by_ref(data: *const ()) {
     thread.unpark();
 }
 
-unsafe fn drop_threadsafe_waker(data: *const ()) {
+unsafe fn drop_waker(data: *const ()) {
     drop(Thread {
         inner: Arc::from_raw(data as *const Inner),
     })
@@ -138,52 +106,23 @@ pub fn block_on_future<F: Future>(mut future: F) -> F::Output {
     // out of this function again.
     let mut future = unsafe { Pin::new_unchecked(&mut future) };
 
-    let mut waker_state = LocalWakeState {
-        is_woken: true,
-        waker_was_cloned: false,
+    // Safety: The `Waker` that we create upholds all guarantees that are expected
+    // from a `Waker`
+    let waker = unsafe {
+        Waker::from_raw(create_threadsafe_raw_waker())
     };
 
-    // Safety: The `Waker` that we create here is references data on the current
-    // callstack. This is safe, since the polled `Future` only gets a reference
-    // to this `Waker`. When it tries to clone the `Waker`, a threadsafe and owned
-    // version is created instead.
-    unsafe {
-        let waker = Waker::from_raw(RawWaker::new(
-            &waker_state as *const LocalWakeState as *const (),
-            current_thread_waker_vtable()));
-
-        let mut cx = Context::from_waker(&waker);
-        loop {
-            while waker_state.is_woken {
-                // Reset is_woken, so that we do not spin if the poll does not
-                // directly wake us up.
-                waker_state.is_woken = false;
-                if let Poll::Ready(task_result) = future.as_mut().poll(&mut cx) {
-                    return task_result;
-                }
-            }
-
-            // The task is not ready, and the `Waker` had not been woken from the
-            // current thread. In order for us to proceed we wait until the
-            // thread gets unparked by another thread. If the `Waker` has not been
-            // cloned this will never happen and represents a deadlock, which
-            // gets reported here.
-            if !waker_state.waker_was_cloned {
-                panic!("Deadlock: Task is not ready, but the Waker had not been cloned");
-                // Note: This flag is never reset, since a `Waker` that had been cloned
-                // once can be cloned more often to wakeup this executor. We don't
-                // have knowledge on how many clones are around - therefore the
-                // deadlock detection only works for the case the `Waker` never
-                // gets cloned.
-            }
-            park();
-            // If thread::park has returned, we have been notified by another
-            // thread. Therefore we are woken.
-            // Remark: This flag can not be set by the other thread directly,
-            // because it may no longer be alive at the point of time when
-            // wake() is called.
-            waker_state.is_woken = true;
+    let mut cx = Context::from_waker(&waker);
+    loop {
+        if let Poll::Ready(task_result) = future.as_mut().poll(&mut cx) {
+            return task_result;
         }
+
+        // The task is not ready. In order for us to proceed we wait until the
+        // thread gets unparked. If the `Waker` had been woken inside `.poll()`,
+        // then `park()` will immediately return, and we will call `.poll()`
+        // again without any wait period.
+        park();
     }
 }
 
@@ -201,31 +140,28 @@ mod tests {
     fn check_refcounts() {
         let original = current_thread_refcount();
 
-        let waker_state = LocalWakeState {
-            is_woken: true,
-            waker_was_cloned: false,
-        };
-
-        let waker = unsafe { Waker::from_raw(RawWaker::new(
-            &waker_state as *const LocalWakeState as *const (),
-            current_thread_waker_vtable())) };
+        let waker = unsafe { Waker::from_raw(create_threadsafe_raw_waker()) };
+        assert_eq!(original + 1, current_thread_refcount());
 
         waker.wake_by_ref();
-        assert_eq!(original, current_thread_refcount());
+        assert_eq!(original + 1, current_thread_refcount());
 
         let clone1 = waker.clone();
-        assert_eq!(original + 1, current_thread_refcount());
-        let clone2 = waker.clone();
         assert_eq!(original + 2, current_thread_refcount());
-        let clone3 = clone1.clone();
+        let clone2 = waker.clone();
         assert_eq!(original + 3, current_thread_refcount());
+        let clone3 = clone1.clone();
+        assert_eq!(original + 4, current_thread_refcount());
 
         drop(clone1);
-        assert_eq!(original + 2, current_thread_refcount());
+        assert_eq!(original + 3, current_thread_refcount());
 
         clone2.wake_by_ref();
-        assert_eq!(original + 2, current_thread_refcount());
+        assert_eq!(original + 3, current_thread_refcount());
         clone2.wake();
+        assert_eq!(original + 2, current_thread_refcount());
+
+        drop(waker);
         assert_eq!(original + 1, current_thread_refcount());
 
         clone3.wake_by_ref();