QtThread: worker for I/O queue

Question

I'm trying to create my own worker thread for I/O operations. As far as I know the low-level management of the serial port is already done in a separate thread, but I want to put my code in another one to handle timeouts and other middleware stuff.

Here the complete code of my class. Please note it's my first attempt with Qt's multi-threading.

EDIT

The code is updated without sub-classing QThread

#ifndef WORKERSERIAL_H
#define WORKERSERIAL_H

#include <QObject>
#include <QThread>
#include <QQueue>
#include <QMutex>
#include <QTimer>
#include <QDebug>

#include "myserial.h"

class WorkerSerial : public QObject
{
    Q_OBJECT

public:
    explicit WorkerSerial(QObject *parent = 0) : QObject(parent)
    {
        m_serial = new MySerial(this);
        connect(m_serial, &MySerial::lineReceived, this, &WorkerSerial::serial_LineReceived);
        m_stop = false;
    }

    bool open(QString port, quint32 baudrate) { return m_serial->open(port, baudrate); }
    bool isOpen() { return m_serial->serial()->isOpen(); }
    QStringList ports() { return m_serial->ports(); }

private:
    MySerial *m_serial;
    QQueue<QString> m_queue;
    QMutex m_mutex;
    bool m_stop;

private slots:
    void serial_LineReceived(QByteArray line)
    {
        emit lineReceived(line);
    }

signals:
    void lineReceived(QByteArray line);

public slots:
    void close() { m_serial->close(); }
    void send(QString data) { m_mutex.lock(); m_queue.enqueue(data); m_mutex.unlock(); }
    void stop() { m_mutex.lock(); m_stop = true; m_mutex.unlock(); }
    void run() {
        forever {
            m_mutex.lock();
            if (m_stop)
            {
                qDebug() << "Closing...";
                QTimer::singleShot(0, this, SLOT(close()));
                while(m_serial->serial()->isOpen());
                m_mutex.unlock();
                return;
            }

            if (!m_queue.isEmpty())
            {
                QString data = m_queue.dequeue();
                qDebug() << m_serial->sendMessage(data.toLocal8Bit(), true);
            }
            m_mutex.unlock();
        }
    }

};

#endif // WORKERSERIAL_H

The MySerial class it's a convenient wrapper for QSerialPort. Here the relevant functions:

bool MySerial::open(QString port, quint32 baudrate) {
    m_serial->setPortName(port);
    if (!m_serial->open(QIODevice::ReadWrite)) return false;
    m_serial->setDataBits(QSerialPort::Data7);
    m_serial->setParity(QSerialPort::EvenParity);
    m_serial->setStopBits(QSerialPort::TwoStop);
    m_serial->setBaudRate(baudrate);
    m_serial->setFlowControl(QSerialPort::NoFlowControl);
    return true;
}

QByteArray MySerial::sendMessage(QByteArray data) {
    m_serial->write(data);
    return data;
}

Finally, here how I start, use and close the worker:

QThread *workerThread = new QThread;
m_worker = new WorkerSerial;
m_worker->moveToThread(workerThread);
workerThread->start();
QTimer::singleShot(0, m_workerDMX, SLOT(run()));
// ...
m_worker->open("COM1", 250000));
// ...
m_worker->send("Hello World!");
// ...
m_worker->stop();
workerThread->quit();
workerThread->wait(1000);
// Here the application ends

This is the output:

QObject: Cannot create children for a parent that is in a different thread.
(Parent is QSerialPort(0x19882160), parent's thread is QThread(0x187fe598), current thread is WorkerSerial(0x19d4c9b8)
[this error appears when it calls the SendMessage() function]
"Hello World!"
Closing...
QMutex: destroying locked mutex
QThread: Destroyed while thread is still running
[the two errors above are due the singleSlot() isn't executed and thus the serial won't close]
Invalid parameter passed to C runtime function.
Invalid parameter passed to C runtime function.

It seems clear I've messed up something! Would you help me to understand my mistakes?

Answer 1

Here is (an incomplete) implementation, which should give you an idea of how to simplify your code. Some additional notes are below:

class WorkerSerial : public QObject
{
    Q_OBJECT

public:
    explicit WorkerSerial(QObject *parent = 0) : QObject(parent)
    {
        // no need for slot -- connect directly to our signal
        connect(&m_serial, &MySerial::lineReceived, this, &WorkerSerial::lineReceived);
    }

    bool open(QString port, quint32 baudrate) { return m_serial.open(port, baudrate); }
    bool isOpen() { return m_serial.serial().isOpen(); }
    QStringList ports() { return m_serial.ports(); }

private:
    // doesn't need to be a pointer
    MySerial m_serial;

signals:
    void lineReceived(const QByteArray& line);

public slots:
    void close() { m_serial.close(); }

    void send(const QByteArray& data)
    {
        // "lengthy" function call
        qDebug() << m_serial.sendMessage(data, true);
    }
};

class Sender : public QObject
{
    Q_OBJECT

public:
    void send(const QByteArray& data) { emit sig_send(data); }

signals:
    void sig_send(const QByteArray& data);
}

...

WorkerSerial m_worker;
m_worker.open("COM1", 250000));

QThread m_thread;
m_worker.moveToThread(&m_thread);
m_thread.start();

Sender m_sender;
QObject::connect(&m_sender, &Sender::sig_send, &m_worker, &WorkerSerial::send, Qt::QueuedConnection);

m_sender.send("Hello World!");

m_thread.quit();
m_thread.wait();

m_worker.close();

Additional notes:

1. I've added a Sender class that queues calls to WorkerSerial::send() on a different thread. If you have a signal in your code, which triggers calls to WorkerSerial::send() you don't need this class.

2. Once you move m_worker to another thread, it is unsafe to access any of its methods, unless (a) they are guaranteed to be thread-safe; or (b) you serialize (protect) access to m_serial with a mutex.

Answer 2

NOTE: this is not a fully working solution (yet). But I think it's better to post it as an answer than add another piece of code in the original question.

I managed to a different approach. Perhaps it looks weird... but it should have some advantages:

• it guarantees the order of the messages to be transmitted
• using a QQueue I can know the queue size or other information about
• the "variable shadowing" mechanism should allow to safely communicate with the thread

Here the code, below the two last issues:

#ifndef WORKERSERIAL_H
#define WORKERSERIAL_H

#include <QObject>
#include <QThread>
#include <QQueue>
#include <QMutex>
#include <QDebug>

#include "myserial.h"

class WorkerSerial : public QThread
{
    Q_OBJECT

public:
    explicit WorkerSerial(QObject *parent = 0, int timeout = 0) : QThread(parent), timeout(timeout)
    {
        quit = false;
        command = None;
    }

    ~WorkerSerial()
    {
        mutex.lock();
        quit = true;
        mutex.unlock();
        wait();
    }

    void open(QString port, quint32 baudrate)
    {
        mutex.lock();
        this->port = port;
        this->baudrate = baudrate;
        command = Open;
        mutex.unlock();
    }

    void close()
    {
        mutex.lock();
        command = Close;
        mutex.unlock();
    }

    void stop()
    {
        mutex.lock();
        command = Quit;
        mutex.unlock();
    }

private:
    enum Commands
    {
        None,
        Open,
        Close,
        Quit
    };

    QQueue<QString> m_queue;
    QMutex mutex;
    int timeout;
    int command;
    QString port;
    int baudrate;
    QString request;
    bool quit;

signals:
    void portStateChanged(bool isOpen);
    void lineReceived(QByteArray line);

public slots:
    void send(QString data)
    {
        mutex.lock();
        m_queue.enqueue(data);
        mutex.unlock();
    }

    void run()
    {
        MySerial serial;
        QString _port;
        int _baudrate;
        int _timeout;
        QString _request = "";
        int _command = None;

        connect(&serial, &MySerial::lineReceived, this, &WorkerSerial::lineReceived);
        mutex.lock();
        _timeout = timeout;
        mutex.unlock();

        forever {
            mutex.lock();
            _command = command;         
            command = None;
            if (!m_queue.isEmpty()) _request = m_queue.dequeue();
            mutex.unlock();

            switch (_command) {
            case Open:
                mutex.lock();
                _port = port;
                _baudrate = baudrate;
                mutex.unlock();
                if (serial.open(_port, _baudrate)) emit portStateChanged(true);
                else portStateChanged(false);
                break;

            case Close:
                serial.close();
                emit portStateChanged(serial.serial()->isOpen());
                break;

            case Quit:
                serial.close();
                return;
                break;

            default:
                break;
            }

            if (!_request.isEmpty())
            {
                qDebug() << serial.sendMessage(_request.toLocal8Bit(), true);
                _request = "";
            }
            msleep(1);
        }
    }

};

#endif // WORKERSERIAL_H

Here how to use it:

WorkerSerial *worker = new WorkerSerial(this);
connect(worker , &WorkerSerial::lineReceived, this, &MainWindow::lineReceived);
connect(worker , &WorkerSerial::portStateChanged, this, &MainWindow::portStateChanged);
worker->start();
// ..
worker->open("COM2", 250000);
worker->send("Hello World");
// ..
worker->stop();
worker->wait(1000);

It "works" but there are two main issues:

1. the serialport is opened but the data are not actually transmitted. The MySerial class is working because if I do the following, bypassing the WorkerSerial:

---

MySerial serial;
serial.open("COM2", 250000);
serial.sendMessage("Hello World!", true);

the data is sent! Hence there is still something wrong in my WorkerSerial.

2. due to the forever loop, I need to insert a small delay, otherwise the CPU will run up to 100% doing nothing.