[동시성 프로그래밍] Ch3. 동기 처리 1 (하)

3.5 조건 변수

어떤 조건을 만족하지 않는 동안에는 프로세스를 대기 상태로 두고 조건이 만족되면 대기 중인 프로세스를 실행하고 싶을 때가 있다. 예를 들면 교차로의 신호등을 떠올릴 수 있는데, 이와 같은 신호등을 동시성 프로그래밍 세계에서는 조건 변수라고 부르며, 조건 변수를 기반으로 프로세스의 대기를 수행한다.

다음 코드는 Pthreads를 이용한 조건 변수의 예다. Pthreads에서는 pthread_cond 계열 타입과 함수를 이용해 조건 변수를 구현한다. 이 코드에는 어떤 데이터를 생성하는 프로세스와 생성된 데이터를 소비하는 프로세스가 있으며, 데이터를 소비하는 프로세스는 데이터가 생성될 때까지 대기한다.

#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

pthread_mutex mut = PTHREAD_MUTEX_INITIALIZER; // 1
pthread_cond_t cond = PTHREAD_COND_INITIALIZER; // 2

volatile bool ready = false; // 3
char buf[256]; // 스레드 사이에서 데이터를 주고 받기 위한 버퍼

void* producer(void *arg) { // 데이터 생성 스레드, 4
    printf("producer: ");
    fgets(buf, sizeof(buf), stdin); // 입력을 받는다.

    pthread_mutex_lock(&mut);
    ready = true; // 5

    if (pthread_cond_broadcast(&cond) != 0) { // 전체에 알림, 6
        perror("pthread_cond_broadcast"); exit(-1);
    }

    pthread_mutex_unlock(&mut);
    return NULL;
}

void* consumer(void *arg) { // 데이터 소비 스레드, 7
    pthread_mutex_lock(&mut);

    while (!ready) { // ready 변수값이 false인 경우 대기
        // 락 반환과 대기를 동시에 실행
        if (pthread_cond_wait(&cond, &mut) != 0) { // 8
            perror("pthread_cond_wait"); exit(-1);
        }
    }

    pthread_mutex_unlock(&mut);
    printf("consumer: %s\n", buf);
    return NULL;
}

int main(int argc, char *argv[]) {
    // 스레드 생성
    pthread_t pr, cn;
    pthread_create(&pr, NULL, producer, NULL);
    pthread_create(&cn, NULL, consumer, NULL);

    // 스레드 종료 대기
    pthread_join(pr, NULL);
    pthread_join(cn, NULL);

    // 뮤텍스 객체 반환
    pthread_mutex_destroy(&mut);

    // 조건 변수 객체 반환, 9
    if (pthread_cond_destroy(&cond) != 0) {
        perror("pthread_cond_destroy"); return -1;
    }

    return 0;
}

1. 조건 변수는 여러 스레드에서 접근하므로 조건 변수의 업데이트 등은 뮤텍스로 락을 얻은 뒤에 수행해야 한다.
2. 조건 변수 cond를 정의한다. 조건 변수 타입은 pthread_cond_t 이며 초기화는 PTHREAD_COND_INITIALIZER에서 수행한다. 초기화에 pthread_cond_init 함수를 이용할 수도 있다.
3. Pthreads가 아닌 커스텀 조건 변수 ready를 정의한다. 이것은 producer 함수를 이용한 데이터 생성이 consumer 스레드 생성 이전에 실행될 가능성이 있기 때문이며 Pthreads의 wait는 의사 각성이라 불리는 현상이 일어날 가능성이 있기 때문이다.
4. 데이터를 생성하는 producer 함수를 정의한다. 이 함수에서는 표준 입력으로 입력을 받아 그것을 생성 데이터로 하여 공유 버퍼인 buf에 저장하고 그 후 조건 변수에 접근하기 위해 락을 취득한다.
5. 버퍼에 데이터가 확실히 들어 있는 것을 나타내기 위해 자신의 조건 변수인 ready에 true를 설정한다.
6. 대기중인 모든 스레드에 알린다.
7. 데이터를 소비하는 consumer 함수를 정의한다. 이 함수에서는 조건 변수를 읽기 위해 먼저 락을 획득한다. 그 후 ready를 확인하여 공유 버퍼에서 읽기가 가능하다면 락을 반환하고 데이터를 읽는다.
8. 읽을 수 없는 경우에는 대기한다. pthread_cond_wait 함수는 락의 반환과 대기를 아토믹하게 수행하는 것을 보증한다. 대기 중에 다른 스레드가 pthread_cond_broadcast 또는 pthread_cond_signal 함수로 알림을 수행한 경우에는 대기를 종료하고 다시 락을 획득해서 처리를 재개한다.
9. 조건 변수의 리소스 반환은 pthread_cond_destroy 함수로 수행한다.

이 코드와 같이 프로세스를 생산자와 소비자로 나누는 프로그래밍 모델을 생산자-소비자(producer-consumer) 모델이라 부른다. 공유 변수로의 접근은 상태 관리가 복잡해지지만 생산자-소비자 모델을 적용하면 변수로의 접근 주체가 명확하게 되므로 간략하게 구현할 수 있다.

3.6 배리어 동기

배리어 동기(barrier synchronization)는 모든 스레드/프로세스가 어떤 시점에서 중지되고 다른 모든 스레드/프로세스가 장벽에 도달할 때까지 도달했다가 한꺼번에 실행되는 개념이다.

3.6.1 스핀락 기반의 배리어 동기

배리어 동기의 개념은 간단하다. 먼저 공유 변수를 준비하고, 프로세스가 어떤 지점에 도달한 시점에 해당 공유 변수를 증가한다. 공유 변수가 계속 증가되어 어떤 일정한 수에 도달하면 배리어를 벗어나 처리를 수행한다. 다음 코드는 스핀락 기반의 배리어 동기를 보여준다.

void barrier(volatile int *cnt, int max) { // 1
  __sync_fetch_and_add(cnt, 1); // 2
    while (*cnt < max); // 3
}

1. 공유 변수에 대한 포인터 cnt와 최댓값 max를 받는다.
2. 공유 변수 cnt를 아토믹하게 증가시킨다.
3. cnt가 가리키는 값이 max가 될 때까지 대기한다.

다음 코드는 배리어 동기를 이용하는 예다.

volatile int num = 0; // 공유 변수

void *worker(void *arg) { // 스레드용 함수
    barrier(&num, 10); // 모든 스레드가 여기에 도달할 때 까지 기다린다. 1
    // 무언가 처리

    return NULL;
}

int main(int argc, char *argv[]) {
    // 스레드 생성
    pthread_t th[10];
    for (int i = 0; i < 10; i++) {
        if (pthread_create(&th[i], NULL, worker, NULL) != 0) {
            perror("pthread_create"); return -1;
        }
    }
    return 0;
}

1. barrier 함수를 호출하여 배리어 동기를 실행한다. 모든 스레드가 barrier에 도달할 때까지는 5행의 처리가 실행되지 않는다. 여기에서 barrier 함수의 두 번째 인수가 10인 것은 10개의 스레드를 실행하기 때문이다.

3.6.2 Pthreads를 이용한 배리어 동기

스핀락을 이용한 배리어 동기에는 대기 중에도 루프 처리를 수행하므로 불필요하게 CPU 리소스를 소비하게 될 가능성이 있다. 그래서 이번엔 Pthreads의 조건 변수를 이용해 배리어 동기를 수행하는 법을 다룬다. 다음 코드는 Pthreads를 이용해 배리어 동기를 구현한 예다.

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

pthread_mutex_t barrier_mut = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t barrier_cond = PTHREAD_COND_INITIALIZER;

void barrier(volatile int *cnt, int max) {
    if (pthread_mutex_lock(&barrier_mut) != 0) {
        perror("pthread_mutex_lock"); exit(-1);
    }

    (*cnt)++; // 1

    if (*cnt == max) { // 2
        // 모든 프로세스가 모였으므로 알림, 3
        if (pthread_cond_broadcast(&barrier_cond) != 0) {
            perror("pthread_cond_broadcast"); exit(-1);
        }
    } else {
        do { // 모든 프로세스가 모일 때까지 대기, 4
            if (pthread_cond_wait(&barrier_cond, &barrier_mut) != 0) {
                perror("pthread_cond_wait"); exit(-1);
            }
        } while (*cnt < max); // 의사 각성을 위한 조건
    }

    if (pthread_mutex_unlock(&barrier_mut) != 0) {
        perror("pthread_mutex_unlock"); exit(-1);
    }
}

1. 락을 획득하고 공유 변수 *cnt를 증가한다.
2. *cnt가 max와 같은지 확인한다.
3. 값이 같으면 pthread_cond_broadcast를 호출하고, 조건 변수 barrier_cond로 대기 중인 스레드를 모두 실행한다.
4. 값이 같지 않으면 pthread_cond_wait를 호출하고 대기한다.

Pthreads를 이용한 배리어 동기는 스핀락 버전과 비교했을 때 다소 복잡하고 *cnt가 max가 될 때 까지 대기한다는 차이가 있지만 나머지는 비슷함을 알 수 있다.

3.7 Readers-Writer 락

레이스 컨디션이 발생하는 원인은 쓰기 때문이며, 쓰기만 배타적으로 수행하면 문제가 발생하지 않는다. 뮤텍스와 세마포어에서는 프로세스에 특별한 역할을 설정하지 않았지만 Readers-Writer락(RW락) 에서는 읽기만 수행하는 프로세스(Reader)와 쓰기만 수행하는 프로세스(Writer)로 분류하고 다음 제약을 만족하도록 배타 제어를 수행한다.

• 락을 획득 중인 Reader는 같은 시각에 다수(0 이상) 존재할 수 있다.
• 락을 획득 중인 Writer는 같은 시각에 1개만 존재할 수 있다.
• Reader와 Writer는 같은 시각에 락 획득 상태가 될 수 없다.

3.7.1 스핀락 기반 RW락

다음 코드는 스핀락 기반 RW락 알고리즘이다. 이 알고리즘은 Readers 수를 나타내는 변수 rcnt(초깃값 0), Writer 수를 나타내는 변수 wcnt(초깃값 0), Writer용 락 변수 lock(초깃값 false)의 3개 공유 변수를 이용해 배타 제어를 수행한다. 또한 Reader용 락 획득과 반환 함수, Writer용 락 획득과 반환 함수는 별도의 인터페이스로 되어 있어 실제로 이용할 때는 공유 리소스의 읽기만 수행할지 쓰기만 수행할 지 적절하게 판단해서 이용해야 한다.

// Reader용 락 획득 함수, 1
void rwlock_read_acquire(int *rcnt, volatile int *wcnt) {
    for(;;) {
        while (*wcnt); // Writer가 있으면 대기, 2
        __sync_fetch_and_add(rcnt, 1); // 3
        if (*wcnt == 0) // Writer가 없으면 락 획득, 4
            break;
        __sync_fetch_and_sub(rcnt, 1);
    }
}

// Reader용 락 반환 함수, 5
void rwlock_read_release(int *rcnt) {
    __sync_fetch_and_sub(rcnt, 1);
}

// Writer용 락 획득 함수, 6
void rwlock_write_acquire(bool *lock, volatile int *rcnt, int *wcnt) {
    __sync_fetch_and_add(wcnt, 1); // 7
    while (*rcnt); // Reader가 있으면 대기
    spinlock_acquire(lock); // 8
}

// Writer용 락 반환 함수, 9
void rwlock_write_release(bool *lock, int *wcnt) {
    spinlock_release(lock);
    __sync_fetch_and_sub(wcnt, 1);
}

1. Reader용 락 획득 함수를 정의한다. 2개의 공유 변수 포인터 rnct와 wnct를 인수로 받는다. rnct와 wnct는 각각 Reader와 Writer 수를 나타내는 공유 변수에 대한 포인터다.
2. *wnct의 값이 0보다 크면 스핀해서 대기한다. wcnt는 락을 얻은(또는 얻으려고 시도하는) Writer 수를 나타내므로 이 수가 0인 경우에만 Reader가 락을 획득할 수 있도록 설계했다.
3. Reader 수를 증가한다.
4. 다시 *wcnt의 값이 0인지 체크한다. 0이면 락을 획득하지만 그렇지 않으면 *rcnt 값을 아토믹하게 감소하고 재시도한다. 다시 *wcnt의 값을 확인하는 이유는 *rcnt를 증가하는 도중에 *wcnt 값이 증가될 가능성이 있기 때문이다.
5. Reader용 락 반환 함수. 단순히 Reader의 수를 감소한다.
6. Writer용 로그 획득 함수. Reader와 Writer의 수를 나타내는 포인트 변수 rcnt, wcnt에 더해 Writer용 로그 변수로서의 포인터인 lock 변수를 인수로 받는다.
7. Writer 수를 증가하고, Reader가 없어질 때까지 대기한다.
8. 뮤텍스용 함수를 이용해 락을 획득한다. 뮤텍스를 이용하므로 동시에 락을 획득할 수 있는 Writer 수를 최대 1개로 제한하는 것이 된다.
9. Writer용 락 해제 함수, 뮤텍스의 락을 해제하고 Writer 수를 감소시킨다.

RW락을 이용해야 할 상황은 대부분 읽기 처리이며, 쓰기는 거의 일어나지 않을 것이다. 여기에서 소개한 알고리즘은 Writer를 우선하도록 설정되어 있으므로 그런 상황에서는 잘 작동하지만 쓰기가 빈번하게 일어난다면 읽기를 전혀 실행하지 못하게 되므로 주의해야 한다. 쓰기도 많이 수행되는 처리인 경우에는 뮤텍스를 이용하는 편이 실행 속도와 안정성 측면에서 좋다.

다음 코드는 RW락 이용 예다.

// 공유 변수
int rcnt = 0;
int wcnt = 0;
bool lock = false;

void reader() { // Reader 용 함수
    for (;;) {
        rwlock_read_acquire(&rcnt, &wcnt);
        // 크리티컬 섹션(읽기만)
        rwlock_read_release(&rcnt);
    }
}

void writer() { // Writer 용 함수
    for (;;) {
        rwlock_write_acquire(&lock, &rcnt, &wcnt);
        // 크리티컬 섹션(읽기 및 쓰기)
        rwlock_write_release(&lock, &rcnt);
    }
}

사용 방법은 뮤텍스와 거의 동일하지만 구현할 때는 읽기만의 처리인지 또는 쓰기도 수행하는 처리인지 올바르게 파악해야 한다.

참고자료

<동시성 프로그래밍(Concurrent Programming> 다카노 유키 저, 한빛미디어