Qt5 Tutorial QThreads and QSemaphore for Producer and Consumer - 2020

In this tutorial, we will learn how we can use QSemaphore for Producer and Consumer.

This Semaphores tutorial will show how to use QSemaphore to control access to a circular buffer shared by a producer thread and a consumer thread.

Video recording - Producer and Consumer model using QSemaphore

The producer writes data to the buffer until it reaches the end of the buffer, at which point it restarts from the beginning, overwriting existing data. The consumer thread reads the data as it is produced and writes it to standard error.

Semaphores make it possible to have a higher level of concurrency than mutexes. If accesses to the buffer were guarded by a QMutex, the consumer thread couldn't access the buffer at the same time as the producer thread. Yet, there is no harm in having both threads working on different parts of the buffer at the same time.
- from http://qt-project.org/doc/qt-5.0/qtcore/semaphores.html

The example comprises mainly three classes: Producer, Consumer, and the QDialog class. Both inherit from QThread. The circular buffer used for communicating between these two classes and the semaphores that protect it are global variables.

The use of semaphores ensure that the producer is never more than BufferSize bytes ahead of the consumer, and that the consumer never reads data that the producer hasn't generated yet. In other words, we use QSemaphore to control access to a circular buffer shared by a producer thread and a consumer thread.

To achieve the goal, we use global variables:

const int DataSize = 100000;

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

QSemaphore freeBytes(BufferSize);
QSemaphore usedBytes;

DataSize is the amout of data that the producer will generate. To keep the example as simple as possible, we make it a constant. BufferSize is the size of the circular buffer. It is less than DataSize, meaning that at some point the producer will reach the end of the buffer and restart from the beginning.

To synchronize the producer and the consumer, we need two semaphores. The freeBytes semaphore controls the "free" area of the buffer (the area that the producer hasn't filled with data yet or that the consumer has already read). The usedBytes semaphore controls the "used" area of the buffer (the area that the producer has filled but that the consumer hasn't read yet).

The freeBytes semaphore is initialized with BufferSize, because initially the entire buffer is empty. The usedBytes semaphore is initialized to 0 (the default value if none is specified).

Producer class looks like this:

// producer.cpp

#include "producer.h"
#include "common.h"

Producer::Producer(QObject *parent) :
    QThread(parent)
{
}

void Producer::run()
{
    qsrand(QTime(0,0,0).secsTo(QTime::currentTime()));
    for (int i = 0; i < DataSize; ++i) {
        freeBytes.acquire();
        buffer[i % BufferSize] = "ACGT"[(int)qrand() % 4];
        usedBytes.release();
        if(i % 20 == 0)
        emit bufferFillCountChanged(usedBytes.available());
        emit producerCountChanged(i);
    }
}

The producer generates DataSize bytes of data. Before it writes a byte to the circular buffer, it must acquire a "free" byte using the freeBytes semaphore. The QSemaphore::acquire() call might block if the consumer hasn't kept up the pace with the producer.

At the end, the producer releases a byte using the usedBytes semaphore. The "free" byte has successfully been transformed into a "used" byte, ready to be read by the consumer.

The usedBytes.available() is for UI purpose. It returns the available bytes for consumer.

This class emits two signals: one for the number of bytes produced so far, and the level of bytes available to consume.

The code is almost symmetric to the Producer class, except that this time we acquire a "used" byte and release a "free" byte, instead of the opposite.

// consumer.cpp

#include "consumer.h"
#include "common.h"

Consumer::Consumer(QObject *parent) :
    QThread(parent)
{
}

void Consumer::run()
{
    for (int i = 0; i < DataSize; ++i) {
        usedBytes.acquire();
        fprintf(stderr, "%c", buffer[i % BufferSize]);
        freeBytes.release();
        emit bufferFillCountChanged(usedBytes.available());
        emit consumerCountChanged(i);
    }
    fprintf(stderr, "\n");
}

When the program fires up QDialog, two new QThreads are created in the constructor. Then, when the Start button is clicked, the two threads call start().

// conproddialog.cpp

#include "conproddialog.h"
#include "ui_conproddialog.h"
#include "myConstants.h"

// BufferSize: maximum bytes that can be stored
char buffer[BufferSize];

QSemaphore freeBytes(BufferSize);
QSemaphore usedBytes;

ConProdDialog::ConProdDialog(QWidget *parent) :
    QDialog(parent),
    ui(new Ui::ConProdDialog)
{
    ui->setupUi(this);

    // progress bar range setup
    ui->producerProgressBar->setRange(0, DataSize);
    ui->consumerProgressBar->setRange(0, DataSize);
    ui->bufferProgressBar->setRange(0, BufferSize);

    // make two threads
    mProducer = new Producer(this);
    mConsumer = new Consumer(this);

    // connect signal/slot for the buffer progress bar
    connect(mConsumer, SIGNAL(bufferFillCountChanged(int)),
              this, SLOT(onBufferValueChanged(int)));
    connect(mProducer, SIGNAL(bufferFillCountChanged(int)),
              this, SLOT(onBufferValueChanged(int)));

    // connect signal/slot for consumer/producer progress bar
    connect(mConsumer, SIGNAL(consumerCountChanged(int)),
              this, SLOT(onConsumerValueChanged(int)));
    connect(mProducer, SIGNAL(producerCountChanged(int)),
              this, SLOT(onProducerValueChanged(int)));
}

ConProdDialog::~ConProdDialog()
{
    delete ui;
}

void ConProdDialog::onBufferValueChanged(int bCount)
{
    ui->bufferProgressBar->setValue(bCount);
}

void ConProdDialog::onProducerValueChanged(int pCount)
{
    ui->producerProgressBar->setValue(pCount);
}

void ConProdDialog::onConsumerValueChanged(int cCount)
{
    ui->consumerProgressBar->setValue(cCount);
}

// start button clicked
void ConProdDialog::on_startButton_clicked()
{
    // disable the start button
    ui->startButton->setEnabled(false);

    // threads starat
    mProducer->start();
    mConsumer->start();
}

The UI looks like this:

Here is the source file: ConProd2.zip.

Here are the other code not shown in this page:

main.cpp:

#include "conproddialog.h"
#include <QApplication>
#include "conproddialog.h"

int main(int argc, char *argv[])
{
    QApplication a(argc, argv);
    ConProdDialog w;
    w.setWindowTitle("Semaphore: Consumer & Producer");
    w.show();

    return a.exec();
}

common.h:

#ifndef COMMON_H
#define COMMON_H

#include <QSemaphore>
#include "myConstants.h"

extern char buffer[BufferSize];

extern QSemaphore freeBytes;
extern QSemaphore usedBytes;

#endif // COMMON_H

myConstants.h:

#ifndef MYCONSTANTS_H
#define MYCONSTANTS_H
const int DataSize = 100000;
const int BufferSize = 8192;
#endif // MYCONSTANTS_H

conproddialog.h:

#ifndef CONPRODDIALOG_H
#define CONPRODDIALOG_H

#include <QDialog>
#include "consumer.h"
#include "producer.h"

#include <QSemaphore>


namespace Ui {
class ConProdDialog;
}

class ConProdDialog : public QDialog
{
    Q_OBJECT

public:
    explicit ConProdDialog(QWidget *parent = 0);
    ~ConProdDialog();

public slots:
    void onBufferValueChanged(int);
    void onProducerValueChanged(int);
    void onConsumerValueChanged(int);

private slots:
    void on_startButton_clicked();

private:
    Ui::ConProdDialog *ui;
    Producer *mProducer;
    Consumer *mConsumer;
};

#endif // CONPRODDIALOG_H

producer.h:

#ifndef PRODUCER_H
#define PRODUCER_H

#include <QThread>
#include <QTime>

class Producer : public QThread
{
    Q_OBJECT
public:
    explicit Producer(QObject *parent = 0);
    void run();

signals:
    void bufferFillCountChanged(int bCount);
    void producerCountChanged(int count);

public slots:
};

#endif // PRODUCER_H

consumer.h:

#ifndef CONSUMER_H
#define CONSUMER_H

#include <QThread>
#include <QTime>


class Consumer : public QThread
{
    Q_OBJECT
public:
    explicit Consumer(QObject *parent = 0);
    void run();

signals:
    //void stringConsumed(const QString &text;);
    void bufferFillCountChanged(int cCount);
    void consumerCountChanged(int count);
public slots:
};

#endif // CONSUMER_H

And finally, .pro:

ConProd2.pro:

QT       += core gui

greaterThan(QT_MAJOR_VERSION, 4): QT += widgets

TARGET = ConProd2
TEMPLATE = app


SOURCES += main.cpp\
        conproddialog.cpp \
    consumer.cpp \
    producer.cpp

HEADERS  += conproddialog.h \
    consumer.h \
    producer.h \
    common.h \
    myConstants.h

FORMS    += conproddialog.ui