PR: Rework taskmanager thread management

qtapputils/managers/taskmanagers.py

@@ -17,7 +17,7 @@
 from time import sleep
 
 # ---- Third party imports
-from qtpy.QtCore import QObject, QThread, Signal
+from qtpy.QtCore import QObject, QThread, Signal, Qt
 
 # ---- Local imports
 from qtapputils.qthelpers import qtwait
@@ -67,7 +67,6 @@ def run_tasks(self):
             self.sig_task_completed.emit(task_uuid4, returned_values)
 
         self._tasks.clear()
-        self.thread().quit()
 
 
 class TaskManagerBase(QObject):
@@ -76,6 +75,7 @@ class TaskManagerBase(QObject):
     """
     sig_run_tasks_started = Signal()
     sig_run_tasks_finished = Signal()
+    sig_run_tasks = Signal()
 
     def __init__(self, verbose: bool = False):
         super().__init__()
@@ -102,9 +102,7 @@ def __init__(self, verbose: bool = False):
 
     @property
     def is_running(self):
-        return not (len(self._running_tasks) == 0 and
-                    len(self._pending_tasks) == 0 and
-                    not self._thread.isRunning())
+        return len(self._running_tasks + self._pending_tasks) > 0
 
     def run_tasks(
             self, callback: Callable = None, returned_values: tuple = None):
@@ -134,13 +132,16 @@ def worker(self) -> WorkerBase:
 
     def set_worker(self, worker: WorkerBase):
         """"Install the provided worker on this manager"""
-        self._worker = worker
         self._thread = QThread()
+
+        self._worker = worker
         self._worker.moveToThread(self._thread)
-        self._thread.started.connect(self._worker.run_tasks)
+        self._worker.sig_task_completed.connect(
+            self._handle_task_completed, Qt.QueuedConnection)
+        self.sig_run_tasks.connect(
+            self._worker.run_tasks, Qt.QueuedConnection)
 
-        # Connect the worker signals to handlers.
-        self._worker.sig_task_completed.connect(self._handle_task_completed)
+        self._thread.start()
 
     # ---- Private API
     def _handle_task_completed(
@@ -202,22 +203,19 @@ def _run_pending_tasks(self):
                 print('Executing {} pending tasks...'.format(
                     len(self._pending_tasks)))
 
-            # Even though the worker has executed all its tasks,
-            # we may still need to wait a little for it to stop properly.
-            i = 0
-            while self._thread.isRunning():
-                sleep(0.1)
-                i += 1
-                if i > 100:
-                    print("Error: unable to stop {}'s working thread.".format(
-                        self.__class__.__name__))
-
             self._running_tasks = self._pending_tasks.copy()
             self._pending_tasks = []
             for task_uuid4 in self._running_tasks:
                 task, args, kargs = self._task_data[task_uuid4]
                 self._worker.add_task(task_uuid4, task, *args, **kargs)
-            self._thread.start()
+
+            self.sig_run_tasks.emit()
+
+    def close(self):
+        if hasattr(self, "_thread"):
+            qtwait(lambda: not self.is_running)
+            self._thread.quit()
+            self._thread.wait()
 
 
 class LIFOTaskManager(TaskManagerBase):

qtapputils/managers/tests/test_taskmanagers.py

@@ -52,8 +52,9 @@ def task_manager(worker, qtbot):
     task_manager.set_worker(worker)
     yield task_manager
 
-    # We wait for the manager's thread to fully stop to avoid segfault error.
-    qtbot.waitUntil(lambda: not task_manager.is_running)
+    task_manager.close()
+    assert not task_manager.is_running
+    assert not task_manager._thread.isRunning()
 
 
 @pytest.fixture
@@ -62,8 +63,9 @@ def lifo_task_manager(worker, qtbot):
     task_manager.set_worker(worker)
     yield task_manager
 
-    # We wait for the manager's thread to fully stop to avoid segfault error.
-    qtbot.waitUntil(lambda: not task_manager.is_running)
+    task_manager.close()
+    assert not task_manager.is_running
+    assert not task_manager._thread.isRunning()
 
 
 # =============================================================================