From 459b11188378de04f95505da950edae8a3df67f3 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Wed, 21 Oct 2026 11:44:01 +0200 Subject: [PATCH] Execute subprocess tasks with pebble and add TaskFuture.abort and TaskFuture.kill Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + doc/explanations/concurrency.rst | 34 +- pyproject.toml | 1 + scripts/ewoks_executor_demo.py | 67 +++- .../gui/concurrency/_controllers/abstract.py | 36 ++- .../gui/concurrency/_controllers/process.py | 51 ++- src/ewoksorange/gui/concurrency/_messages.py | 57 +++- .../gui/concurrency/_runners/abstract.py | 10 +- .../gui/concurrency/_runners/process.py | 60 +++- src/ewoksorange/gui/concurrency/executor.py | 69 ++-- src/ewoksorange/gui/concurrency/future.py | 74 ++++- src/ewoksorange/gui/owwidgets/base.py | 9 +- src/ewoksorange/tests/executor/signals.py | 4 + src/ewoksorange/tests/executor/tasks.py | 148 +++++++++ .../tests/executor/test_executor_abort.py | 304 ++++++++++++++++++ .../tests/executor/test_executor_actions.py | 74 ++++- .../tests/executor/test_executor_order.py | 31 +- .../tests/executor/test_executor_parallel.py | 27 +- .../tests/executor/test_executor_progress.py | 37 +++ 19 files changed, 1030 insertions(+), 64 deletions(-) create mode 100644 src/ewoksorange/tests/executor/test_executor_abort.py diff --git a/CHANGELOG.md b/CHANGELOG.md index dbb9e4d6..44578acb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,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 diff --git a/doc/explanations/concurrency.rst b/doc/explanations/concurrency.rst index 43d5e51b..ac84a463 100644 --- a/doc/explanations/concurrency.rst +++ b/doc/explanations/concurrency.rst @@ -13,12 +13,14 @@ backend, receivers observe the same order of signals for a task: that is cancelled before it starts emits ``submitted`` → ``cancelled`` → ``finished``. A task that is requested to stop before it starts emits ``submitted`` → ``stopRequested`` → ``failed`` → ``finished``. ``finished`` is -the last signal of every submitted task. +the last signal of every submitted task. ``aborted`` and ``killed`` (process +pool only) follow the same rules as ``stopRequested``. In the ``"thread"`` and ``"process"`` backends, the executor does not emit the -completion signals (``cancelled``, ``stopRequested``, ``succeeded``, -``failed``, ``finished``) from the callback that detects the completion. That -callback can run in any thread, so it emits the internal ``_done`` signal, +completion signals (``cancelled``, ``stopRequested``, ``aborted``, +``killed``, ``succeeded``, ``failed``, ``finished``) from the callback that +detects the completion. That callback can run in any thread, including the GUI +thread (in ``kill()``), so it emits the internal ``_done`` signal, which is always queued to the GUI thread. The completion signals are then emitted in the GUI thread, after the signals of the task that were queued before. @@ -119,14 +121,19 @@ is not picklable). The ``ProcessTaskController`` in the GUI process and the object is replaced by a :class:`~ewoksorange.gui.concurrency._messages.QueueProgress`, which puts each progress value on that same queue. + While the task executes, a sender thread of the runner puts the progress + values, so that an abort never interrupts the task thread in the middle of + a queue call. - The queue to the runner carries the decision whether the task may start - (``ToRunner.START`` or a ``StartDenial``), then ``ToRunner.REQUEST_STOP``. + (``ToRunner.START`` or a ``StartDenial``), then ``ToRunner.REQUEST_STOP`` + and ``ToRunner.ABORT``. ``ToRunner.ABORT`` interrupts the task with a + ``KeyboardInterrupt``. In the GUI process, a single relay thread started by :meth:`ProcessTaskController.watch() ` reads the queue to the controller and calls the matching callback. On ``ToController.ABOUT_TO_START``, it decides whether the task may start (it may -not once it was cancelled or requested to stop), replies to the runner, and +not once it was cancelled, stopped, aborted or killed), replies to the runner, and emits ``aboutToStart`` when allowed. On ``ToController.STARTED`` it emits ``started``, and on a progress value it sets the value on the caller's progress object. The relay only starts once ``submitted`` was emitted, so the runner @@ -143,8 +150,8 @@ cannot start the task before that. box GUI process: relay thread participant RL as Relay end - box GUI process: pool management thread - participant PM as ProcessPool mgmt + box GUI process: pebble pool threads + participant PM as pebble.ProcessPool end box Worker process participant R as ProcessTaskRunner + Task @@ -193,7 +200,7 @@ practice both run concurrently. What matters is the ordering: are synchronous, so every event of the runner is already queued ahead of ``END``. Waiting for the relay to reach ``END`` therefore delivers all of them before ``succeeded`` / ``failed`` are emitted. -- **The start decision comes first.** Stop requests are only put on the queue +- **The start decision comes first.** Stop requests and aborts are only put on the queue to the runner once the start was allowed, so the runner always reads the decision first. @@ -202,8 +209,13 @@ The relay's lifetime is tied to the task's future, not to a timeout: - **Cancelled or stopped while queued.** The runner only puts ``ABOUT_TO_START``, is denied and returns or raises, so the done callback runs and puts ``END``. -- **Worker crash.** The pool sets ``BrokenProcessPool`` on the future, so the - done callback still runs and puts ``END``. +- **Worker crash.** The pool sets ``pebble.ProcessExpired`` on the future and + replaces the worker process, so the done callback still runs and puts + ``END``. +- **Killed worker.** ``kill()`` cancels the future of the pool, which runs the + done callback right away, in the thread calling ``kill()``. When a worker + already picked up the task, the pool terminates that worker process and + replaces it. - **Manager shut down.** Reading the queue raises and the relay exits. - **Long-running task.** The relay keeps waiting, since progress may still come. There is deliberately no timeout on reading the queue: a timeout would diff --git a/pyproject.toml b/pyproject.toml index 867a6eef..ca12eb65 100644 --- a/pyproject.toml +++ b/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'", diff --git a/scripts/ewoks_executor_demo.py b/scripts/ewoks_executor_demo.py index 4b076c53..acb2511a 100644 --- a/scripts/ewoks_executor_demo.py +++ b/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, ), } @@ -132,6 +152,8 @@ def __init__( 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) @@ -163,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() @@ -171,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) @@ -183,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) @@ -212,12 +244,19 @@ def _on_cancelled(self, name: str, future: Any) -> None: 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)}") + + 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: @@ -230,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( @@ -258,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. """ @@ -276,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() diff --git a/src/ewoksorange/gui/concurrency/_controllers/abstract.py b/src/ewoksorange/gui/concurrency/_controllers/abstract.py index fe9ca19e..45d53297 100644 --- a/src/ewoksorange/gui/concurrency/_controllers/abstract.py +++ b/src/ewoksorange/gui/concurrency/_controllers/abstract.py @@ -13,6 +13,8 @@ class StartDenial(Enum): CANCELLED = auto() STOPPED = auto() + ABORTED = auto() + KILLED = auto() class StartGate: @@ -53,8 +55,8 @@ class TaskController(ABC): """Interface for controlling an ewoks task. The task (instance) is expected to be executed at most once. Only the first - of cancel and stop that reaches a task before it starts has an effect: the - task never starts. + of cancel, stop, abort and kill that reaches a task before it starts has an + effect: the task never starts. """ def __init__(self) -> None: @@ -79,10 +81,27 @@ def _send_stop_request(self) -> None: """Request the running task to stop.""" raise NotImplementedError + def started(self) -> bool: + """Return True if the task was allowed to start.""" + return self._gate.started + def stop_requested(self) -> bool: """Return True if the stop request prevented the start or reached the task.""" return self._gate.denial is StartDenial.STOPPED or self._stop_sent + def abort(self) -> bool: + """Request to interrupt the task, or prevent it from starting. + + Not supported by default. + + Returns True if the abort request prevented the start or was sent. + """ + return False + + def aborted(self, future: Future) -> bool: + """Return True if an abort request prevented the start or interrupted the task.""" + return False + def cancel(self, future: Future) -> bool: """Prevent execution if the task has not started. @@ -94,6 +113,19 @@ def cancelled(self, future: Future) -> bool: """Return True if the task was cancelled before it started.""" return self._gate.denial is StartDenial.CANCELLED + def kill(self, future: Future) -> bool: + """Kill the task running with `future`, or prevent it from starting. + + Not supported by default. + + Returns True if the task is killed or prevented from starting. + """ + return False + + def killed(self) -> bool: + """Return True if the task was killed or prevented from starting by a kill request.""" + return False + def watch( self, on_about_to_start: Callable[[], None], diff --git a/src/ewoksorange/gui/concurrency/_controllers/process.py b/src/ewoksorange/gui/concurrency/_controllers/process.py index 7d51ff15..6784ae0a 100644 --- a/src/ewoksorange/gui/concurrency/_controllers/process.py +++ b/src/ewoksorange/gui/concurrency/_controllers/process.py @@ -1,11 +1,13 @@ import logging import threading +from concurrent.futures import Future from queue import Queue from typing import Callable from typing import Optional from .._messages import ToController from .._messages import ToRunner +from .abstract import StartDenial from .abstract import TaskController _logger = logging.getLogger(__name__) @@ -19,7 +21,7 @@ class ProcessTaskController(TaskController): start", "started" and progress through the queue to the controller, relayed by a single thread, so the callbacks are called in the order the runner put them. On "about to start", the runner waits for the controller to allow the start through - the queue to the runner, which then carries the stop requests. + the queue to the runner, which then carries the stop and abort requests. """ def __init__(self, to_controller_queue: Queue, to_runner_queue: Queue): @@ -28,6 +30,8 @@ def __init__(self, to_controller_queue: Queue, to_runner_queue: Queue): self._to_runner_queue = to_runner_queue # The start decision must be the first message to the runner. self._to_runner_lock = threading.Lock() + self._abort_sent = False + self._killed = False self._relay_thread: Optional[threading.Thread] = None def watch( @@ -92,6 +96,11 @@ def stop_watching(self) -> None: except Exception: _logger.debug("failed to stop the events relay", exc_info=True) + if threading.current_thread() is self._relay_thread: + # `kill()` from a slot connected with `Qt.DirectConnection`: the relay + # reads the remaining events once that slot returns. + return + self._relay_thread.join(timeout=5.0) if self._relay_thread.is_alive(): _logger.debug("events relay thread did not terminate in time") @@ -99,3 +108,43 @@ def stop_watching(self) -> None: def _send_stop_request(self) -> None: with self._to_runner_lock: self._to_runner_queue.put(ToRunner.REQUEST_STOP) + + def abort(self) -> bool: + if self._gate.deny(StartDenial.ABORTED): + return True + if not self._gate.started: + return False + self._abort_sent = True + with self._to_runner_lock: + self._to_runner_queue.put(ToRunner.ABORT) + return True + + def aborted(self, future: Future) -> bool: + if self._gate.denial is StartDenial.ABORTED: + return True + if not self._abort_sent or not future.done() or future.cancelled(): + return False + # Circular import: the future module imports the controllers. + from ..future import TaskAbortedError + + return isinstance(future.exception(), TaskAbortedError) + + def kill(self, future: Future) -> bool: + if future.done(): + return False + if self._gate.deny(StartDenial.KILLED): + # Completes the future of a queued task right away. Cancelling the pebble + # future terminates the worker when one already picked up the task. + future.cancel() + return True + if not self._gate.started: + return False + # Set before cancelling: the done callbacks run inside `cancel()`. + self._killed = True + if future.cancel(): + return True + self._killed = False + return False + + def killed(self) -> bool: + return self._gate.denial is StartDenial.KILLED or self._killed diff --git a/src/ewoksorange/gui/concurrency/_messages.py b/src/ewoksorange/gui/concurrency/_messages.py index 05e23d32..32ce5064 100644 --- a/src/ewoksorange/gui/concurrency/_messages.py +++ b/src/ewoksorange/gui/concurrency/_messages.py @@ -4,12 +4,14 @@ Each direction has its own queue. The queue to the controller carries "about to start", "started" and progress, so the controller receives them in the order the runner put -them. The queue to the runner carries the start decision and stop requests. +them. The queue to the runner carries the start decision, stop requests and aborts. """ import queue +import threading from enum import Enum from enum import auto +from typing import Optional from ewokscore.progress import BasePercentageProgress @@ -38,14 +40,25 @@ class ToRunner(Enum): REQUEST_STOP = auto() """Request the running task to stop.""" + ABORT = auto() + """Interrupt the running task.""" + END = auto() """End the reading of the queue. Put by the runner itself.""" +_STOP_SENDING = object() + +_MAX_PENDING = 100 +# The task waits for the sender thread when it reports progress faster than it is sent. + + class QueueProgress(BasePercentageProgress): """Picklable progress that forwards updates to the controller. Shares the queue to the controller with the :class:`ToController` messages. + Between :meth:`start_sending` and :meth:`stop_sending`, a sender thread forwards + the updates. """ _TRANSIENT_ERRORS = (EOFError, BrokenPipeError, ConnectionError) @@ -57,10 +70,48 @@ def __init__(self, to_controller_queue: queue.Queue): """ super().__init__() self._progress_queue = to_controller_queue + self._pending: Optional[queue.Queue] = None + self._sender: Optional[threading.Thread] = None + self._discard = False + + def start_sending(self) -> None: + """Start the thread that sends the progress values to the controller.""" + self._pending = queue.Queue(maxsize=_MAX_PENDING) + self._sender = threading.Thread( + target=self._send, args=(self._pending,), daemon=True + ) + self._sender.start() + + def stop_sending(self, discard: bool = False) -> None: + """Send the remaining progress values, or drop them, and stop the sender thread.""" + if self._sender is None: + return + self._discard = discard + self._pending.put(_STOP_SENDING) + self._sender.join() + self._pending = None + self._sender = None def _update(self) -> None: - """Send the current progress to the controller.""" + """Send the current progress, through the sender thread when it runs.""" + if self._pending is None: + self._put(self._progress) + else: + # Not sent from here: an abort interrupts the task thread, which would + # leave the manager connection of that thread out of sync. + self._pending.put(self._progress) + + def _send(self, pending: queue.Queue) -> None: + """Send the queued progress values until :meth:`stop_sending`.""" + while True: + value = pending.get() + if value is _STOP_SENDING: + return + if not self._discard: + self._put(value) + + def _put(self, value: float) -> None: try: - self._progress_queue.put(self._progress) + self._progress_queue.put(value) except self._TRANSIENT_ERRORS: pass diff --git a/src/ewoksorange/gui/concurrency/_runners/abstract.py b/src/ewoksorange/gui/concurrency/_runners/abstract.py index e1973785..9fbaa583 100644 --- a/src/ewoksorange/gui/concurrency/_runners/abstract.py +++ b/src/ewoksorange/gui/concurrency/_runners/abstract.py @@ -10,6 +10,7 @@ from ewokscore.variable import VariableContainer from .._controllers.abstract import StartDenial +from ..future import TaskAbortedError from ..future import TaskStoppedError @@ -91,7 +92,7 @@ def _watch_stop_request(): watcher = threading.Thread(target=_watch_stop_request, daemon=True) watcher.start() try: - task.execute(raise_on_error=True) + self._run_task(task) finally: done.set() # Wake the watcher if the task completed without stop request. @@ -100,10 +101,17 @@ def _watch_stop_request(): return task.output_variables + def _run_task(self, task: Task) -> None: + task.execute(raise_on_error=True) + def _denial_error(task_class: Type[Task], denial: StartDenial) -> Exception: """The error of a task that was not allowed to start.""" name = task_class.__name__ if denial is StartDenial.STOPPED: return TaskStoppedError(f"Task {name} was stopped before it started") + if denial is StartDenial.ABORTED: + return TaskAbortedError(f"Task {name} was aborted before it started") + if denial is StartDenial.KILLED: + return TaskAbortedError(f"Task {name} was killed before it started") raise ValueError(f"Start denial {denial} is not handled") diff --git a/src/ewoksorange/gui/concurrency/_runners/process.py b/src/ewoksorange/gui/concurrency/_runners/process.py index 1d69dcf2..a6b82827 100644 --- a/src/ewoksorange/gui/concurrency/_runners/process.py +++ b/src/ewoksorange/gui/concurrency/_runners/process.py @@ -1,4 +1,7 @@ +import _thread import queue +import signal +import sys import threading from typing import Any from typing import Dict @@ -9,8 +12,10 @@ from ewokscore.variable import VariableContainer from .._controllers.abstract import StartDenial +from .._messages import QueueProgress from .._messages import ToController from .._messages import ToRunner +from ..future import TaskAbortedError from .abstract import TaskRunner @@ -30,6 +35,7 @@ def __init__( super().__init__(task_class, task_kwargs) self._to_controller_queue = to_controller_queue self._to_runner_queue = to_runner_queue + self._abort_sent: Optional[threading.Event] = None def _try_start(self) -> Optional[StartDenial]: # The controller decides, so that a denied task never starts. @@ -49,8 +55,9 @@ def _announce_started(self, task: Task) -> None: self._to_controller_queue.put(ToController.STARTED) def _execute(self, task: Task) -> VariableContainer: - """Execute `task`, handling the stop requests of the controller.""" + """Execute `task`, handling the stop and abort requests of the controller.""" stop_event = threading.Event() + abort_sent = self._abort_sent = threading.Event() def _read_to_runner_queue(): try: @@ -58,19 +65,70 @@ def _read_to_runner_queue(): message = self._to_runner_queue.get() if message is ToRunner.REQUEST_STOP: stop_event.set() + elif message is ToRunner.ABORT: + if not abort_sent.is_set(): + abort_sent.set() + _interrupt_main_thread() elif message is ToRunner.END: return except self._TRANSIENT_ERRORS: pass + progress = self._task_kwargs.get("progress") + if not isinstance(progress, QueueProgress): + progress = None + if progress is not None: + progress.start_sending() reader = threading.Thread(target=_read_to_runner_queue, daemon=True) reader.start() + aborted = False try: return self._execute_with_stop_event(task, stop_event) + except KeyboardInterrupt: + if abort_sent.is_set(): + aborted = True + raise TaskAbortedError( + f"Task {type(task).__name__} was aborted" + ) from None + raise finally: + if progress is not None: + # Before returning: the controller expects all progress values first. + # Those of an aborted task do not matter. + progress.stop_sending(discard=aborted) try: # Wake the reader. self._to_runner_queue.put(ToRunner.END) except self._TRANSIENT_ERRORS: pass reader.join() + + def _run_task(self, task: Task) -> None: + # SIGINT only interrupts the task, not the runner. Outside the task, the + # pebble worker ignores SIGINT. + previous = signal.signal(signal.SIGINT, _raise_keyboard_interrupt) + try: + try: + if self._abort_sent is not None and self._abort_sent.is_set(): + # The SIGINT arrived before the handler was installed and was ignored. + raise KeyboardInterrupt + super()._run_task(task) + finally: + signal.signal(signal.SIGINT, previous) + except KeyboardInterrupt: + # The SIGINT can also interrupt the restore above. + signal.signal(signal.SIGINT, previous) + raise + + +def _raise_keyboard_interrupt(signum: int, frame: Any) -> None: + """Interrupt the task like Ctrl-C.""" + raise KeyboardInterrupt + + +def _interrupt_main_thread() -> None: + if sys.platform == "win32": + # Unlike a real signal, this does not interrupt blocking calls. + _thread.interrupt_main() + else: + signal.pthread_kill(threading.main_thread().ident, signal.SIGINT) diff --git a/src/ewoksorange/gui/concurrency/executor.py b/src/ewoksorange/gui/concurrency/executor.py index 2500b1d7..6d1fa4e1 100644 --- a/src/ewoksorange/gui/concurrency/executor.py +++ b/src/ewoksorange/gui/concurrency/executor.py @@ -17,7 +17,9 @@ from typing import List from typing import Optional from typing import Type +from typing import Union +import pebble from AnyQt.QtCore import QObject from AnyQt.QtCore import Qt from AnyQt.QtCore import Signal @@ -55,8 +57,8 @@ def create_pool_executor( concurrency: Concurrency, max_workers: Optional[int] = None, mp_context: Optional[multiprocessing.context.BaseContext] = None, -) -> Optional[futures.Executor]: - """Create the `concurrent.futures` executor for a given concurrency. +) -> Optional[Union[futures.Executor, pebble.ProcessPool]]: + """Create the pool executor for a given concurrency. :param concurrency: The execution backend. :param max_workers: Maximum number of workers, `None` for the pool default. @@ -67,9 +69,10 @@ def create_pool_executor( if concurrency is Concurrency.SYNC: return None if concurrency is Concurrency.PROCESS: - return futures.ProcessPoolExecutor( - max_workers=max_workers, mp_context=mp_context - ) + kwargs = {"context": mp_context or multiprocessing} + if max_workers is not None: + kwargs["max_workers"] = max_workers + return pebble.ProcessPool(**kwargs) if concurrency is Concurrency.THREAD: return futures.ThreadPoolExecutor(max_workers=max_workers) raise ValueError(f"Concurrency policy {concurrency} is not handled") @@ -78,9 +81,9 @@ def create_pool_executor( class EwoksExecutor(QObject): """Qt-aware executor for ewoks tasks. - Wraps a `concurrent.futures.Executor` (thread, process or None) and emits Qt - signals on task lifecycle events. Returns :class:`TaskFuture` objects to cancel - or stop the task. Their result is the executed task's + Wraps a thread pool (`concurrent.futures.Executor`), a process pool (`pebble.ProcessPool`) + or None and emits Qt signals on task lifecycle events. Returns :class:`TaskFuture` objects + to cancel, stop, abort or kill the task. Their result is the executed task's :class:`~ewokscore.variable.VariableContainer` of output variables. """ @@ -98,7 +101,8 @@ class EwoksExecutor(QObject): """Emitted when a task finishes successfully.""" failed = Signal(TaskFuture) - """Emitted when a task raises an exception, also when a stop prevented its start.""" + """Emitted when a task raises an exception, also when a stop, abort or kill prevented + its start or a kill ended it.""" ignored = Signal() """Emitted when a task submission is ignored due to the DROP_IF_BUSY policy.""" @@ -109,6 +113,12 @@ class EwoksExecutor(QObject): stopRequested = Signal(TaskFuture) """Emitted when a task finishes after :meth:`TaskFuture.request_stop` prevented its start or reached it.""" + aborted = Signal(TaskFuture) + """Emitted when a task finishes after :meth:`TaskFuture.abort` prevented its start or interrupted it.""" + + killed = Signal(TaskFuture) + """Emitted when a task finishes after :meth:`TaskFuture.kill` prevented its start or terminated it.""" + finished = Signal(TaskFuture) """Emitted last, for every submitted task.""" @@ -118,12 +128,13 @@ class EwoksExecutor(QObject): def __init__( self, - executor: Optional[futures.Executor], + executor: Optional[Union[futures.Executor, pebble.ProcessPool]], policy: SubmitPolicy = SubmitPolicy.ALWAYS, mp_context: Optional[multiprocessing.context.BaseContext] = None, ): """ :param executor: Wrapped executor, `None` to execute in the calling thread. + Use `pebble.ProcessPool` to execute in subprocesses. :param policy: What to do with submissions while the executor is busy. :param mp_context: Multiprocessing context used for the IPC manager. Provide the context of `executor` so the manager process is created @@ -131,13 +142,16 @@ def __init__( """ super().__init__() + if isinstance(executor, futures.ProcessPoolExecutor): + raise TypeError("Use 'pebble.ProcessPool' to execute in subprocesses") + self._executor = executor self._policy = policy self._mp_context = mp_context self._running = 0 self._lock = Lock() - self._as_process = isinstance(executor, futures.ProcessPoolExecutor) + self._as_process = isinstance(executor, pebble.ProcessPool) self._done.connect(self._handle_done, Qt.ConnectionType.QueuedConnection) @@ -245,7 +259,7 @@ def on_progress(value: float) -> None: self_ref = weakref.ref(self) holder: List[Optional[TaskFuture]] = [None] - raw_future = self._executor.submit(runner) + raw_future = self._executor.submit(runner, None) def watch() -> None: # Only after the submission (a failing 'submit' is never watched) and @@ -299,7 +313,15 @@ def _finalize_submission( self.submitted.emit(task_future) - after_submitted() + try: + after_submitted() + except BaseException: + # Otherwise a process runner waits forever for the start decision. + controller.cancel(raw_future) + raw_future.cancel() + # Without a done callback: `submit_task` releases the executor. + self._handle_done(raw_future, task_future, controller) + raise queue_done = self._executor is not None raw_future.add_done_callback( @@ -329,6 +351,10 @@ def _handle_done( if task_future.stop_requested(): self.stopRequested.emit(task_future) + if task_future.aborted(): + self.aborted.emit(task_future) + if task_future.killed(): + self.killed.emit(task_future) try: task_future.result() @@ -344,7 +370,10 @@ def shutdown(self, wait: bool = False) -> None: manager, self._manager = self._manager, None def _shutdown() -> None: - if executor is not None: + if isinstance(executor, pebble.ProcessPool): + executor.close() + executor.join() + elif executor is not None: executor.shutdown(wait=True) if manager is not None: manager.shutdown() @@ -354,8 +383,8 @@ def _shutdown() -> None: return # Wait off-thread rather than `executor.shutdown(wait=False)`: on - # Python < 3.9, garbage-collecting a `ProcessPoolExecutor` mid-shutdown - # can orphan a worker process, hanging the interpreter at exit. + # Python < 3.9, garbage-collecting a pool mid-shutdown can orphan a + # worker process, hanging the interpreter at exit. Thread(target=_shutdown, name="EwoksExecutorShutdown").start() @@ -380,9 +409,9 @@ def _done_callback( if queue_done: # The other signals of the task are queued for the thread of the - # executor, also when this callback runs in that thread (a task that - # finished before the callback was added). The completion signals must - # not overtake them. + # executor, also when this callback runs in that thread (a kill, or a + # task that finished before the callback was added). The completion + # signals must not overtake them. executor._done.emit(raw_future, task_future, controller) else: executor._handle_done(raw_future, task_future, controller) @@ -411,7 +440,7 @@ def _sync_submit( raw_future: futures.Future[VariableContainer], ) -> None: """Run `fn` and store its outcome in `raw_future`.""" - # `ThreadPoolExecutor`/`ProcessPoolExecutor` already do this internally + # `ThreadPoolExecutor`/`pebble.ProcessPool` already do this internally # (via `set_running_or_notify_cancel()`) before invoking a submitted # callable, honoring a `cancel()` that raced with the start of execution # instead of silently overwriting it. The sync case has no real diff --git a/src/ewoksorange/gui/concurrency/future.py b/src/ewoksorange/gui/concurrency/future.py index 131aa71d..f985f1c3 100644 --- a/src/ewoksorange/gui/concurrency/future.py +++ b/src/ewoksorange/gui/concurrency/future.py @@ -17,19 +17,29 @@ class TaskStoppedError(RuntimeError): """Raised by :meth:`TaskFuture.result` when a stop request prevented the task from starting.""" +class TaskAbortedError(RuntimeError): + """Raised by :meth:`TaskFuture.result` when the task was aborted or killed.""" + + class TaskFuture: """Future of an ewoks task submitted to an `EwoksExecutor`. - Ways to prevent or end the execution: + Ways to prevent or end the execution, from mild to forceful: 1. :meth:`cancel` (do not run): has no effect on a running task. 2. :meth:`request_stop` (cooperative): the running task is asked to stop and decides whether and how (see `Task.stop_requested` and `Task._on_stop_requested`). + 3. :meth:`abort` (forced, with cleanup): the running task is interrupted like Ctrl-C, + also when it ignores stop requests. Subprocess only. + 4. :meth:`kill` (forced, no cleanup): the subprocess running the task is terminated, + also when the task does not return to the Python interpreter. Subprocess only. - Each of them prevents a queued task from starting. Only the first one that reaches - a queued task has an effect. None of them has an effect on a finished task. + Each of them that is supported prevents a queued task from starting. Only the first + one that reaches a queued task has an effect. None of them has an effect on a + finished task. - :meth:`cancelled` and :meth:`stop_requested` report which of these happened. + :meth:`cancelled`, :meth:`stop_requested`, :meth:`aborted` and :meth:`killed` + report which of these happened. """ def __init__( @@ -66,6 +76,41 @@ def stop_requested(self) -> bool: """Return True if :meth:`request_stop` prevented the start or reached the ewoks task.""" return self._controller.stop_requested() + def abort(self) -> bool: + """Interrupt the ewoks task like Ctrl-C (forced, with cleanup). + + Only supported for tasks running in a subprocess. A queued task never starts. + A running task is interrupted, also when it ignores stop requests. The task can + clean up and the subprocess survives. It only takes effect when the task returns + to the Python interpreter: use :meth:`kill` otherwise. When the task is + interrupted, the result raises :class:`TaskAbortedError`. + + :return: True if this prevented the task from starting or was sent to the running task. + """ + if self.done(): + return False + return self._controller.abort() + + def kill(self) -> bool: + """Terminate the subprocess running the ewoks task (forced, no cleanup). + + Only supported for tasks running in a subprocess. A queued task never starts and + its future completes right away. Unlike :meth:`abort`, this also works when a + running task does not return to the Python interpreter. The task cannot clean up + and the pool replaces the subprocess. The result raises :class:`TaskAbortedError`. + + :return: True if this prevented the task from starting or killed the running task. + """ + return self._controller.kill(self._future) + + def aborted(self) -> bool: + """Return True if :meth:`abort` prevented the start or interrupted the ewoks task.""" + return self._controller.aborted(self._future) + + def killed(self) -> bool: + """Return True if :meth:`kill` prevented the start or terminated the ewoks task.""" + return self._controller.killed() + def cancelled(self) -> bool: """Return True if :meth:`cancel` prevented the execution.""" return self._controller.cancelled(self._future) @@ -83,11 +128,17 @@ def result(self, timeout: Optional[float] = None) -> VariableContainer: :raises TimeoutError: The task did not finish in time. :raises CancelledError: The task was cancelled before it started. :raises TaskStoppedError: A stop request prevented the task from starting. + :raises TaskAbortedError: The task was aborted or killed. :raises Exception: Whatever the task raised. :return: An immutable mapping of output name to :class:`~ewokscore.variable.Variable`. """ - result = self._future.result(timeout=timeout) + try: + result = self._future.result(timeout=timeout) + except CancelledError: + if self._controller.killed(): + raise self._killed_error() from None + raise if self.cancelled(): raise CancelledError() return result @@ -128,7 +179,12 @@ def exception(self, timeout: Optional[float] = None) -> Optional[BaseException]: :raises CancelledError: The task was cancelled before it started. :return: The exception or `None` when the task succeeded. """ - exception = self._future.exception(timeout=timeout) + try: + exception = self._future.exception(timeout=timeout) + except CancelledError: + if self._controller.killed(): + return self._killed_error() + raise if exception is None and self.cancelled(): raise CancelledError() return exception @@ -156,3 +212,9 @@ def add_done_callback(self, fn: Callable[[Future[VariableContainer]], Any]) -> N :param fn: Called with the wrapped future, not with this object. """ self._future.add_done_callback(fn) + + def _killed_error(self) -> TaskAbortedError: + """The error of a killed task.""" + if self._controller.started(): + return TaskAbortedError("The task was killed") + return TaskAbortedError("The task was killed before it started") diff --git a/src/ewoksorange/gui/owwidgets/base.py b/src/ewoksorange/gui/owwidgets/base.py index 6a0f7490..346bf69f 100644 --- a/src/ewoksorange/gui/owwidgets/base.py +++ b/src/ewoksorange/gui/owwidgets/base.py @@ -15,7 +15,9 @@ from typing import Mapping from typing import Optional from typing import Set +from typing import Union +import pebble from AnyQt import QtWidgets from ewokscore import TaskWithProgress from ewokscore import missing_data @@ -182,9 +184,12 @@ def __init__(self, *args, **kwargs): self.__executor.finished.connect(self.__on_finished) @classmethod - def _create_pool_executor(cls) -> Optional[futures.Executor]: + def _create_pool_executor( + cls, + ) -> Optional[Union[futures.Executor, pebble.ProcessPool]]: """ - Create the `concurrent.futures` executor that runs the Ewoks tasks. + Create the executor that runs the Ewoks tasks: a `concurrent.futures` + executor, or a `pebble.ProcessPool` to execute in subprocesses. Override to execute tasks in an executor that :class:`Concurrency` does not cover. diff --git a/src/ewoksorange/tests/executor/signals.py b/src/ewoksorange/tests/executor/signals.py index 7fcc0af9..74fdbbb4 100644 --- a/src/ewoksorange/tests/executor/signals.py +++ b/src/ewoksorange/tests/executor/signals.py @@ -27,6 +27,8 @@ def __init__(self): "failed": 0, "cancelled": 0, "stopRequested": 0, + "aborted": 0, + "killed": 0, "finished": 0, "ignored": 0, } @@ -39,6 +41,8 @@ def connect(self, executor: EwoksExecutor) -> None: executor.failed.connect(self._store_future("failed")) executor.cancelled.connect(self._store_future("cancelled")) executor.stopRequested.connect(self._store_future("stopRequested")) + executor.aborted.connect(self._store_future("aborted")) + executor.killed.connect(self._store_future("killed")) executor.finished.connect(self._store_future("finished")) executor.ignored.connect(self._store_noargs("ignored")) diff --git a/src/ewoksorange/tests/executor/tasks.py b/src/ewoksorange/tests/executor/tasks.py index 60a7a305..335abafb 100644 --- a/src/ewoksorange/tests/executor/tasks.py +++ b/src/ewoksorange/tests/executor/tasks.py @@ -106,6 +106,24 @@ def _on_stop_requested(self): pass +class CleanupTask( + Task, + input_names=["duration", "marker"], + output_names=["result"], +): + """Never checks `self.stop_requested` and creates the file `marker` however it exits.""" + + def run(self): + try: + end = time.monotonic() + self.inputs.duration + while time.monotonic() < end: + time.sleep(0.01) + self.outputs.result = "completed" + finally: + with open(self.inputs.marker, "w"): + pass + + class WaitForFileTask( Task, input_names=["path"], @@ -126,6 +144,65 @@ def _on_stop_requested(self): pass +class StubbornTask( + Task, + input_names=["duration", "marker"], + output_names=["result"], +): + """Never checks `self.stop_requested` and survives an abort: creates the file + `marker` when interrupted, then keeps running.""" + + def run(self): + end = time.monotonic() + self.inputs.duration + while True: + try: + while time.monotonic() < end: + time.sleep(0.01) + break + except KeyboardInterrupt: + with open(self.inputs.marker, "w"): + pass + self.outputs.result = "completed" + + def _on_stop_requested(self): + """Nothing to do: the stop request is ignored.""" + pass + + +class StartMarkerTask( + Task, + input_names=["duration", "marker"], + output_names=["result"], +): + """Creates the file `marker` when it starts, then never checks `self.stop_requested`.""" + + def run(self): + with open(self.inputs.marker, "w"): + pass + end = time.monotonic() + self.inputs.duration + while time.monotonic() < end: + time.sleep(0.01) + self.outputs.result = "completed" + + +class SlowInitTask( + Task, + input_names=["init_duration", "duration"], + output_names=["result"], +): + """Takes `init_duration` seconds to be created, then never checks `self.stop_requested`.""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + time.sleep(self.inputs.init_duration) + + def run(self): + end = time.monotonic() + self.inputs.duration + while time.monotonic() < end: + time.sleep(0.01) + self.outputs.result = "completed" + + class InterruptStopTask( Task, input_names=["duration"], @@ -183,6 +260,77 @@ def run(self) -> None: self.outputs.pid = os.getpid() +class ProgressLoopTask( + TaskWithProgress, + input_names=["duration"], + output_names=["result"], +): + """Reports progress continuously for `duration` seconds.""" + + def run(self) -> None: + end = time.monotonic() + self.inputs.duration + i = 0 + while time.monotonic() < end: + i += 1 + self.progress = i % 100 + time.sleep(0.0001) + self.outputs.result = "completed" + + +class ProgressBurstTask( + TaskWithProgress, + input_names=["duration"], + output_names=["end"], +): + """Reports progress as fast as it can for `duration` seconds. Outputs the + wall-clock time it ended.""" + + def run(self) -> None: + end = time.monotonic() + self.inputs.duration + i = 0 + while time.monotonic() < end: + i += 1 + self.progress = i % 100 + self.outputs.end = time.time() + + +class SurviveAbortProgressTask( + TaskWithProgress, + input_names=["timeout"], + output_names=["result"], +): + """Waits for an abort, survives it and reports up to 100% before it completes.""" + + def run(self) -> None: + end = time.monotonic() + self.inputs.timeout + try: + while time.monotonic() < end: + time.sleep(0.01) + except KeyboardInterrupt: + pass + else: + raise TimeoutError("not aborted") + for percentage in range(1, 101): + self.progress = percentage + self.outputs.result = "completed" + + +class InitProgressTask( + TaskWithProgress, + input_names=["value"], + output_names=["value"], +): + """Reports 50% when it is created and 100% when it runs.""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.progress = 50 + + def run(self) -> None: + self.progress = 100 + self.outputs.value = self.inputs.value + + class PidTask(Task, input_names=["value"], output_names=["value", "pid"]): """Outputs the pid it ran in. Not a `TaskWithProgress`.""" diff --git a/src/ewoksorange/tests/executor/test_executor_abort.py b/src/ewoksorange/tests/executor/test_executor_abort.py new file mode 100644 index 00000000..6f91b882 --- /dev/null +++ b/src/ewoksorange/tests/executor/test_executor_abort.py @@ -0,0 +1,304 @@ +import os +import signal +import sys +import threading +import time + +import pytest +from AnyQt.QtCore import Qt + +from ...gui.concurrency.future import TaskAbortedError +from ...gui.qt_utils.app import wait_until +from ...gui.qt_utils.progress import QProgress +from .tasks import CleanupTask +from .tasks import IgnoreStopTask +from .tasks import PidTask +from .tasks import ProgressLoopTask +from .tasks import SlowInitTask +from .tasks import StartMarkerTask +from .tasks import StubbornTask +from .tasks import SurviveAbortProgressTask + + +def test_abort(ewoksorange_qtapp, executor_context_factory, tmp_path): + """Abort interrupts a task running in a subprocess, even when it ignores stop + requests. The task cleans up and the worker remains usable.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("abort is only supported for tasks running in a subprocess") + + pid = ( + executor.submit_task(PidTask, inputs={"value": 1}) + .result(timeout=10)["pid"] + .value + ) + + marker = tmp_path / "cleanup" + executor.submit_task( + CleanupTask, inputs={"duration": 10, "marker": str(marker)} + ) + future = recorder.wait_future("started", index=1) + + t0 = time.monotonic() + assert future.abort() + + with pytest.raises(TaskAbortedError, match="was aborted"): + future.result(timeout=5) + assert time.monotonic() - t0 < 5 + + assert future.aborted() + assert not future.killed() + assert not future.stop_requested() + assert not future.cancelled() + assert marker.exists() + + recorder.wait_for("finished", 2) + recorder.assert_counts( + submitted=2, + aboutToStart=2, + started=2, + succeeded=1, + failed=1, + aborted=1, + finished=2, + ) + + outputs = executor.submit_task(PidTask, inputs={"value": 2}).result(timeout=10) + assert outputs["value"].value == 2 + assert outputs["pid"].value == pid + + +def test_abort_during_creation(ewoksorange_qtapp, executor_context_factory): + """Abort interrupts a task that is still being created when the abort arrives.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("abort is only supported for tasks running in a subprocess") + + executor.submit_task(SlowInitTask, inputs={"init_duration": 1, "duration": 10}) + future = recorder.wait_future("aboutToStart") + + assert future.abort() + + with pytest.raises(TaskAbortedError, match="was aborted"): + future.result(timeout=5) + assert future.aborted() + + +def test_abort_during_progress(ewoksorange_qtapp, executor_context_factory, tmp_path): + """Abort while the task reports progress leaves the worker usable: a task + cancelled while queued for that worker never starts.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("abort is only supported for tasks running in a subprocess") + + progress = QProgress() + future = executor.submit_task( + ProgressLoopTask, inputs={"duration": 30}, progress=progress + ) + assert wait_until(lambda: progress.progress > 0, timeout=10) + assert future.abort() + assert isinstance(future.exception(timeout=10), TaskAbortedError) + + blocker = executor.submit_task( + StartMarkerTask, inputs={"duration": 1, "marker": str(tmp_path / "blocker")} + ) + marker = tmp_path / "queued" + queued = executor.submit_task( + StartMarkerTask, inputs={"duration": 0, "marker": str(marker)} + ) + assert queued.cancel() + + assert blocker.result(timeout=10)["result"].value == "completed" + recorder.wait_for("finished", 3) + assert queued.cancelled() + assert not marker.exists() + + +def test_abort_survived_keeps_progress(ewoksorange_qtapp, executor_context_factory): + """A task that survives an abort and completes keeps its progress.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("abort is only supported for tasks running in a subprocess") + + progress = QProgress() + executor.submit_task( + SurviveAbortProgressTask, inputs={"timeout": 30}, progress=progress + ) + future = recorder.wait_future("started") + + assert future.abort() + + assert future.result(timeout=10)["result"].value == "completed" + assert not future.aborted() + recorder.wait_for("finished", 1) + assert progress.progress == 100 + recorder.assert_counts( + submitted=1, aboutToStart=1, started=1, succeeded=1, finished=1 + ) + + +def test_kill(ewoksorange_qtapp, executor_context_factory, tmp_path): + """Kill terminates the subprocess running a task. The task does not clean up + and the pool replaces the worker.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("kill is only supported for tasks running in a subprocess") + + pid = ( + executor.submit_task(PidTask, inputs={"value": 1}) + .result(timeout=10)["pid"] + .value + ) + + marker = tmp_path / "cleanup" + executor.submit_task( + CleanupTask, inputs={"duration": 10, "marker": str(marker)} + ) + future = recorder.wait_future("started", index=1) + + t0 = time.monotonic() + assert future.kill() + + with pytest.raises(TaskAbortedError, match="was killed"): + future.result(timeout=5) + assert time.monotonic() - t0 < 5 + assert isinstance(future.exception(), TaskAbortedError) + + assert future.killed() + assert not future.aborted() + assert not future.stop_requested() + assert not future.cancelled() + + recorder.wait_for("finished", 2) + recorder.assert_counts( + submitted=2, + aboutToStart=2, + started=2, + succeeded=1, + failed=1, + killed=1, + finished=2, + ) + + outputs = executor.submit_task(PidTask, inputs={"value": 2}).result(timeout=10) + assert outputs["value"].value == 2 + assert outputs["pid"].value != pid + assert not marker.exists() + + +def test_escalate(ewoksorange_qtapp, executor_context_factory, tmp_path): + """A running task that ignores a stop request and survives an abort is killed.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("abort and kill are only supported in a subprocess") + + marker = tmp_path / "interrupted" + executor.submit_task( + StubbornTask, inputs={"duration": 30, "marker": str(marker)} + ) + future = recorder.wait_future("started") + + assert future.request_stop() + assert future.abort() + t0 = time.monotonic() + while not marker.exists(): + assert time.monotonic() - t0 < 5, "the abort did not reach the task" + time.sleep(0.01) + assert not future.done() + + assert future.kill() + with pytest.raises(TaskAbortedError, match="was killed"): + future.result(timeout=5) + assert future.stop_requested() + assert not future.aborted() + assert future.killed() + + recorder.wait_for("finished", 1) + recorder.assert_counts( + submitted=1, + aboutToStart=1, + started=1, + stopRequested=1, + killed=1, + failed=1, + finished=1, + ) + + +def test_kill_from_relay_thread(ewoksorange_qtapp, executor_context_factory): + """Kill from a slot that runs in the relay thread ends the execution.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("kill is only supported for tasks running in a subprocess") + + executor.started.connect( + lambda future: future.kill(), Qt.ConnectionType.DirectConnection + ) + future = executor.submit_task(IgnoreStopTask, inputs={"duration": 10}) + + with pytest.raises(TaskAbortedError, match="was killed"): + future.result(timeout=5) + + recorder.wait_for("finished", 1) + recorder.assert_counts( + submitted=1, aboutToStart=1, started=1, failed=1, killed=1, finished=1 + ) + + +@pytest.mark.skipif(sys.platform == "win32", reason="requires POSIX signals") +def test_abort_signal_ignored_by_idle_worker( + ewoksorange_qtapp, executor_context_factory +): + """SIGINT does not kill an idle worker, which would break the pool.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("abort is only supported for tasks running in a subprocess") + + pid = ( + executor.submit_task(PidTask, inputs={"value": 1}) + .result(timeout=10)["pid"] + .value + ) + + os.kill(pid, signal.SIGINT) + time.sleep(0.5) + + outputs = executor.submit_task(PidTask, inputs={"value": 2}).result(timeout=10) + assert outputs["value"].value == 2 + assert outputs["pid"].value == pid + + +def test_abort_not_supported(ewoksorange_qtapp, executor_context_factory): + """Abort and kill have no effect on a task that does not run in a subprocess.""" + with executor_context_factory() as (kind, executor, recorder): + if kind == "process": + pytest.skip("abort is supported for tasks running in a subprocess") + + inputs = {"duration": 0.3} + thread = None + if kind == "sync": + thread = threading.Thread( + target=executor.submit_task, + args=(IgnoreStopTask,), + kwargs={"inputs": inputs}, + ) + thread.start() + else: + executor.submit_task(IgnoreStopTask, inputs=inputs) + + future = recorder.wait_future("started") + + assert not future.abort() + assert not future.kill() + + result = future.result(timeout=10) + assert result["result"].value == "completed despite stop request" + assert not future.aborted() + assert not future.killed() + + recorder.wait_for("finished", 1) + recorder.assert_counts( + submitted=1, aboutToStart=1, started=1, succeeded=1, finished=1 + ) + if thread is not None: + thread.join(timeout=10) diff --git a/src/ewoksorange/tests/executor/test_executor_actions.py b/src/ewoksorange/tests/executor/test_executor_actions.py index d7d87543..aac118a2 100644 --- a/src/ewoksorange/tests/executor/test_executor_actions.py +++ b/src/ewoksorange/tests/executor/test_executor_actions.py @@ -3,20 +3,28 @@ import pytest +from ...gui.concurrency._controllers.process import ProcessTaskController +from ...gui.concurrency.future import TaskAbortedError from ...gui.concurrency.future import TaskFuture from ...gui.concurrency.future import TaskStoppedError from .tasks import AddTask -_ACTIONS = ("cancel", "request_stop") +_ACTIONS = ("cancel", "request_stop", "abort", "kill") + +_SUBPROCESS_ONLY = ("abort", "kill") _ERRORS = { "cancel": (CancelledError, None), "request_stop": (TaskStoppedError, "stopped before it started"), + "abort": (TaskAbortedError, "aborted before it started"), + "kill": (TaskAbortedError, "killed before it started"), } _SIGNALS = { "cancel": {"cancelled": 1, "finished": 1}, "request_stop": {"stopRequested": 1, "failed": 1, "finished": 1}, + "abort": {"aborted": 1, "failed": 1, "finished": 1}, + "kill": {"killed": 1, "failed": 1, "finished": 1}, } @@ -31,10 +39,20 @@ def test_queued_task_never_starts(ewoksorange_qtapp, executor_context_factory, a recorder.wait_for("started") queued = executor.submit_task(AddTask, inputs={"a": 2}) - assert getattr(queued, action)() + supported = kind == "process" or action not in _SUBPROCESS_ONLY + assert getattr(queued, action)() is supported assert blocker.result(timeout=10)["result"].value == 1 + if not supported: + assert queued.result(timeout=10)["result"].value == 2 + assert _flags(queued) == dict.fromkeys(_ACTIONS, False) + recorder.wait_for("finished", 2) + recorder.assert_counts( + submitted=2, aboutToStart=2, started=2, succeeded=2, finished=2 + ) + return + exc_type, match = _ERRORS[action] with pytest.raises(exc_type, match=match): queued.result(timeout=10) @@ -62,8 +80,8 @@ def test_queued_task_first_action_wins( ): """Only the first action that reaches a queued task has an effect.""" with executor_context_factory(workers=1) as (kind, executor, recorder): - if kind == "sync": - pytest.skip("sync execution has no queue") + if kind != "process": + pytest.skip("abort and kill are only supported in a subprocess") blocker = executor.submit_task(AddTask, inputs={"a": 1, "delay": 1}) recorder.wait_for("started") @@ -80,6 +98,52 @@ def test_queued_task_first_action_wins( assert _flags(queued) == {name: name == first for name in _ACTIONS} +def test_killed_queued_task_completes(ewoksorange_qtapp, executor_context_factory): + """Kill completes the future of a queued task right away.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("kill is only supported in a subprocess") + + blocker = executor.submit_task(AddTask, inputs={"a": 1, "delay": 2}) + recorder.wait_for("started") + queued = executor.submit_task(AddTask, inputs={"a": 2}) + + assert queued.kill() + assert queued.done() + assert not blocker.done() + with pytest.raises(TaskAbortedError, match="killed before it started"): + queued.result(timeout=0) + + recorder.wait_for("finished", 1) + assert recorder.future("finished") is queued + assert blocker.result(timeout=10)["result"].value == 1 + + +def test_failed_watch_ends_execution( + ewoksorange_qtapp, executor_context_factory, monkeypatch +): + """When the relay fails to start, the submission raises, the execution ends + with `cancelled` and `finished`, and the executor stays usable.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("only the process backend watches the task") + + def _failing_watch(*args, **kwargs) -> None: + raise RuntimeError("cannot start the relay") + + # Last resort: starting a thread cannot be made to fail otherwise. + with monkeypatch.context() as patch: + patch.setattr(ProcessTaskController, "watch", _failing_watch) + with pytest.raises(RuntimeError, match="cannot start the relay"): + executor.submit_task(AddTask, inputs={"a": 1}) + + recorder.assert_counts(submitted=1, cancelled=1, finished=1) + assert recorder.future("finished").cancelled() + + future = executor.submit_task(AddTask, inputs={"a": 2}) + assert future.result(timeout=30)["result"].value == 2 + + def test_finished_task(ewoksorange_qtapp, executor_context_factory): """No action has an effect on a finished task.""" with executor_context_factory() as (kind, executor, recorder): @@ -103,4 +167,6 @@ def _flags(future: TaskFuture) -> Dict[str, bool]: return { "cancel": future.cancelled(), "request_stop": future.stop_requested(), + "abort": future.aborted(), + "kill": future.killed(), } diff --git a/src/ewoksorange/tests/executor/test_executor_order.py b/src/ewoksorange/tests/executor/test_executor_order.py index 501c6120..0c1eff34 100644 --- a/src/ewoksorange/tests/executor/test_executor_order.py +++ b/src/ewoksorange/tests/executor/test_executor_order.py @@ -9,6 +9,7 @@ from ...gui.concurrency.executor import create_pool_executor from .signals import SignalRecorder from .tasks import AddTask +from .tasks import StartMarkerTask def test_completion_after_start_while_gui_thread_busy( @@ -30,11 +31,39 @@ def test_completion_after_start_while_gui_thread_busy( ) +def test_kill_completion_after_start( + ewoksorange_qtapp, executor_context_factory, tmp_path +): + """The completion signals of a task killed from the GUI thread follow the + start signals that are still queued for the GUI thread.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind != "process": + pytest.skip("kill is only supported for tasks running in a subprocess") + + marker = tmp_path / "started" + future = executor.submit_task( + StartMarkerTask, inputs={"duration": 10, "marker": str(marker)} + ) + _wait_without_events(marker.exists, timeout=60) + time.sleep(0.5) # let the relay queue "aboutToStart" and "started" + + assert future.kill() + + recorder.wait_for("finished", 1) + recorder.assert_order( + "submitted", "aboutToStart", "started", "killed", "failed", "finished" + ) + + @pytest.mark.parametrize("concurrency", [Concurrency.THREAD, Concurrency.PROCESS]) def test_failed_submission_is_not_running(ewoksorange_qtapp, concurrency): """A submission that raises does not keep the executor busy.""" pool = create_pool_executor(concurrency, max_workers=1) - pool.shutdown() + if concurrency is Concurrency.PROCESS: + pool.close() + pool.join() + else: + pool.shutdown() executor = EwoksExecutor(pool, SubmitPolicy.DROP_IF_BUSY) recorder = SignalRecorder() recorder.connect(executor) diff --git a/src/ewoksorange/tests/executor/test_executor_parallel.py b/src/ewoksorange/tests/executor/test_executor_parallel.py index df047986..fa0b03ce 100644 --- a/src/ewoksorange/tests/executor/test_executor_parallel.py +++ b/src/ewoksorange/tests/executor/test_executor_parallel.py @@ -1,13 +1,17 @@ import threading +import time from itertools import combinations from typing import Tuple +from .tasks import PidTask from .tasks import TimedTask def test_parallel_execution(ewoksorange_qtapp, executor_context_factory) -> None: workers: int = 2 - with executor_context_factory(workers=workers) as (_, executor, recorder): + with executor_context_factory(workers=workers) as (kind, executor, recorder): + nwarmup = _warmup_workers(executor, workers) if kind == "process" else 0 + inputs_list = [{"value": i, "delay": 1} for i in range(4)] futures = [None] * len(inputs_list) @@ -37,12 +41,29 @@ def _submit(index: int, inputs: dict) -> None: overlapping_pairs = sum(_overlaps(a, b) for a, b in combinations(intervals, 2)) assert overlapping_pairs >= workers - recorder.wait_for("finished", 4) + n = 4 + nwarmup + recorder.wait_for("finished", n) recorder.assert_counts( - submitted=4, aboutToStart=4, started=4, succeeded=4, finished=4 + submitted=n, aboutToStart=n, started=n, succeeded=n, finished=n ) +def _warmup_workers(executor, workers: int, timeout: float = 60) -> int: + """Wait until all workers executed a task, since subprocess workers can become + ready at very different times. Returns the number of submitted tasks.""" + pids = set() + nsubmitted = 0 + t0 = time.monotonic() + while len(pids) < workers: + assert time.monotonic() - t0 < timeout, "not all workers became ready" + futures = [ + executor.submit_task(PidTask, inputs={"value": 0}) for _ in range(workers) + ] + nsubmitted += workers + pids.update(future.result(timeout=timeout)["pid"].value for future in futures) + return nsubmitted + + def _overlaps(interval_a: Tuple[float, float], interval_b: Tuple[float, float]) -> bool: """Check whether two (start, end) time intervals overlap.""" start_a, end_a = interval_a diff --git a/src/ewoksorange/tests/executor/test_executor_progress.py b/src/ewoksorange/tests/executor/test_executor_progress.py index 9c1a5262..5e479b2f 100644 --- a/src/ewoksorange/tests/executor/test_executor_progress.py +++ b/src/ewoksorange/tests/executor/test_executor_progress.py @@ -1,11 +1,14 @@ """Ewoks task progress reaches the caller for every execution backend.""" import os +import time from typing import List from typing import Union from ...gui.qt_utils.progress import QProgress +from .tasks import InitProgressTask from .tasks import PidTask +from .tasks import ProgressBurstTask from .tasks import ProgressTask @@ -44,6 +47,40 @@ def test_progress(ewoksorange_qtapp, executor_context_factory): assert result["pid"].value == os.getpid() +def test_progress_during_creation(ewoksorange_qtapp, executor_context_factory): + """The progress a task reports while it is created arrives between + `aboutToStart` and `started`.""" + with executor_context_factory() as (kind, executor, recorder): + progress = QProgress() + events: List[Union[str, int]] = [] + executor.aboutToStart.connect(lambda _: events.append("aboutToStart")) + executor.started.connect(lambda _: events.append("started")) + progress.sigProgressChanged.connect(events.append) + + future = executor.submit_task( + InitProgressTask, inputs={"value": 1}, progress=progress + ) + + assert future.result(timeout=30)["value"].value == 1 + recorder.wait_for("finished", 1) + assert events == ["aboutToStart", 50, "started", 100] + + +def test_fast_progress_does_not_delay_result( + ewoksorange_qtapp, executor_context_factory +): + """A task reporting progress faster than it can be relayed returns its + result without delay.""" + with executor_context_factory() as (kind, executor, recorder): + progress = QProgress() + future = executor.submit_task( + ProgressBurstTask, inputs={"duration": 1}, progress=progress + ) + + end = future.result(timeout=30)["end"].value + assert time.time() - end < 5 + + def test_progress_for_task_without_progress_support( ewoksorange_qtapp, executor_context_factory ):