Produzent und Konsument mit Wartevorgaben

Der Produzent schreibt Daten in den Puffer, bis er das Ende des Puffers erreicht; an dieser Stelle beginnt er wieder von vorne und überschreibt dabei bereits vorhandene Daten. Der Verbraucher-Thread liest die Daten, sobald sie erzeugt werden, und schreibt sie in die Standardfehlerausgabe.

Wartebedingungen ermöglichen ein höheres Maß an Parallelität, als es mit Mutexen allein möglich ist. Würden Zugriffe auf den Puffer lediglich durch einen ` QMutex` geschützt, könnte der Consumer-Thread nicht gleichzeitig mit dem Producer-Thread auf den Puffer zugreifen. Es schadet jedoch nicht, wenn beide Threads gleichzeitig an verschiedenen Teilen des Puffers arbeiten.

Das Beispiel umfasst zwei Klassen: ` Producer ` und ` Consumer`. Beide erben von ` QThread`. Der für die Kommunikation zwischen diesen beiden Klassen verwendete Ringpuffer und die Synchronisationswerkzeuge, die ihn schützen, sind globale Variablen.

Eine Alternative zur Verwendung von ` QWaitCondition ` und ` QMutex ` zur Lösung des Produzent-Konsument-Problems ist die Verwendung von ` QSemaphore`. Genau das tut das Beispiel „Produzent und Konsument mit Semaphoren “.

Globale Variablen

Betrachten wir zunächst den Ringpuffer und die dazugehörigen Synchronisationswerkzeuge:

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 ist die Datenmenge, die der Produzent generieren wird. Um das Beispiel so einfach wie möglich zu halten, legen wir sie als Konstante fest. ` BufferSize ` ist die Größe des Ringpuffers. Sie ist kleiner als ` DataSize`, was bedeutet, dass der Produzent irgendwann das Ende des Puffers erreicht und von vorne beginnt.

Um den Produzenten und den Konsumenten zu synchronisieren, benötigen wir zwei Wartebedingungen und einen Mutex. Die Bedingung „ bufferNotEmpty “ wird ausgelöst, wenn der Produzent Daten generiert hat, und teilt dem Konsumenten mit, dass er mit dem Lesen beginnen kann. Die Bedingung „ bufferNotFull “ wird ausgelöst, wenn der Konsument Daten gelesen hat, und teilt dem Produzenten mit, dass er weitere Daten generieren kann. Der „ numUsedBytes “ ist die Anzahl der Bytes im Puffer, die Daten enthalten.

Zusammen sorgen die Wartebedingungen, der Mutex und der Zähler „ numUsedBytes “ dafür, dass der Produzent dem Konsumenten nie um mehr als „ BufferSize “ Bytes voraus ist und dass der Konsument niemals Daten liest, die der Produzent noch nicht generiert hat.

Produzentenklasse

Sehen wir uns den Code für die Klasse „ Producer “ noch einmal an:

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

Der Produzent generiert DataSize Bytes an Daten. Bevor er ein Byte in den Ringpuffer schreibt, muss er zunächst prüfen, ob der Puffer voll ist (d. h., ob numUsedBytes gleich BufferSize ist). Ist der Puffer voll, wartet der Thread auf die Bedingung „ bufferNotFull “.

Am Ende erhöht der Produzent den Wert von ` numUsedBytes ` und signalisiert, dass die Bedingung ` bufferNotEmpty ` wahr ist, da ` numUsedBytes ` zwangsläufig größer als 0 ist.

Wir sichern alle Zugriffe auf die Variable ` numUsedBytes ` mit einem Mutex ab. Zudem akzeptiert die Funktion ` QWaitCondition::wait()` einen Mutex als Argument. Dieser Mutex wird entsperrt, bevor der Thread in den Ruhezustand versetzt wird, und gesperrt, wenn der Thread wieder aufwacht. Darüber hinaus erfolgt der Übergang vom gesperrten Zustand in den Wartezustand atomar, um das Auftreten von Race-Bedingungen zu verhindern.

Verbraucherklasse

Wenden wir uns nun der Klasse ` Consumer ` zu:

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

Der Code ist dem des Produzenten sehr ähnlich. Bevor wir das Byte lesen, prüfen wir, ob der Puffer leer ist (numUsedBytes ist 0), anstatt zu prüfen, ob er voll ist, und warten auf die Bedingung ` bufferNotEmpty `, falls er leer ist. Nachdem wir das Byte gelesen haben, verringern wir den Wert von „ numUsedBytes “ (anstatt ihn zu erhöhen) und lösen die Bedingung „ bufferNotFull “ aus (anstelle der Bedingung „ bufferNotEmpty “).

Die Funktion `main()`

In ` main()` erstellen wir die beiden Threads und rufen ` QThread::wait()` auf, um sicherzustellen, dass beide Threads Zeit haben, ihre Ausführung abzuschließen, bevor wir das Programm beenden:

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

Was passiert also, wenn wir das Programm ausführen? Zu Beginn ist der Produzenten-Thread der einzige, der etwas tun kann; der Konsumenten-Thread ist blockiert und wartet darauf, dass die Bedingung „ bufferNotEmpty “ signalisiert wird (numUsedBytes ist 0). Sobald der Produzent ein Byte in den Puffer geschrieben hat, ist numUsedBytes streng größer als 0, und die Bedingung „ bufferNotEmpty “ wird signalisiert. An diesem Punkt können zwei Dinge geschehen: Entweder übernimmt der Consumer-Thread und liest dieses Byte, oder der Producer kann ein zweites Byte produzieren.

Das in diesem Beispiel vorgestellte Produzenten-Konsumenten-Modell ermöglicht die Entwicklung hochgradig parallelisierter Multithread-Anwendungen. Auf einem Multiprozessor-Rechner ist das Programm potenziell bis zu doppelt so schnell wie das entsprechende Mutex-basierte Programm, da die beiden Threads gleichzeitig an verschiedenen Stellen des Puffers aktiv sein können.

Beachten Sie jedoch, dass diese Vorteile nicht immer zum Tragen kommen. Das Sperren und Entsperren eines „ QMutex “ ist mit einem Overhead verbunden. In der Praxis lohnt es sich wahrscheinlich, den Puffer in Blöcke zu unterteilen und mit diesen Blöcken statt mit einzelnen Bytes zu arbeiten. Auch die Puffergröße ist ein Parameter, der auf der Grundlage von Experimenten sorgfältig ausgewählt werden muss.

Beispielprojekt @ code.qt.io