Worker threads with shared resources in Qt application

mkaze

I am working on a Qt application which involves serial communication with one or multiple devices. There are different procedures that can be executed simulteanously and each procedure may send one or unknown number of commands to a device and may receive data in response. To make it more clear, here is a graphical illustration of the scenario:

Clicking on a button triggers the execution of the corresponding procedure. So two or more different procedures may be running at the same time when the user clicks on two or more buttons in a short interval. Actually the only thing that may be shared between them is the serial communication with a single device; otherwise they are mostly independent of one another. And here are two pseudo-code examples of what a procedure may look like:

Procedure A:

begin
write command a1 on serial port
wait for one second
perform some computations
write command a2 on serial port
wait for one second
end

Procedure B:

begin
while true:
    write command b1 on serial port
    read the response from serial port
    perform some computations
    if a condition holds return, otherwise continue
end

My solution and its issue:

To simplify the situation consider that there is only one device which we need to communicate with. Since procedures can be executed simulteanously (and only one of them can communicate with the device through serial port at a time) I have created one thread and one worker class for each of the procedures and have moved the workers to their corresponding threads. To synchronize procedures when accessing the serial port I have created one mutex:

MainWindow.h

class MainWindow : public QMainWindow {

public:
    //...
    QSerialPort*	serial_;
    QMutex		serial_mutex_;

private:
    //...
    ProcAWorker*	proca_worker;
    ProcBWorker*	procb_worker;
    ProcCWorker*	procc_worker;
    ProcDWorker*	procd_worker;

    QThread		proca_thread;
    QThread		procb_thread;
    QThread		procc_thread;
    QThread		procd_thread;

}

MainWindow.cpp

void MainWindow::onConnectButtonClicked()
{
    serial_ = new QSerialPort();
    // configure serial port settings
    
    serial_->open(QIODevice::ReadWrite);
}

void MainWindow::onButtonAClicked()
{
    proca_worker = new ProcAWorker(0, this);   // pass a pointer to this class to be able to access its methods and members
    proca_worker->moveToThread(&proca_thread);
    
    // setup worker-thread connections: started, quit, finished, etc.

    proca_thread.start();    // triggers `proccess` slot in proca_worker
}

// same thing for other buttons and procedures

ProcAWorker.cpp

void ProcAWorker::ProcAWorker(QObject *parent, QMainWindow *wnd) :
	QObject(parent), wnd_(wnd)
{

}

void ProcAWorker::process()
{
    wnd_->serial_mutex_->lock();
    wnd_->serial_->write('Command a1');   // Warning occurs in this line
    bool write_ok = client_->serial_->waitForBytesWritten(SERIAL_WRITE_TIMEOUT);
    wnd_->serial_mutex_->unlock();
    
    QThread::sleep(1);
    // perform some computations
    
    wnd_->serial_mutex_->lock();
    wnd_->serial_->write('Command a2');
    bool write_ok = client_->serial_->waitForBytesWritten(SERIAL_WRITE_TIMEOUT);
    wnd_->serial_mutex_->unlock();

    if (write_ok) {
        // signal successful to main window
        emit success();
    }
}

However, when the write operation is performed on the serial port (i.e. wnd_->serial_->write('Command a1');) the following warning is shown:

QObject: Cannot create children for a parent that is in a different
thread. (Parent is QSerialPort(0x18907d0), parent's thread is
QThread(0x13cbc50), current thread is QThread(0x17d8d08)

My questions:

I have already looked at questions on Stackoverflow regarding this warning, but their answers have only mentioned that signal/slot should be used. I am familiar with using signal/slot to communicate with worker threads. However, I can't figure out how to implement my specific scenario (simultaneous running procedures with shared resources like serial port) using signal/slot or how can I modify my current solution to resolve this issue? Note that the procedures should be allowed to run in parallel (unless in those moments when they want to communicate with the device). Obviously one can run the procedures sequentially (i.e. one after another) but I am not looking for such solutions.
Actually there is also a "Halt" button that stops all the running procedures and sends a halt command to the device. But I could not figure out to implement this functionality as well (set a flag, send a quit signal, etc.). Could you please give me some hints in this regards as well?

VRonin

@mkaze said in Worker threads with shared resources in Qt application:

but their answers have only mentioned that signal/slot should be used.

And they are correct. This is the S of S.O.L.I.D.. the worker shouldn't care about writing to serial, that's not its job, it just have to dump the data to someone else and carry on.
To give you a rule of thumb if you find yourself doing something like:

proca_worker = new ProcAWorker(0, this);   // pass a pointer to this class to be able to access its methods and members

your design is wrong.

in ProcAWorker create a Q_SIGNAL void sendDataToSerial(const QByteArray& data); then in MainWindow add connect(proca_worker ,&ProcAWorker::sendDataToSerial,this,[=](const QByteArray& data){if(serial_) serial_->write(data);}); no need for mutex or wait conditions

Actually there is also a "Halt" button that stops all the running procedures and sends a halt command to the device.

for this it really depends on what the worker does. you should have a slot in the worker that stops everything and connect it the halt signal. however, for it to work properly you either need not to block the event loop of the secondary thread or to call QCoreApplication::processEvents() every once in a while.

P.S.
in MainWindow::onConnectButtonClicked you are leaking serial_ if that method is executed more than once

mkaze

@VRonin Thanks. You are right that by using signal/slots then there is no need to use mutex since the requests of serial write or serial read is queued in the event loop of MainWindow and processed sequentially (i.e. only the request of one procedure is handled at a time and therefore there is no race condition here).
Actually, this solution came to my mind but I thought it would make implementing the procedures fairly complicated. Consider the code for ProcAWorker :

void ProcAWorker::process()
{
    wnd_->serial_mutex_->lock();
    wnd_->serial_->write('Command a1');   // Warning occurs in this line
    bool write_ok = client_->serial_->waitForBytesWritten(SERIAL_WRITE_TIMEOUT);
    wnd_->serial_mutex_->unlock();
    
    QThread::sleep(1);
    // perform some computations
    
    wnd_->serial_mutex_->lock();
    wnd_->serial_->write('Command a2');
    bool write_ok = client_->serial_->waitForBytesWritten(SERIAL_WRITE_TIMEOUT);
    wnd_->serial_mutex_->unlock();

    if (write_ok) {
        // signal successful to main window
        emit success();
    }
}

To use signal and slots it seems to me that the following considerations should be taken into account:
1- This way the workflow of procedure workers would be asynchronous I think. For example considering the code above, first a sendDataToSerial signal is emitted. Then from the corresponding slot in MainWindow the result of writing to serial port should be sent back to the procedure by emitting another signal. So the ProcAWorker would need to store its state i.e. what has been done (the first command has been sent) and what needs to be done (is waiting for the result of first write and also the second command needs to be sent). I think it should be implemented as a state machine, right? If that's the case then implementing complex procedures (like the procedure B I mentioned in my question) would be much more complicated.

2- We need to send back the result of writing to or reading from serial port to the corresponding procedure. One workaround is to pass additional arguments like a flag (or a pointer) to the signal emitted from procedure workers (e.g. sendDataToSerial). What do you suggest?

Additionally, If we use the signal/slot solution then I think it is better to create a serial worker and move that to a new thread as well, in order to keep the GUI (i.e. MainWindow) responsive.

Also thanks for mentioning the leakage of serial_ .

P.S.
You mentioned that passing a class pointer to a class that contains that class is a wrong design. So I was wondering why is it done in Qt, i.e. when the QObject class (and its inherited classes from it) get a parent pointer argument? (actually the first argument of constructor of ProcAWorker is exactly this). How is it different?

jsulm

@mkaze said in Worker threads with shared resources in Qt application:

You mentioned that passing a class pointer to a class that contains that class is a wrong design

Because a worker thread should not call anything from "this" directly what ever this is. Use signals/slots. The purpose of passing this in Qt is completely other topic (parent/child).

mrdebug

Hi. I suggest you this type of architecture:

Create the QSerialPort object inside the run() methot on the only thread that is managing the serial port. The QSerialPort object must be private.
To send commands at the devices please implement a ProcessCommand(int CommandType, int Arg0 ...) function.
To have back data please use signal and slots.

So for example, from the main gui
object->ProcessCommand(FIRMWARE_GET);
The ProcessCommand function, using mutex, will put in the queue the commands to send by the thread

VRonin

@mkaze said in Worker threads with shared resources in Qt application:

Consider the code for ProcAWorker

Using signal/slot it would become:

void ProcAWorker::process()
{
    sendDataToSerial("Command a1");
    QThread::sleep(1);
    // perform some computations
    sendDataToSerial("Command a2");
    emit success();
}

If anything it's easier.
Errors of serial write, again, are not a responsibility of the worker to handle. They should be handle by the object owning serial_

mkaze

@VRonin said in Worker threads with shared resources in Qt application:

@mkaze said in Worker threads with shared resources in Qt application:

Consider the code for ProcAWorker

Using signal/slot it would become:

Thanks for your reply. That's right, but your code considers only write operations. A procedure may want to read from a serial port so it must wait (or be notified) when the response is ready. Even in case of only write operations, a procedure may want to perform some alternative operations or rollbacks when write errors happens. I meant something like the following when I mentioned implementing ProcAWorker as a state machine:

ProcAWorker::ProcAWorker() :
    current_state_(STATE_INITIAL),
    next_state_(STATE1)
{
    // assign a unique id to this procedure to
    // distinguish it from other procedures when
    // communicating with serial worker
    PID_ = PROCA_ID;
    connect(this, &ProcAWorker::forward, this, &ProcAWorker::process);
}

void ProcAWorker::reset_states()
{
    current_state_ = STATE_INITIAL;
    next_state_ = STATE1;
}

void ProcAWorker::process()
{
    current_state_ = next_state_;

    switch (current_state_){

    case STATE1:
    {
        emit serial_write_data(PID_, "command a1");
        QThread::sleep(1);
        next_state_ = STATE2;
        break;
    }

    case STATE2:
        emit serial_write_data(PID_, "command a2");
        // perform some computations
        next_state_ = STATE_FINAL;
        break;

    case STATE_FINAL:
        emit success();
        reset_states();
        break;

    default:
        reset_states();
        break;
    }

    return;
}

void ProcAWorker::serial_write_response(const Stepper::proc_id pid, bool response)
{
    // if the response does not belong
    // to this procedure, ignore it
    if (pid != PID_)
        return;

    if (response) {
        emit forward();
    } else {
        emit error(ERROR_PORT_WRITE);
        // update some internal variables

        reset_states();
    }
}

void ProcAWorker::serial_read_response(const Stepper::proc_id pid, const QByteArray & response)
{
    // if the response does not belong
    // to this procedure, ignore it
    if (pid != PID_)
        return;
    
    // store response in a member variable for later use
    // or perform some calculations and change some internal variables based on it
    
    emit forward();
}

}

And on the other side, in serial worker we have:

void SerialWorker::write_data(const proc_id pid, const QByteArray & data)
{
    serial_->write(data);
    bool write_ok = serial_->waitForBytesWritten(SERIAL_WRITE_TIMEOUT);
    emit write_finished(pid, write_ok);
}


void SerialWorker::read_data(const proc_id pid)
{
    QByteArray data = serial_->readAll();
    emit read_finished(pid, data);
}

// This slot is invoked when we want to read immediately after
// writing a command on serial port. Essentially, we are
// expecting a reponse to the sent command.
void SerialWorker::write_read_data(const proc_id pid, const QByteArray & data)
{
    serial_->write(data);
    bool write_ok = serial_->waitForBytesWritten(SERIAL_WRITE_TIMEOUT);
    if (write_ok) {
        serial_.waitForReadyRead(SERIAL_READ_TIMEOUT);
        QByteArray response = serial_->readAll();
        emit write_read_finished(pid, write_ok, reponse);
    } else {
        emit write_read_finished(pid, write_ok, QByteArray());
    }

}

which the corresponding signals/slots have been connected in main thread (i.e. GUI thread) when creating ProcAWorker, ProcBWorker, ... and SerialWorker.

What do you think about this design?

VRonin

I agree you need to use a state machine for that case. http://doc.qt.io/qt-5/qstatemachine.html probably helps putting down the skeleton already