loading...

Async Rust Beta- Quick Peek

jeikabu profile image jeikabu Originally published at rendered-obsolete.github.io on ・6 min read

As announced on the Rust Blog a few weeks ago, the long awaited async-await syntax hit beta and is slated for release with 1.39 early November. Take a look at the Async Book for an in-depth introduction.

In eager anticipation of async-await, we’ve been using futures 0.1 crate for some time. But, we’re long overdue to finally take a look at where things are headed. The following is some loosely structured notes on the upcoming “stable” state of futures and async/await.

Foot in the Door

cargo new async and in Cargo.toml add:

[dependencies]
futures = { version = "0.3.0-alpha", package = "futures-preview" }

Here using the “mostly final” Futures 0.3 crate.

As explained in The Book there are three release channels: nightly, beta, and stable. You may have used “nightly” before to access the latest and greatest features- this is where async/await previously lived. “beta” is where a version goes before becoming the next “stable” release.

# Update all installed toolchains
rustup update
# List installed toolchains
rustup toolchain list
# Install beta toolchain
rustup install beta

There’s a few ways to use the beta/nightly channels:

# List toolchain overrides
rustup override list
# Set default toolchain for directory
rustup override set beta # or "nightly"
# Now defaults to beta toolchain
cargo build

# Explicitly build using beta/nightly toolchain
cargo +beta build # or "+nightly"

In src/main.rs:

async fn hello_world() {
    println!("Hello world");
}

async fn start() {
    hello_world().await
}

fn main() {
    let future = start();
    futures::executor::block_on(future);
}

cargo run to greet the world.

Notice how await is post-fix instead of await hello_world() as found in many other languages. The syntax was heavily debated, but the rationale boils down to improving: method chaining, co-existance with the ? operator, and precedence rules.

A contrived example with a series of calls (some of which can fail):

let answer = can_fail().await?
    .some_func().await
    .member_can_fail().await?
    .get_answer()?;

You can’t understand async without understanding Rust’s Future trait. Perhaps the first thing to learn about Future is they’re lazy; nothing happens unless something “pumps” them- as executor::block_on does here. Contrast this with std::thread::spawn which creates a running thread. If futures are polled, does that mean Rust async programming isn’t event-driven à la epoll/kqueue? Don’t fret, a Waker can be used to signal the future is ready to be polled again.

Error-Handling

We have test code like:

while !done.load(Ordering::Relaxed) {
    match block_on(ctx.receive()) {
        Ok(msg) => {
            let pipe = msg.get_pipe()?;
            let mut response = msg.dup()?;
            response.set_pipe(&pipe);
            block_on(ctx.send(response))?;
        }
        _ => panic!(),
    }
}

How we might re-write it:

let future = async {
    while !done.load(Ordering::Relaxed) {
        match ctx.receive().await {
            Ok(msg) => {
                let pipe = msg.get_pipe()?;
                let mut response = msg.dup()?;
                response.set_pipe(&pipe);
                ctx.send(response).await?;
            }
            _ => panic!(),
        }
    }
};

block_on(future);

Unfortunately, how the ? operator works in async blocks (i.e. async {}) is not defined, and async closures (i.e. async || {}) are unstable.

If we replace ? with .unwrap() it compiles and runs.

Heterogeneous Returns

Given:

broker_pull_ctx.receive().for_each(|msg| {
    if let Ok(msg) = msg {
        broker_push_ctx.send(msg).then(|msg| {
            // Do something with the message
            future::ready(())
        })
    } else {
        future::ready(())
    }
});

Sadness:

|
144 | / if let Ok(msg) = msg {
145 | | broker_push_ctx.send(msg).then(|res| {
    | | _____________________ -
146 | || res.unwrap().unwrap();
147 | || future::ready(())
148 | || })
    | || ______________________ - expected because of this
149 | | } else {
150 | | future::ready(())
    | | ^^^^^^^^^^^^^^^^^ expected struct `futures_util::future::then::Then`, found struct `futures_util::future::ready::Ready`
151 | | }
    | | _________________ - if and else have incompatible types
    |
    = note: expected type `futures_util::future::then::Then<futures_channel::oneshot::Receiver<std::result::Result<(), runng::result::Error>>, futures_util::future::ready::Ready<_>, [closure@runng/tests/test_main.rs:145:52: 148:22]>`
               found type `futures_util::future::ready::Ready<_>`

Basically, then()- like many Rust combinators- returns a distinct type (Then in this case).

If you reach for a trait object for type erasure via -> Box<dyn Future<Output = ()>> and wrap the returns in Box::new() you’ll run into:

error[E0277]: the trait bound `dyn core::future::future::Future<Output = ()>: std::marker::Unpin` is not satisfied
   --> runng/tests/test_main.rs:155:58
    |
155 | let fut = broker_pull_ctx.receive().unwrap().for_each(|msg| -> Box<dyn Future<Output = ()>> {
    | ^^^^^^^^ the trait `std::marker::Unpin` is not implemented for `dyn core::future::future::Future<Output = ()>`
    |
    = note: required because of the requirements on the impl of `core::future::future::Future` for `std::boxed::Box<dyn core::future::future::Future<Output = ()>>`

Lo, the 1.33 feature “pinning”. Thankfully, the type-insanity that is Pin<Box<dyn Future<Output = T>>> is common enough that a future::BoxFuture<T> alias is provided:

let fut = broker_pull_ctx...for_each(|msg| -> future::BoxFuture<()> {
    if let Ok(msg) = msg {
        Box::pin(broker_push_ctx.send(msg).then(|_| { }))
    } else {
        Box::pin(future::ready(()))
    }
});
block_on(fut);

Alternatively, you can multiplex the return with something like future::Either:

let fut = broker_pull_ctx...for_each(|msg| {
    use futures::future::Either;
    if let Ok(msg) = msg {
        Either::Left(
            broker_push_ctx.send(msg).then(|_| { })
        )
    } else {
        Either::Right(future::ready(()))
    }
});
block_on(fut);

This avoids the boxing allocation, but it might become a bit gnarly if there’s a large number of return types.

block_on() != .await

Our initial, exploratory implementation made heavy use of wait() found in futures 0.1. To transition to async/await it’s tempting to replace wait() with block_on():

#[test]
fn block_on_panic() -> runng::Result<()> {
    let url = get_url();

    let mut rep_socket = protocol::Rep0::open()?;
    let mut rep_ctx = rep_socket.listen(&url)?.create_async()?;

    let fut = async {
        block_on(rep_ctx.receive()).unwrap();
    };
    block_on(fut);

    Ok(())
}

cargo test block_on_panic yields:

---- tests::reqrep_tests::block_on_panic stdout ----
thread 'tests::reqrep_tests::block_on_panic' panicked at 'cannot execute `LocalPool` executor from within another executor: EnterError', src/libcore/result.rs:1165:5

Note this isn’t a compiler error, it’s a runtime panic. I haven’t looked into the details of this, but the problem stems from the nested calls to block_on(). It seems that if the inner future finishes immediately everything is fine, but not if it blocks. However, it works as expected with await:

let fut = async {
    rep_ctx.receive().await.unwrap();
};
block_on(fut);

Async Traits

Try:

trait AsyncTrait {
    async fn do_stuff();
}

Nope:

error[E0706]: trait fns cannot be declared `async`
 --> src/main.rs:5:5
  |
5 | async fn do_stuff();
  | ^^^^^^^^^^^^^^^^^^^^

How about:

trait AsyncTrait {
    fn do_stuff();
}

struct Type;

impl AsyncTrait for Type {
    async fn do_stuff() { }
}

Nope:

error[E0706]: trait fns cannot be declared `async`
  --> src/main.rs:11:5
   |
11 | async fn do_stuff() { }
   | ^^^^^^^^^^^^^^^^^^^^^^^

One work-around involves being explicit about the return types, and using an async block within the impl:

use futures::future::{self, BoxFuture, Future};

trait AsyncTrait {
    fn boxed_trait() -> Box<dyn Future<Output = ()>>;
    fn pinned_box() -> BoxFuture<'static, ()>;
}

impl<T> AsyncTrait for T {
    fn boxed_trait() -> Box<dyn Future<Output = ()>> {
        Box::new(async {
            // .await to your heart's content
        })
    }
    fn pinned_box() -> BoxFuture<'static, ()> {
        Box::pin(async {
            // .await to your heart's content
        })
    }
}

Posted on Oct 28 '19 by:

Discussion

markdown guide
 

Nice writeup. I believe the Rust team and community drew a lot of inspiration from the C# async-await, during their planning and design stages? Good to see a lot of other programming languages incorporating this feature into their various ecosystems. Once more, we'll done Rust.

 

From the various threads I've read they look at a wide range of "prior art", and it was the main argument against postfix ".await"- it would be foreign to everyone.

Should have mentioned this is just the MVP to unblock the ecosystem and async-await is far from "finished". Should probably sneak that into the opening somewhere.