1 producer, n consumer with QThread

Question

I am new to C++ and to threading, so I'm sorry if my code is bad. I'm here to get it better and to learn.

Any kind of advice (both on C++ style, on Qt usage, or on threading handling) is welcome. Please feel free to comment anything, also if the question has to be reformulated.

I have this problem:

• one thread that grab frames from a video/web cam
• some (let's say 3) threads that take the last grabbed frame, do some computer vision analysis, and display it back in the main gui. Each computation thread could have different computation time (because of computation-heavy algorithm like optical flow or similar)

There are few important things I want to care about:

1. the threads must run in parallel (that's why I used the threads)
2. threads takes only the last frame depending on their own job, so if a thread has some heavy optical flow computation, it will take only frames (let's say 1, 15, 21, 30). Instead, a light computation that does only frame differencing will take all frames 1, 2, 3, 4,..
3. I need a way to stop it and restart it. Let's say that I have 3 video to analyze, I want to do the first, wait for everything is finished, to the second, etc..

I have done a little prototype that works with int as a shared resource, and not with a frame. I will use opencv, cv::Mat that are basically containers for uchar array where each uchar represents a pixel.

I think that this is a working prototype, but I'm very scared about threading. I've never done something with multithreading, I don't understand the problems well, and I don't know how to test it correctly.

I'm going to copy the files here, but the project is also on bitbucket.

P.S. If it's worth something, I'm using Qt 4.8 with QtCreator 2.8 under Mac OS/X 10.8, but should be super cross-compiling as I'm just using Qt objects.

main.cpp:

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

int main(int argc, char *argv[])
{
    QApplication a(argc, argv);

    MainWidget w;
    w.show();

    return a.exec();
}

mainwidget.h:

#ifndef MAINWIDGET_H
#define MAINWIDGET_H

#include <QWidget>
#include <QLabel>
#include <QCloseEvent>
#include <QKeyEvent>
#include <QVBoxLayout>
#include "resources.h"
#include "producer.h"
#include "consumer.h"

class MainWidget : public QWidget
{
    Q_OBJECT

public:
    //------------------------------------------------------------------ ctor dtor
    explicit MainWidget(QWidget *parent = 0) : QWidget(parent) {
        initGUI();

        initProducer();
        initConsumers();
    }

    ~MainWidget()
    {
    }

    //------------------------------------------------------------------ 1 prod, n cons
    void initProducer()
    {
        producer = new Producer(&r);
        connect(producer, SIGNAL(newFrame(int)),
                this, SLOT(showProducer(int)));
        connect(producer, SIGNAL(finished()),
                producer, SLOT(deleteLater()));

        producer->start();
    }

    void initConsumers()
    {
        consumerCounter = 0;

        consumer1 = new Consumer(1, &r);
        connect(consumer1, SIGNAL(newComputation(int)),
                this, SLOT(showConsumer1(int)));
        connect(consumer1, SIGNAL(finished()),
                consumer1, SLOT(deleteLater()));
        connect(consumer1, SIGNAL(destroyed()),
                this, SLOT(incrementConsumerCounter()));

        consumer2 = new Consumer(2, &r);
        connect(consumer2, SIGNAL(newComputation(int)),
                this, SLOT(showConsumer2(int)));
        connect(consumer2, SIGNAL(finished()),
                consumer2, SLOT(deleteLater()));
        connect(consumer2, SIGNAL(destroyed()),
                this, SLOT(incrementConsumerCounter()));

        consumer3 = new Consumer(3, &r);
        connect(consumer3, SIGNAL(newComputation(int)),
                this, SLOT(showConsumer3(int)));
        connect(consumer3, SIGNAL(finished()),
                consumer3, SLOT(deleteLater()));
        connect(consumer3, SIGNAL(destroyed()),
                this, SLOT(incrementConsumerCounter()));

        consumer1->start();
        consumer2->start();
        consumer3->start();
    }

    void stopProducer() {
        producer->stopProduce();
    }

    void stopConsumers() {
        consumer1->stopConsume();
        consumer2->stopConsume();
        consumer3->stopConsume();
    }

    void stopAll()
    {
        stopProducer();
        stopConsumers();
        qDebug() << "wake all!";
        r.waitCondition.wakeAll();
    }

    //------------------------------------------------------------------ gui
    void initGUI() {
        QVBoxLayout *l = new QVBoxLayout;
        this->producerLabel = new QLabel;
        l->addWidget(producerLabel);
        this->consumer1Label = new QLabel;
        l->addWidget(consumer1Label);
        this->consumer2Label = new QLabel;
        l->addWidget(consumer2Label);
        this->consumer3Label = new QLabel;
        l->addWidget(consumer3Label);
        this->setLayout(l);
    }

    void keyPressEvent(QKeyEvent* event) {
        if(event->key() == Qt::Key_S) {
            qDebug() << "pressed s, stop all";
            stopAll();
        }
        if(event->key() == Qt::Key_R) {
            qDebug() << "pressed r, restart all";
            initProducer();
            initConsumers();
        }
    }

    void closeEvent(QCloseEvent * event)
    {
        emit shutdown();
        this->close();
    }

private:
    QLabel *producerLabel;
    QLabel *consumer1Label;
    QLabel *consumer2Label;
    QLabel *consumer3Label;

    Resources r;
    Producer *producer;
    Consumer *consumer1;
    Consumer *consumer2;
    Consumer *consumer3;

    int consumerCounter;

private slots:
    void showProducer(int n) {
        this->producerLabel->setText(QString("Producer: %1").arg(n));
    }
    void showConsumer1(int n) {
        this->consumer1Label->setText(QString("Consumer1: %1").arg(n));
    }
    void showConsumer2(int n) {
        this->consumer2Label->setText(QString("Consumer2: %1").arg(n));
    }
    void showConsumer3(int n) {
        this->consumer3Label->setText(QString("Consumer3: %1").arg(n));
    }

    void incrementConsumerCounter() {
        consumerCounter++;
        if(consumerCounter==3) {
            qDebug() << "..... could restart!";
            initProducer();
            initConsumers();
        }
    }

signals:
    void shutdown();
};

#endif // MAINWIDGET_H

resources.h:

#ifndef RESOURCES_H
#define RESOURCES_H

#include <QDebug>
#include <QReadWriteLock>
#include <QWaitCondition>

class Resources {

public:

    Resources() {
        qDebug() << "Resources > ctor";
        lastFrame = 0;
    }

    ~Resources() {
        qDebug() << "Resources > dtor";
    }

    int lastFrame;
    QReadWriteLock lastFrame_read;
    QReadWriteLock lastFrame_write;

    QWaitCondition waitCondition;

};

#endif // RESOURCES_H

producer.h:

#ifndef PRODUCER_H
#define PRODUCER_H

#include <QDebug>
#include <QThread>
#include <QMutex>
#include <QReadWriteLock>
#include <QWaitCondition>
#include "resources.h"

class Producer : public QThread
{
    Q_OBJECT
public:
    explicit Producer(Resources *r, QObject *parent = 0) :
        QThread(parent),
        abort(false)
    {
        qDebug() << "producer > ctor";
        frameCount = 0;
        this->r = r;
    }

    ~Producer() {
        qDebug() << "Producer dtor";
        stopProduce();
        wait();
    }

    void stopProduce() {
        abortMutex.lock();
        abort = true;
        abortMutex.unlock();
    }

protected:
    Resources *r;
    int frameCount;

    void produce()
    {
        frameCount++;
        r->lastFrame = frameCount;
        qDebug() << QString("Producing: %1").arg(frameCount);
        emit newFrame(frameCount);
    }

    void run()
    {
        forever {
            abortMutex.lock();
            if(abort) {
                abortMutex.unlock();
                break;
            } abortMutex.unlock();

            r->lastFrame_read.lockForRead();
            produce();
            r->waitCondition.wakeAll();
            r->lastFrame_read.unlock();

            msleep(200);
        }
        emit finished();
    }

private:
    QMutex abortMutex;
    bool abort;

signals:
    void newFrame(int);

};

#endif // PRODUCER_H

consumer.h:

#ifndef CONSUMER_H
#define CONSUMER_H

#include <QDebug>
#include <QThread>
#include <QMutex>
#include <QWaitCondition>
#include <QReadWriteLock>
#include <QString>
#include "resources.h"

class Consumer : public QThread
{
    Q_OBJECT

public:
    Consumer(int id, Resources *r, QObject *parent=0):
        QThread(parent),
        abort(false)
    {
        this->r = r;
        this->id = id;
    }
    ~Consumer() {
        qDebug() << QString("Consumer %1 dtor").arg(id);
        stopConsume();
        wait();
    }

    void stopConsume() {
        abortMutex.lock();
        abort = true;
        abortMutex.unlock();
    }

protected:
    Resources *r;
    int id;

    int consumeMessage(int val)
    {
        qDebug() << QString("#%1 Consuming : %2").arg(id).arg(val);
        msleep(1000*id);
        emit newComputation(val);

        return val;
    }

    void run()
    {
        forever {
            r->lastFrame_read.lockForRead();
            consumeMessage(r->lastFrame);
            r->lastFrame_read.unlock();

            // I think that this check MUST be done here:
            // between the consume and the wait
            abortMutex.lock();
            if(abort) {
                abortMutex.unlock();
                break;
            } abortMutex.unlock();

            qDebug() << QString("Consumer %1: going to sleep").arg(id);
            r->waitCondition.wait(&r->lastFrame_read);
            qDebug() << QString("Consumer %1: awake").arg(id);
        }
        emit finished();
    }

private:
    QMutex abortMutex;
    bool abort;

signals:
    void newComputation(int);
    void finished();
};

#endif // CONSUMER_H

Answer 1

First of all: your code is not bad; I agree with Wayne Conrad.

I have a minor remark about the connect statement. You are currently using Qt 4.8, so this is fine as it is. However with Qt 5 a new signal/slot syntax was introduced.

Your question about threading: Don't be afraid of QThread, they're just threads too. The problem with QThread is only this: a lack of consistent documentation. You have probably looked up the documentation of QThread and ended up subclassing QThread. There's nothing seriously wrong with this approach but if you want to use the event loop it becomes difficult.

I ended up in exactly the same situation in one of my projects. Then I came across a very interesting post of Maya Posch. You may also want to read Bradley T. Hughes' post - one of the Qt core developers - where he admits the mistakes of the past (and even takes blame for it: i take my hat off to so much honesty). Debao's Blog has a good overview of the QThread history and how to do it wrong and right. (Part 2 here). He has a short conclusion:

• Subclass QThread and reimplement its run() function is intuitive and there are still many perfectly valid reasons to subclass QThread, but when event loop is used in worker thread, it's not easy to do it in the right way.
• Use worker objects by moving them to the thread is easy to use when event loop exists, as it has hidden the details of event loop and queued connection.

To summarize, you would subclass from QObject instead of QThread:

class Producer : public QObject
{
    Q_OBJECT
public:
    explicit Producer(Resources *r, QObject *parent = 0) 
    // ...

and then create your producer this way:

QThread* producerthread = new QThread;
Producer* producer = new Producer();
producer->moveToThread(producerthread);
connect(producer, SIGNAL(error(QString)), this, SLOT(errorString(QString)));
connect(producerthread, SIGNAL(started()), producer, SLOT(process()));
connect(producer, SIGNAL(finished()), producerthread, SLOT(quit()));
connect(producer, SIGNAL(finished()), producer, SLOT(deleteLater()));
connect(producerthread, SIGNAL(finished()), producerthread, SLOT(deleteLater()));

connect(producer, SIGNAL(newFrame(int)), this, SLOT(showProducer(int)));

producerthread->start();