뮤텍스와 조건 변수 - 스레드 안전 작업 큐 만들기

이 장에서 배우는 것

앞 장에서 스레드의 시작과 종료를 객체의 수명에 연결했다. 이제 여러 스레드가 같은 주문에 접근하는 동안 무엇을 보호해야 하는지 살펴본다. 스레드를 모두 합류시켜도 실행 중 공유 자료를 잘못 읽고 쓰면 프로그램은 올바르지 않다. 수명 관리와 공유 상태의 동기화는 함께 설계해야 한다.

이번에는 주문을 넣는 생산자와 주문을 꺼내 검증하는 소비자 사이에 작업 큐를 둔다. 큐가 비었을 때 소비자를 재우고, 새 주문이 들어오거나 접수가 종료되면 깨운다. 구현의 중심은 잠금 함수의 개수가 아니라, 큐의 내용과 종료 상태를 하나의 규칙으로 관리하는 데 있다.

• 데이터 경쟁을 의도적으로 만들고 검사 도구의 보고를 해석한다.
• 뮤텍스와 잠금 객체를 역할에 맞게 선택한다.
• 조건 변수의 대기 조건을 작성하고 가짜 깨움에 대응한다.
• 접수를 닫은 뒤 남은 주문까지 처리하는 생산자·소비자 큐를 구현한다.
• 실행 순서에 의존하지 않는 로그 출력과 종료 절차를 확인한다.

문제 상황

작은 주문 처리 엔진은 지금까지 주문을 하나씩 받아 검증하고 체결 로그를 만들었다. 주문 접수가 잠시 몰리면 검증이 끝날 때까지 다음 접수가 늦어진다. 접수 담당은 주문을 큐에 넣고 돌아가며, 두 작업 스레드가 큐에서 주문을 꺼내 처리하도록 바꾸려 한다.

문제는 표준 컨테이너를 스레드 사이에 공유하는 순간 생긴다. 한 스레드가 큐에 주문을 추가하는 동안 다른 스레드가 맨 앞 주문을 제거하면 컨테이너의 내부 상태에 동시에 접근한다. 서로 다른 주문을 처리한다는 업무상의 구분만으로 컨테이너 접근까지 독립적이 되지는 않는다.

빈 큐도 별도의 문제다. 소비자가 계속 큐를 확인하면 할 일이 없는 동안에도 계산 자원을 사용한다. 반대로 아무 조건 없이 잠들면 새 주문이 도착한 사실을 놓칠 수 있다. 종료 시점에는 빈 큐가 잠깐 비어 있는 것인지, 앞으로도 주문이 오지 않을 것인지 구분해야 한다.

이 장의 큐는 접수 용량에 제한을 두지 않는다. 접수 종료 후에는 새 주문을 거절하고, 이미 들어온 주문은 모두 꺼낼 수 있다. 소비자는 접수가 닫혔고 큐도 비었을 때 종료한다. 체결은 실제 거래소 연결 대신, 수량과 가격이 양수인 주문에 대한 로그 기록으로 단순화한다.

공유 상태와 잠금의 경계

데이터 경쟁은 잘못된 숫자보다 넓은 문제다

데이터 경쟁(data race)은 서로 다른 스레드의 충돌하는 메모리 접근 사이에 필요한 동기화가 없을 때 생긴다. 여기서 다루는 일반 정수의 경우 같은 위치를 동시에 접근하고 적어도 한쪽이 쓰기를 수행하면 문제가 된다. 읽기만 하는 접근끼리는 이 조건에 해당하지 않는다.

int processed = 0;

auto work = [&processed] {
    for (int i = 0; i < 100000; ++i) {
        ++processed;
    }
};

std::jthread first(work);
std::jthread second(work);
first.join();
second.join();

증가식은 기존 값을 읽고, 증가한 값을 계산하고, 다시 저장하는 동작을 포함한다. 두 스레드가 같은 값을 읽은 뒤 같은 결과를 저장하는 그림으로 문제를 이해할 수 있다. 다만 이것은 원인을 설명하기 위한 모형이다. 실제 C++ 프로그램에 데이터 경쟁이 있으면 정의되지 않은 동작이므로, 결과가 몇만큼 줄어든다고 범위를 약속할 수 없다.

두 번의 합류는 주 스레드가 최종 값을 읽기 전에 작업 완료를 기다리게 한다. 그러나 작업 스레드끼리 수행한 증가 연산을 서로 동기화하지는 않는다. 실행 결과가 우연히 기대한 값이어도 경쟁이 없다는 증거가 되지 않는다.

하나의 규칙을 하나의 뮤텍스로 보호한다

뮤텍스(mutex)는 공유 상태에 접근하는 구간을 한 스레드씩 실행하게 한다. 앞선 스레드가 잠금을 해제한 뒤 다음 스레드가 같은 뮤텍스를 획득하면, 잠금 안에서 이루어진 변경을 동기화된 방식으로 관찰할 수 있다. 서로 다른 뮤텍스로 같은 변수를 보호한다고 주장하는 것만으로는 이 관계가 생기지 않는다.

int processed = 0;
std::mutex mutex;

auto work = [&] {
    for (int i = 0; i < 100000; ++i) {
        std::lock_guard<std::mutex> lock(mutex);
        ++processed;
    }
};

잠금으로 보호되는 구간을 임계 구역(critical section)이라 한다. 잠금의 범위는 유지해야 할 상태 규칙에서 정한다. 큐에서는 비어 있는지 확인하기, 앞 원소를 가져오기, 그 원소를 제거하기가 하나의 작업이다. 각각의 컨테이너 호출에만 잠금을 붙이면 호출 사이에 다른 소비자가 끼어들 수 있다.

직접 잠그고 직접 해제하는 코드는 중간 반환이나 예외 경로를 놓치기 쉽다. 잠금 객체를 지역 변수로 두면 생성과 소멸이 잠금의 수명을 표현한다. 기본서에서 배운 RAII가 스레드 동기화에서도 같은 역할을 한다.

잠금 객체는 필요한 소유권 동작에 따라 선택한다

도구	주요 역할	이 장의 사용 위치
std::lock_guard	한 구간 동안 잠금을 소유한다	주문 추가, 접수 종료, 로그 추가
std::unique_lock	잠금 소유권을 유지하며 해제와 재획득을 허용한다	조건 변수 대기
std::scoped_lock	여러 뮤텍스를 교착 회피 방식으로 함께 잠근다	두 공유 객체의 상태 변경

두 객체를 함께 바꿀 때의 잠금

주문 엔진의 두 처리 구역 사이에 작업 할당량을 옮긴다고 가정하자. 한 구역에서 하나를 빼고 다른 구역에 하나를 더하는 동안 두 구역을 함께 보호해야 한다. 다음은 두 구역의 합계가 유지되도록 하는 독립적인 예다. 할당량은 작은 음이 아닌 값이라고 가정한다.

struct Lane {
    std::mutex mutex;
    int quota = 0;
};

void move_one(Lane& from, Lane& to) {
    if (&from == &to) {
        return;
    }

    std::scoped_lock lock(from.mutex, to.mutex);
    if (from.quota > 0) {
        --from.quota;
        ++to.quota;
    }
}

두 뮤텍스를 각자 정한 순서로 잠그면 한 스레드는 첫 번째를, 다른 스레드는 두 번째를 소유한 채 상대의 잠금 해제를 기다릴 수 있다. 이런 교착 상태(deadlock)를 피하기 위해 두 잠금을 하나의 std::scoped_lock에 맡긴다. 같은 객체가 양쪽 인자로 들어오는 경우에는 같은 비재귀 뮤텍스를 중복 전달하지 않도록 먼저 반환한다.

이 도구가 프로그램 전체의 교착을 없애 주는 것은 아니다. 이미 다른 잠금을 가진 채 호출하거나, 잠금 안에서 외부 함수를 호출하면 더 큰 대기 관계가 만들어질 수 있다. 여러 잠금의 획득 규칙은 호출 경로 전체에서 일관되어야 한다.

조건 변수는 상태를 다시 확인하게 한다

조건 변수(condition variable)는 어떤 상태 변화가 있을 때까지 스레드를 대기시키는 도구다. 알림 자체를 작업 항목처럼 쌓아 두지는 않는다. 따라서 “알림을 받았는가” 대신 “지금 작업을 진행할 수 있는가”를 공유 상태로 표현해야 한다. 큐에서는 원소의 존재와 접수 종료 여부가 그 상태다.

std::unique_lock<std::mutex> lock(mutex_);
ready_.wait(lock, [this] {
    return closed_ || !orders_.empty();
});

대기는 잠금을 가진 상태에서 시작한다. 조건이 거짓이면 대기 연산이 잠금 해제와 대기 진입을 하나의 절차로 수행한다. 깨어난 뒤에는 같은 뮤텍스를 다시 획득한다. 조건을 확인하는 코드와 상태를 바꾸는 코드가 같은 뮤텍스를 사용해야, 조건 확인과 대기 사이의 틈으로 상태 변화를 놓치지 않는다.

std::unique_lock이 필요한 이유도 여기에 있다. 대기 중에는 생산자가 뮤텍스를 획득해 큐를 변경할 수 있어야 한다. 대기 함수는 잠금을 잠시 내놓았다가 다시 획득해야 하므로, 범위가 끝날 때만 해제하는 std::lock_guard로는 이 인터페이스를 사용할 수 없다.

대기 중인 스레드는 알림 없이도 깨어날 수 있다. 이를 가짜 깨움(spurious wakeup)이라 한다. 알림을 정상적으로 받았더라도 먼저 실행된 다른 소비자가 마지막 주문을 가져갈 수 있다. 두 경우 모두 잠금을 다시 얻은 뒤 조건을 확인해야 한다.

조건식을 받는 대기 함수는 조건이 참일 때까지 확인과 대기를 반복한다. 따라서 함수에서 돌아왔다는 사실은 조건식이 잠금 안에서 참으로 확인되었다는 뜻이다. 이 큐에서는 “주문이 있다” 또는 “접수가 닫혔다” 중 하나가 성립한다. 접수가 닫혔다고 곧바로 소비자를 끝내면 남은 주문을 버리므로, 반환 후 큐의 상태를 한 번 더 구분한다.

생산자는 먼저 잠금 안에서 주문을 추가하고, 잠금 범위를 나온 뒤 소비자 하나에 알린다. 이렇게 하면 깨어난 소비자가 생산자의 잠금 해제를 다시 기다릴 가능성을 줄일 수 있다. 알림을 잠금 안에서 수행하는 것도 허용된다. 올바름의 핵심은 알림의 위치만이 아니라, 상태 변경과 조건 확인이 같은 뮤텍스로 보호된다는 점이다.

소비자가 아직 대기하지 않을 때 주문이 들어와도 문제가 없다. 나중에 소비자가 뮤텍스를 획득하면 큐가 비어 있지 않음을 확인하고 대기를 건너뛴다. 남아 있어야 하는 것은 알림이 아니라 작업을 진행할 수 있다는 상태다.

생산자·소비자 큐의 종료 계약

생산자·소비자(producer–consumer) 구조에서 큐는 데이터 전달과 종료 전달을 함께 맡는다. 이번 인터페이스는 push, pop, close 세 연산으로 구성한다. push는 접수 성공 여부를 돌려주고, pop은 주문 또는 값이 없는 결과를 돌려준다. close는 다시 호출해도 닫힌 상태를 유지한다.

종료 표식으로 특별한 주문 번호를 예약하지 않는다. 그런 방법은 소비자 수만큼 표식을 넣어야 하는지, 표식 뒤의 주문은 어떻게 처리하는지 추가 규칙을 요구한다. 접수 상태를 별도 값으로 두면 큐의 데이터와 종료 의미가 분리된다.

접수가 열린 빈 큐에서는 소비자가 기다린다. 접수가 닫혀도 주문이 남아 있으면 소비자는 계속 꺼낸다. 접수가 닫히고 큐도 비어야 값이 없는 결과를 반환한다. 한 번 닫힌 큐를 다시 여는 기능은 제공하지 않는다.

push와 close가 겹치면 같은 뮤텍스가 두 연산의 순서를 정한다. 추가가 먼저 잠금 안에서 완료되면 그 주문은 처리 대상이다. 종료 상태가 먼저 설정되면 추가는 실패한다. 종료 요청이 시작된 시각만으로 접수 여부를 추측하지 않고, 잠금으로 보호된 실제 상태를 기준으로 판단한다.

주문 하나를 추가할 때는 소비자 하나가 처리하면 되므로 notify_one을 사용한다. 접수를 닫을 때는 기다리는 모든 소비자가 종료 조건을 확인해야 하므로 notify_all을 사용한다. 조건 변수는 소비자 간의 공평한 분배를 보장하지 않는다. 두 소비자를 만들었다고 주문이 절반씩 나뉘지는 않는다.

이 큐는 저장 공간이 허용하는 동안 생산자를 기다리게 하지 않는다. 처리량보다 접수량이 계속 많으면 메모리 사용이 증가한다. 용량 제한이 필요하면 생산자가 기다릴 조건과, 종료 시 그 생산자까지 깨우는 규칙을 추가해야 한다. 여기서는 빈 큐의 대기와 종료 절차에 집중한다.

완성 코드

다음 내용을 order_queue.cpp로 저장한다. 주 스레드가 생산자 역할을 맡고 두 소비자가 주문을 검증한다. 소비자는 화면에 직접 출력하지 않고 결과를 모은다. 모든 작업이 끝나면 주 스레드가 주문 번호순으로 정렬해 출력하므로 실행마다 같은 결과를 확인할 수 있다.

--race 인자를 주면 앞서 설명한 경쟁 예제를 실행한다. 이 경로는 검사 도구로 문제를 관찰하기 위해 의도적으로 잘못 작성했다. 인자 없이 실행하는 주문 처리 경로와는 별개다.

#include <algorithm>
#include <array>
#include <condition_variable>
#include <deque>
#include <exception>
#include <iostream>
#include <mutex>
#include <optional>
#include <stdexcept>
#include <string_view>
#include <thread>
#include <vector>

struct Order {
    int id;
    int quantity;
    int price;
};

struct Execution {
    int order_id;
    bool accepted;
};

class OrderQueue {
public:
    bool push(Order order) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            if (closed_) {
                return false;
            }
            orders_.push_back(order);
        }
        ready_.notify_one();
        return true;
    }

    std::optional<Order> pop() {
        std::unique_lock<std::mutex> lock(mutex_);
        ready_.wait(lock, [this] {
            return closed_ || !orders_.empty();
        });

        if (orders_.empty()) {
            return std::nullopt;
        }

        Order order = orders_.front();
        orders_.pop_front();
        return order;
    }

    void close() {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            closed_ = true;
        }
        ready_.notify_all();
    }

private:
    std::mutex mutex_;
    std::condition_variable ready_;
    std::deque<Order> orders_;
    bool closed_ = false;
};

void run_race() {
    int processed = 0;
    auto work = [&processed] {
        for (int i = 0; i < 100000; ++i) {
            ++processed;
        }
    };

    std::jthread first(work);
    std::jthread second(work);
    first.join();
    second.join();

    std::cout << "processed=" << processed << '\n';
}

void run_engine() {
    const std::array<Order, 6> input{{
        {101, 4, 1200},
        {102, 0, 1500},
        {103, 2, 900},
        {104, 3, -1},
        {105, 1, 500},
        {106, 8, 1100}
    }};

    OrderQueue queue;
    std::mutex log_mutex;
    std::vector<Execution> log;
    log.reserve(input.size());

    auto consume = [&] {
        while (auto order = queue.pop()) {
            const bool accepted =
                order->quantity > 0 && order->price > 0;

            const Execution result{order->id, accepted};
            std::lock_guard<std::mutex> lock(log_mutex);
            log.push_back(result);
        }
    };

    std::jthread first;
    std::jthread second;

    try {
        first = std::jthread(consume);
        second = std::jthread(consume);

        for (const Order& order : input) {
            if (!queue.push(order)) {
                throw std::logic_error("queue is closed");
            }
        }
    } catch (...) {
        queue.close();
        throw;
    }

    queue.close();
    first.join();
    second.join();

    std::sort(log.begin(), log.end(),
              [](const Execution& left, const Execution& right) {
                  return left.order_id < right.order_id;
              });

    int accepted_count = 0;
    for (const Execution& result : log) {
        std::cout << "order " << result.order_id << ": ";
        if (result.accepted) {
            ++accepted_count;
            std::cout << "executed\n";
        } else {
            std::cout << "rejected\n";
        }
    }

    std::cout << "summary: executed=" << accepted_count
              << ", rejected=" << log.size() - accepted_count
              << '\n';
}

int main(int argc, char* argv[]) {
    try {
        if (argc == 2 && std::string_view(argv[1]) == "--race") {
            run_race();
        } else {
            run_engine();
        }
    } catch (const std::exception& error) {
        std::cerr << "error: " << error.what() << '\n';
        return 1;
    }
    return 0;
}

줄별 해설

주문 자료와 큐의 공개 연산

Order는 세 정수를 가진 값 타입이다. 소비자는 큐에서 가져온 주문을 자기 지역 객체로 소유한다. 큐 내부 원소의 참조나 포인터를 밖으로 돌려주지 않으므로, 잠금이 풀린 뒤 다른 소비자가 원소를 제거해도 현재 주문의 수명에는 영향이 없다.

Execution에는 주문 번호와 검증 결과만 남긴다. 이 예제는 동기화 구조를 보기 위한 것이므로 부분 체결이나 가격 계산은 포함하지 않는다. 입력 순서는 접수 순서지만 소비자 처리 완료 순서는 실행 일정에 따라 달라질 수 있다.

push의 안쪽 중괄호는 잠금의 끝을 눈에 보이게 만든다. 종료 확인과 원소 추가를 같은 잠금 안에서 수행한다. push_back이 예외를 던지면 잠금 객체가 소멸하면서 잠금이 해제되고, 추가 성공을 알리는 코드까지 진행하지 않는다.

pop은 잠금 획득, 조건 대기, 원소 추출을 하나의 연산으로 묶는다. 대기 함수에서 돌아온 뒤 큐가 비어 있다면 조건식의 다른 항인 closed_가 참이다. 따라서 이 위치에서만 값이 없는 결과를 돌려준다. 큐가 닫혔더라도 비어 있지 않으면 계속 주문을 반환한다.

앞 원소를 지역 변수에 복사한 다음 큐에서 제거한다. 이 주문 타입의 정수 복사는 예외를 던지지 않는다. 임의의 타입을 담는 템플릿 큐로 바꾸면 추출 과정의 복사나 이동이 실패할 때 원소를 어떻게 보존할지 다시 검토해야 한다. 여기서는 구체적인 주문 타입을 사용해 그 문제를 늘리지 않는다.

close도 큐와 동일한 뮤텍스를 사용한다. 종료 여부가 단순한 참·거짓 값이라는 이유로 잠금 밖에서 쓰면 소비자의 조건 검사와 데이터 경쟁이 생긴다. 작은 변수인지보다 여러 스레드가 어떻게 접근하는지가 중요하다.

소비자가 잠금을 소유하는 시간

consume의 반복 조건은 주문을 얻었을 때만 본문을 실행한다. 검증 시점에는 pop의 잠금 객체가 이미 소멸했으므로 큐 뮤텍스를 소유하지 않는다. 실제 검증이 더 오래 걸리더라도 생산자와 다른 소비자가 큐에 접근할 수 있다.

로그는 여러 소비자가 같은 벡터에 추가하므로 별도의 뮤텍스로 보호한다. 미리 공간을 확보했더라도 벡터의 크기를 바꾸는 작업은 동시에 수행할 수 없다. 큐 잠금은 로그를 추가하기 전에 해제되므로 이 코드에서는 두 뮤텍스를 동시에 소유하지 않는다.

log.reserve는 스레드 시작 전에 여섯 결과를 담을 공간을 확보한다. 결과 타입은 단순한 값이고, 각 주문은 한 번씩만 꺼내므로 정상 경로의 로그 추가에는 추가 할당이 필요하지 않다. 이는 이 예제의 입력 수가 고정되어 있기 때문에 가능한 구성이다.

종료와 예외 경로

두 std::jthread는 처음에는 실행할 스레드 없이 생성한다. 스레드 시작과 주문 접수를 try 안에서 수행하므로, 두 번째 스레드 생성이나 주문 저장이 실패해도 catch에서 큐를 닫을 수 있다. 이후 예외가 전파되면 이미 시작한 스레드는 지역 객체의 소멸 과정에서 합류한다.

이 예외 경로에서 먼저 큐를 닫는 이유는 대기 중인 소비자를 깨우기 위해서다. std::jthread의 종료 요청만으로 일반 std::condition_variable 대기가 풀리지는 않는다. 이번 소비자는 종료 토큰 대신 큐의 접수 상태를 종료 조건으로 사용한다.

큐와 로그는 스레드 객체보다 먼저 선언했으므로 나중에 소멸한다. 소비자가 사용할 객체가 합류 전에 사라지지 않도록 선언 순서도 수명 설계에 포함한다. 정상 경로에서는 명시적으로 합류한 뒤 로그를 정렬하고 출력한다. 그때는 로그를 변경할 소비자가 없으므로 출력 구간에 로그 잠금이 필요하지 않다.

주 스레드의 catch가 소비자 내부에서 발생한 예외까지 받는 것은 아니다. 스레드 함수 밖으로 예외가 빠져나오면 프로그램이 종료된다. 검증에 외부 호출이나 메모리 할당을 추가한다면 소비자 내부에서 예외를 처리하고 실패 결과를 전달하는 정책도 마련해야 한다.

실행 결과

C++20의 std::jthread를 제공하는 컴파일러와 표준 라이브러리가 필요하다. macOS와 Linux에서 기본 빌드 명령은 다음과 같다. 완성 코드는 다음 경고 옵션에서 경고 없이 컴파일되도록 작성했다.

c++ -std=c++20 -Wall -Wextra -pthread order_queue.cpp -o order_queue
./order_queue

인자 없이 실행한 정상 경로의 출력은 다음과 같다. 소비자별 처리량과 완료 순서는 달라질 수 있지만, 출력 전에 주문 번호순으로 정렬하므로 아래 순서를 유지한다.

order 101: executed
order 102: rejected
order 103: executed
order 104: rejected
order 105: executed
order 106: executed
summary: executed=4, rejected=2

스레드 새니타이저(ThreadSanitizer)는 실행 중 메모리 접근과 동기화 관계를 추적해 데이터 경쟁을 찾는다. 해당 운영체제와 아키텍처에 맞는 런타임을 지원하는 Clang 또는 GCC에서 다음과 같이 별도의 실행 파일을 만든다. 배포 도구 체인에 따라 지원 여부가 다를 수 있다.

c++ -std=c++20 -Wall -Wextra -pthread -O1 -g -fsanitize=thread order_queue.cpp -o order_queue_tsan
./order_queue_tsan --race

경쟁이 관찰되면 진단에는 서로 충돌한 접근과 각 접근의 호출 위치가 나타난다. run_race 안의 증가식을 수행한 두 스레드의 기록을 함께 확인한다. 주소, 스레드 식별자, 줄 번호, 진단 순서는 환경에 따라 달라지므로 고정 출력으로 제시하지 않는다. 이 경로의 processed 값도 보장하지 않는다.

정상 큐 경로는 같은 검사 실행 파일을 인자 없이 실행한다.

./order_queue_tsan

정상 실행의 표준 출력은 위의 일곱 줄과 같다. 데이터 경쟁 진단은 예상하지 않는다. 다만 진단이 없다는 사실은 모든 실행 일정에서 올바르다는 증명이 아니다. 종료 누락으로 인한 무한 대기나 업무 규칙 오류도 별도 확인이 필요하다. 주소 새니타이저와 함께 켜기보다는 스레드 검사 전용 빌드로 분리한다.

플랫폼 지원과 검사 옵션의 사실 확인에는 Clang의 ThreadSanitizer 문서와 GCC의 계측 옵션 문서를 참고할 수 있다.

실무에서 자주 틀리는 것

한 번 확인한 뒤 조건 없이 기다린다

다음은 pop 내부라고 가정한 잘못된 코드다. 가짜 깨움이 생기거나 다른 소비자가 먼저 원소를 가져가면 빈 큐에 접근한다. 종료 상태도 반영하지 않아서 더 이상 주문이 오지 않아도 기다릴 수 있다.

std::unique_lock<std::mutex> lock(mutex_);
if (orders_.empty()) {
    ready_.wait(lock);
}
Order order = orders_.front();

대기 조건을 반복 검사하고, 반환 후 종료와 원소 추출을 나눈다.

std::unique_lock<std::mutex> lock(mutex_);
ready_.wait(lock, [this] {
    return closed_ || !orders_.empty();
});
if (orders_.empty()) {
    return std::nullopt;
}
Order order = orders_.front();
orders_.pop_front();
return order;

검사와 사용을 서로 다른 잠금 구간에 둔다

아래 코드는 개별 컨테이너 접근을 잠갔지만 하나의 추출 연산은 보호하지 못한다. 첫 잠금이 풀린 뒤 다른 소비자가 마지막 주문을 가져갈 수 있다. 이런 오류는 데이터 경쟁이 없어도 발생하므로 새니타이저 진단만으로 확인하기 어렵다.

bool available;
{
    std::lock_guard<std::mutex> lock(mutex_);
    available = !orders_.empty();
}
if (available) {
    std::lock_guard<std::mutex> lock(mutex_);
    Order order = orders_.front();
    orders_.pop_front();
    return order;
}
return std::nullopt;

기다리지 않는 추출 연산이라면 확인부터 제거까지 한 번의 잠금으로 묶는다. 이 코드의 값 없는 결과는 “현재 비었음”이며, 완성 코드의 pop이 표현하는 “접수 종료 후 소진됨”과 의미가 다르다.

std::lock_guard<std::mutex> lock(mutex_);
if (orders_.empty()) {
    return std::nullopt;
}
Order order = orders_.front();
orders_.pop_front();
return order;

접수를 닫을 때 한 소비자만 깨운다

다음 코드는 종료 상태를 올바르게 잠갔지만 깨우는 범위가 부족하다. 여러 소비자가 빈 큐에서 기다리고 있다면 나머지가 계속 대기할 수 있다. 가짜 깨움이 언젠가 해결해 줄 것이라고 기대해서는 안 된다.

{
    std::lock_guard<std::mutex> lock(mutex_);
    closed_ = true;
}
ready_.notify_one();

접수 종료는 모든 대기자에게 관계있는 상태 변화이므로 전체에 알린다. 알림 횟수로 소비자의 종료 여부를 관리하지 않고, 각 소비자가 닫힌 상태와 빈 큐를 확인하게 한다.

{
    std::lock_guard<std::mutex> lock(mutex_);
    closed_ = true;
}
ready_.notify_all();

합류한 다음 큐를 닫는다

소비자가 더 들어올 주문을 기다리는 구조에서 먼저 합류하면, 주 스레드는 소비자 종료를 기다리고 소비자는 주 스레드의 종료 통지를 기다린다. 아래 순서는 생산자의 접수가 끝났다는 사실을 소비자에게 전달하지 못한다.

first.join();
second.join();
queue.close();

먼저 접수를 닫고 대기자를 깨운 다음 합류한다. 큐를 파괴하는 것은 그 뒤다. 큐 소멸자에서 알림만 보내는 방법으로는 다른 스레드의 접근 완료까지 보장할 수 없다.

queue.close();
first.join();
second.join();

한눈에 보기

큐의 상태와 잠금 범위가 대기 및 종료 동작을 결정한다

상황	필요한 규칙	결과
접수가 열려 있고 큐가 비었다	조건식을 사용해 대기한다	주문 추가 또는 종료를 기다린다
큐에 주문이 있다	확인과 추출을 같은 잠금 안에서 수행한다	한 소비자가 주문 하나를 소유한다
접수가 닫혔고 주문이 남았다	종료 상태만 보고 반환하지 않는다	남은 주문을 계속 처리한다
접수가 닫혔고 큐가 비었다	값이 없는 결과를 반환한다	소비자 반복이 끝난다
여러 소비자가 기다린다	종료 시 모두에게 알린다	모든 소비자가 종료 조건을 확인한다
로그를 최종 출력한다	소비자와 합류한 뒤 정렬한다	경쟁 없이 일정한 출력 순서를 얻는다

뮤텍스는 여러 값이 함께 지켜야 하는 규칙을 보호한다. 조건 변수는 그 규칙이 작업을 허용할 때까지 기다리게 한다. 다음 장에서는 독립적인 카운터처럼 더 작은 공유 상태를 다룰 때 어떤 조건이 필요한지 살펴본다. 큐 전체의 상태 규칙을 단일 카운터와 같은 문제로 취급하지 않는 것이 출발점이다.

연습 문제

1. run_race의 증가식에 std::lock_guard를 적용하라. 두 스레드가 같은 뮤텍스를 사용하도록 수정하고, 합류 뒤 예상되는 출력과 잠금이 보장하는 내용을 설명하라.
2. 접수 종료 전후를 확인하는 간단한 순차 검사를 작성하라. 주문 하나를 넣고 큐를 닫은 뒤, 추가 접수는 실패하고 기존 주문은 한 번 반환되며 다음 추출은 값이 없어야 한다.
3. 완성 코드의 입력을 빈 std::array<Order, 0>로 바꿨을 때 예상 출력을 적어라. 두 소비자 중 어느 쪽이 먼저 실행되어도 종료할 수 있는 이유를 설명하라.
4. 대기 조건을 !orders_.empty()로만 바꾸면 접수 종료 시 어떤 문제가 생기는지 설명하라. notify_all 호출은 그대로 있다고 가정한다.

정답과 해설

1. 증가 연산의 동기화

processed와 같은 바깥 범위에 뮤텍스를 하나 선언하고 두 스레드가 참조하게 한다. 각 스레드 안에 별도 뮤텍스를 만들면 같은 값을 함께 보호하지 못한다. 증가 전체가 같은 잠금 안에서 실행되어야 한다.

int processed = 0;
std::mutex mutex;

auto work = [&] {
    for (int i = 0; i < 100000; ++i) {
        std::lock_guard<std::mutex> lock(mutex);
        ++processed;
    }
};

std::jthread first(work);
std::jthread second(work);
first.join();
second.join();
std::cout << "processed=" << processed << '\n';

각 스레드가 십만 번 증가하고 두 스레드가 완료된 뒤 읽으므로 출력은 다음과 같다. 뮤텍스는 증가 연산 사이의 상호 배제와 동기화를 제공하지만 어느 스레드가 먼저 실행될지는 정하지 않는다.

processed=200000

2. 접수 종료 계약의 확인

완성 코드에 <cassert>를 추가하고 다음 함수를 별도로 호출할 수 있다. 상태를 바꾸는 연산은 단언식 밖에서 실행한다. 그러면 단언을 비활성화한 빌드에서도 큐 연산 자체가 생략되지 않는다.

void check_close_contract() {
    OrderQueue queue;

    [[maybe_unused]] const bool before =
        queue.push(Order{201, 1, 700});
    queue.close();
    [[maybe_unused]] const bool after =
        queue.push(Order{202, 1, 800});

    [[maybe_unused]] const auto first = queue.pop();
    [[maybe_unused]] const auto second = queue.pop();

    assert(before);
    assert(!after);
    assert(first && first->id == 201);
    assert(!second);
}

이 검사는 종료 계약을 확인하지만 동시 실행의 모든 순서를 검사하지는 않는다. 소비자가 실제로 기다리는 경우와 접수 종료가 먼저 이루어진 경우도 검토해야 한다. 두 경우 모두 공유 상태를 확인해 같은 종료 결론에 도달하는지가 핵심이다.

3. 입력이 없는 경우

입력 선언을 다음과 같이 바꾼다.

const std::array<Order, 0> input{};

주문별 출력은 없고 합계만 남는다.

summary: executed=0, rejected=0

먼저 기다린 소비자는 접수 종료 알림으로 깨어난다. 종료 후에 실행된 소비자는 조건식에서 이미 닫힌 상태를 보고 대기하지 않는다. 따라서 소비자가 먼저 기다려야 한다는 실행 순서 가정이 필요하지 않다.

4. 대기 조건에서 종료 상태를 빠뜨린 경우

빈 큐에서 접수가 닫히면 알림을 받아도 !orders_.empty()는 거짓이다. 조건식을 받는 대기는 다시 잠들고, 이후 주문은 더 들어오지 않으므로 소비자가 끝나지 않을 수 있다. 알림이 조건식의 결과를 참으로 바꾸는 것은 아니다.

대기 조건은 “주문을 처리할 수 있음”과 “더 기다릴 필요가 없음”을 모두 포함해야 한다. 그래서 closed_ || !orders_.empty()를 사용하고, 깨어난 뒤 빈 큐인지 확인하여 종료와 처리를 구분한다.