Update with review comments

rust-lang · May 25, 2018 · f57acc7 · f57acc7
1 parent 108fc90
commit f57acc7
Show file tree

Hide file tree

Showing 38 changed files with 135 additions and 76 deletions.
diff --git a/futures-core/src/poll.rs b/futures-core/src/poll.rs
@@ -2,7 +2,7 @@
 ///
 /// This macro bakes in propagation of `Pending` signals by returning early.
 #[macro_export]
-macro_rules! try_ready {
+macro_rules! ready {
     ($e:expr) => (match $e {
         $crate::Poll::Ready(t) => t,
         $crate::Poll::Pending => return $crate::Poll::Pending,

diff --git a/futures-util/src/future/flatten_sink.rs b/futures-util/src/future/flatten_sink.rs
@@ -66,7 +66,7 @@ impl<F> Sink for FlattenSink<F> where F: Future, <F as Future>::Item: Sink<SinkE
 
     fn poll_close(&mut self, cx: &mut task::Context) -> Result<Async<()>, Self::SinkError> {
         if let State::Ready(ref mut s) = self.st {
-            try_ready!(s.poll_close(cx));
+            ready!(s.poll_close(cx));
         }
         self.st = State::Closed;
         return Ok(Async::Ready(()));

diff --git a/futures-util/src/io/lines.rs b/futures-util/src/io/lines.rs
@@ -46,7 +46,7 @@ impl<A> Stream for Lines<A>
     type Error = io::Error;
 
     fn poll(&mut self, cx: &mut task::Context) -> Poll<Option<String>, io::Error> {
-        let n = try_ready!(self.io.read_line(&mut self.line));
+        let n = ready!(self.io.read_line(&mut self.line));
         if n == 0 && self.line.len() == 0 {
             return Ok(None.into())
         }

diff --git a/futures-util/src/io/read_until.rs b/futures-util/src/io/read_until.rs
@@ -62,7 +62,7 @@ impl<A> Future for ReadUntil<A>
                 // and just return it, as we are finished.
                 // If we hit "would block" then all the read data so far
                 // is in our buffer, and otherwise we propagate errors.
-                try_ready!(a.read_until(byte, buf));
+                ready!(a.read_until(byte, buf));
             },
             State::Empty => panic!("poll ReadUntil after it's done"),
         }

diff --git a/futures-util/src/sink/buffer.rs b/futures-util/src/sink/buffer.rs
@@ -44,11 +44,11 @@ impl<S: Sink> Buffer<S> {
     }
 
     fn try_empty_buffer(&mut self, cx: &mut task::Context) -> Poll<(), S::SinkError> {
-        try_ready!(self.sink.poll_ready(cx));
+        ready!(self.sink.poll_ready(cx));
         while let Some(item) = self.buf.pop_front() {
             self.sink.start_send(item)?;
             if self.buf.len() != 0 {
-                try_ready!(self.sink.poll_ready(cx));
+                ready!(self.sink.poll_ready(cx));
             }
         }
         Ok(Async::Ready(()))
@@ -93,13 +93,13 @@ impl<S: Sink> Sink for Buffer<S> {
     }
 
     fn poll_flush(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
-        try_ready!(self.try_empty_buffer(cx));
+        ready!(self.try_empty_buffer(cx));
         debug_assert!(self.buf.is_empty());
         self.sink.poll_flush(cx)
     }
 
     fn poll_close(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
-        try_ready!(self.try_empty_buffer(cx));
+        ready!(self.try_empty_buffer(cx));
         debug_assert!(self.buf.is_empty());
         self.sink.poll_close(cx)
     }

diff --git a/futures-util/src/sink/send.rs b/futures-util/src/sink/send.rs
@@ -61,7 +61,7 @@ impl<S: Sink> Future for Send<S> {
 
         // we're done sending the item, but want to block on flushing the
         // sink
-        try_ready!(self.sink_mut().poll_flush(cx));
+        ready!(self.sink_mut().poll_flush(cx));
 
         // now everything's emptied, so return the sink for further use
         Ok(Async::Ready(self.take_sink()))

diff --git a/futures-util/src/sink/send_all.rs b/futures-util/src/sink/send_all.rs
@@ -75,18 +75,18 @@ impl<T, U> Future for SendAll<T, U>
         // If we've got an item buffered already, we need to write it to the
         // sink before we can do anything else
         if let Some(item) = self.buffered.take() {
-            try_ready!(self.try_start_send(cx, item))
+            ready!(self.try_start_send(cx, item))
         }
 
         loop {
             match self.stream_mut().poll_next(cx)? {
-                Async::Ready(Some(item)) => try_ready!(self.try_start_send(cx, item)),
+                Async::Ready(Some(item)) => ready!(self.try_start_send(cx, item)),
                 Async::Ready(None) => {
-                    try_ready!(self.sink_mut().poll_flush(cx));
+                    ready!(self.sink_mut().poll_flush(cx));
                     return Ok(Async::Ready(self.take_result()))
                 }
                 Async::Pending => {
-                    try_ready!(self.sink_mut().poll_flush(cx));
+                    ready!(self.sink_mut().poll_flush(cx));
                     return Ok(Async::Pending)
                 }
             }

diff --git a/futures-util/src/sink/with.rs b/futures-util/src/sink/with.rs
@@ -143,12 +143,12 @@ impl<S, U, Fut, F> Sink for With<S, U, Fut, F>
     }
 
     fn poll_flush(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
-        try_ready!(self.poll(cx));
+        ready!(self.poll(cx));
         self.sink.poll_flush(cx).map_err(Into::into)
     }
 
     fn poll_close(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
-        try_ready!(self.poll(cx));
+        ready!(self.poll(cx));
         self.sink.poll_close(cx).map_err(Into::into)
     }
 }
diff --git a/futures-util/src/sink/with_flat_map.rs b/futures-util/src/sink/with_flat_map.rs
@@ -71,7 +71,7 @@ where
             }
         }
         if let Some(mut stream) = self.stream.take() {
-            while let Some(x) = try_ready!(stream.poll_next(cx)) {
+            while let Some(x) = ready!(stream.poll_next(cx)) {
                 match self.sink.poll_ready(cx)? {
                     Async::Ready(()) => self.sink.start_send(x)?,
                     Async::Pending => {

diff --git a/futures-util/src/stream/buffer_unordered.rs b/futures-util/src/stream/buffer_unordered.rs
@@ -96,7 +96,7 @@ impl<S> Stream for BufferUnordered<S>
         }
 
         // Try polling a new future
-        if let Some(val) = try_ready!(self.queue.poll_next(cx)) {
+        if let Some(val) = ready!(self.queue.poll_next(cx)) {
             return Ok(Async::Ready(Some(val)));
         }
 

diff --git a/futures-util/src/stream/buffered.rs b/futures-util/src/stream/buffered.rs
@@ -106,7 +106,7 @@ impl<S> Stream for Buffered<S>
         }
 
         // Try polling a new future
-        if let Some(val) = try_ready!(self.queue.poll_next(cx)) {
+        if let Some(val) = ready!(self.queue.poll_next(cx)) {
             return Ok(Async::Ready(Some(val)));
         }
 

diff --git a/futures-util/src/stream/chain.rs b/futures-util/src/stream/chain.rs
@@ -38,7 +38,7 @@ impl<S1, S2> Stream for Chain<S1, S2>
 
     fn poll_next(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<Option<Self::Item>> {
         if let Some(first) = self.first().as_pin_mut() {
-            if let Some(item) = try_ready!(first.poll_next(cx)) {
+            if let Some(item) = ready!(first.poll_next(cx)) {
                 return Poll::Ready(Some(item))
             }
         }

diff --git a/futures-util/src/stream/chunks.rs b/futures-util/src/stream/chunks.rs
@@ -1,4 +1,5 @@
 use std::mem::{self, PinMut};
+use std::marker::Unpin;
 use std::prelude::v1::*;
 
 use futures_core::{Poll, Stream};
@@ -76,6 +77,8 @@ impl<S> Chunks<S> where S: Stream {
     unsafe_pinned!(stream -> Fuse<S>);
 }
 
+unsafe impl<S: Unpin + Stream> Unpin for Chunks<S> {}
+
 impl<S> Stream for Chunks<S>
     where S: Stream
 {
@@ -84,7 +87,7 @@ impl<S> Stream for Chunks<S>
     fn poll_next(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<Option<Self::Item>> {
         let cap = self.items.capacity();
         loop {
-            match try_ready!(self.stream().poll_next(cx)) {
+            match ready!(self.stream().poll_next(cx)) {
                 // Push the item into the buffer and check whether it is full.
                 // If so, replace our buffer with a new and empty one and return
                 // the full one.

diff --git a/futures-util/src/stream/collect.rs b/futures-util/src/stream/collect.rs
@@ -1,5 +1,5 @@
 use std::prelude::v1::*;
-
+use std::marker::Unpin;
 use std::mem::{self, PinMut};
 
 use futures_core::{Future, Poll, Stream};
@@ -33,14 +33,16 @@ impl<S: Stream, C: Default> Collect<S, C> {
     unsafe_unpinned!(items -> C);
 }
 
+unsafe impl<S: Unpin + Stream, C> Unpin for Collect<S, C> {}
+
 impl<S, C> Future for Collect<S, C>
     where S: Stream, C: Default + Extend<S:: Item>
 {
     type Output = C;
 
     fn poll(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<C> {
         loop {
-            match try_ready!(self.stream().poll_next(cx)) {
+            match ready!(self.stream().poll_next(cx)) {
                 Some(e) => self.items().extend(Some(e)),
                 None => return Poll::Ready(self.finish()),
             }

diff --git a/futures-util/src/stream/concat.rs b/futures-util/src/stream/concat.rs
@@ -1,6 +1,7 @@
 use core::mem::PinMut;
 use core::fmt::{Debug, Formatter, Result as FmtResult};
 use core::default::Default;
+use core::marker::Unpin;
 
 use futures_core::{Future, Poll, Stream};
 use futures_core::task;
@@ -41,6 +42,8 @@ impl<S: Stream> Concat<S> {
     unsafe_unpinned!(accum -> Option<S::Item>);
 }
 
+unsafe impl<S: Stream + Unpin> Unpin for Concat<S> {}
+
 impl<S> Future for Concat<S>
     where S: Stream,
           S::Item: Extend<<<S as Stream>::Item as IntoIterator>::Item> + IntoIterator + Default,

diff --git a/futures-util/src/stream/empty.rs b/futures-util/src/stream/empty.rs
@@ -1,5 +1,5 @@
 use core::mem::PinMut;
-use core::marker::PhantomData;
+use core::marker::{Unpin, PhantomData};
 
 use futures_core::{Stream, Poll};
 use futures_core::task;
@@ -22,6 +22,8 @@ pub fn empty<T>() -> Empty<T> {
     }
 }
 
+unsafe impl<T> Unpin for Empty<T> {}
+
 impl<T> Stream for Empty<T> {
     type Item = T;
 

diff --git a/futures-util/src/stream/filter.rs b/futures-util/src/stream/filter.rs
@@ -1,4 +1,5 @@
 use core::mem::PinMut;
+use core::marker::Unpin;
 
 use {PinMutExt, OptionExt};
 
@@ -69,6 +70,12 @@ impl<S, R, P> Filter<S, R, P>
     unsafe_unpinned!(pending_item -> Option<S::Item>);
 }
 
+unsafe impl<S, R, P> Unpin for Filter<S, R, P>
+    where S: Stream + Unpin,
+          P: FnMut(&S::Item) -> R,
+          R: Future<Output = bool> + Unpin,
+{}
+
 /* TODO
 // Forwarding impl of Sink from the underlying stream
 impl<S, R, P> Sink for Filter<S, R, P>
@@ -94,7 +101,7 @@ impl<S, R, P> Stream for Filter<S, R, P>
     fn poll_next(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<Option<S::Item>> {
         loop {
             if self.pending_fut().as_pin_mut().is_none() {
-                let item = match try_ready!(self.stream().poll_next(cx)) {
+                let item = match ready!(self.stream().poll_next(cx)) {
                     Some(e) => e,
                     None => return Poll::Ready(None),
                 };
@@ -103,7 +110,7 @@ impl<S, R, P> Stream for Filter<S, R, P>
                 *self.pending_item() = Some(item);
             }
 
-            let yield_item = try_ready!(self.pending_fut().as_pin_mut().unwrap().poll(cx));
+            let yield_item = ready!(self.pending_fut().as_pin_mut().unwrap().poll(cx));
             self.pending_fut().assign(None);
             let item = self.pending_item().take().unwrap();
 

diff --git a/futures-util/src/stream/filter_map.rs b/futures-util/src/stream/filter_map.rs
@@ -1,4 +1,5 @@
 use core::mem::PinMut;
+use core::marker::Unpin;
 
 use {PinMutExt, OptionExt};
 
@@ -66,6 +67,12 @@ impl<S, R, F> FilterMap<S, R, F>
     unsafe_pinned!(pending -> Option<R>);
 }
 
+unsafe impl<S, R, F> Unpin for FilterMap<S, R, F>
+    where S: Stream + Unpin,
+          F: FnMut(S::Item) -> R,
+          R: Future + Unpin,
+{}
+
 /* TODO
 // Forwarding impl of Sink from the underlying stream
 impl<S, R, F> Sink for FilterMap<S, R, F>
@@ -90,15 +97,15 @@ impl<S, R, F, B> Stream for FilterMap<S, R, F>
     fn poll_next(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<Option<B>> {
         loop {
             if self.pending().as_pin_mut().is_none() {
-                let item = match try_ready!(self.stream().poll_next(cx)) {
+                let item = match ready!(self.stream().poll_next(cx)) {
                     Some(e) => e,
                     None => return Poll::Ready(None),
                 };
                 let fut = (self.f())(item);
                 self.pending().assign(Some(fut));
             }
 
-            let item = try_ready!(self.pending().as_pin_mut().unwrap().poll(cx));
+            let item = ready!(self.pending().as_pin_mut().unwrap().poll(cx));
             self.pending().assign(None);
             if item.is_some() {
                 return Poll::Ready(item);

diff --git a/futures-util/src/stream/flatten.rs b/futures-util/src/stream/flatten.rs
@@ -77,12 +77,12 @@ impl<S> Stream for Flatten<S>
     fn poll_next(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<Option<Self::Item>> {
         loop {
             if self.next().as_pin_mut().is_none() {
-                match try_ready!(self.stream().poll_next(cx)) {
+                match ready!(self.stream().poll_next(cx)) {
                     Some(e) => self.next().assign(Some(e)),
                     None => return Poll::Ready(None),
                 }
             }
-            let item = try_ready!(self.next().as_pin_mut().unwrap().poll_next(cx));
+            let item = ready!(self.next().as_pin_mut().unwrap().poll_next(cx));
             if item.is_some() {
                 return Poll::Ready(item);
             } else {

diff --git a/futures-util/src/stream/fold.rs b/futures-util/src/stream/fold.rs
@@ -1,4 +1,5 @@
 use core::mem::PinMut;
+use core::marker::Unpin;
 
 use {PinMutExt, OptionExt};
 
@@ -37,6 +38,8 @@ impl<S, Fut, T, F> Fold<S, Fut, T, F> {
     unsafe_pinned!(fut -> Option<Fut>);
 }
 
+unsafe impl<S: Unpin, Fut: Unpin, T, F> Unpin for Fold<S, Fut, T, F> {}
+
 impl<S, Fut, T, F> Future for Fold<S, Fut, T, F>
     where S: Stream,
           F: FnMut(T, S::Item) -> Fut,
@@ -46,21 +49,23 @@ impl<S, Fut, T, F> Future for Fold<S, Fut, T, F>
 
     fn poll(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<T> {
         loop {
-            if self.accum().is_some() {
-                let item = try_ready!(self.stream().poll_next(cx));
-                let accum = self.accum().take().unwrap();
-
-                if let Some(e) = item {
-                    let fut = (self.f())(accum, e);
-                    self.fut().assign(Some(fut));
-                } else {
-                    return Poll::Ready(accum)
-                }
-            } else {
-                let accum = try_ready!(self.fut().as_pin_mut().unwrap().poll(cx));
+            // we're currently processing a future to produce a new accum value
+            if self.accum().is_none() {
+                let accum = ready!(self.fut().as_pin_mut().unwrap().poll(cx));
                 *self.accum() = Some(accum);
                 self.fut().assign(None);
             }
+
+            let item = ready!(self.stream().poll_next(cx));
+            let accum = self.accum().take()
+                .expect("Fold polled after completion");
+
+            if let Some(e) = item {
+                let fut = (self.f())(accum, e);
+                self.fut().assign(Some(fut));
+            } else {
+                return Poll::Ready(accum)
+            }
         }
     }
 }
diff --git a/futures-util/src/stream/for_each.rs b/futures-util/src/stream/for_each.rs
@@ -1,4 +1,5 @@
 use core::mem::PinMut;
+use core::marker::Unpin;
 
 use {PinMutExt, OptionExt};
 
@@ -35,6 +36,8 @@ impl<S, U, F> ForEach<S, U, F> {
     unsafe_pinned!(fut -> Option<U>);
 }
 
+unsafe impl<S, U, F> Unpin for ForEach<S, U, F> {}
+
 impl<S, U, F> Future for ForEach<S, U, F>
     where S: Stream,
           F: FnMut(S::Item) -> U,
@@ -45,11 +48,11 @@ impl<S, U, F> Future for ForEach<S, U, F>
     fn poll(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<()> {
         loop {
             if let Some(fut) = self.fut().as_pin_mut() {
-                try_ready!(fut.poll(cx));
+                ready!(fut.poll(cx));
             }
             self.fut().assign(None);
 
-            match try_ready!(self.stream().poll_next(cx)) {
+            match ready!(self.stream().poll_next(cx)) {
                 Some(e) => {
                     let fut = (self.f())(e);
                     self.fut().assign(Some(fut));