공유 상태 동시성

메시지 패싱은 동시성을 다루는 좋은 방법이지만, 유일한 수단은 아닙니다.

또 다른 방법은 여러 스레드가 동일한 공유 데이터에 접근하는 것입니다.

고(Go) 언어 문서로부터 나온 슬로건의 일부를 다시 한번 생각해 보세요: ‘메모리를 공유하여 통신하지 마세요.’

메모리를 공유하는 통신은 어떻게 생겼을까요?

더불어서 메시지 패싱 애호가들은 왜 메모리 공유를 쓰지 말라고 경고할까요?

이 절에서 쓰는 러스트 채널은 값을 send에 넘길 때 소유권을 전달합니다. 앞 예제의 String은 Copy가 아니므로 보낸 뒤 원래 변수를 다시 사용할 수 없습니다. Copy 값은 복사해서 보낼 수 있으며, 다른 언어의 채널도 같은 소유권 규칙을 따른다고 일반화할 수는 없습니다.

공유 메모리 동시성은 복수 소유권과 유사합니다.

여러 스레드들이 동시에 동일한 메모리 위치에 접근할 수 있지요.

스마트 포인터가 복수 소유권을 가능하게 하는 내용을 담은 14장에서 보셨듯이, 복수 소유권은 서로 다른 소유자들에 대한 관리가 필요하기 때문에 더 복잡할 수 있습니다.

러스트의 타입 시스템과 소유권 규칙은 이러한 관리가 올바르도록 훌륭히 도와줍니다.

예를 들면, 공유 메모리를 위한 더 일반적인 동시성 기초 재료 중 하나인 뮤텍스(mutex)를 살펴봅시다.

뮤텍스를 사용하여 한 번에 한 스레드에서의 데이터 접근을 허용하기

뮤텍스는 상호 배제(mutual exclusion)의 줄임말로, 뮤텍스에서는 한 번에 하나의 스레드만 데이터에 접근하도록 허용합니다.

뮤텍스 내부의 데이터에 접근하려면 스레드는 먼저 뮤텍스의 락(lock)을 얻는 요청을 해서 접근을 희망하는 신호를 보내야 합니다.

락은 누가 현재 배타적으로 데이터에 접근하는지 추적하는 뮤텍스의 일부에 해당하는 데이터 구조입니다.

그러므로, 뮤텍스는 잠금 시스템을 통해 가지고 있는 데이터를 보호하는 것(guard)으로 묘사됩니다.

뮤텍스는 사용하기 어렵다는 평판이 있는데 이는 다음 두 가지 규칙을 기억해야 하기 때문입니다.

• 데이터를 사용하기 전에는 반드시 락을 얻는 시도를 해야 합니다.
• 만일 뮤텍스가 보호하는 데이터의 사용이 끝났다면, 반드시 언락을 해야 다른 스레드들이 락을 얻을 수 있습니다.

뮤텍스는 마이크가 하나뿐인 컨퍼런스 패널 토의에 비유할 수 있습니다.

패널 참가자가 말하려면 먼저 마이크 사용을 요청하거나 신호를 줘야 합니다.

마이크를 얻었을 때는 원하는 만큼 길게 말한 다음, 말하기를 원하는 다음 패널 참가자에게 마이크를 건네줍니다.

만일 패널 참여자가 마이크 사용을 끝냈을 때 이를 건네주는 것을 잊어먹는다면, 그 외 아무도 말할 수 없게 됩니다.

공유하는 마이크의 관리가 잘못되면, 패널 토의는 계획대로 진행되지 않을 겁니다!

뮤텍스의 관리를 올바르게 하려면 믿을 수 없을 만큼 까다로울 수 있는데, 이것이 바로 많은 사람들이 채널 애호가가 되는 이유입니다.

러스트의 타입 시스템은 보호된 데이터에 접근할 때 guard를 요구하고, 소유권 규칙은 guard가 drop될 때 락을 해제하도록 돕습니다. 다만 데드락이나 지나치게 긴 락 보유까지 막지는 않습니다.

Mutex<T>의 API

뮤텍스 사용 방법에 대한 예제로, 예제 15-12처럼 싱글스레드 컨텍스트에서 뮤텍스를 사용하는 것으로 시작해봅시다.

use std::sync::Mutex;

fn main() {
    let m = Mutex::new(5);

    {
        let mut num = m.lock().unwrap();
        *num = 6;
    }

    println!("m = {:?}", m);
}

많은 타입이 그렇듯 Mutex<T>는 연관 함수 new를 사용하여 만들어집니다.

뮤텍스 내의 데이터에 접근하기 위해서는 lock 메서드를 사용하여 락을 얻습니다.

다른 스레드가 이미 락을 보유하고 있다면 이 호출은 현재 스레드가 락을 얻을 때까지 기다립니다. 현재 스레드는 lock 다음 줄로 진행하지 못하지만, 락을 보유한 스레드나 다른 스레드까지 멈추는 것은 아닙니다. 락을 즉시 얻을 수 있다면 기다리지 않고 guard를 받습니다.

락을 보유한 스레드의 panic이 감지되면 뮤텍스는 poisoned 상태가 될 수 있고, 이후 lock은 PoisonError를 반환합니다.

이 예제는 unwrap으로 오류를 전파하여 현재 스레드도 panic하게 합니다. 영구적으로 잠긴다는 뜻은 아닙니다. 표준 라이브러리 문서처럼 PoisonError::into_inner로 guard를 얻어 데이터를 점검하고 복구할 수도 있습니다.

락을 얻고 난 후에는 그 반환 값, 지금의 경우 num이라는 이름의 값을 내부 데이터에 대한 가변 참조자로 취급할 수 있습니다.

타입 시스템은 m 내부의 값을 사용하기 전에 락을 얻도록 보장합니다.

Mutex<i32>는 i32가 아닙니다. 이 예처럼 공유 참조에서 보호된 i32를 변경하려면 lock으로 guard를 얻습니다. 뮤텍스 자체를 유일하게 가변 대여할 수 있는 별도 상황에는 get_mut 경로도 있으므로 모든 API가 락 획득을 요구한다고 일반화하지 마세요.

이 공유 접근 경로에서 타입 시스템은 guard 없이 내부의 i32를 직접 변경하는 것을 허용하지 않습니다.

짐작하셨을지 모르겠지만 Mutex<T>의 API에는 스마트 포인터가 사용됩니다.

더 정확하게는 lock의 호출이 MutexGuard라는 스마트 포인터를 반환하는데, unwrap 호출을 통해 처리되는 LockResult로 감싸져 있습니다.

MutexGuard 스마트 포인터는 내부 데이터를 가리키도록 Deref가 구현되어 있습니다; 또한 MutexGuard 스마트 포인터에는 Drop 구현체가 있는데, 이것으로 내부 스코프의 끝에서 스코프 밖으로 벗어났을 때 자동으로 락을 해제하는 일이 벌어집니다.

이 예제에서는 내부 스코프가 끝날 때 guard가 drop되어 락이 자동으로 해제됩니다. guard를 오래 보유하거나 여러 락을 잘못된 순서로 획득하면 대기나 데드락이 발생할 수 있습니다.

락이 버려진 후에는 뮤텍스 값을 출력하여 내부의 i32를 6으로 바꿀 수 있음을 확인할 수 있습니다.

예제 15-12에서 락을 얻고, 값을 바꾸고, 스코프 끝에서 락이 풀리는 흐름은 MutexGuard의 수명과 함께 읽으면 다음처럼 정리됩니다.

let m = Mutex::new(5);
{
    let mut num = m.lock().unwrap();
    *num = 6;
} // num drop → unlock

let m = Mutex::new(5);
{
    let mut num = m.lock().unwrap();
    *num = 6;
} // num drop → unlock

여러 스레드 사이에서 Mutex<T> 공유하기

이제 Mutex<T>를 사용하여 여러 스레드 사이에서 값을 공유하는 시도를 해봅시다.

10개의 스레드를 생성하고 각자 카운터 값을 1씩 증가시켜서 카운터가 0에서 10으로 가도록 할 것입니다.

다음 예제 15-13은 컴파일 에러가 날 것이고, 이 에러를 이용하여 Mutex<T>를 사용하는 방법과 러스트가 이를 고치는 것을 돕는 방법에 대해 학습하겠습니다.

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

fn main() {
    let counter = Mutex::new(0);
    let mut handles = vec![];

    for _ in 0..10 {
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();

            *num += 1;
        });
        handles.push(handle);
    }

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

    println!("Result: {}", *counter.lock().unwrap());
}

예제 15-12에서 했던 것처럼 Mutex<T> 내부에 i32를 담고 있는 counter 변수를 만듭니다.

다음으로 숫자 범위만큼 반복하여 10개의 스레드를 만듭니다.

thread::spawn을 사용하고 모든 스레드에게 동일한 클로저를 주었습니다.

이 클로저는 카운터를 스레드로 이동시키고, lock 메서드를 호출하여 Mutex<T>의 락을 얻은 다음, 뮤텍스 내의 값을 1만큼 증가시킵니다.

스레드가 자신의 클로저 실행을 끝냈을 때, num은 스코프 밖으로 벗어나고 락이 해제되어 다른 스레드가 이를 얻을 수 있습니다.

메인 스레드에서는 조인 핸들을 전부 모읍니다.

그러고 나서 예제 15-2에서처럼 각 핸들에 join을 호출하여 모든 스레드가 종료되는 것을 확실히 합니다.

그 시점에서 메인 스레드는 락을 얻고 이 프로그램의 결과를 출력합니다.

이 예제가 컴파일되지 않을 것이라고 암시했었죠.

이제 왜 그런지 알아봅시다!

$ cargo run
   Compiling shared-state v0.1.0 (file:///projects/shared-state)
error[E0382]: use of moved value: `counter`
  --> src/main.rs:9:36
   |
5  |     let counter = Mutex::new(0);
   |         ------- move occurs because `counter` has type `Mutex<i32>`, which does not implement the `Copy` trait
...
9  |         let handle = thread::spawn(move || {
   |                                    ^^^^^^^ value moved into closure here, in previous iteration of loop
10 |             let mut num = counter.lock().unwrap();
   |                           ------- use occurs due to use in closure

For more information about this error, try `rustc --explain E0382`.
error: could not compile `shared-state` due to previous error

이 에러 메시지는 counter 값이 루프의 이전 반복에서 이동되었다고 설명합니다.

러스트는 락 counter의 소유권을 여러 스레드로 옮길 수 없음을 말하고 있습니다.

14장에서 설명했던 복수 소유자 메서드를 가지고 이 컴파일 에러를 고쳐봅시다.

복수 스레드와 함께하는 복수 소유권

14장에서는 스마트 포인터 Rc<T>를 사용하여 참조 카운팅 값을 만들어 값에 여러 소유자를 부여했습니다.

여기서도 똑같이 해서 어떻게 되는지 봅시다.

예제 15-14의 Mutex<T>를 Rc<T>로 감싸서 스레드로 소유권을 넘기기 전에 그 Rc<T>를 복제하겠습니다.

use std::rc::Rc;
use std::sync::Mutex;
use std::thread;

fn main() {
    let counter = Rc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Rc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();

            *num += 1;
        });
        handles.push(handle);
    }

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

    println!("Result: {}", *counter.lock().unwrap());
}

다시 한번 컴파일을 하고 그 결과가... 다른 에러들이네요!

컴파일러는 많은 것을 가르쳐 주고 있습니다.

$ cargo run
   Compiling shared-state v0.1.0 (file:///projects/shared-state)
error[E0277]: `Rc<Mutex<i32>>` cannot be sent between threads safely
  --> src/main.rs:11:36
   |
11 |           let handle = thread::spawn(move || {
   |                        ------------- ^------
   |                        |             |
   |  ______________________|_____________within this `[closure@src/main.rs:11:36: 11:43]`
   | |                      |
   | |                      required by a bound introduced by this call
12 | |             let mut num = counter.lock().unwrap();
13 | |
14 | |             *num += 1;
15 | |         });
   | |_________^ `Rc<Mutex<i32>>` cannot be sent between threads safely
   |
   = help: within `[closure@src/main.rs:11:36: 11:43]`, the trait `Send` is not implemented for `Rc<Mutex<i32>>`
note: required because it's used within this closure
  --> src/main.rs:11:36
   |
11 |         let handle = thread::spawn(move || {
   |                                    ^^^^^^^
note: required by a bound in `spawn`
  --> /rustc/d5a82bbd26e1ad8b7401f6a718a9c57c96905483/library/std/src/thread/mod.rs:704:8
   |
   = note: required by this bound in `spawn`

For more information about this error, try `rustc --explain E0277`.
error: could not compile `shared-state` due to previous error

와우, 이 에러는 정말 장황하네요!

여기 집중할 중요한 부분이 있습니다.

`Rc<Mutex<i32>>` cannot be sent between threads safely(`Rc<Mutex<i32>>`는 스레드간에 안전하게 보낼 수 없습니다).

또한 컴파일러는 그 이유를 말해주고 있습니다.

the trait `Send` is not implemented for `Rc<Mutex<i32>>`(트레이트 `Send`가 `Rc<Mutex<i32>>` 에 대해 구현되지 않았습니다).

Send에 대해서는 다음 절에서 이야기하겠습니다.

이것은 스레드와 함께 사용하는 타입들이 동시적 상황에서 쓰이기 위한 것임을 보장하는 트레이트 중 하나입니다.

안타깝게도, Rc<T>는 스레드를 교차하면서 공유하기에는 안전하지 않습니다.

Rc<T>가 참조 카운트를 관리할 때, 각 clone 호출마다 카운트에 더하고 각 클론이 버려질 때 카운트에서 제합니다.

하지만 Rc<T>는 여러 스레드가 카운트를 갱신할 때 서로 간섭하지 않도록 보장하는 동기화 수단을 사용하지 않습니다.

이는 잘못된 카운트를 야기할 수 있습니다. 결과적으로 메모리 누수를 발생시키거나 아직 다 쓰기 전에 값이 버려질 수 있는 미세한 버그를 낳겠죠.

우리가 원하는 것은 정확히 Rc<T>와 비슷하지만 스레드-안전한 방식으로 참조 카운트를 바꾸는 녀석입니다.

Arc<T>를 이용한 아토믹 참조 카운팅

Arc<T>는 참조 카운트를 원자적으로 갱신하여 여러 스레드가 같은 값의 소유권을 공유하도록 도와줍니다. 다만 Arc<T>의 Send·Sync 구현에는 T: Send + Sync가 필요하며, 내부 데이터를 함께 변경할 때는 이 예제처럼 Mutex<T>로 접근을 보호해야 합니다.

a는 아토믹(atomic)을 의미하는데, 즉 이것이 원자적으로 참조자를 세는 (atomically reference counted) 타입임을 뜻합니다.

아토믹은 추가적인 종류의 동시성 기초 재료로서, 여기서는 자세히 다루지 않을 겁니다.

더 자세히 알고 싶으면 std::sync::atomic에 대한 표준 라이브러리 문서를 보세요.

이 시점에서는 아토믹이 기초 타입처럼 동작하지만 스레드를 교차하며 공유해도 안전하다는 것만 알면 됩니다.

그렇다면 여러분은 왜 모든 기초 타입이 아토믹하지 않은지, 그리고 표준 라이브러리 타입은 왜 기본적으로 Arc<T>를 구현에 이용하지 않는지를 궁금해할지도 모르겠습니다.

그 이유는 스레드 안전성이란 것이 정말로 필요할 때만 감내하고 싶을 성능 저하를 일으키기 때문입니다.

싱글스레드 내에서만 값을 연산하는 경우, 아토믹이 제공하는 보장을 강제할 필요 없이 코드는 더 빠르게 실행될 수 있습니다.

예제로 다시 돌아갑시다.

이 예제에서 사용하는 new와 clone은 Arc<T>와 Rc<T> 모두 제공합니다. 따라서 여기서는 use 라인과 그 호출의 타입 이름을 바꾸어 수정합니다. 두 타입의 모든 API와 스레드 경계가 같다는 뜻은 아닙니다.

예제 15-15의 코드는 마침내 컴파일 및 실행이 될 것입니다.

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

fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();

            *num += 1;
        });
        handles.push(handle);
    }

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

    println!("Result: {}", *counter.lock().unwrap());
}

이 코드는 다음을 출력할 것입니다.

Result: 10

해냈군요!

크게 인상적인 것처럼 보이지 않을지도 모르겠지만 0부터 10까지 세었고, Mutex<T>와 스레드 안전성에 대하여 많은 것을 알게 해 주었습니다.

또한 이 프로그램의 구조를 사용하여 카운터만 증가시키는 것보다 더 복잡한 연산을 할 수도 있습니다.

이 전략을 사용하여 계산을 독립적인 부분들로 나누고, 해당 부분들을 스레드로 쪼갠 다음, Mutex<T>를 사용하여 각 스레드가 해당 부분의 최종 결과를 업데이트하도록 할 수 있습니다.

Arc<Mutex<T>> 조합은 컴파일 에러를 지나 최종 구조에 도달하는 흐름으로 보면 더 선명합니다.

단순한 정수 갱신에는 표준 라이브러리의 std::sync::atomic 모듈의 정수 원자 타입도 선택할 수 있습니다. 이들은 Mutex<T>와 별도 타입이며 Mutex<T> 자체는 std::sync에 있습니다. 여러 값 사이의 불변식을 함께 지키는 연산은 단일 원자 정수만으로 해결되는지 따로 판단해야 합니다.

이들 타입은 기초 타입에 대한 안전하고, 동시적이며, 원자적인 접근을 제공합니다.

이 예제에서는 기초 타입에 대해 Mutex<T>를 사용하여 Mutex<T>가 동작하는 방식에 집중하였습니다.

RefCell<T>/Rc<T>와 Mutex<T>/Arc<T> 간의 유사성

counter가 불변이지만 내부 값에 대한 가변 참조자를 가져올 수 있었음을 알아채셨을 수도 있습니다; 이는 Cell 가족이 그러하듯 Mutex<T>가 내부 가변성을 제공한다는 의미입니다.

14장에서 Rc<T>의 내용물을 변경할 수 있도록 하기 위해 RefCell<T>를 사용한 것과 같은 방식으로, Arc<T> 내부의 값을 변경하기 위해 Mutex<T>를 이용합니다.

주목할 만한 또 다른 세부 사항은 Mutex<T>를 사용할 때 러스트가 모든 종류의 논리 에러로부터 보호해줄 수 없다는 것입니다.

14장에서 Rc<T>를 사용하는 것은 두 Rc<T> 값들이 서로를 참조하여 메모리 누수를 야기하는 순환 참조자를 만들 위험성이 따라오는 것이었음을 상기해봅시다.

이와 유사하게, Mutex<T>에는 데드락(deadlock)을 생성할 위험성이 따라옵니다.

이것은 어떤 연산이 두 개의 리소스에 대한 락을 얻을 필요가 있고 두 개의 스레드가 락을 하나씩 얻는다면, 서로가 서로를 영원히 기다리는 형태로 발생됩니다.

데드락을 분석하려면 먼저 종이에 두 작업의 락 보유·대기를 표시해 보세요. A가 X를 보유하고 Y를 기다리며, B가 Y를 보유하고 X를 기다리는 상태에서는 어느 작업도 다음 락을 얻지 못합니다. 두 작업이 같은 X→Y 순서로 획득하도록 바꾸면 이 역순 획득의 순환 대기 원인을 제거할 수 있습니다.

Mutex<T>와 MutexGuard에 대한 표준 라이브러리 API 문서가 유용한 정보를 제공합니다.

예시

Arc의 소유권 공유, 짧은 임계 구역, 완료 뒤 읽기를 별개의 줄로 확인하는 전체 프로그램입니다.

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

fn main() {
    let total = Arc::new(Mutex::new(0));
    let mut handles = Vec::new();
    for amount in [2, 4, 6] {
        let shared = Arc::clone(&total);
        handles.push(thread::spawn(move || {
            let mut guard = shared.lock().expect("poisoned");
            *guard += amount;
        }));
    }
    for handle in handles {
        handle.join().expect("worker panicked");
    }
    println!("total={}", *total.lock().expect("poisoned"));
}

모든 작업이 정상 종료하면 total=12입니다. Arc::clone은 같은 Mutex 할당의 소유자를 늘리고, guard를 통한 더하기는 한 작업씩 직렬화합니다. 어느 작업이 먼저 락을 얻는지는 정하지 않아도 각 덧셈을 정확히 한 번 완료한 뒤의 합은 같습니다.

어디까지 함께 잠가야 할까요? 여기서 *guard += amount는 현재 값을 읽고 더한 값을 쓰기까지 같은 guard를 유지합니다. 읽기와 쓰기를 각각 잠가도, 그 사이에 guard를 버리면 다른 작업이 값을 바꿀 수 있습니다. 그러면 오래된 복사본으로 나중에 덮어쓰는 갱신 손실이 생깁니다. 데이터에 대한 배타 접근과 여러 단계로 된 업무 연산의 정확성은 함께 검토해야 합니다.

스스로 잠금과 완료 구별하기

별도 정상 실행에서 초기 값을 5, 세 amount를 1·2·3으로 바꾸면 모든 join 뒤의 값은 무엇인가요? 세 join을 생략하고 main이 락만 얻어 읽는다면 그 값도 고정되나요? 마지막으로 main이 먼저 같은 Mutex의 guard를 얻어 살아 있게 둔 채, 그 락이 필요한 작업의 join을 기다린다고 합시다. 작업과 main이 각각 무엇을 기다리는지 설명하세요.

공유 접근과 유일 접근의 차이는 Mutex::lock 및 get_mut, payload의 조건은 Arc의 스레드 안전성을 읽어 확인하세요.

이제 Send와 Sync 트레이트와 이를 커스텀 타입과 함께 사용하는 방법을 이야기하는 것으로 이 장을 마무리하겠습니다.

공유 상태 동시성 학습 판단 기준표

이 다이어그램은 공유 상태 동시성 내용을 예제 코드에 적용할 때 먼저 볼 조건과 나중에 검증할 조건을 나눕니다.

각 접근은 안전한데 합이 틀리는 경우

다음 전체 프로그램의 두 작업은 각각 2와 4를 더하려 합니다. Barrier::new(2)는 두 작업이 모두 wait()에 도착해야 그다음 줄로 진행하게 하는 만남 지점입니다. 같은 시각에 CPU에서 실행된다는 보장은 아닙니다. 이 프로그램에서는 두 작업이 이전 값을 읽은 뒤에만 쓰기를 시작하도록 사용합니다. 두 스레드 생성과 join은 성공하고, 패닉·외부 종료는 없다고 가정하세요.

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

fn main() {
    let total = Arc::new(Mutex::new(0_i32));
    let gate = Arc::new(Barrier::new(2));
    let mut handles = Vec::new();
    for amount in [2, 4] {
        let shared = Arc::clone(&total);
        let gate = Arc::clone(&gate);
        handles.push(thread::spawn(move || {
            let before = {
                let guard = shared.lock().unwrap();
                *guard
            };
            gate.wait();
            let mut guard = shared.lock().unwrap();
            *guard = before + amount;
        }));
    }
    for handle in handles {
        handle.join().unwrap();
    }
    println!("total={}", *total.lock().unwrap());
}

가능한 최종 출력 값을 모두 찾고 total=6이 가능한지 설명하세요. 모든 읽기·쓰기에 락이 있는데도 왜 덧셈이 손실되는지, 클로저 내부를 어떻게 고쳐야 하는지도 답하세요. 마지막으로 읽기부터 쓰기까지 같은 guard를 유지한 채 gate.wait()를 호출하는 수정은 안전한지 대기 관계로 판단하세요.

gate.wait();
let mut guard = shared.lock().unwrap();
*guard += amount;

만남 지점의 조건은 Barrier::wait, 스코프를 벗어날 때의 해제는 MutexGuard의 설명과 연결됩니다.