Why usage of async block in then makes my Stream Unpin?

Question

I'm new to Rust and I'm sorry if I'm using terms incorrect. Maybe my choice of words for question is incorrect.

I was playing with streams and I needed to have some delay between stream elements. So I wrote this:

use futures::stream;
use futures::StreamExt;
use tokio::time;

#[tokio::main]
async fn main() {
    let mut stream = stream::iter(0..1000).then(|x| async move {
        time::delay_for(std::time::Duration::from_millis(500)).await;
        x + 1
    });
    while let Some(x) = stream.next().await {
        println!("{:?}", x)
    }
}

I get a lot of compilation errors, but the most important errors are connected with pinning. Here they are:

error[E0277]: `std::future::from_generator::GenFuture<[static generator@src/main.rs:7:64: 10:6 x:_ _]>` cannot be unpinned
  --> src/main.rs:11:32
   |
11 |     while let Some(x) = stream.next().await {
   |                                ^^^^ within `futures_util::stream::stream::then::_::__Then<'_, futures_util::stream::iter::Iter>, impl core::future::future::Future, [closure@src/main.rs:7:49: 10:6]>`, the trait `std::marker::Unpin` is not implemented for `std::future::from_generator::GenFuture<[static generator@src/main.rs:7:64: 10:6 x:_ _]>`
   |
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `std::option::Option`
   = note: required because it appears within the type `futures_util::stream::stream::then::_::__Then<'_, futures_util::stream::iter::Iter>, impl core::future::future::Future, [closure@src/main.rs:7:49: 10:6]>`
   = note: required because of the requirements on the impl of `std::marker::Unpin` for `futures_util::stream::stream::then::Then>, impl core::future::future::Future, [closure@src/main.rs:7:49: 10:6]>`

error[E0277]: `std::future::from_generator::GenFuture<[static generator@src/main.rs:7:64: 10:6 x:_ _]>` cannot be unpinned
  --> src/main.rs:11:25
   |
11 |     while let Some(x) = stream.next().await {
   |                         ^^^^^^^^^^^^^^^^^^^ within `futures_util::stream::stream::then::_::__Then<'_, futures_util::stream::iter::Iter>, impl core::future::future::Future, [closure@src/main.rs:7:49: 10:6]>`, the trait `std::marker::Unpin` is not implemented for `std::future::from_generator::GenFuture<[static generator@src/main.rs:7:64: 10:6 x:_ _]>`
   |
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `std::option::Option`
   = note: required because it appears within the type `futures_util::stream::stream::then::_::__Then<'_, futures_util::stream::iter::Iter>, impl core::future::future::Future, [closure@src/main.rs:7:49: 10:6]>`
   = note: required because of the requirements on the impl of `std::marker::Unpin` for `futures_util::stream::stream::then::Then>, impl core::future::future::Future, [closure@src/main.rs:7:49: 10:6]>`
   = note: required because of the requirements on the impl of `core::future::future::Future` for `futures_util::stream::stream::next::Next<'_, futures_util::stream::stream::then::Then>, impl core::future::future::Future, [closure@src/main.rs:7:49: 10:6]>>`

If I change my code to this:

use futures::stream;
use futures::StreamExt;
use tokio::time;

#[tokio::main]
async fn main() {
    let mut stream = stream::iter(0..1000).then(|x|  {
        futures::future::ready(x + 1)
    });
    while let Some(x) = stream.next().await {
        println!("{:?}", x)
    }
}

Or to this:

use futures::stream;
use futures::StreamExt;
use tokio::time;

#[tokio::main]
async fn main() {
    stream::iter(0..1000)
        .then(|x| async move {
            time::delay_for(std::time::Duration::from_millis(500)).await;
            x + 1
        })
        .for_each(|x| async move { println!("{:?}", x) })
        .await;
}

It compiles.

I assume it has something to do with pinning and simultaneous usage of then combinator and while, but I can't wrap my head around it.

user_865874 · Answer

I think that the issue boils down to the fact that async blocks are not Unpin. It can be proven with this code:

fn check_unpin(_: F) { }

fn main() {
    check_unpin(async {});
}

that fails with the rather cryptic message:

error[E0277]: `std::future::from_generator::GenFuture<[static generator@src/main.rs:4:23: 4:25 _]>` cannot be unpinned
  --> src/main.rs:4:5
   |
1  | fn check_unpin(_: F) { }
   |                   ----- required by this bound in `check_unpin`
...
4  |     check_unpin(async {});
   |     ^^^^^^^^^^^ within `impl std::future::Future`, the trait `std::marker::Unpin` is not implemented for `std::future::from_generator::GenFuture<[static generator@src/main.rs:4:23: 4:25 _]>`
   |
   = note: required because it appears within the type `impl std::future::Future`

I reckon that GenFuture is the internal type that converts an async block into a impl Future.

Now back to your issue, the combinator then() returns an Unpin value if both the stream and the Future returned by the closure are Unpin (it is not totally clear in the documentation, but I've inferred that from the source code). The stream::iter is Unpin, but when you write |x| async move { x + 1} you are returning an async block that is not Unpin, thus your error.

If you use futures::future::ready(x + 1) it works simply because that Future implements Unpin.

If you use StreamExt::for_each it works because it does not require Self to be Unpin. It is not Unpin itself, but it does not matter because you are sending it up to tokio::main that pins everything internally before polling.

If you want your original code to work you just have to manually pin your stream (playground):

use futures::stream;
use futures::StreamExt;
use tokio::time;
use pin_utils::pin_mut;

#[tokio::main]
async fn main() {
    let stream = stream::iter(0..1000).then(|x| async move {
        time::delay_for(std::time::Duration::from_millis(500)).await;
        x + 1
    });
    pin_mut!(stream); //<---- here, pinned!
    while let Some(x) = stream.next().await {
        println!("{:?}", x)
    }
}

Why usage of async block in then makes my Stream Unpin?

1 Answers