Implements the waker changes for task API stabilization.

.gitignore

@@ -3,3 +3,5 @@ target/
 Cargo.lock
 _site
 .sass-cache
+.idea
+.DS_Store

.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-01-11
+      rust: nightly-2019-02-15
 
     - name: cargo clippy
       rust: nightly

README.md

@@ -39,7 +39,7 @@ Now, you can use futures-rs:
 use futures::future::Future; // Note: It's not `futures_preview`
 ```
 
-The current version of futures-rs requires Rust nightly 2019-01-11 or later.
+The current version of futures-rs requires Rust nightly 2019-02-15 or later.
 
 ### Feature `std`
 

futures-channel/benches/sync_mpsc.rs

@@ -9,16 +9,16 @@ use {
         ready,
         stream::{Stream, StreamExt},
         sink::Sink,
-        task::{LocalWaker, Poll},
+        task::{Waker, Poll},
     },
-    futures_test::task::noop_local_waker_ref,
+    futures_test::task::noop_waker_ref,
     std::pin::Pin,
 };
 
 /// Single producer, single consumer
 #[bench]
 fn unbounded_1_tx(b: &mut Bencher) {
-    let lw = noop_local_waker_ref();
+    let waker = noop_waker_ref();
     b.iter(|| {
         let (tx, mut rx) = mpsc::unbounded();
 
@@ -27,20 +27,20 @@ fn unbounded_1_tx(b: &mut Bencher) {
         for i in 0..1000 {
 
             // Poll, not ready, park
-            assert_eq!(Poll::Pending, rx.poll_next_unpin(lw));
+            assert_eq!(Poll::Pending, rx.poll_next_unpin(waker));
 
             UnboundedSender::unbounded_send(&tx, i).unwrap();
 
             // Now poll ready
-            assert_eq!(Poll::Ready(Some(i)), rx.poll_next_unpin(lw));
+            assert_eq!(Poll::Ready(Some(i)), rx.poll_next_unpin(waker));
         }
     })
 }
 
 /// 100 producers, single consumer
 #[bench]
 fn unbounded_100_tx(b: &mut Bencher) {
-    let lw = noop_local_waker_ref();
+    let waker = noop_waker_ref();
     b.iter(|| {
         let (tx, mut rx) = mpsc::unbounded();
 
@@ -49,26 +49,26 @@ fn unbounded_100_tx(b: &mut Bencher) {
         // 1000 send/recv operations total, result should be divided by 1000
         for _ in 0..10 {
             for i in 0..tx.len() {
-                assert_eq!(Poll::Pending, rx.poll_next_unpin(lw));
+                assert_eq!(Poll::Pending, rx.poll_next_unpin(waker));
 
                 UnboundedSender::unbounded_send(&tx[i], i).unwrap();
 
-                assert_eq!(Poll::Ready(Some(i)), rx.poll_next_unpin(lw));
+                assert_eq!(Poll::Ready(Some(i)), rx.poll_next_unpin(waker));
             }
         }
     })
 }
 
 #[bench]
 fn unbounded_uncontended(b: &mut Bencher) {
-    let lw = noop_local_waker_ref();
+    let waker = noop_waker_ref();
     b.iter(|| {
         let (tx, mut rx) = mpsc::unbounded();
 
         for i in 0..1000 {
             UnboundedSender::unbounded_send(&tx, i).expect("send");
             // No need to create a task, because poll is not going to park.
-            assert_eq!(Poll::Ready(Some(i)), rx.poll_next_unpin(lw));
+            assert_eq!(Poll::Ready(Some(i)), rx.poll_next_unpin(waker));
         }
     })
 }
@@ -84,41 +84,41 @@ struct TestSender {
 impl Stream for TestSender {
     type Item = u32;
 
-    fn poll_next(mut self: Pin<&mut Self>, lw: &LocalWaker)
+    fn poll_next(mut self: Pin<&mut Self>, waker: &Waker)
         -> Poll<Option<Self::Item>>
     {
         let this = &mut *self;
         let mut tx = Pin::new(&mut this.tx);
 
-        ready!(tx.as_mut().poll_ready(lw)).unwrap();
+        ready!(tx.as_mut().poll_ready(waker)).unwrap();
         tx.as_mut().start_send(this.last + 1).unwrap();
         this.last += 1;
-        assert_eq!(Poll::Ready(Ok(())), tx.as_mut().poll_flush(lw));
+        assert_eq!(Poll::Ready(Ok(())), tx.as_mut().poll_flush(waker));
         Poll::Ready(Some(this.last))
     }
 }
 
 /// Single producers, single consumer
 #[bench]
 fn bounded_1_tx(b: &mut Bencher) {
-    let lw = noop_local_waker_ref();
+    let waker = noop_waker_ref();
     b.iter(|| {
         let (tx, mut rx) = mpsc::channel(0);
 
         let mut tx = TestSender { tx, last: 0 };
 
         for i in 0..1000 {
-            assert_eq!(Poll::Ready(Some(i + 1)), tx.poll_next_unpin(lw));
-            assert_eq!(Poll::Pending, tx.poll_next_unpin(lw));
-            assert_eq!(Poll::Ready(Some(i + 1)), rx.poll_next_unpin(lw));
+            assert_eq!(Poll::Ready(Some(i + 1)), tx.poll_next_unpin(waker));
+            assert_eq!(Poll::Pending, tx.poll_next_unpin(waker));
+            assert_eq!(Poll::Ready(Some(i + 1)), rx.poll_next_unpin(waker));
         }
     })
 }
 
 /// 100 producers, single consumer
 #[bench]
 fn bounded_100_tx(b: &mut Bencher) {
-    let lw = noop_local_waker_ref();
+    let waker = noop_waker_ref();
     b.iter(|| {
         // Each sender can send one item after specified capacity
         let (tx, mut rx) = mpsc::channel(0);
@@ -133,11 +133,11 @@ fn bounded_100_tx(b: &mut Bencher) {
         for i in 0..10 {
             for j in 0..tx.len() {
                 // Send an item
-                assert_eq!(Poll::Ready(Some(i + 1)), tx[j].poll_next_unpin(lw));
+                assert_eq!(Poll::Ready(Some(i + 1)), tx[j].poll_next_unpin(waker));
                 // Then block
-                assert_eq!(Poll::Pending, tx[j].poll_next_unpin(lw));
+                assert_eq!(Poll::Pending, tx[j].poll_next_unpin(waker));
                 // Recv the item
-                assert_eq!(Poll::Ready(Some(i + 1)), rx.poll_next_unpin(lw));
+                assert_eq!(Poll::Ready(Some(i + 1)), rx.poll_next_unpin(waker));
             }
         }
     })

futures-channel/src/mpsc/mod.rs

@@ -79,7 +79,7 @@
 // by the queue structure.
 
 use futures_core::stream::{FusedStream, Stream};
-use futures_core::task::{LocalWaker, Waker, Poll};
+use futures_core::task::{Waker, Poll};
 use futures_core::task::__internal::AtomicWaker;
 use std::any::Any;
 use std::error::Error;
@@ -555,7 +555,7 @@ impl<T> SenderInner<T> {
     /// - `Err(SendError)` if the receiver has been dropped.
     fn poll_ready(
         &mut self,
-        lw: &LocalWaker
+        waker: &Waker
     ) -> Poll<Result<(), SendError>> {
         let state = decode_state(self.inner.state.load(SeqCst));
         if !state.is_open {
@@ -564,7 +564,7 @@ impl<T> SenderInner<T> {
             }));
         }
 
-        self.poll_unparked(Some(lw)).map(Ok)
+        self.poll_unparked(Some(waker)).map(Ok)
     }
 
     /// Returns whether this channel is closed without needing a context.
@@ -582,7 +582,7 @@ impl<T> SenderInner<T> {
         self.inner.recv_task.wake();
     }
 
-    fn poll_unparked(&mut self, lw: Option<&LocalWaker>) -> Poll<()> {
+    fn poll_unparked(&mut self, waker: Option<&Waker>) -> Poll<()> {
         // First check the `maybe_parked` variable. This avoids acquiring the
         // lock in most cases
         if self.maybe_parked {
@@ -600,7 +600,7 @@ impl<T> SenderInner<T> {
             //
             // Update the task in case the `Sender` has been moved to another
             // task
-            task.task = lw.map(|lw| lw.clone().into_waker());
+            task.task = waker.map(|waker| waker.clone());
 
             Poll::Pending
         } else {
@@ -649,12 +649,12 @@ impl<T> Sender<T> {
     /// - `Err(SendError)` if the receiver has been dropped.
     pub fn poll_ready(
         &mut self,
-        lw: &LocalWaker
+        waker: &Waker,
     ) -> Poll<Result<(), SendError>> {
         let inner = self.0.as_mut().ok_or(SendError {
             kind: SendErrorKind::Disconnected,
         })?;
-        inner.poll_ready(lw)
+        inner.poll_ready(waker)
     }
 
     /// Returns whether this channel is closed without needing a context.
@@ -679,7 +679,7 @@ impl<T> UnboundedSender<T> {
     /// Check if the channel is ready to receive a message.
     pub fn poll_ready(
         &self,
-        _: &LocalWaker,
+        _: &Waker,
     ) -> Poll<Result<(), SendError>> {
         let inner = self.0.as_ref().ok_or(SendError {
             kind: SendErrorKind::Disconnected,
@@ -904,7 +904,7 @@ impl<T> Stream for Receiver<T> {
 
     fn poll_next(
         mut self: Pin<&mut Self>,
-        lw: &LocalWaker,
+        waker: &Waker,
     ) -> Poll<Option<T>> {
             // Try to read a message off of the message queue.
         match self.next_message() {
@@ -916,7 +916,7 @@ impl<T> Stream for Receiver<T> {
             },
             Poll::Pending => {
                 // There are no messages to read, in this case, park.
-                self.inner.as_ref().unwrap().recv_task.register(lw);
+                self.inner.as_ref().unwrap().recv_task.register(waker);
                 // Check queue again after parking to prevent race condition:
                 // a message could be added to the queue after previous `next_message`
                 // before `register` call.
@@ -971,9 +971,9 @@ impl<T> Stream for UnboundedReceiver<T> {
 
     fn poll_next(
         mut self: Pin<&mut Self>,
-        lw: &LocalWaker,
+        waker: &Waker,
     ) -> Poll<Option<T>> {
-        Pin::new(&mut self.0).poll_next(lw)
+        Pin::new(&mut self.0).poll_next(waker)
     }
 }
 

futures-channel/src/oneshot.rs

@@ -1,7 +1,7 @@
 //! A channel for sending a single message between asynchronous tasks.
 
 use futures_core::future::Future;
-use futures_core::task::{LocalWaker, Poll, Waker};
+use futures_core::task::{Waker, Poll};
 use std::pin::Pin;
 use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
@@ -154,7 +154,7 @@ impl<T> Inner<T> {
         }
     }
 
-    fn poll_cancel(&self, lw: &LocalWaker) -> Poll<()> {
+    fn poll_cancel(&self, waker: &Waker) -> Poll<()> {
         // Fast path up first, just read the flag and see if our other half is
         // gone. This flag is set both in our destructor and the oneshot
         // destructor, but our destructor hasn't run yet so if it's set then the
@@ -176,7 +176,7 @@ impl<T> Inner<T> {
         // `Receiver` may have been dropped. The first thing it does is set the
         // flag, and if it fails to acquire the lock it assumes that we'll see
         // the flag later on. So... we then try to see the flag later on!
-        let handle = lw.clone().into_waker();
+        let handle = waker.clone();
         match self.tx_task.try_lock() {
             Some(mut p) => *p = Some(handle),
             None => return Poll::Ready(()),
@@ -249,7 +249,7 @@ impl<T> Inner<T> {
         }
     }
 
-    fn recv(&self, lw: &LocalWaker) -> Poll<Result<T, Canceled>> {
+    fn recv(&self, waker: &Waker) -> Poll<Result<T, Canceled>> {
         // Check to see if some data has arrived. If it hasn't then we need to
         // block our task.
         //
@@ -260,7 +260,7 @@ impl<T> Inner<T> {
         let done = if self.complete.load(SeqCst) {
             true
         } else {
-            let task = lw.clone().into_waker();
+            let task = waker.clone();
             match self.rx_task.try_lock() {
                 Some(mut slot) => { *slot = Some(task); false },
                 None => true,
@@ -348,8 +348,8 @@ impl<T> Sender<T> {
     /// alive and may be able to receive a message if sent. The current task,
     /// however, is scheduled to receive a notification if the corresponding
     /// `Receiver` goes away.
-    pub fn poll_cancel(&mut self, lw: &LocalWaker) -> Poll<()> {
-        self.inner.poll_cancel(lw)
+    pub fn poll_cancel(&mut self, waker: &Waker) -> Poll<()> {
+        self.inner.poll_cancel(waker)
     }
 
     /// Tests to see whether this `Sender`'s corresponding `Receiver`
@@ -416,9 +416,9 @@ impl<T> Future for Receiver<T> {
 
     fn poll(
         self: Pin<&mut Self>,
-        lw: &LocalWaker,
+        waker: &Waker,
     ) -> Poll<Result<T, Canceled>> {
-        self.inner.recv(lw)
+        self.inner.recv(waker)
     }
 }
 

futures-channel/tests/mpsc.rs

@@ -6,7 +6,7 @@ use futures::future::{FutureExt, poll_fn};
 use futures::stream::{Stream, StreamExt};
 use futures::sink::{Sink, SinkExt};
 use futures::task::Poll;
-use futures_test::task::noop_local_waker_ref;
+use futures_test::task::noop_waker_ref;
 use pin_utils::pin_mut;
 use std::sync::{Arc, Mutex};
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -304,7 +304,7 @@ fn stress_receiver_multi_task_bounded_hard() {
                     } else {
                         // Just poll
                         let n = n.clone();
-                        match rx.poll_next_unpin(noop_local_waker_ref()) {
+                        match rx.poll_next_unpin(noop_waker_ref()) {
                             Poll::Ready(Some(_)) => {
                                 n.fetch_add(1, Ordering::Relaxed);
                             }