Rename TaskFuture.abort to request_stop, execute subprocess tasks with pebble, add abort and kill

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Support OASYS2 (`pip install ewoksorange[oasys2]`) next to the standard
   `orange-canvas-core` + `orange-widget-base` and the optional Orange3.
 - `TaskFuture.output_values`, `TaskFuture.succeeded` and `TaskFuture.task_exception`.
+- `TaskFuture.abort` and `TaskFuture.kill` to interrupt or kill a task running in a subprocess.
 
 ### Changed
 
@@ -25,6 +26,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Support for OASYS1. Use OASYS2 instead.
 
+### Fixed
+
+- Random crash on Windows at the end of a test session that uses the `ewoksorange_qtapp` fixture.
+
 ## [6.0.0rc7] - 2026-08-18
 
 ### Fixed

doc/explanations/execution.rst

@@ -56,8 +56,8 @@ One execution cycle
 
         BG-->>BG: task runs
         Note right of BG: has_pending_task() -> still True<br/>(succeeded/failed not delivered yet)
-        BG--)EX: Future done callback
-        EX--)W: succeeded/failed (queued across threads)
+        BG--)EX: _done (queued across threads)
+        EX-)W: succeeded/failed
 
         Note over W: GUI thread now runs __on_succeeded/__on_failed
         W->>W: propagate_downstream()
@@ -76,9 +76,9 @@ thread. The other backends only change who runs the task:
     Every step runs on the Qt main thread, inside the ``submit_task()`` call, so
     ``handleNewSignals()`` only returns once the outputs have been propagated.
 ``concurrency="process"``
-    A worker process runs the task. Its lifecycle events and progress travel
-    back, in order, through a single ``multiprocessing`` manager queue, relayed
-    onto the Qt main thread by
+    A worker process runs the task. Its "about to start" and "started" events
+    and its progress travel back, in order, through a ``multiprocessing``
+    manager queue, relayed onto the Qt main thread by a thread of
     :class:`~ewoksorange.gui.concurrency._controllers.process.ProcessTaskController`.
 
 See :doc:`concurrency` for how each backend keeps the signals in order.

doc/reference.rst

@@ -5,3 +5,4 @@ Reference
     :maxdepth: 1
 
     reference/api
+    reference/glossary

doc/reference/glossary.rst

@@ -0,0 +1,114 @@
+Glossary
+========
+
+.. glossary::
+    :sorted:
+
+    Task
+        An `ewoks <https://ewoks.readthedocs.io/>`_ task: a class deriving from
+        ``ewokscore.Task`` that declares its inputs and outputs and implements ``run()``.
+        Each execution creates a new instance of the task class.
+
+    Widget
+        An Orange widget that executes an ewoks :term:`task`, deriving from
+        :class:`~ewoksorange.gui.owwidgets.base.OWEwoksBaseWidget`. It submits its task to
+        its :term:`executor` when it is triggered (by default when its inputs change), then
+        propagates the outputs downstream when the task succeeds, or invalidates them when
+        it fails.
+
+    Execution
+        One run of a :term:`task`, from its submission to the :term:`executor` to the
+        ``finished`` signal. Each submission is a new execution, with its own
+        :term:`task future`.
+
+    Queued execution
+        An :term:`execution` that was submitted but did not start yet: it waits for a free
+        :term:`worker` (and, in the ``"process"`` :term:`backend`, for the
+        controller's permission to start).
+
+    Submit policy
+        What the :term:`executor` does with a submission while an :term:`execution` is
+        running, selected by the ``submit_policy`` class argument of a :term:`widget`:
+        ``"always"`` (the default) queues it, ``"drop_if_busy"`` drops it.
+
+    Executor
+        The :class:`~ewoksorange.gui.concurrency.executor.EwoksExecutor` that executes the
+        :term:`tasks <task>` of a :term:`widget` with a :term:`backend` and emits Qt signals
+        for each step of an :term:`execution`. Each signal carries the :term:`task future`
+        of the execution, except ``ignored``.
+
+    Backend
+        Where the :term:`executor` executes a :term:`task`, selected by the ``concurrency``
+        class argument of a :term:`widget`
+        (:class:`~ewoksorange.gui.concurrency.executor.Concurrency`): ``"sync"`` in the
+        :term:`GUI thread`, ``"thread"`` (the default) in a thread pool or ``"process"`` in
+        a process pool.
+
+    Worker
+        A thread of the thread pool or a process of the process pool that executes
+        :term:`tasks <task>`, one at a time.
+
+    Task future
+        The :class:`~ewoksorange.gui.concurrency.future.TaskFuture` returned by
+        :meth:`~ewoksorange.gui.concurrency.executor.EwoksExecutor.submit_task`. It gives
+        the outputs or the error of the execution, and can :term:`cancel`,
+        :term:`stop <stop request>`, :term:`abort` or :term:`kill` it.
+
+    Progress object
+        The optional ``progress`` task argument, usually a
+        :class:`~ewoksorange.gui.qt_utils.progress.QProgress`. A task deriving from
+        ``ewokscore.TaskWithProgress`` reports its progress, as a percentage, by assigning
+        ``self.progress``. A ``QProgress`` emits ``sigProgressChanged`` when the percentage
+        changes, in the thread that assigns it.
+
+    GUI thread
+        The main thread of the Qt application, in which the :term:`widgets <widget>` and the
+        :term:`executor` live.
+
+    Queued connection
+        How Qt delivers a signal emitted in a thread other than the thread of the receiver:
+        the signal is posted to the event queue of the receiver's thread, and the connected
+        slots run later, when that thread processes its events. Signals posted to the same
+        thread are processed in the order they were posted. Qt uses it by default for
+        signals emitted in another thread.
+
+    Cancel
+        Prevent a queued :term:`task` from starting
+        (:meth:`TaskFuture.cancel() <ewoksorange.gui.concurrency.future.TaskFuture.cancel>`).
+        It has no effect on a running task.
+
+    Stop request
+        Ask a :term:`task` to stop
+        (:meth:`TaskFuture.request_stop() <ewoksorange.gui.concurrency.future.TaskFuture.request_stop>`).
+        It is cooperative: a running task decides whether and how to stop, by checking
+        ``Task.stop_requested`` or implementing ``Task._on_stop_requested``.
+
+    Abort
+        Interrupt a :term:`task` running in a :term:`worker` process like Ctrl-C
+        (:meth:`TaskFuture.abort() <ewoksorange.gui.concurrency.future.TaskFuture.abort>`),
+        also when it ignores :term:`stop requests <stop request>`. The task can clean up and
+        the :term:`worker` survives.
+
+    Kill
+        Terminate the :term:`worker` process running a :term:`task`
+        (:meth:`TaskFuture.kill() <ewoksorange.gui.concurrency.future.TaskFuture.kill>`). It
+        also works when the task does not return to the Python interpreter, but the task
+        cannot clean up. The pool replaces the :term:`worker`.
+
+    pebble
+        The `third-party process pool <https://github.com/noxdafox/pebble>`_ of the
+        ``"process"`` :term:`backend`. Unlike ``concurrent.futures.ProcessPoolExecutor``, it
+        can :term:`kill` one task without breaking the pool.
+
+    Manager
+        The ``multiprocessing`` manager: a separate server process holding the queues
+        through which the :term:`GUI thread`'s process and the :term:`worker` process
+        communicate in the ``"process"`` :term:`backend`. Each :term:`executor` starts one
+        manager process on its first :term:`execution` and stops it on shutdown. All its
+        executions share it.
+
+    Relay
+        A thread of the :term:`GUI thread`'s process that reads the messages of a
+        :term:`worker` process from the :term:`manager` queue and turns them into Qt signals.
+        It also replies whether the task may start. One relay thread runs per execution in
+        the ``"process"`` :term:`backend`.

pyproject.toml

@@ -19,6 +19,7 @@ dependencies = [
     "ewokscore >=5.1.0rc5",
     "orange-canvas-core>=0.1.35",
     "orange-widget-base>=4.22.0",
+    "pebble >=5.1",
     "importlib-resources; python_version<'3.9'",
     "importlib-metadata; python_version<'3.10'",
     "backports.strenum; python_version<'3.11'",

scripts/ewoks_executor_demo.py

@@ -4,13 +4,14 @@
 import sys
 import threading
 import time
-from concurrent.futures import ProcessPoolExecutor
 from concurrent.futures import ThreadPoolExecutor
 from functools import partial
 from typing import Any
+from typing import Callable
 from typing import Dict
 from typing import Optional
 
+import pebble
 from AnyQt.QtCore import QEventLoop
 from AnyQt.QtCore import Qt
 from AnyQt.QtCore import QTimer
@@ -30,6 +31,7 @@
 
 from ewoksorange.gui.concurrency.executor import EwoksExecutor
 from ewoksorange.gui.concurrency.executor import SubmitPolicy
+from ewoksorange.gui.concurrency.future import TaskFuture
 
 
 class SumTask(
@@ -38,7 +40,7 @@ class SumTask(
     optional_input_names=["b", "delay", "fail"],
     output_names=["result"],
 ):
-    """Add two numbers with a delay and optional failure."""
+    """Add two numbers with a delay and optional failure. Stops early on request."""
 
     def run(self) -> None:
         result = self.inputs.a
@@ -47,13 +49,29 @@ def run(self) -> None:
             result += self.inputs.b
 
         if self.inputs.delay:
-            time.sleep(self.inputs.delay)
+            end = time.monotonic() + self.inputs.delay
+            while time.monotonic() < end:
+                if self.stop_requested:
+                    raise RuntimeError("Stopped on request")
+                time.sleep(0.05)
 
         if self.inputs.fail:
             raise RuntimeError(f"Intentional failure after {self.inputs.delay}s")
 
         self.outputs.result = result
 
+    def _on_stop_requested(self) -> None:
+        # `run` polls `stop_requested`.
+        pass
+
+
+_ACTIONS: Dict[str, Callable[[TaskFuture], bool]] = {
+    "Cancel (queued only)": TaskFuture.cancel,
+    "Request stop": TaskFuture.request_stop,
+    "Abort (process only)": TaskFuture.abort,
+    "Kill (process only)": TaskFuture.kill,
+}
+
 
 class Window(QWidget):
     def __init__(
@@ -76,7 +94,9 @@ def __init__(
 
         self._pending = QListWidget()
         self._pending_items: Dict[int, QListWidgetItem] = {}
+        self._futures: Dict[int, TaskFuture] = {}
         self._buttons: Dict[str, QPushButton] = {}
+        self._action_buttons: Dict[str, QPushButton] = {}
 
         self._counter: int = 1
         self._total_tasks_triggered: int = 0
@@ -111,15 +131,15 @@ def __init__(
             ),
             # Processes
             "P DROP": EwoksExecutor(
-                ProcessPoolExecutor(max_workers=1),
+                pebble.ProcessPool(max_workers=1),
                 SubmitPolicy.DROP_IF_BUSY,
             ),
             "P QUEUE": EwoksExecutor(
-                ProcessPoolExecutor(max_workers=1),
+                pebble.ProcessPool(max_workers=1),
                 SubmitPolicy.ALWAYS,
             ),
             "P PARALLEL": EwoksExecutor(
-                ProcessPoolExecutor(max_workers=4),
+                pebble.ProcessPool(max_workers=4),
                 SubmitPolicy.ALWAYS,
             ),
         }
@@ -130,7 +150,10 @@ def __init__(
             executor.ignored.connect(partial(self._on_ignored, name))
             executor.succeeded.connect(partial(self._on_succeeded, name))
             executor.failed.connect(partial(self._on_failed, name))
+            executor.cancelled.connect(partial(self._on_cancelled, name))
+            executor.stopRequested.connect(partial(self._on_stop_requested, name))
             executor.aborted.connect(partial(self._on_aborted, name))
+            executor.killed.connect(partial(self._on_killed, name))
             executor.finished.connect(partial(self._on_finished, name))
 
         layout = QVBoxLayout(self)
@@ -162,6 +185,15 @@ def __init__(
 
         layout.addLayout(grid)
 
+        actions = QHBoxLayout()
+        actions.addWidget(QLabel("Apply to all queued and running tasks:"))
+        for text in _ACTIONS:
+            button = QPushButton(text)
+            button.clicked.connect(partial(self._on_action_clicked, text))
+            actions.addWidget(button)
+            self._action_buttons[text] = button
+        layout.addLayout(actions)
+
         body = QHBoxLayout()
 
         log_box = QVBoxLayout()
@@ -170,7 +202,7 @@ def __init__(
         body.addLayout(log_box, 2)
 
         pending_box = QVBoxLayout()
-        pending_box.addWidget(QLabel("Pending tasks"))
+        pending_box.addWidget(QLabel("Queued and running tasks"))
         pending_box.addWidget(self._pending)
         body.addLayout(pending_box, 1)
 
@@ -182,8 +214,9 @@ def _append_log(self, message: str) -> None:
         print(log)
         self._log.append(log)
 
-    def _on_submitted(self, name: str, future: Any) -> None:
+    def _on_submitted(self, name: str, future: TaskFuture) -> None:
         self._append_log(f"[{name}] Submitted -> {id(future)}")
+        self._futures[id(future)] = future
 
         item = QListWidgetItem(f"[{name}] {id(future)}")
         self._pending.addItem(item)
@@ -205,15 +238,25 @@ def _on_failed(self, name: str, future: Any) -> None:
         exc = future.exception()
         self._append_log(f"[{name}] ❌ {id(future)} -> {exc}")
 
+    def _on_cancelled(self, name: str, future: Any) -> None:
+        self._append_log(f"[{name}] 🚫 Cancelled -> {id(future)}")
+
+    def _on_stop_requested(self, name: str, future: Any) -> None:
+        self._append_log(f"[{name}] ⏹ Stop requested -> {id(future)}")
+
     def _on_aborted(self, name: str, future: Any) -> None:
-        self._append_log(f"[{name}] ⏹ Aborted -> {id(future)}")
+        self._append_log(f"[{name}] ⛔ Aborted -> {id(future)}")
+
+    def _on_killed(self, name: str, future: Any) -> None:
+        self._append_log(f"[{name}] 💀 Killed -> {id(future)}")
 
     def _on_finished(self, name: str, future: Any) -> None:
         self._total_tasks_handled += 1
         self._append_log(f"[{name}] ■ Finished -> {id(future)}")
         self._remove_pending(future)
 
     def _remove_pending(self, future: Any) -> None:
+        self._futures.pop(id(future), None)
         item: Optional[QListWidgetItem] = self._pending_items.pop(id(future), None)
 
         if item is not None:
@@ -226,6 +269,12 @@ def _on_button_clicked(
     ) -> None:
         self._submit_many(self._executors[key])
 
+    def _on_action_clicked(self, text: str, _checked: bool = False) -> None:
+        action = _ACTIONS[text]
+        futures = [future for future in self._futures.values() if not future.done()]
+        accepted = sum(action(future) for future in futures)
+        self._append_log(f"{text}: accepted by {accepted}/{len(futures)} tasks")
+
     def _submit_many(self, executor: EwoksExecutor) -> None:
         for _ in range(self._submit_count):
             executor.submit_task(
@@ -254,7 +303,8 @@ def is_finished(self, total_tasks: int) -> bool:
 
 
 def run_self_test(window: Window, submit_count: int, timeout: float) -> bool:
-    """Click every button and wait until all submitted tasks finish.
+    """Click every submit button, then every action button, and wait until all
+    submitted tasks finish.
 
     Used in CI to smoke-test the demo without a real display.
     """
@@ -272,6 +322,9 @@ def on_timeout() -> None:
         QTest.mouseClick(button, Qt.MouseButton.LeftButton)
         total_tasks_expected += submit_count
 
+    for button in window._action_buttons.values():
+        QTest.mouseClick(button, Qt.MouseButton.LeftButton)
+
     loop = QEventLoop()
 
     poll_timer = QTimer()