PyQt4: Interrupt QThread exec when GUI is closed

Question

I have a PyQt4 GUI that has three threads. One thread is a data source, it provides numpy arrays of data. The next thread is a calculation thread, it takes the numpy array (or multiple numpy arrays) via a Python Queue.Queue and calculates what will be displayed on the GUI. The calculator then signals the GUI thread (the main thread) via a custom signal and this tells the GUI to update the matplotlib figure that's displayed.

I'm using the "proper" method described here and here.

So here's the general layout. I tried to shorten my typing time and used comments instead of the actual code in some parts:

class Source(QtCore.QObject):
    signal_finished = pyQtSignal(...)
    def __init__(self, window):
        self._exiting = False
        self._window = window

    def do_stuff(self):
        # Start complicated data generator
        for data in generator:
            if not self._exiting:
                # Get data from generator
                # Do stuff - add data to Queue
                # Loop ends when generator ends
            else:
                break
        # Close complicated data generator

    def prepare_exit(self):
        self._exiting = True

class Calculator(QtCore.QObject):
    signal_finished = pyQtSignal(...)
    def __init__(self, window):
        self._exiting = False
        self._window = window

    def do_stuff(self):
        while not self._exiting:
            # Get stuff from Queue (with timeout)
            # Calculate stuff
            # Emit signal to GUI
            self._window.signal_for_updating.emit(...)

    def prepare_exit(self):
        self._exiting = True

class GUI(QtCore.QMainWindow):
    signal_for_updating = pyQtSignal(...)
    signal_closing = pyQtSignal(...)
    def __init__(self):
        self.signal_for_updating.connect(self.update_handler, type=QtCore.Qt.BlockingQueuedConnection)
    # Other normal GUI stuff
    def update_handler(self, ...):
        # Update GUI
    def closeEvent(self, ce):
        self.fileQuit()
    def fileQuit(self): # Used by a menu I have File->Quit
        self.signal_closing.emit() # Is there a builtin signal for this

if __name__ == '__main__':
    app = QtCore.QApplication([])
    gui = GUI()
    gui.show()

    source_thread = QtCore.QThread() # This assumes that run() defaults to calling exec_()
    source = Source(window)
    source.moveToThread(source_thread)

    calc_thread = QtCore.QThread()
    calc = Calculator(window)
    calc.moveToThread(calc_thread)

    gui.signal_closing.connect(source.prepare_exit)
    gui.signal_closing.connect(calc.prepare_exit)
    source_thread.started.connect(source.do_stuff)
    calc_thread.started.connect(calc.do_stuff)
    source.signal_finished.connect(source_thread.quit)
    calc.signal_finished.connect(calc_thread.quit)

    source_thread.start()
    calc_thread.start()
    app.exec_()
    source_thread.wait() # Should I do this?
    calc_thread.wait() # Should I do this?

...So, my problems all occur when I try to close the GUI before the sources are complete, when I let the data generators finish it closes fine:

• While waiting for the threads, the program hangs. As far as I can tell this is because the closing signal's connected slots never get run by the other thread's event loops (they're stuck on the "infinitely" running do_stuff method).

• When the calc thread emits the updating gui signal (a BlockedQueuedConnection signal) right after the GUI closing, it seems to hang. I'm guessing this is because the GUI is already closed and isn't there to accept the emitted signal (judging by the print messages I put in my actual code).

I've been looking through tons of tutorials and documentation and I just feel like I'm doing something stupid. Is this possible, to have an event loop and an "infinite" running loop that end early...and safely (resources closed properly)?

I'm also curious about my BlockedQueuedConnection problem (if my description makes sense), however this problem is probably fixable with a simple redesign that I'm not seeing.

Thanks for any help, let me know what doesn't make sense. If it's needed I can also add more to the code instead of just doing comments (I was kind of hoping that I did something dumb and it wouldn't be needed).

Edit: I found some what of a work around, however, I think I'm just lucky that it works every time so far. If I make the prepare_exit and the thread.quit connections DirectConnections, it runs the function calls in the main thread and the program does not hang.

I also figured I should summarize some questions:

1. Can a QThread have an event loop (via exec_) and have a long running loop?
2. Does a BlockingQueuedConnection emitter hang if the receiver disconnects the slot (after the signal was emitted, but before it was acknowledged)?
3. Should I wait for the QThreads (via thread.wait()) after app.exec_(), is this needed?
4. Is there a Qt provided signal for when QMainWindow closes, or is there one from the QApplication?

Edit 2/Update on progress: I have created a runnable example of the problem by adapting this post to my needs.

from PyQt4 import QtCore
import time
import sys


class intObject(QtCore.QObject):
    finished = QtCore.pyqtSignal()
    interrupt_signal = QtCore.pyqtSignal()
    def __init__(self):
        QtCore.QObject.__init__(self)
        print "__init__ of interrupt Thread: %d" % QtCore.QThread.currentThreadId()
        QtCore.QTimer.singleShot(4000, self.send_interrupt)
    def send_interrupt(self):
        print "send_interrupt Thread: %d" % QtCore.QThread.currentThreadId()
        self.interrupt_signal.emit()
        self.finished.emit()

class SomeObject(QtCore.QObject):
    finished = QtCore.pyqtSignal()
    def __init__(self):
        QtCore.QObject.__init__(self)
        print "__init__ of obj Thread: %d" % QtCore.QThread.currentThreadId()
        self._exiting = False

    def interrupt(self):
        print "Running interrupt"
        print "interrupt Thread: %d" % QtCore.QThread.currentThreadId()
        self._exiting = True

    def longRunning(self):
        print "longRunning Thread: %d" % QtCore.QThread.currentThreadId()
        print "Running longRunning"
        count = 0
        while count < 5 and not self._exiting:
            time.sleep(2)
            print "Increasing"
            count += 1

        if self._exiting:
            print "The interrupt ran before longRunning was done"
        self.finished.emit()

class MyThread(QtCore.QThread):
    def run(self):
        self.exec_()

def usingMoveToThread():
    app = QtCore.QCoreApplication([])
    print "Main Thread: %d" % QtCore.QThread.currentThreadId()

    # Simulates user closing the QMainWindow
    intobjThread = MyThread()
    intobj = intObject()
    intobj.moveToThread(intobjThread)

    # Simulates a data source thread
    objThread = MyThread()
    obj = SomeObject()
    obj.moveToThread(objThread)

    obj.finished.connect(objThread.quit)
    intobj.finished.connect(intobjThread.quit)
    objThread.started.connect(obj.longRunning)
    objThread.finished.connect(app.exit)
    #intobj.interrupt_signal.connect(obj.interrupt, type=QtCore.Qt.DirectConnection)
    intobj.interrupt_signal.connect(obj.interrupt, type=QtCore.Qt.QueuedConnection)

    objThread.start()
    intobjThread.start()
    sys.exit(app.exec_())

if __name__ == "__main__":
    usingMoveToThread()

You can see by running this code and swapping between the two connection types on interrupt_signal that the direct connection works because its running in a separate thread, proper or bad practice? I feel like that is bad practice because I am quickly changing something that another thread is reading. The QueuedConnection does not work because the event loop must wait until longRunning is finished before the event loop gets back around to the interrupt signal, which is not what I want.

Edit 3: I remembered reading that QtCore.QCoreApplication.processEvents can be used in cases with long running calculations, but everything I read said don't use it unless you know what you are doing. Well here is what I think it's doing (in a sense) and using it seems to work: When you call processEvents it causes the caller's event loop to halt its current operation and continue on processing the pending events in the event loop, eventually continuing the long calculation event. Other recommendations like in this email suggest timers or putting the work in other threads, I think this just makes my job even more complicated, especially since I've proven(I think) timers don't work in my case. If processEvents seems to fix all my problems I will answer my own question later.

Answer 1

After looking through the mailing list archives, google searching, stack overflow searching, and thinking about what my question really was and what the purpose of the question was I came up with this answer:

The short answer being use processEvents(). The long answer is that all my searching results in people saying "be very careful using processEvents()" and "avoid it at all costs". I think it should be avoided if you are using it because you are not seeing results in your GUI main thread fast enough. Instead of using processEvents in this case, the work being done in the main thread that is not UI purposed should be moved to another thread (as my design has done).

The reason my specific problem needs processEvents() is that I want my QThreads to have two way communication with the GUI thread, which means that my QThreads have to have an event loop (exec_()) to accept signals from the GUI. This two way communication is what I meant earlier by "the purpose of the question". Since my QThreads are meant to run "concurrently" with the main GUI thread AND because they need to update the GUI and be "updated" by the GUI (the exit/closing signal in my first example), they need processEvents(). I think this is what processEvents() is for.

My understanding of processEvents(), as decribed above, is that when called in a QThread it will block/pause the current event (my longRunning method) while it continues on through the events in the event loop (only for the QThread processEvents() was called in). After going through the pending events, the event loop wraps back around and continues running the event that it paused (my longRunning method).

I know I didn't answer all my questions, but the main one is answered.

PLEASE CORRECT ME IF I AM WRONG IN ANY WAY

Edit: Please read Ed's answer and the comments.

Answer 2

I honestly did not read all of the code. I would recommend against having loops in your code but instead run each logical chunk at a time. Signals/Slots can work as transparent queues for these things too.

Some producer/consumer example code I've written https://github.com/epage/PythonUtils/blob/master/qt_producer_consumer.py Some different threading code with more advanced utils I've written https://github.com/epage/PythonUtils/blob/master/qt_error_display.py

Yes I used loops, mostly for example purposes but sometimes you can't avoid them (like reading from an pipe). You can either use QTimer with a timeout of 0 or have a flag to mark that things should quit and protect it with a mutex.

RE EDIT 1: 1. Don't mix exec_ with long running loops 3. PySide requires that you wait after quitting a thread. 4. I don't remember there being one, you can set it to Destroy On Close and then monitor for close or you can inherit from QMainWindow, override closeEvent and fire a signal (like I do in the qt_error_display.py example)

RE EDIT 2: I'd recommend using the default connection types.

RE EDIT 3: Don't use processEvents.

Answer 3

You could split your workload into chunks and process them one by one in separate slot calls as suggested here: https://wiki.qt.io/Threads_Events_QObjects

import time
import sys

from PyQt5 import QtCore
from PyQt5.QtCore import pyqtSlot, pyqtSignal, QMetaObject, Qt, QThread


class intObject(QtCore.QObject):
    finished = pyqtSignal()
    interrupt_signal = pyqtSignal()

    def __init__(self):
        QtCore.QObject.__init__(self)
        print("__init__ of interrupt Thread: %d" % QThread.currentThreadId())
        QtCore.QTimer.singleShot(3000, self.send_interrupt)

    @pyqtSlot()
    def send_interrupt(self):
        print("send_interrupt Thread: %d" % QThread.currentThreadId())
        self.interrupt_signal.emit()
        self.finished.emit()

class SomeObject(QtCore.QObject):
    finished = pyqtSignal()
    def __init__(self):
        QtCore.QObject.__init__(self)
        print("__init__ of obj Thread: %d" % QThread.currentThreadId())
        self._exiting = False
        self.count = 0

    @pyqtSlot()
    def interrupt(self):
        print("Running interrupt")
        print("interrupt Thread: %d" % QThread.currentThreadId())
        self._exiting = True

    @pyqtSlot()
    def longRunning(self):
        if self.count == 0:
            print("longrunning Thread: %d" % QThread.currentThreadId())
            print("Running longrunning")
        if self._exiting:
            print('premature exit')
            self.finished.emit()
        elif self.count < 5:
            print(self.count, 'sleeping')
            time.sleep(2)
            print(self.count, 'awoken')
            self.count += 1
            QMetaObject.invokeMethod(self, 'longRunning',  Qt.QueuedConnection)
        else:
            print('normal exit')
            self.finished.emit()


class MyThread(QThread):
    def run(self):
        self.exec_()

def usingMoveToThread():
    app = QtCore.QCoreApplication([])
    print("Main Thread: %d" % QThread.currentThreadId())

    # Simulates user closing the QMainWindow
    intobjThread = MyThread()
    intobj = intObject()
    intobj.moveToThread(intobjThread)

    # Simulates a data source thread
    objThread = MyThread()
    obj = SomeObject()
    obj.moveToThread(objThread)

    obj.finished.connect(objThread.quit)
    intobj.finished.connect(intobjThread.quit)
    objThread.started.connect(obj.longRunning)
    objThread.finished.connect(app.exit)
    #intobj.interrupt_signal.connect(obj.interrupt, type=Qt.DirectConnection)
    intobj.interrupt_signal.connect(obj.interrupt, type=Qt.QueuedConnection)

    objThread.start()
    intobjThread.start()
    sys.exit(app.exec_())

if __name__ == "__main__":
    usingMoveToThread()

Result:

Main Thread: 19940
__init__ of interrupt Thread: 19940
__init__ of obj Thread: 19940
longrunning Thread: 18040
Running longrunning
0 sleeping
0 awoken
1 sleeping
send_interrupt Thread: 7876
1 awoken
Running interrupt
interrupt Thread: 18040
premature exit

Answer 4

Instead of QMetaObject.invokeMethod it is also possible to use QTimer with 0 timeout as suggested here: https://doc.qt.io/qt-5/qtimer.html