동시성

2024.04.09 23:07

대부분의 요즘 운영 체제에서, 실행되는 프로그램의 코드는 프로세스 내에서 실행되고, 프로그램 내에서도 동시에 실행되는 독립적인 스레드를 가진다.
계산 부분을 여러 개의 스레드로 쪼개는 것은 프로그램이 동시에 여러 개의 일을 할 수 있기 때문에 성능을 향상시킬 수 있지만, 프로그램을 복잡하게 만들기도 한다.
러스트 표준 라이브러리는 언어 런타임의 크기를 작게 유지하기 위해 1:1 스레드 구현만 제공한다.

스레드 생성

새로운 스레드를 생성하기 위해서는 thread::spawn 함수를 호출하고, 새로운 스레드 내에서 실행하기를 원하는 코드가 담겨 있는 클로저를 넘기면 된다.

use std::thread;
use std::time::Duration;

fn main() {
    thread::spawn(|| {
        for i in 1..10 {
            println!("hi number {} from the spawned thread!", i);
            thread::sleep(Duration::from_millis(1));
        }
    });

    for i in 1..5 {
        println!("hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(1));
    }
}

JoinHandle

thread::spawn의 반환 타입은 JoinHandle이며, join 메소드를 호출했을 때 그 스레드가 끝날때까지 기다리는 소유된 값이다.
JoinHandle을 저장하면 스레드가 완전히 실행되는 것을 보장할 수 있다.
핸들에 대해 join을 호출하는 것은 핸들에 대한 스레드가 종료될 때까지 현재 실행중인 스레드를 블록하여 그 스레드의 작업을 수행하거나 종료되는 것을 방지한다.

use std::thread;
use std::time::Duration;

fn main() {
    let handle = thread::spawn(|| {
        for i in 1..10 {
            println!("hi number {} from the spawned thread!", i);
            thread::sleep(Duration::from_millis(1));
        }
    });

    for i in 1..5 {
        println!("hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(1));
    }

    handle.join().unwrap();
}

스레드에 move 클로저 사용하기

어떤 스레드의 데이터를 다른 스레드 내에서 사용하도록 하기 위해 move 클로저와 thread::spawn를 함께 사용할 수 있다.
v에 대한 소유권이 메인 스코프에 속하기 때문에 이 예제에서 move 클로저를 사용하지 않으면 컴파일러는 새로 생성한 Thread가 v를 안전하게 사용할 수 없다고 판단한다. 따라서 move를 함께 명시해줘야한다.

use std::thread;

fn main() {
    let v = vec![1, 2, 3];

    let handle = thread::spawn(move || {
        println!("Here's a vector: {:?}", v);
    });

    handle.join().unwrap();
}

메세지 패싱

mpsc::channel 함수를 사용하여 새로운 채널을 생성할 수 있다.
- mpsc는 복수 생성자, 단수 소비자 (multiple producer, single consumer)를 나타낸다.

하위 스레드에서 채널로 string을 send하고, 메인 스레드에서 받아 출력하는 예제이다.

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

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);
}

/*
출력 결과:
Got: hi
*/

채널의 수신 단말은 recv(), try_recv() 두가지의 메서드를 가지고 있다. (receive 의 줄임말이다.)
- recv()는 메인 스레드의 실행을 블록시키고 채널로부터 값이 보내질 때까지 기다릴 것이다.
- 그리고 값이 전달되면 recv()는 Result<T, E> 형태로 이를 반환하고, 채널의 송신 단말이 닫히면 더 이상 어떤 값도 오지 않을 것이라는 의미의 에러를 반환할 것이다.

생성자가 여러개인 예시를 살펴보자.

// --snip--

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

let tx1 = mpsc::Sender::clone(&tx);
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);
}

// --snip--

mpsc::Sender::clone(&tx);을 통해 생성자를 복제해서 각각의 생성자를 두 개의 하위 스레드에서 사용했다.

공유 상태 동시성

뮤텍스는 상호 배제 (mutual exclusion)의 줄임말로서, 주어진 시간에 오직 하나의 스레드만 데이터 접근을 허용한다.
Mutex<T>는 연관함수 new를 사용하여 만들어지고, lock 메소드를 사용하여 락을 얻는다.

Mutex<T>를 사용하여 여러 스레드들 사이에서 값을 공유해보자. 10개의 스레드를 돌리고 이들이 카운터 값을 1만큼씩 증가 시켜서, 카운터를 0에서 10으로 증가시키는 예제이다.

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());
}

사실 이 코드에선 예외가 발생한다! counter의 소유권을 여러 스레드로 이동시킬 수 없기 때문이다.

이 문제를 해결하려면 복수 소유자 메소드인 Rc의 Thread-safe 버전인 Arc를 사용해야한다.

use std::sync::{Mutex, Arc};
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
*/

Sync와 Send Trait

Rust에서는 언어상에서 지원하는 동시성 기능이 매우 적다. 대신 위에서 살펴봤던 기능들은 모두 표준 라이브러리에 구현되어있는 것이다.
Send 마커 트레잇은 Send가 구현된 타입의 소유권이 스레드 사이에서 이전될 수 있음을 나타낸다.
- 대부분의 Rust 타입이 Send이지만 예외가 있다. 대표적으로 Rc<T>는 클론하여 다른 스레드로 복제본의 소유권을 전송하는 경우 두 스레드 모두 동시에 참조 카운트 값을 갱신할 가능성이 있기 때문에 Send가 될 수 없다.
- Send 타입으로 구성된 어떤 타입은 또한 자동적으로 Send로 마킹된다.
Sync 마커 트레잇은 Sync가 구현된 타입이 여러 스레드로부터 안전하게 참조 가능함을 나타낸다.
- 바꿔 말하면, 만일 &T (T의 참조자)가 Send인 경우 (참조자가 다른 스레드로 안전하게 보내질 수 있는 경우) T는 Sync를 수행한다.

참고

메모리-참조소유권과 Lifetime 메모리-참조스마트 포인터 활용 스레드Condvar 스레드멀티 스레드 웹 서버 만들기 예외처리Anyhow rustString