Split multithreaded executor

src/ewoksorange/bindings/taskexecuter.py

@@ -1,6 +1,7 @@
 import warnings
 
 from ..gui.concurrency.base import TaskExecutor  # noqa F401
+from ..gui.concurrency.threaded import MultiThreadedTaskExecutor  # noqa F401
 from ..gui.concurrency.threaded import ThreadedTaskExecutor  # noqa F401
 
 warnings.warn(

src/ewoksorange/gui/concurrency/threaded.py

@@ -1,6 +1,12 @@
+from dataclasses import dataclass
+from typing import Callable
+from typing import Iterable
+from typing import List
 from typing import Optional
 
+from AnyQt.QtCore import QObject
 from AnyQt.QtCore import QThread
+from AnyQt.QtCore import pyqtSignal as Signal
 
 from .base import TaskExecutor
 
@@ -29,3 +35,139 @@ def cancel_running_task(self):
         """
         if self.current_task is not None:
             self.current_task.cancel()
+
+
+@dataclass
+class _TaskExecutorState:
+    callbacks: Iterable[Callable[[ThreadedTaskExecutor], None]]
+    task_kwargs: Dict[str, Any]
+    log_missing_inputs: bool = False
+    task_executor: Optional[ThreadedTaskExecutor] = None
+
+
+class MultiThreadedTaskExecutor(QObject):
+    """Create and execute each Ewoks task in its own dedicated thread."""
+
+    sigComputationStarted = Signal()
+    """Signal emitted when a computation is started"""
+
+    sigComputationEnded = Signal()
+    """Signal emitted when a computation is ended"""
+
+    def __init__(self, ewokstaskclass):
+        super().__init__()
+        self.__ewokstaskclass = ewokstaskclass
+        self.__task_executors: List[_TaskExecutorState] = []
+        self.__last_output_variables = dict()
+        self.__last_task_succeeded = None
+        self.__last_task_done = None
+        self.__last_task_exception = None
+
+    def execute_task(
+        self,
+        _callbacks: Iterable[Callable[[ThreadedTaskExecutor], None]] = tuple(),
+        log_missing_inputs: bool = False,
+        **kwargs,
+    ) -> Optional[ThreadedTaskExecutor]:
+        """Execute a prepared task, or directly create and execute a new one."""
+        task_executor = ThreadedTaskExecutor(ewokstaskclass=self.__ewokstaskclass)
+        task_executor.create_task(
+            log_missing_inputs=log_missing_inputs,
+            **kwargs,
+        )
+
+        state = _TaskExecutorState(
+            callbacks=tuple(_callbacks),
+            task_kwargs=kwargs,
+            log_missing_inputs=log_missing_inputs,
+            task_executor=task_executor,
+        )
+        self.__add_task_executor(state)
+
+        if task_executor.has_task:
+            task_executor.finished.connect(self.__process_ended)
+            self.sigComputationStarted.emit()
+            task_executor.start()
+        else:
+            self.__process_ended_direct(task_executor)
+
+        return task_executor
+
+    def __process_ended(self):
+        self.__process_ended_direct(self.sender())
+
+    def _getState(self, task_executor: ThreadedTaskExecutor) -> _TaskExecutorState:
+        return next(
+            (
+                state
+                for state in self.__task_executors
+                if state.task_executor is task_executor
+            ),
+            None,
+        )
+
+    def __process_ended_direct(self, task_executor: ThreadedTaskExecutor):
+        state = self._getState(task_executor=task_executor)
+        if state is None:
+            return
+
+        self.__last_output_variables = task_executor.output_variables
+        self.__last_task_succeeded = task_executor.succeeded
+        self.__last_task_done = task_executor.done
+        self.__last_task_exception = task_executor.exception
+
+        try:
+            for callback in state.callbacks:
+                callback(task_executor)
+        finally:
+            self.sigComputationEnded.emit()
+            self.__remove_task_executor(task_executor)
+
+    def stop(self, timeout: Optional[float] = None, wait: bool = False) -> None:
+        """Stop all tracked task threads."""
+        for state in list(self.__task_executors):
+            task_executor = state.task_executor
+            if task_executor is None:
+                continue
+            if task_executor.receivers(task_executor.finished) > 0:
+                task_executor.finished.disconnect(self.__process_ended)
+            task_executor.stop(timeout=timeout, wait=wait)
+        self.__task_executors.clear()
+
+    def cancel_running_tasks(self, wait=True):
+        """Request cancellation of all running tasks."""
+        for state in list(self.__task_executors):
+            if state.task_executor is None or not state.task_executor.isRunning():
+                continue
+            task_executor = state.task_executor
+            task_executor.cancel_running_task()
+            task_executor.stop(wait=wait)
+
+    def __add_task_executor(self, state: _TaskExecutorState) -> None:
+        self.__task_executors.append(state)
+
+    def __remove_task_executor(self, task_executor: ThreadedTaskExecutor) -> None:
+        if task_executor is None:
+            return
+        for state in list(self.__task_executors):
+            if state.task_executor is task_executor:
+                if task_executor.receivers(task_executor.finished) > 0:
+                    task_executor.finished.disconnect(self.__process_ended)
+                self.__task_executors.remove(state)
+                break
+
+    @property
+    def task_succeeded(self) -> Optional[bool]:
+        return self.__last_task_succeeded
+
+    @property
+    def task_done(self) -> Optional[bool]:
+        return self.__last_task_done
+
+    @property
+    def task_exception(self) -> Optional[Exception]:
+        return self.__last_task_exception
+
+    @property
+    def output_variables(self) -> dict:
+        return self.__last_output_variables

src/ewoksorange/gui/owwidgets/threaded.py

@@ -6,11 +6,10 @@
 
 import logging
 from contextlib import contextmanager
-from typing import Dict
 from typing import Optional
-from typing import Tuple
 
 from ..concurrency.queued import TaskExecutorQueue
+from ..concurrency.threaded import MultiThreadedTaskExecutor
 from ..concurrency.threaded import ThreadedTaskExecutor
 from ..qt_utils.progress import QProgress
 from .base import OWEwoksBaseWidget
@@ -183,113 +182,66 @@ class OWEwoksWidgetOneThreadPerRun(_OWEwoksThreadedBaseWidget, **ow_build_opts):
 
     def __init__(self, *args, **kwargs):
         """
-        Initialize per-run executor storage.
+        Initialize the multi-thread task executor.
         """
         super().__init__(*args, **kwargs)
-        self.__task_executors: Dict[int, Tuple[ThreadedTaskExecutor, bool]] = dict()
-        self.__last_output_variables = dict()
-        self.__last_task_succeeded = None
-        self.__last_task_done = None
-        self.__last_task_exception = None
+        self.__task_executor = MultiThreadedTaskExecutor(
+            ewokstaskclass=self.ewokstaskclass
+        )
 
     def _execute_ewoks_task(self, propagate: bool, log_missing_inputs: bool) -> None:
         """
-        Create a fresh ThreadedTaskExecutor, register it, and start it if it has work.
+        Submit a task to the multi-thread executor.
 
         :param propagate: Whether to propagate outputs after execution.
         :param log_missing_inputs: Whether to log missing input warnings.
         """
-        task_executor = ThreadedTaskExecutor(ewokstaskclass=self.ewokstaskclass)
-        task_executor.create_task(
-            log_missing_inputs=log_missing_inputs, **self._get_task_arguments()
-        )
-        with self.__init_task_executor(task_executor, propagate):
-            if task_executor.has_task:
-                with self._ewoks_task_start_context():
-                    task_executor.start()
-            else:
-                task_executor.finished.emit()
-
-    @contextmanager
-    def __init_task_executor(self, task_executor, propagate: bool):
-        """
-        Register a task executor and connect its finished callback for safe cleanup.
-
-        :param task_executor: The ThreadedTaskExecutor instance.
-        :param propagate: Propagate flag to store with the executor.
-        """
-        task_executor.finished.connect(self._ewoks_task_finished_callback)
-        self.__add_task_executor(task_executor, propagate)
-        try:
-            yield
-        except Exception:
-            task_executor.finished.disconnect(self._ewoks_task_finished_callback)
-            self.__remove_task_executor(task_executor)
-            raise
-
-    def __disconnect_all_task_executors(self):
-        """Disconnect all connected finished signals from tracked executors."""
-        for task_executor, _ in self.__task_executors.values():
-            if task_executor.receivers(task_executor.finished) > 0:
-                task_executor.finished.disconnect(self._ewoks_task_finished_callback)
+        with self._ewoks_task_start_context():
+            self.__task_executor.execute_task(
+                _callbacks=(
+                    lambda task_executor: self._ewoks_task_finished_callback(
+                        task_executor, propagate
+                    ),
+                ),
+                log_missing_inputs=log_missing_inputs,
+                **self._get_task_arguments(),
+            )
 
-    def _ewoks_task_finished_callback(self):
+    def _ewoks_task_finished_callback(
+        self, task_executor: ThreadedTaskExecutor, propagate: bool
+    ):
         """
         Slot invoked when a per-run executor finishes; stores its outputs and optionally propagates.
         """
         with self._ewoks_task_finished_context():
-            task_executor = None
-            try:
-                task_executor = self.sender()
-                self.__last_output_variables = task_executor.output_variables
-                self.__last_task_succeeded = task_executor.succeeded
-                self.__last_task_done = task_executor.done
-                self.__last_task_exception = task_executor.exception
-                self.__post_task_exception = None
-                propagate = self.__is_task_executor_propagated(task_executor)
-                if propagate:
-                    self.propagate_downstream(succeeded=task_executor.succeeded)
-            finally:
-                self.__remove_task_executor(task_executor)
+            self.__post_task_exception = None
+            if propagate:
+                self.propagate_downstream(succeeded=task_executor.succeeded)
 
     def _cleanup_task_executor(self):
-        """Disconnect and quit all tracked executors on widget cleanup."""
-        self.__disconnect_all_task_executors()
-        for task_executor, _ in self.__task_executors.values():
-            task_executor.quit()
-        self.__task_executors.clear()
-
-    def __add_task_executor(self, task_executor, propagate: bool):
-        """Internal: register a new task executor with its propagate flag."""
-        self.__task_executors[id(task_executor)] = task_executor, propagate
-
-    def __remove_task_executor(self, task_executor: ThreadedTaskExecutor):
-        """Internal: unregister a task executor and disconnect its signals."""
-        if task_executor is None:
-            return
-        if task_executor.receivers(task_executor.finished) > 0:
-            task_executor.finished.disconnect(self._ewoks_task_finished_callback)
-        self.__task_executors.pop(id(task_executor), None)
-
-    def __is_task_executor_propagated(self, task_executor) -> bool:
-        """Return whether the given executor was registered to propagate."""
-        return self.__task_executors.get(id(task_executor), (None, False))[1]
+        """Stop all tracked per-run executors on widget cleanup."""
+        self.__task_executor.stop()
+        self.__task_executor = None
 
     @property
     def task_succeeded(self) -> Optional[bool]:
-        return self.__last_task_succeeded
+        return self.__task_executor.task_succeeded
 
     @property
     def task_done(self) -> Optional[bool]:
-        return self.__last_task_done
+        return self.__task_executor.task_done
 
     @property
     def task_exception(self) -> Optional[Exception]:
-        return self.__last_task_exception
+        return self.__task_executor.task_exception
 
     def _get_task_outputs(self) -> dict:
         """Return the last finished task's outputs."""
-        return self.__last_output_variables
+        return self.__task_executor.output_variables
+
+    def cancel_running_task(self):
+        """Request cancellation of all running per-run tasks."""
+        self.__task_executor.cancel_running_tasks()
 
 
 class OWEwoksWidgetWithTaskStack(_OWEwoksThreadedBaseWidget, **ow_build_opts):

src/ewoksorange/tests/test_task_executor.py

@@ -7,6 +7,7 @@
 
 from ..gui.concurrency.base import TaskExecutor
 from ..gui.concurrency.queued import TaskExecutorQueue
+from ..gui.concurrency.threaded import MultiThreadedTaskExecutor
 from ..gui.concurrency.threaded import ThreadedTaskExecutor
 from ..gui.qt_utils.app import QtEvent
 
@@ -51,6 +52,48 @@ def finished_callback():
     executor.finished.disconnect(finished_callback)
 
 
+def test_multi_threaded_task_executor(qtapp):
+    class MyObject(QObject):
+        def __init__(self, expected_results):
+            self.results = None
+            self.expected_results = expected_results
+            self.finished = QtEvent()
+
+        def finished_callback(self, task_executor):
+            self.results = {
+                k: v.value for k, v in task_executor.output_variables.items()
+            }
+            assert self.results == self.expected_results
+            self.finished.set()
+
+    executor = MultiThreadedTaskExecutor(ewokstaskclass=SumTask)
+    objects = [
+        MyObject({"result": 3}),
+        MyObject({"result": 7}),
+        MyObject({"result": 11}),
+    ]
+    inputs = [
+        {"a": 1, "b": 2},
+        {"a": 3, "b": 4},
+        {"a": 5, "b": 6},
+    ]
+
+    for obj, input_values in zip(objects, inputs):
+        executor.execute_task(
+            inputs=input_values,
+            _callbacks=(obj.finished_callback,),
+        )
+        executor.execute_task()
+
+    for obj in objects:
+        assert obj.finished.wait(timeout=3)
+        assert obj.results == obj.expected_results
+
+    expected_results = [obj.expected_results for obj in objects]
+    results = {k: v.value for k, v in executor.output_variables.items()}
+    assert results in expected_results
+
+
 def test_threaded_task_executor_queue(qtapp):
     class MyObject(QObject):
         def __init__(self):