#2.19. par - 병렬 처리 API

rok::core::par 모듈은 데이터 병렬 처리 API 를 제공합니다. Rust 의 Rayon 에서 영감을 받아 설계되었으며, task::* 와 동일한 work-stealing 스케줄러 위에서 동작합니다.

use rok::core::**; 가 묵시적으로 적용되기 때문에 par::for_each 나 par::map 등은 별도의 import 없이 바로 사용할 수 있습니다. 이때 모듈 prefix par:: 는 그대로 보존됩니다. 다만 병렬 처리 의도를 시각적으로 분명히 드러내는 자리에서는 use rok::core::par::*; 를 명시적으로 적어 주는 편을 권장합니다.

#2.19.1. 핵심 API

핵심 API 는 다음과 같이 여섯 개의 함수로 구성됩니다.

fn for_each<I: IntoIterator<Item = T>, T: Send>(iter: I, f: fn(T) + Send + Sync);
fn map<I: IntoIterator<Item = T>, T: Send, U: Send>(iter: I, f: fn(T) -> U + Send + Sync) -> List<U>;
fn reduce<T: Send + Sync>(iter: List<T>, init: T, f: fn(T, T) -> T + Send + Sync) -> T;
fn filter<T: Send>(iter: List<T>, pred: fn(T) -> bool + Send + Sync) -> List<T>;
fn join<T1: Send, T2: Send>(f1: fn() -> T1 + Send, f2: fn() -> T2 + Send) -> (T1, T2);    // 2 개 병렬
fn scope(f: fn(Scope));                                          // 동적 spawn

for_each / map / reduce / filter 의 f 는 여러 worker 가 동시에 부르므로 + Send + Sync 입니다.

• for_each / map / filter 는 입력 순서를 보존합니다. map / filter 의 결과 List 는 입력과 같은 순서입니다.
• for_each / map / filter 는 항목마다 worker 에 줍니다. 두 항목이 Sync 가 아닌 컬렉션 하나에 닿으면 안 됩니다 (2.13.9). 호출마다 먼저 두 항목에서 닿는 컬렉션이 있는지 찾습니다. 넘김의 복사처럼 항목의 구성 요소와 컬렉션의 원소를 깊이에 상관없이 따라가되, plain 값 (수, bool, str) 의 컬렉션 안으로는 들어가지 않습니다. 찾으면 항목들을 부른 스레드에서 하나씩 차례로 실행하므로 경쟁이 없고 제자리에서 바꾼 것은 worker 로 돌릴 때와 같이 남으며, 경고가 호출 위치를 알려 줍니다 (위치마다 한 번). slice 는 읽기만 합니다: 어느 항목도 리스트 자체를 들고 있지 않으면 항목들이 한 리스트의 slice (큰 리스트 하나의 조각) 를 통해 그것을 공유해도 되며, slice 가 닿는 원소는 하나씩 따라갑니다. 검사는 항목이 닿는 컬렉션마다 객체 header 에 표시하는 만큼 들며 (리스트 하나짜리 항목은 컬렉션마다 약 15 ns, 중첩된 항목은 약 50 ns, 대부분 메모리 읽기, 컬렉션이 없는 항목은 들지 않음), 일이 적은 작은 항목이 많으면 일보다 클 수 있습니다. 경고는 대신 쓸 것을 알려 줍니다: 항목마다 자기 값을 주거나 (공유하는 것은 저장하기 전에 .clone()), 읽기만 하는 데이터는 검사가 따라가지 않는 Frozen (2.16.4b) 이나 다른 Sync 타입으로 공유합니다. 항목 가운데 future 는 그 frame 으로 칩니다: 한 future 를 든 두 항목도 하나씩 실행합니다. par 는 항목을 넘기지 않으므로, frame 이 컬렉션을 캡처하는 future 는 항목이 될 수 없습니다 (컴파일 에러): 그런 future 는 task 로 spawn 합니다 (2.18.1). filter 의 결과는 입력의 항목을 공유합니다.
• reduce 는 입력을 chunk 로 나눠 chunk 마다 병렬로 접은 뒤, init 과 chunk 결과들을 순서대로 접습니다. 입력이 비어 있으면 init 을 돌려줍니다.
• scope 안에서 spawn 한 자식은 scope 가 끝날 때 모두 join 됩니다.
• par 는 task 와 같은 work-stealing pool (2.18.3) 위에서 실행됩니다. par 안에서 기다리는 worker 는 그동안 다른 task 를 실행하므로, task 안에서 par 를 불러도 deadlock 이 생기지 않습니다. scope 에 넘기는 Scope 와 그 핸들은 다음과 같습니다.

struct Scope { /* scope 안의 작업 목록 */ }
impl Scope {
    fn spawn<T: Send>(self, f: fn() -> T + Send) -> ScopeHandle<T>;
}

struct ScopeHandle<T> { /* spawn 된 작업 하나 */ }
impl<T> ScopeHandle<T> {
    fn join(self) -> T;          // 그 작업이 끝날 때까지 기다림
}

scope 가 끝날 때 join 되지 않은 작업도 모두 기다립니다.

#사용 예

실제 사용 모습은 다음과 같습니다.

use rok::core::par::*;

let xs: List<i32> = (0..1000).collect();

par::for_each(xs, |x| { process(x) });

let squared: List<i32> = par::map(xs, |x| x * x);

let total: i32 = par::reduce(squared, 0, |a, b| a + b);

let (a, b) = par::join(
    || compute_a(),
    || compute_b(),
);

par::scope(|s| {
    for chunk in chunks {
        s.spawn(|| process_chunk(chunk));
    }
});           // ← scope 끝에서 모든 spawn 완료 보장

#2.19.2. par::for_each vs Iterator

순차 처리와 병렬 처리는 호출 형태로 구분됩니다.

xs.into_iter().for_each(f);     // 순차 (single thread)
par::for_each(xs, f);            // 병렬 (worker pool)

작업은 chunk 단위의 data parallel 방식으로 분배되며, work-stealing 스케줄러가 작업량이 균일하지 않은 unbalanced workload 도 자동으로 균등화합니다.

#2.19.3. par::scope - 구조적 병렬

par::scope 는 동적으로 작업을 spawn 할 수 있는 구조적 병렬 API 입니다.

par::scope(|s| {
    let h1 = s.spawn(|| task1());
    let h2 = s.spawn(|| task2());
    // h1, h2 결과 사용 가능 (scope 안에서)
    // scope 끝에서 자동 join
});

이는 structured concurrency 의 형태를 따르며, scope 는 모든 자식 spawn 이 완료될 때까지 block 합니다.

#par::try_scope - 첫 에러에서 취소

에러를 던지는 자식을 두려면 try_scope 를 씁니다.

fn try_scope<E: Send>(f: fn(TryScope<E>) throws (E)) throws (E);

impl<E> TryScope<E> {
    fn spawn<T: Send>(self, f: fn() throws (E) -> T + Send) -> TryScopeHandle<T, E>;
    fn is_cancelled(self) -> bool;
}

impl<T, E: Send> TryScopeHandle<T, E> {
    fn join(self) throws (E) -> T;     // 자기 에러, 또는 시작 전에 취소됐으면 scope 의 첫 에러
}

자식 하나가 에러를 던지면 scope 가 취소됩니다. 아직 시작하지 않은 자식은 실행되지 않고, 이미 실행 중인 자식은 끝까지 실행됩니다. 모든 자식이 끝나면 try_scope 가 그 첫 에러를 다시 던집니다. 실행 중인 자식은 is_cancelled() 를 보고 일찍 끝낼 수 있습니다.

try par::try_scope(|s| {
    let a = s.spawn(|| try load("a"));
    let b = s.spawn(|| try load("b"));
    use_both(try a.join(), try b.join());
});

#2.19.4. par::reduce - 결합 법칙 요구

par::reduce 의 시그니처는 다음과 같으며, 사용 시 몇 가지 수학적 조건을 함께 고려해야 합니다.

fn reduce<T: Send + Sync>(iter: List<T>, init: T, f: fn(T, T) -> T + Send + Sync) -> T;

먼저 결합 함수 f 가 결합 법칙(associative) 을 만족해야 결과가 결정론적으로 산출됩니다. 그리고 init 값은 가능하면 identity 로 두기를 권장하는데, 예를 들어 + 연산이라면 0, * 연산이라면 1 이 적절합니다. 반대로 부동소수점 합산처럼 큰 값이 누적되는 비결합 함수의 경우에는 결과가 비결정적 이 될 수 있으므로 명시적인 주의가 필요합니다.

let sum: i64 = par::reduce([1, 2, 3, 4, 5], 0, |a, b| a + b);    //→ 15

// 부동소수점 - 결과 약간 다를 수 있음
let total: f64 = par::reduce(values, 0.0, |a, b| a + b);

#2.19.5. par::join - 정확히 2 개

두 개의 작업을 동시에 수행하고 결과 쌍을 받아야 할 때 par::join 을 사용합니다.

let (a, b) = par::join(
    || expensive_computation_1(),
    || expensive_computation_2(),
);

두 개의 작업은 fork-join 패턴 중 가장 흔한 형태이므로 별도의 함수로 제공됩니다. 만약 N 개의 작업을 다루어야 한다면 par::for_each 나 par::scope 를 사용하면 됩니다.

#2.19.6. SIMD / GPU 통합

par 모듈은 CPU 데이터 병렬을 중심에 둡니다. SIMD multiversion 은 par::map 내부에서 target_feature 에 따라 자동으로 적용되며, GPU 가속은 표준 라이브러리에 속하지 않으며, 라이브러리가 rok::bare::mlir (1.9) 로 다룹니다.

par::map(xs, |x| heavy_op(x));     // CPU work-stealing + SIMD 자동

#2.19.7. 모듈 매핑

각 API 의 모듈 위치와 내부 전략은 아래 표와 같이 정리됩니다.

내부적으로는 task::spawn 위에 한 겹 layer 를 얹어 동일한 스케줄러를 공유하는 구조입니다.

GitHub에서 이 페이지 고치기