tokio-macros

Procedural macros for Tokio including

tokio-macros provides Tokio’s procedural macros that simplify setting up and testing async code. These macros eliminate boilerplate when creating async entry points and tests.

Installation

The macros are re-exported by the main tokio crate, so you typically don’t need to add tokio-macros directly:

Cargo.toml

[dependencies]
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

The macros feature flag is required to use #[tokio::main] and #[tokio::test].

The #[tokio::main] Macro

The #[tokio::main] macro transforms an async main function into a synchronous one that runs on the Tokio runtime.

Basic Usage

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

This expands to:

fn main() {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            println!("Hello world");
        })
}

The macro handles runtime setup automatically, making async code more ergonomic.

Runtime Flavors

Multi-threaded Runtime (Default)

The default flavor uses multiple threads:

#[tokio::main]
async fn main() {
    // Runs on multi-threaded runtime
    println!("Hello from multiple threads!");
}

Configure Worker Threads

#[tokio::main(worker_threads = 4)]
async fn main() {
    println!("Running on 4 worker threads");
}

By default, the number of worker threads equals the number of CPU cores.

Current Thread Runtime

Use a single-threaded runtime:

#[tokio::main(flavor = "current_thread")]
async fn main() {
    println!("Running on a single thread");
}

This expands to:

fn main() {
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            println!("Running on a single thread");
        })
}

Multi-threaded with Explicit Flavor

#[tokio::main(flavor = "multi_thread", worker_threads = 8)]
async fn main() {
    println!("Explicitly multi-threaded with 8 workers");
}

Advanced Configuration

Start with Time Paused

Requires the test-util feature flag.

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    // Time starts paused - useful for testing
    tokio::time::advance(tokio::time::Duration::from_secs(1)).await;
}

Custom Crate Name

If you’ve renamed the tokio crate:

use tokio as tokio1;

#[tokio1::main(crate = "tokio1")]
async fn main() {
    println!("Hello world");
}

The #[tokio::test] Macro

The #[tokio::test] macro enables writing async tests without manual runtime setup.

Basic Test

#[tokio::test]
async fn my_test() {
    let result = async_operation().await;
    assert_eq!(result, expected_value);
}

This expands to:

#[test]
fn my_test() {
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            let result = async_operation().await;
            assert_eq!(result, expected_value);
        })
}

Multi-threaded Tests

#[tokio::test(flavor = "multi_thread")]
async fn concurrent_test() {
    let handle1 = tokio::spawn(async { 1 });
    let handle2 = tokio::spawn(async { 2 });

    let result1 = handle1.await.unwrap();
    let result2 = handle2.await.unwrap();

    assert_eq!(result1 + result2, 3);
}

Configure Test Workers

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn parallel_test() {
    // Test runs with 2 worker threads
    assert!(true);
}

Tests with Paused Time

Requires the test-util feature.

#[tokio::test(start_paused = true)]
async fn time_based_test() {
    let start = tokio::time::Instant::now();

    // Advance time by 1 second instantly
    tokio::time::advance(tokio::time::Duration::from_secs(1)).await;

    let elapsed = start.elapsed();
    assert!(elapsed >= tokio::time::Duration::from_secs(1));
}

Starting tests with paused time is extremely useful for testing time-dependent code without actual delays.

Complete Examples

Simple HTTP Server

use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Server running on port 8080");

    loop {
        let (mut socket, addr) = listener.accept().await?;
        println!("New connection from: {}", addr);

        tokio::spawn(async move {
            let mut buf = [0; 1024];
            loop {
                let n = socket.read(&mut buf).await.unwrap();
                if n == 0 {
                    return;
                }
                socket.write_all(&buf[0..n]).await.unwrap();
            }
        });
    }
}

Testing Async Functions

use tokio::time::{sleep, Duration};

async fn fetch_data() -> Result<String, ()> {
    sleep(Duration::from_millis(100)).await;
    Ok("data".to_string())
}

#[tokio::test]
async fn test_fetch_data() {
    let result = fetch_data().await;
    assert!(result.is_ok());
    assert_eq!(result.unwrap(), "data");
}

#[tokio::test(start_paused = true)]
async fn test_fetch_data_no_delay() {
    // With paused time, no actual delay occurs
    let result = fetch_data().await;
    assert!(result.is_ok());
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_concurrent_fetches() {
    let handles: Vec<_> = (0..10)
        .map(|_| tokio::spawn(fetch_data()))
        .collect();

    for handle in handles {
        assert!(handle.await.unwrap().is_ok());
    }
}

Current Thread for Deterministic Tests

#[tokio::test(flavor = "current_thread")]
async fn deterministic_test() {
    // Single-threaded ensures deterministic execution order
    let mut counter = 0;

    let task1 = async {
        counter += 1;
        counter
    };

    let task2 = async {
        counter += 2;
        counter
    };

    let result1 = task1.await;
    let result2 = task2.await;

    assert_eq!(result1, 1);
    assert_eq!(result2, 3);
}

Comparison Table

Macro · Purpose

Macro	Purpose	Default Runtime
#[tokio::main]	Async main function	Multi-threaded
#[tokio::test]	Async test function	Current-thread

Runtime Flavors Comparison

Multi-threaded Runtime

• Best for: Production applications, CPU-bound tasks
• Features: Work-stealing scheduler, scales with CPU cores
• Trade-off: More overhead than current-thread

Current-thread Runtime

• Best for: Tests, single-threaded apps, WASM
• Features: Lightweight, deterministic execution
• Trade-off: Cannot utilize multiple cores

When to Use Each Macro

Use #[tokio::main]

• Application entry points
• Production services
• When you need full runtime control

Use #[tokio::test]

• Unit tests for async code
• Integration tests
• When testing time-based logic

Common Patterns

Error Handling

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let result = risky_operation().await?;
    println!("Success: {}", result);
    Ok(())
}

Non-main Async Functions

You can use #[tokio::main] on non-main functions, but this creates a new runtime each time:

#[tokio::main]
async fn process_data() {
    // This spawns a new runtime every call
    // Consider passing a runtime handle instead
}

Using #[tokio::main] on frequently-called functions is inefficient. Consider using Runtime::block_on or passing runtime handles instead.

Requirements

1. Enable macros feature

   Add macros to your Tokio features in Cargo.toml

2. Choose runtime flavor

   Use rt-multi-thread for multi-threaded or rt for current-thread

3. Add test-util for testing

   Include test-util feature for start_paused and time control

Utility Macros

The macros feature also includes essential utility macros for working with multiple concurrent operations.

The join! Macro

Waits on multiple concurrent branches, returning when all branches complete.

use tokio::join;

async fn fetch_user() -> String {
    // async work
    "user".to_string()
}

async fn fetch_posts() -> Vec<String> {
    // async work
    vec!["post1".to_string(), "post2".to_string()]
}

#[tokio::main]
async fn main() {
    let (user, posts) = join!(
        fetch_user(),
        fetch_posts()
    );
    
    println!("User: {}, Posts: {:?}", user, posts);
}

join! runs all futures concurrently on the same task and waits for all to complete, even if some return errors.

Fairness Control

By default, join! rotates which future is polled first. Use biased; for deterministic order:

let (a, b, c) = tokio::join!(
    biased;
    future_a(),
    future_b(),
    future_c()
);

Use join! when you need results from multiple independent operations and all must complete.

The try_join! Macro

Similar to join!, but returns early on the first Err:

use tokio::try_join;

async fn fetch_user() -> Result<String, Box<dyn std::error::Error>> {
    Ok("user".to_string())
}

async fn fetch_settings() -> Result<String, Box<dyn std::error::Error>> {
    Ok("settings".to_string())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (user, settings) = try_join!(
        fetch_user(),
        fetch_settings()
    )?;
    
    println!("User: {}, Settings: {}", user, settings);
    Ok(())
}

All futures in try_join! must return Result with the same error type.

When to Use try_join!

// Good: All operations must succeed
let (user, profile, settings) = try_join!(
    db.fetch_user(),
    db.fetch_profile(),
    db.fetch_settings()
)?;

// Bad: Different error handling needed
// Use join! and handle errors individually instead

The select! Macro

Waits on multiple concurrent branches, returning when the first branch completes:

use tokio::select;
use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<i32>(100);
    
    loop {
        select! {
            Some(value) = rx.recv() => {
                println!("Got: {}", value);
            }
            _ = tokio::signal::ctrl_c() => {
                println!("Shutting down");
                break;
            }
        }
    }
}

select! cancels the remaining branches when the first one completes. Ensure operations are cancellation-safe.

Pattern Matching

select! {
    // Pattern match on result
    Ok(n) = reader.read(&mut buf) => {
        println!("Read {} bytes", n);
    }
    // Bind to variable
    msg = rx.recv() => {
        if let Some(m) = msg {
            process(m);
        }
    }
    // Multiple patterns
    result = operation() => match result {
        Ok(val) => handle_success(val),
        Err(e) => handle_error(e),
    }
}

Conditional Branches

let mut do_shutdown = false;

loop {
    select! {
        msg = rx.recv(), if !do_shutdown => {
            // Only poll rx when not shutting down
            process(msg);
        }
        _ = shutdown_rx.recv() => {
            do_shutdown = true;
        }
    }
}

Biased Polling

select! {
    biased;
    
    // Shutdown is checked first every time
    _ = shutdown_signal.recv() => {
        return;
    }
    // Then check for messages
    msg = message_rx.recv() => {
        process(msg);
    }
}

Biased polling can cause starvation. Place high-priority operations first, but ensure fairness for all branches.

The else Branch

select! {
    msg = rx.recv() => {
        println!("Received: {:?}", msg);
    }
    else => {
        // All branches disabled, execute fallback
        println!("No messages available");
    }
}

Without an else branch, select! panics if all branches are disabled.

Cancellation Safety

Some operations are cancellation-safe, others are not:

Cancellation-safe operations:

• tokio::sync::mpsc::Receiver::recv
• tokio::sync::broadcast::Receiver::recv
• tokio::sync::watch::Receiver::changed
• tokio::net::TcpListener::accept
• tokio::io::AsyncReadExt::read (on TcpStream)

NOT cancellation-safe:

• tokio::io::AsyncWriteExt::write_all
• tokio::io::AsyncReadExt::read_exact

use tokio::io::{AsyncReadExt, AsyncWriteExt};

// ❌ Bad: write_all may partially write
select! {
    _ = socket.write_all(&buf) => { }
    _ = shutdown.recv() => { }
}

// ✅ Good: Complete the write, then check shutdown
let write_fut = socket.write_all(&buf);
tokio::pin!(write_fut);

loop {
    select! {
        result = &mut write_fut => {
            result?;
            break;
        }
        _ = shutdown.recv() => {
            // Wait for write to complete before shutdown
            write_fut.await?;
            break;
        }
    }
}

Comparison: join! vs try_join! vs select!

Macro	Waits for	Returns on	Use Case
join!	All futures	All complete	Independent operations, all results needed
try_join!	All futures	First Err or all Ok	Operations that must all succeed
select!	First future	First complete	Racing operations, timeouts, shutdown signals

Complete Example: Using All Three

use tokio::{join, try_join, select};
use tokio::time::{sleep, Duration, timeout};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // join! - Wait for all independent operations
    let (users, posts, comments) = join!(
        fetch_users(),
        fetch_posts(),
        fetch_comments()
    );
    
    // try_join! - All must succeed
    let (profile, settings) = try_join!(
        save_profile(&users),
        save_settings(&posts)
    )?;
    
    // select! - First to complete wins
    select! {
        result = process_data() => {
            println!("Processing completed: {:?}", result);
        }
        _ = sleep(Duration::from_secs(10)) => {
            println!("Processing timed out");
        }
    }
    
    Ok(())
}

async fn fetch_users() -> Vec<String> { vec![] }
async fn fetch_posts() -> Vec<String> { vec![] }
async fn fetch_comments() -> Vec<String> { vec![] }
async fn save_profile(users: &[String]) -> Result<(), Box<dyn std::error::Error>> { Ok(()) }
async fn save_settings(posts: &[String]) -> Result<(), Box<dyn std::error::Error>> { Ok(()) }
async fn process_data() -> Result<(), Box<dyn std::error::Error>> { Ok(()) }

Resources

API Documentation

Complete API reference on docs.rs

Tokio Runtime

Learn more about runtime configuration

GitHub Repository

View source code

Testing Guide

Best practices for testing async code