for_await macro

[yoshuawuyts]

Description

Adds a for_await macro that can be used to parallelize spawning of items in a stream. Depends on rust-lang/futures-rs#1548 to land first.

Motivation and Context

This introduces a superset of the futures::for_await macro, where a concurrency mode can now be set in the attribute, which in turn can spawn individual tasks back on the executor.

This makes iterating over streams much more legible:

#[runtime::main]
async fn main() -> std::io::Result<()> {
    let mut listener = TcpListener::bind("127.0.0.1:8080")?;
    println!("Listening on {}", listener.local_addr()?);

    #[for_await(try_parallel)]
    for stream in listener.incoming() {
        println!("Accepting from: {}", stream.peer_addr()?);

        let (reader, writer) = &mut stream.split();
        await!(reader.copy_into(writer))?;
        Ok::<(), std::io::Error>(())
    }
    Ok(())
}

as compared to:

#[runtime::main]
async fn main() -> std::io::Result<()> {
    let mut listener = TcpListener::bind("127.0.0.1:8080")?;
    println!("Listening on {}", listener.local_addr()?);

    // accept connections and process them in parallel
    await!(listener
        .incoming()
        .try_for_each_concurrent(!0, async move |stream| {
            await!(runtime::spawn(async move {
                println!("Accepting from: {}", stream.peer_addr()?);

                let (reader, writer) = &mut stream.split();
                await!(reader.copy_into(writer))?;
                Ok::<(), std::io::Error>(())
            }))
        }))?;
    Ok(())
}

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)

[taiki-e]

@yoshuawuyts rust-lang/futures-rs#1548 was closed and published as futures-async-stream. Thanks for your patience!