[Concurrency] Future✨에 대해 더 알아보자.

Future the #

2023.10.25 - [언어/C++] - [thread] Promise /Future 모델 필요성 및 사용 방법

2023.10.31 - [언어/C++] - [thread] Promise /Future 모델 적용

이전 글에서 Promise / Future 모델 사용법과 기존코드에 Poromise / Future를 적용시키는 내용을 소개했었는데 이번 글에서는 Future의 세부 기능을 소개하고자 한다.

 

Promise / Future 모델을 써도 연산이 종료되지 않아 무한대기하는 상황에서는 어떻게 해야 할까?🤔

 

이를 위해 Future 클래스에서 이런 문제를 해결하기 위한 기능을 제공하고 있다. 우선 멤버 함수를 확인해 보자.

 

Future의 멤버 함수

---

 

share	shared_future 얻기 / transfers the shared state from *this to a shared_future and returns it (public member function)
Getting the result
get	결과 꺼내기 /returns the result (public member function)
State
valid	유효성 확인 / checks if the future has a shared state (public member function)
wait	결과값이 준비 될 때까지 대기 / waits for the result to become available (public member function)
wait_for	주어진 시간만큼 대기 / waits for the result, returns if it is not available for the specified timeout duration (public member function)
wait_until	주어진 시간까지 대기 /waits for the result, returns if it is not available until specified time point has been reached (public member function)

 

 

wait_for

---

get 함수와 마찬가지로 결과를 대기하지만, wait_for 함수는 주어진 시간까지만 기다리기에 기대한 응답시간을 초과한 것에 대한 예외처리를 아래와 같이 할 수가 있다.

 

예제 1
아래 예제는 2개의 값을 받아 합산한 결과를 리턴하는 add() 함수를 새로운 스레드로 실행시키는데 연산 결과를 5초 뒤에 응답하게 하고, future에 wait_for 함수를 3초로 설정하여 timeout 동작을 하게 하는 프로그램 코드이다.

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

void add(std::promise<int>&& p, int a, int b)
{
    std::this_thread::sleep_for(5s);
    p.set_value(a + b);
}

int main()
{
    std::promise<int> pm;
    std::future<int> ft = pm.get_future();

    std::thread t(add, std::move(pm), 10, 20);

    std::future_status ret = ft.wait_for(3s);
	std::cout << "wait_for() 실행됨\n";

    if (ret == std::future_status::ready)
    {
        std::cout << "set_value() 되어서 결과 있음\n";

        int n = ft.get();
        std::cout << "데이타가 있으므로 대기 안함\n";        
    }
    else if (ret == std::future_status::timeout)
    {
        std::cout << "대기시간초과\n";
    }
    else if (ret == std::future_status::deferred)
    {
        std::cout << "스레드 실행안됨\n";
         // => std::async() 사용시에만 나올수 있는값
         // => std::thread 로 스레드 생성시는 이값이 나오지 않음
    }

    t.join();
}

 

결과

 

실행결과를 보면 get()처럼 wait_for() 함수에서 대기를 하고 설정한 시간을 넘어서면(timeout 되면) 결과 값과 상관없이 대기를 종료하고 ret 값을 받아온다. wait_for 응답 값은 3가지가 있으며 정의는 다음과 같다.

Return value

Constant	Explanation
future_status::deferred	연산을 수행할 함수가 아직 시작 안됨, thread의 경우는 없고 async() 함수에서 발생할 수 있음 / The shared state contains a deferred function using lazy evaluation, so the result will be computed only when explicitly requested
future_status::ready	결과 값이 준비 됨 / The result is ready
future_status::timeout	시간 초과 / The timeout has expired

 

만약에 응답이 timeout 이전에 오는 경우에는 어떻게 동작할까?🤔

    std::this_thread::sleep_for(1s);// 5s -> 1s

add() 안의 코드를 1초로 변경하여 실행해 보자.

 

결과

 

timeout 이전에 연산결과가 전달되었기에 ret 값이 ready로 전달되고, 코드를 보면 값을 받기 위해 ft.get()을 하지만 이미 값이 준비가 되어있기에 대기를 하지 않는 것을 확인할 수 있다.

if (ret == std::future_status::ready)
{
    std::cout << "set_value() 되어서 결과 있음\n";
    int n = ft.get();
    std::cout << "데이타가 있으므로 대기 안함\n";
}

 

* wait, wait_until 그 wait_for과 사용방법이 같다.

valid

---

Promise / Future 모델 콘셉트에서 얘기했지만, promise와 future는 페어 콘셉트이기에 페어가 깨질 수 있는 아래 코드는 오류가 발생한다. valid를 사용하면 이런 문제를 예방할 수 있다.

 

get() 두 번 호출

    int ret = ft.get();
    int ret2 = ft.get(); // 예외

 

결과
future에서 이미 결과를 가져갔기에 다시 시도하면 예외 발생

terminate called after throwing an instance of 'std::future_error'
what(): std::future_error: No associated state

 

 

예제 2
아래 예제는 2개의 값을 받아 합산한 결과를 리턴하는 add() 함수를 새로운 스레드로 실행시키는데 future의 valid() 함수의 결과 값이 get() 전/후 어떻게 달라지는지, 확인할 수 있는 프로그램 코드이다.

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

void add(std::promise<int>&& p, int a, int b)
{
    std::this_thread::sleep_for(1s);

    p.set_value(a + b);
}

int main()
{
    std::promise<int> pm;

    std::future<int> ft = pm.get_future();

    std::thread t(add, std::move(pm), 10, 20);
        
    std::cout << "ft.valid() : " << ft.valid() << std::endl; // true

    int ret = ft.get();

    std::cout << "ft.valid() : " << ft.valid() << std::endl; // false

    t.join();
}

 

결과

ft.valid() : 1
ft.valid() : 0

 

이처럼 valid 함수를 용도를 활용하여 get() 함수가 두 번 호출되지 않도록 조건처리를 할 수 있다.

 

오류 발생 코드

---

그 외에도 아래와 같이 사용을 하면 오류가 발생한다. 

 

get_future() 두 번 호출

    std::future<int> ft = pm.get_future();
    std::future<int> ft2 = pm.get_future(); // 예외 발생(exception)

 

결과
promise에서 future는 한 번만 꺼낼 수 있기에 다시 시도하면 예외발생

terminate called after throwing an instance of 'std::future_error'
  what():  std::future_error: Future already retrieved

 

future 복사

    std::future<int> ft = pm.get_future();
    std::future<int> ft2 = ft; // error. 복사 생성자가 삭제

 

결과
future 복사 생성자가 삭제되어 컴파일 에러 발생

error: use of deleted function 'std::future<_Res>::future(const std::future<_Res>&) [with _Res = int]'
30 | std::future<int> ft2 = ft;

 

set_value() 두 번 호출

    p.set_value(a + b);
    p.set_value(a + b); // 예외 발생

 

결과
이미 결과 값을 전달했다고 예외가 발생된다.

terminate called after throwing an instance of 'std::future_error'
  what():  std::future_error: Promise already satisfied

 

한 개의 promise는 한 개의 스레드에만 전달해야 한다. (여러 개 스레드 공유 안됨) 여러 번 전달하려면 std::condition_variable 등의 도구로 직접 만들어야 한다.

 

C++ 답지 않게 너무 자유도가 없는 것 같기도 하지만 콘셉트를 이해하고 있다면 당연한 오류이기도 하다.

 

share

---

promise의 결과를 여러 스레드에 알려야 한다면 어떻게 해야 할까? 🤔

share는 future를 여러 개로 만들기 위한 함수이다.

 

예제 3
예제 2와 동일한 동작을 하면서, share() 함수를 이용해서 future를 shared_future로 만드는 방법과 shared_future 특징을 확인하는 프로그램 코드이다.

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

void add(std::promise<int>&& p, int a, int b)
{
    std::this_thread::sleep_for(1s);
    p.set_value(a + b);
}
int main()
{
    std::promise<int> pm;
    std::future<int> ft = pm.get_future();
    std::shared_future<int> sft1 = ft.share(); //이 순간 ft는 더이상 사용해선 안된다.
    //std::shared_future<int> sft2 = ft.share(); //가능은 하지만 sft2 사용 시 예외발생
    std::shared_future<int> sft3 = sft1;  // 복사 가능

    std::thread t(add, std::move(pm), 10, 20);

    // shared_future 를 꺼낸 경우는 future가 아닌 shared_future를 사용해야 한다.
    // int ret1 = ft.get();
    
    int ret1 = sft1.get();
    std::cout << "ret1 = " << ret1 << std::endl;

    // 사용 시 예외 발생
    //int ret2 = sft2.get();
    //std::cout << "ret2 = " << ret2 << std::endl;

    int ret3 = sft3.get();
    std::cout << "ret3 = " << ret3 << std::endl;

    t.join();
}

 

결과

ret1 = 30
ret3 = 30

 

shared_future는 복사가 가능하기에(복사생성자 정의) 여러 스레드에 shared_future 전달이 가능하며

std::shared_future<int> sft3 = sft1;  // 복사도 가능

 

shared_future 각각 결과를 대기(get) 할 수 있다.

    int ret1 = sft1.get();
    std::cout << "ret1 = " << ret1 << std::endl;
    
    int ret3 = sft3.get();
    std::cout << "ret3 = " << ret3 << std::endl;

 

*주의사항

std::shared_future<int> sft1 = ft.share(); //이 순간 ft는 더이상 사용해선 안된다.

future에서 share() 함수를 호출하면 더 이상 future를 사용해선 안된다.

왜냐하면 share()를 호출하면 shared_future로 return 이 되는데, 이때 future의 소유권을 return 해버리기에 이후 ft 사용 시 예외가 발생되게 설계되어 있다.(혼용 불가능)

    _NODISCARD shared_future<_Ty> share() noexcept {
        return shared_future<_Ty>(_STD move(*this));
    }

 

예제 4
예제 3에서 shared_future를 이용하여 여러 스레드에서 결과를 대기하는 프로그램 코드이다.

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

void add(std::promise<int>&& p, int a, int b)
{
    std::this_thread::sleep_for(1s);
    p.set_value(a + b);
}

void consume(std::shared_future<int> sf)
{
    sf.get();
    std::cout << "finish foo" << std::endl;
}
int main()
{
    std::promise<int> pm;
    std::future<int> ft = pm.get_future();
    std::shared_future<int> sft = ft.share();

//    std::shared_future<int> sft = pm.get_future();
 
    std::thread t(add, std::move(pm), 10, 20);

    std::thread t1(consume, sft);
    std::thread t2(consume, sft);

    t.join();
    t1.join();
    t2.join();
}

 

결과

finish foo
finish foo

 

이처럼 하나의 결과를 여러 스레드에서 대기하여 실행가능하다.

 

참고

share() 함수대신 future를 꺼낼 때 아예 shared_future를 꺼내서 사용하는 것도 가능하다.

//    std::future<int> ft = pm.get_future();
//    std::shared_future<int> sft = ft.share();

    std::shared_future<int> sft = pm.get_future();

 

* 본 글에 예제 코드는 코드누리 교육을 받으면서 제공된 샘플코드를 활용하였습니다.