Asynchronous programming is essential for high-performance I/O-bound applications. This chapter explores how Rust's async/await works under the hood and how to use it effectively.
At the core of async Rust is the Future trait. Let's understand what it really means:
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
// The actual Future trait (simplified)
pub trait SimpleFuture {
type Output;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}
// Poll is an enum
pub enum Poll<T> {
Ready(T), // Future completed with value
Pending, // Future not ready yet
}
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
struct Delay {
when: Instant,
}
impl Delay {
fn new(duration: Duration) -> Self {
Delay {
when: Instant::now() + duration,
}
}
}
impl Future for Delay {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
if Instant::now() >= self.when {
Poll::Ready(())
} else {
// Tell the executor to poll again
cx.waker().wake_by_ref();
Poll::Pending
}
}
}
#[tokio::main]
async fn main() {
let delay = Delay::new(Duration::from_secs(1));
delay.await;
println!("Done!");
}
// Async function
async fn fetch_data(url: &str) -> Result<String, String> {
// Simulated async work
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
Ok(format!("Data from {}", url))
}
// Async block
fn create_future() -> impl Future<Output = i32> {
async {
42
}
}
// Using async/await
#[tokio::main]
async fn main() {
// Sequential
let data1 = fetch_data("api.example.com/1").await.unwrap();
let data2 = fetch_data("api.example.com/2").await.unwrap();
println!("{}\n{}", data1, data2);
}
use tokio::join;
use tokio::time::{sleep, Duration};
async fn task1() -> String {
sleep(Duration::from_millis(100)).await;
"Task 1 done".to_string()
}
async fn task2() -> String {
sleep(Duration::from_millis(100)).await;
"Task 2 done".to_string()
}
#[tokio::main]
async fn main() {
// Run concurrently (not parallel - same thread!)
let (result1, result2) = join!(task1(), task2());
println!("{}", result1);
println!("{}", result2);
}
// For dynamic number of tasks
async fn fetch_all(urls: Vec<String>) -> Vec<Result<String, String>> {
let tasks: Vec<_> = urls
.into_iter()
.map(|url| tokio::spawn(async move {
fetch_data(&url).await
}))
.collect();
let mut results = Vec::new();
for task in tasks {
results.push(task.await.unwrap());
}
results
}
use tokio::runtime::Runtime;
fn main() {
// Multi-threaded runtime (default)
let runtime = Runtime::new().unwrap();
runtime.block_on(async {
println!("Running on multi-threaded runtime");
});
// Current thread runtime (single-threaded)
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
runtime.block_on(async {
println!("Running on current thread");
});
// Custom configuration
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(4)
.thread_name("my-worker")
.thread_stack_size(3 * 1024 * 1024)
.enable_all()
.build()
.unwrap();
}
use tokio::task;
#[tokio::main]
async fn main() {
// Spawn a task
let handle = task::spawn(async {
println!("Task running on tokio thread pool");
42
});
let result = handle.await.unwrap();
println!("Result: {}", result);
// Spawn blocking (for CPU-intensive work)
let handle = task::spawn_blocking(|| {
// This runs on a dedicated thread pool
println!("CPU-intensive work");
expensive_computation()
});
handle.await.unwrap();
}
fn expensive_computation() -> u64 {
(0..1_000_000).sum()
}
use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
let (tx, mut rx) = mpsc::channel(32);
// Spawn sender task
tokio::spawn(async move {
for i in 0..10 {
tx.send(i).await.unwrap();
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
}
});
// Receive in main task
while let Some(value) = rx.recv().await {
println!("Received: {}", value);
}
}
// Bounded vs Unbounded
async fn channel_types() {
// Bounded: sender waits if full
let (tx, rx) = mpsc::channel::<i32>(100);
// Unbounded: never blocks (use carefully!)
let (tx_unbounded, rx_unbounded) = mpsc::unbounded_channel::<i32>();
}
use tokio::sync::{broadcast, watch};
#[tokio::main]
async fn main() {
// Broadcast: multiple receivers
let (tx, _rx) = broadcast::channel(16);
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();
tokio::spawn(async move {
tx.send("Hello").unwrap();
});
println!("rx1: {}", rx1.recv().await.unwrap());
println!("rx2: {}", rx2.recv().await.unwrap());
// Watch: single-value state channel
let (tx, mut rx) = watch::channel("initial");
tokio::spawn(async move {
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
tx.send("updated").unwrap();
});
rx.changed().await.unwrap();
println!("Value changed to: {}", *rx.borrow());
}
use tokio::sync::Mutex;
use std::sync::Arc;
#[tokio::main]
async fn main() {
let counter = Arc::new(Mutex::new(0));
let mut handles = vec![];
for _ in 0..10 {
let counter = Arc::clone(&counter);
let handle = tokio::spawn(async move {
let mut num = counter.lock().await;
*num += 1;
});
handles.push(handle);
}
for handle in handles {
handle.await.unwrap();
}
println!("Counter: {}", *counter.lock().await);
}
tokio::sync::Mutex: For holding locks across await pointsstd::sync::Mutex: For short, non-async critical sections (faster)use tokio::sync::{RwLock, Semaphore};
#[tokio::main]
async fn main() {
// RwLock: multiple readers or single writer
let lock = Arc::new(RwLock::new(5));
let read = lock.read().await;
println!("Value: {}", *read);
drop(read);
let mut write = lock.write().await;
*write += 1;
// Semaphore: limit concurrent access
let semaphore = Arc::new(Semaphore::new(3));
let mut handles = vec![];
for i in 0..10 {
let sem = semaphore.clone();
handles.push(tokio::spawn(async move {
let _permit = sem.acquire().await.unwrap();
println!("Task {} acquired permit", i);
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
}));
}
for handle in handles {
handle.await.unwrap();
}
}
Race multiple futures and handle whichever completes first:
use tokio::select;
use tokio::time::{sleep, Duration};
#[tokio::main]
async fn main() {
let mut interval = tokio::time::interval(Duration::from_secs(1));
select! {
_ = sleep(Duration::from_secs(5)) => {
println!("5 second timeout");
}
_ = interval.tick() => {
println!("Interval fired");
}
}
// With cancellation
let operation = async {
sleep(Duration::from_secs(10)).await;
"Completed"
};
let timeout = sleep(Duration::from_secs(2));
select! {
result = operation => {
println!("Operation: {}", result);
}
_ = timeout => {
println!("Timed out!");
}
}
}
use tokio_stream::{self as stream, StreamExt};
#[tokio::main]
async fn main() {
// Create a stream
let mut stream = stream::iter(vec![1, 2, 3, 4, 5]);
// Process with combinator methods
while let Some(value) = stream.next().await {
println!("{}", value);
}
// Advanced stream processing
let stream = stream::iter(1..=10)
.filter(|x| x % 2 == 0)
.map(|x| x * 2);
let results: Vec<_> = stream.collect().await;
println!("{:?}", results);
// Merge multiple streams
let s1 = stream::iter(vec![1, 2, 3]);
let s2 = stream::iter(vec![4, 5, 6]);
let merged = stream::iter(vec![s1, s2]).flatten();
}
use tokio::try_join;
async fn fallible_task1() -> Result<i32, String> {
Ok(1)
}
async fn fallible_task2() -> Result<i32, String> {
Err("Task 2 failed".to_string())
}
#[tokio::main]
async fn main() {
// Early return on first error
match try_join!(fallible_task1(), fallible_task2()) {
Ok((r1, r2)) => println!("Both succeeded: {}, {}", r1, r2),
Err(e) => println!("Error: {}", e),
}
}
// Timeout pattern
use tokio::time::{timeout, Duration};
async fn with_timeout() -> Result<String, String> {
match timeout(Duration::from_secs(5), slow_operation()).await {
Ok(result) => result,
Err(_) => Err("Timeout!".to_string()),
}
}
async fn slow_operation() -> Result<String, String> {
tokio::time::sleep(Duration::from_secs(10)).await;
Ok("Done".to_string())
}