Producer und Consumer mit Semaphoren

Der Producer-Thread schreibt Daten in den Puffer, bis er das Ende des Puffers erreicht, woraufhin er von vorne beginnt und die vorhandenen Daten überschreibt. Der Consumer-Thread liest die Daten, während sie produziert werden, und schreibt sie in den Standardfehler.

Semaphore ermöglichen ein höheres Maß an Gleichzeitigkeit als Mutexe. Wären 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 kann jedoch nicht schaden, wenn beide Threads gleichzeitig an verschiedenen Teilen des Puffers arbeiten.

Das Beispiel umfasst zwei Klassen: Producer und Consumer. Beide erben von QThread. Der zirkuläre Puffer, der für die Kommunikation zwischen diesen beiden Klassen verwendet wird, und die Semaphoren, die ihn schützen, sind globale Variablen.

Eine Alternative zur Verwendung von QSemaphore zur Lösung des Producer-Consumer-Problems ist die Verwendung von QWaitCondition und QMutex. Dies wird im Beispiel Producer und Consumer mit Wait Conditions gemacht.

Globale Variablen

Beginnen wir mit der Betrachtung des Ringpuffers und der zugehörigen Semaphoren:

constexpr int DataSize = 100000;

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

QSemaphore freeBytes(BufferSize);
QSemaphore usedBytes;

DataSize ist die Menge der Daten, die der Producer erzeugt. Um das Beispiel so einfach wie möglich zu halten, setzen wir sie als Konstante ein. BufferSize ist die Größe des Ringpuffers. Sie ist kleiner als DataSize, was bedeutet, dass der Producer irgendwann das Ende des Puffers erreicht und wieder von vorne beginnt.

Um den Producer und den Consumer zu synchronisieren, benötigen wir zwei Semaphoren. Die Semaphore freeBytes kontrolliert den "freien" Bereich des Puffers (den Bereich, den der Producer noch nicht mit Daten gefüllt hat oder den der Consumer bereits gelesen hat). Die Semaphore usedBytes steuert den "benutzten" Bereich des Puffers (der Bereich, den der Produzent gefüllt, der Konsument aber noch nicht gelesen hat).

Zusammen stellen die Semaphore sicher, dass der Producer nie mehr als BufferSize Bytes vor dem Consumer liegt und dass der Consumer nie Daten liest, die der Producer noch nicht erzeugt hat.

Die Semaphore freeBytes wird mit BufferSize initialisiert, da anfangs der gesamte Puffer leer ist. Die Semaphore usedBytes wird auf 0 initialisiert (der Standardwert, wenn keine Semaphore angegeben ist).

Producer-Klasse

Schauen wir uns den Code für die Klasse Producer 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 Producer erzeugt DataSize Bytes an Daten. Bevor er ein Byte in den Ringpuffer schreibt, muss er ein "freies" Byte mit Hilfe der Semaphore freeBytes erwerben. Der Aufruf QSemaphore::acquire() kann blockieren, wenn der Konsument nicht mit dem Produzenten Schritt halten kann.

Am Ende gibt der Producer ein Byte über das Semaphor usedBytes frei. Das "freie" Byte wurde erfolgreich in ein "benutztes" Byte umgewandelt und kann nun vom Verbraucher gelesen werden.

Verbraucher-Klasse

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, nur dass wir diesmal ein "gebrauchtes" Byte erfassen und ein "freies" Byte freigeben, statt umgekehrt.

Die main()-Funktion

In main() erstellen wir die beiden Threads und rufen QThread::wait() auf, um sicherzustellen, dass beide Threads Zeit haben, sich zu beenden, 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 nun, wenn wir das Programm ausführen? Zu Beginn ist der Producer-Thread der einzige, der etwas tun kann; der Consumer-Thread ist blockiert und wartet auf die Freigabe des usedBytes Semaphors (seine anfängliche available() Anzahl ist 0). Sobald der Producer ein Byte in den Puffer gelegt hat, ist freeBytes.available() BufferSize - 1 und usedBytes.available() ist 1. Zu diesem Zeitpunkt können zwei Dinge passieren: Entweder übernimmt der Consumer-Thread und liest dieses Byte, oder der Producer-Thread kann ein zweites Byte produzieren.

Das in diesem Beispiel vorgestellte Producer-Consumer-Modell macht es möglich, hochgradig nebenläufige Multithread-Anwendungen zu schreiben. Auf einer Multiprozessormaschine ist das Programm potenziell bis zu doppelt so schnell wie das entsprechende Mutex-basierte Programm, da die beiden Threads gleichzeitig auf verschiedenen Teilen des Puffers aktiv sein können.

Beachten Sie jedoch, dass diese Vorteile nicht immer zum Tragen kommen. Das Anfordern und Freigeben von QSemaphore ist mit Kosten verbunden. In der Praxis würde es sich wahrscheinlich lohnen, den Puffer in Chunks aufzuteilen und auf Chunks statt auf einzelnen Bytes zu arbeiten. Auch die Puffergröße ist ein Parameter, der sorgfältig und auf der Grundlage von Experimenten ausgewählt werden muss.

Beispielprojekt @ code.qt.io