비동기 프로그래밍

앞의 글들에서는 스레드를 직접 만들고, 결과를 공유 변수에 쓰고, 뮤텍스와 조건 변수로 준비되었는지를 알렸습니다. 그런데 하고 싶은 일이 "이 계산을 다른 곳에서 해 두고, 나중에 결과만 받기"라면 이 모든 것을 직접 짜는 것은 번거롭고 실수하기 쉽습니다. <future> 헤더는 작업을 실행하는 쪽과 결과를 받는 쪽을 이어 주는 도구를 제공합니다.

이 글은 작업을 비동기로 실행하고 결과를 std::future로 받는 std::async, 결과를 직접 정해 넘기는 std::promise, 호출 가능한 것과 결과를 묶어 원하는 곳에서 실행하는 std::packaged_task를 다루고, 세 도구 가운데 무엇을 고를지 정리합니다.

결과를 나중에 받는 방법

비동기(asynchronous) 실행은 작업을 시작해 두고, 그 작업이 끝나기를 기다리지 않은 채 다른 일을 하다가 필요할 때 결과를 받는 방식입니다. 세 도구는 모두 결과를 담는 공유 상태(shared state)를 만들고, 결과를 받는 쪽은 그 상태에 연결된 std::future<T>를 들고 있습니다.

future의 get()은 결과가 준비될 때까지 기다렸다가 값을 돌려주며, 작업이 예외를 던졌다면 그 예외를 다시 던집니다. 결과는 한 번만 꺼낼 수 있어서, get()을 부른 뒤에는 valid()가 false가 되고 다시 get()을 부를 수 없습니다. 여러 곳에서 같은 결과를 읽어야 한다면 fut.share()로 얻는 std::shared_future를 씁니다.

std::async와 std::future

std::async는 함수와 인자를 받아 작업을 실행하고, 그 결과를 받을 future를 돌려줍니다. 스레드를 만들고 결과를 전달하는 일을 대신 해 주므로 가장 간단한 방법입니다.

#include <future>
#include <iostream>
#include <stdexcept>

long long calculateSum(int from, int to) {
    long long sum = 0;
    for (int i = from; i <= to; ++i) {
        sum += i;
    }
    return sum;
}

int main() {
    auto fut = std::async(std::launch::async, calculateSum, 1, 500);
    std::cout << std::boolalpha << fut.valid() << '\n';
    long long result = fut.get();
    std::cout << result << ' ' << fut.valid() << '\n';

    auto failing = std::async(std::launch::async, []() -> int {
        throw std::runtime_error("bad input");
    });
    try {
        failing.get();
    } catch (const std::exception& e) {
        std::cout << "caught " << e.what() << '\n';
    }
}

true
125250 false
caught bad input

std::launch::async를 주면 작업이 새 스레드에서 바로 시작되고, main은 get()에서 결과를 기다립니다. 1부터 500까지의 합은 500 × 501 / 2 = 125250이며, get()으로 꺼낸 뒤에는 valid()가 false가 됩니다. 작업이 던진 예외는 future에 담겼다가 get()에서 다시 던져지므로, 결과를 받는 쪽에서 try-catch 구문으로 처리할 수 있습니다.

실행 정책을 주지 않으면 std::launch::async | std::launch::deferred가 쓰여, 새 스레드에서 실행할지 get()을 부를 때 그 자리에서 실행할지를 구현이 고릅니다. 구현이 지연 실행(deferred)을 고르면 get()이나 wait()을 부르기 전까지 작업이 시작되지 않으므로, 동시에 실행되어야 한다면 std::launch::async를 직접 적습니다. 또 새 스레드에서 실행하기로 한 작업이라면 std::async가 돌려준 future는 소멸할 때 작업이 끝나기를 기다리므로, 돌려받은 값을 버리면 그 줄에서 작업이 끝날 때까지 멈춥니다.

std::promise

std::promise<T>는 결과를 직접 정해 넣는 쪽입니다. get_future()로 같은 공유 상태에 연결된 future를 얻고, 생산자가 set_value()로 값을, set_exception()으로 예외를 넣으면 그 상태가 준비됩니다.

#include <future>
#include <iostream>
#include <thread>
#include <utility>

int main() {
    std::promise<int> prom;
    std::future<int> fut = prom.get_future();

    std::thread producer([p = std::move(prom)]() mutable {
        p.set_value(42);
    });

    std::cout << fut.get() << '\n';
    producer.join();
}

42

아래 그림처럼 main은 fut.get()에서 결과를 기다리고, 생산자 스레드가 set_value(42)로 값을 넣으면 공유 상태가 준비되어 get()이 42를 돌려줍니다. 값을 꺼낸 뒤에는 future와 상태의 연결이 끝납니다. promise는 복사할 수 없으므로 람다 표현식에서 본 초기화 캡처로 생산자 스레드에 옮겨 주었습니다.

prom.set_value(data);
// 다른 스레드
received_data = fut.get();

prom.set_value(data);
// 다른 스레드
received_data = fut.get();

promise가 값을 넣지 않은 채 사라지면, 기다리던 future의 get()은 오류 코드가 std::future_errc::broken_promise인 std::future_error 예외를 던집니다. 그래서 생산자는 작업이 성공하면 set_value()를, 실패하면 set_exception(std::current_exception())처럼 예외를 넣어, 결과를 받는 쪽이 어느 경우든 기다림에서 빠져나오게 해야 합니다. 값이나 예외는 한 번만 넣을 수 있으며, 두 번 넣으면 std::future_error가 던져집니다.

std::packaged_task

std::packaged_task는 호출 가능한 것과 결과를 담을 공유 상태를 하나로 묶습니다. 작업을 언제 어디서 호출할지는 직접 정하고, 호출하면 반환값이나 예외가 공유 상태에 들어갑니다.

#include <future>
#include <iostream>
#include <thread>
#include <utility>

int main() {
    std::packaged_task<int(int, int)> task([](int a, int b) { return a * b; });
    std::future<int> fut = task.get_future();

    std::thread worker(std::move(task), 6, 7);
    std::cout << fut.get() << '\n';
    worker.join();
}

42

task를 스레드로 옮겨 6과 7로 호출하면 곱 42가 공유 상태에 들어가고, main은 fut.get()으로 받습니다. 작업을 바로 실행하지 않고 큐에 넣어 두었다가 정해진 작업 스레드가 꺼내 실행하는 스레드 풀(thread pool)을 만들 때, 이렇게 호출할 곳을 고를 수 있는 packaged_task가 쓰입니다.

도구 고르기

세 도구는 모두 future로 결과를 받지만, 누가 작업을 시작하고 누가 결과를 정하는지가 다릅니다.

std::async(calculate_sum,
  1, 500);

auto fut = prom.get_future();
prom.set_value(data);

std::packaged_task<int(int)>
  task(process_heavy_task);

result_future.get();
worker_thread.join();

결과가 준비되었는지 기다리지 않고 확인하려면 fut.wait_for(std::chrono::seconds(0))의 결과가 std::future_status::ready인지 봅니다. 여러 작업을 동시에 시작하고 각자의 future를 모았다가 차례로 get()하면, 전체 시간이 가장 오래 걸리는 작업 정도로 줄어듭니다. 다만 작업마다 새 스레드를 만드는 비용이 있으므로, 아주 짧은 작업을 수없이 나누는 것은 오히려 느릴 수 있습니다.

더 깊이: std::future의 한계와 C++26의 std::execution

std::future에는 결과가 준비되면 다음 작업을 시작하라고 등록하는 멤버 함수가 없습니다. 그래서 여러 비동기 작업을 차례로 잇거나 하나가 끝나는 대로 다른 작업을 시작하려면, 어느 스레드가 get()이나 wait()에서 기다려야 합니다. 또 std::async에 줄 수 있는 표준 실행 정책은 async와 deferred 두 가지뿐이어서, 작업을 보낼 스레드 풀 같은 실행 자원을 고를 수도 없습니다.

C++26은 이 문제를 다루는 실행 제어 라이브러리 std::execution을 더했습니다(P2300R10). 비동기 작업을 설명하는 송신자(sender), 그 결과를 받는 일반화된 콜백인 수신자(receiver), 작업을 실행 자원에 보내는 스케줄러(scheduler)를 조합해, then 같은 단계를 |로 이어 붙인 작업 그래프를 만들고 원하는 실행 자원에서 시작합니다. 결과는 값, 오류, 멈춤의 세 갈래로 전달되므로, 취소도 같은 틀 안에서 다룹니다.

다만 cppreference의 C++26 지원표에는 아직 이 라이브러리를 제공하는 GCC, Clang, MSVC의 표준 라이브러리가 없습니다. cppreference의 실행 제어 라이브러리 문서도 예제를 참조 구현 가운데 하나인 NVIDIA의 stdexec으로 실행하고 있으므로, 이 교재의 C++20 환경에서는 본문의 future 도구를 쓰고 새 라이브러리는 쓰는 구현의 지원 상태를 확인한 뒤 검토합니다.

정리

• 비동기 도구는 결과를 담는 공유 상태를 만들고, 결과를 받는 쪽은 그 상태에 연결된 std::future를 듭니다.
• future::get()은 결과를 기다려 한 번만 꺼내며, 작업이 던진 예외를 다시 던지고, 꺼낸 뒤에는 valid()가 false가 됩니다.
• std::async는 작업의 실행과 결과 전달을 맡으며, 동시에 실행해야 한다면 std::launch::async를 적습니다.
• std::promise는 생산자가 값이나 예외를 직접 넣으며, 넣지 않고 사라지면 broken_promise 오류가 됩니다.
• std::packaged_task는 호출 가능한 것과 결과를 묶어, 호출할 곳과 때를 직접 정하게 합니다.
• 여러 곳에서 같은 결과를 읽을 때는 std::shared_future를 씁니다.

연습 문제

1. 결과를 두 번 꺼내기

async_sum.cpp에서 fut.get()을 한 번 더 부르면 어떻게 되나요? 같은 결과를 두 함수에서 읽어야 한다면 어떻게 하나요?

2. 실행 정책을 빼면

async_sum.cpp의 첫 std::async에서 std::launch::async를 빼고, get()을 부르지 않은 채 main이 끝나면 calculateSum은 실행되나요?

3. 값을 넣지 않은 promise

promise_demo.cpp에서 생산자 스레드가 p.set_value(42);를 하지 않고 끝나면 main의 fut.get()은 어떻게 되나요?

이것으로 15장의 멀티스레딩 기초를 마칩니다. 다음 장의 성능 프로파일링 도구 소개에서는 프로그램이 어디에서 시간을 쓰는지 측정하는 프로파일링으로, 성능을 개선하는 방법의 첫 단계를 다룹니다.