Tokio로 하는 Async Rust: async/await, task 및 channel

1. 개요

Rust는 async/await 문법이 기본으로 포함되어 있지만 Future를 실행할 runtime이 없습니다. Tokio는 이를 위한 가장 인기 있는 runtime입니다. Scheduler, timer, I/O 비동기 처리, 그리고 동기화 도구들을 제공합니다.

이 글은 첫 번째 async 함수부터 시작하여 자주 사용되는 작업들을 다룹니다: task spawn, join!를 사용한 병렬 실행, select!를 사용한 타임아웃, task 간 상태 공유, channel을 통한 데이터 송신, 그리고 blocking 코드 처리. 마지막 부분에서는 테스트 작성 방법과 흔한 오류들을 다룹니다.

2. 설치

Cargo.toml에 Tokio를 추가합니다:

[dependencies]
tokio = { version = "1", features = ["full"] }

[dev-dependencies]
tokio = { version = "1", features = ["full", "test-util"] }

full feature는 모든 모듈을 활성화합니다: multi-threaded runtime, time, sync, net, macros 등. 이 방식은 학습할 때 편리합니다. 실제 프로젝트에서는 정말 필요한 feature만 활성화하여 컴파일 시간을 줄이세요. 예를 들어 ["rt-multi-thread", "macros", "time", "sync"]. test-util feature는 테스트에서 시간을 제어할 수 있게 합니다. 섹션 8에서 사용할 것입니다.

3. 첫 번째 async 함수

3.1. #[tokio::main] 및 .await

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

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100)).await; // 데이터베이스 호출 시뮬레이션
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let order = fetch_order(42).await;
    println!("{order}");
}

설명:

• async fn fetch_order는 String을 즉시 반환하지 않습니다. 대신 impl Future<Output = String>을 반환합니다.
• tokio::time::sleep은 async 버전의 sleep입니다. 대기하는 동안 thread는 runtime에 반환되어 다른 작업을 실행할 수 있습니다.
• #[tokio::main]은 매크로입니다. multi-threaded runtime을 생성한 후 block_on을 사용하여 그 안에서 main을 실행합니다.

Order #42

3.2. Future는 lazy입니다

이것이 JavaScript의 Promise와의 큰 차이점입니다: Rust의 future는 .await 또는 poll되기 전까지 아무것도 하지 않습니다.

async fn load_customer(id: u32) -> String {
    println!("고객 {id} 로딩 중");
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let fut = load_customer(7); // 함수 내부의 코드는 아직 실행되지 않음
    println!("await 전");
    let customer = fut.await;
    println!("{customer}");
}

await 전
고객 7 로딩 중
Customer 7

“고객 7 로딩 중”은 .await 호출 후에만 나타납니다. .await을 빠뜨리면 함수는 절대 실행되지 않습니다. Compiler는 unused implementer of Future that must be used 경고를 표시하니 이 경고를 무시하지 마세요.

4. Task spawn

.await은 future를 순차적으로 실행합니다. 동시에 실행하려면 tokio::spawn을 사용하여 task를 만듭니다. Task는 Tokio가 관리하는 가벼운 “green thread”입니다.

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

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100 * id as u64)).await;
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let mut handles = Vec::new();

    for id in 1..=3 {
        handles.push(tokio::spawn(async move { fetch_order(id).await }));
    }

    for handle in handles {
        let order = handle.await.expect("task가 panic됨");
        println!("{order}");
    }
    println!("전체 시간: ~{} ms", start.elapsed().as_millis());
}

Order #1
Order #2
Order #3
전체 시간: ~300 ms

순차 실행하면 전체 시간은 100 + 200 + 300 = 600 ms입니다. Spawn할 때 세 task가 동시에 실행되므로 전체 시간은 약 300 ms입니다.

몇 가지 주의할 점:

• tokio::spawn은 JoinHandle<T>을 반환합니다. handle을 .await할 때 Result<T, JoinError>를 받습니다. task가 panic되거나 취소되면 Err입니다.
• 전달되는 future는 Send + 'static을 만족해야 합니다. task가 다른 thread에서 실행될 수 있고 현재 scope보다 오래 살 수 있기 때문입니다. 따라서 async move가 필요합니다.
• Task는 spawn되는 순간 즉시 실행 시작합니다. handle을 .await할 때까지 기다리지 않습니다.

주의: JoinHandle을 drop하면 task를 취소하지 않습니다. Task는 detached 모드로 계속 실행됩니다. 취소하려면 handle.abort()를 호출하세요.

task 수가 변할 때는 tokio::task::JoinSet을 사용할 수 있습니다. 이를 통해 join_set.join_next().await를 사용하여 완료된 task 순서대로 결과를 받을 수 있습니다.

5. 같은 task 내에서 동시 실행: join!, select!, timeout

5.1. join! 및 try_join!

항상 spawn이 필요한 것은 아닙니다. tokio::join!은 여러 future를 같은 task 내에서 동시에 poll하고 모두 완료될 때까지 기다립니다:

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

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

async fn fetch_customer(id: u32) -> String {
    sleep(Duration::from_millis(150)).await;
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let (order, customer) = tokio::join!(fetch_order(1), fetch_customer(7));
    println!("{order} - {customer} ({} ms)", start.elapsed().as_millis());
}

Order #1 - Customer 7 (200 ms)

전체 시간은 더 오래 걸리는 future와 같습니다. 약 200 ms입니다. Future는 'static일 필요가 없으므로 로컬 변수를 빌릴 수 있습니다.

함수가 Result를 반환할 때는 tokio::try_join!을 사용합니다. 이 매크로는 첫 오류에서 즉시 멈추고 남은 future를 drop합니다:

let (order, customer) = tokio::try_join!(fetch_order_checked(1), fetch_customer_checked(7))?;

여기서 fetch_order_checked와 fetch_customer_checked는 Result<String, AppError>를 반환하는 async 함수입니다.

5.2. timeout

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

match timeout(Duration::from_millis(50), fetch_order(1)).await {
    Ok(order) => println!("{order}"),
    Err(_) => println!("타임아웃"),
}

타임아웃

fetch_order는 200 ms가 걸리는데 50 ms 제한을 초과합니다. 이 경우 timeout은 Err(Elapsed)를 반환하고 내부 future는 drop되어 취소됩니다.

5.3. select!

select!는 첫 번째 분기가 완료될 때까지 기다리고 나머지를 취소합니다:

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    tokio::spawn(async move {
        sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(());
    });

    tokio::select! {
        order = fetch_order(1) => println!("처리 완료 {order}"),
        _ = shutdown_rx => println!("shutdown 신호 수신, 처리 중단"),
    }
}

shutdown 신호 수신, 처리 중단

Shutdown 신호가 100 ms 후에 도착하는데 이는 fetch_order(200 ms)보다 빠릅니다. 따라서 두 번째 분기가 이기게 됩니다.

주의: Rust에서 취소는 future를 가장 가까운 .await 지점에서 drop하는 것을 의미합니다. 취소된 분기가 불완전한 연산(예: socket에서 메시지의 절반을 읽음)을 하고 있었다면 데이터가 손실될 수 있습니다. loop 내에서 select!에 함수를 넣기 전에 각 함수의 문서에서 “cancel safety”를 확인하세요.

6. Task 간 데이터 공유

6.1. Arc<Mutex<T>>

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

#[tokio::main]
async fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("laptop", 10u32)])));
    let mut handles = Vec::new();

    for _ in 0..5 {
        let inventory = Arc::clone(&inventory);
        handles.push(tokio::spawn(async move {
            let mut stock = inventory.lock().unwrap();
            *stock.get_mut("laptop").unwrap() -= 1;
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    println!("남은 수량: {}", inventory.lock().unwrap()["laptop"]);
}

남은 수량: 5

여기서는 tokio::sync::Mutex가 아니라 std::sync::Mutex를 사용합니다. 이것이 권장되는 선택입니다. critical section이 짧고 내부에 .await이 없을 때입니다. mutex는 표준 버전이 더 빠르기 때문입니다.

tokio::sync::Mutex가 필요한 경우는 .await을 지나가면서 lock을 유지해야 할 때입니다. 예를 들어 I/O 대기 중 공유 연결을 잠그고 있을 때입니다.

6.2. Channel mpsc

메모리를 공유하는 대신 channel을 통해 task 간에 메시지를 전달할 수 있습니다:

use tokio::sync::mpsc;

#[derive(Debug)]
struct Order {
    id: u32,
    amount: u64,
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<Order>(32);

    for id in 1..=3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(Order { id, amount: id as u64 * 100 }).await.unwrap();
        });
    }
    drop(tx); // 루프가 끝날 수 있도록 원본 sender 닫기

    let mut total = 0;
    while let Some(order) = rx.recv().await {
        println!("수신 {order:?}");
        total += order.amount;
    }
    println!("합계: {total}");
}

수신 Order { id: 1, amount: 100 }
수신 Order { id: 3, amount: 300 }
수신 Order { id: 2, amount: 200 }
합계: 600

메시지 수신 순서는 실행마다 다를 수 있습니다. 32 매개변수는 버퍼 용량입니다. 버퍼가 가득 차면 send는 대기하여 자연스러운 back-pressure 메커니즘을 만듭니다. recv()는 모든 sender가 drop되면 None을 반환합니다. drop(tx)를 빠뜨리면 프로그램이 while 루프에서 영원히 멈춰있을 것입니다.

mpsc 외에도 Tokio는 다음을 제공합니다:

• oneshot: 단 하나의 값을 보냅니다.
• broadcast: 여러 consumer, 각 consumer가 모든 메시지를 받습니다.
• watch: consumer가 최신 값만 관심 있습니다.

7. Blocking 코드 처리

Tokio는 cooperative scheduling을 사용합니다. Task는 .await 지점에서만 thread를 양보합니다. CPU 비용이 많이 드는 연산이나 동기 I/O는 worker thread를 독점하여 다른 task가 굶게 됩니다.

fn hash_password(password: &str) -> String {
    // CPU 비용이 많이 드는 연산을 시뮬레이션, 예: bcrypt/argon2
    std::thread::sleep(std::time::Duration::from_millis(500));
    format!("hashed:{password}")
}

#[tokio::main]
async fn main() {
    let password = String::from("s3cret");
    let hashed = tokio::task::spawn_blocking(move || hash_password(&password))
        .await
        .expect("blocking task가 panic됨");
    println!("{hashed}");
}

hashed:s3cret

spawn_blocking은 closure를 blocking 작업 전용 스레드 풀로 보냅니다. 이렇게 하면 runtime의 worker thread는 다른 요청을 처리할 수 있습니다.

주의: async 함수 내에서 직접 std::thread::sleep, std::fs::read, 또는 동기 HTTP client를 호출하지 마세요. async 버전(tokio::time::sleep, tokio::fs)을 사용하거나 spawn_blocking으로 감싸세요. 대규모 병렬 계산의 경우 Rayon이 더 적합한 경우가 많습니다.

8. Async 코드 테스트

매크로 #[tokio::test]는 각 테스트에 대해 runtime을 생성합니다. 기본적으로 이는 단일 스레드 runtime입니다.

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

pub async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn fetch_order_returns_formatted_id() {
        assert_eq!(fetch_order(42).await, "Order #42");
    }

    #[tokio::test(start_paused = true)]
    async fn timeout_fires_when_order_service_is_slow() {
        let result = timeout(Duration::from_millis(50), fetch_order(1)).await;
        assert!(result.is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn order_arrives_within_generous_timeout() {
        let result = timeout(Duration::from_secs(1), fetch_order(1)).await;
        assert_eq!(result.unwrap(), "Order #1");
    }
}

cargo test

running 3 tests
test tests::fetch_order_returns_formatted_id ... ok
test tests::timeout_fires_when_order_service_is_slow ... ok
test tests::order_arrives_within_generous_timeout ... ok

start_paused = true (test-util feature 필요)는 Tokio의 시계를 일시 중지합니다. 모든 task가 timer를 기다리고 있으면 runtime이 자동으로 다음 시점으로 시간을 건너뜁니다. 덕분에 timeout 관련 테스트는 거의 즉시 실행되고 결과가 일정하므로 머신 성능에 의존하지 않습니다.

여러 스레드에서의 동작을 확인해야 하면 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]를 사용합니다.

9. 흔한 오류

.await을 지나면서 std::sync::MutexGuard 유지. 이 guard는 Send가 아니므로 future도 Send가 아닙니다. 그러면 tokio::spawn은 future cannot be sent between threads safely 컴파일 오류를 냅니다. 수정 방법은 .await 전에 { ... } 블록으로 guard의 scope를 좁히거나 tokio::sync::Mutex로 전환하는 것입니다.

중첩된 runtime 생성. Tokio에서 실행 중인 코드 내에서 Runtime::new().block_on(...)을 호출하면 Cannot start a runtime from within a runtime panic이 발생합니다. async 문맥에서는 그냥 .await을 사용하세요.

필요 없을 때 async 사용. CLI 도구가 몇 개의 파일을 읽거나 순수 계산 작업을 하면 Tokio의 이점이 없습니다. 대신 컴파일 시간과 복잡성만 증가시킵니다. Async는 많은 I/O 연산이 동시에 일어날 때 가치 있습니다: 웹 서버, crawler, proxy, 여러 서비스를 호출하는 client.

제한 없는 spawn. Task는 저렴하지만 무료가 아닙니다. 백만 개 항목 리스트의 각 항목에 대해 task를 spawn하면 downstream의 리소스(예: 데이터베이스 connection pool)가 소진될 수 있습니다. tokio::sync::Semaphore나 futures crate의 buffer_unordered로 동시성을 제한하세요.

JoinError 무시. handle을 .await할 때 let _ =을 사용하면 task 내의 panic이 삼켜집니다. 오류를 로그하거나 위로 전파하세요.

10. 결론

이 글은 Tokio를 사용한 async Rust의 핵심을 다루었습니다:

• Future는 lazy이고 .await될 때만 실행됩니다.
• tokio::spawn은 동시 실행 task를 만듭니다. 전달되는 future는 Send + 'static이어야 합니다.
• join!과 try_join!은 같은 task 내에서 여러 future를 실행합니다. select!와 timeout은 경쟁 조건과 취소를 처리합니다.
• 공유 데이터는 Arc<Mutex<T>>로 공유하거나 mpsc와 oneshot 같은 channel을 통해 전달할 수 있습니다.
• Blocking 코드는 spawn_blocking에 넣어야 합니다.
• #[tokio::test(start_paused = true)]는 시간 관련 테스트를 빠르고 일관성 있게 만듭니다.

이 부분들을 숙달한 후에는 tokio::net을 통한 네트워크 I/O나 Axum 같은 framework를 사용한 web service 구축으로 나아갈 수 있습니다.