세마포어를 사용하는 프로듀서와 컨슈머

프로듀서는 버퍼의 끝까지 데이터를 쓰다가, 버퍼 끝에 도달하면 처음부터 다시 시작하여 기존 데이터를 덮어씁니다. 컨슈머 스레드는 데이터가 생성되는 대로 이를 읽어서 표준 오류 출력(standard error)에 기록합니다.

세마포어를 사용하면 뮤텍스보다 더 높은 수준의 동시성을 확보할 수 있습니다. 버퍼에 대한 접근이 ` QMutex`로 보호된다면, 소비자 스레드는 생산자 스레드와 동시에 버퍼에 접근할 수 없습니다. 그러나 두 스레드가 버퍼의 서로 다른 부분을 동시에 처리한다고 해도 아무런 문제가 없습니다.

이 예제는 ` Producer `와 ` Consumer`라는 두 개의 클래스로 구성되어 있습니다. 두 클래스 모두 ` QThread`를 상속받습니다. 이 두 클래스 간의 통신에 사용되는 순환 버퍼와 이를 보호하는 세마포어는 전역 변수입니다.

프로듀서-컨슈머 문제를 해결하기 위해 ` QSemaphore `을 사용하는 대신, ` QWaitCondition `과 ` QMutex`을 사용할 수도 있습니다. 이것이 바로 ‘대기 조건을 사용하는 프로듀서와 컨슈머’ 예제에서 수행하는 방식입니다.

전역 변수

먼저 순환 버퍼와 관련 세마포어를 살펴보겠습니다.

constexpr int DataSize = 100000;

constexpr int BufferSize = 8192;
char buffer[BufferSize];

QSemaphore freeBytes(BufferSize);
QSemaphore usedBytes;

DataSize 는 생산자가 생성할 데이터의 양입니다. 예제를 최대한 간단하게 유지하기 위해 이를 상수로 설정합니다. BufferSize 는 순환 버퍼의 크기입니다. 이 값은 DataSize 보다 작으므로, 생산자는 어느 시점에서 버퍼의 끝에 도달한 후 처음부터 다시 시작하게 됩니다.

프로듀서와 컨슈머를 동기화하려면 두 개의 세마포어가 필요합니다. ` freeBytes ` 세마포어는 버퍼의 “사용 가능한” 영역(프로듀서가 아직 데이터를 채우지 않았거나 컨슈머가 이미 읽은 영역)을 제어합니다. usedBytes 세마포어는 버퍼의 “사용된” 영역(프로듀서가 데이터를 채웠지만 컨슈머가 아직 읽지 않은 영역)을 제어합니다.

이 두 세마포어는 함께 작동하여, 프로듀서가 컨슈머보다 BufferSize 바이트 이상 앞서지 않도록 하고, 컨슈머가 프로듀서가 아직 생성하지 않은 데이터를 읽지 않도록 보장합니다.

freeBytes 세마포어는 BufferSize 로 초기화되는데, 이는 처음에는 전체 버퍼가 비어 있기 때문입니다. usedBytes 세마포어는 0으로 초기화됩니다(값이 지정되지 않은 경우의 기본값).

프로듀서 클래스

Producer 클래스의 코드를 살펴보겠습니다.

class Producer : public QThread
{
public:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            freeBytes.acquire();
            buffer[i % BufferSize] = "ACGT"[QRandomGenerator::global()->bounded(4)];
            usedBytes.release();
        }
    }
};

프로듀서는 DataSize 바이트 분량의 데이터를 생성합니다. 순환 버퍼에 바이트를 기록하기 전에, freeBytes 세마포어를 사용하여 “사용 가능한” 바이트를 확보해야 합니다. 소비자가 프로듀서의 속도를 따라가지 못하면 QSemaphore::acquire() 호출이 차단될 수 있습니다.

마지막으로, 프로듀서는 usedBytes 세마포어를 사용하여 1바이트를 해제합니다. “사용 가능(free)” 바이트는 성공적으로 “사용 중(used)” 바이트로 전환되어, 컨슈머가 읽을 준비가 되었습니다.

컨슈머 클래스

이제 Consumer 클래스를 살펴보겠습니다.

class Consumer : public QThread
{
public:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            usedBytes.acquire();
            fprintf(stderr, "%c", buffer[i % BufferSize]);
            freeBytes.release();
        }
        fprintf(stderr, "\n");
    }
};

이 코드는 프로듀서의 코드와 매우 유사하지만, 이번에는 그 반대로 “사용된” 바이트를 확보하고 “사용 가능한” 바이트를 해제한다는 점이 다릅니다.

main() 함수

main() 에서 두 개의 스레드를 생성하고, 프로그램을 종료하기 전에 두 스레드 모두 작업을 완료할 시간을 확보하기 위해 ` QThread::wait()`를 호출합니다:

int main(int argc, char *argv[])
{
    QCoreApplication app(argc, argv);
    Producer producer;
    Consumer consumer;
    producer.start();
    consumer.start();
    producer.wait();
    consumer.wait();
    return 0;
}

그렇다면 프로그램을 실행하면 어떤 일이 일어날까요? 처음에는 프로듀서 스레드만이 작업을 수행할 수 있으며, 컨슈머는 usedBytes 세마포어가 해제되기를 기다리며 차단된 상태입니다( available()의 초기 카운트 값은 0입니다). 프로듀서가 버퍼에 1바이트를 넣으면, ` freeBytes.available() `는 ` BufferSize - 1`이 되고 ` usedBytes.available() `는 1이 됩니다. 이 시점에서 두 가지 상황이 발생할 수 있습니다. 소비자 스레드가 작업을 이어받아 해당 바이트를 읽거나, 프로듀서 스레드가 두 번째 바이트를 생성하게 되는 것입니다.

이 예제에서 제시된 프로듀서-컨슈머 모델을 통해 높은 동시성을 가진 멀티스레드 애플리케이션을 작성할 수 있습니다. 멀티프로세서 시스템에서 두 스레드가 버퍼의 서로 다른 부분에서 동시에 활동할 수 있으므로, 이 프로그램은 동등한 뮤텍스 기반 프로그램보다 잠재적으로 최대 두 배까지 빠를 수 있습니다.

다만 이러한 이점이 항상 실현되는 것은 아니라는 점에 유의해야 합니다. 뮤텍스( QSemaphore )를 획득하고 해제하는 데는 비용이 발생합니다. 실제로는 버퍼를 여러 청크(chunk)로 나누고, 개별 바이트 대신 청크 단위로 처리하는 것이 더 효율적일 수 있습니다. 또한 버퍼 크기는 실험을 통해 신중하게 선택해야 하는 매개변수입니다.

예제 프로젝트 @ code.qt.io