[FIXED] PyQt6 multithreading mixing concurrent and sequantial tasks

Issue

I have a PyQt6 application in which I have implemented multithreading. I'm looking to achieve the following. Suppose I have a step1, which needs to be executed first. When finished, step2 can start. However, step2 relies on two other tasks requirement_a and requirement_b. In contrast to step1 and step2, requirement_a and requirement_b are supposed to run concurrently.

I've tried implementing this using two QThreadPools, but I'm not yet getting the expected results. Consider the code below:

import sys
from datetime import datetime
import time
from PyQt6.QtCore import QThreadPool, QRunnable, QObject, pyqtSignal, pyqtSlot
from PyQt6.QtWidgets import QMainWindow, QPushButton, QApplication


class WorkerSignals(QObject):
    """Defines the signals available from a running worker thread.

    Supported signals are:
    - started: no data
    - finished: no data
    """
    started = pyqtSignal()
    finished = pyqtSignal()


class Worker(QRunnable):
    """Worker thread"""
    def __init__(self, fun, *args, **kwargs):
        """Initialize method of the Worker thread"""
        super(Worker, self).__init__()

        # Store constructor arguments (re-used for processing)
        self.fun = fun
        self.args = args
        self.kwargs = kwargs
        self.signals = WorkerSignals()

    @pyqtSlot()
    def run(self):
        """Execute the function with passed args and kwargs"""
        self.signals.started.emit()
        self.fun(*self.args, **self.kwargs)
        self.signals.finished.emit()


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()

        self.button = QPushButton('Start', self)
        self.button.setGeometry(10, 10, 100, 30)
        self.button.clicked.connect(self.start)
        self.sequential_thread_pool = QThreadPool()
        self.sequential_thread_pool.setMaxThreadCount(1)
        self.concurrent_thread_pool = QThreadPool()

    def start(self):
        self.start_worker(self.step1, 'sequential')
        self.start_worker(self.step2, 'sequential')

    def start_worker(self, fun, thread_pool):
        worker = Worker(fun)
        worker.signals.started.connect(lambda: print(f'[{datetime.now()}] Started {fun.__name__}'))
        worker.signals.finished.connect(lambda: print(f'[{datetime.now()}] Finished {fun.__name__}'))
        if thread_pool == 'sequential':
            self.sequential_thread_pool.start(worker)
        elif thread_pool == 'concurrent':
            self.concurrent_thread_pool.start(worker)

    def step1(self):
        # this is a step that needs to execute first
        time.sleep(1) # do stuff
        print(f'[{datetime.now()}]   Executing step1')
        time.sleep(1) # do stuff

    def requirement_a(self):
        # this is a step that needs to execute after step1 finished, but can happen concurrently with requirement_b
        time.sleep(1) # do stuff
        print(f'[{datetime.now()}]   Executing requirement_a')
        time.sleep(1) # do stuff

    def requirement_b(self):
        # this is a step that needs to execute after step1 finished, but can happen concurrently with requirement_a
        time.sleep(1) # do stuff
        print(f'[{datetime.now()}]   Executing requirement_b')
        time.sleep(1) # do stuff

    def step2(self):
        # this is a step that needs to execute after step1, requirement_a and requirement_b finished
        self.start_worker(self.requirement_a, 'concurrent')
        self.start_worker(self.requirement_b, 'concurrent')

        self.concurrent_thread_pool.waitForDone() # wait for requirement_a and requirement_b to finish

        time.sleep(1) # do stuff
        print(f'[{datetime.now()}]   Executing step2')
        time.sleep(1) # do stuff


if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app.exec())

Note that currently step2 is given to the sequential threadpool, from where it sends requirement_a and requirement_b to the concurrent threadpool. This is some kind of thread nesting I suppose? Running this code and clicking the button prints the following:

[2023-06-16 17:15:54.792738] Started step1
[2023-06-16 17:15:55.794494]   Executing step1
[2023-06-16 17:15:56.804084] Finished step1
[2023-06-16 17:15:56.804084] Started step2
[2023-06-16 17:15:57.817431]   Executing requirement_b[2023-06-16 17:15:57.817431]   Executing requirement_a

[2023-06-16 17:15:59.848227]   Executing step2
[2023-06-16 17:16:00.859409] Finished step2

The order of the steps are being executed in the way that I desire, however the "Started requirement_a", "Finished requirement_a", "Started requirement_b" and "Finished requirement_b" prints are missing. This indicates that these print statements connected to worker.signals.started and worker.signals.finished for the worker that is given to self.concurrent_thread_pool, do not seem to be executed.

I've spent quite some time on trying to fix this. I have also done quite some digging and reading in trying to find a best practice for mixing sequential and concurrent threading. Both without any luck... What am I doing wrong and how should this be done? Hopefully the community can clarify and help me on this!

Solution

It's because there's no event loop which can process the started and finished signals emitted when the workers run in the step2(). You need to understand the thread affinity of a QObject. You didn't hint Qt that the signals should be queued into the main thread.

There are many possible solutions to your problem. I'll mention just one which uses slots of the MainWindow as hints.

...
class WorkerSignals(QObject):
    started = pyqtSignal(object)
    finished = pyqtSignal(object)

class Worker(QRunnable):
    ...
    def run(self):
        self.signals.started.emit(self)
        self.fun(*self.args, **self.kwargs)
        self.signals.finished.emit(self)

class MainWindow(QMainWindow):
    ...
    def on_worker_started(self, worker):
        print(f'[{datetime.now()}] Started {worker.fun.__name__}')

    def on_worker_finished(self, worker):
        print(f'[{datetime.now()}] Finished {worker.fun.__name__}')

    def start_worker(self, fun, thread_pool):
        worker = Worker(fun)
        worker.signals.started.connect(self.on_worker_started)
        worker.signals.finished.connect(self.on_worker_finished)
        ...
...

Answered By - relent95