15. 동시성 제어의 고전적 문제들: Chapter 7. Synchronization Examples (Part 1)

7.1 고전적인 동기화 문제들

• 많은 클래스의 병행 제어 문제에 대한 예로서 중요한 여러가지 다른 동기화 문제 제시
  1. 유한 버퍼 문제
  2. Readers-Writers 문제
  3. 식사하는 철학자 문제

 

 

유한 버퍼 문제

• 6.1절의 유한 버퍼 환경의 생산자-소비자 문제 참고 (https://coding-rocker.tistory.com/133)

11. 프로세스 동기화: Chapter 6. Synchronization Tools (Part 1)

• 동기화 문제 - 유한버퍼문제(생산자 소비자 문제)
  • 생산자producer는 생산할 공간이 있을 경우에만 버퍼에 접근할 수 있어야 함
  • 소비자consumer는 소비할 아이템이 있는 경우에만 버퍼에 접근할 수 있어야 함
• 문제 해결
  • n개의 버퍼로 구성된 풀pool이 있다고 가정
  • 각 버퍼는 한개의 아이템을 저장할 수 있다고 가정
  • mutex 이진 세마포가 있다고 가정
    • 버퍼 풀에 접근하기 위한 상호 배제 기능을 제공
    • 1으로 초기화
  • empty 세마포가 있다고 가정
    • 비어있는 버퍼의 수를 기록
    • n으로 초기화
  • full 세마포가 있다고 가정
    • 꽉 찬 버퍼의 수를 기록
    • 0으로 초기화

int n;
semaphore mutex = 1;
semaphore empty = n;
semaphore full = 0;

// 생산자 코드
while(true){
  /* 버퍼에 저장될 다음 아이템 생산 */
  ...
  wait(empty); // 소비자가 버퍼를 다 비울 때 까지 기다림
  wait(mutex); // 여러 생산자 중 한 개의 생산자만 버퍼에 접근할 수 있도록 동기화 해 줌
  ...
  /* 다음 아이템을 버퍼에 추가 */
  ...
  signal(mutex); // 다음 생산자가 버퍼에 접근할 수 있도록 함
  signal(full); // 소비자에게 꽉 찬 버퍼가 생겼음을 알림
}

// 소비자 코드
while(true){
  wait(empty); // 생산자가 버퍼를 다 채울 때 까지 기다림
  wait(mutex); // 여러 소비자 중 한 개의 소비자 버퍼에 접근할 수 있도록 동기화 해 줌
  ...
  /* 버퍼로부터 다음 아이템을 삭제 */
  ...
  signal(mutex); // 다음 소비자 버퍼에 접근할 수 있도록 함
  signal(full); // 생산자에게 비어있는 버퍼가 생겼음을 알림
  ...
  /* 다음 아이템을 소비 */
}

• 실습 코드

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

#define true 1
#define BUFFER_SIZE 1

// 공유 버퍼
int buffer[BUFFER_SIZE];

pthread_mutex_t mutex;
sem_t empty, full;

int in = 0, out = 0;

void insert_item(int item){
    sem_wait(&empty); // wait
    pthread_mutex_lock(&mutex);

    // ciritcal section
    buffer[in] = item;
    in = (in + 1) % BUFFER_SIZE;
    printf("Producer: inserted $%d\n", item);

    pthread_mutex_unlock(&mutex);
    sem_post(&full); // signal
}

void remove_item(int *item){
    sem_wait(&full); // wait
    pthread_mutex_lock(&mutex);

    *item = buffer[out];
    out = (out + 1) % BUFFER_SIZE;
    printf("Consumer: removed $%d\n", *item);

    pthread_mutex_unlock(&mutex);
    sem_post(&empty); // signal
}

void *producer(void *param){
    int item;
    while (true){
        // 0.1초 ~ 0.5초 사이를 슬립하게 해줌
        usleep((1+rand()%5)*100000);
        // 1000 ~ 2000까지의 수를 랜덤하게 만들어줌
        item = 1000 + rand()%1000;
        // 랜덤값을 버퍼에 집어 넣음
        insert_item(item); // critical section
    }
}

void *consumer(void *param){
    int item;
    while (true){
        // 0.1초 ~ 0.5초 사이를 슬립하게 해줌
        usleep((1+rand()%5)*100000);
        // item을 하나 꺼내옴
        remove_item(&item);  // critical section
    }
}

int main(int argc, char const *argv[])
{
    int i, numOfProducers = 1, numOfConsumers = 5;
    pthread_t tid;

    // 뮤텍스 초기화
    pthread_mutex_init(&mutex, NULL);
    // 버퍼 사이즈로 empty 초기화
    sem_init(&empty, 0, BUFFER_SIZE);
    // full 초기화
    sem_init(&full, 0, 0);

    for (int i=0; i<numOfProducers; i++){
        pthread_create(&tid, NULL, producer, NULL);
    }
    for (int i=0; i<numOfConsumers; i++){
        pthread_create(&tid, NULL, consumer, NULL);
    }
    
    sleep(5);
    return 0;
}

 

 

Readers-Writers 문제

• 하나의 데이터베이스가 다수의 병행 프로세스간에 공유된다고 가정
• 동기화 문제(Readers-Writers 문제)
  • reader: 데이터베이스의 내용을 읽기만 하는 프로세스
  • writer: 데이터 베이스의 내용을 갱신(읽고 쓰기)하는 프로세스 
  • 하나의 Writer와 어떤 다른 스레드(reader 또는 writer)가 동시에 데이터베이스에 접근하면 혼란이 야기될 수 있음
• 첫 번째 Readers-Writers 문제
  • writer가 공유 객체를 사용할 수 있는 허가를 아직 얻지 못했다면, 어느 reader도 기다리게 해서는 안됨
  • writer가 기아starvation 상태에 빠질 수 있음 
• 두 번째 Readers-Writers 문제
  • writer가 준비 되면 가능한 빨리 쓰기를 수행해야함
  • reader가 기아starvation 상태에 빠질 수 있음
• 첫 번째 Readers-Writers 문제 해결
  • read_count가 있다고 가정
    • 현재 몇 개의 프로세스들이 객체를 읽고 있는지 알려줌
    • 0으로 초기화
  • rw_mutex 이진 세마포가 있다고 가정
    • writer들을 위한 상호 배제 세마포
    • 첫 번째 reader와 임계구역을 빠져나오는 마지막 reader에 의해서도 사용됨
    • reader와 writer가 모두 공유
    • 1으로 초기화
  • mutex 세마포가 있다고 가정
    
    • read_count를 갱신할 때 상호 배제를 보장하기 위해 사용
    • 1으로 초기화

semaphore rw_mutex = 1;
semaphore mutex = 1;
int read_count = 0;

// writer 프로세스 코드

while(true){
  wait(rw_mutex); // reader 혹은 writer가 signal(rw_mutex)를 수행할 때 까지 대기
  ...
  /* 쓰기 작업 수행 */
  ...
  signal(rw_mutex); // signal(rw_mutex)를 수행하여 대기중인 관련 큐에 있는 reader, writer 프로세스들을 깨움
}

while(true){
  // n개의 reader가 기다리고 있다고 할 때
  // wait(mutext)에 의해 n-1개의 reader는 mutex 관련 큐에 삽입됨
  wait(mutex); // critical section인 read_count에 진입하기 위해 동기화
  read_count++;
  if (read_count == 1) 
    wait(rw_mutex); // 한 개의 reader(첫번째로 임계구역에 진입하는 reader 프로세스)가 rw_mutex관련 큐에 진입했다가 writer가 signal(rw_mutex)를 수행하면 깨어남
  signal(mutex); // critical section인 read_count에 진입해도 됨을 알림
  ...
  /* 읽기 작업 수행 */
  ...
  wait(mutex); // c
  read_count--; // critical section인 read_coun에 진입하기 위해 동기화
  if (read_count == 0)
    signal(rw_mutex); // 한 개의 reader(마지막으로 임계구역에 진입하는 reader 프로세스)가 rw_mutex관련 큐에 있는 writer 프로세스들을 깨움
  signal(mutex); // critical section인 read_count에 진입해도 됨을 알림
}

• CashBox를 이용한 Reader-Writer 실습 코드

package com.os_study;

public class BoundedBuffer {
    public static void main(String[] args) {
        CashBox cashBox = new CashBox(1);
        Thread[] producers = new Thread[5];
        Thread[] consumers = new Thread[5];

        for (int i = 0; i < producers.length; i++) {
            producers[i] = new Thread(new ProdRunner(cashBox));
            producers[i].start();
        }
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = new Thread(new ConsRunner(cashBox));
            consumers[i].start();
        }
    }

    static class CashBox {
        private int[] buffer;
        private int count, in, out;

        public CashBox(int buffersize){
            buffer = new int[buffersize];
            count = in = out = 0;
        }
        synchronized public void give(int money) throws InterruptedException {
            while (count == buffer.length) {
                try {
                    wait();
                } catch (InterruptedException e) {
                }
            }
            buffer[in] = money;
            in = (in+1)% buffer.length;
            count++;
            System.out.printf("여기있다, 용돈: %d\n", money);
            notify();
        }
        synchronized public int take() throws InterruptedException {
            while(count == 0){
                try {
                    wait();
                }
                catch (InterruptedException e){}
            }
            int money = buffer[out];
            out = (out+1)%buffer.length;
            count--;
            System.out.printf("고마워요, 용돈: %d원\n", money);

            notify();
            return money;
        }

    }

    private static class ProdRunner implements Runnable {
        CashBox cashBox;
        public ProdRunner(CashBox cashBox) {
            this.cashBox = cashBox;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    Thread.sleep((long) (Math.random() * 500));
                    int money = ((int) (1 + Math.random() * 9)) * 10000;
                    cashBox.give(money);
                }
            } catch (InterruptedException e) {
            }
        }
    }

    private static class ConsRunner implements Runnable {
        CashBox cashBox;

        public ConsRunner(CashBox cashBox) {
            this.cashBox = cashBox;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    Thread.sleep((long) (Math.random() * 500));
                    int money = cashBox.take();
                    cashBox.give(money);
                }
            } catch (InterruptedException e) {}
        }
    }
}

CashBox = 1, Producer = 1, Consumer = 1일 때 producer와 consumer가 순서대로 동작하는 것을 확인

CashBox = 5, Producer = 5, Consumer = 5일 때 producer와 consumer가 순서대로 동작하지 않고 reader나 writer가 임의의 순서대로 동작하는 경우가 있는 것을 확인

• SharedDB를 이용한 Reader-Writer 실습 코드

package com.os_study;

public class ReadersCounters {
    static class SharedDB{
        private int readerCount = 0;
        private boolean isWriting = false;

        public void read(){
            // read from the database here.
        }

        public void write(){
            // write into database here.
        }

        synchronized public void acquiredReadLock(){
            while(isWriting == true){
                try {
                    wait();
                } catch (InterruptedException e) {}
            }
            readerCount++;
        }

        synchronized public void releasedReadLock(){
            readerCount--;
            if (readerCount == 0){
                notify();
            }
        }

        synchronized public void acquiredWriteLock(){
            while(readerCount > 0 || isWriting == true){
                try {
                    wait();
                } catch (InterruptedException e) {}
            }
            isWriting = true;
        }

        synchronized public void releasedWriteLock(){
            isWriting = false;
            notifyAll();
        }
    }

    public static void main(String[] args) {
        SharedDB sharedDB = new SharedDB();
        sharedDB.acquiredReadLock();
        sharedDB.read();
        sharedDB.releasedReadLock();

        sharedDB.acquiredWriteLock();
        sharedDB.write();
        sharedDB.releasedWriteLock();
    }

}