Rust에서 Mutex와 channel로 안전한 동시성 프로그래밍

1. 개요

Rust는 자신의 능력을 “두려움 없는 동시성”이라고 부릅니다. 컴파일러는 프로그램이 실행되기 전에 빌드 타임에 데이터 경합을 차단합니다. 이 글에서는 표준 라이브러리의 두 가지 주요 도구를 소개합니다: 상태 공유를 위한 Mutex (함께 Arc), 그리고 스레드 간 메시지 전송을 위한 mpsc channel입니다. 이 글을 다 읽으면 각각의 사용법, 테스트 작성, 그리고 데드락이나 과도한 lock 유지 같은 흔한 함정을 피하는 방법을 알게 될 것입니다.

모든 예제는 표준 라이브러리만 사용하므로 외부 의존성이 필요 없습니다. cargo new orders-demo로 프로젝트를 생성하고 코드를 src/main.rs에 붙여넣으면 됩니다.

2. 문제: 스레드 간 데이터 공유

여러 스레드가 주문을 처리하면서 총 주문 수를 세고 싶다고 가정해봅시다. 순진한 방식은 다음과 같습니다:

use std::thread;

fn main() {
    let mut total_orders = 0;

    let handles: Vec<_> = (0..4)
        .map(|_| thread::spawn(|| total_orders += 1))
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

이 코드는 컴파일되지 않습니다:

error[E0373]: closure may outlive the current function, but it borrows `total_orders`, which is owned by the current function

컴파일러가 두 가지 이유를 제시합니다:

• thread::spawn은 클로저가 'static이어야 합니다. 자식 스레드가 main보다 오래 살 수 있으므로 지역 변수를 빌릴 수 없습니다.
• move를 추가하더라도 각 스레드는 total_orders의 자체 복사본을 받습니다. 실제로는 하나의 공유 변수에 누적되지 않습니다.

이를 해결하려면 두 가지가 필요합니다: Arc로 ownership 공유, Mutex로 접근 동기화.

3. Arc<Mutex<T>>로 상태 공유

3.1. 여러 스레드에서 주문 세기

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let total_orders = Arc::new(Mutex::new(0u32));
    let mut handles = Vec::new();

    for _ in 0..4 {
        let counter = Arc::clone(&total_orders);
        handles.push(thread::spawn(move || {
            for _ in 0..1_000 {
                *counter.lock().unwrap() += 1;
            }
        }));
    }

    for h in handles {
        h.join().unwrap();
    }

    println!("총 주문: {}", *total_orders.lock().unwrap());
}

총 주문: 4000

각 부분의 설명:

• Arc::new(Mutex::new(0u32)): Arc (Atomic Reference Counted)는 여러 스레드가 값을 공동 소유하도록 합니다. Mutex는 한 번에 하나의 스레드만 내부 값에 접근하도록 보장합니다.
• Arc::clone(&total_orders)은 참조 카운트를 증가시키기만 하고 데이터를 복사하지 않습니다. 각 스레드는 move를 통해 자체 Arc를 받습니다.
• counter.lock()은 lock을 획득할 때까지 현재 스레드를 차단합니다. 함수는 Result<MutexGuard<u32>, PoisonError<…>>를 반환합니다.
• MutexGuard는 &mut u32로 deref되므로 *... += 1로 값을 직접 수정합니다. Guard는 문장 끝에서 drop되고 lock은 자동으로 해제됩니다.

Rust에서는 “unlock을 잊을” 수 없습니다. 데이터가 Mutex 내부에 있으므로 lock을 건너뛰고 값을 읽을 수도 없습니다.

주의: Arc를 Rc로 바꿀 수 없습니다. Rc는 Send를 구현하지 않으므로 thread::spawn에 전달할 때 컴파일 오류가 발생합니다. 마찬가지로 Mutex를 RefCell로 바꿀 수 없습니다. RefCell은 Sync가 아니기 때문입니다.

3.2. MutexGuard의 범위

Lock은 guard가 drop될 때까지 유지됩니다. guard를 변수에 할당하면 lock은 변수의 scope이 끝날 때까지 유지됩니다:

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

fn reserve(inventory: &Mutex<HashMap<String, u32>>, sku: &str) -> bool {
    let reserved = {
        let mut stock = inventory.lock().unwrap();
        match stock.get_mut(sku) {
            Some(qty) if *qty > 0 => {
                *qty -= 1;
                true
            }
            _ => false,
        }
    }; // guard는 여기서 drop되고 lock이 해제됩니다

    if reserved {
        // PaymentService 호출 시뮬레이션, lock을 유지하지 않음
        thread::sleep(Duration::from_millis(50));
    }
    reserved
}

fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("AO-THUN".to_string(), 2)])));

    let handles: Vec<_> = (0..3)
        .map(|i| {
            let inv = Arc::clone(&inventory);
            thread::spawn(move || println!("고객 {i}: {}", reserve(&inv, "AO-THUN")))
        })
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

고객 0: true
고객 1: true
고객 2: false

줄의 순서는 실행할 때마다 다를 수 있습니다. 결과는 항상 true 둘과 false 하나입니다.

블록 { ... }은 lock 유지 시간을 최소한으로 제한합니다. 느린 부분(결제 호출)은 lock이 해제된 후 실행되므로 다른 스레드가 기다릴 필요가 없습니다. 중간에 더 일찍 해제해야 하는 경우 명시적으로 drop(stock)을 호출할 수 있습니다.

3.3. Mutex poisoning

스레드가 lock을 유지하는 동안 panic이 발생하면 Rust는 Mutex를 poisoned로 표시합니다. 이유는 데이터가 불완전한 상태에 있을 수 있기 때문입니다:

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let balance = Arc::new(Mutex::new(100i64));

    let b = Arc::clone(&balance);
    let _ = thread::spawn(move || {
        let mut guard = b.lock().unwrap();
        *guard -= 30;
        panic!("PaymentService 중간 오류");
    })
    .join();

    match balance.lock() {
        Ok(v) => println!("잔액: {}", *v),
        Err(poisoned) => {
            let v = poisoned.into_inner();
            println!("Mutex가 poisoned되었습니다. 현재 잔액: {}", *v);
        }
    }
}

thread '<unnamed>' panicked at src/main.rs:10:9:
PaymentService 중간 오류
Mutex가 poisoned되었습니다. 현재 잔액: 70

따라서 lock() 후 .unwrap()은 실제로 “다른 스레드가 lock을 유지하는 동안 panic이 발생했으면 panic을 따라한다”는 뜻입니다. 대부분의 애플리케이션에서는 합리적인 선택입니다. 데이터를 복구하는 방법을 알고 있다면 into_inner()로 guard를 반환받아 확인하고 수정할 수 있습니다.

4. Channel로 메시지 전송

Mutex는 “여러 스레드가 하나의 데이터를 수정한다”는 문제를 해결합니다. Channel은 반대 방향으로 갑니다: 각 데이터는 한 번에 하나의 소유자만 가지며, ownership이 channel을 통해 전달됩니다. 많은 문제에서는 이 방식이 더 간단하고 오류가 적습니다.

4.1. 여러 producer, 하나의 consumer

std::sync::mpsc는 multiple producer, single consumer의 약자입니다:

use std::sync::mpsc;
use std::thread;

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

fn main() {
    let (tx, rx) = mpsc::channel::<Order>();

    for shop in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let order = Order {
                id: shop * 100,
                customer: format!("고객-{shop}"),
                amount: 150_000 * shop as u64,
            };
            tx.send(order).unwrap();
        });
    }
    drop(tx); // 중요: 원본 sender 해제

    let mut revenue = 0;
    for order in rx {
        println!("주문 #{} 수신, 고객: {}", order.id, order.customer);
        revenue += order.amount;
    }
    println!("수익: {revenue}");
}

주문 #100 수신, 고객: 고객-1
주문 #300 수신, 고객: 고객-3
주문 #200 수신, 고객: 고객-2
수익: 900000

설명:

• mpsc::channel()은 (Sender, Receiver) 쌍을 반환합니다. 각 producer는 tx.clone()의 복사본을 받습니다.
• tx.send(order)은 order를 channel로 move합니다. 이 줄 후에는 송신 스레드가 더 이상 order를 사용할 수 없으며, 시도하면 컴파일 오류가 발생합니다. 따라서 두 스레드가 같은 Order를 수정하지 않습니다.
• for order in rx는 모든 Sender가 drop될 때까지 반복됩니다. drop(tx) 줄을 빼먹으면 원본 sender가 main에서 계속 살아있고 루프는 계속 기다립니다.

수신 순서는 스레드 스케줄에 따라 다릅니다. 총 수익은 항상 900000입니다.

4.2. 용량 제한 channel과 backpressure

mpsc::channel()은 무제한 버퍼입니다. producer가 consumer보다 빠르면 메모리가 계속 증가합니다. sync_channel(n)은 버퍼를 n개 요소로 제한하고, 버퍼가 가득 차면 send가 블록됩니다:

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

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u32>(2);

    let producer = thread::spawn(move || {
        for id in 1..=5 {
            tx.send(id).unwrap();
            println!("주문 {id} 전송됨");
        }
    });

    thread::sleep(Duration::from_millis(100));
    for id in rx {
        println!("주문 {id} 처리 중");
        thread::sleep(Duration::from_millis(20));
    }
    producer.join().unwrap();
}

주문 1 전송됨
주문 2 전송됨
주문 1 처리 중
주문 3 전송됨
주문 2 처리 중
...

Producer는 두 개의 주문을 전송한 후 consumer가 일부를 처리할 때까지 기다려야 합니다. 이것이 자연스러운 backpressure 메커니즘으로 데이터 처리 파이프라인에 매우 유용합니다. sync_channel(0)으로는 각 send가 누군가 recv할 때까지 기다려야 합니다 (rendezvous).

5. Arc가 불필요한 경우: scoped thread

스레드가 함수 반환 전에 반드시 종료되면 thread::scope은 Arc 없이 데이터를 직접 빌릴 수 있게 합니다:

use std::sync::Mutex;
use std::thread;

fn main() {
    let processed = Mutex::new(Vec::new());
    let orders = vec![101, 102, 103, 104];

    thread::scope(|s| {
        for chunk in orders.chunks(2) {
            s.spawn(|| processed.lock().unwrap().extend_from_slice(chunk));
        }
    }); // scope 내의 모든 스레드가 join되었습니다

    let mut result = processed.into_inner().unwrap();
    result.sort();
    println!("{result:?}");
}

[101, 102, 103, 104]

scope 덕분에 컴파일러는 스레드가 processed와 orders보다 오래 살지 않음을 알 수 있습니다. 따라서 & 빌림이 합법입니다. scope 후 into_inner()로 lock 없이 Mutex에서 데이터를 꺼냅니다.

6. 동시성 코드 테스트 작성

테스트하기 쉽도록 로직을 순수 함수로 분리해야 합니다. 다음 코드를 src/lib.rs에 넣으세요:

use std::sync::{mpsc, Arc, Mutex};
use std::thread;

pub fn count_orders(workers: u32, per_worker: u32) -> u32 {
    let total = Arc::new(Mutex::new(0));
    let handles: Vec<_> = (0..workers)
        .map(|_| {
            let total = Arc::clone(&total);
            thread::spawn(move || {
                for _ in 0..per_worker {
                    *total.lock().unwrap() += 1;
                }
            })
        })
        .collect();
    handles.into_iter().for_each(|h| h.join().unwrap());
    let result = *total.lock().unwrap();
    result
}

pub fn sum_payments(payments: Vec<u64>) -> u64 {
    let (tx, rx) = mpsc::channel();
    for amount in payments {
        let tx = tx.clone();
        thread::spawn(move || tx.send(amount).unwrap());
    }
    drop(tx);
    rx.iter().sum()
}

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

    #[test]
    fn counts_every_order_exactly_once() {
        assert_eq!(count_orders(8, 10_000), 80_000);
    }

    #[test]
    fn sums_all_payments_from_channel() {
        assert_eq!(sum_payments(vec![50_000, 120_000, 30_000]), 200_000);
    }

    #[test]
    fn empty_payments_returns_zero() {
        assert_eq!(sum_payments(vec![]), 0);
    }
}

cargo test

running 3 tests
test tests::empty_payments_returns_zero ... ok
test tests::sums_all_payments_from_channel ... ok
test tests::counts_every_order_exactly_once ... ok

test result: ok. 3 passed; 0 failed

테스트 empty_payments_returns_zero는 edge case를 확인합니다: drop(tx)을 빼먹으면 0을 반환하는 대신 무한 대기합니다. 이것이 오류를 감지하는 명확한 신호입니다.

count_orders에서 작은 포인트 하나: 반환하기 전에 값을 result 변수에 할당합니다. 이유는 함수 마지막 표현식의 임시 guard가 total보다 오래 살 수 있고 컴파일러가 borrow 오류를 보고하기 때문입니다.

7. 흔한 오류

순서가 잘못된 lock으로 인한 데드락. 스레드 A는 orders 다음 customers를 잠글 때, 스레드 B는 customers 다음 orders를 잠급니다. 둘 다 서로를 기다리며 무한 대기합니다. Rust는 데이터 경합을 방지하지만 데드락은 방지하지 않습니다. 고정된 lock 순서를 정하거나 두 데이터를 하나의 Mutex<(A, B)>로 결합해야 합니다.

임시 guard가 생각보다 오래 산다. 다음 코드를 보세요:

while let Some(order) = queue.lock().unwrap().pop() {
    process(order); // lock이 여기서도 유지됩니다!
}

while let (및 match)의 표현식 내 임시값은 루프 본문이 끝날 때까지 살아있습니다. 다른 스레드가 queue에 주문을 푸시하려 하면 process가 실행되는 동안 계속 차단됩니다. 수정 방법은 분리하는 것입니다: let next = queue.lock().unwrap().pop(); 다음 처리합니다.

같은 스레드에서 lock()을 다시 호출. std::sync::Mutex는 재진입 불가입니다. 같은 스레드에서 lock()을 두 번 호출하면 데드락이나 panic이 발생합니다.

I/O 또는 recv() 대기 중 lock 유지. guard를 유지하는 동안의 모든 blocking 작업은 Mutex를 병목으로 만듭니다. 필요한 데이터를 복사하거나 clone한 후 lock을 해제하고 느린 작업을 수행해야 합니다.

Sender drop 빼먹기. 위에서 본 것처럼 consumer는 무한 대기합니다.

주의: async 코드(Tokio)에서는 .await를 거쳐 std::sync::MutexGuard를 유지하지 마세요. tokio::sync::Mutex나 Tokio의 channel을 사용하세요. 또한 간단한 정수 카운터처럼 AtomicU32가 Mutex<u32>보다 훨씬 빠릅니다.

언제 어느 것을 선택할까?

• Mutex는 여러 스레드가 캐시나 재고 같은 하나의 상태를 오랫동안 읽기/수정할 때 사용하세요. 읽기가 명백히 더 많으면 RwLock을 고려하세요.
• Channel은 주문 큐, 파이프라인, worker pool처럼 데이터가 한 곳에서 다른 곳으로 흐를 때 사용하세요. Channel은 각 값이 하나의 소유자만 가지도록 하므로 로직을 더 쉽게 추론할 수 있습니다.

8. 결론

Rust는 일반적인 동시성 오류를 컴파일 오류로 변환합니다. Arc 없이 데이터를 공유할 수 없고, Mutex 없이 데이터를 수정할 수 없으며, channel을 통해 전송된 값을 다시 사용할 수 없습니다. Arc<Mutex<T>>는 공유 상태에 적합하고 mpsc channel은 스레드 간 ownership 전달에 적합합니다. sync_channel은 backpressure를 추가하고 thread::scope은 스레드 수명이 명확할 때 Arc를 제거하는 데 도움이 됩니다. 컴파일러가 보호하지 못하는 것은 데드락, 장시간 유지되는 guard, 그리고 drop되지 않은 sender입니다. 이런 부분은 신중하게 검토하고 세심하게 테스트를 작성해야 합니다.