HSPACE Rust Lectures #11 — Async Programming, Part 1

We already did threads in the concurrency session — so why async? That's where this one starts.

Synchronous → asynchronous

Reading a socket synchronously:

let mut buf = [0u8; 1024];
let n = socket.read(&mut buf)?;   // the thread stops until data arrives

This thread can do nothing until data shows up. Fine for one connection. What about ten thousand concurrent ones?

• One thread each: every thread needs a stack. 8MB default × 10,000 = 80GB. Not all of it gets touched in practice, but kernel resources and context switching still cost.
• Context switches go through the kernel and cost microseconds. Ten thousand threads switching among themselves burn CPU on the switching alone.

The async idea: give the thread back while waiting. If the socket isn't ready, fold that job away and run another one on the same thread. When data arrives, unfold it.

That's the basis for "async for I/O-bound, threads for CPU-bound." Work that keeps the CPU busy has no point at which to fold away, so async buys it nothing.

Futures

Asynchronous computations are represented by the Future trait.

trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

enum Poll<T> {
    Ready(T),
    Pending,
}

Polling gives you one of two answers:

• Ready(value) — done, here's the value
• Pending — not yet, ask again later

Which leads to the defining property: Rust futures are lazy. Creating one does nothing; it only progresses when something polls it. The exact opposite of a JavaScript Promise, which starts running on creation.

let fut = do_something();   // nothing has run yet
fut.await;                  // now it runs

This looked inconvenient at first, but it makes cancellation free: drop the future and the work never happened. Safely stopping work in progress is a hard problem in general, and lazy futures reduce it to "stop polling."

Of course, repeatedly asking a Pending future is wasteful, which is why Context carries a Waker. A future registers "call this waker when I'm ready" and returns Pending. Part 2 implements that machinery by hand.

async functions and await expressions

Hand-implementing futures is painful. The async syntax does it for you.

async fn read_line(socket: &mut TcpStream) -> io::Result<String> {
    let mut line = String::new();
    socket.read_line(&mut line).await?;
    Ok(line)
}

An async fn compiles its body into a state machine. The return type isn't io::Result<String> but impl Future<Output = io::Result<String>>. Every .await becomes a suspension point, and any local that lives across one becomes a field of the state machine.

In other words, async fn is syntactic sugar for coroutines: write sequential code with no callback pyramid and let the compiler turn it into a state machine. Same idea as C#'s async/await, except Rust does it with no heap allocation — the state machine is one struct whose size is known at compile time.

.await is only usable inside an async context, so the top level needs something else.

block_on

The bridge between the sync and async worlds.

use async_std::task;

fn main() {
    let result = task::block_on(async {
        let socket = TcpStream::connect("127.0.0.1:8080").await?;
        // ...
        Ok(())
    });
}

block_on polls the future to completion, blocking the current thread while it does. Since main can't be async, you use it once at the entry point. Attribute macros like #[tokio::main] are wrappers around the same thing.

One warning: never call block_on inside an async function. That thread stalls and every other task on the same executor stalls with it. In practice this shows up as "why is the program completely unresponsive?"

Spawning async tasks

To make several things progress concurrently, spawn them as tasks.

use async_std::task;

let handle = task::spawn(async {
    expensive_io().await
});

let result = handle.await;   // like join, but non-blocking

The name resembles thread::spawn but the meaning is different: it doesn't create an OS thread, it registers a future with the executor's queue. A task is far cheaper than a thread — hundreds of thousands are fine.

The HTTP client example shows the shape well:

pub async fn many_requests(urls: &[String]) -> Vec<Result<String, surf::Error>> {
    let client = surf::Client::new();

    let mut handles = vec![];
    for url in urls {
        let request = client.get(&url).recv_string();
        handles.push(async_std::task::spawn(request));   // start them all
    }

    let mut results = vec![];
    for handle in handles {
        results.push(handle.await);                      // collect in order
    }

    results
}

fn main() {
    let requests = &[
        "http://example.com".to_string(),
        "https://www.red-bean.com".to_string(),
        "https://en.wikipedia.org/wiki/Main_Page".to_string(),
    ];

    let results = async_std::task::block_on(many_requests(requests));
    for result in results {
        match result {
            Ok(response) => println!("*** {response}\n"),
            Err(err) => eprintln!("error: {err}\n"),
        }
    }
}

The first loop starts all three requests; the second collects them. All three are in flight at once, so total time is the slowest single request.

What if you dropped the spawn and just wrote request.await in the first loop? You'd wait for each in turn and the times would add up. .await does not create concurrency — spawn (or a combinator like join!) does. Missing this is the usual reason for "I made it async and it got no faster."

async blocks

To make part of a function async rather than all of it, use an async block.

let fut = async {
    let data = fetch().await?;
    process(data).await
};

Blocks produce futures too, which is especially handy when passing them to spawn.

When the block must own its captures, use async move — the same reason as a closure's move, since the task can outlive the surrounding scope.

let name = String::from("Chris");
task::spawn(async move {
    println!("{name}");
});

Writing async functions by hand

Sometimes you need to spell out the return type — inside a trait, or when returning different futures conditionally.

// roughly equivalent to an async fn
fn read_line(socket: &mut TcpStream) -> impl Future<Output = io::Result<String>> + '_ {
    async move {
        // ...
    }
}

Note the '_. The future borrows socket, so socket must outlive the future. Writing async fn lets the compiler handle that for you.

Running on a thread pool

async_std::task::spawn puts the task on a thread pool, which means you don't know which thread runs it and it may migrate at any .await.

Long CPU work inside an async task blocks that pool thread, so push it aside with spawn_blocking:

let result = task::spawn_blocking(|| {
    heavy_computation()      // runs on a dedicated thread
}).await;

Why Send shows up

The topic that took the most slides — and the one that trips people up most in real async Rust.

If tasks run on a thread pool and may migrate at .await points, then everything the task holds must be Send.

use std::rc::Rc;

task::spawn(async {
    let rc = Rc::new(42);
    do_something().await;      // the thread may change here
    println!("{rc}");          // error: Rc isn't Send
});

The question is always which values live across an .await. Values dropped before it never enter the state machine and cause no trouble.

task::spawn(async {
    {
        let rc = Rc::new(42);
        println!("{rc}");
    }                          // rc dropped here
    do_something().await;      // OK — rc doesn't cross this point
});

The message reads future cannot be sent between threads safely, and the compiler points at the offending value. It looked intimidating at first; reducing it to one sentence — "some value crossing an .await isn't Send" — made it readable.

MutexGuard hits this constantly:

let guard = mutex.lock().unwrap();
do_something().await;              // holding a lock across await → trouble

It's a Send problem, but more importantly a design problem: holding a lock across .await blocks every other task from acquiring it. Either use an async mutex like tokio::sync::Mutex, or end the lock's scope before the .await.

Long computations and async design strategy

Summarized:

Kind of work	Tool
Waiting on I/O (network, files)	async / .await
Short CPU work	inline in the async fn
Long CPU work	spawn_blocking or a dedicated thread
Data parallelism	Rayon

Blocking an executor thread with CPU work starves every other task. It's the most common performance accident in async runtimes.

A real async chat client

The back half of the session builds a chat application — client here, server in Part 2.

The protocol first. Messages between client and server are enums, serialized to JSON with serde.

#[derive(Debug, Serialize, Deserialize, PartialEq)]
pub enum FromClient {
    Join { group_name: Arc<String> },
    Post { group_name: Arc<String>, message: Arc<String> },
}

#[derive(Debug, Serialize, Deserialize, PartialEq)]
pub enum FromServer {
    Message { group_name: Arc<String>, message: Arc<String> },
    Error(String),
}

The Arc<String> stands out. When the server fans one message out to every subscriber of a group, it clones an Arc instead of copying the string. Enabling serde's rc feature makes Arc serializable.

The error type collapses into Box<dyn Error + Send + Sync + 'static>:

pub type ChatError = Box<dyn Error + Send + Sync + 'static>;
pub type ChatResult<T> = Result<T, ChatError>;

The Send + Sync is now obvious: errors have to cross task boundaries. It's the Box<dyn Error> from the error-handling session plus the async requirements.

Sending and receiving packets.

pub async fn send_as_json<S, P>(outbound: &mut S, packet: &P) -> ChatResult<()>
where
    S: async_std::io::Write + Unpin,
    P: Serialize,
{
    let mut json = serde_json::to_string(&packet)?;
    json.push('\n');
    outbound.write_all(json.as_bytes()).await?;
    Ok(())
}

pub fn receive_as_json<S, P>(inbound: S) -> impl Stream<Item = ChatResult<P>>
where
    S: async_std::io::BufRead + Unpin,
    P: DeserializeOwned,
{
    inbound.lines().map(|line_result| -> ChatResult<P> {
        let line = line_result?;
        let parsed = serde_json::from_str::<P>(&line)?;
        Ok(parsed)
    })
}

The Stream returned by receive_as_json is the important concept. Where a Future asynchronously produces one value, a Stream asynchronously produces many — the async counterpart of an iterator, pulled with .next().await.

The framing is worth noticing too: one JSON object per line. The lines() stream handles the framing for free.

The client's main.

fn main() -> ChatResult<()> {
    let address = std::env::args().nth(1).expect("Usage: client ADDRESS:PORT");

    task::block_on(async {
        let socket = net::TcpStream::connect(address).await?;
        socket.set_nodelay(true)?;

        let to_server = send_commands(socket.clone());
        let from_server = handle_replies(socket);

        from_server.race(to_server).await?;

        Ok(())
    })
}

This structure is clean. The client has to do two things at once:

• read stdin and send commands to the server (send_commands)
• print messages arriving from the server (handle_replies)

race takes whichever finishes first and drops the other. The user hitting Ctrl-D or the server dropping the connection — either one ends the client. Written with threads this would need explicit shutdown signaling; in async it's one combinator.

Inside send_commands:

async fn send_commands(mut to_server: net::TcpStream) -> ChatResult<()> {
    let mut command_lines = io::BufReader::new(io::stdin()).lines();

    while let Some(command_result) = command_lines.next().await {
        let command = command_result?;
        let request = match parse_command(&command) {
            Some(request) => request,
            None => continue,
        };

        utils::send_as_json(&mut to_server, &request).await?;
        to_server.flush().await?;
    }

    Ok(())
}

while let Some(x) = stream.next().await is the idiom for iterating an async stream. It reads almost exactly like a synchronous for loop, except each iteration can fold away at an .await. That's where the core value of async/await landed for me: async code that reads like sync code.

Takeaways

• async is about reclaiming I/O wait time; it does nothing for CPU-bound work
• Rust futures are lazy — nothing happens without polling, which makes cancellation cheap
• async fn compiles a function into a state machine, with no heap allocation
• .await doesn't create concurrency; spawn and combinators like race/join do
• because tasks run on a thread pool, anything living across an .await must be Send
• Stream is the multi-value Future, i.e. the async iterator

Next: async programming, part 2 — the chat server, plus implementing block_on and spawn_blocking by hand to get at wakers and Pin.