adds async stream rfc #2996

nellshamrell · 2020-09-29T16:57:52Z

Rendered

Pre-RFC Discussions

Draft 1
Draft 2
Draft 3
Draft 4

Signed-off-by: Nell Shamrell nellshamrell@gmail.com

Current FCP: #2996 (comment)

Status: merged

Signed-off-by: Nell Shamrell <nellshamrell@gmail.com>

text/0000-async-stream.md

Co-authored-by: Florian Gilcher <florian.gilcher@asquera.de>

Co-authored-by: Waffle Lapkin <waffle.lapkin@gmail.com>

text/0000-async-stream.md

Co-authored-by: kennytm <kennytm@gmail.com>

Co-authored-by: Taiki Endo <te316e89@gmail.com>

Signed-off-by: Nell Shamrell <nellshamrell@gmail.com>

nikomatsakis · 2020-09-30T16:50:14Z

We discussed this amongst the lang team at some point and concluded that this does not require @rust-lang/lang signoff. I've tagged it for @rust-lang/libs.

nikomatsakis · 2020-09-30T16:51:34Z

Also cc @rust-lang/wg-async-foundations

joshtriplett · 2020-09-30T19:47:56Z

👍 to adding try_next() if possible.

Also: could we please standardize a method for converting from Iterator to Stream? That may not be able to use IntoStream (because a blanket impl for Iterator would conflict with more specific impls people may want to write), but having a function that takes an impl Iterator<Item = T> and returns an impl Stream<Item = T> would be really helpful.

Signed-off-by: Nell Shamrell <nellshamrell@gmail.com>

text/0000-async-stream.md

yoshuawuyts · 2020-10-01T21:31:06Z

👍 to adding try_next() if possible.

@joshtriplett The conversation around whether to add Stream::next took quite a while to resolve, and was only added because poll_next by itself is not useful for end-users. The proposed try_next adapter has no counterpart in Iterator, is a minor helper to avoid writing .transpose(), and doesn't include functionality unique to async Rust. I would prefer if the scope of this PR would not be expanded beyond what's already been proposed so we don't need to find consensus on the design of further stream adapters in order to stabilize Stream.

Also: could we please standardize a method for converting from Iterator to Stream?

@joshtriplett async-std has stream::from_iter and futures-rs has stream::iter. I don't know if you think this should be mentioned in the RFC, but adding a function with this functionality would be straightforward once we expose Stream.

joshtriplett · 2020-10-01T21:42:12Z

@yoshuawuyts To clarify, neither of those comments was intended as a blocker to this RFC. I think it makes sense to add a minimal implementation of Stream, and after doing so, add additional helper functions like these to make it easier to work with.

nellshamrell · 2020-10-01T22:38:18Z

Information about converting an iterator to a stream has been added :)

LucioFranco · 2020-10-05T17:58:58Z

I finally got around to reading, this looks pretty good, I think including next is +1. I would really really like to see try_next, I think this is by far the most useful fn I have used for streams. I don't really understand the reason to not include it? I don't think there is much consensus needed for it, we've already agreed how to do next and try_next shouldn't be much more complicated as far as I can tell.

Another example of !Unpin streams are done via async-stream crate https://github.com/tokio-rs/async-stream/ which is usable on stable today.

Thanks a ton @nellshamrell and other for pushing this through 😄

ratmice · 2020-10-12T12:26:40Z

I'm not sure where to comment on this, but given that this RFC discusses LendStream, as a generalization, in theory at least there is a further generalization involving different lifetimes for the Lender, and the Item, with a lifetime bounds that 'lender: 'item. Which probably requires additional control mechanisms to regain ownership in the Lender.

With the idea that the single liftime LendingStream is a specialization of it where the lifetimes are the same.
I haven't thought enough about the extent to which the further generalization can be expressed currently though.
It seemed since the discussion of LendingStream as a generalization of Stream up it might/might not be worth thinking about this further generalization

l4l

Just some minor typos, please check it out, I'm not a native speaker there might be mistakes.
That's an amazing job by the way, really like reading the rfc. Hope it would be approved soon.

text/0000-async-stream.md

yoshuawuyts · 2020-10-13T08:49:08Z

I would really really like to see try_next [...] I don't really understand the reason to not include it?

@LucioFranco I covered that earlier in the thread #2996 (comment):

The proposed try_next adapter has no counterpart in Iterator, is a minor helper to avoid writing .transpose(), and doesn't include functionality unique to async Rust. I would prefer if the scope of this PR would not be expanded beyond what's already been proposed so we don't need to find consensus on the design of further stream adapters in order to stabilize Stream.

I see try_next as being desirable only as long as we don't have async iteration syntax, and will fall out of fashion the moment we do. I know you'll see this differently, and discussing the scope and priorities of stream adapters will make for an interesting conversation. But do you really want to push us to find consensus on that now?

We're both part of the Async Foundations WG, and we've had the privilege of being able to set the direction of this RFC. And that was a success; we have consensus on everything included here! I think now is the time to ensure what's in this RFC is accurate, but not seek to expand its scope with subjects we then newly need to find consensus on.

nikomatsakis · 2020-10-13T14:51:32Z

To ensure I understand, the role of try_next is to allow you to write code like this?

while let Some(foo) = stream.try_next().await? {
}

as opposed to this?

while let Some(foo) = stream.try_next().await {
    let foo = foo?;
}

Hmm, I'm torn. I agree with @yoshuawuyts that we should avoid increasing the scope of the RFC in general, which is intentionally targeted. But I also think that the point of including next was that it was something people would "immediately want" in practice, and I can imagine that try_next also fits in that bucket. I'm not sure how important it is that it would become less useful if/when async iteration syntax is added, given that we don't have a clear timeline on that. Is the primary objection that the method will not be as useful in the future or are there other concerns?

I would want to hold the line against other forms of "scope creep", the only reason I can see to consider try_next is that it feels very analogous to next.

nikomatsakis · 2020-10-13T14:56:17Z

Thinking on it more, I think what I'd prefer is to leave the RFC unaltered, land it, and consider try_next separately as a PR. It feels like a fairly standard "libs addition". The only difference is that, if we plan to make the futures crate redirect to core, before we actually do that we will need to have a clear idea of what methods we have, because any additional methods will have to be carefully coordinated.

LucioFranco · 2020-10-13T15:39:46Z

@yoshuawuyts sure, I am pretty sure I voiced my opinion during that period. It's also my fault for not pushing it harder, I got busy and that happens. I still don't follow your argument you referenced with iterators. I think Streams are not 1:1 with iterators but I rest my case.

@nikomatsakis right, I think most streams I have worked with return some sort of result so avoiding the first example in favor of the second is much nicer.

I think its fine to punt on this for now, just knowing that I think this is extremely useful. My question then is, what does that path look like to add try_next and what does that look like in conjunction with the futures crate? For example, I want to use this try_next and when/if we add it to std will it just replace the one on the futures crate? To me I see this method as one of the foundational ones for this trait and I think that path needs to be clear.

Nemo157 · 2020-10-13T16:41:56Z

The alternative to try_next is

while let Some(foo) = stream.next().await.transpose()? {
}

which is a pattern I've started using with fallible iterators as well, having to use a while let instead of a for loop is less annoying than having the let foo = foo?; line inside the loop.

Nemo157 · 2021-01-20T11:56:56Z

If streams can register themselves once at the beginning of a for_each and de-register at the end, the stream may be implemented considerably more efficiently.

for_each takes a closure returning a future which may "block" for as long as it likes, so it would need to pause the stream before running that future and unpause it again afterwards. Similarly as soon as you have a .then(_) in a stream chain that Then stream would have to pre-emptively pause the prior stream in case the future it runs for each item "blocks" for too long. I don't think there are many situations in which you could poll a stream multiple times without pausing it between each poll.

Thomasdezeeuw · 2021-01-20T13:20:58Z

Re: the use of the Stream trait when receiving from channels.

Maybe Stream is simply not the correct abstraction for it. If you need to know (for performance reasons) what receivers are actively attempting to receive a value maybe a short-lived Future like the Stream::next method (now removed) returned would be a better fit.

I realise that Stream could provide some nice syntax sugar similar to what for with Iterators, but the following also works fine.

while let Ok(msg) = channel.receive_next().await {
    // Handle message...
}

impl<M> Channel<M> {
    fn receive_next(&mut self) -> impl Future<Output = Result<M, RecvError>> { /* ... */ }
}

It creates short lived Futures allowing the implementation to know which receivers are actively trying to receive a value.

KodrAus · 2021-01-21T05:15:03Z

text/0000-async-stream.md

+consider `next`, which when called, returns a future which yields
+`Option<Item>`.
+
+The future returned by `next` will yield `Some(Item)` as long as there are


The future returned by next will yield Some(Item) as long as there are elements

How does this play with fuse considering the future exhausted after the first Poll::Ready? Is the idea that you can either rely on that implementation detail for next and continue to pull items using the same future or get a new one after the previous next resolved?

Hi @KodrAus - the answer on this one is I honestly don't know. I'm happy to include a note that this will need to be considered in the eventual RFC and implementation of the next method. Would that be sufficient, since this is in the Future Possibilities section?

KodrAus · 2021-01-21T05:20:36Z

text/0000-async-stream.md

+}
+```
+
+## Initial impls


I found myself wanting to see the contents of the More Usage Examples section here first after we defined our Counter to see what it's like to interact with on the consuming side.

Hi @KodrAus - I intentionally did not include the consuming side here since the next method has been moved to the "Future Possibilities" section. I felt including that here would introduce more confusion about what is and is not covered by this RFC.

KodrAus · 2021-01-21T05:25:00Z

@rust-lang/libs I've dropped a link to the FCP in the OP here, since it's gotten quite buried.

zesterer · 2021-01-21T13:17:44Z

Maybe Stream is simply not the correct abstraction for it. If you need to know (for performance reasons) what receivers are actively attempting to receive a value maybe a short-lived Future like the Stream::next method (now removed) returned would be a better fit.

Perhaps this is the case, but I think it would be a shame for Stream to not accommodate this common use-case when the cost of the features required to allow it to accommodate it is so low.

This also means that there is no longer a natural duality between Stream and Iterator as the RFC claims.

nellshamrell · 2021-01-26T02:52:01Z

Hello @rust-lang/libs :) For those of you who have not yet approved FCP in this comment - are there any concerns you have or is there anything blocking approving this for FCP?

nellshamrell · 2021-01-26T02:55:56Z

@zesterer thank you for your comments, they encouraged us to think even more carefully about the design of this feature. I am going to leave the design as it is now as a minimal viable implementation, but we have your concerns about channels recorded here for future reference. I have no doubt we will be opening more RFCs around Async streams in the near future and we will certainly keep your concerns in mind. Thank you again, the community input and context is vital to any RFC.

Add `core::stream::Stream` [[Tracking issue: rust-lang#79024](rust-lang#79024)] This patch adds the `core::stream` submodule and implements `core::stream::Stream` in accordance with [RFC2996](rust-lang/rfcs#2996). The RFC hasn't been merged yet, but as requested by the libs team in rust-lang/rfcs#2996 (comment) I'm filing this PR to get the ball rolling. ## Documentatation The docs in this PR have been adapted from [`std::iter`](https://doc.rust-lang.org/std/iter/index.html), [`async_std::stream`](https://docs.rs/async-std/1.7.0/async_std/stream/index.html), and [`futures::stream::Stream`](https://docs.rs/futures/0.3.8/futures/stream/trait.Stream.html). Once this PR lands my plan is to follow this up with PRs to add helper methods such as `stream::repeat` which can be used to document more of the concepts that are currently missing. That will allow us to cover concepts such as "infinite streams" and "laziness" in more depth. ## Feature gate The feature gate for `Stream` is `stream_trait`. This matches the `#[lang = "future_trait"]` attribute name. The intention is that only the APIs defined in RFC2996 will use this feature gate, with future additions such as `stream::repeat` using their own feature gates. This is so we can ensure a smooth path towards stabilizing the `Stream` trait without needing to stabilize all the APIs in `core::stream` at once. But also don't start expanding the API until _after_ stabilization, as was the case with `std::future`. __edit:__ the feature gate has been changed to `async_stream` to match the feature gate proposed in the RFC. ## Conclusion This PR introduces `core::stream::{Stream, Next}` and re-exports it from `std` as `std::stream::{Stream, Next}`. Landing `Stream` in the stdlib has been a mult-year process; and it's incredibly exciting for this to finally happen! --- r? `````@KodrAus````` cc/ `````@rust-lang/wg-async-foundations````` `````@rust-lang/libs`````

nellshamrell · 2021-02-04T19:54:42Z

Gentle ping to @rust-lang/libs - this only needs one more approval for FCP!

dtolnay · 2021-02-08T01:46:51Z

Approving on behalf of withoutboats, as per rust-lang/team#526.

rfcbot · 2021-02-08T01:56:46Z

🔔 This is now entering its final comment period, as per the review above. 🔔

rfcbot · 2021-02-18T01:58:29Z

The final comment period, with a disposition to merge, as per the review above, is now complete.

As the automated representative of the governance process, I would like to thank the author for their work and everyone else who contributed.

The RFC will be merged soon.

jplatte · 2021-03-22T14:36:58Z

FCP has been complete for a bit over a month now, is something else missing for this to be merged?

nikomatsakis · 2021-03-22T14:51:57Z

No! I thought I did the merge! Maybe I forgot to push?

nikomatsakis · 2021-03-22T14:57:59Z

Hmm, maybe I removed the merge commit or something. I'm not sure why github doesn't show this as merged but .. it is. Going to close!

jplatte · 2021-03-22T15:11:54Z

Looks like a rebase instead of a merge, and GitHub only "understands" command-line PR merging if the exact original commits are included (i.e. if it's a merge or fast-forward, not rebase or squash).

nikomatsakis · 2021-03-22T19:39:43Z

Yeah I guess I rebased without realizing it. Oh well!

pickfire · 2021-03-31T01:59:55Z

text/0000-async-stream.md

+
+Unfortunately, async methods in traits are not currently supported,
+and there [are a number of challenges to be
+resolved](https://rust-lang.github.io/wg-async-foundations/design_notes/async_fn_in_traits.html)


Broken link

pickfire · 2021-03-31T02:02:25Z

text/0000-async-stream.md

+with them.
+
+Unfortunately, the use of poll does mean that it is harder to write
+stream implementations. The long-term fix for this, discussed in the [Future possiblilities](future-possibilities) section, is dedicated [generator syntax].


Broken link

any chance you can open a PR with fixes?

Recently tokio got a first stable release and many libraries & applications already migrated to the newest version. This changes upgrades tokio version to 1.0.2: * Tokio renamed some of its features, e.g `rt-util` and `rt-core` now combined into `rt`. * `stream` feature got extracted to a separate crate [tokio-stream](https://docs.rs/tokio-stream/0.1.2/tokio_stream), waiting for eventual `Stream` landing to the Rust std library. [RFC](rust-lang/rfcs#2996) [TODO] Actix's integration test_actix_ws_integration test still fails due to, I guess, async rt is not being initialized. Apart from that, the tests are green. Please feel free to take over this effort.

adds async stream rfc

8152e62

Signed-off-by: Nell Shamrell <nellshamrell@gmail.com>

skade reviewed Sep 29, 2020

View reviewed changes

text/0000-async-stream.md Outdated Show resolved Hide resolved

text/0000-async-stream.md Outdated Show resolved Hide resolved

WaffleLapkin reviewed Sep 29, 2020

View reviewed changes

text/0000-async-stream.md Outdated Show resolved Hide resolved

nellshamrell and others added 3 commits September 29, 2020 10:31

Update text/0000-async-stream.md

16c4870

Co-authored-by: Florian Gilcher <florian.gilcher@asquera.de>

Update text/0000-async-stream.md

20a064b

Co-authored-by: Florian Gilcher <florian.gilcher@asquera.de>

Update text/0000-async-stream.md

5a3de0f

Co-authored-by: Waffle Lapkin <waffle.lapkin@gmail.com>

kennytm reviewed Sep 29, 2020

View reviewed changes

text/0000-async-stream.md Show resolved Hide resolved

text/0000-async-stream.md Outdated Show resolved Hide resolved

text/0000-async-stream.md Outdated Show resolved Hide resolved

taiki-e reviewed Sep 29, 2020

View reviewed changes

text/0000-async-stream.md Outdated Show resolved Hide resolved

nellshamrell and others added 3 commits September 29, 2020 16:33

Update text/0000-async-stream.md

a145be9

Co-authored-by: kennytm <kennytm@gmail.com>

Update text/0000-async-stream.md

f65f099

Co-authored-by: Taiki Endo <te316e89@gmail.com>

remove boilerplate questions

310b0b3

Signed-off-by: Nell Shamrell <nellshamrell@gmail.com>

nikomatsakis added the T-libs-api Relevant to the library API team, which will review and decide on the RFC. label Sep 30, 2020

nellshamrell added 2 commits September 30, 2020 22:06

adds more information about coverting an iterator into a stream

e0d66d4

Signed-off-by: Nell Shamrell <nellshamrell@gmail.com>

adds clarifying statement regarding !Unpin streams

7b84d1c

Signed-off-by: Nell Shamrell <nellshamrell@gmail.com>

Nemo157 reviewed Oct 1, 2020

View reviewed changes

text/0000-async-stream.md Outdated Show resolved Hide resolved

text/0000-async-stream.md Outdated Show resolved Hide resolved

joshtriplett mentioned this pull request Oct 5, 2020

async foundations rust-lang/lang-team#33

Closed

l4l reviewed Oct 12, 2020

View reviewed changes

bdash mentioned this pull request Jan 21, 2021

Update signal-hook-tokio for tokio v1.0 vorner/signal-hook#87

Merged

KodrAus reviewed Jan 21, 2021

View reviewed changes

rfcbot added final-comment-period Will be merged/postponed/closed in ~10 calendar days unless new substational objections are raised. and removed proposed-final-comment-period Currently awaiting signoff of all team members in order to enter the final comment period. labels Feb 8, 2021

samsartor mentioned this pull request Feb 15, 2021

Unified coroutines a.k.a. Generator resume arguments #2781

Open

rfcbot added finished-final-comment-period The final comment period is finished for this RFC. to-announce and removed final-comment-period Will be merged/postponed/closed in ~10 calendar days unless new substational objections are raised. labels Feb 18, 2021

nikomatsakis closed this Mar 22, 2021

pickfire reviewed Mar 31, 2021

View reviewed changes

couchand mentioned this pull request Oct 27, 2021

Fix broken links in RFC 2996 #3188

Merged

dmitrii-ubskii mentioned this pull request Feb 23, 2023

TypeDB Client Rust dependencies: async runtimes, tokio-stream vaticle/dependencies#423

Merged

adds async stream rfc #2996

adds async stream rfc #2996

Conversation

nellshamrell commented Sep 29, 2020 • edited by nikomatsakis

Pre-RFC Discussions

nikomatsakis commented Sep 30, 2020

nikomatsakis commented Sep 30, 2020

joshtriplett commented Sep 30, 2020

yoshuawuyts commented Oct 1, 2020 • edited

joshtriplett commented Oct 1, 2020

nellshamrell commented Oct 1, 2020

LucioFranco commented Oct 5, 2020 • edited

ratmice commented Oct 12, 2020

l4l left a comment

Choose a reason for hiding this comment

yoshuawuyts commented Oct 13, 2020

nikomatsakis commented Oct 13, 2020

nikomatsakis commented Oct 13, 2020

LucioFranco commented Oct 13, 2020

Nemo157 commented Oct 13, 2020

Nemo157 commented Jan 20, 2021

Thomasdezeeuw commented Jan 20, 2021

KodrAus Jan 21, 2021

Choose a reason for hiding this comment

nellshamrell Jan 26, 2021

Choose a reason for hiding this comment

KodrAus Jan 21, 2021

Choose a reason for hiding this comment

nellshamrell Jan 26, 2021

Choose a reason for hiding this comment

KodrAus commented Jan 21, 2021

zesterer commented Jan 21, 2021 • edited

nellshamrell commented Jan 26, 2021

nellshamrell commented Jan 26, 2021

nellshamrell commented Feb 4, 2021

dtolnay commented Feb 8, 2021

rfcbot commented Feb 8, 2021

rfcbot commented Feb 18, 2021

jplatte commented Mar 22, 2021

nikomatsakis commented Mar 22, 2021

nikomatsakis commented Mar 22, 2021

jplatte commented Mar 22, 2021

nikomatsakis commented Mar 22, 2021

pickfire Mar 31, 2021

Choose a reason for hiding this comment

pickfire Mar 31, 2021

Choose a reason for hiding this comment

nikomatsakis Apr 1, 2021

Choose a reason for hiding this comment

nellshamrell commented Sep 29, 2020 •

edited by nikomatsakis

yoshuawuyts commented Oct 1, 2020 •

edited

LucioFranco commented Oct 5, 2020 •

edited

zesterer commented Jan 21, 2021 •

edited