How to make Qt signals work using Python's multiprocessing interface

[tpane3]

Unfortunately I'm not able to post my code so I will do my best explain the problem:

I'm using PyQt to develop a front-end for an application that does some heavy duty modeling/simulation. I have a QMainWindow instantiated as the primary interface for the user. When the user is ready to execute the simulation, a QThread is created to allow the user to interact with the GUI and progress to be reported while the simulation is running. Here's where I run into the issue: Because the simulation is process-intensive, I leverage the Python multiprocessing library to spawn multiple processes in order to speed up runtime. In order to report progress back to the GUI, I need to send a QtSignal Object into my main simulation thread. However, the Python multiprocessing library requires objects to be serialized and unfortunately, QtSignal objects cannot be pickled as far as I know. I am at a loss as to how to maintain the multiprocessing functionality while being able to report progress back to the GUI. Are there any known solutions to this problem? Thanks.

[SGaist]

Hi and welcome to devnet,

How do you get progress information currently from your multiprocessing implementation ?

[tpane3]

Prior to using PyQt I would create a multiprocessing Pool and pass a Queue and Lock object to the function that kicks off the Monte Carlo simulation (I'll call it main()). The number of processes created is configurable by the user, so the user can break up a 1000 trial simulation into 10 separate processes each running 100 trials. The results are combined after all the processes are finished. I was putting an integer object into Queue (initially set to 0) and updating it after each trial finished. I was printing out a progress bar to the console after each trial finished based on the current value in Queue divided by the total number of trials for the entire sim.

Below is a simple example I put together with PyQt that should illustrate what I'm trying to achieve in principal. With NUM_PROCESS == 1 everything works fine... however anything > 1 will result in "TypeError: can't pickle PyQt5.QtCore.pyqtBoundSignal objects". Is there another implementation that is feasible that would allow the multiprocessing functionality to remain in place while also allowing for progress reporting (using PyQt)?:

import sys
import multiprocessing as mp

from worker import Worker
from gui_example import *

NUM_PROCESSES = 2
ITERATIONS = 20000

def compute_primes(args):
    process_idx, iters, total_iters, queue, lock, progress_callback = args

    first = process_idx * iters
    last = first + iters

    all_primes = []
    for i in range(first, last):
        flag = True
        for j in range(1, i):
            if % j == 0 and j != 1:
                flag = False
                break
            if flag:
                all_primes.append(i)

            current_iter = i + 1 if not queue else queue.get() + 1
            progress = current_ter / total_iters * 100
            if progress % 10 == 0:
                progress_callback.emit(progress)

        return all_primes

def run_sim(iterations, progress_callback):
    sim_tracker = list()
    if NUM_PROCESSES == 1:
        queue, lock = None, None
    else:
        manager = mp.Manager()
        queue = manager.Queue()
        lock = manager.Lock()
        queue.put(0)

    iters_per_process = iterations // NUM_PROCESSES
    for process_idx in range(NUM_PROCESSES):
        sim_tracker.append((process_idx, iters_per_process, iterations, queue, lock, progress_callback))
    
    if NUM_PROCESSES == 1:
        output = compute_primes(sim_tracker[0])
    else:
        with mp.Pool(NUM_PROCESSES) as p:
            output = p.map(compute_primes, sim_tracker)

class ExampleDialog(QtWidgets.QDialog):

    def __init__(self, parent=None):
        super(ExampleDialog, self).__init__(parent)
        self._ui = Ui_Test()
        self._ui.setupUi(self)
        self._ui.progress_bar.setValue(0)
        self.worker = None

        self._ui.run_pb.clicked.connect(self.create_process)

    @QtCore.pyqtSlot()
    def create_process(self):
        self._ui.run_pb.setEnabled(False)
        self._ui.progress_bar.setValue(0)
        self._ui.status_lbl.setText('Executing...')
        self.worker = Worker(parent=self, func=run_sim, iterations=ITERATIONS)
        self.worker.signals.finished.connect(self.thread_complete)
        self.worker.signals.progress.connect(self.update_progress)
        self.worker.start()

def update_progress(self, n):
        self._ui.progress_bar.setValue(n)

def thread_complete(self):
       self._ui.status_lbl.setText('Thread Complete!')
       self._ui.run_pb.setEnabled(True)
       self.worker.quit()

if __name__ == '__main__':
    app = QtWidget.QApplication(sys.argv)
    myapp = ExampleDialog()
    myapp.show()
    sys.exit(app.exec_())

Module containing code for worker thread is below:

from PyQt5 import QtCore

class Worker(QtCore.QThread):

    def __init__(self, func, *args, **kwargs):
        super(Worker, self).__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.signals = WorkerSignals()

       self.kwargs['progress_callback'] = self.signals.progress

    @QtCore.pyqtSlot()
    def run(self):
        self.func(*self.args, **self.kwargs)
        self.signals.finished.emit()

class WorkerSignals(QtCore.QObject)
    finished = QtCore.pyqtSignal()
    progress = QtCore.pyqtSignal(int)