On "Async Streams" in Rust
Tags: rust, languagedesign, async
Published: 2019-04-15 20:36:10

---
title: On "Async Streams" in Rust
published: true
tags: rust,language-design,async
---

Context: [For await loops](https://internals.rust-lang.org/t/for-await-loops/9819?u=cad97) discussion started by @withoutboats on Internals.

# Step 1: What is a `Stream`?

I found the most clear consise explanation comes from @withoutboats again [on a different Internals thread](https://internals.rust-lang.org/t/pre-rfc-await-generators-directly/7202/10?u=cad97):

```
            | Evaluated immediately     | Evaluated asynchronously
------------------------------------------------------------------------
Return once | `fn() -> T`    (Function) | `async fn() -> T`    (Future)
Yield many  | `fn() yield T` (Iterator) | `async fn() yield T` (Stream)
```

- When you use a Function, you evaluate some routine to retrieve a value.
- When you use an Iterator, you evaluate some routine to retrieve the next value, multiple times.
- When you use a Future, you await some routine to retrieve a value.
- When you use a Stream, you await some routine to retrieve the next value, multiple times.

# Step 2: How do we use a `Stream`?

To determine this, let's look at the other points in the matrix.

## Function

Trivial: `function()`.

## Iterator

A for loop: `for item in iterator()`.

This effectively desugars to code using the [`Iterator`](https://doc.rust-lang.org/stable/std/iter/trait.Iterator.html) trait (simplified):

```rust
let __iter = iterator();
while let Some(item) = __iter.next() {
    // your block
}
```

## Future

The exact syntax to be used to await a [`Future`](https://doc.rust-lang.org/stable/std/future/trait.Future.html) is still being decided by the lang team. The only fairly decided thing is that it will involve the reserved keyword `await`. I'll use `await!(future)` in this document as it's the current unstable syntax and fully unambiguous.

This is extremely simplified, but awaiting a future expands to roughly the following in concept:

```rust
let __future = __stack_pin!(future);
loop {
    match __future.poll(__async_context_waker!()) {
        Poll::Ready(value) => break value,
        Poll::Pending => __async_yield_execution!(),
    }
}
```

## Stream

So, how are [`Stream`](https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.14/futures/prelude/trait.Stream.html)s used _before_ syntax sugar? This is important to understand the semantics we're trying to represent.

We have `Iterator`, but instead of the function `Iterator::next`, we have the future `Stream::poll_next`.

```rust
let __stream = __stack_pin!(stream);
while let Some(item) = await!(__future_from_poll!(__stream::poll_next)) {
    // your block
}
```

This still hides some details in the `await!(__future_from_poll!())`, so lets inline that:

```rust
let __stream = __stack_pin!(stream);
while let Some(item) = loop {
    match __stream.poll_next(__async_context_waker!()) {
        Poll::Ready(value) => break value,
        Poll::Pending => __async_yield_execution!(),
    }
} {
    // your block
}
```

_Wow_, everything makes some amount of sense! But how do we actually use this, because this is a lot of boilerplate to use what seems like a useful data type.

# Step 3: What Not

The logical operation happening here is awaiting the next value from the stream, for every element in the stream: a combination of iterators (`for item in iterator`) and futures (`await!(future)`).

The "obvious" solution would be `async for item in stream`. But this is the wrong direction: `async` is the enabler, but `await` is how you poll a future to completion.

Another enticing solution is to make `for` just work with `Stream`s and do the equivalent of `impl<I: Iterator> Stream for I`. But we made (are making) the await operation explicit on futures for good reason, and shouldn't lose that for streams, if for no other reason than it would be inconsistent.

I've seen in multiple places the suggestion to use `for item in await!(stream)`. Hopefully, the above expansions illustrate why that doesn't work semantically. This would instead be awaiting a `Future<Item=Iterator>`.

# Step 4: Possibilities

I personally think that `await for item in stream` reads perfectly for what the semantics are. Those semantics are that we `await` "something" to get each item out of the stream, or in other words, we "(a)wait for each item in the stream". There's also no possibility of being confused with `await!(for item in stream {})`, as `for` loops don't (and can't) return a value.

@withoutboats also suggests that we could make `await` a pattern, giving us `for await item in stream`. I don't hate it, but there's a problem with this.

`await` as a pattern (here I'll still use `await!(pattern)` for clarity) I would expect to be the same as awaiting the rhs of the pattern. Or in code, the following two statements would be equivalent:

```
let value = await!(future);
let await!(value) = future;
```

This was actually proposed previously in the venerable `async`/`await` bikeshedding thread.

The problem is that this now suggests that `Stream<Item=T>` is `Iterator<Item=Future<Item=T>>`. Maybe some streams could meet this type. But the current definition of `Stream` is closer to a `StreamingIterator<Item=Future<Item=T>>`, as the `poll_next` requires a pin mut ref to the stream. This borrowing is required for many real streams. (I expect `Stream` to be abused for `StreamingIterator` if it works for such and we don't have `StreamingIterator` yet.)

This could be made to work fairly simply: make `for` work with `StreamingIterator` with a fallback for `Iterator` behavior, and let streams be used by the above composition of `StreamingIterator` and `Future`.

But this still lacks the "`await?`" behavior that many real futures will want (as most async is so because of IO). And even if we add `?` to patterns as well, this still means we have two ways of doing the same await operation: expr position (for chaining, which was a focus of the bikeshedding thread) and pat position.

# Step 5: Conclusions

I think that `await for` is the best of the currently proposed syntaxes, at least while awaiting on futures includes the keyword `await`.

In any case, this is a difficult choice with many tradeoffs, and one that's even harder to make without knowing what the final `await` syntax is going to be. But it's one that needs to be at least considered when stabilizing `await`.