공유 상태 동시성

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

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

고(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

여러 스레드 사이에서 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의 설명과 연결됩니다.