Chapter 4. Synchronizing Concurrent Operations. (2) - future

1. Future vs Shared Future

뮤택스(Mutualy Exclusive)는 Race Condition(경쟁 상태)을 방지하기 위해서 공유 자원에 접근을 하나로 제한하는 것이었다.
Conditional Variable(조건 변수)는 스레드 간의 순서를 맞출 때 사용했다. (Synchronizing)
future 멀티 스레드에서 결과 값을 받아 올 때 쓰는 class이다. future 자체로는 Race Condition을 막을 수 없다. 그래서 Read only 함수만을 이용하면 더 편리한 멀티 스레딩이 된다. 만약 여러 개의 함수가 Read only로 결과를 내서 합친다면 shared_future를 사용할 수 있다. future는 movable ^[각주:1] 하지만 shared_future는 movable과 동시에 copyable 하다. future 객체의 get()은 한 번만 실행이 가능하다. shared_future는 여러 번 get()할 수 있다.

아래 async, promise, packaged_task는 future와 함께 일회성 이벤트의 결괏값을 받아올 때 사용한다. 세 가지 모두 get, ready

2. future + async

간단한 멀티스레딩을 할 때엔 async를 쓰는 것이 좋다. async는 future를 return하고 해당 future에 wait()과 get() 멤버 변수를 사용해서 쓰레딩이 완료된 값을 받는다. async는 즉시 thread를 생성해서 해당 값을 구할 수도 있고, 나중에 get()이나 wait()이 호출될 때 해당 값을 구할 수 도 있다. std::launch 파리미터를 통해서 즉시 thread를 만들어서 실행할지 wait()이나 get()이 실행될 때 실행할지 수동으로 결정할 수 있다.

auto f6=std::async(std::launch::async,Y(),1.2); 
auto f7=std::async(std::launch::deferred,baz,std::ref(x));
auto f8=std::async(
   std::launch::deferred | std::launch::async,
   baz,std::ref(x));
auto f9=std::async(baz,std::ref(x));
f7.wait();

아래는 async의 사용 예시이다.

#include <string>
#include <future>
struct X
{
    void foo(int,std::string const&);
    std::string bar(std::string const&);
};
X x;
auto f1=std::async(&X::foo,&x,42,"hello");
auto f2=std::async(&X::bar,x,"goodbye");
struct Y
{
    double operator()(double);
};
Y y;
auto f3=std::async(Y(),3.141);
auto f4=std::async(std::ref(y),2.718);
X baz(X&);
std::async(baz,std::ref(x));
class move_only
{
public:
    move_only();
    move_only(move_only&&)
    move_only(move_only const&) = delete;
    move_only& operator=(move_only&&);
    move_only& operator=(move_only const&) = delete;
    void operator()();
};
auto f5=std::async(move_only());

즉 async는 main thread를 차단(block) 하지 않고 비동기 처리하는 것에 초점이 맞춰져 있었다. 그러나 여러 개의 스레드를 써서 thread 간의 값을 받아오려면 Promise가 필요하다.

3. future + promise

promise는 thread 간 통신을 위해 만들어졌다. 가령 네트워크 프로그래밍에서 몇 개의 thread는 데이터를 받아오고 해당 데이터를 다른 스레드가 쓸 때, promise를 사용할 수 있다. async보다 더 low level의 기능을 제공한다. set_exception(), shared_future와의 연계 등등..

1) promise 객체제 get_future()를 통해 future를 얻는다.
2) 실행하고 싶은 함수에 promise를 인자로 넘긴다.
3) 함수 안에서 set_value 한다.
4) main thread에서 future에 get 혹은 wait을 한다.

#include <future>

void process_connections(connection_set& connections) {
    while (!done(connections)) {
        for (connection_iterator connection = connections.begin(), end = connections.end();
             connection != end;
             ++connection) {
            if (connection->has_incoming_data()) {
                data_packet data = connection->incoming();
                std::promise<payload_type>& p = connection->get_promise(data.id);
                p.set_value(data.payload);
            }
            if (connection->has_outgoing_data()) {
                outgoing_packet data = connection->top_of_outgoing_queue();
                connection->send(data.payload);
                data.promise.set_value(true);
            }
        }
    }
}

4. future + packaged_task

package_task는 thread pool, 또는 Function Reactive Programming(Async Dataflow programming)를 만들 때 쓰인다. 함수를 template specialization을 통해 return 값을 future로 만든다. future로 된 return 값들이 해당 데스크들이 끝날 때까지 wait 할 것이기 때문에, lambda로 함수들을 묶은 것과는 차이가 생긴다.
아래 코드는 Function Reactive Programming으로 만든 GUI이다. task는 queue에 차곡차곡 쌓인다. 그리고 queue에 들어간 task들이 차례대로 실행된다.

#include <utility>
#include <mutex>
#include <deque>
#include <packaged_task>

std::mutex m;
std::deque<std::packaged_task<void()>> tasks;
bool gui_shutdown_message_received();
void get_and_process_gui_message();

void gui_thread() {
    while (!gui_shutdown_message_received()) {
        get_and_process_gui_message();
        std::packaged_task<void()> task;
        {
            std::lock_guard<std::mutex> lk(m);
            if (tasks.empty()) continue;
            task = std::move(tasks.front());
            tasks.pop_front();
        }
        task();
    }
}
std::thread gui_bg_thread(gui_thread);

template<typename Func>
std::future<void> post_task_for_gui_thread(Func f) {
    std::packaged_task<void()> task(f);
    std::future<void> res = task.get_future();
    std::lock_guard<std::mutex> lk(m);
    tasks.push_back(std::move(task));
    return res;
}

5. 시간 관련된 것들

wait에는 _for와 _until 2가지 suffix가 있다. for는 해당 시간 동안 기다리는 것이며, until은 해당 시간마다 알람 스케줄링 돼서 실행되는 것이다. 해당 wait에 결과는 future_status에 담긴다. future status는 ready, defer(아직 다른 스레드가 안 끝남), timeout(시간 종료)가 담긴다. 

ps.

• Cuncurrency TS (Technical Specification #include <experimental/future>
• multi thread에서 exception이 일어나면 다른 thead에게 알릴 수 있다.

ref.

• C++ concurrency in action
• https://en.cppreference.com/w/cpp/thread/future_status

1. r value로 만들고 ownership을 옮길 수 있다. ownership의 정의는 (Synchronizing Concurrent Operations. (1) 참조) [본문으로]