[Rust 실전 로드맵 22] Rust 채널과 공유 상태: 동시성 구조 선택하기

작업마다 스레드를 하나씩 만들면 작업 수가 곧 스레드 수가 됩니다. 이번에는 고정된 작업자 두 개가 제한된 큐에서 엔드포인트 네 개를 꺼내도록 구조를 바꿉니다. 작업 제출, 결과 회수, 종료를 각각 어떤 소유권 경계로 표현하는지 살펴봅니다.

이 글은 Rust 2024 edition과 rustc·Cargo 1.98.1을 기준으로 합니다. 표준 라이브러리의 동기 채널과 OS 스레드만 사용합니다. 비동기 런타임은 다음 편의 범위로 남겨 둡니다.

1. 제한된 채널로 작업자 풀 만들기

mpsc::sync_channel은 용량이 고정된 multi-producer, single-consumer 채널을 만듭니다. 용량보다 메시지가 적을 때는 send가 바로 큐에 넣을 수 있지만 버퍼가 가득 차면 소비자가 자리를 만들 때까지 호출 스레드를 막습니다. 용량이 0이면 송신자와 수신자가 직접 만나는 rendezvous 채널이 됩니다. 이 blocking 지점이 생산 속도를 제한하는 backpressure 경계입니다.

예제는 작업자 수와 큐 용량을 따로 받습니다.

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct PoolConfig {
    pub workers: usize,
    pub queue_capacity: usize,
}

1.1. 단일 수신자를 여러 작업자가 나누어 쓰기

표준 mpsc의 Receiver는 하나입니다. 예제는 Arc<Mutex<Receiver<Job>>>로 수신 권한을 공유합니다. 한 작업자가 잠금을 잡고 recv에서 기다리는 동안 다른 작업자는 같은 수신기를 받을 수 없습니다. 메시지를 하나 받은 뒤에는 guard가 블록 끝에서 해제되므로 실제 check는 여러 작업자가 동시에 수행할 수 있습니다.

let (work_sender, work_receiver) = mpsc::sync_channel::<Job>(config.queue_capacity);
let work_receiver = Arc::new(Mutex::new(work_receiver));
let (result_sender, result_receiver) = mpsc::channel::<(usize, CheckOutcome)>();

for _ in 0..config.workers {
    let work_receiver = Arc::clone(&work_receiver);
    let result_sender = result_sender.clone();
    let checker = Arc::clone(&checker);
    handles.push(thread::spawn(move || {
        loop {
            let job = {
                let receiver = work_receiver
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                receiver.recv()
            };
            let Ok(job) = job else {
                break WorkerExit::WorkChannelDisconnected;
            };
            // Run the check and send the result after releasing the receiver lock.
        }
    }));
}

교육용으로는 작고 명시적인 구성입니다. 다만 Mutex로 감싼 단일 수신기는 작업 획득을 직렬화하며 공정성을 보장하지 않습니다. 큐 용량 역시 처리량을 자동으로 높이는 숫자가 아닙니다. 허용할 대기 작업 수와 생산자가 막혀도 되는 위치를 보고 정해야 합니다.

1.2. 어디서 막힐 수 있는가

이 예제에는 세 가지 blocking 경계가 있습니다. 생산자는 가득 찬 sync_channel의 send에서 기다릴 수 있습니다. 작업자는 빈 작업 채널의 recv에서 기다립니다. 조정 스레드는 결과 recv와 마지막 join에서 기다릴 수 있습니다. 결과 채널은 제한되지 않았기 때문에 작업자가 결과를 보내며 막히지는 않지만 결과가 매우 크거나 소비가 늦은 운영 환경이라면 별도 용량 정책이 필요합니다.

예제는 모든 작업을 제출한 다음 결과를 받습니다. 작업 채널만 제한되어 있고 결과 채널은 제한되지 않아 이 순서가 진행됩니다. 두 채널을 모두 제한한 채 생산자가 결과를 읽기 전에 제출만 계속하면 작업자는 가득 찬 결과 큐에서, 생산자는 가득 찬 작업 큐에서 서로 기다리는 설계를 만들 수 있습니다.

2. 소유권과 연결 해제로 종료하기

채널 종료는 별도의 종료 플래그 없이도 소유권으로 표현할 수 있습니다. 모든 Sender 또는 SyncSender가 drop되면 수신기는 버퍼에 남은 메시지를 모두 받은 뒤 연결 해제를 반환합니다. 반대로 Receiver가 drop되면 이후 전송은 오류로 돌아옵니다.

2.1. 마지막 송신자를 명시적으로 놓기

조정 스레드는 입력 순서와 함께 Endpoint의 소유권을 Job으로 옮깁니다. 제출을 마치면 반드시 원본 송신자를 놓습니다.

for (index, endpoint) in endpoints.into_iter().enumerate() {
    if work_sender.send(Job { index, endpoint }).is_err() {
        channel_error = Some(MonitorError::WorkChannelDisconnected);
        break;
    }
}
drop(work_sender);

각 작업자는 복제된 작업 송신자를 갖지 않습니다. 따라서 조정 스레드의 drop(work_sender) 뒤에는 살아 있는 작업 송신자가 없습니다. 큐를 비운 작업자는 recv의 연결 해제를 관찰해 루프를 끝냅니다. 종료 메시지를 작업자 수만큼 세어 넣는 방식과 달리 종료 신호 개수와 작업자 수를 맞출 필요가 없습니다.

2.2. 결과 송신자와 합류 순서

작업자를 만들 때마다 결과 송신자를 복제합니다. 조정 스레드에 남은 원본은 바로 drop합니다. 모든 작업자 복제본이 사라지면 결과 수신기도 연결 해제를 볼 수 있습니다. 성공 경로에서는 예상 작업 수만큼 결과를 받은 뒤 모든 핸들을 합류합니다.

연결 해제는 종료를 알리지만 스레드 정리 완료를 증명하지는 않습니다. JoinHandle::join까지 호출해야 작업자가 끝났고 관련 메모리 효과를 합류 뒤에서 관찰할 수 있습니다. 예제는 한 합류가 실패해도 나머지 핸들을 모두 합류한 뒤 WorkerPanicked를 반환합니다.

3. 완료 순서와 출력 순서 분리하기

작업자가 끝나는 순서는 스케줄러와 검사 시간에 따라 달라집니다. 결과 채널에서 먼저 받은 항목을 그대로 출력하면 실행마다 순서가 바뀔 수 있습니다. 그래서 제출할 때 입력 인덱스를 메시지에 넣고 결과에도 돌려받습니다.

let mut indexed = Vec::with_capacity(expected);
for _ in 0..expected {
    match result_receiver.recv() {
        Ok(result) => indexed.push(result),
        Err(_) => {
            channel_error = Some(MonitorError::ResultChannelDisconnected);
            break;
        }
    }
}

indexed.sort_by_key(|(index, _)| *index);
let outcomes = indexed.into_iter().map(|(_, outcome)| outcome).collect();

정렬은 완료 순서를 바꾸지 않습니다. 표현 순서만 입력 순서로 복원합니다. 테스트도 어느 작업자가 먼저 끝나는지 단정하지 않고 최종 outcomes의 식별자 순서와 내용을 확인합니다. 이 구분 덕분에 sleep이나 실행 시간 임곗값 없이 결과를 결정적으로 검증할 수 있습니다.

4. 오류와 패닉을 정책으로 다루기

검사 실패, 채널 실패, 패닉은 같은 사건이 아닙니다. 예제는 Checker::check가 반환한 CheckError를 해당 엔드포인트의 OutcomeKind::CheckError로 바꾸고 다음 작업을 계속합니다. 작업 또는 결과 채널의 연결 해제는 전체 실행을 끝내는 MonitorError입니다.

4.1. 잡을 패닉의 범위를 좁히기

검사 구현이 unwind panic을 일으키면 예제는 검사 호출만 catch_unwind로 감싸 Panicked 결과로 바꿉니다.

let kind = match catch_unwind(AssertUnwindSafe(|| checker.check(&job.endpoint))) {
    Ok(Ok(status)) if (200..400).contains(&status) => OutcomeKind::Healthy { status },
    Ok(Ok(status)) => OutcomeKind::Unhealthy { status },
    Ok(Err(error)) => OutcomeKind::CheckError {
        message: error.message,
    },
    Err(_) => OutcomeKind::Panicked,
};

catch_unwind가 패닉 결과를 잡더라도 기본 panic hook은 패닉 시점에 먼저 실행됩니다. 따라서 잡힌 패닉도 기본 설정에서는 stderr에 메시지를 쓸 수 있습니다. 조용한 출력을 원한다는 이유만으로 프로세스 전역 hook을 작업자마다 바꾸면 다른 스레드와 충돌할 수 있으므로, hook 정책은 애플리케이션 경계에서 따로 설계해야 합니다.

또한 이 코드는 panic payload를 API 계약으로 노출하지 않습니다. panic=abort 빌드에서는 unwind를 잡을 수 없습니다. 패닉 전에 외부 상태가 이미 바뀌었다면 결과 하나로 바꿨다고 복구가 끝나는 것도 아닙니다. 예상 가능한 도메인 실패는 계속 Result로 반환하는 편이 맞습니다.

5. 채널·Mutex·원자 타입 선택하기

메시지 전달이 항상 우월하다고 볼 수는 없습니다. 바뀌는 데이터의 소유자, 여러 필드 사이의 불변식, 관찰 시점이 무엇인지에 따라 도구가 달라집니다.

5.1. 작업의 소유권을 넘길 때는 채널

생산자가 작업을 만들고 작업자가 그 값을 한 번 소비한다면 채널이 잘 맞습니다. 큐 용량으로 대기 작업 수를 제한하고 송신자 소유권으로 종료를 표현할 수 있습니다. 반면 여러 작업자가 같은 최신 설정을 반복해서 읽거나 한 객체를 제자리에서 갱신해야 한다면 매번 메시지로 복제하거나 별도 소유자 루프를 두는 편이 오히려 복잡할 수 있습니다.

5.2. 여러 필드를 한 불변식으로 묶을 때는 `Arc<Mutex<_>>`

MutexMetrics는 completed, healthy, unhealthy, check_errors, panics를 하나의 Summary 안에서 갱신합니다. lock guard가 살아 있는 동안 관련 필드의 변경을 한 임계 구역으로 묶을 수 있습니다. 여러 값의 관계를 한 시점의 일관된 snapshot으로 읽어야 한다면 이런 구조가 원자 카운터 여러 개보다 설명하기 쉽습니다.

다만 잠금 범위에서 느린 검사나 채널의 blocking 연산까지 수행하면 경합과 deadlock 가능성이 커집니다. 이 예제도 수신기 잠금은 recv 한 번까지만 유지합니다. 결과 집계용 잠금은 카운터 변경에만 씁니다. poisoning은 다른 스레드가 panic한 흔적이지 자동 복구가 아닙니다. into_inner를 사용할 때는 보호하던 불변식을 다시 검사할 책임이 남습니다.

5.3. 독립 이벤트 계수에는 원자 타입

AtomicMetrics의 각 필드는 서로 독립적인 telemetry count입니다. fetch_add와 최종 load에 Ordering::Relaxed를 사용합니다. 이 Relaxed 연산은 각 카운터 자체의 원자성만 보장하며 다른 데이터를 동기화하지 않습니다. 결과 객체를 publish하거나 여러 카운터의 transactional snapshot을 만드는 수단도 아닙니다.

예제는 모든 기록 스레드를 join한 뒤 snapshot_after_join을 호출하므로 최종 테스트 값을 결정적으로 읽습니다. 이 합류가 별도의 완료 경계입니다. 실행 중간에 completed == healthy + unhealthy + errors + panics 같은 필드 간 관계를 항상 관찰해야 한다면 독립 원자 카운터는 맞지 않으며 하나의 잠금으로 묶거나 별도 집계 소유자에게 메시지를 보내는 편이 낫습니다.

6. 경계를 검증하고 구조 선택하기

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

cd examples/article-22-channels-shared-state
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에서 다섯 명령은 종료 코드 0을 반환해야 합니다. 테스트 모음에는 라이브러리 단위 테스트 2개와 통합 테스트 7개, 모두 9개가 있습니다. 실행 출력은 다음과 같이 고정됩니다.

endpoint monitor: workers=2 queue_capacity=2
home: healthy (200)
health: unhealthy (503)
metrics: check-error (fixture timeout)
admin: healthy (204)
summary: completed=4 healthy=2 unhealthy=1 errors=1 panics=0
shutdown: disconnected_workers=2 joined_workers=2

6.1. 선택 기준

작업을 한 소유자에서 다른 소유자로 넘기고 대기열에 상한이 필요하면 제한된 채널부터 검토합니다. 여러 필드를 함께 바꾸고 같은 시점의 일관된 값을 읽어야 하면 Arc<Mutex<_>>가 더 직접적입니다. 다른 상태를 publish할 필요가 없는 독립 카운터나 플래그에는 원자 타입이 맞을 수 있습니다.

어느 구조든 blocking 위치와 종료 책임부터 적어 두면 판단하기 쉽습니다. 누가 마지막 송신자를 drop하는지, 누가 결과를 끝까지 비우는지, 누가 모든 핸들을 join하는지가 드러나야 합니다. 작업자 수와 큐 용량은 부하 측정과 자원 한도를 근거로 조정해야 하며 이 작은 예제의 값 2를 운영 기본값으로 받아들여서는 안 됩니다.

전체 소스 코드

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