[Effective Modern C++] 39. void future 객체

항목 39. 단발성 이벤트 통신에는 void future 객체를 고려하라

 

 

 

스레드 간 통신을 처리하는 방법

• 특정 이벤트를 감지하여 브로드캐스팅 하는 task가 유용할 때가 있다.
  • 자료구조의 초기화
  • 계산 과정 중 특정 단계의 완료 등.
• 이런 통신은 어떻게 처리해야 할까?

 

 

 

조건 변수(condition variable, condvar)

• 조건 변수를 사용하여 해결해 본다.
• 검출 task(detecting task)
  • 조건을 검출하는 task
• 반응 task(reacting task)
  • 조건에 반응하는 task
• 반응 task는 하나의 조건 변수를 기다리고,
• 검출 task는 사건이 발생하면 그 조건 변수를 노티한다.

std::condition_variable cv; // 이벤트를 위한 조건 변수

std::mutex m; // cv와 함께 사용할 뮤텍스

// 검출 task
{
    ... // 이벤트 검출

    cv.notify_one(); // 반응 task에 노티한다.
}

// 반응 task
{ // 범위 시작
    std::unique_lock<std::mutex> lk(m); // 뮤텍스를 잠근다.
    
    cv.wait(lk); // 노티를 기다린다.(잘못된 방식!)
    
    ... // 이벤트에 반응한다.(m이 잠금 상태)
    
} // lk의 소멸자가 m을 해제한다.

... // 계속 반응한다.(m은 잠금 상태가 풀렸다.)

• 코드 악취(code smell), 즉 뭔가 잘못된 게 있는 것 같다.
  • 뮤텍스가 필요하다는 부분이다.
  • 뮤텍스는 공유 자료에 대한 접근을 제어하는 데 쓰이지만,
  • 검출 task, 반응 task에는 그런 접근 제어가 전혀 필요없을 수도 있다.
• 그 뿐 아니라 반드시 처리해야 할 문제점도 두 가지가 있다.

 

 

 

문제점 두 가지

• 반응 task가 wait를 실행하기 전에 검출 task가 조건 변수를 노티하면 반응 task가 행잉된다.
  • 검출 task가 노티를 실행했는데, 반응 task가 wait를 실행하기 전이라면?
  • 그 노티를 놓치고, 영원히 노티를 기다리게 된다.
• wait 호출문은 가짜 기상(spurious wakeup)을 고려하지 않는다.
  • 조건 변수가 노티되지 않았는데도 조건 변수를 기다리는 코드가 깨어날 수 있다.
    • 이는 스레드 적용 API들에서 흔히 있는 일이다.
  • 깨어난 후 가장 먼저, 기다리던 조건이 정말로 발생했는지 확인해야 한다.
  • 기다리던 조건을 판정하는 람다를 wait에 넘겨줄 수 있다.

    • cv.wait(lk, []{ return 이벤트 발생 여부 체크; });

  • 위 람다를 사용하려면...
    • 반응 task 쪽에서 기다리던 조건의 true, false를 판정할 수 있어야 한다.
    • 헌데 그 조건이란, 특정 사건의 발생인데 그 발생 여부를 검출하는 건 검출 task의 몫이다.
    • 즉, 반응 task는 그 조건을 판단할 수 없을 확률이 높다.
    • 애초에 반응 task가 그 조건을 판단할 수 있었다면 노티를 기다릴 필요도 없었을 것이다.

 

 

 

공유 bool 플래그

• 조건 변수가 힘들다면, 공유 bool 플래그를 사용하면 어떨까.

std::atomic<bool> flab(false); // 공유 플래그

// 검출 task
{
    ... // 사건을 검출한다.
    flag = true; // 반응 task에 노티한다.
}

// 반응 task
{
    ... // 반응 준비

    while(!flag); // 이벤트를 기다린다.

    ... // 이벤트에 반응한다.
}

• 이제 조건 변수 기반 설계의 단점은 없다.
  • 뮤텍스를 사용할 필요가 없다.
  • 반응 task가 폴링을 시작하기 전에 검출 task가 플래그를 설정해도 문제 없다.
  • 가짜 기상이 존재하지 않는다.
• 단점은 반응 task의 폴링 비용이 높다는 점이다.
  • 플래그가 설정되길 기다리는 동안 반응 task는 사실상 차단된 상태지만,
  • 그래도 여전히 실행 중이다.
    • 여전히 하드웨어 스레드를 점유하고 있고,
    • 자신의 시간 조각의 시작이나 끝에서 문맥 전환 비용을 유발하고,
    • 전원 절약을 위해 닫아도 될 코어를 계속 돌리게 된다.
  • task가 진짜 차단되었다면 전혀 생기지 않을 일이다.

 

 

 

조건 변수 + 플래그

• 이벤트 발생 여부를 플래그로 나타내되,
  • 그 플래그에 대한 접근을 뮤텍스로 동기화한다.
• 이 경우, 플래그에 대한 동시 접근을 뮤텍스가 방지하기 때문에 플래그가 std::atomic일 필요는 없다.(항목 40 참조)

std::condition_variable cv;
std::mutex m;

bool flag(false); // 플래그

// 검출 task
{
    ... // 이벤트 검출
    {
        std::lock_guard<std::mutex> g(m); // g의 생성자에서 m을 뮤텍스 잠금
    
        flag = true; // 반응 task에 노티(1), 가짜 기상 방지용
        
    } // g의 소멸자에서 뮤텍스 해제
    
    cv.notify_one; // 반응 task에 노티(2)
}

// 반응 task
{
    ... // 반응 준비
    {
        std::unique_lock<std::mutex> lk(m);
        
        cv.wait(lk, [] { return flag; }}; // 가짜 기상을 방지하기 위한 플래그 사용
        
        ... // 이벤트에 반응(m은 잠긴 상태)
        
    } // m을 풀어준다.
    
    ... // 계속 반응
}

• 가짜 기상도 막았고, 폴링도 수행하지 않는다.
• 다만 코드 악취가 아직 남아있다.
  • 검출 task가 아주 기묘한 방식으로 반응 task와 통신한다는 점이다.
  • 검출 task는 기다리던 이벤트가 발생했음을 노티하기 위해 위해 조건 변수 / 플래그 모두 사용한다.
  • 반응 task는 조건 변수 노티만으로는 확신하지 못하고 반드시 플래그를 점검해야 한다.

 

 

 

반응 task가 검출 task의 future 객체를 기다리게 하기

• future 객체는 통신 채널의 호출자 쪽 핸들에 해당한다.(항목 38 참조)
• 검출 task - 반응 task 는 호출자 - 피호출자 관계가 아니다.
  
  • 그렇지만, 전송 단자가 std::Promise이고 수신 단자가 future 객체인 통신 채널을
  • 반드시 호출자 - 피호출자 통신에서만 사용해야 하는 것은 아니다.
  • 프로그램의 한 장소에서 다른 장소로 정보를 전송해야 하는 모든 상황에서 사용할 수 있다.
• 설계
  • 검출 task에서 std::promise 객체(통신 채널 전송 단자)를 하나 두고
  • 반응 task에는 그에 대응되는 future 객체를 하나 둔다.
  • 이벤트가 발생했음을 인식하면 검출 task는 자신의 std::promise를 설정한다.(통신 채널에 정보 기록)
  • 반응 task는 자신의 future 객체에 대해 wait를 호출해둔 상태이다.
    • wait 호출은 std::promise가 설정될 때까지 차단된다.
• 헌데, std::promise와 std::future 모두 템플릿이다.
  • 즉, 타입 매개변수를 요구한다.
  • 그러나 지금 예에서는 딱히 전송할 자료가 없다.
    • 반응 task는 그저 turue 객체가 설정되었는지만 확인 가능하면 된다.
  • 그럴 때 void 타입을 사용하면 된다.
• 결론적으로 검출 task는 std::promise<void>
  • 반응 task는 std::future<void>(혹은 std::shared_future<void>)를 사용하면 된다.

std::promise<void> p;

// 검출 task
{
    ...
    p.set_value();
}

// 반응 task
{
    ... // 반응 준비
    p.get_future().wait();  // p에 해당하는 future 객체를 기다린다.
    ... // 이벤트에 반응
}

• 장점
  • 더이상 뮤텍스가 필요하지 않다.
  • 반응 task가 wait로 대기하기 전에 검출 task가 자신의 std::promise를 설정해도 작동한다.
  • 가짜 기상이 없다.
  • 반응 task는 wait 호출 후 진짜로 차단된다.
• 하지만 여전히 난관이 있다.
  
  •  std::promise와 future 객체 사이에는 공유 상태가 있다.
    • 대체로 공유 상태는 동적으로 할당된다.
    • 즉, 이 설계는 힙 기반 할당 해제 비용을 유발한다.
  • std::promise를 한 번만 설정할 수 있다.
    • std::promise와 future 객체 사이의 통신 채널은 단발성(one-shot) 매커니즘이다.

 

 

 

일시정지된 스레드 생성

• 일시정지된 스레드가 필요한 이유
  • 스레드 실행 전에 스레드 생성에 관련된 모든 추가부담을 미리 처리하고 싶을 때.
  • 실행 전에 스레드를 좀 더 세밀하게 구성하고 싶을 때.
• 스레드를 한 번만 일시정지 하면 된다면, void future 객체가 유용하다.

std::promise<void> p;

void react(); // 반응 task 함수

void detect() // 검출 task 함수
{
    // 스레드 생성
    std::thread t([]{
                        p.get_future().wait(); // future 객체 설정 전까지 t를 일시정지
                        react();
                    });
                    
    ... // t는 여전히 일시정지. 필요한 작업을 한다.
    
    p.set_value(); // t의 일시정지를 푼다.(이제 react가 호출된다.)
    
    ... // 추가작업을 수행한다.
    
    t.join(); // t를 합류 불가능으로 만든다.(항목 37 참조)
}

• detect 바깥의 모든 경로에서 t를 합류 불가능으로 만들고 싶으므로 ThreadRAII를 사용한다.(항목 37 참조)

void detect()
{
    ThreadRAII tr(
        std::thread([]{
                           p.get_future().wait();
                           react();
                      }),
        ThreadRAII::DtorAction::join // 위험!!!!
    );
    
    ... // 만일 여기서 예외가 발생하면? 아래에서 설명
    
    p.set_value(); // 일시정지 해제
    
    ...
}

• 만일 예외가 발생하면 p에 대한 set_value 호출이 일어나지 않는다.
  • 따라서 람다 안의 wait 호출은 계속해서 차단된다.
  • 람다를 실행하는 스레드는 결코 완료되지 않는다.
  • 이는 반드시 해결되어야 한다.
    • RAII 객체의 소멸자가 그 스레드에 대해 join을 호출하기 때문이다.
    • 이 부분은 연습 문제로 남겨두고 패스.
• 반응 task 여러개를 일시정지 시키고 풀도록 확장할 수 있다.
  • std::promise와 future 객체 사이의 통신 채널은 단발성(one-shot) 매커니즘이라는 단점을 상쇄시킬 수 있다.
  • 첫 번째 버전의 코드를 간단하게 일반화 하기만 하면 된다.
  • react 코드에서 std::futre 대신 std::shared_future를 사용하게 바꾼다.
    • std::future의 share 멤버 함수는 해당 함수가 리턴하는 std::shared_future 객체에 자기 공유 상태의 소유권을 넘겨준다.

std::promise<void> p;

void detect() // 여러 개의 반응 task에 노티한다.
{
    auto sf = p.get_future().share(); // sf의 타입은 std::shared_future<void>
    
    std::vector<std::thread> vt; // 반응 스레드들을 담는 컨테이너
    
    for(int i = 0 ; i < threadsToRun ; ++i)
    {
        vt.emplace_back([sf]{
                                sf.wait(); // sf의 지역 복사본을 기다린다.
                                react();
                            });
    }
    
    ... // 예외가 발생하면 프로그램이 종료되는 문제는 여전.
    
    p.set_value(); // 모든 스레드의 일시 정지를 해제한다.
    
    ...
    
    for(auto& t : vt) // 모든 스레드를 합류 불가능으로 만든다.
    {
        t.join();
    }
}