</>PatchNote
목록으로

2024-09-21

HSPACE Rust 특강 #9 — 동시성

RustHSPACE 2024ConcurrencyThreads

가장 분량이 많았던 회차. 그리고 지금까지 배운 것들이 전부 여기서 맞물린다는 게 느껴진 회차이기도 하다. Rust가 "두려움 없는 동시성"이라고 말할 때 그게 정확히 무슨 뜻인지 확인하는 시간이었다.

강의는 동시성을 쓰는 방식을 셋으로 나눈다.

  1. Fork-Join — 일을 쪼개 스레드에 나눠주고 결과를 모은다
  2. 채널 — 스레드끼리 값을 주고받는다. 공유하지 않는다
  3. 공유 가변 상태 — 락으로 보호되는 데이터를 함께 쓴다

Fork-Join 병렬 처리

가장 단순한 형태다. 큰 작업을 독립적인 조각으로 나눌 수 있을 때 쓴다.

use std::thread;

let handle = thread::spawn(move || {
    // 다른 스레드에서 실행
    process(chunk)
});

let result = handle.join().unwrap();   // 끝날 때까지 기다리고 결과를 받는다

spawn 이 move 클로저를 요구하는 이유는 클로저 회차에서 본 대로다. 새 스레드가 원래 스택 프레임보다 오래 살 수 있으므로 캡처한 값을 소유해야 한다. 정확히는 F: FnOnce() -> T + Send + 'static 이라는 바운드가 붙는다.

스레드 간 오류 처리

join() 은 Result 를 돌려준다. 스레드가 패닉하면 Err 가 온다.

match handle.join() {
    Ok(value) => println!("결과: {value:?}"),
    Err(_)    => println!("스레드가 패닉했다"),
}

한 스레드의 패닉이 프로세스 전체를 죽이지 않는다. 패닉한 스레드만 되감기를 거쳐 종료되고, 부모는 join() 에서 그 사실을 알게 된다. 다만 오류의 원인 정보는 Box<dyn Any> 형태라 다루기가 번거로워서, 실제로는 오류를 채널로 보내는 쪽이 편하다.

변경할 수 없는 데이터 공유하기

모든 스레드가 같은 읽기 전용 데이터를 봐야 하는 경우가 흔하다. Arc 를 쓴다.

use std::sync::Arc;

let data = Arc::new(build_huge_table());

for _ in 0..8 {
    let data = Arc::clone(&data);   // 참조 카운트만 증가
    thread::spawn(move || {
        lookup(&data, key);
    });
}

데이터를 복사하지 않고, 카운터만 원자적으로 올린다. 마지막 스레드가 끝날 때 해제된다. 읽기만 한다면 Mutex 가 전혀 필요 없다. 공유 레퍼런스가 곧 불변이라는 규칙 덕분이다.

Rayon 라이브러리

직접 스레드를 관리하는 건 번거롭다. Rayon은 이터레이터를 병렬로 바꿔준다.

use rayon::prelude::*;

let sum: i64 = numbers.par_iter().map(|&n| expensive(n)).sum();

iter() 를 par_iter() 로 바꾸는 것만으로 작업 훔치기(work-stealing) 스레드 풀 위에서 돌아간다. 놀라운 건 이게 안전하다는 걸 컴파일러가 보장한다는 점이다. 클로저가 Send 가 아닌 걸 캡처하면 컴파일이 안 된다.

채널

두 번째 접근. Go의 슬로건 "메모리를 공유해서 통신하지 말고, 통신해서 메모리를 공유하라"와 같은 발상이다.

use std::sync::mpsc::channel;

let (sender, receiver) = channel();

thread::spawn(move || {
    for line in read_lines() {
        sender.send(line).unwrap();
    }
    // sender가 여기서 드롭 → 채널이 닫힌다
});

for line in receiver {          // 채널이 닫힐 때까지 순회
    process(line);
}

mpsc 는 multi-producer, single-consumer다. Sender 는 복제할 수 있지만 Receiver 는 하나다.

값을 보내는 것은 이동이다. sender.send(value) 는 값의 소유권을 넘긴다. 그래서 보낸 뒤에 원본을 건드릴 수 없고, 두 스레드가 같은 값을 동시에 볼 일이 없다. 채널이 안전한 이유가 새로운 장치가 아니라 이미 있던 소유권 규칙이라는 게 인상적이었다.

채널 닫기

Sender 가 전부 드롭되면 채널이 닫히고, for 루프가 자연스럽게 끝난다. "이제 끝"이라는 종료 신호를 따로 보낼 필요가 없다. 반대로 Sender 를 실수로 붙잡고 있으면 소비자가 영원히 기다린다 — 이건 실제로 자주 하는 실수다.

파이프라인 실행하기

예제로 나온 fingertips(역색인 생성기)가 좋은 사례다. 문서를 읽고 → 색인을 만들고 → 임시 파일로 쓰고 → 병합하는 다섯 단계를 채널로 잇는다.

fn run_pipeline(documents: Vec<PathBuf>, output_dir: PathBuf) -> io::Result<()> {
    // 각 단계가 별도 스레드에서 돌고, 채널로 연결된다
    let (texts,   h1) = start_file_reader_thread(documents);
    let (pints,   h2) = start_file_indexing_thread(texts);
    let (gallons, h3) = start_in_memory_merge_thread(pints);
    let (files,   h4) = start_index_writer_thread(gallons, &output_dir);
    let result = merge_index_files(files, &output_dir);

    // 모든 스레드를 join하고 오류를 확인
    // ...
}

각 단계가 자기 스레드에서 돌면서 앞 단계의 출력을 소비하고 다음 단계로 넘긴다. CPU 코어가 여러 개면 단계들이 동시에 진행된다. 채널이 자연스러운 백프레셔로도 작동한다는 게 좋았다. 동기 채널을 쓰면 소비자가 느릴 때 생산자가 자동으로 기다린다.

같은 저장소에 단일 스레드 버전(run_single_threaded)도 함께 있어서 비교하기 좋았다. 로직은 같고 연결 방식만 다르다.

채널의 성능

채널이 마법은 아니다.

  • send/recv 는 잠금과 조건 변수를 쓰므로 공짜가 아니다
  • 아주 작은 값을 아주 자주 보내면 오버헤드가 커진다. 덩어리로 묶어 보내는 게 대개 낫다
  • 기본 채널은 무한 버퍼라, 소비자가 느리면 메모리가 계속 늘어난다. sync_channel(n) 으로 용량을 제한할 수 있다

Send와 Sync

여기가 Rust 동시성의 핵심이다. 두 마커 트레잇이 스레드 안전성을 타입 수준에서 정의한다.

  • Send — 이 타입의 값을 다른 스레드로 옮겨도 안전하다
  • Sync — 이 타입의 &T 를 여러 스레드가 동시에 봐도 안전하다

정확히는 T: Sync ⟺ &T: Send 다.

거의 모든 타입이 자동으로 둘 다 구현한다(auto trait). 안 되는 것들이 중요하다.

타입SendSync이유
Rc<T>✗✗참조 카운트가 원자적이지 않음
RefCell<T>✓✗빌림 플래그가 원자적이지 않음
Arc<T>✓✓원자적 카운터 (T: Send + Sync 일 때)
Mutex<T>✓✓락으로 보호됨 (T: Send 일 때)
원시 포인터✗✗무엇을 가리키는지 알 수 없음

thread::spawn 의 시그니처가 F: Send, T: Send 를 요구하므로, Rc 를 캡처한 클로저는 아예 컴파일되지 않는다. 첫 회차에서 본 그 예제다.

여기서 감탄한 부분: Rust 컴파일러는 동시성에 대해 아무것도 모른다. 데드락도 경합도 이해하지 못한다. 그저 Send/Sync 바운드와 소유권 규칙을 검사할 뿐이다. 그런데 그것만으로 데이터 경합이 사라진다. 동시성 안전이 일반적인 타입 검사의 따름정리로 나오는 구조다.

공유 가변 상태

세 번째 접근. 여러 스레드가 같은 데이터를 읽고 쓴다.

뮤텍스

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

let counter = Arc::new(Mutex::new(0));

for _ in 0..10 {
    let counter = Arc::clone(&counter);
    thread::spawn(move || {
        let mut n = counter.lock().unwrap();   // 락 획득
        *n += 1;
    });                                        // 스코프 끝 → 자동 해제
}

Rust의 Mutex<T> 가 C의 뮤텍스와 결정적으로 다른 점은 보호 대상을 뮤텍스가 소유한다는 것이다.

C에서는 뮤텍스와 데이터가 별개의 변수다. 락을 안 잡고 데이터에 접근해도 컴파일러는 모른다. Rust에서는 데이터가 Mutex 안에 들어 있고, 꺼내려면 lock() 을 불러야 한다. lock() 은 MutexGuard 를 주고, 이 가드가 드롭될 때 락이 풀린다.

락을 잊는 게 불가능하고, 해제를 잊는 것도 불가능하다.

mut와 Mutex

재미있는 지점이 하나 있다.

let counter = Mutex::new(0);       // mut가 없다!
let mut n = counter.lock().unwrap();
*n += 1;                           // 그런데 내용을 바꾼다

counter 는 mut 가 아닌데 안의 값을 바꾼다. Mutex 도 내부 가변성 타입이기 때문이다. RefCell 이 단일 스레드에서 런타임 빌림 검사로 이를 제공한다면, Mutex 는 멀티 스레드에서 락으로 제공한다.

교착 상태

Rust도 데드락은 막지 못한다.

// 스레드 A: lock(a) → lock(b)
// 스레드 B: lock(b) → lock(a)     ← 데드락

같은 뮤텍스를 두 번 잠그는 것도 데드락이다.

let guard1 = m.lock().unwrap();
let guard2 = m.lock().unwrap();   // 자기 자신을 기다린다

빌림 검사기는 메모리 안전을 보장하지 데드락 자유를 보장하지 않는다. 데드락은 정의상 미정의 동작이 아니라 "그냥 멈춘 프로그램"이라 안전한 동작으로 분류된다. 락 순서를 정하는 것 같은 고전적 규율은 여전히 개발자 몫이다.

오염된 뮤텍스

lock() 이 Result 를 돌려주는 이유가 이거다. 락을 잡은 스레드가 패닉하면 그 뮤텍스는 오염(poisoned) 된다. 보호 중이던 데이터가 절반만 갱신된 상태일 수 있기 때문이다.

이후 lock() 호출은 Err(PoisonError) 를 돌려준다. 그래서 코드에 .unwrap() 이 붙는 것이고, 그건 "다른 스레드가 이 데이터를 망가뜨린 채 죽었으면 나도 죽겠다"는 선언이다. 데이터 무결성을 언어가 진지하게 다룬다는 게 여기서도 보인다.

RwLock

읽기가 압도적으로 많으면 뮤텍스는 과하다.

use std::sync::RwLock;

let config = RwLock::new(Config::default());

let c = config.read().unwrap();       // 여러 스레드가 동시에 읽기 가능
// ...
let mut c = config.write().unwrap();  // 쓰기는 배타적

&T 여러 개 vs &mut T 하나라는 빌림 규칙이 런타임 락으로 그대로 옮겨온 형태다. 같은 규칙이 컴파일 시점(레퍼런스), 단일 스레드 런타임(RefCell), 멀티 스레드 런타임(RwLock) 세 곳에서 반복된다는 걸 보고 설계의 일관성이 느껴졌다.

조건 변수

"조건이 만족될 때까지 기다린다"를 표현한다.

use std::sync::Condvar;

let (lock, cvar) = &*pair;
let mut started = lock.lock().unwrap();

while !*started {
    started = cvar.wait(started).unwrap();   // 락을 풀고 대기, 깨면 다시 잡음
}

wait 이 가드를 소비하고 다시 돌려준다는 게 인상적이다. 대기 중에는 락을 놓아야 하고, 깨어나면 다시 잡아야 한다는 프로토콜이 타입 시그니처에 그대로 들어 있다. 그리고 가짜 깨어남(spurious wakeup) 때문에 if 가 아니라 while 로 검사해야 한다.

원자성

가장 단순한 카운터라면 뮤텍스도 과하다.

use std::sync::atomic::{AtomicUsize, Ordering};

static COUNTER: AtomicUsize = AtomicUsize::new(0);

COUNTER.fetch_add(1, Ordering::SeqCst);

락 없이 CPU의 원자적 명령을 직접 쓴다. Ordering 매개변수가 메모리 순서를 정하는데, SeqCst 가 가장 강하고 느리다. Relaxed, Acquire, Release 를 정확히 쓰려면 메모리 모델을 알아야 해서, 강의에서도 "확신이 없으면 SeqCst" 라는 조언이 나왔다.

전역 변수

static PACKETS_SERVED: AtomicUsize = AtomicUsize::new(0);   // OK

// static mut COUNTER: usize = 0;   // 접근하려면 unsafe

static mut 은 데이터 경합이 가능하므로 unsafe 가 필요하다. 대신 원자적 타입이나 Mutex 를 static 으로 두면 안전하게 쓸 수 있다. 초기화에 계산이 필요하면 OnceLock 이나 LazyLock 을 쓴다.

과제: 만델브로트 집합

같은 프로그램을 네 가지 방식으로 병렬화하는 과제였다. 비교가 정말 좋았다.

1. 단일 스레드 — 기준선.

2. 멀티 스레드 — 이미지를 가로 밴드로 쪼개 스레드에 하나씩 준다.

let bands = pixels.chunks_mut(bounds.0).enumerate().collect::<Vec<_>>();
let mut handles = Vec::new();

for (i, band) in bands.into_iter() {
    let mut band = band.to_owned();
    let handle = thread::spawn(move || {
        render(&mut band, band_bounds, band_upper_left, band_lower_right);
        band
    });
    handles.push(handle);
}

for (i, handle) in handles.into_iter().enumerate() {
    let band = handle.join().expect("Thread panicked");
    pixels[start..end].copy_from_slice(&band);
}

chunks_mut 이 벡터를 겹치지 않는 가변 슬라이스들로 쪼갠다는 게 핵심이다. 겹치지 않으니 각각을 다른 스레드에 줘도 안전하고, 컴파일러가 그걸 안다.

3. Rayon — 같은 일을 한 줄로.

use rayon::prelude::*;

let bands: Vec<(usize, &mut [u8])> = pixels.chunks_mut(bounds.0).enumerate().collect();

bands.into_par_iter().for_each(|(i, band)| {
    render(band, band_bounds, band_upper_left, band_lower_right);
});

into_iter() 를 into_par_iter() 로 바꾼 게 전부다. 스레드 개수도, join 도 신경 쓰지 않는다. 게다가 2번과 달리 밴드를 복사(to_owned)했다가 되쓸 필요가 없어서 더 빠르다.

4. 락 프리 — 밴드를 미리 나누지 않고, 원자적 카운터로 다음 작업을 집어간다.

pub struct AtomicChunksMut<'a, T> {
    slice: ...,
    step: usize,
    next: AtomicUsize,      // 다음 밴드 인덱스
}

let bands = AtomicChunksMut::new(&mut pixels, rows_per_band * bounds.0);

crossbeam::scope(|scope| {
    for _ in 0..threads {
        scope.spawn(|_| {
            for (i, band) in &bands {       // 원자적으로 다음 밴드를 가져온다
                render(band, band_bounds, band_upper_left, band_lower_right);
            }
        });
    }
});

만델브로트는 밴드마다 계산량이 크게 다르다. 집합 안쪽 픽셀은 반복 한도까지 전부 돌지만 바깥쪽은 금방 끝난다. 그래서 밴드를 미리 균등 분배하면 어떤 스레드는 놀고 어떤 스레드는 계속 일한다. 원자적 카운터로 작업을 동적으로 집어가게 하면 부하가 저절로 맞춰진다.

crossbeam::scope 도 중요한 도구다. 표준 thread::spawn 은 'static 을 요구해서 지역 변수를 빌릴 수 없는데, 스코프 스레드는 스코프가 끝나기 전에 반드시 join된다는 걸 보장하므로 지역 데이터를 빌릴 수 있다. 그래서 여기서는 to_owned() 복사가 필요 없다.

네 버전을 직접 돌려보면 성능 차이가 그대로 보인다. 그리고 그 차이가 어디서 오는지(작업 분배 방식, 복사 여부)를 코드로 설명할 수 있다는 게 이 과제의 핵심이었다.

정리

  • 접근법은 셋: Fork-Join, 채널, 공유 가변 상태
  • Send/Sync 가 스레드 안전성을 타입으로 정의하고, 나머지는 기존 소유권 규칙이 처리한다
  • 채널로 값을 보내는 건 이동이라, 공유 자체가 일어나지 않는다
  • Mutex<T> 는 데이터를 소유하므로 락을 잊을 수 없다. 다만 데드락은 막지 못한다
  • 빌림 규칙이 RwLock 으로 런타임에 그대로 재현된다
  • 작업량이 불균등하면 정적 분배보다 동적 분배(작업 훔치기)가 낫다

다음 편은 매크로 — macro_rules! 로 문법 자체를 확장하기.