[Rust 실전 로드맵 24] Tokio task와 async/await: 동시 실행을 안전하게 구성하기

tokio::spawn으로 future를 여러 개 띄우는 일은 어렵지 않습니다. 더 중요한 문제는 따로 있습니다. 각 task가 소유할 데이터를 어떻게 넘길지, 생성한 JoinHandle을 빠짐없이 기다릴지, 실패와 panic을 어디서 분류할지 정해야 합니다. 동시에 실행할 작업 수에도 상한이 필요합니다.

이 글은 Rust 2024 edition, rustc·Cargo 1.98.1, Tokio 1.53.1을 기준으로 합니다. 작은 검사 runner를 만들어 task 시작과 join, 결과 순서, Semaphore 기반 동시성 제한을 살펴봅니다. 네트워크와 시간 측정은 쓰지 않으며 select!, 취소, shutdown은 다음 편에서 다룹니다.

1. Tokio task는 독립적으로 스케줄되는 future입니다

Tokio의 task는 가볍고 비차단인 실행 단위입니다. tokio::spawn에 future를 넘기면 runtime이 다른 task와 함께 실행하도록 스케줄하고 JoinHandle을 돌려줍니다. spawn 직후 호출한 쪽과 새 task의 실행이 겹칠 수 있지만 어떤 task가 먼저 끝날지는 계약에 들어 있지 않습니다.

1.1. `spawn`이 요구하는 `Send + 'static`

tokio::spawn의 핵심 경계는 다음과 같습니다.

pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
where
    F: Future + Send + 'static,
    F::Output: Send + 'static,

'static은 task가 프로그램이 끝날 때까지 반드시 살아 있다는 뜻이 아닙니다. spawn된 future가 호출 함수의 짧은 지역 참조를 빌리지 않아도 된다는 타입 경계입니다. runtime은 task를 다른 thread로 옮길 수 있으므로 future와 출력에는 Send도 필요합니다.

예제는 각 검사 이름과 예정된 결과를 소유한 CheckPlan으로 만듭니다. loop 안에서는 async move가 plan과 복제한 공유 handle을 가져갑니다.

for (index, plan) in plans.into_iter().enumerate() {
    let barrier = Arc::clone(&barrier);
    let semaphore = Arc::clone(&semaphore);
    let started_tx = started_tx.clone();
    let active = Arc::clone(&active);
    let peak = Arc::clone(&peak);
    let join_name = plan.name.clone();
    let handle = tokio::spawn(async move {
        started_tx.send(index).await.expect("coordinator is alive");
        barrier.wait().await;
        let _permit = semaphore.acquire_owned().await.expect("semaphore is open");
        // ...
    });
    handles.push((index, join_name, handle));
}

여기서 task가 빌린 &str 대신 소유한 String을 받는 선택은 단순한 문법 우회가 아닙니다. 호출자의 stack frame과 task 수명을 분리하는 API 설계입니다.

1.2. spawn은 시작 순서나 완료 순서를 보장하지 않습니다

예제는 task마다 bounded mpsc channel로 시작 사실을 알린 뒤 Barrier에서 기다리게 합니다. coordinator가 모든 시작 메시지를 받은 후에야 barrier를 해제합니다. 이 방식이면 경과 시간을 재거나 scheduler 순서를 추측하지 않고도 여러 task가 release 전에 rendezvous에 도달했음을 확인할 수 있습니다.

Barrier는 참여 task가 모두 rendezvous할 때까지 완료되지 않습니다. 다만 이 검증이 특정 시작 순서나 완료 순서를 보장하는 것은 아닙니다. channel의 목적도 작업 결과 운반이 아니라 결정적인 test handshake입니다.

2. JoinHandle은 task 종료를 기다리는 소유권입니다

tokio::spawn이 반환한 JoinHandle<T>를 await하면 task 종료와 출력 T를 기다립니다. 반환형은 중첩될 수 있습니다. task 자체의 join 결과가 바깥 Result이고 task가 수행한 업무의 성공·실패는 안쪽 값으로 따로 표현할 수 있습니다.

이 예제는 업무 실패를 CheckStatus::CheckFailed로, task panic을 CheckStatus::Panicked로 구분합니다. 취소는 일부러 만들지 않습니다.

2.1. 하나가 실패해도 모든 handle을 join하기

동시 작업을 구조화하려면 첫 오류에서 곧바로 return하는 습관을 경계해야 합니다. 이미 spawn한 task가 남아 있다면 cleanup 경계가 흐려지기 때문입니다. runner는 handle loop를 끝까지 돌며 모든 task를 await합니다.

for (index, name, handle) in handles {
    joined_count += 1;
    match handle.await {
        Ok((returned_index, thread_id, result)) => {
            worker_threads.insert(thread_id);
            indexed_results.push((returned_index, result));
        }
        Err(error) if error.is_panic() => {
            indexed_results.push((
                index,
                CheckResult {
                    name,
                    status: CheckStatus::Panicked,
                },
            ));
        }
        Err(error) => panic!("unexpected cancelled task: {error}"),
    }
}

Tokio는 spawn된 task의 panic을 잡아 JoinError로 전달합니다. is_panic()으로 panic을 분류할 수 있고 is_cancelled()는 취소 여부를 판별합니다. 여기서는 어떤 task도 abort하지 않으므로 panic 외 JoinError를 예제 범위 밖 상태로 취급합니다. 실제 서비스라면 취소 정책과 shutdown 규칙을 먼저 정한 뒤 그 분기를 명시적으로 처리해야 합니다.

2.2. 결과 순서와 완료 순서는 다릅니다

handle을 입력 순서로 보관했다고 해서 task가 그 순서로 끝났다는 뜻은 아닙니다. runner는 각 join 결과를 입력 index와 함께 저장합니다. 모든 handle을 기다린 다음 index로 정렬해 report를 만듭니다.

indexed_results.sort_by_key(|(index, _)| *index);

Ok(RunReport {
    started_before_release,
    worker_threads,
    results: indexed_results
        .into_iter()
        .map(|(_, result)| result)
        .collect(),
    joined_count,
    peak_active: peak.load(Ordering::SeqCst),
})

따라서 report의 결정적인 입력 순서는 scheduler의 완료 순서를 관찰한 결과가 아니라 API가 의도적으로 복원한 순서입니다. 운영 코드에서도 둘을 섞어 설명하면 안 됩니다.

3. Semaphore로 실행 중인 작업 수를 제한합니다

모든 입력을 task로 만들 수 있다고 해서 모두를 한꺼번에 외부 자원에 진입시켜도 된다는 뜻은 아닙니다. Semaphore는 비동기 permit 획득을 제공하며 mutex와 달리 설정한 수만큼 여러 호출자를 동시에 통과시킬 수 있습니다.

예제의 각 task는 barrier를 지난 뒤 owned permit을 얻습니다. permit을 가진 구간에 들어올 때 active counter를 올리고 guard가 drop될 때 내립니다.

let _permit = semaphore.acquire_owned().await.expect("semaphore is open");
let now_active = active.fetch_add(1, Ordering::SeqCst) + 1;
peak.fetch_max(now_active, Ordering::SeqCst);
let _active_guard = ActiveGuard(active);
tokio::task::yield_now().await;

limit가 2이면 측정된 peak_active도 2를 넘지 않습니다. yield_now().await는 다른 task가 실행될 기회를 만들지만 다음에 어떤 task가 선택될지, 공정하게 번갈아 실행될지는 보장하지 않습니다. 이 test는 순서가 아니라 permit 상한만 검사합니다.

3.1. 잘못된 limit는 task를 만들기 전에 거부하기

Semaphore::new(0) 자체를 작업 대기 정책으로 해석하면 모든 task가 permit을 영원히 기다릴 수 있습니다. runner는 빈 입력인지와 상관없이 limit == 0을 먼저 검사해 typed error를 반환합니다.

pub async fn run_checks(
    plans: Vec<CheckPlan>,
    limit: usize,
) -> Result<RunReport, RunChecksError> {
    if limit == 0 {
        return Err(RunChecksError::ZeroLimit);
    }

    let task_count = plans.len();
    // ...
}

검증 순서도 중요합니다. channel, semaphore, task를 만든 뒤 오류를 알아내는 대신 configuration boundary에서 즉시 멈춥니다.

4. 동시성과 병렬성은 같은 말이 아닙니다

동시성(concurrency)은 여러 작업이 같은 기간에 진행될 수 있도록 구성하는 방식입니다. 병렬성(parallelism)은 실제로 여러 계산이 같은 순간에 서로 다른 실행 자원에서 돌아가는 상태입니다.

예제의 concurrent-start test는 current_thread runtime을 사용합니다. 한 OS thread에서도 task 세 개가 barrier에 도달해 진행 중 상태를 함께 가질 수 있으므로 동시성은 성립합니다. 그러나 OS thread가 하나뿐인 이 test로 병렬 실행을 주장할 수는 없습니다.

Tokio의 multi-thread runtime은 task를 여러 worker thread에서 실행할 수 있게 합니다. 가능성이지 개별 task의 thread 배치나 시작·완료 순서 보장은 아닙니다. task는 현재 thread에서 실행될 수도, 다른 thread로 보내질 수도 있습니다.

4.1. `.await`를 넣는다고 자동으로 병렬화되지는 않습니다

.await는 현재 future가 진행할 수 없을 때 runtime에 제어를 돌려줄 수 있는 지점입니다. CPU 작업을 여러 core에 자동 분산하는 지시문은 아닙니다. 긴 동기 CPU 작업이나 blocking 호출이 async task 안에서 양보하지 않으면 runtime thread를 계속 점유할 수 있습니다.

이번 runner에는 network I/O, blocking 작업, spawn_blocking이 없습니다. 그래서 I/O 처리량이나 CPU 병렬 성능을 이 결과에서 추론해서는 안 됩니다. 목적은 task lifecycle과 join 경계를 확인하는 데 있습니다.

5. 결정적인 test로 task 경계를 확인합니다

프로젝트 디렉터리에서 다음 명령을 실행합니다.

cd examples/article-24-tokio-tasks-async-await
cargo fmt --all -- --check
cargo check --all-targets --all-features
cargo clippy --all-targets --all-features -- -D warnings
cargo test --all-features
cargo run --quiet

Rust 1.98.1, Cargo 1.98.1, Tokio 1.53.1에서 다섯 명령은 종료 코드 0을 반환해야 합니다. 테스트 모음에는 통합 test 7개가 있습니다. 실행 출력은 입력 순서로 고정됩니다.

joined=3
alpha: passed
beta: check failed: invalid input
gamma: passed

5.1. test가 고정하는 계약

concurrent-start test는 current-thread runtime에서 task 세 개가 coordinator의 release 전에 모두 barrier에 도착했는지 검사합니다. join test는 성공, 업무 실패, panic이 섞여도 handle 세 개를 모두 기다리는지 확인합니다. input-order test는 report가 입력 순서를 복원하는지, semaphore test는 peak admission이 limit를 넘지 않는지 검증합니다. 나머지 두 test는 빈 plan과 비어 있지 않은 plan 모두에서 zero limit가 RunChecksError::ZeroLimit로 거부되는지 다룹니다. binary output test까지 합쳐 7개입니다.

이 결과가 말해 주지 않는 범위도 분명합니다. 실제 scheduler의 공정성, task 완료 순서, multi-thread 병렬 속도, 취소 후 cleanup, network backpressure는 측정하지 않았습니다. 특히 mpsc는 시작 rendezvous를 결정적으로 관찰하려고 썼을 뿐, long-lived worker 설계나 backpressure 정책의 예제가 아닙니다.

Tokio task를 안전하게 구성할 때는 spawn 개수보다 수명 경계를 먼저 보아야 합니다. task에는 소유한 데이터를 넘기고 만든 JoinHandle은 정해진 cleanup 지점에서 모두 기다리며 외부 자원 진입에는 명시적인 상한을 둡니다. 그런 다음 report 순서와 실제 완료 순서를 구분해야 합니다. 취소와 shutdown이 필요한 순간부터는 별도 정책이 필요하며 그 경계는 다음 편에서 이어집니다.

전체 소스 코드

이 글의 전체 실행 가능한 소스는 GitHub의 Chapter 24 프로젝트에서 확인할 수 있습니다.