세마포어

앞 글의 뮤텍스는 한 번에 한 스레드만 임계 영역에 들여보냅니다. 그런데 때로는 정해진 수까지는 동시에 들여보내야 합니다. 데이터베이스 연결 풀에 연결이 10개 있다면 스레드 10개까지는 동시에 연결을 쓸 수 있고, 11번째 스레드부터는 누군가 연결을 돌려줄 때까지 기다려야 합니다. 또 어떤 스레드가 일을 끝냈다는 사실을 다른 스레드에 알려야 할 때도 있습니다. 이런 일에 쓰는 도구가 1965년 무렵 에츠허르 데이크스트라(Edsger W. Dijkstra)가 제안한 세마포어(semaphore)입니다.

뮤텍스가 잠겼는지 풀렸는지만 아는 스위치라면, 세마포어는 남은 자원의 수를 세는 계수기입니다. 이 글에서는 세마포어의 연산과 쓰임새를 익히고, 동기화의 대표적인 고전 문제인 생산자와 소비자 문제, 독자와 저자 문제를 풀어 봅니다.

세마포어의 두 연산

세마포어는 0 이상의 정수 값 하나와 두 연산으로 이루어집니다. wait 연산은 값이 0보다 크면 1을 빼고 계속 진행하고, 0이면 값이 양수가 될 때까지 기다립니다. signal 연산은 값에 1을 더하고, 기다리는 스레드가 있으면 하나를 깨웁니다. 데이크스트라는 네덜란드어 단어에서 따와 두 연산을 P와 V라고 불렀으며, API에 따라 down과 up, acquire와 release라고도 부릅니다.

wait(S) {
    while (S <= 0)
        ;        /* 값이 양수가 될 때까지 기다린다. */
    S--;
}

signal(S) {
    S++;
}

이 코드는 연산의 뜻을 보여 줄 뿐, 이대로 구현하면 안전하지 않습니다. 두 스레드가 동시에 S가 1인 것을 확인하고 둘 다 1을 빼면 값이 -1이 되어 버립니다. 6장 1편에서 본 확인 후 행동의 경쟁입니다. 그래서 실제 세마포어는 값을 확인하고 빼는 과정을 원자적으로 처리하고, 기다리는 스레드를 돌며 기다리게 하지 않고 잠재웁니다. 그 구현 방법은 이 글의 마지막에서 봅니다.

동시에 쓰는 수를 세는 카운팅 세마포어

초깃값이 N인 세마포어를 카운팅 세마포어(counting semaphore)라고 하며, 똑같은 자원 N개를 관리할 때 씁니다. 자원을 쓰기 전에 wait로 하나를 얻고, 다 쓰면 signal로 돌려줍니다. 다음 코드는 연결이 3개인 풀에 요청 6개가 동시에 들어올 때, 실제로 동시에 쓰인 연결의 최대 수를 셉니다.

import threading
import time

connections = threading.Semaphore(3)     # 연결 3개짜리 풀
in_use = 0
peak = 0
count_lock = threading.Lock()

def query(n):
    global in_use, peak
    with connections:                    # 연결 하나를 얻을 때까지 기다린다
        with count_lock:
            in_use += 1
            peak = max(peak, in_use)
        time.sleep(0.1)                  # 질의를 처리하는 시간
        with count_lock:
            in_use -= 1

threads = [threading.Thread(target=query, args=(n,)) for n in range(6)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print("요청 6개, 동시에 쓴 연결의 최대 수:", peak)

s = threading.Semaphore(1)
s.release()                              # 실수로 한 번 더 반납한다
print("Semaphore(1)에 한 번 더 반납한 뒤:", s.acquire(blocking=False), s.acquire(blocking=False))

b = threading.BoundedSemaphore(1)
b.acquire()
b.release()
try:
    b.release()
except ValueError as error:
    print("BoundedSemaphore(1)에 한 번 더 반납:", error)

요청 6개, 동시에 쓴 연결의 최대 수: 3
Semaphore(1)에 한 번 더 반납한 뒤: True True
BoundedSemaphore(1)에 한 번 더 반납: Semaphore released too many times

요청 6개가 동시에 들어와도 세마포어가 3에서 시작하므로 세 요청만 들어가고, 나머지 셋은 앞의 요청이 연결을 돌려줄 때까지 기다립니다. with connections:는 들어갈 때 acquire()를, 나올 때 release()를 부릅니다. 세마포어 하나로 정하는 것은 동시에 몇 개까지 쓸 수 있느냐는 상한이며, 1초에 몇 개를 처리하느냐는 속도 제한은 시간에 따라 허가를 채워 주는 다른 방법이 필요합니다.

출력의 둘째와 셋째 줄은 세마포어를 쓸 때 흔한 실수를 보여 줍니다. 세마포어는 얻은 횟수보다 많이 돌려주어도 막지 않으므로, 초깃값이 1인 세마포어에 한 번 더 반납하면 값이 2가 되어 두 스레드가 함께 들어갈 수 있게 됩니다. Python의 BoundedSemaphore는 초깃값보다 많이 반납하면 ValueError를 일으켜 이런 실수를 바로 알려 줍니다.

값이 0과 1뿐인 이진 세마포어

값을 0과 1로만 쓰는 세마포어를 이진 세마포어(binary semaphore)라고 합니다. 뮤텍스처럼 상호 배제에 쓸 수 있지만, 둘 사이에는 중요한 차이가 있습니다. POSIX의 뮤텍스는 소유권(ownership)이 있어서, 잠근 스레드가 풀어야 합니다. 다른 스레드가 풀면 뮤텍스 종류에 따라 오류가 나거나 동작이 정의되지 않습니다. 세마포어에는 소유권이 없어서, 한 스레드가 wait하고 다른 스레드가 signal해도 됩니다. Python의 기본 threading.Lock도 소유권을 확인하지 않아 다른 스레드가 풀 수 있다는 점은 알아 둘 만합니다.

소유권이 없다는 성질 덕분에 세마포어는 스레드 사이에 일이 일어났음을 알리는 데 알맞습니다. 0에서 시작하는 세마포어를 두고, 결과를 기다리는 스레드는 wait를, 일을 끝낸 스레드는 signal을 부르면, 기다리는 쪽은 일이 끝날 때까지 잠들어 있다가 깨어납니다. 6장 1편에서 본 "B가 끝나야 D를 시작한다" 같은 실행 순서의 조건을 이렇게 만들 수 있습니다.

생산자와 소비자 문제

생산자와 소비자(producer-consumer) 문제는 동기화의 가장 대표적인 고전 문제입니다. 생산자는 데이터를 만들어 크기가 정해진 버퍼에 넣고, 소비자는 버퍼에서 데이터를 꺼내 처리합니다. 메시지 큐, 로그 수집, 처리 단계를 이은 파이프라인처럼 실무 곳곳에 이 모양이 있습니다. 이 문제를 풀려면 세 가지 동기화가 함께 필요합니다.

1. 버퍼가 가득 차면 생산자가 기다려야 합니다. 빈 자리의 수를 세는 세마포어 empty로 풉니다.
2. 버퍼가 비어 있으면 소비자가 기다려야 합니다. 채워진 자리의 수를 세는 세마포어 full로 풉니다.
3. 버퍼 자체를 고치는 동안에는 다른 스레드가 끼어들면 안 됩니다. 뮤텍스 mutex로 풉니다.

import threading
import time

CAPACITY = 5
buffer = []
mutex = threading.Lock()
empty = threading.Semaphore(CAPACITY)    # 빈 자리 수
full = threading.Semaphore(0)            # 채워진 자리 수
consumed, longest = [], 0

def producer():
    global longest
    for item in range(10):
        empty.acquire()                  # 빈 자리를 먼저 확보하고
        with mutex:                      # 버퍼를 고칠 때만 잠근다
            buffer.append(item)
            longest = max(longest, len(buffer))
        full.release()                   # 꺼낼 수 있는 항목이 하나 늘었다

def consumer():
    for _ in range(10):
        full.acquire()                   # 꺼낼 항목이 생길 때까지 기다리고
        with mutex:
            consumed.append(buffer.pop(0))
        empty.release()                  # 빈 자리를 돌려준다
        time.sleep(0.02)                 # 소비가 생산보다 느리다

threads = [threading.Thread(target=producer), threading.Thread(target=consumer)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print("소비한 순서:", consumed)
print("버퍼에 동시에 있던 항목의 최대 수:", longest)

소비한 순서: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
버퍼에 동시에 있던 항목의 최대 수: 5

소비자가 생산자보다 느리므로 버퍼는 곧 가득 차지만, 항목 수는 용량 5를 넘지 않습니다. 버퍼가 가득 차면 empty가 0이 되어 생산자가 empty.acquire()에서 기다리고, 소비자가 하나를 꺼내 empty.release()를 부를 때마다 생산자가 하나씩 더 넣기 때문입니다. 소비자는 처음에 full이 0이므로 생산자가 첫 항목을 넣을 때까지 기다리며, 항목을 빠짐없이 넣은 순서대로 꺼냅니다.

허가와 버퍼 길이의 관계

세마포어의 값과 버퍼의 길이가 늘 일치하는 것은 아닙니다. 생산자가 empty에서 허가를 얻은 뒤 아직 항목을 넣지 않은 순간이 있고, 항목을 넣은 뒤 아직 full을 올리지 않은 순간도 있기 때문입니다. 다음 단계 개념도는 용량 5인 버퍼에서 생산과 소비를 한 번씩 할 때, 허가의 수와 버퍼 길이가 어떻게 바뀌는지 따라갑니다.

처음에는 empty가 5, full이 0입니다. 생산자가 empty에서 허가를 하나 얻으면 empty는 4가 되지만, 버퍼 길이는 아직 0입니다. 이렇게 얻었지만 아직 반대쪽 세마포어로 넘기지 않은 허가를 held라고 세면, 어느 순간에나 empty + full + held = 5가 성립합니다. 항목을 넣으면 버퍼 길이만 1이 되고, full을 올려야 허가가 full 쪽으로 넘어갑니다. 소비도 같은 방식으로 full에서 허가를 얻고, 항목을 꺼내고, empty에 허가를 돌려줍니다. 그래서 "full의 값은 언제나 버퍼 길이와 같다"는 생각은 틀렸고, 둘은 허가를 넘기는 단계가 끝난 시점에만 같아집니다.

잠그는 순서가 바뀌면

위 코드에서 생산자는 먼저 empty로 빈 자리를 확보한 뒤 mutex를 잠급니다. 이 순서를 바꾸어 mutex를 먼저 잠그고 그 안에서 빈 자리를 기다리면 어떻게 될까요? 다음 코드는 버퍼가 가득 찬 상황을 만들어 확인합니다. 영원히 멈추지 않도록 기다리는 시간에 제한을 두었습니다.

import threading
import time

mutex = threading.Lock()
empty = threading.Semaphore(0)           # 버퍼가 가득 차 빈 자리가 없다

def bad_producer():
    with mutex:                          # 잘못된 순서: 잠금을 먼저 쥐고
        got = empty.acquire(timeout=0.3) # 빈 자리를 기다린다
        print("생산자: 빈 자리", "얻음" if got else "끝내 얻지 못함")

def consumer():
    got = mutex.acquire(timeout=0.1)     # 항목을 꺼내려면 잠금이 필요하다
    print("소비자: 잠금", "얻음" if got else "얻지 못함, 빈 자리를 돌려줄 수 없음")
    if got:
        empty.release()
        mutex.release()

p = threading.Thread(target=bad_producer)
p.start()
time.sleep(0.05)                         # 생산자가 먼저 잠금을 쥐게 한다
c = threading.Thread(target=consumer)
c.start()
p.join()
c.join()

소비자: 잠금 얻지 못함, 빈 자리를 돌려줄 수 없음
생산자: 빈 자리 끝내 얻지 못함

생산자는 mutex를 쥔 채 빈 자리를 기다립니다. 빈 자리를 만들어 줄 수 있는 것은 소비자뿐인데, 소비자는 항목을 꺼내려면 mutex가 필요하므로 생산자가 잠금을 풀 때까지 기다립니다. 서로가 상대의 다음 동작을 기다리는 교착 상태입니다. 이 예제는 시간 제한 때문에 둘 다 포기하고 끝나지만, 실제 코드처럼 제한 없이 기다렸다면 영원히 멈췄을 것입니다. 그래서 기다릴 수 있는 세마포어는 뮤텍스 밖에서 먼저 얻고, 뮤텍스는 버퍼를 고치는 짧은 구간에만 쥐어야 합니다. 교착 상태는 7장에서 자세히 다룹니다.

독자와 저자 문제

독자와 저자(readers-writers) 문제는 데이터베이스, 파일 시스템, 캐시에서 자주 만나는 모양입니다. 규칙은 다음과 같습니다. 데이터를 읽기만 하는 독자는 여럿이 동시에 읽어도 안전합니다. 데이터를 고치는 저자가 쓰는 동안에는 다른 저자는 물론 독자도 들어오면 안 됩니다. 저자는 읽고 있는 독자가 모두 나갈 때까지 기다려야 합니다. 모든 접근을 뮤텍스 하나로 막으면 규칙은 지키지만, 독자들이 동시에 읽을 수 있는 기회를 버리게 됩니다.

이 문제에는 우선순위에 따라 변형이 있습니다. 독자 우선 방식에서는 다른 독자가 읽고 있으면 새 독자가 바로 들어가므로, 독자가 끊임없이 오면 저자가 기아에 빠집니다. 저자 우선 방식에서는 기다리는 저자가 있으면 새 독자를 막아 두므로, 저자가 끊임없이 오면 이번에는 독자가 기아에 빠집니다. 다음 코드는 독자 우선 방식을 구현하고, 실행 순서를 정해 두 독자와 한 저자의 상호작용을 따라갑니다.

import threading

read_count = 0
count_lock = threading.Lock()            # read_count를 보호한다
resource = threading.Lock()              # 저자를 막는 잠금

def reader_enter(name):
    global read_count
    with count_lock:
        read_count += 1
        if read_count == 1:
            resource.acquire()           # 첫 독자가 저자를 막는다
        print(f"{name} 읽기 시작, 독자 {read_count}명")

def reader_leave(name):
    global read_count
    with count_lock:
        read_count -= 1
        if read_count == 0:
            resource.release()           # 마지막 독자가 저자를 들여보낸다
        print(f"{name} 읽기 끝, 독자 {read_count}명")

def writer_try(name):
    got = resource.acquire(blocking=False)
    print(f"{name} 쓰기 시도:", "들어감" if got else "독자가 있어 기다림")
    if got:
        resource.release()

reader_enter("R1"); reader_enter("R2"); writer_try("W")
reader_leave("R1"); writer_try("W")
reader_leave("R2"); writer_try("W")

R1 읽기 시작, 독자 1명
R2 읽기 시작, 독자 2명
W 쓰기 시도: 독자가 있어 기다림
R1 읽기 끝, 독자 1명
W 쓰기 시도: 독자가 있어 기다림
R2 읽기 끝, 독자 0명
W 쓰기 시도: 들어감

첫 독자 R1이 들어오면 read_count가 1이 되면서 resource를 잠가 저자를 막습니다. 둘째 독자 R2는 이미 막혀 있으므로 resource를 다시 잠그지 않고 들어와 함께 읽습니다. R1이 나가도 R2가 남아 있으므로 저자는 여전히 들어가지 못하고, 마지막 독자 R2가 나가면서 read_count가 0이 될 때 비로소 resource를 풀어 저자가 들어갑니다. 여기서 resource를 잠근 것은 R1이고 푼 것은 R2라는 점에 주목하세요. 실제 여러 스레드에서 쓰면 잠근 스레드와 푼 스레드가 다르므로, 이 구현은 소유권을 따지지 않는 Python의 Lock이나 세마포어로는 되지만, 소유권이 있는 POSIX 뮤텍스로 그대로 옮기면 안 됩니다. 또 이 코드는 R2가 나가기 전에 새 독자가 계속 들어오면 저자가 끝없이 기다리는 기아 문제를 그대로 가지고 있습니다.

읽기 쓰기 잠금

이 모양이 워낙 흔해서 많은 언어가 읽기 쓰기 잠금(read-write lock)을 제공합니다. C의 pthread_rwlock_t, Java의 ReentrantReadWriteLock, Rust의 std::sync::RwLock이 그 예입니다. 읽기 잠금은 여럿이 함께 쥘 수 있고, 쓰기 잠금은 혼자만 쥘 수 있습니다. 독자와 저자 가운데 누구를 우선할지는 구현마다 다르므로, 기아가 걱정된다면 쓰는 라이브러리의 설명을 확인해야 합니다. Python 표준 라이브러리에는 읽기 쓰기 잠금이 없으며, threading.RLock은 같은 스레드가 여러 번 잠글 수 있는 배타적 잠금일 뿐 여러 독자의 동시 읽기를 허락하지 않습니다.

데이터베이스의 공유 잠금(shared lock)과 배타 잠금(exclusive lock)도 같은 생각입니다. 다만 모든 조회가 공유 잠금을 거는 것은 아닙니다. MySQL의 InnoDB는 기본 격리 수준인 REPEATABLE READ와 그보다 약한 READ COMMITTED에서는 일반 SELECT를 잠금 없이 다중 버전 동시성 제어(MVCC)의 스냅샷으로 읽고, SELECT ... FOR SHARE처럼 명시한 경우에 공유 잠금을 겁니다. 가장 엄격한 SERIALIZABLE에서는 autocommit을 끄면 일반 SELECT도 SELECT ... FOR SHARE로 바꾸어 공유 잠금을 겁니다. 일반 SELECT의 스냅샷 읽기와 FOR SHARE·FOR UPDATE 같은 잠금 읽기가 어떻게 다른지는 데이터베이스 교재의 격리 수준과 잠금에서 다룹니다.

더 깊이: 라이브러리마다 다른 읽기 쓰기 잠금의 우선순위

glibc의 pthread_rwlock_t는 기본으로 독자를 앞세웁니다. pthread_rwlockattr_setkind_np(3)에 따르면 기본 종류인 PTHREAD_RWLOCK_PREFER_READER_NP에서는 독자가 남아 있는 한 저자가 굶을 수 있습니다. 이름만 보면 저자를 앞세울 것 같은 PTHREAD_RWLOCK_PREFER_WRITER_NP는 glibc가 무시합니다. POSIX는 같은 스레드가 읽기 잠금을 겹쳐 얻을 수 있어야 한다고 요구하는데, 이를 허락하면서 저자를 앞세우면 쉽게 교착 상태가 생기기 때문입니다. 저자의 기아를 막으려면 읽기 잠금을 겹쳐 얻지 않는다는 조건 아래 PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP를 고릅니다.

그 교착 상태는 이런 순서로 생깁니다. 스레드 1이 읽기 잠금을 쥔 상태에서 스레드 2가 쓰기 잠금을 기다리고, 스레드 1이 읽기 잠금을 한 번 더 얻으려 한다고 합시다. 기다리는 저자가 새 독자를 막는 구현이라면 스레드 1의 두 번째 읽기는 스레드 2 뒤에서 기다리고, 스레드 2는 스레드 1의 첫 번째 읽기 잠금이 풀리기를 기다립니다. Rust의 RwLock 문서는 우선순위 정책이 운영체제의 구현에 달려 있다고 밝히면서, 바로 이 순서가 교착 상태에 빠질 수 있다고 경고합니다.

Java의 ReentrantReadWriteLock은 독자와 저자 어느 쪽도 앞세우지 않는 대신 공정성(fairness) 정책을 고르게 합니다. 기본인 비공정 모드에서는 들어가는 순서가 정해져 있지 않아, 경합이 계속되면 어떤 독자나 저자가 한없이 밀릴 수 있지만 처리량은 보통 더 높습니다. 공정 모드에서는 대략 도착한 순서로 들어가며, 저자가 기다리고 있으면 새로 온 독자는 그 저자가 쓰기를 마칠 때까지 기다립니다.

대기 큐로 세마포어 구현하기

세마포어를 돌며 기다리는 방식으로 구현하면 CPU를 낭비하므로, 실제 구현은 기다리는 스레드를 대기 큐에 넣고 잠재웁니다. 다음은 교과서에서 흔히 쓰는 모델로, 값이 음수가 되는 것을 허락해 그 절댓값으로 기다리는 스레드의 수를 나타냅니다.

typedef struct {
    int value;
    struct list_head wait_queue;   /* 기다리는 스레드들 */
} semaphore;

void wait(semaphore *S)
{
    S->value--;
    if (S->value < 0) {
        add_current_thread(&S->wait_queue);
        block();                    /* 잠든다. */
    }
}

void signal(semaphore *S)
{
    S->value++;
    if (S->value <= 0)
        wakeup(remove_one_thread(&S->wait_queue));   /* 하나를 깨워 준비 상태로 */
}

value가 -3이라면 스레드 셋이 큐에서 기다리고 있다는 뜻입니다. signal이 값을 올린 뒤에도 0 이하라면 기다리는 스레드가 남아 있다는 뜻이므로 하나를 깨웁니다. 이 음수 값은 모델 안에서의 표현일 뿐이며, Python의 Semaphore처럼 값을 0 이상으로 유지하는 구현도 많습니다.

그런데 wait와 signal 자체도 값을 고치고 큐를 다루므로 임계 영역입니다. 실제 구현은 이 짧은 구간을 스핀락으로 보호하며, 구간이 몇 줄뿐이어서 돌며 기다리는 비용이 거의 없습니다. 다만 스핀락을 쥔 채로 잠들면 아무도 그 스핀락을 얻지 못하므로, 큐에 자신을 등록하고 스핀락을 푸는 것과 잠드는 것을 깨우기 신호를 놓치지 않도록 이어 주어야 합니다. 커널은 이 일을 스케줄러와 함께 처리하고, 사용자 공간의 구현은 앞 글에서 본 futex에 맡깁니다.

정리

• 세마포어는 0 이상의 정수 값과 wait, signal 두 연산으로 이루어지며, wait는 값이 양수일 때 1을 빼고 0이면 기다리고, signal은 1을 더하고 기다리는 스레드를 깨웁니다.
• 카운팅 세마포어는 똑같은 자원 N개를 동시에 쓰는 수를 제한하고, 이진 세마포어는 소유권이 없어 스레드 사이에 일이 끝났음을 알리는 데도 씁니다. 얻은 횟수보다 많이 반납하는 실수는 BoundedSemaphore로 잡을 수 있습니다.
• 생산자와 소비자 문제는 빈 자리를 세는 empty, 채워진 자리를 세는 full, 버퍼를 보호하는 mutex로 풉니다. 기다릴 수 있는 세마포어는 뮤텍스 밖에서 먼저 얻어야 교착 상태를 피합니다.
• 독자와 저자 문제는 여러 독자의 동시 읽기를 허락하되 저자는 혼자 쓰게 하는 문제이며, 독자와 저자 가운데 누구를 우선하느냐에 따라 상대쪽의 기아가 생길 수 있습니다.
• 실제 세마포어는 기다리는 스레드를 대기 큐에 넣고 잠재우며, 값과 큐를 고치는 짧은 구간은 스핀락 같은 저수준 잠금으로 보호합니다.

연습 문제

1. 소비 한 번의 허가 추적

용량 5인 버퍼에 항목이 3개 있고, empty가 2, full이 3, 반대쪽으로 넘기지 않은 허가 held가 0입니다. 다른 스레드의 동작 없이 소비자 하나가 full 획득, 버퍼에서 항목 꺼내기, empty 반환을 차례로 마칠 때, 각 단계 뒤의 (empty, full, held)와 버퍼 길이를 적으세요.

2. 일이 끝났음을 알리기

메인 스레드가 작업 스레드 셋을 시작한 뒤, 셋이 모두 초기화를 마칠 때까지 기다렸다가 다음 일을 하려 합니다. join은 쓸 수 없는데, 작업 스레드들은 초기화를 마친 뒤에도 계속 실행되기 때문입니다. 세마포어 하나로 이를 구현하는 방법을 설명하고, 초깃값을 얼마로 해야 하는지 답하세요.

3. 끝나지 않는 쓰기

독자 우선 방식의 읽기 쓰기 잠금을 쓰는 캐시가 있습니다. 조회 요청은 초당 수천 개가 들어오고, 각 조회는 몇 밀리초씩 읽기 잠금을 쥡니다. 1분에 한 번 캐시를 갱신하는 저자가 몇 분째 쓰기 잠금을 얻지 못하고 있습니다. 원인을 설명하고 해결 방법을 두 가지 제시하세요.

다음 글 모니터와 고전적 동기화 문제에서는 잠금과 조건 변수를 묶은 모니터, 조건을 다시 확인해야 하는 이유, 식사하는 철학자 문제를 살펴봅니다.