qthreads - qthread daemon python

Async como patrón en pyqt? O un patrón de llamada de fondo más limpio?

Question

Intento escribir un programa corto (un archivo pyqt) que sea receptivo (por lo que las dependencias fuera de python / lxml / qt, especialmente las que no puedo incluir en el archivo, tienen algunas desventajas para este caso de uso, pero aún podría estarlo dispuesto a probarlos). Estoy intentando realizar operaciones posiblemente largas (y cancelables) en un hilo de trabajo (en realidad, la operación en segundo plano tiene un bloqueo para evitar múltiples operaciones a la vez (ya que la biblioteca que utiliza solo puede usarse una llamada a la vez) y tiempos de espera así que engendrar múltiples hilos también estaría bien).

Hasta donde puedo descubrir la forma "básica" de hacer esto con qt es. (el código de la nota no se ha probado, por lo que puede ser incorrecto)

class MainWindow(QWidget):
#self.worker moved to background thread
def initUI(self):
...
self.cmd_button.clicked.connect(self.send)
...
@pyqtslot()
def send(self):
...
...#get cmd from gui
QtCore.QTimer.singleShot(0, lambda : self.worker(cmd))
@pyqtslot(str)
def end_send(self, result):
...
...# set some gui to display result
...
class WorkerObject(QObject):
def send_cmd(self, cmd):
... get result of cmd
QtCore.QTimer.singleShot(0, lambda: self.main_window.end_send())

(¿Estoy usando QTimer a la derecha (se ejecuta en un hilo diferente a la derecha)?)

Realmente preferiría tener algo más simple y más abstracto a lo largo de las líneas de asincronización de c #. (nota que no he usado asyncio así que podría estar equivocando algunas cosas)

class MainWindow(QWidget):
...
@asyncio.coroutine
def send(self):
...
...#get cmd from gui
result = yield from self.worker(cmd)
#set gui textbox to result
class WorkerObject(QObject):
@asyncio.coroutine
def send_cmd(self, cmd):
... get result of cmd
yield from loop.run_in_executor(None, self.model.send_command, cmd)

Escuché que Python 3 tenía características similares y que había un puerto posterior, pero ¿funciona correctamente con qt?

Si alguien sabe de otro patrón más saludable. eso también sería útil / una respuesta aceptable.

Answer 1

Para ejecutar ONE, el procesamiento prolongado crea objetos complejos basados ​​en eventos dependientes de QThreads. Hay dos enfoques para llamar a métodos de procesamiento únicos una vez:

El primer enfoque es QtConcurrent() . Si solo ejecuta una función larga, este sería un buen enfoque. Sin embargo, no estoy seguro si esto está disponible en pyqt.

El segundo enfoque sería subclasificar QThread e implementar su código de procesamiento en el método run() de la subclase. luego simplemente llame a QThreadSubclass.start() . Esto debería estar disponible en PyQt y probablemente sea el camino a seguir. La complejidad se reduce a una subclase simple. La comunicación al hilo es fácil de implementar ya que se comunicaría con cualquier otra clase.

Cuando se utiliza un Objeto asignado a QThread, que probablemente no sea la mejor manera, en lugar de QTimer debe emitir una señal usando Qt.QueuedConnection. Al usar QueuedConnection se asegurará de que la ranura se esté ejecutando, el objeto reside en.

Answer 2

La respuesta breve a su pregunta ("¿hay alguna forma de utilizar un patrón tipo asyncio en PyQt?") Es sí, pero es bastante complicado y podría decirse que no vale la pena para un programa pequeño. Aquí hay un código de prototipo que le permite usar un patrón asíncrono como el que describió:

import types
import weakref
from functools import partial
from PyQt4 import QtGui
from PyQt4 import QtCore
from PyQt4.QtCore import QThread, QTimer
## The following code is borrowed from here:
# http://.com/questions/24689800/async-like-pattern-in-pyqt-or-cleaner-background-call-pattern
# It provides a child->parent thread-communication mechanism.
class ref(object):
"""
A weak method implementation
"""
def __init__(self, method):
try:
if method.im_self is not None:
# bound method
self._obj = weakref.ref(method.im_self)
else:
# unbound method
self._obj = None
self._func = method.im_func
self._class = method.im_class
except AttributeError:
# not a method
self._obj = None
self._func = method
self._class = None
def __call__(self):
"""
Return a new bound-method like the original, or the
original function if refers just to a function or unbound
method.
Returns None if the original object doesn''t exist
"""
if self.is_dead():
return None
if self._obj is not None:
# we have an instance: return a bound method
return types.MethodType(self._func, self._obj(), self._class)
else:
# we don''t have an instance: return just the function
return self._func
def is_dead(self):
"""
Returns True if the referenced callable was a bound method and
the instance no longer exists. Otherwise, return False.
"""
return self._obj is not None and self._obj() is None
def __eq__(self, other):
try:
return type(self) is type(other) and self() == other()
except:
return False
def __ne__(self, other):
return not self == other
class proxy(ref):
"""
Exactly like ref, but calling it will cause the referent method to
be called with the same arguments. If the referent''s object no longer lives,
ReferenceError is raised.
If quiet is True, then a ReferenceError is not raise and the callback
silently fails if it is no longer valid.
"""
def __init__(self, method, quiet=False):
super(proxy, self).__init__(method)
self._quiet = quiet
def __call__(self, *args, **kwargs):
func = ref.__call__(self)
if func is None:
if self._quiet:
return
else:
raise ReferenceError(''object is dead'')
else:
return func(*args, **kwargs)
def __eq__(self, other):
try:
func1 = ref.__call__(self)
func2 = ref.__call__(other)
return type(self) == type(other) and func1 == func2
except:
return False
class CallbackEvent(QtCore.QEvent):
"""
A custom QEvent that contains a callback reference
Also provides class methods for conveniently executing
arbitrary callback, to be dispatched to the event loop.
"""
EVENT_TYPE = QtCore.QEvent.Type(QtCore.QEvent.registerEventType())
def __init__(self, func, *args, **kwargs):
super(CallbackEvent, self).__init__(self.EVENT_TYPE)
self.func = func
self.args = args
self.kwargs = kwargs
def callback(self):
"""
Convenience method to run the callable.
Equivalent to:
self.func(*self.args, **self.kwargs)
"""
self.func(*self.args, **self.kwargs)
@classmethod
def post_to(cls, receiver, func, *args, **kwargs):
"""
Post a callable to be delivered to a specific
receiver as a CallbackEvent.
It is the responsibility of this receiver to
handle the event and choose to call the callback.
"""
# We can create a weak proxy reference to the
# callback so that if the object associated with
# a bound method is deleted, it won''t call a dead method
if not isinstance(func, proxy):
reference = proxy(func, quiet=True)
else:
reference = func
event = cls(reference, *args, **kwargs)
# post the event to the given receiver
QtGui.QApplication.postEvent(receiver, event)
## End borrowed code
## Begin Coroutine-framework code
class AsyncTask(QtCore.QObject):
""" Object used to manage asynchronous tasks.
This object should wrap any function that you want
to call asynchronously. It will launch the function
in a new thread, and register a listener so that
`on_finished` is called when the thread is complete.
"""
def __init__(self, func, *args, **kwargs):
super(AsyncTask, self).__init__()
self.result = None # Used for the result of the thread.
self.func = func
self.args = args
self.kwargs = kwargs
self.finished = False
self.finished_cb_ran = False
self.finished_callback = None
self.objThread = RunThreadCallback(self, self.func, self.on_finished,
*self.args, **self.kwargs)
self.objThread.start()
def customEvent(self, event):
event.callback()
def on_finished(self, result):
""" Called when the threaded operation is complete.
Saves the result of the thread, and
executes finished_callback with the result if one
exists. Also closes/cleans up the thread.
"""
self.finished = True
self.result = result
if self.finished_callback:
self.finished_ran = True
func = partial(self.finished_callback, result)
QTimer.singleShot(0, func)
self.objThread.quit()
self.objThread.wait()
class RunThreadCallback(QtCore.QThread):
""" Runs a function in a thread, and alerts the parent when done.
Uses a custom QEvent to alert the main thread of completion.
"""
def __init__(self, parent, func, on_finish, *args, **kwargs):
super(RunThreadCallback, self).__init__(parent)
self.on_finished = on_finish
self.func = func
self.args = args
self.kwargs = kwargs
def run(self):
try:
result = self.func(*self.args, **self.kwargs)
except Exception as e:
print "e is %s" % e
result = e
finally:
CallbackEvent.post_to(self.parent(), self.on_finished, result)
def coroutine(func):
""" Coroutine decorator, meant for use with AsyncTask.
This decorator must be used on any function that uses
the `yield AsyncTask(...)` pattern. It shouldn''t be used
in any other case.
The decorator will yield AsyncTask objects from the
decorated generator function, and register itself to
be called when the task is complete. It will also
excplicitly call itself if the task is already
complete when it yields it.
"""
def wrapper(*args, **kwargs):
def execute(gen, input=None):
if isinstance(gen, types.GeneratorType):
if not input:
obj = next(gen)
else:
try:
obj = gen.send(input)
except StopIteration as e:
result = getattr(e, "value", None)
return result
if isinstance(obj, AsyncTask):
# Tell the thread to call `execute` when its done
# using the current generator object.
func = partial(execute, gen)
obj.finished_callback = func
if obj.finished and not obj.finished_cb_ran:
obj.on_finished(obj.result)
else:
raise Exception("Using yield is only supported with AsyncTasks.")
else:
print("result is %s" % result)
return result
result = func(*args, **kwargs)
execute(result)
return wrapper
## End coroutine-framework code

Si coloca el código anterior en un módulo (digamos qtasync.py ), puede importarlo a un script y usarlo para obtener asyncio comportamiento similar al asyncio :

import sys
import time
from qtasync import AsyncTask, coroutine
from PyQt4 import QtGui
from PyQt4.QtCore import QThread
class MainWindow(QtGui.QMainWindow):
def __init__(self):
super(MainWindow, self).__init__()
self.initUI()
def initUI(self):
self.cmd_button = QtGui.QPushButton("Push", self)
self.cmd_button.clicked.connect(self.send_evt)
self.statusBar()
self.show()
def worker(self, inval):
print "in worker, received ''%s''" % inval
time.sleep(2)
return "%s worked" % inval
@coroutine
def send_evt(self, arg):
out = AsyncTask(self.worker, "test string")
out2 = AsyncTask(self.worker, "another test string")
QThread.sleep(3)
print("kicked off async task, waiting for it to be done")
val = yield out
val2 = yield out2
print ("out is %s" % val)
print ("out2 is %s" % val2)
out = yield AsyncTask(self.worker, "Some other string")
print ("out is %s" % out)
if __name__ == "__main__":
app = QtGui.QApplication(sys.argv)
m = MainWindow()
sys.exit(app.exec_())

Salida (cuando se presiona el botón):

in worker, received ''test string''
in worker, received ''another test string''
kicked off async task, waiting for it to be done
out is test string worked
out2 is another test string worked
in worker, received ''Some other string''
out is Some other string worked

Como puede ver, el worker se ejecuta de forma asíncrona en un hilo cada vez que recibe una llamada a través de la clase AsyncTask , pero su valor de retorno se puede yield directamente desde send_evt , sin necesidad de utilizar devoluciones de llamada.

El código usa las características de soporte de coroutine ( generator_object.send ) de los generadores Python, y una receta que encontré en ActiveState que proporciona un mecanismo de comunicación hijo-> principal, para implementar algunas corutinas muy básicas. Las corutinas son bastante limitadas: no puede devolver nada de ellas, y no puede encadenar llamadas de corotine juntas. Probablemente sea posible implementar ambas cosas, pero probablemente tampoco valga la pena, a menos que realmente las necesites. Tampoco he hecho muchas pruebas negativas con esto, por lo que las excepciones en los trabajadores y en otros lugares pueden no ser manejadas adecuadamente. Lo que sí funciona bien, sin embargo, es permitirle llamar a métodos en hilos separados a través de la clase AsyncTask , y luego yield un resultado del hilo cuando uno está listo, sin bloquear el ciclo de eventos Qt . Normalmente, este tipo de cosas se haría con devoluciones de llamadas, que pueden ser difíciles de seguir y, en general, son menos legibles que tener todo el código en una sola función.

Le invitamos a utilizar este enfoque si las limitaciones que mencioné son aceptables para usted, pero esto es realmente solo una prueba de concepto; Tendría que hacer un montón de pruebas antes de pensar en ponerlo en producción en cualquier lugar.

Como mencionaste, Python 3.3 y 3.4 hacen que la programación asíncrona sea más fácil con la introducción de yield from y asyncio , respectivamente. Creo que el yield from sería en realidad bastante útil aquí para permitir encadenar corutinas (es decir, hacer que una corrutina llame a otra y obtener un resultado de ella). asyncio no tiene integración de bucle de eventos PyQt4, por lo que su utilidad es bastante limitada.

Otra opción sería eliminar la pieza de corutina de este todo y simplemente usar directamente el mecanismo de comunicación entre hilos basado en devolución de llamada :

import sys
import time
from qtasync import CallbackEvent # No need for the coroutine stuff
from PyQt4 import QtGui
from PyQt4.QtCore import QThread
class MyThread(QThread):
""" Runs a function in a thread, and alerts the parent when done.
Uses a custom QEvent to alert the main thread of completion.
"""
def __init__(self, parent, func, on_finish, *args, **kwargs):
super(MyThread, self).__init__(parent)
self.on_finished = on_finish
self.func = func
self.args = args
self.kwargs = kwargs
self.start()
def run(self):
try:
result = self.func(*self.args, **self.kwargs)
except Exception as e:
print "e is %s" % e
result = e
finally:
CallbackEvent.post_to(self.parent(), self.on_finished, result)
class MainWindow(QtGui.QMainWindow):
def __init__(self):
super(MainWindow, self).__init__()
self.initUI()
def initUI(self):
self.cmd_button = QtGui.QPushButton("Push", self)
self.cmd_button.clicked.connect(self.send)
self.statusBar()
self.show()
def customEvent(self, event):
event.callback()
def worker(self, inval):
print("in worker, received ''%s''" % inval)
time.sleep(2)
return "%s worked" % inval
def end_send(self, cmd):
print("send returned ''%s''" % cmd)
def send(self, arg):
t = MyThread(self, self.worker, self.end_send, "some val")
print("Kicked off thread")
if __name__ == "__main__":
app = QtGui.QApplication(sys.argv)
m = MainWindow()
sys.exit(app.exec_())

Salida:

Kicked off thread
in worker, received ''some val''
send returned ''some val worked''

Esto puede ser un poco difícil si se trata de una cadena de devolución de llamada larga, pero no depende del código de coroutine no coroutine .

Answer 3

Si desea un modo simple de ejecutar dead-simple (en términos de líneas de código), puede simplemente crear un QThread y usar una pyqtSignal para alertar al padre cuando el hilo está hecho. Aquí hay dos botones. Uno controla un hilo de fondo que se puede cancelar. El primer impulso desactiva el hilo, y un segundo empuje cancela el hilo de fondo. El otro botón se desactivará automáticamente mientras se ejecuta el hilo de fondo y se volverá a habilitar una vez que se complete.

from PyQt4 import QtGui
from PyQt4 import QtCore
class MainWindow(QtGui.QMainWindow):
def __init__(self):
super(MainWindow, self).__init__()
self.initUI()
self.task = None
def initUI(self):
self.cmd_button = QtGui.QPushButton("Push/Cancel", self)
self.cmd_button2 = QtGui.QPushButton("Push", self)
self.cmd_button.clicked.connect(self.send_cancellable_evt)
self.cmd_button2.clicked.connect(self.send_evt)
self.statusBar()
self.layout = QtGui.QGridLayout()
self.layout.addWidget(self.cmd_button, 0, 0)
self.layout.addWidget(self.cmd_button2, 0, 1)
widget = QtGui.QWidget()
widget.setLayout(self.layout)
self.setCentralWidget(widget)
self.show()
def send_evt(self, arg):
self.t1 = RunThread(self.worker, self.on_send_finished, "test")
self.t2 = RunThread(self.worker, self.on_send_finished, 55)
print("kicked off async tasks, waiting for it to be done")
def worker(self, inval):
print "in worker, received ''%s''" % inval
time.sleep(2)
return inval
def send_cancellable_evt(self, arg):
if not self.task:
self.task = RunCancellableThread(None, self.on_csend_finished, "test")
print("kicked off async task, waiting for it to be done")
else:
self.task.cancel()
print("Cancelled async task.")
def on_csend_finished(self, result):
self.task = None # Allow the worker to be restarted.
print "got %s" % result
def on_send_finished(self, result):
print "got %s. Type is %s" % (result, type(result))
class RunThread(QtCore.QThread):
""" Runs a function in a thread, and alerts the parent when done.
Uses a pyqtSignal to alert the main thread of completion.
"""
finished = QtCore.pyqtSignal(["QString"], [int])
def __init__(self, func, on_finish, *args, **kwargs):
super(RunThread, self).__init__()
self.args = args
self.kwargs = kwargs
self.func = func
self.finished.connect(on_finish)
self.finished[int].connect(on_finish)
self.start()
def run(self):
try:
result = self.func(*self.args, **self.kwargs)
except Exception as e:
print "e is %s" % e
result = e
finally:
if isinstance(result, int):
self.finished[int].emit(result)
else:
self.finished.emit(str(result)) # Force it to be a string by default.
class RunCancellableThread(RunThread):
def __init__(self, *args, **kwargs):
self.cancelled = False
super(RunCancellableThread, self).__init__(*args, **kwargs)
def cancel(self):
self.cancelled = True # Use this if you just want to signal your run() function.
# Use this to ungracefully stop the thread. This isn''t recommended,
# especially if you''re doing any kind of work in the thread that could
# leave things in an inconsistent or corrupted state if suddenly
# terminated
#self.terminate()
def run(self):
try:
start = cur_time = time.time()
while cur_time - start < 10:
if self.cancelled:
print("cancelled")
result = "cancelled"
break
print "doing work in worker..."
time.sleep(1)
cur_time = time.time()
except Exception as e:
print "e is %s" % e
result = e
finally:
if isinstance(result, int):
self.finished[int].emit(result)
else:
self.finished.emit(str(result)) # Force it to be a string by default.
if __name__ == "__main__":
app = QtGui.QApplication(sys.argv)
m = MainWindow()
sys.exit(app.exec_())

Salida (desde presionar "Empujar"):

in worker, received ''test''kicked off async tasks, waiting for it to be done
in worker, received ''55''
got 55. Type is <type ''int''>
got test. Type is <class ''PyQt4.QtCore.QString''>
in worker, received ''test''
in worker, received ''55''

Salida (al presionar "Empujar / Cancelar"):

kicked off async task, waiting for it to be done
doing work in worker...
doing work in worker...
doing work in worker...
doing work in worker...
doing work in worker...
doing work in worker...
<I pushed the button again here>
Cancelled async task.
cancelled
got cancelled

Aquí hay un par de limitaciones molestas:

1. La señal finished no facilita el manejo de tipos arbitrarios. Debe declarar explícitamente y conectar un controlador para cada tipo que desee devolver, y luego asegurarse de emit al controlador correcto cuando obtenga un resultado. Esto se llama sobrecargar la señal. Algunos tipos de Python que tienen la misma firma de C ++ no se comportarán correctamente cuando se usan juntos de esta manera , por ejemplo, pyqtSignal([dict], [list]) . Puede ser más fácil para usted crear un par de subclase QThread diferente para manejar los diferentes tipos que puede devolver de las cosas que está ejecutando en un hilo.
2. RunThread guardar una referencia al objeto RunThread que cree o, de lo contrario, se destruirá inmediatamente cuando salga del ámbito que finaliza al trabajador que se ejecuta en el hilo antes de que se complete. Esto es un poco derrochador porque mantiene una referencia al objeto QThread completado QThread vez que haya terminado con él (a menos que lo limpie en el controlador on_finish , o por algún otro mecanismo).

Answer 4

Esta es la solución que estoy considerando actualmente

Está vagamente inspirado en c # ''s

task.ContinueWith(() => { /*some code*/ }, TaskScheduler.FromCurrentSynchronizationContext());

y algo adaptado a mi situación de subproceso principal de subproceso / trabajo único, aunque se puede generalizar fácilmente creando subprocesos nuevos (eventualmente querrás usar algún tipo de grupo de subprocesos, creo q tiene uno pero no lo he probado).

1. Parece funcionar
2. Creo que entiendo cómo funciona
3. Es bastante corto, recuento de líneas

Básicamente, se programan funciones para ejecutar en subprocesos particulares y se utilizan cierres con anidadas llamadas run_on_thread para cambiar el ui. Para simplificar las cosas, no agregué valores devueltos (solo asigno cosas a objetos / use cierres en vair local.

Realmente no he pensado en cómo pasar excepciones en la cadena.

¿Alguna crítica / sugerencia para mejoras?

Código que incluye código de prueba:

#!/usr/bin/env python
from __future__ import print_function
import sys
from PyQt4 import QtCore
from PyQt4 import QtGui
class RunObjectContainer(QtCore.QObject):
#temporarily holds references so objects don''t get garbage collected
def __init__(self):
self._container = set()
def add(self, obj):
self._container.add(obj)
@QtCore.pyqtSlot(object)
def discard(self, obj):
self._container.discard(obj)
container = RunObjectContainer()
class RunObject(QtCore.QObject):
run_complete = QtCore.pyqtSignal(object)
def __init__(self, parent=None,f=None):
super(RunObject, self).__init__(parent)
self._f = f
@QtCore.pyqtSlot()
def run(self):
self._f()
self.run_complete.emit(self)
main_thread = None
worker_thread = QtCore.QThread()
def run_on_thread(thread_to_use, f):
run_obj = RunObject(f=f)
container.add(run_obj)
run_obj.run_complete.connect(container.discard)
if QtCore.QThread.currentThread() != thread_to_use:
run_obj.moveToThread(thread_to_use)
QtCore.QMetaObject.invokeMethod(run_obj, ''run'', QtCore.Qt.QueuedConnection)
def print_run_on(msg):
if QtCore.QThread.currentThread() == main_thread:
print(msg + " -- run on main thread")
elif QtCore.QThread.currentThread() == worker_thread:
print(msg + " -- run on worker thread")
else:
print("error " + msg + " -- run on unkown thread")
raise Exception(msg + " -- run on unkown thread")
class Example(QtGui.QWidget):
def __init__(self):
super(Example, self).__init__()
self.initUI()
def initUI(self):
self.button = QtGui.QPushButton(''Test'', self)
self.button.clicked.connect(self.handleButton)
self.show()
def handleButton(self):
run_on_thread(main_thread, lambda: print_run_on("main_thread"))
run_on_thread(worker_thread, lambda: print_run_on("worker_thread"))
def n():
a = "yoyoyo"
print_run_on("running function n on thread ")
run_on_thread(main_thread, lambda: print_run_on("calling nested from n "))
run_on_thread(worker_thread, lambda: print_run_on("a is " + a))
run_on_thread(worker_thread, n)
print("end of handleButton")
def gui_main():
app = QtGui.QApplication(sys.argv)
ex = Example()
worker_thread.start()
global main_thread
main_thread = app.thread()
sys.exit(app.exec_())
if __name__ == ''__main__'':
gui_main()

Editar: con un pero de abstracción extra

añadir

def run_on_thread_d(thread_to_use):
def decorator(func):
run_on_thread(thread_to_use, func)
return func
return decorator

y puedes reemplazar el método handleButton

def handleButton(self):
@run_on_thread_d(worker_thread)
def _():
a = "yoyoyo"
print_run_on("running function n on thread ")
@run_on_thread_d(main_thread)
def _():
print_run_on("calling nested from n ")
@run_on_thread_d(worker_thread)
def _():
print_run_on("a is " + a)