[전공생이 설명하는 OS] 동기화 - (4) 생산자 소비자 문제

Reviews

• 쓰레드(Thread)와 동기화 문제
• 동기화 - (1) 용어 및 개념정리
• 동기화 - (2) 상호배제 전략(Mutex)
• 동기화 - (3) 조건동기화 (pthread 예시)

 여태까지, 멀티프로세스 또는 멀티쓰레드 환경에서 공유자원을 사용하다보면 race condition 이 발생할 수 있고 이를 해결하기 위해서는 동기화가 필요함을 설명했다. 동기화에는 상호배제(Mutual Exclusion) 전략과 조건 동기화(Condition Synchronization) 전략이 있다.

 

  상호배제는 공유자원을 한번에 하나씩 사용하는데 집중한다. 비슷하지만 조건동기화는 다수의 실행흐름을 제어하는데 더 유리하다. Critical Section에 대기하는 프로세스(또는 쓰레드)를 Ordering 하는 전략이다. 

 

   지난 포스팅까지는 조건동기화 관점에서 동기화의 의미를 파악하고, pthread 라이브러리를 이용해서 join() 함수를 간단하게 구현해보았다. 이번에는 조건동기화를 이용하면서 발생하는 문제들을 알아본다.

---

생산자-소비자 문제

  조건동기화에서는 대기큐를 사용한다. Critical Section 에 들어가기 전에, 조건을 만족하지 않는 모든 스레드들이 대기하는 장소이다. 이 대기큐의 길이가 한정되어 있다보니 발생하는 문제가 있다. 이를 Bounded Buffer problem 또는 Producer-Consumer problem 이라고 한다.

 

  가장 흔한 예시는 웹 서버를 예로 들 수 있다. 멀티 스레드 기반의 웹서버를 생각해보자. 웹서버는 다수의 요청을 처리해야한다. 따라서 요청을 받기 위해 대기(listen)하는 스레드가 다수 존재한다. 요청이 들어오면 해당 요청을 받은 쓰레드가 그 요청(httpRequest)을 대기큐에 요청을 쌓아둔다. 또 요청을 처리하기 위한 다수의 스레드가 존재해, 대기큐에 있는 요청을 하나씩 꺼내서 처리한다. 

 

  대기큐에 요청을 집어넣는 스레드를 생산자, 대기큐에서 요청을 꺼내 처리하는 스레드를 소비자라고 할 수 있다.

• 생산자 : 웹서버에서 요청을 받기 위해 대기(listen) 하면서, 요청이 발생하면 대기큐에 해당 요청을 집어넣는 스레드
• 소비자 : 요청을 처리하기 위해 기다리면서, 대기큐에 요청이 있으면 하나씩 꺼내는 스레드

  이런 상황이 문제가 되는 이유는 바로 대기큐가 공유자원이고 race condition이 발생할 수 있기 때문이다. 생산자는 대기큐가 가득 차 있는지 확인해야하고, 소비자는 대기큐에 요청이 하나라도 있는지 계속 확인해야한다. 다수의 소비자와 다수의 생산자가 존재하는 대규모 소프트웨어 환경에서는 빈번하게 발생하는 상황이다. 이 문제를 해결하기 위해서는 상호배제 측면의 동기화보다는 대기큐의 상태를 고려하는 조건 동기화 측면으로 접근해야한다. 

 

  이제부터는 condition variable 을 이용하여 생산자-소비자 문제를 해결하는 코드를 설명한다. pthread 라이브러리를 사용하는 버전과 semaphore 를 사용하는 버전이 있다.

 

---

 

 1) pthread 를 이용

  처음부터 완성된 솔루션을 보면 이해하기가 힘들기 때문에 점진적으로 설명하도록 한다.

(1) Infinite Buffer

  대기큐 (대기 버퍼) 의 크기가 무한일 때에는, put 할 때는 신경쓰지 않아도 된다. get() 할 때만 버퍼가 비어있는지 확인하면 된다. 이전 포스팅에서 설명했듯이, 조건 동기화를 위해서는 (1) 상태변수, (2) 상태변수에 대한 lock, (3) loop checking 이 필요하다. count 라는 상태변수를 선언하고, lock을 설정한 것을 볼 수 있다.

  다만 아래의 코드에서는 while 대신 if 를 사용했는데, loop checking 의 필요성을 추후에 설명하기 위함이다.

void producer() {
    while (1) {
        pthread_mutex_lock(&m);  // 대기큐c 에 대한 lock (상호배제)
        put(value);
        pthread_cond_signal(&c);    // 대기하는 consumer 가 있으면 깨운다.
        pthread_mutex_unlock(&m);// 대기큐c 에 대한 lock (상호배제)
    }
}
void consumer() {
    while (1) {
        pthread_mutex_lock(&m);    // 대기큐c 와 count 변수 에 대한 lock (상호배제)
        if (count == 0)
    		pthread_cond_wait(&c, &m); // 소비할 것이 없으면 대기큐 c에서 대기한다.
        get();
        pthread_mutex_unlock(&m);  // 대기큐c 와 count 변수 에 대한 lock (상호배제)
    }
}

 

(2) 크기가 고정된 버퍼

  버퍼의 크기가 고정된다면, 가득 찼을 때 더 넣을 수 없다. 따라서 producer에서 공유변수 count 가 MAX 인지 확인하는 코드를 추가해야한다.

  또 생산자가 대기하고 있고 누군가 소비해서 공간이 생겨난다면, 대기중인 생산자를 깨워야한다. 따라서 소비자가 소비한 이후에 생산자를 깨우는 코드를 추가해야한다. 

void producer() {
    while (1) {
/*p1*/    pthread_mutex_lock(&m);  // 대기큐c 와 count 변수 에 대한 lock (상호배제)
/*p2*/    if (count == MAX)              // 추가된 코드
/*p3*/        pthread_cond_wait(&c, &m);    // 추가된 코드
/*p4*/    put(value);
/*p5*/    pthread_cond_signal(&c);
/*p6*/    pthread_mutex_unlock(&m);// 대기큐c 와 count 변수 에 대한 lock (상호배제)
    }
}
void consumer() {
    while (1) {
/*c1*/    pthread_mutex_lock(&m);    // 대기큐c 와 count 변수 에 대한 lock (상호배제)
/*c2*/    if (count == 0)
/*c3*/        pthread_cond_wait(&c, &m);
/*c4*/    get();
/*c5*/    pthread_cond_signal(&c);       // 추가된 코드, 대기하는 생산자가 있으면 깨운다.
/*c6*/    pthread_mutex_unlock(&m);  // 대기큐c 와 count 변수 에 대한 lock (상호배제)
    }
}

 

   하지만 위의 코드는 두가지의 문제점이 있어서 수정해야한다.

    : 생산자 1명, 소비자 2명, 버퍼 크기가 1개라고 가정

    : 생산자가 소비자A를 깨웠는데, os 스케줄러에 의해 소비자B가 consume 하게 되면 발생. (signal 은 thread의 state를 변경시키는 것일 뿐, 누가 실행될지는 os가 실행가능한 상태의 thread들 중에서 결정한다.)

1. 버퍼가 비어있을 때 소비자A 가 소비 시도 -> c3 코드에서 wait()
2. 생산자가 1개를 생산하고 signal 을 보내서, 소비자 A가 ready 상태로 변경됨.
   (즉시 실행되지 않는데, 이것 때문에 문제가 발생한다.)
3. 생산자가 하나 더 생산하려는데 버퍼가 가득참 -> p3 에서 wait()
4. 소비자 A, 소비자 B 중에서 OS 가 소비자 B 를 실행시킴 -> 1개 소비 후에 버퍼가 비어서 wait()
5. 이제야 소비자 A 가 실행함. (1번)에서 c3 코드라인에서 대기했으니, c3 이후부터 실행한다. get() 을 호출하는데 버퍼가 비어있으므로 오류를 발생시킨다. 

  문제가 되는 근본적인 이유는 생산자가 소비자A 를 깨웠는데, OS 스케줄러에 의해 다음에 실행된 스레드는 소비자B 가 된 것이다. 새치기를 당한다. 이를 해결하는 법은 두가지가 있다.

• Mesa Sementics : 5번에서 깨어났을 때, 즉시 get() 하지 않고 다시 한번 버퍼(count)를 확인하는 방법 -> if 대신 while 을 사용한다.
• Hoare Sementics : 시그널을 통해 깨어난 스레드를, 즉시 실행시킨다. -> os 를 수정해야함.

  대부분의 os는 Mesa Sementic 으로 구현되어 있기 때문에 if 대신 무조건 while을 사용하는 것이 좋다. 따라서 위의 코드에서 if 를 while 로 바꾸면 된다.

    : 생산자 1명, 소비자 2명, 버퍼 크기가 1개라고 가정

    : 생산자와 소비자가 사용하는 condition variable이 같기 때문에 발생.

 

1. 소비자A 가 실행할 차례이고, 소비자B와 생산자가 wait() 으로 sleep 상태에 있다고 해보자.
2. 소비자A 가 성공적으로 버퍼에서 데이터를 소비한다. 생산자가 깨어날 것을 기대하고, 데이터를 소비했다는 시그널을 보낸다.
3. 소비자A 가 한번 더 수행하려다 버퍼가 비어있어서 sleep 한다.
4. condition variable 을 생산자와 소비자가 모두 같은 것을 사용하기 때문에, 해당 시그널을, 생산자가 아닌, 소비자B 가 받아서 깨어난다. => 소비자A 와 생산자는 sleep 상태
5. 소비자B 도 버퍼가 비어있어서 sleep 한다.
6. 결과적으로 모든 스레드가 sleep 하게 된다.

 

생산자와 소비자가 기다리는 이벤트는 엄연히 다르다.

• 생산자 - (기다림) 버퍼에 잔여공간이 생겨났다는 이벤트를 기다림 (empty 이벤트)
            - (발생) 버퍼에 새로운 데이터를 넣었다는 이벤트를 발생시킴 (fill 이벤트)
• 소비자 - (기다림) 버퍼에 소비가능한 데이터가 생겼다는 이벤트. (fill 이벤트)
            - (발생) 버퍼에서 데이터를 소비했다는 이벤트를 발생시킴 (empty 이벤트)

따라서 condition variable 을 구별해서 사용해야한다.

 

최종 코드는 다음과 같다.

void producer() {
    while (1) {
        pthread_mutex_lock(&m);  // 대기큐 empty,fill 와 count 변수에 대한 lock (상호배제)
        while (count == MAX)              
    		pthread_cond_wait(&empty, &m);    // 수정된 부분 (소비이벤트를 기다림)
        put(value);
        pthread_cond_signal(&fill);           // 수정된 부분 (생산이벤트 발생시킴)
        pthread_mutex_unlock(&m);// 대기큐 empty,fill 와 count 변수에 대한 lock (상호배제)
    }
}
void consumer() {
    while (1) {
        pthread_mutex_lock(&m);    // 대기큐 empty,fill 와 count 변수에 대한 lock (상호배제)
        while (count == 0)
        	pthread_cond_wait(&fill, &m);  // 수정된 부분 (생산이벤트 기다림)
        get();
        pthread_cond_signal(&empty);       // 수정된 부분 (소비이벤트 발생시킴)
        pthread_mutex_unlock(&m);  // 대기큐 empty,fill 와 count 변수에 대한 lock (상호배제)
    }
}

 

---

2) Semaphore 이용

  세마포는 상호배제와 조건동기화에 모두 적용할 수 있는 일반화된 동기화 방법이다. 상호배제의 측면으로 보면 가용한 자원의 수를 중심으로 장원은 guard 하는 역할이고, 조건동기화에서는 다수의 스레드를 조건에 따라 대기시키면서 줄을 서게하는 역할로 볼 수 있다. 

  이런 일반화 동기화 방법인 세마포는, 두가지의 의미가 혼용되어 사용될 때 코드를 해석하기가 힘들다. 아래의 코드에서도 상호배제의 의미와 조건동기화의 의미로 모두 세마포가 사용된다.

  세마포에서도 대기하는 wait(), 시그널 보내는 post() 함수가 존재한다. (P, V 함수라고도 이야기한다.) 

wait() 하면 조건변수를 1 감소시키고, post() 하면 조건변수를 1 증가시킨다. 조건변수가 0 이하면 대기하거나 시그널을 보낸다.

 

  위에서 pthread 라이브러리를 사용했던 것과 동일하게 세마포로 구현하면 아래와 같다.

/* 상호배제 */
sem_init(&m, 0, 1);    // critical section 에는 한 스레드만 들어갈 수 있다.

/* 조건 동기화 조건변수 */
sem_init(&empty, 0, MAX); // 처음에는 빈공간(생산할 수 있는 버퍼 남은 크기)이 MAX 이다.
sem_init(&fill, 0, 0);  // (소비할 수 있는 자원, 생산된 자원)는 처음에 아무것도 없다.

//[Producer]
void producer() {
    while (1) {                
        sem_wait(&m);          // 상호배제, 공유자원인 버퍼 guard
        sem_wait(&empty);
        put(value);
        sem_post(& fill);            // 생산했다는 이벤트 발행
        sem_post(&m);          // 상호배제, 공유자원인 버퍼 guard
    }
}

//[Consumer]
void consumer() {
    while (1) {
        sem_wait(&m);          // 상호배제, 공유자원인 버퍼 guard
        sem_wait(& fill);            // 생산된 것이 있어야 소비가능
        get();
        sem_post(& empty);           // 소비했다는 이벤트 발행
        sem_post(&m);          // 상호배제, 공유자원인 버퍼 guard
    }
}

  하지만 여기에 문제가 하나 존재하는데, 상호배제에 대한 lock 변수인 m을 잡고 wait 하게 되면 생산자와 소비자 모두 critical section 으로 진입하지 못하는 데드락 상황이 발생한다. pthread_wait(&c, &m) 에서는 내부적으로 m에 대한 lock 을 풀고 sleep 했기 때문에 문제가 되지 않았지만, 세마포는 general 한 버전이기 때문에 그런 작업을 지원하지 않는다. 따라서 lock 변수 m 을 잡고 sleep 하지 않도록 아래와 같이 수정해야한다

/* 상호배제 */
sem_init(&m, 0, 1);    // critical section 에는 한 스레드만 들어갈 수 있다.

/* 조건 동기화 조건변수 */
sem_init(&empty, 0, MAX); // 빈공간(생산할 수 있는 버퍼 남은 크기)이 MAX 이다.
sem_init(&fill, 0, 0);  // (소비할 수 있는 자원, 생산된 자원)는 처음에 아무것도 없다.

//[Producer]
void producer() {
    while (1) {                
        sem_wait(&empty);
        sem_wait(&m);          // 상호배제, 공유자원인 버퍼 guard
        put(value);
        sem_post(&m);          // 상호배제, 공유자원인 버퍼 guard
        sem_post(& fill);   
    }
}

//[Consumer]
void consumer() {
    while (1) {  
        sem_wait(& fill);  
        sem_wait(&m);          // 상호배제, 공유자원인 버퍼 guard
        get();
        sem_post(&m);          // 상호배제, 공유자원인 버퍼 guard
        sem_post(& empty); 
    }
}

---

마무리

  지금까지 생산자-소비자 문제를 다뤘다. 다수의 생산자와 다수의 소비자가 있을 때, 공유자원을 race condition 없이 접근하기 위해서는 조건동기화 기법을 사용해야한다. 생산자가 발생시키는 이벤트와, 소비자가 발생시키는 이벤트를 구분하고, 이벤트의 개수가 특정 조건에 맞을때만 공유자원에 접근하도록 한다. 이런 방법을 통해서 생산자, 소비자 상관없이 공유자원에는 한번에 하나의 스레드만 접근하도록 보장할 수 있다.

  

  pthread 와 semaphore 를 이용해서 버퍼를 동기화하는 예시 코드를 살펴보았다. state variable 이 음수가 되지 않으면서 상한이 존재하도록 하는 기법이다. 이런 코드는 구조화된 형태를 띄는데, 이런 특성 때문에 java 에서는 synchronize 라는 키워드만 붙이면 동기화가 가능하다. (컴파일러가 bytecode 를 전후로 붙여준다.) 

 

  공유자원을 수정하지 않고 read-only 할 때는, 동기화가 필요하지 않다. 이런 경우에 reader 들에 대해서도 모두 동기화를 걸어주는 것은 비효율적이다. 이를 해결하려는 문제를 Readers-Writers 라고 하며 다양한 알고리즘들이 존재한다. 여태까지 설명했던 조건동기화와 상호배제의 개념으로 해결할 수 있다.

 

  다음 포스팅에서는 동기화 부작용으로 발생할 수 있는 데드락에 대해서 알아보도록 하자.