C++ Promise, package_task

promise - future

promise, future은 thread 간 단방향 전달 도구이다. 쉽게 말해 promise는 "값을 약속" 하고, future은 "값을 받는"역할을 수행한다.

#include <future>
#include <iostream>
#include <string>
#include <thread>
#include <chrono>
using namespace std;

void worker(promise<string>* p) {
    // 시간 소비
    this_thread::sleep_for(chrono::seconds(2));
    // 약속을 이행
    p->set_value("some data");
}

void example_function() {
    // 1초 대기
    this_thread::sleep_for(chrono::seconds(1));
    cout << "This is an example function." << endl;
}

int main() {
    promise<string> p;
    future<string> data = p.get_future();
    thread t(worker, &p);
    
    example_function();
	
    // 미래에 약속된 데이터를 받을 때 까지 기다린다.
    data.wait();
    
    // wait 이 리턴했다는 뜻이 future 에 데이터가 준비되었다는 의미.
	// 참고로 wait 없이 그냥 get 해도 wait 한 것과 같다.
    cout << "데이터 수신: " << data.get() << endl;

    t.join();

• data.wait()은 worker의 set_value를 기다리게 된다. -> promise에 값을 전달하기 때문에.
• 참고로 thread를 할당을 안해주면 data.get()을 했을때 프로그램이 블로킹 된다.
• 여기서 만약 이미 set_value()된 promise에 또다시 set_value()를 하면 오류가 발생한다.
• 마찬가지로 get또한 한번만 가능하고 다시 get()을 한다면 에러가 발생한다.
• 그 이유는 move semantics 때문에 그렇다. (소유권 이전)

std::condition_variable cv;
std::mutex m;
bool done = false;
std::string info;

void worker() {
  {
    std::lock_guard<std::mutex> lk(m);
    info = "some data";  // 위의 p->set_value("some data") 에 대응
    done = true;
  }
  cv.notify_all();
}

int main() {
  std::thread t(worker);

  std::unique_lock<std::mutex> lk(m);
  cv.wait(lk, [] { return done; });  // 위의 data.wait() 이라 보면 된다.
  lk.unlock();

  std::cout << "받은 데이터 : " << info << std::endl;

  t.join();
}

• 참고로 이전에 배웠던 condition_variable을 통해서 promise-future 패턴을 구현할 수 있다. 근데 봐서 알겠지만 p-f패턴이 훨씬 편하다..!

void worker(std::promise<string>* p) {
  try {
    throw std::runtime_error("Some Error!");
  } catch (...) {
    // set_exception 에는 exception_ptr 를 전달해야 한다.
    p->set_exception(std::current_exception());
  }
}
int main() {
  std::promise<string> p;

  // 미래에 string 데이터를 돌려 주겠다는 약속.
  std::future<string> data = p.get_future();

  std::thread t(worker, &p);

  // 미래에 약속된 데이터를 받을 때 까지 기다린다.
  data.wait();

  try {
    data.get();
  } catch (const std::exception& e) {
    std::cout << "예외 : " << e.what() << std::endl;
  }
  t.join();
}

• 게다가 promise 패턴을 이용하면 예외까지 전달할 수 있다.
• set_exception을 통해 예외를 받으면 try-catch문을 이용해서 실제 예외를 처리할 수 있게 된다.

wait_for

wait_for을 사용하면 그냥 wait과 달리 특정 시간동안만 대기하고 아니면 그냥 진행하게 할 수 있다.

void worker(promise<string>* p) {
    // 시간 소비
    this_thread::sleep_for(chrono::seconds(5));
    // 약속을 이행
    p->set_value("some data");
}

void example_function() {
    // 1초 대기
    this_thread::sleep_for(chrono::seconds(1));
    cout << "This is an example function." << endl;
}

int main() {
    promise<string> p;
    future<string> data = p.get_future();
    thread t(worker, &p);
    
    example_function();

    while(true) {
        future_status status = data.wait_for(chrono::seconds(1));
        if (status == future_status::timeout) {
            cout << "데이터 준비 중..." << endl;
        } else if (status == future_status::ready) {
            cout << "데이터 수신: " << data.get() << endl;
            break;
        }
    }
    t.join();
}

• future_status는 총 3가지의 형태를 가질 수 있다.
  • ready : future에 값이 설정 된 상태.
  • timeout : wait_for의 시간이 지났지만 값이 설정되지 않은 상태.
  • deferred : 아직 스레드가 시작되지 않았다는 뜻. 추후에 등장하는 개념으로 일반적인 promise, thread에서는 나오지 않는다.
• 여기서는 while을 통해 1초마다 한번씩 상태를 보고 값을 받았는지 확인한다.

shared_future

future의 경우에는 한번만 get할 수 있었는데, 그 이유는 future의 객체 내부가 이동하기 때문이다. 하지만 여러 스레드에서 future의 값을 여러번 get할 수 있게 해주는 것이 shared_future이다.

void runner(std::shared_future<void> start, int id) {
  start.get();
  printf("%d: 출발!\n", id);
}

int main() {
  std::promise<void> p;
  std::shared_future<void> start = p.get_future();

  thread t1(runner, start, 1);
  thread t2(runner, start, 2);
  thread t3(runner, start, 3);
  thread t4(runner, start, 4);

  // 참고로 cerr 는 std::cout 과는 다르게 버퍼를 사용하지 않기 때문에 터미널에
  // 바로 출력된다.
  std::cerr << "준비...";
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cerr << "땅!" << std::endl;

  p.set_value();

  t1.join();
  t2.join();
  t3.join();
  t4.join();
}

• 이렇게 하면 각 shared_future인 start를 각 스레드에서 여러번 get해도 정상적으로 동작한다.
• get()을 통해 각 future은 대기를 하다가 main에서 p.set_value()가 호출이 되면 각 스레드에서 동작한다.

packaged_task

특정 함수를 스레드에서 실핼할 수 있게 promise로 감싼 wrapper이다.

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

int some_task(int x) { return 10 + x; }

int main() {
  // int(int) : int 를 리턴하고 인자로 int 를 받는 함수. (std::function 참조)
  std::packaged_task<int(int)> task(some_task);

  std::future<int> start = task.get_future();

  std::thread t(std::move(task), 5);

  std::cout << "결과값 : " << start.get() << std::endl;
  t.join();
}

• packaged_task는 비동기적으로 수행할 함수 자체를 인자로 받는다. (감싼다.)
• 이렇게 감싸게 되면 promise가 자동으로 생겨 get_future()을 할 수 있다.
• packaged_task는 복사가 불가능하기 때문에 move를 통해서 소유권을 줘야 한다. (promise도 복사 불가능)

std::async

Promise나 packaged_task를 사용하는 방법은 우리가 스레드를 별도로 만들어 줘야 했지만 async를 사용하면 자동으로 스레드를 만들어 비동기적으로 실행하고 결과를 future에 전달하게 된다.

int heavy_calc(int x) {
    std::this_thread::sleep_for(2s);  // 무거운 계산 흉내
    return x * x;
}

• 만약 이런 함수가 존재한다면 promise와 packaged_task를 이용하면 어떻게 할 수 있을까?

promise<int> p;
future<int> f = p.get_future();

thread t([&p] { p.set_value(heavy_calc(5)); });

int result = f.get();  // 25
t.join();

• 아마 promise를 사용하면 이렇게 할 수 있을 것이다.
• 람다 캡쳐를 이용해 함수의 결과를 set_value해준다.

packaged_task<int(int)> task(heavy_calc);
future<int> f = task.get_future();

thread t(std::move(task), 5);

int result = f.get();  // 25
t.join();

• packaged_task로 만들게 된다면 위와 같이 작성할 수 있을 것이다.

auto f = std::async(std::launch::async, heavy_calc, 5);
int result = f.get();  // 25 (자동 정리!)

• 이제 async로 만드는것의 장점이 나오게 된다.
• async 함수는 첫번째 인자로 어떠한 형태로 함수를 실행할지 받는다.
  • std::launch::async : 새 스레드 만들어서 실행.
  • std::launch::deferred : 지연 실행 -> future의 get함수가 호출되면 그 때 실행. (새 스레드 x)
• 두번째 인자로는 호출할 함수가 나오게 되고.
• 그 후에는 그 함수에 들어갈 인자들을 넣는다.
  

void do_work(int x) {
  std::this_thread::sleep_for(std::chrono::seconds(x));
  std::cout << "작업 끝남\n";
}

void do_work_parallel() {
  auto f1 = std::async(std::launch::deferred, []() { do_work(3); });
  auto f2 = std::async(std::launch::deferred,[]() { do_work(3); });
  do_work(0);

  f1.get();
  f2.get();
}

int main() { do_work_parallel(); }

• 이해를 위해 이 코드를 한번 봐보자. async를 호출했는데 launch를 deferred로 했다. 이렇게 하면 새로운 스레드를 만드는게 아니라 일단 저장을 해두는 것이다. (대기)
• 추후에 wait이나 get을 만나면 그 때 기존 스레드를 이용해 작업을 수행하게 된다.
• 즉 현재 코드는 do_work(0)이 먼저 실행이 되고 그 다음 f1.get()을 하기 때문에 do_work(3)을 실행하고 끝나면 f2.get()을 만나 f2의 do_work(3)를 수행하게 된다.

void do_work(int x) {
  std::this_thread::sleep_for(std::chrono::seconds(x));
  std::cout << "작업 끝남\n";
}

void do_work_parallel() {
  auto f1 = std::async(std::launch::async, []() { do_work(1); });
  auto f2 = std::async(std::launch::async,[]() { do_work(1); });
  do_work(3);

  f1.wait();
  f2.wait();
}

int main() { do_work_parallel(); }

• 만약에 이렇게 launch를 async로 하게 된다면, 새로운 스레드를 만들어 do_work(1)을 가진 f1과 f2를 비동기로 처리하면서 do_work(3)를 처리하고 있을 것이다.

⚠️ std::async를 사용할때 launch는 생략하지 않는 것이 좋다. 생략하게 되면 구현체마다 다른 기본값으로 설정이 되어서 예측하기 어려워진다.

참고자료

씹어먹는 C++ - 15.4 future, promise, packaged_task