[운영체제] Concurrency : Condition Variables

10 Jun 2018

Reading time ~8 minutes

출처 : http://pages.cs.wisc.edu/~remzi/OSTEP/

Concurrency

Condition Variables를 사용하는 상황

많은 상황에서 쓰레드가 실행되기에 앞서 실행이 될 수 있는 상태인지를 확인을 해야하는 것이 필요합니다. 예를 들면, 자식 쓰레드가 끝난지를 확인하는 부모 쓰레드가 있을 수 있습니다.(join()이란 함수로 구현됩니다.) 코드로 보면 아래와 같을 수 있습니다.

void *child(void *arg){
    printf("child\n");
    return NULL;
}

int main(int argc, char *argv[]){
    printf("parent: begin\n");
    pthread_t c;
    /*
     * int pthread_create(pthread_t *thread, const pthread_attr_t *attr,
     *                     void *(*start_routine) (void *), void *arg);
     *
     * pthread_attr_t를 조정해서 유저 모드, 커널 모드 쓰레드를 지정해줄 수 있습니다.
     * NULL은 시스템에서 세팅된 기본값이 됩니다.
     */
    pthread_create(&c, NULL, child, NULL);
    printf("parent: end\n");
    return 0;
}

하지만 아쉽게도 이 코드는 정상적인 동작을 보장하지 못합니다. 왜냐하면, 상황에 따라 자식 쓰레드가 먼저 끝날 수도 있기 때문입니다. 그렇다면 이를 해결하기 위해 어떤 방법들이 나왔는 지 알아보도록 하겠습니다.

Spin-based Approach

이전에 락에서 우리는 스핀을 사용하여 문제를 푸는 방법을 알아봤습니다. 여기서도 그 방법을 적용을 해서 문제를 접근해보도록 하겠습니다.

volatile int done = 0;

void *child(void *arg){
    printf("child\n");
    done = 1;
    return NULL;
}

int main(int argc, char *argv[]){
    printf("parent: begin\n");
    pthread_t c;
    pthread_create(&c, NULL, child, NULL);
    while(done == 0) ;
    printf("parent: end\n");
    return 0;
}

이 방법은 정확한 동작을 보장할지는 모르지만 앞선 락과 동일하게 CPU 시간을 많이 잡아먹는 비효율적인 방법입니다.

Condition Variable

앞서 이야기한 조건을 만족하기 위해서 조건 변수를 사용하도록 합니다. 이는 2가지의 주요 기능으로 나뉠 수 있습니다.

Waiting on the condition
- 실행 상태가 만족되지 않는 쓰레드의 경우 명시적인 큐(Explicit Queue)에 넣어지는 것을 의미합니다.
Signaling on the condition
- 실행 상태가 만족되었다고 다른 쓰레드에게 작업 허용을 보내는 것을 의미합니다.

POSIX

POSIX체계 상에서 어떻게 작업 틀(Routines)이 정의가 되는 지 알아보도록 하겠습니다.

조건 변수를 선언하도록 합니다.
- pthread_cond_t c;
적절한 명령어를 실행하도록 합니다.
- pthread_cond_wait(pthread_cond_t *c, pthread_mutex_t *,); → wait()
- pthread_cond_signal(pthread_cond_t *c); → signal()
- wait()는 mutex를 매개 변수로 부릅니다.
  - wait()가 불러지면 뮤텍스(mutex)를 가집니다.
  - wait()가 호출되면 락을 해제¹하고, 쓰레드를 잠자는 상태로 만들도록 합니다.
  - 쓰레드가 깨어나면 다시 락을 얻도록 합니다.
- signal()은 대기 상태의 조건 변수인 쓰레드 하나를 깨우도록 합니다.

이런 조건 변수를 활용한 구현은 아래와 같습니다.

int done = 0;
pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t c = PTHREAD_COND_INITIALIZED;

// 조건 변수를 통해 작업이 종료될 때까지 기다립니다.
void thr_exit() {
    pthread_mutex_lock(&m);
    done = 1;
    pthread_cond_signal(&c);
    pthread_mutex_unlock(&m);
}

void *child(void *arg){
    printf("child\n");
    thr_exit();
    return NULL;
}

/*
 * 부모 프로세스가 자식을 생성하고 자식이 끝날 때까지
 * 기다리도록 합니다.
 */
void thr_join(){
    pthread_mutex_lock(&m);
    while(done == 0)
        pthread_cond_wait(&c, &m);
    // 자식에 의해서 깬 뒤 시작하는 위치입니다.
    pthread_mutex_unlock(&m);
}

int main(){
    printf("parent: begin\n");
    pthread_t p;
    pthread_create(&p, NULL, child, NULL);
    thr_join();
    printf("parent: end\n");
    return 0;
}

이 코드의 수행은 2가지로 나뉠 수 있습니다.

부모가 먼저 수행되는 경우
자식이 먼저 수행되는 경우

1과 2는 유사하게 동작합니다. 하지만 1의 경우 부모가 wait() 과정이 있는 반면에, 2의 경우 그런 과정이 존재하지 않습니다. 왜냐하면, 2에서 thr_exit()에 의해 설정된 done 변수로 인해 thr_join()의 while이 무시되기 때문입니다. 여기서도 나타나지만 저 done변수가 매우 중요합니다. 만약 done변수가 존재하지 않게 되면 2의 경우에 문제가 생기게 됩니다. 자식이 먼저 수행되면서 신호를 보내지만 신호를 받는 쓰레드는 아무 없게 됩니다. 그 후에 부모가 다시 원래대로 돌아가서 wait()작업을 수행을 하게 됩니다. 그렇게 되면 부모를 깨울 수 있는 쓰레드가 영영 잠자게 됩니다. 좀 더 이해를 돕기위해 thr_join()과 thr_exit()이 아래와 같이 되어있다고 생각하면 쉬울 겁니다.

void thr_exit() {
    pthread_mutex_lock(&m);
    // done = 1; // 삭제
    pthread_cond_signal(&c);
    pthread_mutex_unlock(&m);
}

void thr_join(){
    pthread_mutex_lock(&m);
    // while(done == 0) // 삭제
    pthread_cond_wait(&c, &m);
    pthread_mutex_unlock(&m)
}

그렇다면 done만 표현해도 괜찮을까요? 그렇지 않습니다. done만을 가지고 표현을 해도 2의 경우에서 문제가 발생하게 됩니다. 만약 if문에서 문맥 교환이 일어나는 순간 위에서 이야기 한 문제가 다시 나타나서 부모는 여전히 잠자는 상태가 만들어지게 됩니다.

void thr_exit() {
    // pthread_mutex_lock(&m); // 삭제
    done = 1;
    pthread_cond_signal(&c);
    // pthread_mutex_unlock(&m); // 삭제
}

void thr_join(){
    // pthread_mutex_lock(&m); // 삭제
    if(done == 0) // while을 if로 변경 
        pthread_cond_wait(&c, &m);
    // pthread_mutex_unlock(&m) // 삭제
}

따라서 조건 변수를 사용한다 하여도 락이 필요하며, 작업이 종료되었는 지를 판별해야하는 변수가 존재해야 함을 알 수 있습니다.

Producer / Consumer (Bound Buffer) Problem

여기서 생산자(Producer)는 데이터 아이템들(data items)을 생산을 하고, 그 값을 버퍼에 넣기를 원하는 것을 지칭합니다. 소비자(Consumer)는 데이터 아이템들을 버퍼로부터 꺼내어 어떤 방식으로 사용하기 원하는 것입니다. 예를 들어, 생산자는 작업 큐에 HTTP 요청을 넣는 일을 하고 소비자는 작업 큐에서 요청을 꺼내고 그것을 바탕으로 작업을 수행하기 원하는 경우가 있을 수 있습니다. 그리고 여기서 유한 버퍼(Bounded buffer)라는 것이 있습니다. 이는 하나의 프로그램이 다른 하나의 프로그램으로 출력을 보낼 때 사용되는 것입니다. 이런 유한 버퍼는 공유 자원이므로 동기화 접근이 필수적입니다. 가장 기본적인 방식은 아래와 같습니다.

int buffer;
int count = 0;

void put(int value){
    // buffer에 공간이 있는가?
    assert(count == 0);
    count = 1;
    buffer = value;
}

void get(){
    // buffer에 내용이 존재하는가?
    assert(count == 1);
    count = 0;
    return buffer;
}

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

void *consumer(void *arg){
    int i;
    while(1){
        int temp = get();
        printf("%d", temp);
    }
}

여기서 좀 더 개선하여 pthread를 추가하도록 하겠습니다.

pthread_cond_t cond;
pthread_mutex_t mutex;

void *producer(void *arg){
    int i;
    int loops = (int) arg;
    for(i = 0; i < loops; i++){
        pthread_mutex_lock(&mutex);                         // p1
        if(count == 1)                                      // p2
            pthread_cond_wait(&cond, &mutex);               // p3
        put(i);                                             // p4
        pthread_cond_signal(&cond);                         // p5
        pthread_mutex_unlock(&mutex);                       // p6
    }
}

void *consumer(void *arg){
    int i;
    int loops = (int) arg;
    for(i = 0; i < loops; i++){
        pthread_mutex_lock(&mutex);                         // c1
        if(count == 0)                                      // c2
            pthread_cond_wait(&cond, &mutex);               // c3
        int tmp = get();                                    // c4
        pthread_cond_signal(&cond);                         // c5
        pthread_mutex_unlock(&mutex);                       // c6
        printf("%d\n", tmp);
    }
}

이렇게 하면 좀 더 생산자, 소비자 문제가 안정적으로 풀릴 수 있습니다. 하지만 불행이도 이것도 생산자 하나, 소비자 하나일 때 옳은 말입니다. 만약, 생산자 하나에 소비자 여럿이 붙게 된다면 이는 성립할 수 없습니다. 아래의 표를 참고하도록 하겠습니다.

invalid_case

여기서 소비자1(\(T_{c1}\)), 소비자2(\(T_{c2}\)), 생산자(\(T_p\))로 정의하도록 하겠습니다. 먼저, 소비자1이 동작을 했으나 아무것도 얻을 수 없어서 소비자1이 잠을 자게 됩니다. 그리고 생산자가 적절한 내용을 생산을 하였습니다. 그리고 소비자1을 깨웠습니다. 하지만 소비자1이 미처 깨기도 전에 소비자2가 수행되었고, 소비자1은 전혀 내용을 얻을 수 없는 상태인데도 get()을 부르는 촌극이 만들어지게 됩니다. 이런 P6 ~ C6의 구역과 같이 쓰레드가 당장 깨서 작업을 할 것이라 예상을 하나 그 예상이 깨어지는(보장되지 않는) 부분을 메사 시맨틱(mesa semantic)이라고 합니다. 여기서 가장 큰 문제는 소비자가 생산자의 부름에도 바로 일어나지 못한다는 점에 있습니다. 그래서 만들어진 해결책은 wait가 끝나도 다시 한 번 확인을 해보는 방법이 나오게 되었습니다. 즉, 아래와 같이 생산자와 소비자의 내용이 수정됩니다.

pthread_cond_t cond;
pthread_mutex_t mutex;

void *producer(void *arg){
    int i;
    int loops = (int) arg;
    for(i = 0; i < loops; i++){
        pthread_mutex_lock(&mutex);                         // p1
        while(count == 1) // if를 while로 변경              // p2
            pthread_cond_wait(&cond, &mutex);               // p3
        put(i);                                             // p4
        pthread_cond_signal(&cond);                         // p5
        pthread_mutex_unlock(&mutex);                       // p6
    }
}

void *consumer(void *arg){
    int i;
    int loops = (int) arg;
    for(i = 0; i < loops; i++){
        pthread_mutex_lock(&mutex);                         // c1
        while(count == 0) // if를 while로 변경              // c2
            pthread_cond_wait(&cond, &mutex);               // c3
        int tmp = get();                                    // c4
        pthread_cond_signal(&cond);                         // c5
        pthread_mutex_unlock(&mutex);                       // c6
        printf("%d\n", tmp);
    }
}

하지만 이 코드는 치명적인 문제가 있습니다. 소비자1가 신호를 보냈는 데(pthread_cond_signal) 정작 일어날 생산자가 안 일어나고 소비자2가 일어나는 문제가 발생할 수 있습니다. 이렇게 된다면 소비자1이 작업을 끝내고 잠자는 상태에 들어가고 소비자2는 내용이 없음을 확인하고 잠을 자게 되고 생산자는 계속 자게 되어 결국 모두가 자고 있는 앞에서 이야기한 것보다 더 큰 촌극이 벌어질 수 있습니다. 그래서 이를 해결하기 위해서 기존의 cond변수를 목적에 따라 empty, fill의 두 개의 변수를 만드는 방법으로 해결 방법을 찾아냈습니다. 결국 아래와 같은 코드가 만들어지게 됩니다.

cond_t empty, fill;
mutex_t mutex;

void *producer(void *arg){
    int i;
    int loops = (int) arg;
    for(i = 0; i < loops; i++){
        pthread_mutex_lock(&mutex);
        while(count == 1) 
            pthread_cond_wait(&empty, &mutex); // cond → empty
        put(i);
        pthread_cond_signal(&fill); // cond → fill
        pthread_mutex_unlock(&mutex);
    }
}

void *consumer(void *arg){
    int i;
    int loops = (int) arg;
    for(i = 0; i < loops; i++){
        pthread_mutex_lock(&mutex);
        while(count == 0) 
            pthread_cond_wait(&fill, &mutex); // cond → fill
        int tmp = get();
        pthread_cond_signal(&empty); // cond → empty
        pthread_mutex_unlock(&mutex);
        printf("%d\n", tmp);
    }
}

이제 모든 문제는 해결되었습니다. 이제 성능을 높혀 보도록 하겠습니다. 버퍼의 공간을 늘리고 문맥 교환을 최소화 시키도록 하겠습니다.

int buffer[MAX];
int fill = 0;
int use = 0;
int count = 0;

void put(int value){
    buffer[fill] = value;
    fill = (fill + 1) % MAX;
    count++;
}

void get(){
    int tmp = buffer[use];
    use = (use + 1) % MAX;
    count--;
    return tmp;
}

cond_t empty, fill;
mutex_t mutex;

void *producer(void *arg){
    int i;
    int loops = (int) arg;
    for(i = 0; i < loops; i++){
        pthread_mutex_lock(&mutex);
        while(count == MAX)
            pthread_cond_wait(&empty, &mutex);
        put(i);
        pthread_cond_signal(&fill);
        pthread_mutex_unlock(&mutex);
    }
}

void *consumer(void *arg){
    int i = 0;
    int loops = (int) arg;
    for(i = 0; i < loops; i++){
        pthread_mutex_lock(&mutex);
        while(count == 0)
            pthread_cond_wait(&fill, &mutex);
        int tmp = get();
        pthread_cond_signal(&empty);
        pthread_mutex_unlock(&mutex);
        printf("%d\n",tmp);
    }
}

하지만 이 방식은 무조건 버퍼가 비어있어야만 소비자가 할당을 받을 수 있는 부족한 점이 있습니다. 따라서 서로 다른 할당량을 필요로 하는 소비자들 사이에서의 자원 할당 방식에 대한 해결 방법은 얼만큼 할당되어 있는 지 확인하고, 소비자 본인이 할당 받을 수 있는 양이면 할당 받도록 하는 방법을 이용하기로 하였습니다. 이에 따라 구현을 하면 아래와 같이 됩니다.

int bytesLeft = MAX_HEAP_SIZE;

cond_t c;
mutex_t m;

void *allocate(int size){
    pthread_mutex_lock(&m);
    while(bytesLeft < size)
        pthread_cond_wait(&c, &m);
    void *ptr = ...; // heap에서 메모리를 할당 받습니다.
    bytesLeft -= size;
    pthread_mutex_unlock(&m);
    return ptr;
}

void free(void *ptr, int size){
    pthread_mutex_lock(&m);
    bytesLeft += size;
    pthread_cond_signal(&c);
    pthread_mutex_unlock(&m);
}

아쉽게도 이 방식도 부족한 점이 있습니다. pthread_cond_signal()이 누구를 깨울 지 모르기 때문에 필요로 하는 놈이 깰 때까지 기다리는 것은 큰 낭비가 됩니다. 이에 대한 해결책으로 기존의 pthread_cond_signal()을 pthread_cond_broadcast()라는 전체를 깨워주는 명령어를 사용하도록 합니다. 이 방식의 유일한 단점은 너무 많은 쓰레드가 순간적으로 일어남에 있습니다. 이를 효율적으로 해결하기 위한 방법을 다음에 알아보도록 하겠습니다.

락을 둔 채로 잠자는 상태로 들어가게 되면 데드락이 발생할 수 있습니다. ↩

os Concurrency