Tracking Issue for #![feature(async_iterator)]

View all comments

This is a tracking issue for the RFC "2996" (rust-lang/rfcs#2996).
The feature gate for the issue is #![feature(async_iterator)].

About tracking issues

Tracking issues are used to record the overall progress of implementation.
They are also used as hubs connecting to other relevant issues, e.g., bugs or open design questions.
A tracking issue is however not meant for large scale discussion, questions, or bug reports about a feature.
Instead, open a dedicated issue for the specific matter and add the relevant feature gate label.

Steps

Implement the RFC (cc @rust-lang/XXX -- can anyone write up mentoring
instructions?)
Adjust documentation (see instructions on rustc-dev-guide)
Stabilization PR (see instructions on rustc-dev-guide)

Unresolved Questions

Clarify the panic behavior of Stream and Iterator Add core::stream::Stream #79023 (comment)
- Add a panic section to Iterator, clarifying panic behavior. The panic behavior between Stream and Iterator should be consistent.
Restore Stream::next. This was removed from the RFC because it prevents dynamic dispatch, and subsequently removed from the implementation. This should be resolved before stabilizing.
- Investigate whether we can move the trait from fn poll_next to async fn next once we can use async in traits.
Investigate as part of keyword-generics whether we can merge Iterator and AsyncIterator into a single trait which is generic over "asyncness".
Should we name this API AsyncIterator instead? Tracking Issue for #![feature(async_iterator)] #79024 (comment)
Should size_hint be async? Would there ever be a use-case for that?
poll_progress?

Implementation history

Rollup merge of rust-lang#79023 - yoshuawuyts:stream, r=KodrAus

Is there a plan to include a core::stream::StreamExt (similar to the one in the futures crate) as well?

Is there a plan to include a core::stream::StreamExt (similar to the one in the futures crate) as well?

The plan is to add methods directly onto Stream much like methods exist on Iterator, but we want to do so in a way that won't cause ambiguities with ecosystem-defined methods in order to not accidentally break existing codebases when upgrading to newer Rust versions.

Some discussion on Zulip raised the question of naming. With full acknowledgement to the fact that the name Stream has a long history in the async ecosystem, multiple people observed that something like AsyncIterator or similar might be much more evocative for new users. Such a name would allow people to map their existing understanding of iterators. "I understand Iterator, and I understand async, and this is an async version of Iterator".

As a new, inexperienced user, I find AsyncIterator way more foreign than Stream, to be honest... perhaps I'm not too involved in compiler discussions and I don't see how the jargon clicks together though :-S ... also, blogposts are already being written about Stream, so changing the name mid-flight will only breed confusion, I reckon?

Prior art on "iterator" and "async iterator" naming schemes in other languages:

JavaScript: Symbol.Iterator, Symbol.AsyncIterator
C#: IEnumerable, IAsyncEnumerable
Python: __iter__, __aiter__
Swift: Sequence, AsyncSequence

Context: We're prototyping some simple CLI app functionality. We're following the mini-redis example code where we can.

We've bumped out head on streams, and the need for the crates async-streams, and parallel-streams.

In our experience some *Iterator terminology would have helped clarify what streams are.

Given the need for the crates we cited, especially the async-streams RFC, we wonder if there isn't a need for:

Iterator
ConcurrentIterator
ParallelIterator

Or is the intention that async-stream and parallel-stream crate efforts all able to converge into a stream that covers the concurrent and parallel use cases?

Rayon provides a whole ecosystem of parallel iterators on top of a work-stealing threadpool, and is currently the de facto Rust standard for parallel iteration. But I don't foresee a parallel iterator trait getting into the Rust standard library anytime soon.

Streams are supposed to cover the use case of concurrent iterators only (per my understanding). Hopefully once streams are stabilized into the Rust standard library we can have some syntax to use them in concurrent for loops just like we currently use synchronous iterators in for loops. However, there are still some issues to work out like what to do if a a stream panics or is dropped. Settling on a name (Stream vs AsyncIterator vs ConcurrentIterator) will be the easy part! 😜

Unfortunately, it's difficult to combine parallel and concurrent iteration at the moment. This would require Rayon to support a way to move tasks from worker threads to an async executor thread, or for an async executor like Tokio to support parallel thread pools in a more sophisticated way than tokio::task::spawn_blocking(). Until that happens, most programmers try to get all their data into memory first on a concurrent executor-driven threadpool and then offload the synchronous computation to a parallel threadpool (e.g. managed by Rayon).

multiple people observed that something like AsyncIterator or similar might be much more evocative for new users. Such a name would allow people to map their existing understanding of iterators. "I understand Iterator, and I understand async, and this is an async version of Iterator".

I've filed a PR to the RFCs repo, updating the "streams" terminology to "async iterator" instead: rust-lang/rfcs#3208.

why did .next make it. where can i get more info on this ?

poll_next() requires the iterator to be pinned while async fn next() doesn't

This makes poll_next more powerful. A Stream like https://docs.rs/tokio-util/latest/tokio_util/codec/struct.FramedRead.html currently works with any ! Unpin AsyncRead type. If we moved to async fn next(), the AsyncRead would need to be Unpin.

Similar story is for https://docs.rs/futures-concurrency/latest/futures_concurrency/stream/trait.Merge.html which would now need to cache the futures of each async fn next(). Since they are all !Unpin, Merge is now forced to box these next futures

Similar story is for https://docs.rs/futures-concurrency/latest/futures_concurrency/stream/trait.Merge.html which would now need to cache the futures of each async fn next(). Since they are all !Unpin, Merge is now forced to box these next futures

The main issue here is that with poll_next(), the iterator itself must contain the async state machine. Whereas with async fn next(), the async state machine is transiently created by the iterator. Self-contained state machines are much easier to cache and poll in an arbitrary order, which is what Merge does. Managing transient state machines created by async functions is a pain point of async traits in general.

Similar story is for https://docs.rs/futures-concurrency/latest/futures_concurrency/stream/trait.Merge.html which would now need to cache the futures of each async fn next(). Since they are all !Unpin, Merge is now forced to box these next futures

I would be hesitant to take this as an inherent constraint. I'm fairly certain that with unsafe binders we can formulate an implementation of Merge based on async fn next which does not require any allocations. We'll need to play around a little with the traits and bounds, but I believe it should be possible.

Unsafe binders are going to be a great help for async traits regardless of their shape. They are the same feature required to e.g. make fn poll_next work with async closures, which is something we need for example for AsyncIterator::filter.

If we use async fn next, Merge will have to store the futures generated by each call to next. That means Merge will need to be pinned. I don't see how unsafe lifetime binders can get rid of the pinning requirement. The example of unsafe binders from the design doc also uses Pin<&mut Self>.

I believe there might be a miscommunication: I didn't mean to comment on the pinning requirement, I meant to say it shouldn't be necessary to allocate. While immovability may be inherent to self-references, allocation should not be.

Right. I suggested box-pinning because at present the async fn next(&mut self) does not present a pinning interface. That is unless we expect to implement

impl AsyncIterator for Pin<&mut Merge<...>> { ... }

otherwise we'd need to instead offer

trait AsyncIterator {
    async fn next(self: Pin<&mut Self>) -> Option<Self::Item>;
}

Tracking issues aren't meant for discussion of this kind. From above:

A tracking issue is however not meant for large scale discussion, questions, or bug reports about a feature.

If you want to discuss this generally, probably the right thing to do would be to open a thread in Zulip under #wg-async.

Just to get an idea on the scale of this and to see if I understood correctly:

Is this going to be the foundation for any async implementations of the current iterators on - for example - containers?

As I see, completion of this, should uncover a next step in the roadmap to implement things like Vec<T: Send + 'static>::iter_async()::map(A).await where map takes an async closure.

Is this also planned?

Unrelated to this trait specifically but this would throw up a question on if an async version of Option::map also has rights to exist as it's fundamentally the same idea of asynchronously mapping.

@withoutboats

After the Futurelock post came out in January, I went down kind of a rabbit hole collecting related deadlocks. The end result was another post: https://jacko.io/snooze.html. Several of the examples in there are related to streams. I came out wondering whether the Stream/AsyncIterator trait might need to change in a couple of ways to protect callers from these issues. Here's a condensed summary of the two categories of problems I'm worried about.

The first problem is "buffered" streams, that is FuturesOrdered/FuturesUnordered and the helper methods that wrap them. These are deadlock-prone if the futures they're buffering hold any locks across await points. For example (playground link):

async fn foo() {
    static LOCK: Mutex<()> = Mutex::const_new(());
    let _guard = LOCK.lock().await;
    sleep(Duration::from_millis(10)).await;
}

#[tokio::main]
async fn main() {
    futures::stream::iter([foo(), foo()])
        .buffered(2)
        .for_each(async |_| {
            foo().await; // Deadlock!
        })
        .await;
}

To make this work reliably in the presence of async locking, I think the stream returned by .buffered(_) needs to expose something like @withoutboats' poll_progress method, and the for_each future (and in general any stream combinator) needs to be honor-bound to call it. This would be an expansion of the basic "contract" for streams, in addition to the trait API.

The second problem is related to the .next() method on any stream, not just buffered ones. Consider this example, using the same foo function from above (playground link):

#[tokio::main]
async fn main() {
    let mut stream = pin!(stream::once(foo()));
    _ = timeout(Duration::from_millis(5), stream.next()).await;
    foo().await; // Deadlock!
}

The problem there is that if .next() is cancelled, it can suspend a stream (and any futures it might contain) at potentially any .await point, not necessarily a well-behaved "yield point" where suspension is expected. Unless .next() is paired with some sort of "uncancellable futures" feature, I worry this makes it too error-prone to recommend for general use. I realize that's a radical thing to suggest, but maybe most existing callers of .next() could be replaced with for await, or in general any helper that takes ownership of the stream. (for await would also presumably participate in any sort of poll_progress solution to the first problem.)

If these problems are worth a discussion, maybe I should open a separate issue thread, but I wanted to at least mention them here at the top level.

Edit: Examples with Mutex and sleep can feel too contrived, so here's a more "natural" example with OnceCell and reqwest and here's one with a bounded channel and no cancellation.

Edit again: I opened a thread on the wg-async Zulip, which is a better place for discussion than this tracking issue.

Edit August: I'm working on a draft RFC for this.

The things we're running into here also bear some resemblance to DirectService in tower from way back in the day. The discussion around its introduction and its eventual abandonment may also be of use here, even if it's not quite the same.

I've filed two RFCs to address the deadlocks above, rust-lang/rfcs#4019 regarding Future in general and rust-lang/rfcs#4020 regarding AsyncIterator specifically. The latter proposes adding poll_progress.

Navigation Menu

Repository navigation

Tracking Issue for #![feature(async_iterator)] #79024

About tracking issues

Steps

Unresolved Questions

Implementation history

23 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Tracking Issue for #![feature(async_iterator)] #79024

Description

About tracking issues

Steps

Unresolved Questions

Implementation history

Activity

kaimast commented on Feb 8, 2021

yoshuawuyts commented on Feb 9, 2021

joshtriplett commented on Apr 9, 2021

brainstorm commented on Apr 10, 2021

yoshuawuyts commented on Apr 15, 2021

bbros-dev commented on Aug 17, 2021

benkay86 commented on Aug 17, 2021

yoshuawuyts commented on Dec 14, 2021

noelzubin commented on Jan 5, 2022

23 remaining items

conradludgate commented on Nov 1, 2024

YuhanLiin commented on Nov 1, 2024

yoshuawuyts commented on Nov 3, 2024

YuhanLiin commented on Nov 4, 2024

yoshuawuyts commented on Nov 4, 2024

conradludgate commented on Nov 5, 2024

21aslade commented on Mar 15, 2025

traviscross commented on Mar 15, 2025

Kek5chen commented on Jul 5, 2025

oconnor663 commented on Apr 16, 2026

jonhoo commented on Apr 18, 2026

oconnor663 commented on Oct 20, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions