C++에서 멀티 스레드 프로그래밍을 하다 보면 한 스레드가 특정 이벤트나, 특정 조건이 true가 될 때까지 기다리길 원하는 상황이 흔하게 발생한다.
그런 상황을 어떻게 구현하는지에 대해 알아본다.
std::condition_variable
표준 C++ 라이브러리는 두 개의 condition_variable의 구현을 제공한다.
std::condition_variable과 std::condition_variable_any이다.
전자는 무조건 std::unique_lock<std::mutex>와 함께 사용해야 한다.
후자는 어떤 락 타입과도 함께 동작할 수 있다.
std::mutex뿐만 아니라 사용자 정의 뮤텍스나 다른 락 타입(std::shared_lock, std::scoped_lock 등)을 사용할 수 있다. 더 유연하지만, 속도가 약간 느릴 수 있다.
대부분 추가적인 유연성이 필요하지 않기 때문에 std::condition_variable을 사용한다.
예제 코드
#include <iostream>
#include <mutex>
#include <vector>
#include <thread>
#include <condition_variable>
std::mutex mut;
std::queue<data_chunk> data_queue; //두 스레드 사이에 데이터를 전달하기 위한 큐
std::condition_variable data_cond;
//데이터 준비 스레드의 엔트리 함수
void data_preparation_thread()
{
while(more_data_to_prepare())
{
data_chunk const data = prepare_data();
//데이터가 준비 되면 데이터 준비 스레드는 lock_gurad를 사용해서
//플래그를 보호하는 뮤텍스를 잠그고 데이터를 큐에 push한다.
std::lock_gurad<std::mutex> lk(mut);
data_queue.push(data);
//푸시 후 이 조건 변수에 대해 대기하고 있는 다른 스레드에게 이벤트를 알린다.
data_cond.notify_one();
}
}
//데이터 처리 스레드의 엔트리 함수
void data_processing_thread()
{
while(true)
{
//std::unique_lock을 써야 하는 이유는
//중간에 임의로 unlock을 해야 하기 때문이다.
//일단 뮤텍스를 잠그고 wait함수에서 람다로 전달된 조건을 검사한다.
std::unique_lock<std::mutex> lk(mut);
//조건을 검사해서 true가 반환되면 곧바로 리턴한다.
//false라면 unlock을 하고 다시 알림이 올 때까지 스레드는 대기한다.
//여기서 unlock을 하는 이유는 그래야 다른 스레드가
//다시 그 공유 자원에 대한 lock을 취득할 수 있기 때문이다.
//조건 부분은 람다가 아니어도 호출 가능한 객체면 다 가능하다.
//조건 확인은 wait 함수 내부에서 여러 번 동작할 수 있고,
//항상 뮤텍스를 잠근 상태에서 확인한다.
data_cond.wait(lk, []{return !data_queue.empty();});
data_chunk data = data_queue.front();
data_queue.pop();
lk.unlock();
process(data);
if(is_last_chunk(data))
break;
}
}std::condition_varialbe에서 다른 스레드를 깨우는 방법은 두 가지가 있다.
대기 중인 스레드 하나를 깨우는 방법은 위의 예제에서처럼 notify_one()함수를 호출하는 것이다.
대기 중인 모든 스레드에게 알림을 주는 것은 notify_all()함수를 호출하면 된다.
대기 스레드가 단 한 번만 대기할 경우, 즉 조건이 참일 때 이 조건 변수를 다시는 기다리지 않을 경우, 조건 변수는 동기화 메커니즘의 최상의 선택이 아닐 수 있다.
이런 시나리오에서는 future가 더 적합할 수 있다.
future를 사용하면 이제 task 기반의 비동기 프로그래밍을 추상화해서 쉽게 처리할 수 있게 된다.
여기서 중요한 것은 task에서 나온 데이터가 caller로 안전하게 전달되고 시그널을 잘 보내는지이다.
이는 future와 promise라는 도구를 활용해 C++에서 구현할 수 있다.
Future를 사용한 일회성 이벤트 대기
C++ 라이브러리는 일회성 이벤트를 future라는 매커니즘으로 모델링한다.
스레드가 특정 일회성 이벤트를 기다려야 하는 경우, 어떻게든 이 이벤트를 나타내는 future를 얻는다. 그런 다음 future를 폴링하여 이벤트가 발생했는지 확인할 수 있다. 다른 작업을 수행하는 도중에서 future를 확인할 수 있다.
future에는 연관된 데이터가 있을 수도 있고 없을 수도 있다.
이벤트가 발생하고 future가 준비되면 future를 재설정할 수 없다.
future는 비동기 작업에 대한 결과를 받을 수 있는 객체라고 할 수 있다.
항공에서 비행기를 타기위해 기다리는 승객 스레드가 여러 명 있다고 해보자.
비행기가 출발 준비가 될 때까지 승객들은 기다린다. 어떤 승객은 계속 기다리고 있을 수 있고, 다른 승객은 준비 알림이 올 때까지 카페에서 뭘 먹고 있을 수도 있고, 면세점에서 물건을 사고 있을 수도 있을 것이다. 이를 코드로 나타내면 다음과 같다.
void wait_for_flight1(flight_number flight)
{
//future를 사용해서 flight에 알림이 올 때까지 기다린다.
std::shared_future<boarding_information> boarding_info =
get_boarding_info(flight);
//get()함수는 future가 아직 알림을 받지 못 했으면 스레드를 대기 시킨다.
board_flight(boarding_info.get());
}
void wait_for_flight2(flight_number flight)
{
std::shared_future<boarding_information> boarding_info =
get_boarding_info(flight);
//바로 get()호출하지 않고 다른 일을 하면서 주기적으로 준비가 되었는지
//체크하다가 준비가 되면 get()을 한다.
while(!boarding_info.is_ready())
{
eat_in_cafe();
buy_duty_free_goods();
}
board_flight(boarding_info.get());
}
이 예제에서는 여러 승객 스레드가 하나의 이벤트를 기다리고 모든 스레드는 동시에 알림을 받아야 하고 모두 동일한 탑승 정보기 필요하기 때문에std::shared_future를 사용했다.
여러 스레드가 아니라 하나의 스레드만 대기를 한다면std::unique_future를 사용하는 것이 적합하다. std::shared_future에 비해 오버헤드가 낮을 가능성이 있기 때문이다.
또한 std::shared_future은 future와 관련된 데이터를 복사만 할 수 있는 반면에 std::unique_future는 데이터를 이동하여 future에서 대기 코드로 소유권을 이전할 수 있다. 이는 이동이 복사보다 비용이 싼 경우 유용하다.
지금까지 대기 스레드 관점에서 본 future들이었다.
이벤트를 트리거 하는 스레드는 어떻게 모델링해야 할까?
future를 준비하려면 어떻게 해야 할까?
연관 데이터는 어떻게 전달할까?
C++ 표준은 이런 질문에 대한 답을 두 가지 함수 템플릿인 std::packaged_task<>와 std::promise<>의 형태로 제공한다.
std::packaged_task<>
이 함수 템플릿은 future를 이 함수 호출 결과에 연결한다.
함수가 완료되면 future는 준비가 되고, 연관된 데이터는 함수의 반환값이 된다.
이는 자체 포함 작업 집합으로 세분화할 수 있는 전체 작업에 이상적이다.
이러한 작업은 함수로 작성되고, std::packaged_task를 사용하여 각 작업이 future와 연관되도록 패키징된 다음 별도의 스레드에서 실행될 수 있다.
그런 다음 구동 함수는 이러한 하위 작업의 결과를 처리하기 전에 future를 기다릴 수 있다.
쉽게 말해 하나의 큰 작업을 쪼개서 여러 작업으로 만들고 그 여러 작업이 끝나면 그걸 취합하는 시나리오에서 유용하다는 것이다.
말이 다소 어려워서 GPT에게 설명을 요청했다.
std::packaged_task는 특정 작업(함수)을 포장(package) 해서, 나중에 실행하거나 다른 스레드로 전달한 뒤 결과를 가져올 수 있도록 만드는 도구입니다. 쉽게 말해, 작업을 "포장 상자" 안에 넣어두고 필요할 때 실행할 수 있게 해주는 역할을 합니다.
예제 코드
#include <iostream>
#include <thread>
#include <future>
int calculateSum(int a, int b) {
return a + b;
}
int main() {
// 1. 작업을 포장 (packaged_task)
std::packaged_task<int(int, int)> task(calculateSum);
// 2. 미래 결과를 가져올 수 있는 열쇠 (future)
std::future<int> result = task.get_future();
// 3. 포장된 작업을 다른 스레드에서 실행
std::thread t(std::move(task), 10, 20); // calculateSum(10, 20) 실행
// 4. 결과 가져오기
std::cout << "Result: " << result.get() << '\n';
t.join();
return 0;
}
std::promise
std::promise는 C++에서 값을 약속(promise) 하고, 나중에 그 값을 전달(deliver) 할 수 있도록 만들어주는 도구이다.
쉽게 말해, "나중에 결과를 줄게!" 라고 약속하고, 결과가 준비되면 전달하는 역할을 한다.
비유로 설명
📩 편지와 우체통 비유
- 우체통 설치:
std::promise는 값을 전달할 수 있는 우체통을 설치한다.
이 우체통에는 편지를 꺼내 볼 수 있는 열쇠(std::future)가 함께 제공된다. - 값을 약속:
편지를 우체통에 넣기로 약속(set_value)한다.
즉, 값을 나중에 줄 것을 약속한다. - 결과 전달:
우체통에 편지를 넣는다. - 결과 확인:
다른 곳에서 열쇠(std::future)를 사용해 우체통에서 편지를 꺼내 내용을 확인한다.
#include <iostream>
#include <thread>
#include <future>
void deliverResult(std::promise<int> promise) {
// 3초 후 값을 약속
std::this_thread::sleep_for(std::chrono::seconds(3));
promise.set_value(42); // 값을 전달
}
int main() {
// 1. 우체통 설치 (promise)
std::promise<int> promise;
// 2. 열쇠(future) 받기
std::future<int> result = promise.get_future();
// 3. 값을 약속 (다른 스레드에서)
std::thread t(deliverResult, std::move(promise));
// 4. 결과 확인
std::cout << "Waiting for result...\n";
std::cout << "Result: " << result.get() << '\n';
t.join();
return 0;
}
이런 추상화된 task 기반의 비동기 함수 호출 메커니즘은 내부적으로 결국 조건 변수와 뮤텍스, 힙 할당, 해제가 이뤄지기 때문에 성능이 중요한 코드에서는 잘 사용되지 않는다고 한다.
