Async Basics
Concurrency · Lesson 45
An async fn returns a future: a value that describes some work and has not done it. Nothing runs until something polls the future, and .await is how one future waits for another. The standard library defines the Future trait and the types around it, and has no runtime to poll futures for you: you write a small one to see the idea, and use a crate such as Tokio for real programs.
Coming from GoGo has no async and await, and no future value to hold. go f() starts f running alongside you, and a call that has to wait is an ordinary call. A Rust async fn differs on both counts: calling it builds a future that has not started, and .await marks each place where it waits. The scheduler that goroutines get from the language is, in Rust, a runtime that you choose.
Coming from KotlinA suspend fun is the nearest thing: it can stop and be resumed later, and a suspend call looks like a plain call where Rust marks each wait with .await. The program above builds a coroutine with the standard library's createCoroutine, which does not start it, as an async call builds a future. Everyday Kotlin uses launch and async from kotlinx.coroutines, a library as Rust's runtimes are; they are scheduled at once, like tokio::spawn, unlike a bare Rust future (shown below).
Coming from Pythonasync def and await work as you know them, and a coroutine object is lazy like a Rust future: calling task() runs nothing, as the program above shows. asyncio.run is the event loop that Python includes; Rust includes none, so you add a runtime. Python warns when an unawaited coroutine is collected, at run time; Rust warns at compile time, when the result of an async call is discarded.
Coming from TypeScriptA promise is eager: calling an async function runs its body up to the first await before you get the promise back, as the program above prints. A Rust future is lazy: nothing runs until it is polled. Node's event loop is built in; Rust's runtime is a library you add.
Coming from JavaJava has no async and await. CompletableFuture.supplyAsync starts at once on a pool, and virtual threads let you write plain blocking code. A Rust future needs no thread of its own: it is a value that something polls, on whatever thread the runtime chooses. A thread that is built but not started is the nearest to "created, not running" (the program above).
Coming from C#async and await come from the same idea, and a Task is the nearest type. A C# async method starts running before the call returns, up to its first await that has to wait, so the Task you get is already running, as the program above prints. A Rust future has not started. The runtime is a crate, not a part of the framework.
Coming from CC has no futures or async/await. A function pointer is the nearest thing in the language: holding it runs nothing until you call it. A Rust async fn becomes a value that remembers where it stopped, and a runtime calls poll on it again and again until it is ready.
Coming from C++C++20 coroutines are the nearest match. The compiler does the transformation, and a promise type decides what happens at creation: the program above is lazy because initial_suspend returns std::suspend_always, as a Rust future is. Rust's standard library has the Future trait and Poll; the runtime that drives them is a crate.
The quick comparison
use std::future::Future;use std::pin::pin;use std::task::{Context, Poll, Waker};
async fn task() { println!("task body runs");}
fn main() { println!("start"); let mut future = pin!(task()); println!("task created");
// a runtime does this for you: ask the future to run let mut cx = Context::from_waker(Waker::noop()); let state = future.as_mut().poll(&mut cx); assert_eq!(state, Poll::Ready(())); println!("end");}start task created task body runs end
package main
import ( "fmt" "sync")
func main() { fmt.Println("start") // Go has no future value. A function value is the nearest thing: // holding it runs nothing until it is called. task := func() { fmt.Println("task body runs") } fmt.Println("task created")
// `go` starts it running alongside main; WaitGroup waits for it. var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done() task() }() wg.Wait() fmt.Println("end")}start task created task body runs end
import kotlin.coroutines.Continuationimport kotlin.coroutines.EmptyCoroutineContextimport kotlin.coroutines.createCoroutineimport kotlin.coroutines.resume
fun main() { println("start") // createCoroutine builds the coroutine without starting it val coroutine = suspend { println("task body runs") } .createCoroutine(Continuation(EmptyCoroutineContext) { }) println("task created") coroutine.resume(Unit) println("end")}start task created task body runs end
import asyncio
async def task(): print("task body runs")
print("start")coroutine = task() # a coroutine object: the body has not runprint("task created")asyncio.run(coroutine)print("end")start task created task body runs end
async function task(): Promise<void> { console.log("task body runs");}
async function main(): Promise<void> { console.log("start"); const promise = task(); // the body has already run, up to its first await console.log("task created"); await promise; console.log("end");}
main();start task body runs task created end
public class Main { public static void main(String[] args) throws InterruptedException { System.out.println("start"); // Java has no async/await. A virtual thread that is built but not started // is the nearest thing to "created, not running". Thread task = Thread.ofVirtual().unstarted(() -> System.out.println("task body runs")); System.out.println("task created"); task.start(); task.join(); System.out.println("end"); }}start task created task body runs end
using System;using System.Threading.Tasks;
static async Task TaskBody(){ Console.WriteLine("task body runs"); await Task.CompletedTask;}
Console.WriteLine("start");Task task = TaskBody(); // runs on this thread until its first await that has to waitConsole.WriteLine("task created");await task;Console.WriteLine("end");start task body runs task created end
#include <stdio.h>
static void task(void) { puts("task body runs"); }
int main(void) { puts("start"); /* C has no futures or async/await. A function pointer is the nearest thing: holding it runs nothing until it is called. */ void (*created)(void) = task; puts("task created"); created(); puts("end"); return 0;}start task created task body runs end
#include <coroutine>#include <iostream>
// C++20 gives the compiler side of a coroutine; the type that says what it does// when created, resumed and finished is yours (or a library's).struct Task { struct promise_type { Task get_return_object() { return Task{std::coroutine_handle<promise_type>::from_promise(*this)}; } std::suspend_always initial_suspend() { return {}; } // lazy: stop before the body std::suspend_never final_suspend() noexcept { return {}; } void return_void() {} void unhandled_exception() {} }; std::coroutine_handle<promise_type> handle;};
Task task() { std::cout << "task body runs\n"; co_return;}
int main() { std::cout << "start\n"; Task t = task(); std::cout << "task created\n"; t.handle.resume(); std::cout << "end\n";}start task created task body runs end
What's different
- Calling an
async fndoes not run it. It returns a future, and the compiler warns when you ignore the result of the call (futures do nothing unless you .await or poll them). - A future is polled and answers
Ready(value)orPending..awaitpolls an inner future and, if it isPending, makes the outer futurePendingtoo. AWakeris how a future asks to be polled again. - The standard library has no runtime. The
block_onon this page is about ten lines, and shows all a runtime does at its core: poll until ready. The real one, with timers, I/O and spawning tasks, is Tokio. - A task that never awaits holds its thread until it finishes, so blocking inside async code stalls the other tasks on that thread. A future that is moved to another thread must be
Send, which includes everything it keeps across an.await.
Using the result of an async function as if it were the value
Looks fine: in Go and Kotlin it runs fine; in Python it crashes at runtime. Rust won't compile it.
Works in Go. In Rust it's just a rule.Works in Kotlin. In Rust it's just a rule.Broken in Python. Rust catches it.TypeScript refuses this too. Same idea, different words.Java refuses this too. Same idea, different words.C# refuses this too. Same idea, different words.No direct equivalent in C. Just a Rust rule.C++ refuses this too. Same idea, different words.async fn load_count() -> u32 { 3}
async fn report() -> u32 { let n: u32 = load_count(); n + 1}
fn main() { let _ = report();}package main
import "fmt"
// Go has no async functions: a call that waits is an ordinary call.func loadCount() int { return 3}
func report() int { n := loadCount() return n + 1}
func main() { fmt.Println("count:", report())}count: 4
suspend fun loadCount(): Int = 3
suspend fun report(): Int { val n = loadCount() // a suspend call looks like a plain call return n + 1}
suspend fun main() { println("count: ${report()}")}count: 4
import asyncio
async def load_count(): return 3
async def report(): n = load_count() return n + 1
print("count:", asyncio.run(report()))Traceback (most recent call last):
File "main.py", line 13, in <module>
print("count:", asyncio.run(report()))
~~~~~~~~~~~^^^^^^^^^^
File "/usr/lib/python3.13/asyncio/runners.py", line 195, in run
return runner.run(main)
~~~~~~~~~~^^^^^^
File "/usr/lib/python3.13/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
File "/usr/lib/python3.13/asyncio/base_events.py", line 725, in run_until_complete
return future.result()
~~~~~~~~~~~~~^^
File "main.py", line 10, in report
return n + 1
~~^~~
TypeError: unsupported operand type(s) for +: 'coroutine' and 'int'
<sys>:0: RuntimeWarning: coroutine 'load_count' was never awaitedtsc --strict refuses: Type 'Promise<number>' is not assignable to type 'number' (TS2322), because n is annotated as a number.
async function loadCount(): Promise<number> { return 3;}
async function report(): Promise<number> { const n: number = loadCount(); return n + 1;}
report().then((count) => console.log("count:", count));main.ts(6,9): error TS2322: Type 'Promise<number>' is not assignable to type 'number'.
javac refuses: CompletableFuture<Integer> cannot be converted to int.
import java.util.concurrent.CompletableFuture;
public class Main { static CompletableFuture<Integer> loadCount() { return CompletableFuture.supplyAsync(() -> 3); }
static int report() { int n = loadCount(); return n + 1; }
public static void main(String[] args) { System.out.println("count: " + report()); }}Main.java:9: error: incompatible types: CompletableFuture<Integer> cannot be converted to int
int n = loadCount();
^
1 errorC# refuses: Cannot implicitly convert type 'Task<int>' to 'int' (CS0029).
using System;using System.Threading.Tasks;
static async Task<int> LoadCount(){ await Task.Delay(1); return 3;}
static async Task<int> Report(){ int n = LoadCount(); return n + 1;}
Console.WriteLine($"count: {await Report()}");Program.cs(12,13): error CS0029: Cannot implicitly convert type 'System.Threading.Tasks.Task<int>' to 'int'
C has no async functions, so there is no such mistake to make.
g++ refuses: cannot convert 'std::future<int>' to 'int' in initialization. The value returned by std::async is a future too.
#include <future>#include <iostream>
std::future<int> load_count() { return std::async(std::launch::async, [] { return 3; });}
int report() { int n = load_count(); return n + 1;}
int main() { std::cout << "count: " << report() << "\n";}main.cpp: In function ‘int report()’:
main.cpp:9:23: error: cannot convert ‘std::future<int>’ to ‘int’ in initialization
9 | int n = load_count();
| ~~~~~~~~~~^~
| |
| std::future<int>What does the compiler say? Guess first, then open.
error[E0308]: mismatched types --> src/main.rs:6:18 | 6 | let n: u32 = load_count(); | --- ^^^^^^^^^^^^ expected `u32`, found future | | | expected due to this | note: calling an async function returns a future --> src/main.rs:6:18 | 6 | let n: u32 = load_count(); | ^^^^^^^^^^^^ help: consider `await`ing on the `Future` | 6 | let n: u32 = load_count().await; | ++++++
Go has no async functions. loadCount() is an ordinary call that returns the number, and the program prints count: 4.
Kotlin runs this and prints count: 4. A suspend call looks like a plain call and returns the Int, with no await to write.
Python raises TypeError: unsupported operand type(s) for +: 'coroutine' and 'int' when it reaches n + 1, and warns that the coroutine was never awaited. The mistake is the same; Python finds it while the program runs.
tsc --strict refuses: Type 'Promise<number>' is not assignable to type 'number' (TS2322), because n is annotated as a number.
javac refuses: CompletableFuture<Integer> cannot be converted to int.
C# refuses: Cannot implicitly convert type 'Task<int>' to 'int' (CS0029).
C has no async functions, so there is no such mistake to make.
g++ refuses: cannot convert 'std::future<int>' to 'int' in initialization. The value returned by std::async is a future too.
load_count() is a call to an async function, and an async function returns a future, not a u32. The compiler says so ("calling an async function returns a future") and suggests .await.
The type of the call is impl Future<Output = u32>, so putting it where a u32 is wanted is a type error and not a silent wrong result. .await is only legal inside an async function or block, and main is not one, so fixed.rs also needs something to run the future. It uses block_on, ten lines of standard library code; a real program uses a runtime.
The fix
Compiles and runs.
use std::future::Future;use std::pin::pin;use std::task::{Context, Poll, Waker};
async fn load_count() -> u32 { 3}
async fn report() -> u32 { let n: u32 = load_count().await; n + 1}
/// A runtime in miniature: poll until the future is ready.fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return value; } }}
fn main() { println!("count: {}", block_on(report()));}count: 4
The toolkit
Outputs come from Linux on 64-bit ARM, Rust 1.99.1, debug mode. The programs use no crates, so you see what a runtime hides. After this page you can read async fn, .await, Send errors and Pin in library code, but not yet a typical Tokio handler: Async with Tokio comes next, and the last section lists what is missing.
A future is a value that has not run
Calling an async fn runs none of its body. It returns a future, a value that stands for the work. The Future docs say: “Futures alone are inert; they must be actively polled for the underlying computation to make progress.” Polling is calling the future’s poll method, which runs the body until it finishes or has to wait. Whatever does the polling is the executor, the core of a runtime; a future it owns and polls is a task. Read block_on, the smallest executor, as “call poll until it answers Ready”; the other lines are explained below.
use std::future::Future;use std::pin::pin;use std::task::{Context, Poll, Waker};
/// Polls `fut` until it is ready. Returns the output and the number of polls it took.fn block_on<F: Future>(fut: F) -> (F::Output, u32) { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); let mut polls = 0; loop { polls += 1; if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return (value, polls); } }}
async fn fetch_name() -> String { println!(" fetch_name is running"); String::from("ferris")}
fn main() { println!("calling fetch_name()"); let future = fetch_name(); println!("called it, and nothing ran");
let (name, polls) = block_on(future); println!("name: {name}, polls: {polls}");}calling fetch_name() called it, and nothing ran fetch_name is running name: ferris, polls: 1
fetch_name()built the future.fetch_name is runningprinted only afterblock_onpolled it.pin!fixes the future in place, becausepolldemands it (Pin). AContextcarries a waker, how a future asks to be polled again (next section).Waker::noop()does nothing; stable since Rust 1.85 (docs).- This future finished on its first poll. The loop busy-waits.
If you call an async fn and ignore the result, the compiler warns (unused_must_use). deny turns the warning into an error the page can show:
// By default this is a warning. `deny` turns it into the error the page can show.#![deny(unused_must_use)]
async fn save() { println!("saving");}
fn main() { save(); println!("done");}error: unused implementer of `Future` that must be used --> src/main.rs:9:5 | 9 | save(); | ^^^^^^ | = note: futures do nothing unless you `.await` or poll them note: the lint level is defined here --> src/main.rs:2:9 | 2 | #![deny(unused_must_use)] | ^^^^^^^^^^^^^^^
The lint fires only on a discarded statement: let _ = save(); and drop(save()); stay silent (check-unused-let.rs), though the work never runs.
Pending, Ready and the waker
poll answers Ready(value) when the future is finished and Pending when it is not. Here is a future written by hand that is not ready the first time, and an async fn that waits on it:
use std::future::{Future, ready};use std::pin::{Pin, pin};use std::task::{Context, Poll, Waker};
/// Not ready the first time it is polled, ready the second time.struct YieldNow { yielded: bool,}
impl Future for YieldNow { type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { if self.yielded { Poll::Ready(()) } else { self.yielded = true; cx.waker().wake_by_ref(); // "poll me again soon" Poll::Pending } }}
/// Polls `fut` until it is ready. Returns the output and the number of polls it took.fn block_on<F: Future>(fut: F) -> (F::Output, u32) { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); let mut polls = 0; loop { polls += 1; if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return (value, polls); } }}
async fn wait_then_answer() -> u32 { println!(" before the await"); YieldNow { yielded: false }.await; println!(" after the await"); 42}
fn main() { let mut cx = Context::from_waker(Waker::noop()); let mut fut = pin!(YieldNow { yielded: false }); println!("poll 1: {:?}", fut.as_mut().poll(&mut cx)); println!("poll 2: {:?}", fut.as_mut().poll(&mut cx));
let (answer, polls) = block_on(wait_then_answer()); println!("answer: {answer}, polls: {polls}");
// awaiting a future that is already ready does not stop the caller let ((), polls) = block_on(async { ready(()).await }); println!("await on a ready future, polls: {polls}");}poll 1: Pending poll 2: Ready(()) before the await after the await answer: 42, polls: 2 await on a ready future, polls: 1
YieldNowreturnedPendingonce andReady(())on the second poll..awaitis that conversation. In the Reference’s words, if the innerpollreturnsPending, the outer future returnsPendingtoo, and when polled again it resumes at the same.await.before the awaitprinted once, though the body was polled twice.- An
.awaiton a future that is alreadyReadydoes not stop: the last line shows one poll. wake_by_refis the future saying “poll me again soon”.
A real executor sleeps instead of looping, and the waker wakes it. This one records whether it was woken and gives up rather than sleep forever. Wake turns an Arc of your own type into a Waker:
use std::future::Future;use std::pin::{Pin, pin};use std::sync::Arc;use std::sync::atomic::{AtomicBool, Ordering};use std::task::{Context, Poll, Wake, Waker};
/// A waker that remembers it was woken.struct Flag(AtomicBool);
impl Wake for Flag { fn wake(self: Arc<Self>) { self.0.store(true, Ordering::SeqCst); }}
/// Pending `pending` times, then ready. Tells the waker only if `wake` is true.struct Slow { pending: u32, wake: bool,}
impl Future for Slow { type Output = &'static str;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { if self.pending == 0 { return Poll::Ready("ready"); } self.pending -= 1; if self.wake { cx.waker().wake_by_ref(); } Poll::Pending }}
/// Polls again only when the future asked to be. Gives up instead of waiting forever.fn run(fut: impl Future<Output = &'static str>) -> Option<&'static str> { let flag = Arc::new(Flag(AtomicBool::new(false))); let waker = Waker::from(flag.clone()); let mut cx = Context::from_waker(&waker); let mut fut = pin!(fut); loop { if let Poll::Ready(v) = fut.as_mut().poll(&mut cx) { return Some(v); } if !flag.0.swap(false, Ordering::SeqCst) { return None; // a real executor would sleep here, with nothing to wake it } }}
fn main() { println!( "wakes before each Pending: {:?}", run(Slow { pending: 3, wake: true }) ); println!( "never wakes: {:?}", run(Slow { pending: 3, wake: false }) );}wakes before each Pending: Some("ready")
never wakes: NoneThe first future woke the executor before each Pending, so it was polled again and finished. The second never asked: a real executor would wait for a wake-up that never comes. A runtime stores the waker and calls it when a timer fires or a socket has data (Tokio guide).
The standard library has no runtime
Nothing in std::future runs a future, so main cannot be async: nothing would poll it.
async fn main() { println!("hello");}error[E0752]: `main` function is not allowed to be `async` --> src/main.rs:1:1 | 1 | async fn main() { | ^^^^^^^^^^^^^^^ `main` function is not allowed to be `async`
That is why #[tokio::main] exists. A runtime is an event loop, which learns when timers fire and sockets are readable, plus a scheduler, which decides which task to poll next. Node and asyncio.run hide this loop; block_on is the smallest. The language gives you async, .await and Future, and the scheduler is a library you choose.
Two futures on one thread: join
To make progress on two futures, poll both. std::future::join! is experimental on 1.99.1 (E0658; check-join-unstable.fail.rs), so here it is by hand; poll_fn turns a closure into a future. A finished future is left alone: the Future docs warn that polling after Ready may panic or block forever.
use std::cell::RefCell;use std::future::{Future, poll_fn};use std::pin::{Pin, pin};use std::rc::Rc;use std::task::{Context, Poll, Waker};
struct YieldNow(bool);
impl Future for YieldNow { type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { if self.0 { return Poll::Ready(()); } self.0 = true; cx.waker().wake_by_ref(); Poll::Pending }}
fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(v) = fut.as_mut().poll(&mut cx) { return v; } }}
/// Runs both futures at once on this thread and returns both results.async fn join_two<A: Future, B: Future>(a: A, b: B) -> (A::Output, B::Output) { let mut a = pin!(a); let mut b = pin!(b); let (mut a_out, mut b_out) = (None, None); poll_fn(|cx| { if a_out.is_none() { if let Poll::Ready(v) = a.as_mut().poll(cx) { a_out = Some(v); } } if b_out.is_none() { if let Poll::Ready(v) = b.as_mut().poll(cx) { b_out = Some(v); } } if a_out.is_some() && b_out.is_some() { Poll::Ready(()) } else { Poll::Pending } }) .await; (a_out.unwrap(), b_out.unwrap())}
async fn count(name: &'static str, n: u32, log: Rc<RefCell<Vec<String>>>) -> u32 { for i in 1..=n { log.borrow_mut().push(format!("{name}{i}")); YieldNow(false).await; } n * 10}
fn main() { let log = Rc::new(RefCell::new(Vec::new())); let (a, b) = block_on(async { let a = count("a", 3, log.clone()).await; let b = count("b", 2, log.clone()).await; (a, b) }); println!( "one after the other: {a} {b} steps: {}", log.borrow().join(" ") );
log.borrow_mut().clear(); let (a, b) = block_on(join_two( count("a", 3, log.clone()), count("b", 2, log.clone()), )); println!( "joined: {a} {b} steps: {}", log.borrow().join(" ") );}one after the other: 30 20 steps: a1 a2 a3 b1 b2 joined: 30 20 steps: a1 b1 a2 b2 a3
One after the other, the steps run a1 a2 a3 b1 b2. Joined, they take turns: a1 b1 a2 b2 a3, with no second thread.
Async blocks capture like closures
async { ... } is a future too, and it captures like a closure (Closures): by reference, unless you write move.
use std::future::Future;use std::pin::pin;use std::task::{Context, Poll, Waker};
fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(v) = fut.as_mut().poll(&mut cx) { return v; } }}
fn greeting(name: String) -> impl Future<Output = usize> { async move { println!(" the block runs now, with name = {name}"); name.len() }}
fn main() { let name = String::from("ferris");
// an async block borrows what it uses, like a closure let borrowing = async { name.len() }; println!("block built; name is still ours: {name}"); println!("length: {}", block_on(borrowing));
// `async move` takes ownership, so the future can outlive this scope let future = greeting(name); println!("future built; its body has not run"); println!("length: {}", block_on(future));}block built; name is still ours: ferris length: 6 future built; its body has not run the block runs now, with name = ferris length: 6
The first block borrowed name, so name was still ours. async move took it, so the future can leave the function. A returned future that borrows a local is refused:
use std::future::Future;fn make() -> impl Future<Output = usize> { let s = String::from("hello"); async { s.len() }}fn main() { let _ = make();}error[E0373]: async block may outlive the current function, but it borrows `s`, which is owned by the current function --> src/main.rs:4:5 | 4 | async { s.len() } | ^^^^^ - `s` is borrowed here | | | may outlive borrowed value `s` | note: async block is returned here --> src/main.rs:4:5 | 4 | async { s.len() } | ^^^^^^^^^^^^^^^^^ help: to force the async block to take ownership of `s` (and any other referenced variables), use the `move` keyword | 4 | async move { s.len() } | ++++
A future that must cross threads is Send
Tokio’s default runtime is multi-threaded: the #[tokio::main] docs say multi_thread “is the default flavor”. A task can then resume on another worker thread after any .await, so spawn asks for Send (Send and Sync) and 'static, as thread::spawn does (Reading signatures; 'static). The spawn below calls thread::spawn, so the bound is real. A future holds everything alive across an .await, so an Rc there makes it not Send:
use std::future::{Future, ready};use std::pin::pin;use std::rc::Rc;use std::task::{Context, Poll, Waker};use std::thread;
fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return value; } }}
/// Like a runtime's `spawn`: the future may move to another thread, so it must be `Send`.fn spawn(fut: impl Future<Output = ()> + Send + 'static) { thread::spawn(move || block_on(fut)).join().unwrap();}
async fn count_words() { let words = Rc::new(vec!["one", "two"]); ready(()).await; // an await point: the future may be moved to another thread here println!("words: {}", words.len());}
fn main() { spawn(count_words());}error: future cannot be sent between threads safely --> src/main.rs:29:11 | 29 | spawn(count_words()); | ^^^^^^^^^^^^^ future returned by `count_words` is not `Send` | = help: within `impl Future<Output = ()>`, the trait `Send` is not implemented for `Rc<Vec<&str>>` note: future is not `Send` as this value is used across an await --> src/main.rs:24:15 | 23 | let words = Rc::new(vec!["one", "two"]); | ----- has type `Rc<Vec<&str>>` which is not `Send` 24 | ready(()).await; // an await point: the future may be moved to another thread here | ^^^^^ await occurs here, with `words` maybe used later note: required by a bound in `spawn` --> src/main.rs:18:42 | 18 | fn spawn(fut: impl Future<Output = ()> + Send + 'static) { | ^^^^ required by this bound in `spawn`
The first note says “future is not Send as this value is used across an await”. End the Rc’s life before the wait:
use std::future::{Future, ready};use std::pin::pin;use std::rc::Rc;use std::task::{Context, Poll, Waker};use std::thread;
fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return value; } }}
/// Like a runtime's `spawn`: the future may move to another thread, so it must be `Send`.fn spawn(fut: impl Future<Output = ()> + Send + 'static) { thread::spawn(move || block_on(fut)).join().unwrap();}
async fn count_words() { let n = { let words = Rc::new(vec!["one", "two"]); words.len() }; // words is gone before the await ready(()).await; println!("words: {n}");}
fn main() { spawn(count_words());}words: 2
An Arc there compiles too (check-send-arc.rs). Dropping early is less reliable than ending a block. On 1.99.1 a drop(words) before the wait does not help once the value was used:
use std::future::{Future, ready};use std::pin::pin;use std::rc::Rc;use std::task::{Context, Poll, Waker};use std::thread;
fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return value; } }}
/// Like a runtime's `spawn`: the future may move to another thread, so it must be `Send`.fn spawn(fut: impl Future<Output = ()> + Send + 'static) { thread::spawn(move || block_on(fut)).join().unwrap();}
async fn count_words() { let words = Rc::new(vec!["one", "two"]); let n = words.len(); drop(words); ready(()).await; println!("words: {n}");}
fn main() { spawn(count_words());}error: future cannot be sent between threads safely --> src/main.rs:31:11 | 31 | spawn(count_words()); | ^^^^^^^^^^^^^ future returned by `count_words` is not `Send` | = help: within `impl Future<Output = ()>`, the trait `Send` is not implemented for `Rc<Vec<&str>>` note: future is not `Send` as this value is used across an await --> src/main.rs:26:15 | 23 | let words = Rc::new(vec!["one", "two"]); | ----- has type `Rc<Vec<&str>>` which is not `Send` ... 26 | ready(()).await; | ^^^^^ await occurs here, with `words` maybe used later note: required by a bound in `spawn` --> src/main.rs:18:42 | 18 | fn spawn(fut: impl Future<Output = ()> + Send + 'static) { | ^^^^ required by this bound in `spawn`
A value that was only created and then dropped is accepted (check-send-drop-unused.rs), but I found no documented rule, so end the block. A lock is the same: Mutex::lock returns a guard, and one held across an .await makes the future not Send:
use std::future::{Future, ready};use std::pin::pin;use std::sync::Mutex;use std::task::{Context, Poll, Waker};use std::thread;
fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return value; } }}
/// Like a runtime's `spawn`: the future may move to another thread, so it must be `Send`.fn spawn(fut: impl Future<Output = ()> + Send + 'static) { thread::spawn(move || block_on(fut)).join().unwrap();}
async fn count_words() { let words = Mutex::new(vec!["one", "two"]); let guard = words.lock().unwrap(); ready(()).await; println!("words: {}", guard.len());}
fn main() { spawn(count_words());}error: future cannot be sent between threads safely --> src/main.rs:30:11 | 30 | spawn(count_words()); | ^^^^^^^^^^^^^ future returned by `count_words` is not `Send` | = help: within `impl Future<Output = ()>`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, Vec<&str>>` note: future is not `Send` as this value is used across an await --> src/main.rs:25:15 | 24 | let guard = words.lock().unwrap(); | ----- has type `std::sync::MutexGuard<'_, Vec<&str>>` which is not `Send` 25 | ready(()).await; | ^^^^^ await occurs here, with `guard` maybe used later note: required by a bound in `spawn` --> src/main.rs:18:42 | 18 | fn spawn(fut: impl Future<Output = ()> + Send + 'static) { | ^^^^ required by this bound in `spawn`
A guard used and then dropped still fails (check-send-guard-drop.fail.rs); one only locked and dropped compiles (check-send-guard-unused.rs). Lock in a block, or in a temporary that ends at the semicolon, and lock again after the wait (exercise 3).
What the keyword costs: function colour
A plain function cannot .await. It needs a runtime or a block_on:
async fn read_count() -> u32 { 3}
fn total() -> u32 { read_count().await + 1}
fn main() { let _ = total();}error[E0728]: `await` is only allowed inside `async` functions and blocks --> src/main.rs:6:18 | 5 | fn total() -> u32 { | ----------------- this is not `async` 6 | read_count().await + 1 | ^^^^^ only allowed inside `async` functions and blocks
This is function colour: async and plain functions are two kinds, and a plain one cannot wait on an async one, so callers turn async or carry a runtime. Libraries split too: std::fs and std::net block the calling thread, so async code uses a runtime’s own I/O types, and a crate usually serves one world. Some runtimes avoid the split by parking a blocked call themselves. Rust has no runtime in std to do that, so it marks each wait in the types. You pay with the split, and get a future you can store, drop or poll by hand.
Translating from what you know
Going deeper
What the compiler builds, and why Pin exists
An async fn becomes a value, and its size is the data it must keep while it waits. The program prints the sizes of three futures whose only wait is one YieldNow:
use std::future::Future;use std::mem::{size_of, size_of_val};use std::pin::Pin;use std::task::{Context, Poll};
/// Pending once, then ready: the smallest thing an `.await` can wait on.struct YieldNow(bool);
impl Future for YieldNow { type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { if self.0 { return Poll::Ready(()); } self.0 = true; cx.waker().wake_by_ref(); Poll::Pending }}
async fn nothing_kept() -> usize { YieldNow(false).await; 1}
async fn array_kept() -> usize { let buf = [0u8; 1000]; YieldNow(false).await; buf.len()}
async fn array_finished_first() -> usize { let n = { let buf = [0u8; 1000]; buf.len() }; YieldNow(false).await; n}
fn main() { println!( "the YieldNow it waits on: {} byte", size_of::<YieldNow>() ); println!( "nothing kept across the await: {} bytes", size_of_val(¬hing_kept()) ); println!( "[u8; 1000] kept across it: {} bytes", size_of_val(&array_kept()) ); println!( "[u8; 1000] used up before it: {} bytes", size_of_val(&array_finished_first()) );}the YieldNow it waits on: 1 byte nothing kept across the await: 2 bytes [u8; 1000] kept across it: 1002 bytes [u8; 1000] used up before it: 16 bytes
With nothing to keep, the future is 2 bytes. A [u8; 1000] used after the wait makes it 1,002. Used up before the wait, the array is not kept and the future is 16 bytes, because the usize n is still needed. The layout is the compiler’s business and may change.
The Rust Book calls this a state machine (Futures and the Async Syntax). The future is that state, stored wherever its owner puts it, and not a pointer to a frame allocated elsewhere, which is how a C++20 coroutine works (see the C++ note). That is why Rust needs Pin. A future may hold a reference into its own fields across an .await, so moving it after the first poll could leave the reference pointing at the old place. poll therefore takes Pin<&mut Self>, a promise that the value will not move (std::pin), unless its type is Unpin, the marker for types that do not mind being moved. The future of an async fn is not Unpin. pin! pins on the stack and Box::pin on the heap; Pin::new needs Unpin, and the compiler points to the first two:
use std::future::Future;use std::pin::Pin;use std::task::{Context, Poll, Waker};async fn f() -> u32 { 1}fn main() { let mut fut = f(); let mut cx = Context::from_waker(Waker::noop()); let p = Pin::new(&mut fut).poll(&mut cx); println!("{:?}", p);}error[E0277]: `{async fn body of f()}` cannot be unpinned --> src/main.rs:10:22 | 4 | async fn f() -> u32 { | --- within this `impl Future<Output = u32>` ... 10 | let p = Pin::new(&mut fut).poll(&mut cx); | -------- ^^^^^^^^ within `impl Future<Output = u32>`, the trait `Unpin` is not implemented for `{async fn body of f()}` | | | required by a bound introduced by this call | = note: consider using the `pin!` macro consider using `Box::pin` if you need to access the pinned value outside of the current scope note: required because it appears within the type `impl Future<Output = u32>` --> src/main.rs:4:17 | 4 | async fn f() -> u32 { | ^^^ ... (notes pointing into the standard library trimmed)
Dropping a future cancels it
Dropping a future between polls ends the work: everything it held is dropped and the rest of the body never runs. This forward takes an item off a queue, waits, then delivers it:
use std::cell::RefCell;use std::collections::VecDeque;use std::future::Future;use std::pin::{Pin, pin};use std::task::{Context, Poll, Waker};
struct YieldNow(bool);
impl Future for YieldNow { type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { if self.0 { return Poll::Ready(()); } self.0 = true; cx.waker().wake_by_ref(); Poll::Pending }}
struct Item(u32);
impl Drop for Item { fn drop(&mut self) { println!(" dropped item {}", self.0); }}
type Queue = RefCell<VecDeque<Item>>;
/// Takes one item off the queue, waits, then delivers it.async fn forward(queue: &Queue) { let item = queue.borrow_mut().pop_front().unwrap(); YieldNow(false).await; // a drop can land here println!(" delivered item {}", item.0);}
fn queue_of_three() -> Queue { RefCell::new((1..=3).map(Item).collect())}
fn main() { let mut cx = Context::from_waker(Waker::noop());
println!("polled until ready:"); let queue = queue_of_three(); let mut fut = pin!(forward(&queue)); while fut.as_mut().poll(&mut cx).is_pending() {} println!(" left in the queue: {}", queue.borrow().len());
println!("polled once, then dropped:"); let queue = queue_of_three(); { let mut fut = pin!(forward(&queue)); let _ = fut.as_mut().poll(&mut cx); println!(" dropping the future now"); } println!(" left in the queue: {}", queue.borrow().len()); println!("end of main: both queues drop what is left");}polled until ready: delivered item 1 dropped item 1 left in the queue: 2 polled once, then dropped: dropping the future now dropped item 1 left in the queue: 2 end of main: both queues drop what is left dropped item 2 dropped item 3 dropped item 2 dropped item 3
In the second run delivered never printed and dropped item 1 did: cleanup runs through Drop (Common traits), which is a plain function and cannot .await (check-drop-await.fail.rs). The item was also lost: it left the queue and was never delivered. A future is cancellation safe if being dropped at an .await loses nothing; forward is not. The place this bites is select!. Tokio’s select! returns “when the first branch completes, cancelling the remaining branches”, and its docs list which methods are cancellation safe. timeout cancels its future the same way (Tokio guide).
Many tasks on one thread, and blocking
This executor loops over a Vec<Pin<Box<dyn Future>>>, one pointer type for futures of different types (Trait objects), and its tasks write to a shared log:
use std::cell::RefCell;use std::future::Future;use std::pin::Pin;use std::rc::Rc;use std::sync::{Mutex, TryLockError};use std::task::{Context, Poll, Waker};use std::thread;use std::time::{Duration, Instant};
type Task = Pin<Box<dyn Future<Output = ()>>>;
/// What the tasks write down, and how long after the start they wrote it.#[derive(Clone)]struct Log { start: Instant, entries: Rc<RefCell<Vec<(String, Duration)>>>,}
impl Log { fn new() -> Log { Log { start: Instant::now(), entries: Rc::default(), } } fn push(&self, text: impl Into<String>) { self.entries .borrow_mut() .push((text.into(), self.start.elapsed())); } fn text(&self) -> String { let names: Vec<String> = self .entries .borrow() .iter() .map(|(t, _)| t.clone()) .collect(); names.join(" ") } fn when(&self, text: &str) -> Duration { self.entries .borrow() .iter() .find(|(t, _)| t == text) .unwrap() .1 }}
/// Pending once, then ready. `.await`ing it hands the thread back to the executor.struct YieldNow(bool);
impl Future for YieldNow { type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { if self.0 { return Poll::Ready(()); } self.0 = true; cx.waker().wake_by_ref(); Poll::Pending }}
/// One thread, tasks polled in turn: the smallest executor.fn run_all(mut tasks: Vec<Task>) { let mut cx = Context::from_waker(Waker::noop()); while !tasks.is_empty() { let mut i = 0; while i < tasks.len() { match tasks[i].as_mut().poll(&mut cx) { Poll::Ready(()) => { drop(tasks.remove(i)); // finished: drop the future } Poll::Pending => i += 1, } } }}
/// Three steps, awaiting between them.async fn polite(name: &'static str, log: Log) { for step in 1..=3 { log.push(format!("{name}{step}")); YieldNow(false).await; }}
/// Three steps with no `.await` between them. The sleep stands in for a call that blocks,/// such as a read from a slow disk.async fn blocker(name: &'static str, log: Log) { for step in 1..=3 { log.push(format!("{name}{step}")); thread::sleep(Duration::from_millis(20)); }}
/// Takes a std lock and waits with the guard still alive.async fn holder(lock: Rc<Mutex<u32>>, log: Log) { let _guard = lock.lock().unwrap(); log.push("holder: has the lock, waits"); YieldNow(false).await; log.push("holder: done");}
/// `try_lock` reports what `lock` would have done: block this thread.async fn asker(lock: Rc<Mutex<u32>>, log: Log) { match lock.try_lock() { Err(TryLockError::WouldBlock) => log.push("asker: lock() would block the only thread"), _ => log.push("asker: got the lock"), }}
fn main() { let log = Log::new(); run_all(vec![ Box::pin(polite("a", log.clone())), Box::pin(polite("b", log.clone())), ]); println!("both await between steps: {}", log.text());
let log = Log::new(); run_all(vec![ Box::pin(blocker("a", log.clone())), Box::pin(polite("b", log.clone())), ]); println!("a never awaits: {}", log.text()); println!( "b1 came after a's three 20 ms sleeps: {}", log.when("b1") >= Duration::from_millis(60) );
let log = Log::new(); let lock = Rc::new(Mutex::new(0)); run_all(vec![ Box::pin(holder(lock.clone(), log.clone())), Box::pin(asker(lock.clone(), log.clone())), ]); for (line, _) in log.entries.borrow().iter() { println!("{line}"); }}both await between steps: a1 b1 a2 b2 a3 b3 a never awaits: a1 a2 a3 b1 b2 b3 b1 came after a's three 20 ms sleeps: true holder: has the lock, waits asker: lock() would block the only thread holder: done
polite awaits between steps, so the tasks interleave. blocker has no .await, and thread::sleep stands in for a blocking call. It ran to the end before b got a turn, and b1 came after the three sleeps: while a poll has not returned, the other tasks on that thread wait. Node’s event loop has the same rule. The Future docs say slow work should be “offloaded to a thread pool (or something similar)”; the Tokio guide names the tool, spawn_blocking.
The last demo holds a std Mutex guard across an .await. The second task’s lock() would block the only thread, and the holder could never resume: I ran that version and it hung until a timeout. The page uses try_lock, which reports WouldBlock. On a multi-thread runtime lock() ties up a worker thread instead.
Async or threads?
std has native OS threads (Threads) and futures without a runtime. The Async Book says async programming is “a good fit for systems which need to handle very many concurrent tasks and where those tasks spend a lot of time waiting”. Two other reasons: cancellation, since a JoinHandle can join a thread or check is_finished, not stop it, while a future can be dropped; and crates, since axum, reqwest, tonic and sqlx are built on Tokio (guide).
Async does not split one task’s computation across threads, and long CPU work stalls the thread it runs on. Rayon and Tokio together sends parallel CPU work to rayon and a blocking call to spawn_blocking. A reasonable default is threads, until your program mostly waits on many things or needs timeouts.
What you can read now
The Tokio guide covers #[tokio::main], spawn, join!, timeout, select! (with what it drops), JoinHandle errors, spawn_blocking, and a lock held across an .await. It does not cover the rest of this list. ? works in an async fn that returns a Result (Errors; check-question-mark.rs). Async closures (async |n| ...) compile on stable 1.99.1 (check-async-closure.rs). Streams (async iterators) have no stable std trait (check-stream-unstable.fail.rs), so you meet them in crates. An async fn in a trait works, but the trait is not dyn compatible (check-dyn-async.fail.rs).
Practice
Try it
Replace the todo!() calls until main prints all checks passed. The starters compile and stop at the first todo!().
Lazy, in your head first
Three scenarios create futures and run them in different orders. Before you run anything, write in predictions() what each log will hold; the checks run the real code and compare. One scenario never runs its future.
#![forbid(unsafe_code)]
use std::cell::RefCell;use std::future::Future;use std::pin::pin;use std::rc::Rc;use std::task::{Context, Poll, Waker};
type Log = Rc<RefCell<Vec<&'static str>>>;
fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return value; } }}
/// An async fn that records its name when its body runs.async fn step(name: &'static str, log: Log) { log.borrow_mut().push(name);}
/// Scenario 1: create one future, record "created", then run it.fn create_then_run() -> Vec<&'static str> { let log: Log = Rc::default(); let future = step("body", log.clone()); log.borrow_mut().push("created"); block_on(future); log.borrow().clone()}
/// Scenario 2: create `a` and then `b`, but run `b` first.fn run_in_reverse() -> Vec<&'static str> { let log: Log = Rc::default(); let a = step("a", log.clone()); let b = step("b", log.clone()); block_on(b); block_on(a); log.borrow().clone()}
/// Scenario 3: create a future, record "created", and never run it.fn never_run() -> Vec<&'static str> { let log: Log = Rc::default(); let future = step("body", log.clone()); log.borrow_mut().push("created"); drop(future); log.borrow().clone()}
/// What each scenario's log holds at the end, in order. Write your prediction,/// then let the checks in `main` run the real code.fn predictions() -> [Vec<&'static str>; 3] { todo!("one Vec per scenario: the words each log holds, in order")}
fn main() { let [first, second, third] = predictions(); assert_eq!(first, create_then_run(), "scenario 1"); assert_eq!(second, run_in_reverse(), "scenario 2"); assert_eq!(third, never_run(), "scenario 3"); println!("all checks passed");}Open in the Rust Playground ↗Run it first: it stops at todo!() with not yet implemented and a hint, and the compiler warns that the parameter is unused. Both are expected. Replace the todo!() with your code and run it again until it prints all checks passed.
Show a solution
#![forbid(unsafe_code)]
use std::cell::RefCell;use std::future::Future;use std::pin::pin;use std::rc::Rc;use std::task::{Context, Poll, Waker};
type Log = Rc<RefCell<Vec<&'static str>>>;
fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return value; } }}
/// An async fn that records its name when its body runs.async fn step(name: &'static str, log: Log) { log.borrow_mut().push(name);}
/// Scenario 1: create one future, record "created", then run it.fn create_then_run() -> Vec<&'static str> { let log: Log = Rc::default(); let future = step("body", log.clone()); log.borrow_mut().push("created"); block_on(future); log.borrow().clone()}
/// Scenario 2: create `a` and then `b`, but run `b` first.fn run_in_reverse() -> Vec<&'static str> { let log: Log = Rc::default(); let a = step("a", log.clone()); let b = step("b", log.clone()); block_on(b); block_on(a); log.borrow().clone()}
/// Scenario 3: create a future, record "created", and never run it.fn never_run() -> Vec<&'static str> { let log: Log = Rc::default(); let future = step("body", log.clone()); log.borrow_mut().push("created"); drop(future); log.borrow().clone()}
/// What each scenario's log holds at the end, in order. Write your prediction,/// then let the checks in `main` run the real code.fn predictions() -> [Vec<&'static str>; 3] { [vec!["created", "body"], vec!["b", "a"], vec!["created"]]}
fn main() { let [first, second, third] = predictions(); assert_eq!(first, create_then_run(), "scenario 1"); assert_eq!(second, run_in_reverse(), "scenario 2"); assert_eq!(third, never_run(), "scenario 3"); println!("all checks passed");}all checks passed
Race two futures
Finish race_two: poll two futures in turn on one thread, a first, and return the output of whichever is ready first. The loser is cancelled: dropped, and never polled again. The checks fail a version that awaits one after the other, keeps polling the loser, polls b first, or leaks the loser (its Drop must run). This is the shape of select!.
#![forbid(unsafe_code)]
use std::cell::RefCell;use std::future::{Future, poll_fn};use std::pin::{Pin, pin};use std::rc::Rc;use std::task::{Context, Poll, Waker};
type Log = Rc<RefCell<Vec<String>>>;
fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return value; } }}
/// Polls `a` and `b` in turn on this thread, `a` first, and returns the output of/// whichever is ready first. The other future is dropped, and never polled again.async fn race_two<A, B>(a: A, b: B) -> A::Outputwhere A: Future, B: Future<Output = A::Output>,{ todo!( "pin both futures, then poll_fn: poll a; if it is not ready, poll b; return the first Ready" )}
/// Pending `steps` times, recording "name1", "name2", ..., then ready with `answer`./// Records "drop name" when it is dropped. Panics if it is polled after it has finished.struct Steps { name: &'static str, steps: u32, done: u32, answer: u32, finished: bool, log: Log,}
fn steps(name: &'static str, steps: u32, answer: u32, log: &Log) -> Steps { Steps { name, steps, done: 0, answer, finished: false, log: log.clone(), }}
impl Future for Steps { type Output = u32;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> { assert!(!self.finished, "{} was polled after it finished", self.name); if self.done == self.steps { self.finished = true; return Poll::Ready(self.answer); } self.done += 1; let entry = format!("{}{}", self.name, self.done); self.log.borrow_mut().push(entry); cx.waker().wake_by_ref(); Poll::Pending }}
impl Drop for Steps { fn drop(&mut self) { let entry = format!("drop {}", self.name); self.log.borrow_mut().push(entry); }}
fn main() { let log: Log = Rc::default();
// a finishes first: its answer wins, and b is dropped half-way let winner = block_on(race_two(steps("a", 2, 7, &log), steps("b", 3, 8, &log))); assert_eq!(winner, 7); let text = log.borrow().join(" "); // they took turns, a first, and nobody was polled after the race was decided assert!(text.starts_with("a1 b1 a2 b2 "), "turns: {text}"); assert!(!text.contains("b3"), "b ran on after a had won: {text}"); // by the time race_two has returned, the loser has been dropped (cancelled) assert_eq!( text.matches("drop b").count(), 1, "the loser must be dropped: {text}" );
// b finishes first log.borrow_mut().clear(); let winner = block_on(race_two(steps("a", 3, 7, &log), steps("b", 1, 8, &log))); assert_eq!(winner, 8); let text = log.borrow().join(" "); assert!(text.starts_with("a1 b1 a2 "), "turns: {text}"); assert!(!text.contains("a3"), "a ran on after b had won: {text}"); assert_eq!( text.matches("drop a").count(), 1, "the loser must be dropped: {text}" );
// both are ready at once: a is polled first, so a wins and b is never polled log.borrow_mut().clear(); let winner = block_on(race_two(steps("a", 0, 7, &log), steps("b", 0, 8, &log))); assert_eq!(winner, 7); let text = log.borrow().join(" "); assert!(!text.contains("b1"), "b was polled: {text}"); assert_eq!( text.matches("drop b").count(), 1, "the loser must be dropped: {text}" );
println!("all checks passed");}Open in the Rust Playground ↗Run it first: it stops at todo!() with not yet implemented and a hint, and the compiler warns that the parameter is unused. Both are expected. Replace the todo!() with your code and run it again until it prints all checks passed.
Show a solution
#![forbid(unsafe_code)]
use std::cell::RefCell;use std::future::{Future, poll_fn};use std::pin::{Pin, pin};use std::rc::Rc;use std::task::{Context, Poll, Waker};
type Log = Rc<RefCell<Vec<String>>>;
fn block_on<F: Future>(fut: F) -> F::Output { let mut fut = pin!(fut); let mut cx = Context::from_waker(Waker::noop()); loop { if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { return value; } }}
/// Polls `a` and `b` in turn on this thread, `a` first, and returns the output of/// whichever is ready first. The other future is dropped, and never polled again.async fn race_two<A, B>(a: A, b: B) -> A::Outputwhere A: Future, B: Future<Output = A::Output>,{ let mut a = pin!(a); let mut b = pin!(b); poll_fn(|cx| { if let Poll::Ready(value) = a.as_mut().poll(cx) { return Poll::Ready(value); } b.as_mut().poll(cx) }) .await}
/// Pending `steps` times, recording "name1", "name2", ..., then ready with `answer`./// Records "drop name" when it is dropped. Panics if it is polled after it has finished.struct Steps { name: &'static str, steps: u32, done: u32, answer: u32, finished: bool, log: Log,}
fn steps(name: &'static str, steps: u32, answer: u32, log: &Log) -> Steps { Steps { name, steps, done: 0, answer, finished: false, log: log.clone(), }}
impl Future for Steps { type Output = u32;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> { assert!(!self.finished, "{} was polled after it finished", self.name); if self.done == self.steps { self.finished = true; return Poll::Ready(self.answer); } self.done += 1; let entry = format!("{}{}", self.name, self.done); self.log.borrow_mut().push(entry); cx.waker().wake_by_ref(); Poll::Pending }}
impl Drop for Steps { fn drop(&mut self) { let entry = format!("drop {}", self.name); self.log.borrow_mut().push(entry); }}
fn main() { let log: Log = Rc::default();
// a finishes first: its answer wins, and b is dropped half-way let winner = block_on(race_two(steps("a", 2, 7, &log), steps("b", 3, 8, &log))); assert_eq!(winner, 7); let text = log.borrow().join(" "); // they took turns, a first, and nobody was polled after the race was decided assert!(text.starts_with("a1 b1 a2 b2 "), "turns: {text}"); assert!(!text.contains("b3"), "b ran on after a had won: {text}"); // by the time race_two has returned, the loser has been dropped (cancelled) assert_eq!( text.matches("drop b").count(), 1, "the loser must be dropped: {text}" );
// b finishes first log.borrow_mut().clear(); let winner = block_on(race_two(steps("a", 3, 7, &log), steps("b", 1, 8, &log))); assert_eq!(winner, 8); let text = log.borrow().join(" "); assert!(text.starts_with("a1 b1 a2 "), "turns: {text}"); assert!(!text.contains("a3"), "a ran on after b had won: {text}"); assert_eq!( text.matches("drop a").count(), 1, "the loser must be dropped: {text}" );
// both are ready at once: a is polled first, so a wins and b is never polled log.borrow_mut().clear(); let winner = block_on(race_two(steps("a", 0, 7, &log), steps("b", 0, 8, &log))); assert_eq!(winner, 7); let text = log.borrow().join(" "); assert!(!text.contains("b1"), "b was polled: {text}"); assert_eq!( text.matches("drop b").count(), 1, "the loser must be dropped: {text}" );
println!("all checks passed");}all checks passed
A task that can change threads
register adds a name to a shared list, waits, and returns the list’s length after the wait. main requires the future to be Send and checks that nobody holds the lock during the wait. Write the natural draft first, guard held across the .await, and read the error; then fix it. drop(guard) after guard.push(..) does not fix it on 1.99.1. The file forbids unsafe.
#![forbid(unsafe_code)]
use std::future::Future;use std::pin::{Pin, pin};use std::sync::{Arc, Mutex};use std::task::{Context, Poll, Waker};
/// Pending once, then ready: stands in for waiting on a file or a socket.struct YieldNow(bool);
impl Future for YieldNow { type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { if self.0 { return Poll::Ready(()); } self.0 = true; cx.waker().wake_by_ref(); Poll::Pending }}
type Names = Arc<Mutex<Vec<String>>>;
/// Adds `name` to the shared list, waits (`YieldNow(false).await`), and returns how/// long the list is AFTER the wait. The returned future must be `Send`, so a runtime/// may move it to another thread while it waits. `main` checks that, and checks that/// other tasks can use the list while this one waits.async fn register(names: Names, name: &str) -> usize { todo!("push the name, wait, then read the length. Mind what is alive across the await")}
/// Compile-time check: only accepts futures that are `Send`.fn assert_send<T: Send>(_: &T) {}
fn main() { let names: Names = Arc::default(); let mut cx = Context::from_waker(Waker::noop());
let first = register(names.clone(), "ann"); let second = register(names.clone(), "bob"); assert_send(&first); assert_send(&second); let mut first = pin!(first); let mut second = pin!(second);
// the first task adds its name, then waits assert_eq!(first.as_mut().poll(&mut cx), Poll::Pending); assert_eq!(*names.lock().unwrap(), ["ann"]);
// while it waits, nobody may be holding the lock assert!( names.try_lock().is_ok(), "the lock is still held while the task waits" );
// the second task gets in during that wait assert_eq!(second.as_mut().poll(&mut cx), Poll::Pending); assert_eq!(*names.lock().unwrap(), ["ann", "bob"]);
// both finish and see both names: the length is read after the wait assert_eq!(first.as_mut().poll(&mut cx), Poll::Ready(2)); assert_eq!(second.as_mut().poll(&mut cx), Poll::Ready(2));
// a task that starts after the others are done sees all three names let third = register(names.clone(), "cy"); assert_send(&third); let mut third = pin!(third); assert_eq!(third.as_mut().poll(&mut cx), Poll::Pending); assert_eq!(third.as_mut().poll(&mut cx), Poll::Ready(3));
println!("all checks passed");}Open in the Rust Playground ↗Run it first: it stops at todo!() with not yet implemented and a hint, and the compiler warns that the parameter is unused. Both are expected. Replace the todo!() with your code and run it again until it prints all checks passed.
Show a solution
#![forbid(unsafe_code)]
use std::future::Future;use std::pin::{Pin, pin};use std::sync::{Arc, Mutex};use std::task::{Context, Poll, Waker};
/// Pending once, then ready: stands in for waiting on a file or a socket.struct YieldNow(bool);
impl Future for YieldNow { type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { if self.0 { return Poll::Ready(()); } self.0 = true; cx.waker().wake_by_ref(); Poll::Pending }}
type Names = Arc<Mutex<Vec<String>>>;
/// Adds `name` to the shared list, waits (`YieldNow(false).await`), and returns how/// long the list is AFTER the wait. The returned future must be `Send`, so a runtime/// may move it to another thread while it waits. `main` checks that, and checks that/// other tasks can use the list while this one waits.async fn register(names: Names, name: &str) -> usize { names.lock().unwrap().push(name.to_string()); // the guard is a temporary: gone at the semicolon YieldNow(false).await; let n = names.lock().unwrap().len(); n}
/// Compile-time check: only accepts futures that are `Send`.fn assert_send<T: Send>(_: &T) {}
fn main() { let names: Names = Arc::default(); let mut cx = Context::from_waker(Waker::noop());
let first = register(names.clone(), "ann"); let second = register(names.clone(), "bob"); assert_send(&first); assert_send(&second); let mut first = pin!(first); let mut second = pin!(second);
// the first task adds its name, then waits assert_eq!(first.as_mut().poll(&mut cx), Poll::Pending); assert_eq!(*names.lock().unwrap(), ["ann"]);
// while it waits, nobody may be holding the lock assert!( names.try_lock().is_ok(), "the lock is still held while the task waits" );
// the second task gets in during that wait assert_eq!(second.as_mut().poll(&mut cx), Poll::Pending); assert_eq!(*names.lock().unwrap(), ["ann", "bob"]);
// both finish and see both names: the length is read after the wait assert_eq!(first.as_mut().poll(&mut cx), Poll::Ready(2)); assert_eq!(second.as_mut().poll(&mut cx), Poll::Ready(2));
// a task that starts after the others are done sees all three names let third = register(names.clone(), "cy"); assert_send(&third); let mut third = pin!(third); assert_eq!(third.as_mut().poll(&mut cx), Poll::Pending); assert_eq!(third.as_mut().poll(&mut cx), Poll::Ready(3));
println!("all checks passed");}all checks passed
Read more
- The Rust Book, chapter 17: Fundamentals of Asynchronous Programming, and Asynchronous Programming in Rust.
- The standard library:
Future,std::task,std::pin,poll_fn. - The Reference: await expressions, and RFC 230 on why
stdhas no runtime. - The runtime itself: Async with Tokio. Related lessons: Threads, Shared state, Send and Sync.
- Errors: E0373, E0728, E0752, or this site’s index.
Learn Rust is an independent project, not affiliated with the Rust Project or the Rust Foundation. Code and text are under the MIT licence.Source on GitHub.