3-6장. 컨디션 변수

지금까지 락의 개념을 학습하고 하드웨어와 운영체제의 적절한 지원을 통해 제대로 된 락을 만드는 법을 살펴보았다. 불행히도 "락"만으로는 병렬 프로그램을 제대로 작성할 수 없다.

 

쓰레드가 계속 진행하기 전에 어떤 조건(Condition)이 참인지를 검사해야 하는 경우가 많이 있다. 예를 들어 부모 쓰레드가 작업을 시작하기 전에 자식 쓰레드가 작업을 끝냈는지를 검사하기를 원할 수 있다.(보통 join() 연산이라고 불린다) 이러한 대기문은 어떻게 구현해야 할까?

void child()
{
    std::cout << "child\n";
    // 작업이 끝났음을 어떻게 알리는가?
}

void main()
{
    std::cout << "parent : begin\n";
    std::thread c{ child }; // 자식 쓰레드 생성
    // 자식 쓰레드를 어떻게 기다리는가?
    std::cout << "parent : end\n";
}

우리는 이 코드에 대해서 다음과 같은 출력을 원한다.

parent : begin
child
parent : end

이를 위해 스핀 기반 접근으로 자식이 종료되기를 기다리는 방법을 사용할 수도 있다.

volatile int done = 0;

void child
{
    cout << "child\n";
    done = 1;
}

int main()
{
     cout << "parent : begin\n";
     thread c {child}; // 자식 쓰레드 생성
     while(done == 0); // 자식 쓰레드가 끝날때까지 스핀
     cout << "parent : end\n";
}

 

이 방법은 제대로 동작하지만 부모 쓰레드가 스핀하면서 CPU 시간을 낭비하기 때문에 비효율적이다. 

 

컨디션 변수 정의와 루틴들

조건이 참이 될 때까지 기다리기 위해서 컨디션 변수(conditional variable)를 활용할 수 있다. 컨디션 변수는 일종의 큐 자료 구조로서, 어떤 실행의 상태(조건)가 원하는 것과 다를 때 조건이 참이 되기를 기다리며 쓰레드가 대기(Waiting)할 수 있는 큐이다. 다른 쓰레드가 상태를 변경시켰을 때, 대기 중이던 쓰레드를 깨우고(Signal, 신호), 계속 진행할 수 있도록 해준다.

 

컨디션 변수는 다음과 같이 선언한다.

std::condition_variable cv

컨디션 변수에는 wait()와 notify_one() 이라는 두 개의 연산자가 있다. 

cv.wait(std::unique_lock lk);
cv.notify_one();

wait()은 쓰레드 스스로를 잠재우기 위해서 호출하는 것이다. wait()은 unique_lock 클래스의 mutex를 매개변수로 넣어줘야 한다. wait()은 매개변수 락을 해제하고 자신을 대기 상태로 변경한다. 그리고 notify_one()으로 깨어날 때, 락을 다시 잠근다.

 

다음은 컨디션 변수를 이용한 join() 연산을 함수로 만든 코드이다.

#include <thread>
#include <mutex>
#include <condition_variable>
#include <iostream>

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

bool done = false;

void thr_exit()
{
   // 부모를 깨우기 위해 호출
    m.lock();
    done = true;
    // 신호를 보내어 부모를 깨운다.
    cv.notify_one();
    m.unlock();
}

void child()
{
    std::cout << "child\n"
    thr_exit(); // 자식 쓰레드가 끝났음을 알린다.
}

void thr_join()
{
    std::unique_lock<std::mutex> lk{ m };
    // 자식 쓰레드가 종료되지 않았을 경우 wait()을 호출하여 자신을 대기 상태에 둠.
    while( false == done ) cv.wait(lk);
}

int main()
{
    std::cout << "parent : begin\n";
    std::thread p { child };
    thr_join(); // 자식 쓰레드의 종료를 기다리기 위해 호출.
    std::cout << "parent : end\n";
}

이렇게 구현을 해봤지만 실제로는 처리할 일들이 더 있다. 때문에 실제로는 이런 식으로 사용자 정의 함수를 만들어 사용하지는 않고 std::thread 객체 내에 존재하는 join() 멤버 함수를 사용한다.(위의 경우에는 p.join() ) 

 

상태 변수의 중요성

위의 코드에서 상태 변수는 done이다. 만약 자식 쓰레드가 done 없이 즉시 실행된다면 어떻게 될 것인가? 그런 경우에 자식의 신호를 받은 컨디션 변수에서 대기 중인 쓰레드는 없을 것이다. 또한 나중에 자식이 끝났음을 신호로 보내 부모가 실행될 때, wait에서 멈추고 누구도 깨우지 못하는 상황이 벌어진다.

void thr_exit()
{
    m.lock();
    cv.notify_one();
    m.unlock();
}

void thr_join()
{
    std::unique_lock<std::mutex> lk{ m };
    cv.wait(lk);
}

 

다른 잘못된 구현

다음과 같이 락을 획득하지 않는 경우에도 문제가 생긴다.

void thr_exit()
{
    done = true;
    cv.notify_one();
}

void thr_join()
{
    if(false == done) cv.wait(lk);
}

 

미묘한 경쟁 조건 문제가 생긴다. 부모가 thr_join()을 호출하면 부모가 done값을 검사하고 false가 되었음을 확인하고 대기 상태로 전환한다. 하지만 wait을 호출하기 바로 전에, 문맥 교환이 일어나 자식 쓰레드를 실행하게 될 수도 있다.  

 

자식 쓰레드 작업이 완료되면 done을 1로 변경한 후 신호를 보낸다. 하지만 wait을 호출하지 않아 대기하고 있는 쓰레드가 없기 때문에 부모가 다시 실행되면, 영원히 잠들게 된다.

 

생산자 / 소비자 (유한 버퍼) 문제

위의 문제보다 좀 더 복잡한 동기화 문제인 생산자 / 소비자 문제이다. (유한 버퍼 문제라고도 알려져 있다)

 

여러 개의 생산자 쓰레드와 소비자 쓰레드가 있다고 하자. 생산자는 데이터를 만들어(Produce) 버퍼에 넣고, 소비자는 버퍼에서 데이터를 꺼내어 사용(Consume)한다.

 

이러한 관계는 실제 시스템에서 자주 일어난다. 예를 들면, 멀티 쓰레드 웹 서버의 경우 생산자는 HTTP 요청을 작업 큐(유한 버퍼)에 넣고, 소비자 쓰레드는 이 큐에서 요청을 꺼내어 처리한다.

 

유한 버퍼는 공유 자원이다. 경쟁 조건의 발생을 방지하기 위해 동기화가 필요하다. 이 문제를 좀 더 잘 이해하기 위해서 실제 코드를 살펴보도록 하자.

int buffer;
int count = 0; // 처음에는 비어있음

void put(int value)
{
    assert(count == 0);
    count = 1;
    buffer = value;
}

int get()
{
    assert(count == 1);
    count = 0;
    return buffer;
}

생산자는 넣고 소비자는 꺼내어 쓸 수 있는 공유 버퍼 buffer가 필요하다. 한 개의 정수를 사용하고, 공유 버퍼에 값을 넣는(put) 루틴과 버퍼에서 값을 꺼내는 루틴(get) 두 개가 있다.

count가 0인 경우에만(비었을 경우) 데이터를 버퍼에 넣는다. 그리고 count가 1인 경우에만(꽉 차있는 경우) 버퍼에서 데이터를 꺼낸다. 

 

이 작업은 두 종류의 쓰레드에 의해 수행될 것이다. 하나는 생산자 쓰레드들이고 다른 하나는 소비자 쓰레드들이다. 

void producer(int loops)
{
    for(int i = 0; i < loops; i++) put(i);
}

void consumer() 
{
    while(true)
    {
        int tmp = get();
        std::cout << tmp << std::endl;
    }
}

생산자 쓰레드는 정수를 loops 개수 만큼 버퍼에 넣고, 소비자 쓰레드는 공유 버퍼에서 데이터를 꺼낸다. 당연히 put()과 get() 루틴에는 임계 영역이 존재한다.(put()은 버퍼를 갱신, get()은 버퍼에서 읽기 때문에) 하지만 락만 추가한다고 해서 제대로 동작하는 것은 아니다. 무언가가 더 필요하다. 그 무언가는 바로 컨디션 변수이다. 다음 코드처럼 컨디션 변수 하나와 그것과 연결된 mutex 락을 사용한다.

conditional_variable cond;
mutex cv_mutex;

void producer(int loops)
{
    for(int i = 0; i < loops; i++)
    {
        unique_lock<mutex> lk (cv_mutex);
        if(count == 1)
            cond.wait(lk);
        put(i);
        cond.notify_one();
        // 자동으로 cv_mutex.unlock();
    }
}

void consumer(int loops)
{
    for(int i = 0; i < loops; i++) 
    {
        unique_lock<mutex> lk (cv_mutex);
        if(count == 0)
            cond.wait(lk);
        int tmp = get();
        cond.notify_one();
        lk.unlock();
        cout << tmp << endl;
    }
}

생산자 쓰레드는 버퍼가 빌 때까지 기다린다. 소비자 쓰레드는 버퍼가 차기를 기다린다.

 

생산자, 소비자 쓰레드가 각각 하나씩인 경우에는 위 코드는 제대로 동작한다. 하지만 두 개 이상의 같은 종류의 쓰레드가 있다면, 이런 해법에는 두 가지 문제점이 존재한다.

 

첫번째 문제점은 대기 명령 전의 if문과 관련이 있다. T(c1), T(c2)라는 두 개의 소비자가 있고 T(p)라는 생산자가 하나 있다고 가정해보자. 소비자 T(c1)이 먼저 실행된다. 락을 획득하고 버퍼를 소비할 수 있는지 검사한다. 비어있음을 확인한 후에 대기하며 락을 해제한다.

 

그리고 생산자 T(p)가 실행된다. 락을 획득하고 버퍼가 비었는지 확인한다. 비었음을 발견하고, 버퍼를 채운다. 생산자는 버퍼가 가득 찼다는 신호를 보낸다. 대기 중인 첫 째 소비자 T(c1)은 깨어나 준비 큐(ready queue)로 이동한다. T(c1)은 이제 실행할 수 있는 상태지만 아직 실행 상태는 아니다. 생산자 쓰레드는 실행을 계속한다. 버퍼가 차 있으므로 대기 상태로 바꾼다.

 

여기서 문제가 발생한다. 다른 소비자 T(c2)가 끼어들어서 실행하면서 버퍼 값을 소비한다.  T(c1)이 실행된다고 해보자. 대기에서 리턴하기 전에 락을 획득한다. 그리고 get()을 호출하지만 버퍼는 비어있다. 코드는 의도한 대로 돌아가지 않았다. 생산자 쓰레드가 버퍼에 넣어 둔 값을 T(c2)가 끼어들어서 소비하였기 때문에 T(c1)이 비어 있는 버퍼를 읽는 행위를 막았어야 했다. 

 

문제의 원인은 단순하다. T(c1)이 깨어나서 실행되기까지 사이에 유한 버퍼의 상태가 변경되었다. 신호는 쓰레드를 깨우기만 한다. 깨어난 쓰레드가 실제 실행되는 시점에도 그 상태가 유지된다는 보장은 없다. 

 

개선, 하지만 아직도 불완전

이 문제는 쉽게 해결할 수 있다. 생산자, 소비자 쓰레드에 있는 if 문을 while 문으로 바꾸면 된다. 유한 버퍼를 항상 검사하는 것이 더 안전하다.

 

하지만 아직도 코드에는 버그가 있다. 이제 두번째 문제점을 설명할 때가 왔다. 컨디션 변수가 하나뿐이라는 사실과 관계가 있다.  

conditional_variable cond;
mutex cv_mutex;

void producer(int loops)
{
    for(int i = 0; i < loops; i++)
    {
        unique_lock<mutex> lk (cv_mutex);
        while(count == 1)
            cond.wait(lk);
        put(i);
        cond.notify_one();
        // 자동으로 cv_mutex.unlock();
    }
}

void consumer(int loops)
{
    for(int i = 0; i < loops; i++) 
    {
        unique_lock<mutex> lk (cv_mutex);
        while(count == 0)
            cond.wait(lk);
        int tmp = get();
        cond.notify_one();
        lk.unlock();
        cout << tmp << endl;
    }
}

두 번째 문제는 소비자 쓰레드 T(c1)과 T(c2)가 먼저 실행한 후 둘 다 대기 상태에 있을 때 발생한다. 생산자 쓰레드가 실행되어 버퍼에 값을 넣고 대기 중인 쓰레드 하나( T(c1) )를 깨우고, 자신은 대기한다. 이제 하나의 소비자 쓰레드( T(c1) )가 실행할 준비가 되었고 조건에 의해 T(c2)와 T(p)는 대기 중이다. 

 

소비자 T(c1)이 wait()에서 리턴을 받아 깨어나고 조건을 재확인한다. 버퍼가 차있다는 것을 발견하고 값을 소비한다. 이 소비자는 신호를 전송하여 대기 중인 쓰레드 중 하나를 깨운다. 이때 어떤 쓰레드를 깨울 것인가?

 

소비자가 버퍼를 비웠기 때문에 생산자를 당연히 깨워야 한다. 하지만, 만약 소비자 T(c2)를 깨운다면 문제가 발생한다. 소비자 T(c2)가 깨어나면 버퍼가 비어 있다는 것을 발견한 후에 다시 대기 상태로 들어간다. 버퍼에 값을 넣어야 하는 생산자 T(p)는 대기 중이다. 다른 소비자 T(c1) 역시 대기 상태로 들어간다. 즉, 세 개의 쓰레드 모두 대기 상태이다. 

 

신호를 보내는 것은 꼭 필요하지만 대상이 명확해야 한다. 소비자는 다른 소비자를 깨울 수 없고 생산자만 깨워야 하며, 반대인 경우도 마찬가지이다.

 

이 문제 역시 약간의 변경만 필요로 한다. 두 개의 컨디션 변수를 사용하여 시스템의 상태가 변경되었을 때 깨워야 하는 쓰레드에게만 신호를 제대로 전달한다.

conditional_variable cv_empty, cv_fill;
mutex cv_mutex;

void producer(int loops)
{
    for(int i = 0; i < loops; i++)
    {
        unique_lock<mutex> lk (cv_mutex);
        while(count == 1)
            cv_empty.wait(lk);
        put(i);
        cv_fill.notify_one();
        // 자동으로 cv_mutex.unlock();
    }
}

void consumer(int loops)
{
    for(int i = 0; i < loops; i++) 
    {
        unique_lock<mutex> lk (cv_mutex);
        while(count == 0)
            cv_fill.wait(lk);
        int tmp = get();
        cv_empty.notify_one();
        lk.unlock();
        cout << tmp << endl;
    }
}