스레드 간 메시지 전달

안전한 동시성을 보장하는 방법으로 점점 인기를 얻는 것이 메시지 전달(message passing)입니다. 스레드나 액터(actor)가 데이터를 담은 메시지를 서로 보내며 소통하는 방식입니다. Go 언어 문서의 구호가 이 생각을 잘 보여 줍니다. 메모리를 공유해서 소통하지 말고, 소통해서 메모리를 공유하라는 것입니다.

메시지를 보내는 동시성을 위해, 러스트의 표준 라이브러리는 채널(channel)의 구현을 제공합니다. 채널은 한 스레드에서 다른 스레드로 데이터를 보내는 일반적인 프로그래밍 개념입니다.

프로그래밍에서 채널은 개울이나 강 같은 물길처럼, 방향이 있는 물줄기라고 생각할 수 있습니다. 고무 오리 같은 것을 강에 띄우면 물길을 따라 끝까지 흘러갑니다.

채널에는 두 반쪽이 있습니다. 송신자(transmitter)와 수신자(receiver)입니다. 송신자 쪽은 고무 오리를 강에 띄우는 상류의 위치이고, 수신자 쪽은 고무 오리가 흘러 도착하는 하류의 위치입니다. 코드의 한 부분이 보내려는 데이터로 송신자의 메서드를 부르고, 다른 부분이 수신하는 쪽에서 도착한 메시지를 확인합니다. 송신자나 수신자의 반쪽이 버려지면 채널이 닫혔다(closed)고 합니다.

메시지 패싱을 사용하여 스레드 간 데이터 전송하기

여기서는 한 스레드가 값을 만들어 채널로 보내고, 다른 스레드가 값을 받아 출력하는 프로그램을 만들어 갑니다. 기능을 보여 주기 위해 스레드 사이에 간단한 값을 보냅니다. 이 기법에 익숙해지면, 채팅 시스템이나 여러 스레드가 계산의 일부를 맡아 그 결과를 하나의 스레드로 보내 모으는 시스템처럼, 서로 소통해야 하는 스레드에 채널을 쓸 수 있습니다.

먼저 다음 코드로 채널을 만들지만 아무것도 하지 않습니다. 러스트가 채널로 어떤 타입의 값을 보내고 싶은지 모르므로, 이 코드는 아직 컴파일되지 않습니다.

use std::sync::mpsc;

fn main() {
    let (tx, rx) = mpsc::channel();
}

mpsc::channel 함수로 새 채널을 만듭니다. mpsc는 다중 생산자, 단일 소비자(multiple producer, single consumer)를 뜻합니다. 간단히 말해, 러스트의 표준 라이브러리가 채널을 구현한 방식은 한 채널에 값을 만드는 송신 끝이 여럿 있을 수 있지만, 그 값을 소비하는 수신 끝은 하나뿐이라는 뜻입니다. 여러 개울이 모여 하나의 큰 강이 되는 것을 떠올려 보세요. 어느 개울에 띄운 것이든 끝에서는 하나의 강에 이릅니다. 지금은 생산자 하나로 시작하고, 이 예제가 동작하면 생산자를 여럿 더합니다.

mpsc::channel 함수는 튜플을 돌려줍니다. 첫 번째 원소는 송신하는 끝, 곧 송신자이고 두 번째 원소는 수신하는 끝, 곧 수신자입니다. tx와 rx라는 줄임말은 전통적으로 많은 분야에서 각각 송신자와 수신자를 뜻하는 데 쓰이므로, 각 끝을 나타내도록 변수 이름을 그렇게 지었습니다. 튜플을 분해하는 패턴과 함께 let 문을 썼는데, let 문에서의 패턴과 분해는 패턴 문법에서 다룹니다. 지금은 이런 식으로 let 문을 쓰는 것이 mpsc::channel이 돌려주는 튜플의 조각들을 꺼내는 편리한 방법이라는 것만 알면 됩니다.

송신하는 끝을 생성된 스레드로 옮기고 문자열 하나를 보내게 해, 생성된 스레드가 메인 스레드와 소통하게 합니다. 다음 코드가 그렇게 합니다. 상류에서 강에 고무 오리를 띄우거나, 한 스레드에서 다른 스레드로 채팅 메시지를 보내는 것과 같습니다.

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

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

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
    });
}

다시 thread::spawn으로 새 스레드를 만들고 move로 tx를 클로저 안으로 옮겨, 생성된 스레드가 tx를 소유하게 합니다. 생성된 스레드가 채널로 메시지를 보내려면 송신자를 소유해야 합니다.

송신자에는 보내고 싶은 값을 받는 send 메서드가 있습니다. send 메서드는 Result<T, E> 타입을 돌려주므로, 수신자가 이미 버려져 값을 보낼 곳이 없다면 보내는 작업이 오류를 돌려줍니다. 이 예제에서는 오류가 나면 패닉을 내도록 unwrap을 부릅니다. 실제 애플리케이션에서는 제대로 처리해야 합니다. 알맞게 오류를 처리하는 방법은 Result와 복구 가능한 오류를 다시 보세요.

다음 코드에서는 메인 스레드에서 수신자의 값을 받습니다. 강의 끝에서 물에 떠내려온 고무 오리를 건지거나 채팅 메시지를 받는 것과 같습니다.

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

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

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("Got: {received}");
}

수신자에는 쓸모 있는 메서드가 두 개 있습니다. recv와 try_recv입니다. 여기서는 receive의 줄임말인 recv를 쓰며, 이 메서드는 메인 스레드의 실행을 막고 채널로 값이 올 때까지 기다립니다. 값이 오면 recv는 그 값을 Result<T, E>에 담아 돌려줍니다. 송신자가 닫히면 recv는 더 이상 값이 오지 않는다는 신호로 오류를 돌려줍니다.

try_recv 메서드는 막지 않고, 곧바로 Result<T, E>를 돌려줍니다. 메시지가 있으면 그것을 담은 Ok 값을, 이번에는 메시지가 없으면 Err 값을 돌려줍니다. try_recv는 이 스레드가 메시지를 기다리는 동안 다른 일을 해야 할 때 쓸모가 있습니다. 가끔 try_recv를 부르는 루프를 쓰고, 메시지가 있으면 처리하고, 없으면 다시 확인하기 전까지 잠깐 다른 일을 하게 할 수 있습니다.

이 예제에서는 간단하게 하려고 recv를 썼습니다. 메인 스레드가 메시지를 기다리는 것 말고 할 일이 없으므로, 메인 스레드를 막는 것이 알맞습니다. 이 코드를 실행하면 메인 스레드에서 출력한 값을 볼 수 있습니다.

Got: hi

완벽합니다.

채널과 소유권 이동

소유권 규칙은 메시지를 보내는 데 중요한 역할을 합니다. 안전한 동시성 코드를 쓰도록 도와주기 때문입니다. 러스트 프로그램 전체에서 소유권을 생각하는 것의 이점이 동시성 프로그래밍에서 오류를 막는 것입니다. 채널과 소유권이 함께 어떻게 문제를 막는지 실험해 봅니다. 채널로 val 값을 보낸 뒤에, 생성된 스레드에서 그 값을 쓰려고 해 봅니다. 다음 코드를 컴파일해 왜 허용되지 않는지 확인합니다.

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

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

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
        println!("val is {val}");
    });

    let received = rx.recv().unwrap();
    println!("Got: {received}");
}

여기서는 tx.send로 채널에 val을 보낸 뒤에 val을 출력하려 합니다. 이것을 허용하는 것은 나쁜 생각입니다. 값을 다른 스레드로 보내고 나면, 그 스레드가 값을 다시 쓰기 전에 값을 바꾸거나 버릴 수 있습니다. 그러면 다른 스레드가 바꾼 데이터 때문에 일관되지 않거나 존재하지 않는 데이터로 오류나 예상하지 못한 결과가 생길 수 있습니다. 하지만 이 코드를 컴파일하려 하면 러스트가 오류를 냅니다.

error[E0382]: borrow of moved value: `val`
  --> src/main.rs:10:27
   |
 8 |         let val = String::from("hi");
   |             --- move occurs because `val` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(val).unwrap();
   |                 --- value moved here
10 |         println!("val is {val}");
   |                           ^^^ value borrowed here after move

For more information about this error, try `rustc --explain E0382`.

동시성의 실수가 컴파일 시점 오류가 되었습니다. send 함수는 매개변수의 소유권을 가져가고, 값이 옮겨지면 수신자가 그 소유권을 가집니다. 이렇게 하면 값을 보낸 뒤에 실수로 다시 쓰는 일을 막을 수 있습니다. 소유권 시스템은 모든 것이 괜찮은지 확인해 줍니다. 아래 그림은 송신자가 여럿일 때 값의 소유권과 채널의 종료가 어떻게 이어지는지 보여 줍니다.

let val = String::from("hi");
tx.send(val).unwrap();
// 이 뒤의 val 사용은 E0382
let received = rx.recv().unwrap();

let val = String::from("hi");
tx.send(val).unwrap();
// 이 뒤의 val 사용은 E0382
let received = rx.recv().unwrap();

std::sync::mpsc는 복제할 수 있는 여러 Sender와 하나의 Receiver를 제공합니다. send는 값을 소비하며, 받는 쪽이 끊어져 보내기가 실패하면 SendError 안에 원래 값이 들어 있어 되찾을 수 있습니다. 한 Sender가 보낸 값의 순서는 유지되지만, 여러 Sender의 값이 어떤 순서로 섞일지는 정해져 있지 않습니다. 그림에서 B1이 A1과 A2 사이에 들어간 것은 한 가지 경우일 뿐입니다. 버퍼에 값이 있으면 recv가 곧바로 꺼내고, 버퍼가 비었는데 Sender가 하나라도 남아 있으면 새 값을 기다립니다. Sender 하나가 버려졌다고 채널이 끝나지는 않습니다. 모든 Sender가 버려진 뒤에도 버퍼에 남은 값은 계속 받을 수 있고, 버퍼까지 비어야 recv가 RecvError를 돌려주며 for received in rx 반복도 끝납니다.

여러 값 보내기와 수신자가 기다리는지 알아보기

앞의 코드는 컴파일되고 실행되었지만, 두 스레드가 채널로 서로 이야기한다는 것을 분명히 보여 주지는 못했습니다. 다음 코드에서는 바꾼 코드가 동시에 실행된다는 것을 보여 주도록 고쳤습니다. 생성된 스레드가 이제 메시지 여러 개를 보내고, 각 메시지 사이에 1초씩 멈춥니다.

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

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

    thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for received in rx {
        println!("Got: {received}");
    }
}

이번에는 생성된 스레드에 메인 스레드로 보내고 싶은 문자열의 벡터가 있습니다. 그 벡터를 돌며 각 문자열을 따로 보내고, 그 사이에 Duration 값 1초로 thread::sleep 함수를 불러 멈춥니다.

메인 스레드에서는 더 이상 recv 함수를 직접 부르지 않습니다. 대신 rx를 반복자처럼 다룹니다. 받은 값마다 그것을 출력합니다. 채널이 닫히면 반복이 끝납니다.

이 코드를 실행하면, 각 줄 사이에 1초씩 멈추며 다음 출력이 나와야 합니다.

Got: hi
Got: from
Got: the
Got: thread

메인 스레드의 for 루프에는 멈추거나 지연시키는 코드가 없으므로, 메인 스레드가 생성된 스레드에서 값을 받기를 기다리고 있다는 것을 알 수 있습니다.

송신자를 복제하여 여러 생산자 만들기

앞에서 mpsc가 다중 생산자, 단일 소비자의 줄임말이라고 했습니다. mpsc를 쓰도록 앞의 코드를 넓혀, 같은 수신자에게 값을 보내는 스레드를 여럿 만듭니다. 다음 코드처럼 송신자를 복제하면 됩니다.

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

fn main() {
    // --생략--

    let (tx, rx) = mpsc::channel();

    let tx1 = tx.clone();
    thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx1.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    thread::spawn(move || {
        let vals = vec![
            String::from("more"),
            String::from("messages"),
            String::from("for"),
            String::from("you"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for received in rx {
        println!("Got: {received}");
    }

    // --생략--
}

이번에는 첫 번째 생성된 스레드를 만들기 전에 송신자에 clone을 부릅니다. 그러면 첫 번째 생성된 스레드에 넘길 새 송신자가 생깁니다. 원래의 송신자는 두 번째 생성된 스레드에 넘깁니다. 이렇게 하면 두 스레드가 생기고, 각 스레드는 같은 수신자에게 서로 다른 메시지를 보냅니다.

이 코드를 실행하면 출력은 다음과 비슷할 것입니다.

Got: hi
Got: more
Got: from
Got: messages
Got: for
Got: the
Got: thread
Got: you

시스템에 따라 값의 순서가 다르게 나올 수 있습니다. 이것이 동시성을 흥미로우면서도 어렵게 만드는 점입니다. thread::sleep으로 스레드마다 다른 값을 주며 실험해 보면, 실행할 때마다 더 비결정적이 되어 매번 다른 출력이 나올 것입니다. 두 스레드가 모두 끝나며 tx1과 tx가 모두 버려지면 채널이 닫히고, 메인 스레드의 for 루프도 끝납니다.

채널이 어떻게 동작하는지 보았으니, 이제 다른 동시성 방법을 봅니다.

더 깊이: 버퍼가 무한한 채널과 크기가 정해진 채널

mpsc::channel이 만드는 채널은 문서의 표현대로 버퍼가 무한한 채널이어서, send는 보내는 스레드를 결코 막지 않습니다. 그래서 받는 쪽이 보내는 쪽보다 느리면 보낸 값이 채널 안에 계속 쌓여 메모리를 씁니다. 이것을 막으려면 mpsc::sync_channel에 버퍼의 크기를 넘겨 크기가 정해진 채널을 만듭니다. 이 채널은 버퍼가 차면 send가 자리가 날 때까지 막히므로, 보내는 쪽이 받는 쪽의 속도에 맞춰집니다. 크기를 0으로 주면 랑데부 채널(rendezvous channel)이 되어, 모든 send는 짝이 되는 recv가 있을 때까지 돌아오지 않습니다.

표준 채널의 구현은 API를 그대로 둔 채 바뀌기도 했습니다. Rust 1.67 발표에 따르면 1.0 이전부터 있던 std::sync::mpsc의 구현이 이 버전에서 crossbeam-channel 크레이트를 바탕으로 한 구현으로 바뀌었습니다. 발표는 이것으로 여러 버그를 고치고 성능과 유지 보수성을 높였으며, 사용자는 동작에서 큰 차이를 느끼지 않을 것이라고 설명합니다. 수신자가 하나뿐이라는 제약은 그대로이므로, 여러 스레드가 한 채널에서 받아야 할 때는 채널과 execute 구현에서처럼 수신자를 Arc와 Mutex로 감싸 나누어 씁니다.

정리

• mpsc::channel()은 송신자 tx와 수신자 rx의 튜플을 돌려주며, 송신자는 여럿일 수 있지만 수신자는 하나입니다.
• send는 값의 소유권을 가져가 수신자에게 넘기므로, 보낸 뒤에 그 값을 쓰면 E0382 오류가 납니다. 수신자가 없으면 send는 원래 값을 담은 Err를 돌려줍니다.
• recv는 값이 올 때까지 막고, try_recv는 막지 않고 곧바로 결과를 돌려줍니다. rx를 for로 반복하면 받은 값을 차례로 처리합니다.
• tx.clone()으로 송신자를 복제해 여러 생산자를 만들 수 있으며, 한 송신자의 순서는 유지되지만 송신자 사이의 순서는 보장되지 않습니다.
• 모든 송신자가 버려지면 버퍼에 남은 값을 다 받은 뒤 recv가 RecvError를 돌려주고 for 반복이 끝납니다.

연습 문제

1. 반복이 끝나지 않는다

다음 코드는 네 값을 출력한 뒤 끝나지 않고 멈춰 있습니다. 까닭은 무엇이고 어떻게 고치나요?

let (tx, rx) = mpsc::channel();
for id in 0..4 {
    let tx = tx.clone();
    thread::spawn(move || {
        tx.send(id).unwrap();
    });
}
for received in rx {
    println!("Got: {received}");
}

2. 보낸 뒤에도 쓰고 싶다

생성된 스레드가 String을 보낸 뒤에도 그 내용을 로그로 남기고 싶습니다. 어떤 방법이 있나요?

다음 글 공유 상태 동시성에서는 여러 스레드가 같은 데이터에 접근하는 공유 상태 동시성을 Mutex<T>와 Arc<T>로 다룹니다.