Async Programming with Tokio

15 min read•

Advanced

Learn how to write asynchronous Rust code with Tokio, comparing it to Python's asyncio.

Setting Up Tokio

First, add Tokio to your Cargo.toml:

Basic Setup

Python

# Python: Built into the standard library
import asyncio

# Python 3.7+
async def main():
    print("Hello, asyncio!")

asyncio.run(main())

Rust

// Rust: Add to Cargo.toml
// [dependencies]
// tokio = { version = "1.0", features = ["full"] }

#[tokio::main]
async fn main() {
    println!("Hello, Tokio!");
}

Async Functions

Python

import asyncio

async def fetch_data():
    print("Fetching data...")
    await asyncio.sleep(1)
    return {"data": 42}

async def main():
    result = await fetch_data()
    print(f"Got data: {result}")

asyncio.run(main())

Rust

use std::time::Duration;
use tokio::time::sleep;

async fn fetch_data() -> serde_json::Value {
    println!("Fetching data...");
    sleep(Duration::from_secs(1)).await;
    serde_json::json!({ "data": 42 })
}

#[tokio::main]
async fn main() {
    let result = fetch_data().await;
    println!("Got data: {}", result);
}

Tasks and Concurrency

Concurrent Tasks

Python

import asyncio

async def task(name, seconds):
    print(f"Task {name} started")
    await asyncio.sleep(seconds)
    print(f"Task {name} completed")
    return f"Task {name} result"

async def main():
    # Run tasks concurrently
    tasks = [
        asyncio.create_task(task("A", 2)),
        asyncio.create_task(task("B", 1)),
        asyncio.create_task(task("C", 3))
    ]
    
    # Wait for all tasks to complete
    results = await asyncio.gather(*tasks)
    print("All tasks completed:", results)

asyncio.run(main())

Rust

use std::time::Duration;
use tokio::time::sleep;

async fn task(name: &str, seconds: u64) -> String {
    println!("Task {} started", name);
    sleep(Duration::from_secs(seconds)).await;
    println!("Task {} completed", name);
    format!("Task {} result", name)
}

#[tokio::main]
async fn main() {
    // Spawn tasks to run concurrently
    let task_a = tokio::spawn(task("A", 2));
    let task_b = tokio::spawn(task("B", 1));
    let task_c = tokio::spawn(task("C", 3));

    // Wait for all tasks to complete
    let (a, b, c) = tokio::join!(
        task_a,
        task_b,
        task_c
    );

    println!("All tasks completed: {:?}", (a, b, c));
}

Making HTTP Requests

HTTP Requests

Python

import aiohttp
import asyncio

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = [
        'https://httpbin.org/get',
        'https://api.github.com',
        'https://www.rust-lang.org'
    ]
    
    tasks = [fetch_url(url) for url in urls]
    results = await asyncio.gather(*tasks)
    
    for url, content in zip(urls, results):
        print(f"{url} returned {len(content)} bytes")

asyncio.run(main())

Rust

use reqwest;
use std::error::Error;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let urls = [
        "https://httpbin.org/get",
        "https://api.github.com",
        "https://www.rust-lang.org"
    ];
    
    let client = reqwest::Client::new();
    let mut tasks = vec![];
    
    for url in urls.iter() {
        let client = client.clone();
        let url = url.to_string();
        
        tasks.push(tokio::spawn(async move {
            match client.get(&url).send().await {
                Ok(resp) => {
                    let text = resp.text().await.unwrap_or_default();
                    (url, text.len())
                }
                Err(_) => (url, 0)
            }
        }));
    }
    
    for task in tasks {
        if let Ok((url, len)) = task.await {
            println!("{} returned {} bytes", url, len);
        }
    }
    
    Ok(())
}

Note: Don't forget to add reqwest = { version = "0.11", features = ["json"] } to your Cargo.toml for the HTTP client.

Channels for Communication

Channels

Python

import asyncio

async def producer(queue, id):
    for i in range(3):
        msg = f"Message {i} from producer {id}"
        await queue.put(msg)
        print(f"Produced: {msg}")
        await asyncio.sleep(0.5)
    await queue.put(None)  # Sentinel value

async def consumer(queue, id):
    while True:
        msg = await queue.get()
        if msg is None:  # Check for sentinel
            await queue.put(None)  # Notify other consumers
            break
        print(f"Consumer {id} got: {msg}")
        await asyncio.sleep(0.1)  # Simulate work
    print(f"Consumer {id} done")

async def main():
    queue = asyncio.Queue()
    
    # Start producers and consumers
    producers = [asyncio.create_task(producer(queue, i)) for i in range(2)]
    consumers = [asyncio.create_task(consumer(queue, i)) for i in range(3)]
    
    # Wait for producers to finish
    await asyncio.gather(*producers)
    
    # Signal consumers to exit
    await queue.put(None)
    
    # Wait for consumers to finish
    await asyncio.gather(*consumers)

asyncio.run(main())

Rust

use tokio::sync::mpsc;
use std::time::Duration;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);
    let mut handles = vec![];
    
    // Spawn producers
    for i in 0..2 {
        let tx = tx.clone();
        handles.push(tokio::spawn(async move {
            for j in 0..3 {
                let msg = format!("Message {} from producer {}", j, i);
                tx.send(Some(msg.clone())).await.unwrap();
                println!("Produced: {}", msg);
                tokio::time::sleep(Duration::from_millis(500)).await;
            }
            // Send None to signal completion
            let _ = tx.send(None).await;
        }));
    }
    
    // Drop the original sender
    drop(tx);
    
    // Spawn consumers
    let mut consumer_handles = vec![];
    for i in 0..3 {
        let mut rx = rx.clone();
        consumer_handles.push(tokio::spawn(async move {
            while let Some(Some(msg)) = rx.recv().await {
                println!("Consumer {} got: {}", i, msg);
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
            println!("Consumer {} done", i);
        }));
    }
    
    // Wait for all producers and consumers to complete
    for handle in handles.into_iter().chain(consumer_handles) {
        let _ = handle.await;
    }
}

Error Handling

Python

import asyncio

async def might_fail():
    await asyncio.sleep(1)
    if True:  # Simulate an error
        raise ValueError("Something went wrong")
    return "Success"

async def main():
    try:
        result = await might_fail()
        print(f"Result: {result}")
    except Exception as e:
        print(f"Error: {e}")
    
    # Handle multiple errors with gather
    tasks = [might_fail() for _ in range(3)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i} failed: {result}")
        else:
            print(f"Task {i} succeeded: {result}")

asyncio.run(main())

Rust

use std::time::Duration;
use tokio::time::sleep;

async fn might_fail() -> Result<String, String> {
    sleep(Duration::from_secs(1)).await;
    if true { // Simulate an error
        return Err("Something went wrong".to_string());
    }
    Ok("Success".to_string())
}

#[tokio::main]
async fn main() {
    // Handle single error
    match might_fail().await {
        Ok(result) => println!("Result: {}", result),
        Err(e) => println!("Error: {}", e),
    }
    
    // Handle multiple errors with join_all
    let tasks = vec![
        tokio::spawn(might_fail()),
        tokio::spawn(might_fail()),
        tokio::spawn(might_fail()),
    ];
    
    let results = futures::future::join_all(tasks).await;
    
    for (i, result) in results.into_iter().enumerate() {
        match result {
            Ok(Ok(value)) => println!("Task {} succeeded: {}", i, value),
            Ok(Err(e)) => println!("Task {} failed: {}", i, e),
            Err(join_err) => println!("Task {} panicked: {}", i, join_err),
        }
    }
}

When to Use Python vs Rust

Both Python's asyncio and Rust's Tokio are powerful tools, but they excel in different scenarios. Here's when you might choose one over the other:

Choose Python's asyncio when:

You need rapid development and prototyping
Your team is more familiar with Python
You're building web applications (especially with frameworks like FastAPI or Django)
You need access to Python's extensive ecosystem of libraries
Your application is I/O bound but not CPU intensive
You're working on data processing scripts or automation

Choose Rust's Tokio when:

You need maximum performance and low-level control
Memory safety without garbage collection is important
You're building high-performance network services or systems programming
You need predictable performance with no garbage collection pauses
You want to minimize resource usage (CPU, memory, energy)
You need to handle thousands of concurrent connections efficiently

Hybrid Approach

Consider using both! Many projects use Python for high-level application logic and Rust for performance-critical components. You can expose Rust code to Python using PyO3 for the best of both worlds.

Key Takeaways

Tokio provides a powerful runtime for async Rust, similar to Python's asyncio but with Rust's performance benefits
Async/await syntax is similar between Python and Rust, but Rust requires explicit handling of errors and lifetimes
Use tokio::spawn to run concurrent tasks (like asyncio.create_task)
Channels in Tokio (mpsc) are similar to Python's asyncio.Queue but more type-safe
For HTTP clients, reqwest is the most popular choice in the Rust ecosystem

Previous: Rust Modules vs Python Packages Back to All Tutorials