대기 조건을 사용하는 생산자와 소비자

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

대기 조건을 사용하면 뮤텍스만으로 가능한 것보다 더 높은 수준의 동시성을 확보할 수 있습니다. 버퍼에 대한 액세스가 단순히 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 조건이 시그널링됩니다. 이 시점에서 두 가지 일이 일어날 수 있습니다: 소비자 스레드가 인계받아 해당 바이트를 읽거나 생산자가 두 번째 바이트를 생성하게 됩니다.

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

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

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