세마포어를 사용하는 생산자와 소비자

생산자는 버퍼의 끝에 도달할 때까지 버퍼에 데이터를 쓰고, 그 시점에서 처음부터 다시 시작하여 기존 데이터를 덮어씁니다. 소비자 스레드는 데이터가 생성되는 대로 데이터를 읽고 표준 오류에 기록합니다.

세마포어를 사용하면 뮤텍스보다 더 높은 수준의 동시성을 확보할 수 있습니다. 버퍼에 대한 액세스가 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 세마포어를 사용하여 바이트 하나를 해제합니다. "사용되지 않은" 바이트가 "사용된" 바이트가 성공적으로 변환되어 소비자가 읽을 준비가 되었습니다.

소비자 클래스

이제 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입니다. 이 시점에서 두 가지 일이 일어날 수 있습니다: 소비자 스레드가 인계받아 해당 바이트를 읽거나 생산자 스레드가 두 번째 바이트를 생성하게 됩니다.

이 예제에 제시된 생산자-소비자 모델을 사용하면 동시성이 높은 멀티스레드 애플리케이션을 작성할 수 있습니다. 멀티프로세서 머신에서는 버퍼의 서로 다른 부분에서 두 개의 스레드를 동시에 활성화할 수 있기 때문에 동등한 뮤텍스 기반 프로그램보다 최대 2배 더 빠를 수 있습니다.

하지만 이러한 이점이 항상 실현되는 것은 아니라는 점에 유의하세요. QSemaphore 를 획득하고 해제하는 데는 비용이 듭니다. 실제로는 버퍼를 청크로 나누고 개별 바이트가 아닌 청크 단위로 작업하는 것이 좋습니다. 버퍼 크기는 실험을 통해 신중하게 선택해야 하는 매개변수이기도 합니다.

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