Produzent und Verbraucher mit Semaphoren

Der Produzent schreibt Daten in den Puffer, bis er das Ende des Puffers erreicht. An diesem Punkt 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.

Semaphoren ermöglichen ein höheres Maß an Parallelität als Mutexe. Würden die Zugriffe auf den Puffer durch ein ` 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 Stellen 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 Semaphoren, die ihn schützen, sind globale Variablen.

Eine Alternative zur Verwendung von ` QSemaphore ` zur Lösung des Produzenten-Konsumenten-Problems ist die Verwendung von ` QWaitCondition ` und ` QMutex`. Genau das tut das Beispiel „Producer and Consumer using Wait Conditions “.

Globale Variablen

Betrachten wir zunächst den Ringpuffer und die zugehörigen Semaphoren:

constexpr int DataSize = 100000;

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

QSemaphore freeBytes(BufferSize);
QSemaphore usedBytes;

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 Semaphoren. Das Semaphor ` freeBytes ` steuert den „freien“ Bereich des Puffers (den Bereich, den der Produzent noch nicht mit Daten gefüllt hat oder den der Konsument bereits gelesen hat). Das Semaphor „ usedBytes “ steuert den „belegten“ Bereich des Puffers (den Bereich, den der Produzent gefüllt hat, den der Konsument aber noch nicht gelesen hat).

Zusammen stellen die Semaphoren sicher, 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 erzeugt hat.

Das Semaphor „ freeBytes “ wird mit „ BufferSize “ initialisiert, da der gesamte Puffer anfangs leer ist. Das Semaphor „ usedBytes “ wird auf 0 initialisiert (der Standardwert, wenn kein Wert angegeben wird).

Produzentenklasse

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

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

Der Produzent generiert „ DataSize “ Bytes an Daten. Bevor er ein Byte in den Ringpuffer schreibt, muss er sich mithilfe des Semaphors „ freeBytes “ ein „freies“ Byte sichern. Der Aufruf von „ QSemaphore::acquire()“ kann blockieren, wenn der Konsument mit dem Produzenten nicht Schritt gehalten hat.

Am Ende gibt der Produzent ein Byte über den Semaphor ` usedBytes ` frei. Das „freie“ Byte wurde erfolgreich in ein „belegtes“ Byte umgewandelt und steht nun dem Konsumenten zum Lesen bereit.

Konsumentenklasse

Wenden wir uns nun der Klasse ` Consumer ` zu:

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

Der Code ist dem des Produzenten sehr ähnlich, außer dass wir diesmal ein „belegtes“ Byte anfordern und ein „freies“ Byte freigeben, anstatt umgekehrt.

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 der Semaphor „ usedBytes “ freigegeben wird (sein anfänglicher Zählerstand von ` available()` ist 0). Sobald der Produzenten-Thread ein Byte in den Puffer geschrieben hat, ist „ freeBytes.available() “ gleich „ BufferSize “ minus 1 und „ usedBytes.available() “ gleich 1. An diesem Punkt können zwei Dinge geschehen: Entweder übernimmt der Verbraucher-Thread und liest dieses Byte, oder der Produzenten-Thread kann ein zweites Byte produzieren.

Das in diesem Beispiel vorgestellte Produzenten-Konsumenten-Modell ermöglicht die Erstellung 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 Erlangen und Freigeben eines „ QSemaphore “ ist mit einem Aufwand 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