대기 조건을 사용하는 프로듀서와 컨슈머

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

대기 조건(wait condition)을 사용하면 뮤텍스만 사용할 때보다 더 높은 수준의 동시성을 확보할 수 있습니다. 만약 버퍼에 대한 접근이 단순히 QMutex 로만 보호된다면, 소비자 스레드는 생산자 스레드와 동시에 버퍼에 접근할 수 없을 것입니다. 그러나 두 스레드가 동시에 버퍼의 서로 다른 부분을 처리하는 데는 아무런 문제가 없습니다.

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

프로듀서-컨슈머 문제를 해결하기 위해 ` QWaitCondition `와 ` QMutex `를 사용하는 대신 ` QSemaphore`를 사용할 수도 있습니다. ‘세마포어를 사용하는 프로듀서와 컨슈머’ 예제에서는 바로 이 방법을 사용합니다.

전역 변수

먼저 순환 버퍼와 관련 동기화 도구를 살펴보겠습니다.

constexpr int DataSize = 100000;
constexpr int BufferSize = 8192;

QMutex mutex; // protects the buffer and the counter
char buffer[BufferSize];
int numUsedBytes;

QWaitCondition bufferNotEmpty;
QWaitCondition bufferNotFull;

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

프로듀서와 컨슈머를 동기화하기 위해서는 두 개의 대기 조건과 하나의 뮤텍스가 필요합니다. bufferNotEmpty 조건은 프로듀서가 데이터를 생성했을 때 신호가 발생하여, 컨슈머에게 데이터 읽기를 시작해도 좋다는 것을 알립니다. bufferNotFull 조건은 컨슈머가 데이터를 읽었을 때 신호가 발생하여, 프로듀서에게 더 많은 데이터를 생성해도 좋다는 것을 알립니다. numUsedBytes 는 버퍼 내에 데이터가 포함된 바이트 수를 나타냅니다.

대기 조건, 뮤텍스, 그리고 numUsedBytes 카운터가 함께 작동하여, 프로듀서가 컨슈머보다 BufferSize 바이트 이상 앞서지 않도록 하고, 컨슈머가 프로듀서가 아직 생성하지 않은 데이터를 읽지 않도록 보장합니다.

프로듀서 클래스

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

class Producer : public QThread
{
public:
    explicit Producer(QObject *parent = nullptr)
        : QThread(parent)
    {
    }

private:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            {
                const QMutexLocker locker(&mutex);
                while (numUsedBytes == BufferSize)
                    bufferNotFull.wait(&mutex);
            }

            buffer[i % BufferSize] = "ACGT"[QRandomGenerator::global()->bounded(4)];

            {
                const QMutexLocker locker(&mutex);
                ++numUsedBytes;
                bufferNotEmpty.wakeAll();
            }
        }
    }
};

프로듀서는 DataSize 바이트 분량의 데이터를 생성합니다. 순환 버퍼에 바이트를 쓰기 전에, 먼저 버퍼가 가득 찼는지(즉, numUsedBytes 가 BufferSize 와 같은지) 확인해야 합니다. 버퍼가 가득 찼다면, 스레드는 bufferNotFull 컨디션에서 대기합니다.

마지막으로, 프로듀서는 ` numUsedBytes `를 증가시키고, ` numUsedBytes `가 반드시 0보다 크기 때문에 ` bufferNotEmpty ` 조건이 참임을 알립니다.

우리는 numUsedBytes 변수에 대한 모든 액세스를 뮤텍스로 보호합니다. 또한, QWaitCondition::wait() 함수는 뮤텍스를 인수로 받아들입니다. 이 뮤텍스는 스레드가 대기 상태로 들어가기 전에 잠금이 해제되고, 스레드가 깨어날 때 잠깁니다. 게다가, 경합 조건이 발생하지 않도록 잠금 상태에서 대기 상태로의 전환은 원자적으로 이루어집니다.

컨슈머 클래스

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

class Consumer : public QThread
{
public:
    explicit Consumer(QObject *parent = nullptr)
        : QThread(parent)
    {
    }

private:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            {
                const QMutexLocker locker(&mutex);
                while (numUsedBytes == 0)
                    bufferNotEmpty.wait(&mutex);
            }

            fprintf(stderr, "%c", buffer[i % BufferSize]);

            {
                const QMutexLocker locker(&mutex);
                --numUsedBytes;
                bufferNotFull.wakeAll();
            }
        }
        fprintf(stderr, "\n");
    }
};

이 코드는 프로듀서 클래스와 매우 유사합니다. 바이트를 읽기 전에 버퍼가 가득 찼는지 여부가 아니라 비어 있는지(numUsedBytes 가 0인지)를 확인하고, 비어 있다면 bufferNotEmpty 조건에서 대기합니다. 바이트를 읽은 후에는 ` numUsedBytes `를 증가시키는 대신 감소시키고, ` bufferNotEmpty ` 조건 대신 ` bufferNotFull ` 조건을 신호로 보냅니다.

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;
}

그렇다면 프로그램을 실행하면 어떤 일이 일어날까요? 처음에는 프로듀서 스레드만이 작업을 수행할 수 있으며, 컨슈머 스레드는 bufferNotEmpty 조건이 신호될 때까지 차단된 상태로 대기합니다(numUsedBytes 가 0인 경우). 프로듀서가 버퍼에 1바이트를 넣으면 numUsedBytes 가 0보다 엄격히 커지므로 bufferNotEmpty 조건이 신호됩니다. 이 시점에서 두 가지 상황이 발생할 수 있습니다. 소비자 스레드가 제어권을 인수하여 해당 바이트를 읽거나, 프로듀서가 두 번째 바이트를 생성하게 되는 경우입니다.

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

다만 이러한 이점이 항상 실현되는 것은 아니라는 점에 유의해야 합니다. 버퍼 블록( QMutex )을 잠그고 잠금을 해제하는 데는 비용이 발생합니다. 실제로는 버퍼를 여러 청크(chunk)로 나누어 개별 바이트 단위가 아닌 청크 단위로 처리하는 것이 더 효율적일 것입니다. 또한 버퍼 크기는 실험을 통해 신중하게 선택해야 하는 매개변수입니다.

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