[sp] Synchronization: Advanced

**Semaphore review

Semaphore : s >= 0인 글로벌 정수 synchronization 변수, P와 V operation에 의해 조절

P(s) :

 - s != 0 -> s를 1 감소, 즉시 리턴

 - s == 0 -> s가 0이 아닐 때까지 스레드가 suspend => V operation에 의해 스레드가 wakeup

 - 스레드가 wakeup 후, P가 s를 감소시키고 caller에게 제어권을 넘겨줌

 

V(s) : 

 - s를 1 증가

 - P에서 s가 0이 아닐 때까지 기다리면서 block된 스레드가 있으면, 이 스레드 중 하나를 wakeup시켜주고, s를 감소시키면서 P를 완료

 

semaphore s는 무조건 0 이상

 

mutual exclusion을 통해 공유된 리소스를 보호하기 위해 semaphore를 사용하는 경우

 - basic idea : 1로 초기화된 semaphore인 mutex를 각 공유된 변수 ( 또는 공유된 변수와 관련된 집합)과 결합

  : P(mutex)와 V(mutex)가 있는 공유된 변수(s)로 각 access를 둘러쌈

 

mutex = 1
//mutex lock을 걸어줌
P(mutex)
cnt++
V(mutex)

 

1. Semaphore을 사용하여 공유된 리소스 (data structures)에 대한 access를 조정

 - basic idea : 스레드는 semaphore을 사용하여 몇몇 조건이 true가 되었음을 다른 스레드에게 알림

  - counting semaphore을 사용하여 리소스의 state를 유지하고 다른 스레드에게 알림

  - mutex를 사용하여 리소스에 대한 접근을 보호

 - examples : The producer-consumer problem, The Readers-Writes Problem

 

2. Producer-Consumer problem

producer thread : 생산, 버퍼를 채움, 만약 버퍼가 가득 차면 더이상 생산을 하지 않고 중단

consumer thread : 소비, 버퍼가 empty이면 consume 할 게 없으므로 중단

 

(1) Producer는 slot이 빌 때까지 wait, 버퍼에 item을 넣어주고 consumer에게 알림

(2) Consumer는 item을 기다리고, 버퍼로 부터 item을 remove, producer에게 알림

 

(3) ex

 - 멀티미디어 processing : 프로듀서는 MPEG 비디오 프레임을 생산하고 consumer에게 제공

 - 이벤트를 기반으로 한 graphical user interfaces : 프로듀서는 마우스의 클릭, 이동, 키보드 힛을 감지하고 버퍼에서 일치하는 이벤트에 넣음

 : consumer는 버퍼로 부터 이벤트를 찾고, 디스플레이에 paint

 

(4) Producer-Consumer on an n-element Buffer

 - mutex와 2개의 counting semaphore 필요

 : mutex - 버퍼에 대한 접근을 mutually exclusive하게 강요

 : slots - 버퍼의 가능한 slot을 셈, 즉 버퍼에서 남아있는 자리를 count, Producer의 입장에서 몇 개를 insert 할 수 있는지

 : items - 버퍼에서 가능한 item을 셈, 버퍼에 몇 개의 item이 들어와 있는지, Consumer 입장에서 몇 개를 소비할 수 있는 지

 

 - shared buffer package(sbuf)를 통해 구현

Declarations

#include "csapp.h”

typedef struct {
    int *buf; /* Buffer array */
    int n; /* Maximum number of slots */
    int front; /* buf[(front+1)%n] is first item */
    int rear; /* buf[rear%n] is last item */
    
    //binary semaphore
    sem_t mutex; /* Protects accesses to buf */
    
    //counting semaphore
    sem_t slots; /* Counts available slots */
    sem_t items; /* Counts available items */
} sbuf_t;

void sbuf_init(sbuf_t *sp, int n);
void sbuf_deinit(sbuf_t *sp);
void sbuf_insert(sbuf_t *sp, int item);
int sbuf_remove(sbuf_t *sp);

 

공유된 버퍼를 intializing 및 deinitializing

/* Create an empty, bounded, shared FIFO buffer with n slots */
void sbuf_init(sbuf_t *sp, int n)
{
    sp->buf = Calloc(n, sizeof(int));
    sp->n = n; /* Buffer holds max of n items */
    sp->front = sp->rear = 0; /* Empty buffer iff front == rear */
    Sem_init(&sp->mutex, 0, 1); /* Binary semaphore for locking */
    Sem_init(&sp->slots, 0, n); /* Initially, buf has n empty slots */
    Sem_init(&sp->items, 0, 0); /* Initially, buf has 0 items */
}

/* Clean up buffer sp */
void sbuf_deinit(sbuf_t *sp)
{
	Free(sp->buf);
}

초기 버퍼는 빈 버퍼이므로 front와 rear는 맨 처음을 같이 가르키고 있음

mutex는 처음에 1로 초기화

slots는 비어있는 버퍼의 자리를 의미하므로 n으로 초기화

item은 버퍼에 들어가 있는 item을 의미하므로 0으로 초기화

 

구현

공유된 버퍼에 아이템 삽입

/* Insert item onto the rear of shared buffer sp */
void sbuf_insert(sbuf_t *sp, int item)
{
    P(&sp->slots); /* Wait for available slot */
    //slot = 0이면 wait, 0이 아니면 감소하고 다음 코드 실행
    
    //mutex lock을 걸어줌
    P(&sp->mutex); /* Lock the buffer */
    sp->buf[(++sp->rear)%(sp->n)] = item; /* Insert the item */
    V(&sp->mutex); /* Unlock the buffer */
    //
    
    //block된 스레드를 깨워줌, item 증가시킴
    V(&sp->items); /* Announce available item */
}

 

공유된 버퍼로 부터 아이템 제거

/* Remove and return the first item from buffer sp */
int sbuf_remove(sbuf_t *sp)
{
    int item;
    P(&sp->items); /* Wait for available item */
    //item이 0이면 suspend, 0이 아니면 decrement하고 다음 코드 실행
    
    //mutex lock
    P(&sp->mutex); /* Lock the buffer */
    item = sp->buf[(++sp->front)%(sp->n)]; /* Remove the item */
    V(&sp->mutex); /* Unlock the buffer */
    
    //slot을 증가하고 slot이 없어서 wait하고 있는 스레드가 있으면 wakeup하고 작업 수행
    V(&sp->slots); /* Announce available slot */
    return item;
}

 

(5) Readers-Writers Problem

 - mutual exclusion problem의 일반화

 - Problem statement

 : Reader 스레드는 오직 object를 read만 함

 : Writer 스레드는 object를 수정

 : Writer는 반드시 object에 대해 exclusive access를 가짐

 -> 내가 write를 하는 중이면 read x, write x -> 즉, 혼자만 write할 수 있음

 : 무한한 reader는 object에 접근할 수 있어서 누구나 read할 수 있음

 

하지만 누군가가 read하고 있을 때 write는 wait

 

(6) Variants of Readers-Writers

- 문제1 : favors readrs ( reader에게 우선순위를 줌)

 -> writer가 object를 사용하기 위해 허가된 permission을 가질 때까지 reader는 wait하지 않음

 -> waiting writer 후에 도착한 reader는 writer보다 우선 순위를 가짐

즉 R->W->R일 때 W는 Wait하지만 W 뒤에 온 R은 object를 read할 수 있음

 

=> writer가 계속 wait하므로 starvation (스레드가 무한정으로 wait하는 상태) 발생

 

- 문제2 : favors writers ( writers에게 우선순위를 줌)

 -> writer가 읽을 준비가 되면, 가능한 write을 빠르게 완료

 -> writer가 wait하고 있는 중이라도 writer 후에 도착한 reader는 반드시 wait해야 함

 

=> reader가 계속 wait하므로 starvation 발생

 

- Solution to First Readers-Writers problem

//Readers
int readcnt; /* Initially = 0 */
sem_t mutex, w; /* Initially = 1 */
void reader(void)
{
    while (1) {
    P(&mutex);
    readcnt++;
    if (readcnt == 1) /* First in */
    	P(&w);
    V(&mutex);
    
    /* Critical section */
    /* Reading happens */
    
    P(&mutex);
    readcnt--;
    if (readcnt == 0) /* Last out */
    	V(&w);
    V(&mutex);
    }
}

//Writers
void writer(void)
{
    while (1) {
    	P(&w);
    /* Critical section */
    /* Writing happens */
        V(&w);
    }
}

 

 

(7) Putting it all together: Pthreaded Concurrent Server

 

쓰레드를 미리 생성해서 안쓰는건 sleep, 필요하면 깨움

 

- Prethreaded Concurrent Server

sbuf_t sbuf; /* Shared buffer of connected descriptors */
int main(int argc, char **argv)
{
    int i, listenfd, connfd;
    socklen_t clientlen;
    struct sockaddr_storage clientaddr;
    pthread_t tid;
    
    listenfd = Open_listenfd(argv[1]);
    sbuf_init(&sbuf, SBUFSIZE); 
    
    for (i = 0; i < NTHREADS; i++) /* Create worker threads */
    	Pthread_create(&tid, NULL, thread, NULL); 
        
    while (1) {
        clientlen = sizeof(struct sockaddr_storage);
        connfd = Accept(listenfd, (SA *) &clientaddr, &clientlen);
        sbuf_insert(&sbuf, connfd); /* Insert connfd in buffer */
    }
}

 

worker thread routine

void *thread(void *vargp)
{
    Pthread_detach(pthread_self()); //쓰레드 종료시 쓰레드 자원 정리
    while (1) {
        int connfd = sbuf_remove(&sbuf); /* Remove connfd from buf */ //connfd를 버퍼에서 꺼냄
        echo_cnt(connfd); /* Service client */
        Close(connfd);
    }
}

 

echo_cnt initialization routine

static int byte_cnt; /* Byte counter */
static sem_t mutex; /* and the mutex that protects it */

static void init_echo_cnt(void)
{
    Sem_init(&mutex, 0, 1);
    byte_cnt = 0;
}

 

worker thread service routine

void echo_cnt(int connfd)
{
    int n;
    char buf[MAXLINE];
    rio_t rio;
    static pthread_once_t once = PTHREAD_ONCE_INIT;
    
    Pthread_once(&once, init_echo_cnt); //쓰레드 초기화를 한 번만
    Rio_readinitb(&rio, connfd); 
    while((n = Rio_readlineb(&rio, buf, MAXLINE)) != 0) {
        P(&mutex);
        byte_cnt += n; //mutex lock으로 잡아둠
        printf("thread %d received %d (%d total) bytes on fd %d\n",(int) pthread_self(), n, byte_cnt, connfd); 
        V(&mutex);
        Rio_writen(connfd, buf, n);
    }
}