모니터와 조건 변수로 구성하는 스레드 동기화

공유 상태와 접근 절차를 한곳에 묶는 모니터

모니터는 공유 자원과 그 자원에 접근하는 절차를 하나의 추상 자료형으로 감싼 동기화 방식이다. 세마포어보다 높은 수준에서 상호배제와 조건 동기화를 다루며, 모니터 내부에는 한 번에 하나의 프로세스만 들어갈 수 있다.

공유 변수는 모니터 내부에서만 접근하고, 이를 조작하는 경로는 프로시저(메서드)로 제한된다. 조건 변수는 어떤 상태가 충족될 때까지 기다리는 큐를 제공한다. 이 구조 덕분에 동기화 규칙과 데이터가 분리되지 않는다.

⤢✕

세마포어는 호출자가 P/V 연산으로 상호배제를 직접 조정해야 한다. 반면 모니터는 진입 자체에 상호배제가 적용되고, 조건 대기를 조건 변수로 표현한다.

특성	Semaphore	Monitor
추상화 수준	낮음	높음
상호배제	명시적 P/V 호출	자동 제공
조건 대기	추가 세마포어 필요	Condition Variable 내장
프로그래밍 난이도	높음 (실수 가능)	낮음 (구조화됨)
언어 지원	라이브러리	Java, C# 등
데이터 캡슐화	없음	있음

조건이 충족될 때까지 기다리는 방법

조건 변수는 상태가 바뀌기를 기다리는 스레드를 다룬다. wait(condition)은 조건을 기다리는 동안 모니터 Lock을 해제하고, signal(condition)은 대기 중인 프로세스 하나를 깨운다. broadcast(condition)은 해당 조건을 기다리는 모든 프로세스를 깨운다.

생산자와 소비자가 유한 버퍼를 공유한다고 하면, not_full과 not_empty가 서로 다른 대기 조건을 표현한다.

// 모니터 정의
monitor BoundedBuffer {
    int buffer[N];
    int count = 0;
    condition not_full, not_empty;

    procedure produce(int item) {
        // 버퍼가 가득 찬 경우 대기
        while (count == N)
            wait(not_full);

        buffer[count] = item;
        count++;

        // 소비자에게 알림
        signal(not_empty);
    }

    procedure consume() returns int {
        // 버퍼가 빈 경우 대기
        while (count == 0)
            wait(not_empty);

        count--;
        int item = buffer[count];

        // 생산자에게 알림
        signal(not_full);

        return item;
    }
}

signal() 이후의 제어권을 누구에게 넘기는지에 따라 Hoare와 Mesa 시맨틱이 나뉜다. Hoare 시맨틱에서는 signal을 호출한 프로세스가 대기하고, 깨어난 프로세스가 즉시 실행한다. 따라서 조건이 엄격하게 유지된다.

if (count == 0)  // if 사용 가능
    wait(not_empty);

Mesa 시맨틱에서는 signal을 호출한 현재 프로세스가 계속 실행하고, 깨어난 프로세스는 준비 큐로 이동한다. 그 사이 다른 프로세스가 조건을 바꿀 수 있으므로 재검사가 필요하다.

while (count == 0)  // while 사용 필수
    wait(not_empty);

Hoare 방식은 이론적으로 깔끔하지만 구현이 복잡하다. Mesa 방식은 실용적이며 대부분의 실제 시스템이 채택한다.

언어 런타임에서 만나는 모니터

Java의 synchronized 메서드는 자동 상호배제를 제공한다. 객체의 wait()와 notifyAll()은 조건 변수 대기와 broadcast에 해당한다.

public class BoundedBuffer {
    private int[] buffer = new int[10];
    private int count = 0;

    // synchronized 키워드로 자동 상호배제
    public synchronized void produce(int item) throws InterruptedException {
        // 버퍼 가득 참 - 대기
        while (count == buffer.length) {
            wait();  // 조건 변수 대기 (모니터 lock 해제)
        }

        buffer[count++] = item;

        // 소비자 깨우기
        notifyAll();  // broadcast
    }

    public synchronized int consume() throws InterruptedException {
        // 버퍼 비어있음 - 대기
        while (count == 0) {
            wait();
        }

        int item = buffer[--count];

        // 생산자 깨우기
        notifyAll();

        return item;
    }
}

여기서 synchronized는 메서드 단위의 상호배제를, wait()는 현재 객체의 조건 변수 대기를 맡는다. notify()는 하나의 대기 쓰레드를 깨우고 notifyAll()은 모든 대기 쓰레드를 깨운다.

Python에서는 threading.Lock()과 이를 공유하는 threading.Condition()으로 같은 패턴을 구성할 수 있다.

import threading

class BoundedBuffer:
    def __init__(self, size=10):
        self.buffer = []
        self.size = size
        self.lock = threading.Lock()
        self.not_full = threading.Condition(self.lock)
        self.not_empty = threading.Condition(self.lock)

    def produce(self, item):
        with self.not_full:  # Lock 자동 획득/해제
            while len(self.buffer) == self.size:
                self.not_full.wait()  # 대기

            self.buffer.append(item)
            self.not_empty.notify()  # 소비자 깨움

    def consume(self):
        with self.not_empty:
            while len(self.buffer) == 0:
                self.not_empty.wait()

            item = self.buffer.pop(0)
            self.not_full.notify()  # 생산자 깨움

            return item

C++11에서는 std::mutex, std::condition_variable, std::unique_lock을 조합한다. Lambda로 대기 조건을 표현할 수 있다.

#include <mutex>
#include <condition_variable>
#include <queue>

class BoundedBuffer {
    std::queue<int> buffer;
    const size_t max_size = 10;
    std::mutex mtx;
    std::condition_variable not_full;
    std::condition_variable not_empty;

public:
    void produce(int item) {
        std::unique_lock<std::mutex> lock(mtx);

        // 버퍼 가득 참 대기
        not_full.wait(lock, [this] {
            return buffer.size() < max_size;
        });

        buffer.push(item);
        not_empty.notify_one();
    }

    int consume() {
        std::unique_lock<std::mutex> lock(mtx);

        // 버퍼 비어있음 대기
        not_empty.wait(lock, [this] {
            return !buffer.empty();
        });

        int item = buffer.front();
        buffer.pop();
        not_full.notify_one();

        return item;
    }
};

읽기·쓰기와 식사 문제에 적용하기

Readers-Writers 문제에서는 읽기 중인 수와 쓰기 상태를 모니터 내부 상태로 관리한다. 쓰기가 진행 중이면 읽기를 기다리게 하고, 읽는 프로세스가 남아 있으면 쓰기 역시 기다린다.

public class ReadersWriters {
    private int readers = 0;
    private boolean writing = false;

    public synchronized void startRead() throws InterruptedException {
        while (writing) {
            wait();  // 쓰기 중이면 대기
        }
        readers++;
    }

    public synchronized void endRead() {
        readers--;
        if (readers == 0) {
            notifyAll();  // 대기 중인 writer 깨움
        }
    }

    public synchronized void startWrite() throws InterruptedException {
        while (writing || readers > 0) {
            wait();  // 읽기/쓰기 중이면 대기
        }
        writing = true;
    }

    public synchronized void endWrite() {
        writing = false;
        notifyAll();  // 모든 대기자 깨움
    }
}

Dining Philosophers 문제에서는 각 철학자의 상태를 바꾸고, 양쪽 철학자가 식사 중이 아닐 때만 식사 상태로 전환한다.

public class DiningPhilosophers {
    private enum State { THINKING, HUNGRY, EATING }
    private State[] state = new State[5];
    private Object[] self = new Object[5];

    public DiningPhilosophers() {
        for (int i = 0; i < 5; i++) {
            state[i] = State.THINKING;
            self[i] = new Object();
        }
    }

    private void test(int i) {
        if (state[i] == State.HUNGRY &&
            state[(i + 4) % 5] != State.EATING &&
            state[(i + 1) % 5] != State.EATING) {
            state[i] = State.EATING;
            synchronized (self[i]) {
                self[i].notify();
            }
        }
    }

    public void pickup(int i) throws InterruptedException {
        synchronized (this) {
            state[i] = State.HUNGRY;
            test(i);
        }

        synchronized (self[i]) {
            while (state[i] != State.EATING) {
                self[i].wait();
            }
        }
    }

    public void putdown(int i) {
        synchronized (this) {
            state[i] = State.THINKING;
            test((i + 4) % 5);  // 왼쪽 철학자 확인
            test((i + 1) % 5);  // 오른쪽 철학자 확인
        }
    }
}

하위 동기화 도구로 모니터를 만들 때

모니터는 세마포어로 구현할 수 있다. 아래 구현은 signal() 호출자가 next에서 대기하는 Hoare semantics를 따른다.

// 모니터를 세마포어로 구현
typedef struct {
    Semaphore mutex;       // 상호배제
    Semaphore next;        // signal() 호출자 대기
    int next_count;        // next에서 대기 중인 수
} Monitor;

// 조건 변수 구조
typedef struct {
    Semaphore sem;         // 조건 대기 큐
    int count;             // 대기 중인 프로세스 수
} Condition;

void init_monitor(Monitor *m) {
    init_semaphore(&m->mutex, 1);
    init_semaphore(&m->next, 0);
    m->next_count = 0;
}

// 모니터 프로시저 진입
void enter_monitor(Monitor *m) {
    P(&m->mutex);
}

// 모니터 프로시저 퇴출
void exit_monitor(Monitor *m) {
    if (m->next_count > 0)
        V(&m->next);
    else
        V(&m->mutex);
}

// wait(c) 구현
void wait_condition(Monitor *m, Condition *c) {
    c->count++;
    if (m->next_count > 0)
        V(&m->next);
    else
        V(&m->mutex);

    P(&c->sem);  // 조건 대기
    c->count--;
}

// signal(c) 구현 (Hoare semantics)
void signal_condition(Monitor *m, Condition *c) {
    if (c->count > 0) {
        m->next_count++;
        V(&c->sem);      // 대기자 깨움
        P(&m->next);     // 자신은 대기
        m->next_count--;
    }
}

운영체제의 스레드 라이브러리에서는 mutex와 condition variable을 직접 묶는 형태가 일반적이다. pthread_cond_wait()은 대기 중 mutex를 자동으로 해제하고, 깨어난 뒤 다시 획득한다.

typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t cond_var;
} Monitor;

void init_monitor(Monitor *m) {
    pthread_mutex_init(&m->mutex, NULL);
    pthread_cond_init(&m->cond_var, NULL);
}

// 모니터 프로시저
void monitor_procedure(Monitor *m) {
    pthread_mutex_lock(&m->mutex);

    // 조건 대기
    while (!condition) {
        pthread_cond_wait(&m->cond_var, &m->mutex);
        // wait 중 mutex 자동 해제, 깨어나면 재획득
    }

    // Critical Section

    pthread_cond_signal(&m->cond_var);  // 대기자 깨움
    pthread_mutex_unlock(&m->mutex);
}

캡슐화가 주는 이점과 락 설계의 제약

모니터는 데이터와 동기화 로직을 캡슐화하고, 자동 상호배제로 프로그래밍 오류를 줄인다. 조건 변수로 복잡한 동기화 조건을 표현할 수 있으며, 세마포어보다 의도가 명확하다. 언어 지원이 있으면 컴파일러가 정확성 검증을 할 수 있다.

반대로 모든 메서드 호출에서 Lock이 필요하므로 성능 오버헤드가 생긴다. 큰 모니터는 병목이 될 수 있고, 낮은 우선순위가 Lock을 보유하면 우선순위 역전이 발생할 수 있다. 중첩 모니터 호출은 Deadlock 가능성을 만들며, C 같은 언어에서는 별도 라이브러리가 필요하다.

Java와 C#의 명시적 조건 처리

Java에서는 ReentrantLock과 Condition을 사용해 여러 조건 변수를 분리할 수 있다.

// ReentrantLock + Condition
import java.util.concurrent.locks.*;

public class BoundedBufferLock {
    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    private final int[] buffer = new int[10];
    private int count = 0;

    public void produce(int item) throws InterruptedException {
        lock.lock();
        try {
            while (count == buffer.length) {
                notFull.await();  // wait()와 유사
            }
            buffer[count++] = item;
            notEmpty.signal();    // notify()와 유사
        } finally {
            lock.unlock();
        }
    }

    public int consume() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) {
                notEmpty.await();
            }
            int item = buffer[--count];
            notFull.signal();
            return item;
        } finally {
            lock.unlock();
        }
    }
}

이 방식은 타임아웃 설정을 위한 tryLock(), await(timeout), 공정성 옵션인 ReentrantLock(true), 락 상태 조회용 isLocked()와 getQueueLength()를 제공한다.

C#에서는 Monitor.Enter, Monitor.Wait, Monitor.Pulse, Monitor.Exit으로 같은 대기와 신호 흐름을 구성한다.

using System;
using System.Threading;

public class BoundedBuffer {
    private int[] buffer = new int[10];
    private int count = 0;
    private readonly object lockObj = new object();

    public void Produce(int item) {
        Monitor.Enter(lockObj);
        try {
            while (count == buffer.Length) {
                Monitor.Wait(lockObj);
            }
            buffer[count++] = item;
            Monitor.Pulse(lockObj);  // signal
        } finally {
            Monitor.Exit(lockObj);
        }
    }

    public int Consume() {
        Monitor.Enter(lockObj);
        try {
            while (count == 0) {
                Monitor.Wait(lockObj);
            }
            int item = buffer[--count];
            Monitor.Pulse(lockObj);
            return item;
        } finally {
            Monitor.Exit(lockObj);
        }
    }
}

병목과 불필요한 깨우기를 줄이는 선택

하나의 큰 모니터가 병목이 되면 여러 작은 모니터로 나누는 fine-grained locking을 고려할 수 있다. ConcurrentHashMap의 Segment별 Lock이 예시다. 다만 분할한 락을 함께 획득하는 경우 Deadlock을 막기 위한 Lock 순서를 정의해야 한다.

단순 자료구조나 읽기 위주 작업에는 Atomic 연산과 CAS 루프를 사용하는 lock-free 대안도 적용할 수 있다. 높은 동시성과 Deadlock이 없다는 장점이 있지만, 구현이 복잡하고 ABA 문제가 있다.

조건 변수에서는 가능한 경우 broadcast보다 signal()을 사용하고, 정확한 조건 검사로 불필요한 wakeup을 줄인다. Spurious wakeup에 대비해 항상 while 루프로 조건을 재검사하며, 서로 다른 조건은 별도 조건 변수로 분리한다.