Update to new futures_api (wake-by-ref)

Author: taiki-e

.travis.yml

@@ -25,7 +25,7 @@ matrix:
 
     # When updating this, the reminder to update the minimum required version in README.md.
     - name: cargo test (minimum required version)
-      rust: nightly-2019-04-08
+      rust: nightly-2019-04-13
 
     - name: cargo clippy
       rust: nightly

futures-core/src/task/__internal/atomic_waker.rs

@@ -241,7 +241,7 @@ impl AtomicWaker {
                 // Currently in the process of waking the task, i.e.,
                 // `wake` is currently being called on the old task handle.
                 // So, we call wake on the new waker
-                waker.wake();
+                waker.wake_by_ref();
             }
             state => {
                 // In this case, a concurrent thread is holding the

futures-executor/benches/thread_notify.rs

@@ -24,7 +24,7 @@ fn thread_yield_single_thread_one_wait(b: &mut Bencher) {
                 Poll::Ready(())
             } else {
                 self.rem -= 1;
-                cx.waker().wake();
+                cx.waker().wake_by_ref();
                 Poll::Pending
             }
         }
@@ -52,7 +52,7 @@ fn thread_yield_single_thread_many_wait(b: &mut Bencher) {
                 Poll::Ready(())
             } else {
                 self.rem -= 1;
-                cx.waker().wake();
+                cx.waker().wake_by_ref();
                 Poll::Pending
             }
         }

futures-executor/src/local_pool.rs

@@ -50,7 +50,7 @@ thread_local! {
 }
 
 impl ArcWake for ThreadNotify {
-    fn wake(arc_self: &Arc<Self>) {
+    fn wake_by_ref(arc_self: &Arc<Self>) {
         arc_self.thread.unpark();
     }
 }

futures-executor/src/thread_pool.rs

@@ -339,7 +339,7 @@ impl fmt::Debug for Task {
 }
 
 impl ArcWake for WakeHandle {
-    fn wake(arc_self: &Arc<Self>) {
+    fn wake_by_ref(arc_self: &Arc<Self>) {
         match arc_self.mutex.notify() {
             Ok(task) => arc_self.exec.state.send(Message::Run(task)),
             Err(()) => {}

futures-executor/tests/local_pool.rs

@@ -147,7 +147,7 @@ fn tasks_are_scheduled_fairly() {
                 return Poll::Ready(());
             }
 
-            cx.waker().wake();
+            cx.waker().wake_by_ref();
             Poll::Pending
         }
     }
@@ -167,4 +167,3 @@ fn tasks_are_scheduled_fairly() {
 
     pool.run();
 }
-

futures-io/src/lib.rs

@@ -100,7 +100,7 @@ mod if_std {
         ///
         /// If no data is available for reading, the method returns
         /// `Ok(Poll::Pending)` and arranges for the current task (via
-        /// `cx.waker().wake()`) to receive a notification when the object becomes
+        /// `cx.waker().wake_by_ref()`) to receive a notification when the object becomes
         /// readable or is closed.
         ///
         /// # Implementation
@@ -122,7 +122,7 @@ mod if_std {
         ///
         /// If no data is available for reading, the method returns
         /// `Ok(Poll::Pending)` and arranges for the current task (via
-        /// `cx.waker().wake()`) to receive a notification when the object becomes
+        /// `cx.waker().wake_by_ref()`) to receive a notification when the object becomes
         /// readable or is closed.
         /// By default, this method delegates to using `poll_read` on the first
         /// buffer in `vec`. Objects which support vectored IO should override
@@ -160,7 +160,7 @@ mod if_std {
         ///
         /// If the object is not ready for writing, the method returns
         /// `Ok(Poll::Pending)` and arranges for the current task (via
-        /// `cx.waker().wake()`) to receive a notification when the object becomes
+        /// `cx.waker().wake_by_ref()`) to receive a notification when the object becomes
         /// readable or is closed.
         ///
         /// # Implementation
@@ -182,7 +182,7 @@ mod if_std {
         ///
         /// If the object is not ready for writing, the method returns
         /// `Ok(Poll::Pending)` and arranges for the current task (via
-        /// `cx.waker().wake()`) to receive a notification when the object becomes
+        /// `cx.waker().wake_by_ref()`) to receive a notification when the object becomes
         /// readable or is closed.
         ///
         /// By default, this method delegates to using `poll_write` on the first
@@ -213,7 +213,7 @@ mod if_std {
         ///
         /// If flushing cannot immediately complete, this method returns
         /// `Ok(Poll::Pending)` and arranges for the current task (via
-        /// `cx.waker().wake()`) to receive a notification when the object can make
+        /// `cx.waker().wake_by_ref()`) to receive a notification when the object can make
         /// progress towards flushing.
         ///
         /// # Implementation
@@ -230,7 +230,7 @@ mod if_std {
         ///
         /// If closing cannot immediately complete, this function returns
         /// `Ok(Poll::Pending)` and arranges for the current task (via
-        /// `cx.waker().wake()`) to receive a notification when the object can make
+        /// `cx.waker().wake_by_ref()`) to receive a notification when the object can make
         /// progress towards closing.
         ///
         /// # Implementation

futures-sink/src/lib.rs

@@ -59,7 +59,7 @@ pub trait Sink<Item> {
     ///
     /// This method returns `Poll::Ready` once the underlying sink is ready to
     /// receive data. If this method returns `Poll::Pending`, the current task
-    /// is registered to be notified (via `cx.waker().wake()`) when `poll_ready`
+    /// is registered to be notified (via `cx.waker().wake_by_ref()`) when `poll_ready`
     /// should be called again.
     ///
     /// In most cases, if the sink encounters an error, the sink will
@@ -95,7 +95,7 @@ pub trait Sink<Item> {
     /// via `start_send` have been flushed.
     ///
     /// Returns `Ok(Poll::Pending)` if there is more work left to do, in which
-    /// case the current task is scheduled (via `cx.waker().wake()`) to wake up when
+    /// case the current task is scheduled (via `cx.waker().wake_by_ref()`) to wake up when
     /// `poll_flush` should be called again.
     ///
     /// In most cases, if the sink encounters an error, the sink will
@@ -108,7 +108,7 @@ pub trait Sink<Item> {
     /// has been successfully closed.
     ///
     /// Returns `Ok(Poll::Pending)` if there is more work left to do, in which
-    /// case the current task is scheduled (via `cx.waker().wake()`) to wake up when
+    /// case the current task is scheduled (via `cx.waker().wake_by_ref()`) to wake up when
     /// `poll_close` should be called again.
     ///
     /// If this function encounters an error, the sink should be considered to

futures-test/src/future/pending_once.rs

@@ -39,7 +39,7 @@ impl<Fut: Future> Future for PendingOnce<Fut> {
             self.as_mut().future().poll(cx)
         } else {
             *self.as_mut().polled_before() = true;
-            cx.waker().wake();
+            cx.waker().wake_by_ref();
             Poll::Pending
         }
     }

futures-test/src/task/panic_waker.rs

@@ -2,20 +2,20 @@ use futures_core::task::{Waker, RawWaker, RawWakerVTable};
 use core::cell::UnsafeCell;
 use core::ptr::null;
 
-unsafe fn clone_panic_waker(_data: *const()) -> RawWaker {
+unsafe fn clone_panic_waker(_data: *const ()) -> RawWaker {
     raw_panic_waker()
 }
 
-unsafe fn noop(_data: *const()) {
-}
+unsafe fn noop(_data: *const ()) {}
 
-unsafe fn wake_panic(_data: *const()) {
+unsafe fn wake_panic(_data: *const ()) {
     panic!("should not be woken");
 }
 
 const PANIC_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
     clone_panic_waker,
     wake_panic,
+    wake_panic,
     noop,
 );
 
@@ -37,7 +37,7 @@ fn raw_panic_waker() -> RawWaker {
 /// waker.wake(); // Will panic
 /// ```
 pub fn panic_waker() -> Waker {
-    unsafe { Waker::new_unchecked(raw_panic_waker()) }
+    unsafe { Waker::from_raw(raw_panic_waker()) }
 }
 
 /// Get a global reference to a
@@ -52,7 +52,7 @@ pub fn panic_waker() -> Waker {
 /// use futures_test::task::panic_waker_ref;
 ///
 /// let waker = panic_waker_ref();
-/// waker.wake(); // Will panic
+/// waker.wake_by_ref(); // Will panic
 /// ```
 pub fn panic_waker_ref() -> &'static Waker {
     thread_local! {

futures-test/src/task/wake_counter.rs

@@ -30,7 +30,7 @@ struct WakerInner {
 }
 
 impl ArcWake for WakerInner {
-    fn wake(arc_self: &Arc<Self>) {
+    fn wake_by_ref(arc_self: &Arc<Self>) {
         let _ = arc_self.count.fetch_add(1, Ordering::SeqCst);
     }
 }
@@ -49,7 +49,7 @@ impl ArcWake for WakerInner {
 ///
 /// assert_eq!(count, 0);
 ///
-/// waker.wake();
+/// waker.wake_by_ref();
 /// waker.wake();
 ///
 /// assert_eq!(count, 2);

futures-util/src/compat/compat01as03.rs

@@ -316,7 +316,7 @@ impl<'a> From<WakerToHandle<'a>> for NotifyHandle01 {
 
 impl Notify01 for NotifyWaker {
     fn notify(&self, _: usize) {
-        self.0.wake();
+        self.0.wake_by_ref();
     }
 }
 

futures-util/src/compat/compat03as01.rs

@@ -174,15 +174,15 @@ impl Current {
         }
 
         let ptr = current_to_ptr(self);
-        let vtable = &RawWakerVTable::new(clone, wake, drop);
+        let vtable = &RawWakerVTable::new(clone, wake, wake, drop);
         unsafe {
-            WakerRef::new(task03::Waker::new_unchecked(RawWaker::new(ptr, vtable)))
+            WakerRef::new(task03::Waker::from_raw(RawWaker::new(ptr, vtable)))
         }
     }
 }
 
 impl ArcWake03 for Current {
-    fn wake(arc_self: &Arc<Self>) {
+    fn wake_by_ref(arc_self: &Arc<Self>) {
         arc_self.0.notify();
     }
 }

futures-util/src/future/shared.rs

@@ -315,7 +315,7 @@ where
 }
 
 impl ArcWake for Notifier {
-    fn wake(arc_self: &Arc<Self>) {
+    fn wake_by_ref(arc_self: &Arc<Self>) {
         arc_self.state.compare_and_swap(POLLING, REPOLL, SeqCst);
 
         let wakers = &mut *arc_self.wakers.lock().unwrap();

futures-util/src/stream/futures_unordered/mod.rs

@@ -311,7 +311,7 @@ impl<Fut: Future> Stream for FuturesUnordered<Fut> {
                     // At this point, it may be worth yielding the thread &
                     // spinning a few times... but for now, just yield using the
                     // task system.
-                    cx.waker().wake();
+                    cx.waker().wake_by_ref();
                     return Poll::Pending;
                 }
                 Dequeue::Data(task) => task,

futures-util/src/stream/futures_unordered/task.rs

@@ -36,7 +36,7 @@ unsafe impl<Fut> Send for Task<Fut> {}
 unsafe impl<Fut> Sync for Task<Fut> {}
 
 impl<Fut> ArcWake for Task<Fut> {
-    fn wake(arc_self: &Arc<Self>) {
+    fn wake_by_ref(arc_self: &Arc<Self>) {
         let inner = match arc_self.ready_to_run_queue.upgrade() {
             Some(inner) => inner,
             None => return,

futures-util/src/stream/select_next_some.rs

@@ -34,7 +34,7 @@ impl<'a, St: Stream + FusedStream + Unpin> Future for SelectNextSome<'a, St> {
             Poll::Ready(item)
         } else {
             debug_assert!(self.stream.is_terminated());
-            cx.waker().wake();
+            cx.waker().wake_by_ref();
             Poll::Pending
         }
     }

futures-util/src/task/arc_wake.rs

@@ -17,26 +17,41 @@ pub trait ArcWake {
     ///
     /// Executors generally maintain a queue of "ready" tasks; `wake` should place
     /// the associated task onto this queue.
-    fn wake(arc_self: &Arc<Self>);
+    fn wake(self: Arc<Self>) {
+        Self::wake_by_ref(&self)
+    }
+
+    /// Indicates that the associated task is ready to make progress and should
+    /// be `poll`ed.
+    ///
+    /// This function can be called from an arbitrary thread, including threads which
+    /// did not create the `ArcWake` based `Waker`.
+    ///
+    /// Executors generally maintain a queue of "ready" tasks; `wake_by_ref` should place
+    /// the associated task onto this queue.
+    ///
+    /// This function is similar to `wake`, but must not consume the provided data
+    /// pointer.
+    fn wake_by_ref(arc_self: &Arc<Self>);
 
     /// Creates a `Waker` from an Arc<T>, if T implements `ArcWake`.
     ///
     /// If `wake()` is called on the returned `Waker`,
     /// the `wake()` function that is defined inside this trait will get called.
     fn into_waker(self: Arc<Self>) -> Waker where Self: Sized
     {
-        let ptr = Arc::into_raw(self) as *const();
+        let ptr = Arc::into_raw(self) as *const ();
 
         unsafe {
-            Waker::new_unchecked(RawWaker::new(ptr, waker_vtable!(Self)))
+            Waker::from_raw(RawWaker::new(ptr, waker_vtable!(Self)))
         }
     }
 }
 
 // FIXME: panics on Arc::clone / refcount changes could wreak havoc on the
 // code here. We should guard against this by aborting.
 
-unsafe fn increase_refcount<T: ArcWake>(data: *const()) {
+unsafe fn increase_refcount<T: ArcWake>(data: *const ()) {
     // Retain Arc by creating a copy
     let arc: Arc<T> = Arc::from_raw(data as *const T);
     let arc_clone = arc.clone();
@@ -46,19 +61,25 @@ unsafe fn increase_refcount<T: ArcWake>(data: *const()) {
 }
 
 // used by `waker_ref`
-pub(super) unsafe fn clone_arc_raw<T: ArcWake>(data: *const()) -> RawWaker {
+pub(super) unsafe fn clone_arc_raw<T: ArcWake>(data: *const ()) -> RawWaker {
     increase_refcount::<T>(data);
     RawWaker::new(data, waker_vtable!(T))
 }
 
-unsafe fn drop_arc_raw<T: ArcWake>(data: *const()) {
+unsafe fn drop_arc_raw<T: ArcWake>(data: *const ()) {
     drop(Arc::<T>::from_raw(data as *const T))
 }
 
 // used by `waker_ref`
-pub(super) unsafe fn wake_arc_raw<T: ArcWake>(data: *const()) {
+pub(super) unsafe fn wake_arc_raw<T: ArcWake>(data: *const ()) {
+    let arc: Arc<T> = Arc::from_raw(data as *const T);
+    ArcWake::wake(arc);
+}
+
+// used by `waker_ref`
+pub(super) unsafe fn wake_by_ref_arc_raw<T: ArcWake>(data: *const ()) {
     let arc: Arc<T> = Arc::from_raw(data as *const T);
-    ArcWake::wake(&arc);
+    ArcWake::wake_by_ref(&arc);
     mem::forget(arc);
 }
 
@@ -84,7 +105,7 @@ mod tests {
     }
 
     impl ArcWake for CountingWaker {
-        fn wake(arc_self: &Arc<Self>) {
+        fn wake_by_ref(arc_self: &Arc<Self>) {
             let mut lock = arc_self.nr_wake.lock().unwrap();
             *lock += 1;
         }
@@ -96,13 +117,13 @@ mod tests {
 
         let w1: Waker = ArcWake::into_waker(some_w.clone());
         assert_eq!(2, Arc::strong_count(&some_w));
-        w1.wake();
+        w1.wake_by_ref();
         assert_eq!(1, some_w.wakes());
 
         let w2 = w1.clone();
         assert_eq!(3, Arc::strong_count(&some_w));
 
-        w2.wake();
+        w2.wake_by_ref();
         assert_eq!(2, some_w.wakes());
 
         drop(w2);

futures-util/src/task/mod.rs

@@ -9,6 +9,7 @@ cfg_target_has_atomic! {
             &RawWakerVTable::new(
                 clone_arc_raw::<$ty>,
                 wake_arc_raw::<$ty>,
+                wake_by_ref_arc_raw::<$ty>,
                 drop_arc_raw::<$ty>,
             )
         };