Skip to content

Instantly share code, notes, and snippets.

@mchubby
Forked from jazzycamel/thread2.py
Last active February 20, 2017 01:02
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save mchubby/36b254e1110f15db566da9229e5ce34f to your computer and use it in GitHub Desktop.
Save mchubby/36b254e1110f15db566da9229e5ce34f to your computer and use it in GitHub Desktop.
Simple example of threaded loop with abort handling (QMutex.tryLock) -- using standalone Qt.py (untested on PyQt4)
import sys
from random import uniform
import time
import threading
from itertools import count, islice
from vendor.Qt.Qt import QtCore, QtGui, QtWidgets
from vendor.Qt.Qt.QtWidgets import (
QApplication,
QLabel,
QPushButton,
QVBoxLayout,
QWidget)
def logthread(caller):
print('%-25s: %s, %s,' % (caller, threading.current_thread().name,
threading.current_thread().ident))
class ThreadedQObject(QtCore.QObject):
# this signal can be connected (handled) in a UI slot,
# using the new syntax at object creation time (see code in consumer class)
state_changed = QtCore.Signal(int)
work_finished = QtCore.Signal(int)
def __init__(self, parent=None, **kwargs):
super().__init__(parent, **kwargs)
logthread('ThreadedQObject.__init__')
self._count = 0
self._halt_requested = 0
self._cleanup_mutex = QtCore.QMutex()
@QtCore.Slot()
def start(self):
# this slot is linked to the QThread instance's started signal
logthread('ThreadedQObject.start (W slot)')
@QtCore.Slot()
def compute_mainloop(self):
logthread('ThreadedQObject.compute_mainloop (W slot)')
self._count = 0
self._halt_requested = 0
self._cleanup_mutex.lock()
is_completed = False
while self._halt_requested == 0 and not is_completed:
time.sleep(uniform(1., 8.))
self._count += 1
self.state_changed.emit(self._count)
QApplication.instance().processEvents()
if self._count > 10:
is_completed = True
if is_completed:
self.work_finished.emit(self._count)
else:
if self._halt_requested > 1:
print("ThreadedQObject: Do fast cleanup")
time.sleep(0.7)
else:
print("ThreadedQObject: Do normal cleanup")
time.sleep(1.8)
print("W: Lock released")
self._cleanup_mutex.unlock()
@QtCore.Slot(bool)
def request_mainloop_halt(self, is_abort=False):
logthread('ThreadedQObject.request_mainloop_halt (W slot)')
self._halt_requested = 2 if is_abort else 1
def acquire_cleanup_mutex(self, timeout):
if self._cleanup_mutex.tryLock(timeout):
print("GUI: Lock acquired")
self._cleanup_mutex.unlock()
return True
return False
class GUI(QWidget):
startLoop = QtCore.Signal()
stopLoop = QtCore.Signal(bool)
def __init__(self, parent=None, **kwargs):
super().__init__(parent, **kwargs)
logthread('GUI.__init__')
self._thread = QtCore.QThread()
# object is referenced to avoid it being GC'd
self._threaded = ThreadedQObject(
state_changed=self.on_state_changed,
work_finished=self.on_work_finished)
self._thread.started.connect(self._threaded.start)
self.startLoop.connect(self._threaded.compute_mainloop)
self.stopLoop.connect(self._threaded.request_mainloop_halt)
self._threaded.moveToThread(self._thread)
QApplication.instance().aboutToQuit.connect(self.on_about_to_quit)
self._thread.start()
l = QVBoxLayout(self)
l.addWidget(QPushButton("Start", self, clicked=self.startLoop))
l.addWidget(QPushButton("Stop", self, clicked=self.stopLoop))
self._countLbl=QLabel("Count:", self)
l.addWidget(self._countLbl)
@QtCore.Slot()
def on_about_to_quit(self):
logthread('GUI.on_about_to_quit')
self.stopLoop.emit(True)
time.sleep(.5)
if self._threaded.acquire_cleanup_mutex(5000):
print("on_about_to_quit: thread returned control before deadline (good)")
self._thread.quit()
else:
print("on_about_to_quit: terminate")
self._thread.terminate()
@QtCore.Slot(int)
def on_state_changed(self, int_value):
logthread('GUI.on_state_changed (UI slot)')
self._countLbl.setText("Count: <b>{}</b>".format(int_value))
@QtCore.Slot(int)
def on_work_finished(self, int_value):
logthread('GUI.on_work_finished (UI slot)')
self._countLbl.setText("Completed, count = <b>{}</b>".format(int_value))
if __name__=="__main__":
from sys import exit, argv
app = QApplication(argv)
g = GUI()
g.show()
exit(app.exec_())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment