#13. 생산자-소비자 문제

Producer-Consumer Problem

- 생산자-소비자 문제

  * 생산자가 데이터를 생산하면 소비자는 그것을 소비

  * 예: 컴파일러 > 어셈블러, 파일 서버 > 클라이언트, 웹 서버 > 웹 클라이언트

(소스 코드를 컴파일러가 컴파일(생산)해서 어셈블리어 코드를 만들어내고 이를 어셈블러가 기계어로 번역(소비)해서 기계어 코드를 만들어 낸다.)

 

- Bounded Buffer(유한 버퍼)

  * 생산된 데이터는 버퍼에 일단 저장 (속도 차이 등)

  * 현실 시스템에서 버퍼 크기는 유한

  * 생산자는 버퍼가 가득 차면 더 넣을 수 없다.

  * 소비자는 버퍼가 비면 뺼 수 없다.

 (생산자-소비자 문제의 제일 핵심은 생산자는 버퍼가 꽉차있으면 더이상 생산을 하지 않고,

서비자는 버퍼가 비어있으면 더이상 소비하지 않는다.)

class Buffer {
    int[] buf;
    int size;
    int count;
    int in;
    int out;
    
    
    Buffer(int size) {
        buf = new int[size];
        this.size = size;
        count = in = out = 0;
    }
    
    void insert(int item) {
        /* check if buf is full */
        while(count == size) { }
        /* buf is not full */
        buf[in] = item;
        in = (in + 1) % size;
        count++;
    }
    
    int remove() {
        /* check if buf is empty */
        while (count == 0) { }
        /* buf is not empty */
        int item = buf[out];
        out = (out + 1) % size;
        count--;
        return item;
    }
}

 

count : buf에 생산된 데이터의 갯수

size : buf의 크기

in : buf에 데이터가 in되는 위치(index)

out : buf에 데이터가 out되는 위치(index)

(이렇게 동작하면 circural하게 동작한다.)

class Producer extends Thread {
    Buffer buf;
    int sz;
    Producer(Buffer buf, int sz) {
        this.buf = buf;
        this.sz = sz;
    }
    
    @Override
    public void run() {
        for(int i = 0; i < sz; i++) {
            buf.insert(i);
        }
    }
}

class Consumer extends Thread {
    Buffer buf;
    int sz;
    Consumer(Buffer buf, int sz) {
        this.buf = buf;
        this.sz = sz;
    }
    
    @Override
    public void run() {
        int item = 0;
        for(int i = 0; i < sz; i++) {
            item = buf.remove();
        }
    }
}

class Test {
    public static void main(String[] args) {
        Buffer buf = new Buffer(100);
        Producer p = new Producer(buf, 10000);
        Consumer c = new Consumer(buf, 10000);
        p.start();
        c.start();
        try {
            p.join();
            c.join();
        } catch(InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Number of items in the buf is " + buf.count);
    }
}

 

 

- 잘못된 결과

  * 실행 불가, 또는 count = 0 (생산된 항목 숫자 = 소비된 항목 숫자)

  * 최종적으로 버퍼 내에는 0개의 항목이 있어야 함.

- 이유

  * 공통변수 count, buf[]에 대한 동시 업데이트

  * 공통변수 업데이트 구간(=임계구역)에 대한 동시 진입

- 해결법

  * 임계구역에 대한 동시 접근 방지(상호배타)

  * 세마포를 사용한 상호배타(mutual exclusion)

  * 세마포어: mutex.value = 1 

 

이를 세마포어를 이용하여 해결

import java.util.concurrent.Semaphore;

class Buffer {
    Semaphore mutex; // mutual exclusion
    int[] buf;
    int size;
    int count;
    int in;
    int out;
    
    
    Buffer(int size) {
        mutex = new Semaphore(1);
        buf = new int[size];
        this.size = size;
        count = in = out = 0;
    }
    
    void insert(int item) {
        /* check if buf is full */
        while(count == size) { }
        
        /* buf is not full */
        try {
            mutex.acquire();
            //////////////////////////////////////
            buf[in] = item;
            in = (in + 1) % size;
            count++;
            //////////////////////////////////////
            mutex.release();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    int remove() {
        /* check if buf is empty */
        while (count == 0) { }
        
        /* buf is not empty */
        try {
            mutex.acquire();
            //////////////////////////////////////
            int item = buf[out];
            out = (out + 1) % size;
            count--;
            //////////////////////////////////////
            mutex.release();
            return item;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return -1;
    }
}

실행결과 : Number of Items in the buf is 0

 

- Busy-wait

  * 생산자 : 버퍼가 가득 차면 기다려야 한다. = empty(빈) 공간이 있어야 한다.

  * 소비자 : 버퍼가 비어있으면 기다려야 한다. = full(찬) 공간이 있어야 한다.

- 세마포어를 사용한 busy-wait 회피

  * 생산자: empty.acquire(); // # of permit = BUF_SIZE

  * 소비자: full.acquire(); // # of permit = 0

[생산자]	[소비자]
empty.acquire(); PRODUCE; full.release();	full.acquire(); CONSUME; empty.release();

(버퍼가 가득차거나, 비어있을 때 생산자, 소비자의 대기 시간이 길어지는 현상 Busy-wait

따라서 cpu가 기다리는 동안 아무것도 못하고 무한 루프를 돌아야 하므로 이러한 낭비를 줄이기 위해(performance) semaphore 추가)

import java.util.concurrent.Semaphore;

class Buffer {
    Semaphore mutex; // mutual exclusion
    Semaphore empty, full;
    int[] buf;
    int size;
    int count;
    int in;
    int out;
    
    
    Buffer(int size) {
        mutex = new Semaphore(1);
        empty = new Semaphore(size);
        full = new Semaphore(0);
        buf = new int[size];
        this.size = size;
        count = in = out = 0;
    }
    
    void insert(int item) {
        try {
            empty.acquire();
            mutex.acquire();
            //////////////////////////////////////
            buf[in] = item;
            in = (in + 1) % size;
            count++;
            //////////////////////////////////////
            mutex.release();
            full.release();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    int remove() {
        try {
            full.acquire();
            mutex.acquire();
            //////////////////////////////////////
            int item = buf[out];
            out = (out + 1) % size;
            count--;
            //////////////////////////////////////
            mutex.release();
            empty.release();
            return item;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return -1;
    }
}