Communication between multiple threads that run at different frequencies

[EvheMary]

Hi everyone,
I am a newbie in using pyqt (i'm using python) and want to ask for suggestions and advice. I'm making an application that stream data through LSL and then process it. The plan is that while the application is running, there are going to be multiples thread running in the background that handles LSL stream in, LSL stream out, processing, and plotting.

1. From what i have read online, there are qthreadpool that can be used to handle multiple worker, but i'm not sure about using qrunnable for the threads since i want the threads to be flexible since the streams need to be changeable, and or termindated. But i'm also not sure on how the application performance when i manually created or terminate the qthreads.
2. I'm also confused on how to handle the sharing of data from these multiple threads. Right now i plan to use object (not qobject) to shares these data because i am not sure on how to handle using signal and slots for this application.

Thank you in advance and any advice is appreciated.
[in case this is the wrong thread to post, please forgive me]

[Axel Spoerl]

The timer is living in one thread and its stop() slot is called directly from another thread.
Don't want the neighbor entering my flat without ringing the door bell, just to turn the music louder (or lower) - right?

I am not a python guru, but something like
QMetaObject.invokeMethod(self.timer, 'stop', Qt.QueuedConnection)
should do what self.timer.stop()doesn't when the timer lives elswhere.
QueuedConnectiondoesn't wait for the timer to be stopped. You can use other connection types if required.

[jsulm]

@EvheMary

1. If you have a fix number of threads do not use QThreadPool but create your QThreads manually
2. Use signals/slots to exchange data between threads

[EvheMary]

Thanks for the reply
I have fixed number of threads so i will follow your advice.
For the second one, since the data for the plot and process is the same, but has different window length and taken at different rate how should i do the thread run?

1. The 'stream in' thread has a flexible frequency, depending on the stream source
2. The plot thread is about ~10 Hz
3. The process thread is about 2 Hz

I plan to emit the data from 'stream in' thread, but how do i receive data from this thread at different rate for each thread (since as far as i know, the one controlling the frequency is the signal thread)?
Do i make a counter / qtimer?

[jsulm]

@EvheMary If you want to emit signal at some specific frequency then use a QTimer and on each timeout emit the signal if there is new data.

[Axel Spoerl]

If "share data" points at objects shared between threads, atomic versions of simple types are a safe way with limitations.
If you want to pass arguments from one thread to another, you can also a suitable overload of QMetaObject::invokeMethod(). Besides arguments, you can pass a connection type, depending on which you may (for blocking types) or may not (for non blocking types) retrieve return values.
The threads' frequency doesn't and shouldn't play a role in this concept. Waiting a specific amount of time in one thread and expecting that the other one has provided a valid result in e.g. a shared object makes an application vulnerable to race conditions and flakiness. The safe way is to emit a signal (with arguments) in the worker thread and connect to it.

[Kent-Dorfman]

My architecture would mirror what @jsulm wrote, except that I'd spawn short lived worker threads in the Qtimer callback for each function.

[EvheMary]

Thanks for the replies.
Just want to make clear, i want to create multiple threads, each with different purpose with flexible lifetime.

1. The 'stream in' thread will create 1 or 2 inlet objects at init, then run pull data (this function is the object's)
2. Thread for the plot (after reviewing my code, this might not be needed, since i can just emit signal to the plotter function, the problem is that the pyqtgraph is in different qdialog, should i connect the signal when the qdialog is called or when i start the stream thread and is there any consequences when i emit signal but the slot is not yet created?)
3. Process thread (same case with plot thread, but after reading kent's comment, i think its better to create short lived worker for every function call, or simpler: just qtimer and a function)
4. Other thread

While i dont think the frequency doesn't affect the thread functions, it will affect the data that will be emitted (or the data in the object which i planned to share before)

After the first comment, i use the stream in thread to create multiple qtimer, each with their own purpose, but the problem i mentioned above appear (2)
Any suggestions is appreciated

[Axel Spoerl]

If problem (2) is what you write at point 2:

2. Thread for the plot (after reviewing my code, this might not be needed

=> We don't have you code, so we can't say much.

since i can just emit signal to the plotter function, the problem is that the pyqtgraph is in different qdialog,

=> That's not a problem if they run in the same application

should i connect the signal when the qdialog is called or when i start the stream thread

Depends on your code: Signals should be connected before they are fired for the first time.

is there any consequences when i emit signal but the slot is not yet created?)

The information that the signal was fired will be lost. No buffering, no crashes etc - if that's what you mean.

[EvheMary]

Thanks for all the replies, i created a dummy demo of what my application is (I combine some codes from stacks). Unfortunately, there are some errors such as:

'QObject::~QObject: Timers cannot be stopped from another thread

This happens when i tried to start a new worker after stopping the previous one, o when i closed the app
Here is the code:

import sys
from PySide2.QtCore import *
from PySide2.QtWidgets import *

class Inlet_Worker(QObject):

    data = Signal(int)

    def __init__(self):
        super().__init__()
        self._stopped = False
        self._registered = False
        self.timer = QTimer(self)
        self.timer.timeout.connect(self.routine)
        self.c = 0

    def starter(self):
        self.timer.start(1000)

    '''
    def run(self):
        count = 0
        self._stopped = False
        while not self._stopped:
            #if client.read_coils(address = 0x0802).bits[0]:
            count += 1
            if count % 20 == 0 and not self._registered:
                self.updateBarcodeRegistration(True)
                self.timer.start(2000)
            QCoreApplication.processEvents()
            QThread.msleep(100)
        self.updateBarcodeRegistration(False)
        self.timer.stop()
        print('Stopped')
    '''

    def routine(self):
        print(self.c)
        self.data.emit(self.c)
        self.c += 1

    def stop(self):
        self.timer.stop()
        print('stopped')


class Window(QWidget):
    def __init__(self):
        super().__init__()
        self.dlg = Dialog()
        self.thread = QThread()
        self.button = QPushButton('Start')
        self.button2 = QPushButton('Stop')
        self.button3 = QPushButton('Window')
        layout = QHBoxLayout(self)
        layout.addWidget(self.button)
        layout.addWidget(self.button2)
        layout.addWidget(self.button3)
        self.button.clicked.connect(self.startThr)
        self.button2.clicked.connect(self.stopThr)
        self.button3.clicked.connect(self.showDlg)

    def showDlg(self):
        if not self.dlg.isVisible():
            self.dlg.show()
    
    def startThr(self):
        self.worker = Inlet_Worker()
        self.worker.moveToThread(self.thread)
        self.worker.data.connect(self.dlg.update)
        self.thread.started.connect(self.worker.starter)
        self.thread.start()

    def stopThr(self):
        self.worker.stop()
        self.thread.terminate()
        self.thread.wait()

    def closeEvent(self, event):
        self.worker.stop()
        self.thread.quit()
        self.thread.wait()

class Dialog(QWidget):
    def __init__(self):
        super().__init__()
        self.text1 = QLabel("Label 1 : ")
        self.text2 = QLabel("Text2")
        layout = QHBoxLayout(self)
        layout.addWidget(self.text1)
        layout.addWidget(self.text2)
        self.setGeometry(400, 100, 100, 50)

    def update(self, sig):
        self.text2.setText(str(sig))

    def closeEvent(self, event):
        pass

if __name__ == '__main__':

    app = QApplication(sys.argv)
    window = Window()
    window.setGeometry(750, 100, 200, 50)
    window.show()
    sys.exit(app.exec_())

Can you explain why is this happening? Thanks in advance.

[Axel Spoerl]

The timer is living in one thread and its stop() slot is called directly from another thread.
Don't want the neighbor entering my flat without ringing the door bell, just to turn the music louder (or lower) - right?

I am not a python guru, but something like
QMetaObject.invokeMethod(self.timer, 'stop', Qt.QueuedConnection)
should do what self.timer.stop()doesn't when the timer lives elswhere.
QueuedConnectiondoesn't wait for the timer to be stopped. You can use other connection types if required.

[EvheMary]

Thanks a lot for the solution. It fixed the error. It seems I misunderstood what a Qtimer is. QTimer might also be a threading interface, hence why I can't just call its start and stop methods. I'm just still confused about why I need this method to stop the timer, but I can do it directly for the start method?
And is QMetaObject works similarly with what a signal is?

[jsulm]

@EvheMary said in Communication between multiple threads that run at different frequencies:

QTimer might also be a threading interface

What does this mean? QTimer is just a timer and has nothing to do with threads.
"why I can't just call its start and stop" - because QTimer is not thread safe (only reentrant as documentation states), so you should not call its methods from other threads than the thread where the QTimer instance is living.

[JonB]

@EvheMary
@Axel-Spoerl may well know more than I, but I have never had to use QMetaObject.invokeMethod(). Fine if that's good/safe/the right to do this/works for you. But my first thought would have been to send a signal from wherever to the thread which the thread has a slot on and does its own self.timer.stop() when received, if that is what you are trying to achieve. Am I wrong?

[Axel Spoerl]

my first thought would have been to send a signal

As usual, @JonB , your (first) thoughts are excellent!
QMetaObject.invokeMethod() spares the definition of a signal, especially when you call a void slot without arguments, like QTimer::stop(). It can also be quite handy if you call it with the BlockingConnection argument, in which case you can block'n'wait for a return value :-)

[JonB]

@Axel-Spoerl
Totally respect your answer. I have seen mentions of QMetaObject.invokeMethod() on the web, but I know it's used in "advanced" situations (including a lot from Python/PyQt/PySide, probably understandably) so haven't touched it! I can tell that from the 'stop' argument being a literal string it's going to look up the method to call by name ("reflection"), like the old SIGNAL/SLOT() macros approach did; so being more a C++ purist I would tend to avoid that, preferring compile-time safety/argument checking etc.

[Axel Spoerl]

@JonB
Totally fair point! A kitten dies, each time you resolve a symbol with a string search at run time....in Python.
There are more invokeMethod() overloads for the C++ lovers from 6.4 onward.

[JonB]

@Axel-Spoerl said in Communication between multiple threads that run at different frequencies:

A kitten dies, each time you resolve a symbol with a string search at run time....in Python.

:)

[EvheMary]

Thanks for all these comments. I might not fully understand since I'm still a beginner in qt, so sorry if I mentioned something wrong. I thought that since I use a signal to communicate with the Qtimer thread it should be safe. And I should be calling the stop method from the thread that it was created. Is my implementation of the signals and slot not correct?
my understanding of using the signal and slot is :
UI Thread --> send signal --> stop method (slot) at another thread --> QTimer stop

[JonB]

@EvheMary
You call self.worker.stop() (which calls self.timer.stop()) directly from the UI thread.

[EvheMary]

@JonB But if i call self.worker.stop() that calls self.timer.stop() from the UI Thread, the error QObject::~QObject: Timers cannot be stopped from another thread appears and that's what I want to know what causes it and how to remove it. If i change self.timer.stop() with QMetaObject.invokeMethod(self.timer, 'stop', Qt.AutoConnection) the error don't appear.