Definition and Routines
Issue: Spin-based Approach
- Parent Waiting For Child: 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); // create child while (done == 0) ; // spin printf("parent: end\n"); return 0; }
main()에서 child process를 만들고 spin을 이용해 done이 1이 되길 기다린다. chid에서 할 일을 끝내고 done을 1로 만들면 흐름이 다시 main() 함수로 돌아온다.
전역변수 done을 이용한다. 하지만 spin으로 기다리기 때문에 비용이 많이 든다.
condition variable
condition variable은 스레드가 들어가서 실행(condition)이 끝나길 기다리는 queue이다.
condition variable이 객체라면 객체의 상태를 나타내는 것이 큐, 상태를 변화시키는 방법이 wait(sleep)와 signal(wake up).
Parent Waiting For Child: Use A Condition Variable
int done = 0; pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER; pthread_cond_t c = PTHREAD_COND_INITIALIZER; 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(int argc, char *argv[]) { printf("parent: begin\n"); pthread_t p; Pthread_create(&p, NULL, child, NULL); thr_join(); printf("parent: end\n"); return 0; }구조체
pthread_cond_t를 이용한다.먼저 main() 함수에서 스레드를 만들고 child() 함수를 호출한다. 이후 main() 함수는 thr_join()으로 기다리고 있는데, 이 함수에서 lock을 걸고 done이 1이 될 때까지 atomic하게 Pthread_cond_wait()를 호출한다. Pthread_cond_wait()로 main() 스레드는 spin하지 않고 sleep 상태로 기다린다.
child() 함수에서 printf()를 호출하고 thr_exit() 함수를 호출한다. child()가 호출한 thr_exit() 함수에서 lock을 걸고 done을 1로 만든 뒤에 atomic하게 Pthread_cond_signal(&c) 함수를 호출한다.
Pthread_cond_signal(&c)는 구조체 pthread_cond_t의 변수 c의 queue에서 sleep 중인 main()을 깨운다.
condition variable을 이용하면 spin하지 않고 대기할 수 있다.
만약 child 스레드가 먼저 done을 1로 만들고 signal()을 보내면 어떻게 될까?
signal로 깨울 스레드가 없고, 이후 메인 스레드는 done이 1이므로 wait()을 호출하지 않아서 결과는 같다.
이 경우 잘 동작한다.
만약 done을 쓰지 않으면 어떻게 될까?
void thr_exit() { Pthread_mutex_lock(&m); Pthread_cond_signal(&c); Pthread_mutex_unlock(&m); } void thr_join() { Pthread_mutex_lock(&m); Pthread_cond_wait(&c, &m); Pthread_mutex_unlock(&m); }child 스레가 exit()을 먼저 호출해서 signal()을 보낸 뒤, main 스레드가 wait을 할 수 있다.
wake up 할 방법없이 sleep해버린다.
만약 lock이 없다면 어떻게 될까?
void thr_exit() { done = 1; Pthread_cond_signal(&c); } void thr_join() { if (done == 0) Pthread_cond_wait(&c); }join()을 먼저 호출해서 done이 0인 것을 확인했다. 이후 wait() 함수를 호출하지 직전 interrupt가 발생해서 child 스레드가 exit()을 호출했다고 하자. 그러면 child가 done을 1로 바꾸고 signal을 보낸다. main() 스레드로 context switching을 하면 main 스레드는 wake할 방법 없이 sleep 해버린다.
→ race condition으로 인해 sleep 하는 스레드를 깨울 수 없게 된다.
- hold the lock when calling signal or wait
wait()는 반드시 lock을 걸고 호출 해야 한다. 이후 lock을 풀고 sleep 해야 한다. 마지막으로 return 하기 직전에 lock을 다시 얻어야 한다.
- condition check
state variable을 if와 while로 check.
(위 코드의 done을 if or while로 check하고 나서 wait() or signal() 할 때 lock() 걸기 )
The Producer/Consumer Problem
Producer/Consumer(Bounded Buffer) Problem이란 제한된 버퍼에 입력과 출력이 반복될 때 생길 수 있는 문제를 뜻한다.
- ex)
grep foo file.txt | wc -l: grep 프로세스는 producer로써 버퍼를 넣어주고, wc 프로세스가 consumer로서 버퍼를 처리한다.
먼저 잘못 설계한 p/c 코드이다.
The Put And Get Routines (Version 1: without if and while)
int buffer; int count = 0; // initially, empty void put(int value) { assert(count == 0); count = 1; buffer = value; } int get() { assert(count == 1); count = 0; return buffer; }count는 버퍼가 비어있으면 0, 버퍼가 있으면 1로 표시한다.
put()은 버퍼가 비어있는지 확인하고 버퍼를 넣는다.
get()은 버퍼가 채워져 있는지 확인하고 버퍼를 꺼낸다.
Producer/Consumer Threads (Version 1: with while)
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 tmp = get(); printf("%d\n", tmp); } }
producer()에서 count를 체크하지 않고 put()을 호출했기 때문에 assert()에 걸린다.
consumer()에서도 count를 체크하지 않고 get()을 호출했기 때문에 잘못된 코드이다.
이런 문제는 condition variable을 이용해 해결할 수 있다.
A Broken Solution (with condition variable)
아래는 if를 이용해 count를 확인하는 코드다.
Producer/Consumer: Single CV And If Statement
// int loops; // must init.. cond_t cond; mutex_t mutex; void *producer(void *arg) { int i; 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; 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); } }producer는 put() 하기 전에 count가 0인지 확인하고 만약 1이면 기다린다. 이후 큐를 기다리는 스레드가 있을 수도 있으니 signal을 보낸다.
consumer는 get() 하기 전에 count가 1인지 확인하고 만약 0이면 기다린다. 이후 빈자리를 기다리는 스레드가 있을 수도 있으니 signal을 보낸다.
if만 사용해서 state를 체크했으며, put()과 get()을 수행하는 condition variable이 하나이다.
Trace: Single CV with If Statement
그러나 문제가 발생한다.
먼저 Tc1이 c3까지 실행해서 wait()를 이용해 sleep 중이다.
이후 Tp가 signal()을 이용해 sleep 중이던 Tc1을 깨우고 wait()으로 sleep한다.
그런데 다음으로 Tc2가 스케쥴링 되면서 get()으로 버퍼를 가져간다.
Tc1이 스케줄링 되었을 때 get()을 호출했지만, 가져갈 버퍼가 남아있지 않다.
→ race condition으로 인한 assert() 발생
Better, But Still Broken: While, Not If
if 대신 while을 사용해보자.
Producer/Consumer: Single CV And While
cond_t cond; mutex_t mutex; void *producer(void *arg) { int i; for (i = 0; i < loops; i++) { Pthread_mutex_lock(&mutex); // p1 while (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; for (i = 0; i < loops; i++) { Pthread_mutex_lock(&mutex); // c1 while (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); } }
Trace: Thread Trace: Broken Solution (Version 2)
Tc1과 Tc2가 consume할 버퍼가 없어서 둘 다 wait()으로 sleep한다.
이후 Tp가 wake해서 signal을 보낸다.(Tc1이 signal을 받음) 이후 Tp가 버퍼를 하나 만들고 wait()로 sleep한다.
signal을 받아 Ready 상태였던 Tc1이 스케줄 되서 get()을 한 뒤 while()을 돌아 signal을 보내고 다시 wait()로 sleep한다.
Tc1의 signal을 받은 Tc2가 일어났지만 count가 0(버퍼에 아무것도 없음)이어서 다시 sleep한다.
이제 모두 sleep해서 아무도 signal을 보내거나 받을 수 없게 된다.
하지만 signal을 다른 스레드가 받아서 잘못 깨어나면 모든 스레드가 sleep하는 문제가 발생한다.
이런 문제는 condition variable을 두 개 사용해서 해결할 수 있다.
The Single Buffer Producer/Consumer Solution (with Empty and Fill)
Producer와 Consumer가 같은 condition variable(Queue)에서 기다리면 signal을 받는 스레드가 엉키는 문제가 발생했다. 따라서 Producer와 Consumer가 기다리는 큐를 분리해서 구현할 수 있다.
Producer가 버퍼를 넣으려 할 때 꽉 차 있다면 Empty 큐에 들어가서 빈자리가 생기길 기다린다.
Consumer가 버퍼를 빼려 할 때 비어 있다면 fill 큐에 들어가서 버퍼가 생기길 기다린다.
Producer/Consumer: Two CVs And While(empty and fill)
empty, fill; mutex_t mutex; void *producer(void *arg) { int i; for (i = 0; i < loops; i++) { Pthread_mutex_lock(&mutex); while (count == 1) Pthread_cond_wait(&empty, &mutex); put(i); Pthread_cond_signal(&fill); Pthread_mutex_unlock(&mutex); } } void *consumer(void *arg) { int i; 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); } }- wait() with empty : producer는 빈 변수를 기다린다.
- wait() with fill : consumer는 채워지길 기다린다.
Trace: Two condition variable - empty and fill
Tp가 p1~p6을 수행하면서 버퍼를 하나 만들고, 다시 p1~p3를 수행할 때 빈자리가 없어서 sleep에 들어간다. 이때 Tp는 empty이라는 condition variable에 들어가서 sleep한다.
이후 스케줄된 Tc1이 버퍼를 가져오고
Pthread_cond_signal(&empty)로 empty에게 signal을 보낸다. 그러면 empty에서 기다리던 Tp이 깨어난다.
이렇게 condition variable을 두개 사용하면 정상적으로 동작한다.
The Final Producer/Consumer Solution
이제는 concurrency와 efficiency를 극대화 하기 위해 buffer slot을 늘려보자.
The Final Put And Get Routines
int buffer[MAX]; int fill_ptr = 0; int use_ptr = 0; int count = 0; void put(int value) { buffer[fill_ptr] = value; fill_ptr = (fill_ptr + 1) % MAX; count++; } int get() { int tmp = buffer[use_ptr]; use_ptr = (use_ptr + 1) % MAX; count--; return tmp; }modular MAX를 이용해 circular linked list처럼 구현했다. count는 사용중이 버퍼의 수를 나타낸다.
fill_ptr은 맨 뒤를 가리키고 있고, put()을 이용해 fill_ptr에 value를 넣으면 fill_ptr은 그 다음 주소를 가리킨다.
get()은 use_ptr이 가리키는 값을 가져오고 use_ptr를 증가시킨다.
The Final Working Solution
cond_t empty, fill; mutex_t mutex; void *producer(void *arg) { int i; for (i = 0; i < loops; i++) { Pthread_mutex_lock(&mutex); while (count == 1) Pthread_cond_wait(&empty, &mutex); put(i); Pthread_cond_signal(&fill); Pthread_mutex_unlock(&mutex); } } void *consumer(void *arg) { int i; 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); } }
Covering Conditions
앞에서 producer와 consumer가 구분 없이 signal을 보내는 문제가 있어서, condition variable을 fill과 empty로 분리했다.
이 방법 대신 broadcast로 모두에게 signal을 보내는 방법이 있다.(broadcast를 해야 하는 경우가 있다.)
Covering Conditions: An Example
// how many bytes of the heap are free? int bytesLeft = MAX_HEAP_SIZE; // need lock and condition too cond_t c; mutex_t m; void * allocate(int size) { Pthread_mutex_lock(&m); while (bytesLeft < size) Pthread_cond_wait(&c, &m); void *ptr = ...; // get mem from 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); // whom to signal?? Pthread_mutex_unlock(&m); }
위 코드는 memory allocate()와 free()를 하는 예시이다.
Trace: condition variable을 하나만 사용하는 경우
Ta가 alloc(100), Tb가 alloc(10)을 호출했지만 남은 공간이 없어서 모두 sleep했다. 이후 Tc가 free(50)으로 메모리를 확보한 뒤 Ta에게 signal을 보냈다. 하지만 alloc(100)을 하기에는 여전히 size가 부족해서 모두 sleep한다.
이런 문제를 해결하기 위해 제시된 방법이 pthread_cond_signal() 대신 sleep 중인 모든 스레드를 깨우는 pthread_cond_broadcast() 함수를 사용하는 것이다.
물론 효율은 떨어진다.
Uploaded by Notion2Tistory v1.1.0