HSPACE Rust Lectures #12 — Async Programming, Part 2

The final session. We build the server to match Part 1's chat client, and then implement by hand the block_on and spawn_blocking we'd been taking from libraries. That last part was the best bit — you get to see futures from the inside.

The server's main

fn main() -> ChatResult<()> {
    let address = std::env::args().nth(1).expect("Usage: server ADDRESS");
    let chat_group_table = Arc::new(group_table::GroupTable::new());

    async_std::task::block_on(async {
        let listener = net::TcpListener::bind(address).await?;
        let mut new_connections = listener.incoming();

        while let Some(socket_result) = new_connections.next().await {
            let socket = socket_result?;
            let groups = chat_group_table.clone();

            task::spawn(async {
                log_error(serve(socket, groups).await);
            });
        }

        Ok(())
    })
}

A classic accept loop, async edition:

• listener.incoming() is a stream, consumed with while let ... .next().await
• each connection gets a task::spawn — a task per connection, not a thread per connection, which is the defining shape of an async server
• an Arc<GroupTable> clone goes to each task, since they all share the group table

A thread-based server would need ten thousand threads for ten thousand connections; here ten thousand tasks ride on a thread pool.

Handling a chat connection

Each connection's job is simple: read requests and act on them.

pub async fn serve(socket: TcpStream, groups: Arc<GroupTable>) -> ChatResult<()> {
    let outbound = Arc::new(Outbound::new(socket.clone()));
    let buffered = BufReader::new(socket);
    let mut from_client = utils::receive_as_json(buffered);

    while let Some(request_result) = from_client.next().await {
        let request = request_result?;
        let result = match request {
            FromClient::Join { group_name } => {
                let group = groups.get_or_create(group_name);
                group.join(outbound.clone());
                Ok(())
            }
            FromClient::Post { group_name, message } => match groups.get(&group_name) {
                Some(group) => {
                    group.post(message);
                    Ok(())
                }
                None => Err(format!("Group '{group_name}' does not exist")),
            },
        };

        if let Err(message) = result {
            outbound.send(FromServer::Error(message)).await?;
        }
    }

    Ok(())
}

Note socket.clone(). Cloning an async_std TcpStream gives another handle to the same socket, so you can hold one for reading and one for writing.

The outgoing side is wrapped in Outbound:

pub struct Outbound(Mutex<TcpStream>);

impl Outbound {
    pub async fn send(&self, packet: FromServer) -> ChatResult<()> {
        let mut guard = self.0.lock().await;
        utils::send_as_json(&mut *guard, &packet).await?;
        guard.flush().await?;
        Ok(())
    }
}

Why a mutex? Because one client can join several groups, and each group sends from its own task. Without the lock, two messages' JSON could interleave on the wire.

Crucially it's an async_std::sync::Mutex: .lock().await, not .lock().unwrap(). Part 1 said "don't hold a lock across .await" — and here we must, because the write itself is async. That's exactly what an async mutex is for: waiting for the lock folds the task away instead of blocking the thread.

Seeing both kinds in one program made the distinction concrete:

• sync std::sync::Mutex — touch the data briefly and release. Fast.
• async async_std::sync::Mutex — when you must .await while holding it.

The group table

And here's a place where the sync mutex is right:

pub struct GroupTable(Mutex<HashMap<Arc<String>, Arc<Group>>>);

impl GroupTable {
    pub fn get(&self, name: &String) -> Option<Arc<Group>> {
        self.0.lock().unwrap().get(name).cloned()
    }

    pub fn get_or_create(&self, name: Arc<String>) -> Arc<Group> {
        self.0
            .lock()
            .unwrap()
            .entry(name.clone())
            .or_insert_with(|| Arc::new(Group::new(name)))
            .clone()
    }
}

This one uses std::sync::Mutex. A hash lookup finishes instantly and there's no .await, so a sync lock is plenty. Both methods clone an Arc out and drop the lock immediately — no lingering inside the critical section.

entry().or_insert_with().clone() is the idiom for "get it or create it" in a single hash lookup.

Chat groups

A group is a broadcast channel.

pub struct Group {
    name: Arc<String>,
    sender: broadcast::Sender<Arc<String>>,
}

impl Group {
    pub fn new(name: Arc<String>) -> Group {
        let (sender, _receiver) = broadcast::channel(1000);
        Group { name, sender }
    }

    pub fn join(&self, outbound: Arc<Outbound>) {
        let receiver = self.sender.subscribe();
        task::spawn(handle_subscriber(self.name.clone(), receiver, outbound));
    }

    pub fn post(&self, message: Arc<String>) {
        // Only errors when there are no subscribers.
        let _ignored = self.sender.send(message);
    }
}

This differs from the mpsc channel of the concurrency session. mpsc is many producers, one consumer. A chat group is the reverse: many consumers, each of which must receive every message — hence tokio's broadcast channel.

Each join spawns a subscriber task:

async fn handle_subscriber(
    group_name: Arc<String>,
    mut receiver: broadcast::Receiver<Arc<String>>,
    outbound: Arc<Outbound>,
) {
    loop {
        let packet = match receiver.recv().await {
            Ok(message) => FromServer::Message {
                group_name: group_name.clone(),
                message: message.clone(),
            },
            Err(RecvError::Lagged(n)) => {
                FromServer::Error(format!("Dropped {n} messages from {group_name}."))
            }
            Err(RecvError::Closed) => break,
        };

        if outbound.send(packet).await.is_err() {
            break;
        }
    }
}

Handling RecvError::Lagged(n) was the most instructive part of this design. The broadcast buffer holds 1000 messages, and a slow client that falls further behind loses the oldest ones. It gets a Lagged(n), and the server tells it "you missed n messages."

One slow consumer doesn't stall everyone else. For chat that's the right trade — seeing recent messages matters more than guaranteeing every message — and it's a clean illustration that backpressure policy is a design decision.

The comment on post explains ignoring its error too: a connection's outgoing side can exit slightly before its incoming side, so posting to an empty group is a normal occurrence.

Built-in futures and executors

Here's the real subject of the session. We've been calling async_std::task::block_on all along; now we build it.

Something has to call poll to drive a future, and that something is the executor. And the thing that says when to call again after a Pending is the Waker.

Calling the waker: implementing spawn_blocking

Writing spawn_blocking yourself makes the waker's role obvious. Its job: run a blocking operation on a separate thread and hand back a future for the result.

pub struct SpawnBlocking<T>(Arc<Mutex<Shared<T>>>);

struct Shared<T> {
    value: Option<T>,       // the result (None until it exists)
    waker: Option<Waker>,   // who to wake when it does
}

The shared state holds exactly two things: a value and a waker. That's all.

pub fn spawn_blocking<T, F>(closure: F) -> SpawnBlocking<T>
where
    F: FnOnce() -> T + Send + 'static,
    T: Send + 'static,
{
    let inner = Arc::new(Mutex::new(Shared { value: None, waker: None }));

    std::thread::spawn({
        let inner = inner.clone();

        move || {
            let value = closure();                  // run the blocking work

            let maybe_waker = {
                let mut guard = inner.lock().unwrap();
                guard.value = Some(value);          // store the result
                guard.waker.take()                  // take the registered waker
            };

            if let Some(waker) = maybe_waker {
                waker.wake();                       // notify the executor
            }
        }
    });

    SpawnBlocking(inner)
}

And the Future impl is remarkably short:

impl<T: Send> Future for SpawnBlocking<T> {
    type Output = T;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
        let mut guard = self.0.lock().unwrap();

        if let Some(value) = guard.value.take() {
            return Poll::Ready(value);      // result available → hand it over
        }

        guard.waker = Some(cx.waker().clone());   // otherwise leave a waker
        Poll::Pending                              // and ask to be woken later
    }
}

The entire async protocol is in these twenty lines:

1. the executor calls poll
2. not ready, so the future takes the waker from Context, stores it, returns Pending
3. the executor forgets about this future for now (no busy waiting)
4. when the work finishes, the worker thread calls waker.wake()
5. the executor puts the task back on its run queue
6. poll again → Ready(value)

Note also that guard.waker.take() happens under the lock while waker.wake() happens outside it. Calling foreign code while holding a lock invites deadlock.

Implementing block_on

Now the executor side — and it's shorter still.

use crossbeam::sync::Parker;
use futures_lite::pin;
use waker_fn::waker_fn;

fn block_on<F: Future>(future: F) -> F::Output {
    let parker = Parker::new();
    let unparker = parker.unparker().clone();
    let waker = waker_fn(move || unparker.unpark());   // waking = unparking
    let mut context = Context::from_waker(&waker);

    pin!(future);

    loop {
        match future.as_mut().poll(&mut context) {
            Poll::Ready(value) => return value,
            Poll::Pending => parker.park(),            // sleep until woken
        }
    }
}

Read it and it really is that simple:

• build a waker whose only job is "wake this thread"
• loop, polling
• Ready → return the value, done
• Pending → park(), i.e. put the thread to sleep. Not a busy loop: until the waker calls unpark, this thread uses no CPU at all

"Async runtime" sounds like a monstrous piece of machinery, but a minimal single-future executor is this. What makes async_std and tokio complex is everything stacked on top: multi-task scheduling, work stealing, an I/O event loop (epoll/kqueue), timers. The skeleton is this loop.

Pinning and a future's two life stages

Why is pin!(future) there? That's the session's last concept. Pin has a reputation as the hard part of async Rust, and it does make sense once you see why it exists.

An async block compiles into a state machine. Now consider:

async fn f() {
    let value = compute();
    let r = &value;          // a reference to a local
    something().await;       // suspension point — both value and r are stored
    println!("{r}");
}

r points at value, and both are fields of the same state machine struct. That is, the struct points into itself — a self-referential struct.

So what happens if you move that struct in memory? value's address changes while r still holds the old one. A dangling pointer.

Hence a future's two life stages:

1. Freshly created — never polled, no self-references yet, free to move
2. After the first poll — self-references may exist, so it must not move

Pin<P> is how "must not move" becomes a type. Holding a Pin<&mut F> means you cannot get F back out, and therefore cannot move it. That's precisely why Future::poll takes self: Pin<&mut Self> — to poll it, you must pin it first.

Two ways to pin:

• Box::pin(future) — put it on the heap, whose address never changes. Yields Pin<Box<F>>
• pin!(future) — keep it on the stack; the macro shadows the original binding so it can't be touched. No allocation, so it's faster

That's why block_on uses pin!: pinning on the stack, allocation-free.

The Unpin trait

Not every type needs to worry about this. i32 and String have no self-references, so moving them is harmless. Such types implement Unpin.

• most types are automatically Unpin
• futures produced by async blocks are not

For T: Unpin you can freely extract &mut T from a Pin<&mut T> — the pin imposes nothing. That's why Part 1's send_as_json signature carried S: ... + Unpin: handling a stream freely is much easier when it's Unpin.

Ultimately Pin is the minimum machinery needed to safely handle in-place, self-referential state machines. Scary name, but all it does is mark "don't move this."

When is async actually worth it?

The closing slides tackle this, and since it's the practical part, it's worth recording.

Async fits:

• servers with many concurrent connections that mostly wait on I/O
• clients dominated by latency (hitting several APIs at once)
• many tasks that are individually idle most of the time

Threads fit:

• CPU-bound computation (Rayon is better still)
• a small number of concurrent jobs — dozens are fine with threads
• when you want the code to stay simple

The costs of async

I appreciated that the lecture was honest about these.

• Function coloring — async functions are callable only from async contexts, so making one thing async infects the whole call chain
• Runtime choice — tokio or async-std, and the ecosystem splits along that line
• Debugging — stack traces go through state machines and get hard to read
• Send bounds and Pin — compile errors are much more cryptic than in sync code

So the conclusion was "only when you need it." Unless you're carrying ten thousand connections, threads are simpler and faster to write. Async is buying performance with complexity, not a better way to write code.

Wrapping up the series

Looking back over twelve posts, everything in this language derives from a single idea:

• Ownership — exactly one party is responsible for a value
• Borrowing — using it without transferring that responsibility: many sharers, or one mutator
• Lifetimes — checking that references never outlive their referents
• extended across threads, those rules become Send/Sync, and data races vanish
• extended into async, they become the constraints on values crossing an .await
• Option/Result replace null and exceptions with types, and pattern matching takes them apart
• traits give polymorphism without inheritance, with dispatch as a choice
• closures, iterators and macros layer abstraction on top at no runtime cost

I did spend a lot of time fighting the compiler. But most of that fight came down to not having decided who owns a value at a given moment. The compiler isn't being difficult; it's asking a question I'd been putting off.

The Linux Kernel half of the course's name builds on this foundation — unsafe, FFI, no_std, and the Rust for Linux project. That's the next thing to study.

Takeaways

• an async server is a task per connection, tens of thousands on a thread pool
• async mutex when you must .await while holding the lock, sync mutex otherwise
• backpressure policy is a design decision, as the broadcast channel's Lagged handling shows
• the future protocol is one line: "if not ready, leave a waker and return Pending"
• a minimal block_on is a poll → park loop
• Pin marks a self-referential state machine as unmovable
• async isn't free — use it when you need it