Bring stream combinators over to 0.3 #1022

aturon · 2018-05-22T19:19:31Z

As with the future combinators, this PR pushes through all the simple combinators, to explore patterns. I'll do follow-up PRs for the more complicated cases.

Here I've developed a new set of macros for easily adding accessor methods that work on PinMut<Self>, similar to @withoutboats's derive. For flat structs, this works very well, and safety is trivial to audit.

cramertj · 2018-05-22T19:28:01Z

futures-core/src/poll.rs

+///
+/// This macro bakes in propagation of `Pending` signals by returning early.
+#[macro_export]
+macro_rules! try_ready {


Heh-- I had added this as ready! in my Sink CL, with try_ready! being the one that included error handling. We definitely want both since we can't use ? anymore-- do you have a naming preference?

I'll switch to ready

cramertj · 2018-05-22T19:29:39Z

futures-util/src/pin_mut.rs

+    fn as_pin_mut(self) -> Option<PinMut<'a, T>>;
+}
+
+impl<'a, T> OptionExt<'a, T> for PinMut<'a, Option<T>> {


TODO: add to core (both this and assign)

Opened rust-lang/rust#50984

i'll rename to set to match your PR

cramertj · 2018-05-22T19:51:03Z

futures-util/src/stream/chunks.rs

@@ -73,54 +71,41 @@ impl<S> Chunks<S> where S: Stream {
    pub fn into_inner(self) -> S {
        self.stream.into_inner()
    }
+
+    unsafe_unpinned!(items ->  Vec<S::Item>);
+    unsafe_pinned!(stream -> Fuse<S>);


Can you add an unsafe impl<S: Unpin> Unpin for Chunks<S> {} so that the Unpin-edness won't depend on the types of the items?

cramertj · 2018-05-22T19:53:52Z

futures-util/src/stream/collect.rs

    }
+
+    unsafe_pinned!(stream -> S);
+    unsafe_unpinned!(items -> C);


unsafe impl<S: Unpin, C> Unpin for Collect<S, C> {}

cramertj · 2018-05-22T20:05:12Z

futures-util/src/stream/concat.rs

+// These methods are the only points of access going through `PinMut`
+impl<S: Stream> Concat<S> {
+    unsafe_pinned!(stream -> S);
+    unsafe_unpinned!(accum -> Option<S::Item>);


unsafe impl<S: Unpin> Unpin for Concat<S> {}

cramertj · 2018-05-22T20:19:41Z

futures-util/src/stream/fold.rs

                }
+            } else {
+                let accum = try_ready!(self.fut().as_pin_mut().unwrap().poll(cx));


Can you make this unwrap into expect("Fold polled after completion")?

cramertj · 2018-05-22T20:21:36Z

futures-util/src/stream/for_each.rs

+    unsafe_pinned!(stream -> S);
+    unsafe_unpinned!(f -> F);
+    unsafe_pinned!(fut -> Option<U>);
+}


unsafe impl<S: Unpin, U: Unpin, F> Unpin for ForEach<S, U, F> {}

cramertj · 2018-05-22T20:23:34Z

futures-util/src/stream/fuse.rs

@@ -14,6 +15,7 @@ pub struct Fuse<S> {
    done: bool,


Not new to this CL, but can you change this to use an Option (or create an issue to follow up and assign me)? That should allow for a more optimized representation, in addition to allowing the stream to drop before the Fuse is ready to be dropped.

cramertj · 2018-05-22T20:25:07Z

futures-util/src/stream/inspect.rs

@@ -45,8 +46,12 @@ impl<S: Stream, F> Inspect<S, F> {
    pub fn into_inner(self) -> S {
        self.stream
    }
+
+    unsafe_pinned!(stream -> S);
+    unsafe_unpinned!(inspect -> F);


unsafe impl<S: Stream + Unpin, F> Unpin for Inspect<S, F> {}

cramertj · 2018-05-22T20:26:05Z

futures-util/src/stream/map.rs

@@ -46,8 +47,12 @@ impl<S, F> Map<S, F> {
    pub fn into_inner(self) -> S {
        self.stream
    }
+
+    unsafe_pinned!(stream -> S);
+    unsafe_unpinned!(f -> F);


unsafe impl<S: Unpin, F> Unpin for Map<S, F> {}

cramertj · 2018-05-22T20:56:41Z

Can you also update StreamExt::next?

cramertj · 2018-05-22T20:58:11Z

futures-util/src/stream/zip.rs

+    unsafe_pinned!(stream1 -> Fuse<S1>);
+    unsafe_pinned!(stream2 -> Fuse<S2>);
+    unsafe_unpinned!(queued1 -> Option<S1::Item>);
+    unsafe_unpinned!(queued2 -> Option<S2::Item>);


unsafe impl<S1: Unpin, S2: Unpin> Unpin for Zip<S1, S2> {}

cramertj · 2018-05-22T20:59:14Z

futures-util/src/stream/unfold.rs

+impl<T, F, Fut> Unfold<T, F, Fut> {
+    unsafe_unpinned!(f -> F);
+    unsafe_unpinned!(state -> Option<T>);
+    unsafe_pinned!(fut -> Option<Fut>);


unsafe impl<T, F, Fut: Unpin> Unpin for Unfold<T, F, Fut> {}

cramertj · 2018-05-22T21:01:28Z

futures-util/src/stream/unfold.rs

@@ -55,11 +57,12 @@ use futures_core::task;
 /// ```
 pub fn unfold<T, F, Fut, It>(init: T, f: F) -> Unfold<T, F, Fut>
    where F: FnMut(T) -> Fut,


Now that borrowed futures are the way of the world, can we change this to use FnMut(&mut T) -> Fut where Fut yields an Option<It>? This matches the itertools::unfold method: https://docs.rs/itertools/0.7.8/itertools/fn.unfold.html

Edit: I'm not sure if that's actually possible to write-- definitely need to sort that out: for<'a> FnMut(&'a mut T) -> Fut::<'a> or something?

Edit2: we might be able to do it with a custom trait:

trait UnfoldFunc<'a, T, Res> { type Fut: Future<Output = Option<Res>> + 'a; fn next(&'a mut self, state: &'a mut T) -> Fut; }

and then bound by F: for<'a> UnfoldFunc<'a, T, Res>. I'm not sure how to get <F as UnfoldFunc<'a, T, Res>>::Fut stored in the same place as a <F as UnfoldFunc<'b, T, Res>>::Fut used to be -- seems like we'd need manual erasure of some kind. Maybe it's enough to know that it works for any given lifetime, so we know it must work for a particular lifetime of our choosing that we keep the same across all the invocations.

I made this nonsense work: cramertj@4bc297b

cramertj · 2018-05-22T21:28:44Z

futures-util/src/stream/then.rs

+impl<S, U, F> Then<S, U, F> {
+    unsafe_pinned!(stream -> S);
+    unsafe_pinned!(future -> Option<U>);
+    unsafe_unpinned!(f -> F);


unsafe impl<S: Unpin, U: Unpin, F> Unpin for Then<S, U, F> {}

cramertj · 2018-05-22T21:29:35Z

futures-util/src/stream/take_while.rs

+    unsafe_unpinned!(pred -> P);
+    unsafe_pinned!(pending_fut -> Option<R>);
+    unsafe_unpinned!(pending_item -> Option<S::Item>);
+    unsafe_unpinned!(done_taking -> bool);


unsafe impl<S: Unpin, R: Unpin, P> Unpin for TakeWhile<S, R, P> {}

cramertj · 2018-05-22T21:31:05Z

futures-util/src/stream/skip_while.rs

+    unsafe_unpinned!(pred -> P);
+    unsafe_pinned!(pending_fut -> Option<R>);
+    unsafe_unpinned!(pending_item -> Option<S::Item>);
+    unsafe_unpinned!(done_skipping -> bool);


unsafe impl<S: Unpin, R: Unpin, P> Unpin for SkipWhile<S, R, P> {}

cramertj · 2018-05-22T21:41:45Z

futures-util/src/stream/repeat.rs

    where T: Clone
 {
    item: T,
-    error: marker::PhantomData<E>,
 }


unsafe impl<T> Unpin for Repeat<T> {}

cramertj · 2018-05-22T22:43:20Z

futures-util/src/stream/peek.rs

            }
        }
    }
+
+    unsafe_pinned!(stream -> Fuse<S>);
+    unsafe_unpinned!(peeked -> Option<S::Item>);


unsafe impl<S: Unpin> Unpin for Peakable<S> {}

MajorBreakfast · 2018-05-24T11:42:40Z

@cramertj @aturon

Out of interest: In the futures RFC thread, your most recent approach is the impl Future for PinMut<..>approach. However, here i see many fn poll(mut self: PinMut<Self>, ..). Is this PR's implementation a stopgap solution/experiment? I see that you now have two branches: "0.3" and "0.3-PinFuture" Are you pursuing both approaches at the same time for the time being or did you run into a problem with the other approach?

aturon · 2018-05-24T19:52:05Z

@cramertj most review comments addressed. I left Fuse as-is, because a bunch of stuff depends on being able to extract out the underlying stream.

@MajorBreakfast we've been pursuing both in parallel, and yes, there are some blockers on the other side. I plan to update the RFC thread soon...

cramertj · 2018-05-25T18:47:23Z

Bring stream combinators over to 0.3

4ac2cb0

aturon requested a review from cramertj May 22, 2018 19:19

cramertj reviewed May 22, 2018

View reviewed changes

Update with review comments

1146a3e

cramertj merged commit f57acc7 into rust-lang:0.3 May 25, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Bring stream combinators over to 0.3 #1022

Bring stream combinators over to 0.3 #1022

aturon commented May 22, 2018

cramertj May 22, 2018

aturon May 23, 2018

cramertj May 22, 2018

cramertj May 23, 2018

aturon May 23, 2018

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018

cramertj commented May 22, 2018 •

edited

Loading

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018 •

edited

Loading

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018

cramertj May 22, 2018

MajorBreakfast commented May 24, 2018

aturon commented May 24, 2018 •

edited

Loading

cramertj commented May 25, 2018

Bring stream combinators over to 0.3 #1022

Bring stream combinators over to 0.3 #1022

Conversation

aturon commented May 22, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cramertj commented May 22, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cramertj May 22, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MajorBreakfast commented May 24, 2018

aturon commented May 24, 2018 • edited Loading

cramertj commented May 25, 2018

cramertj commented May 22, 2018 •

edited

Loading

cramertj May 22, 2018 •

edited

Loading

aturon commented May 24, 2018 •

edited

Loading