diff --git a/CHANGELOG.md b/CHANGELOG.md index ba78c673..3de1c14a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/doc/explanations/concurrency.rst b/doc/explanations/concurrency.rst index d15ad752..ab6d3679 100644 --- a/doc/explanations/concurrency.rst +++ b/doc/explanations/concurrency.rst @@ -1,26 +1,236 @@ Execution Backends ================== -:class:`~ewoksorange.gui.concurrency.executor.EwoksExecutor` runs a task in one -of three ways (see :class:`~ewoksorange.gui.concurrency.executor.Concurrency`): -in the calling thread, in a thread pool or in a process pool. Whatever the -backend, receivers observe the same order of signals for a task: +A :term:`widget` executes its :term:`task` through an :term:`executor`, in one of three +:term:`backends `: in the :term:`GUI thread`, in a thread pool or in a process +pool. This page has two parts: -``submitted`` → ``aboutToStart`` → ``started`` → progress values → -``aborted`` (if any) → ``succeeded`` / ``failed`` → ``finished`` +- :ref:`concurrency-contract`, for widget authors: which backend to choose, how to end an + :term:`execution`, and which signals arrive in which order. +- :ref:`concurrency-implementation`, for maintainers: how each backend keeps that order. -This page explains how each backend preserves that order. In the diagrams -below, ``progress?`` stands for the optional ``progress`` task argument: it is -only forwarded when the task class is a ``TaskWithProgress``, and dropped -otherwise. Progress steps marked ``[0..n]`` happen once per progress value the -task reports, so never without progress. +See :doc:`execution` for how an execution fits in the life of a workflow. + +.. _concurrency-contract: + +Contract +-------- + +Executing from a widget +^^^^^^^^^^^^^^^^^^^^^^^ + +A widget executes its task with +:meth:`~ewoksorange.gui.owwidgets.base.OWEwoksBaseWidget.execute_ewoks_task`, which +returns the :term:`task future` of the execution, or ``None`` when the submission was +dropped. The executor of the widget is +:attr:`~ewoksorange.gui.owwidgets.base.OWEwoksBaseWidget.task_executor`. + +:class:`~ewoksorange.gui.owwidgets.base.OWEwoksBaseWidget` already handles the signals of +its executor: it starts the progress bar on ``aboutToStart``, and on ``succeeded`` or +``failed`` it propagates the outputs downstream or invalidates them, then stops the +progress bar. Connect to the signals only to do more than that. + +A widget configures its executor with class arguments: + +- ``concurrency``: the :term:`backend`, see below. +- ``max_workers``: how many executions can run at the same time (``None`` for the pool + default). Ignored by the ``"sync"`` backend. +- ``submit_policy``: what to do with a submission while an execution is running: + + - ``"always"`` (the default): queue it until a :term:`worker` is free. + - ``"drop_if_busy"``: drop it. + +- ``mp_context``: the start method of the worker processes (``"spawn"`` by default). Only + used by the ``"process"`` backend. + +Choosing a backend +^^^^^^^^^^^^^^^^^^ + +.. list-table:: + :header-rows: 1 + + * - ``concurrency`` + - The task runs in + - The GUI stays responsive + - Ways to end an execution + - Requirements + * - ``"sync"`` + - the :term:`GUI thread`, inside ``submit_task()`` + - no, not even to display progress + - :term:`cancel` and :term:`stop request`, only from a slot connected to the + signals of the execution + - + * - ``"thread"`` (default) + - a :term:`worker` thread + - yes + - :term:`cancel`, :term:`stop request` + - + * - ``"process"`` + - a :term:`worker` process + - yes + - :term:`cancel`, :term:`stop request`, :term:`abort`, :term:`kill` + - the task class must be importable and its inputs and outputs picklable + +Ending an execution +^^^^^^^^^^^^^^^^^^^ + +The :term:`task future` can end an execution, from mild to forceful: + +.. list-table:: + :header-rows: 1 + + * - Method + - On a :term:`queued ` task + - On a running task + - Then ``result()`` + * - ``cancel()`` + - the task never starts + - no effect + - raises ``CancelledError`` + * - ``request_stop()`` + - the task never starts + - the task decides whether and how to stop + - raises :class:`~ewoksorange.gui.concurrency.future.TaskStoppedError` when the task + never started, otherwise returns or raises what the task does + * - ``abort()`` + - the task never starts + - interrupted like Ctrl-C (the task sees a ``KeyboardInterrupt``) once it runs + Python code again, so it can clean up + - raises :class:`~ewoksorange.gui.concurrency.future.TaskAbortedError` when the task + was interrupted + * - ``kill()`` + - the task never starts + - its process is terminated, it cannot clean up + - raises :class:`~ewoksorange.gui.concurrency.future.TaskAbortedError` + +``CancelledError`` is ``concurrent.futures.CancelledError``. + +Each method returns ``True`` when it prevented the start or reached the running task. On +a queued task, the first call decides: the task never starts, and later calls return +``False``. On a running task, ``request_stop()`` and ``abort()`` can be repeated, and the +calls can escalate, for example ``request_stop()``, then ``abort()``, then ``kill()``. +None of them has an effect on a finished task, which includes a task ended by ``kill()``. +``abort()`` and ``kill()`` have no effect in the ``"sync"`` and ``"thread"`` backends. + +A task honours ``request_stop()`` by checking ``self.stop_requested`` or by implementing +``_on_stop_requested()`` (see `ewokscore `_). + +How a running task reacts to a stop request is up to the task: it can raise, return early +with missing or partial outputs, or complete normally. After a stop request, +``succeeded`` therefore does not mean that the outputs are complete: check +``stop_requested()`` before using them. + +Signals +^^^^^^^ + +The :term:`executor` emits these signals, each carrying the :term:`task future` except +``ignored``: + +- ``submitted``: the task was submitted. +- ``aboutToStart``: the task is about to be created. +- ``started``: the task was created and is about to execute. +- ``cancelled``: ``cancel()`` prevented the start. +- ``stopRequested``: ``request_stop()`` prevented the start or reached the running task. +- ``aborted`` / ``killed``: ``abort()`` / ``kill()`` prevented the start or ended the task. +- ``succeeded`` / ``failed``: the task finished without / with an exception. +- ``finished``: the execution is over. +- ``ignored`` (no argument): a submission was dropped by ``submit_policy="drop_if_busy"``. + +Progress values are not executor signals: the :term:`progress object` emits +``sigProgressChanged`` when the progress reported by the task changes. They keep their +order with respect to the executor signals in all backends. + +These signals arrive in the following order, after ``submitted``: + +.. list-table:: + :header-rows: 1 + + * - Case + - Signals after ``submitted`` + * - The task succeeds + - ``aboutToStart``, ``started``, progress values, ``succeeded``, ``finished`` + * - The task raises + - ``aboutToStart``, ``started`` (not when creating the task raises), progress + values, ``failed``, ``finished`` + * - Cancelled before it starts + - ``cancelled``, ``finished`` + * - Stop requested before it starts + - ``stopRequested``, ``failed``, ``finished`` + * - Aborted before it starts + - ``aborted``, ``failed``, ``finished`` + * - Killed before it starts + - ``killed``, ``failed``, ``finished`` + * - Stop requested while running + - ``aboutToStart``, ``started``, progress values, ``stopRequested``, then + ``failed`` when the task raises, otherwise ``succeeded`` (with complete, partial + or missing outputs), ``finished`` + * - Aborted while running + - ``aboutToStart``, ``started``, progress values, ``aborted``, ``failed``, + ``finished`` + * - Killed while running + - ``aboutToStart``, ``started``, progress values, ``killed``, ``failed``, + ``finished`` + * - The worker process crashes (``"process"`` only) + - ``aboutToStart``, ``started``, progress values, ``failed``, ``finished`` + +The signals arrive only as far as the execution got: a stop request, an abort or a kill +can arrive before ``aboutToStart`` or ``started`` were emitted. An abort that does not +interrupt the task in time (the task ends first, or it catches ``KeyboardInterrupt``) +gives the signals of a task that was not aborted. A submission dropped by +``submit_policy="drop_if_busy"`` only emits ``ignored``. + +What this means for a widget: + +- ``finished`` is the last signal of every submitted execution, so it is the place to + release what the execution uses. +- No ``aboutToStart``, ``started`` or progress value arrives after the other signals of the + execution, unless the relay of the ``"process"`` backend times out (see below). +- An execution can finish without ever emitting ``aboutToStart`` or ``started``. +- ``failed`` is also emitted for an execution that was stopped, aborted or killed. The + task future tells why with ``stop_requested()``, ``aborted()`` and ``killed()``, and + ``cancelled()`` for a cancelled execution. +- Connect to the signals with the default connection type (plain ``connect()``): all slots + then run in the GUI thread. Avoid ``Qt.DirectConnection``: such a slot runs in the + thread that emits the signal, so it must not touch widgets, and it must return quickly + for the order above to hold (see :ref:`concurrency-implementation`). + +.. _concurrency-implementation: + +Implementation +-------------- + +In the diagrams below: + +- A solid line with a filled arrowhead is a call or an action. +- A dotted line is a return value, or a message read from a queue. +- A line with an open arrowhead is a Qt signal: + + - ⚡ emitted in the GUI thread: the connected slots run right away. + - 📨 emitted in another thread: Qt posts it to the event queue of the GUI thread (a + :term:`queued connection`) and the connected slots run later, when the GUI thread + processes its events. + +- ``_done`` is an internal signal of the executor, not one of the signals above. +- ``progress?`` is the optional ``progress`` task argument. It is kept when the task + class derives from ``ewokscore.TaskWithProgress``, and dropped otherwise. +- ``[0..n]`` marks a step that is repeated for each progress value, and skipped without a + :term:`progress object`. + +The diagrams show an execution that is not ended early. + +In the ``"thread"`` and ``"process"`` backends, the executor does not emit the 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()``, for example), 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 +execution that were posted before. Sync ----- +^^^^ -Everything runs in the calling (GUI) thread, through direct calls. The order -is the order of the calls, and ``submit_task()`` only returns once the task is -done. +The task runs in the GUI thread, through direct calls. The order is the order of the +calls, and ``submit_task()`` only returns once ``finished`` was emitted. .. mermaid:: @@ -33,29 +243,30 @@ done. end W->>E: submit_task(task, progress?) - E->>E: _handle_progess_arg: keep or drop progress - E-->>W: submitted - E->>R: _sync_submit(runner) - R-->>W: aboutToStart (direct) + E->>E: keep or drop progress + E-)W: ⚡ submitted + E->>E: mark the future as running + E->>R: run + R-)W: ⚡ aboutToStart R->>T: create task - R-->>W: started (direct) + R-)W: ⚡ started R->>T: execute - T->>T: [0..n] progress = v, QProgress emits sigProgressChanged(v) (direct) - T-->>R: result / exception - R->>E: raw_future.set_result / set_exception - E->>E: add_done_callback runs right away (future already done) - E->>E: _done_callback: stop_watching() (no-op), then _handle_done - E-->>W: succeeded / failed - E-->>W: finished - E-->>W: return TaskFuture (already done) + T->>T: [0..n] progress = v + T-)W: [0..n] ⚡ sigProgressChanged(v) of the progress object + T-->>R: outputs / exception + R-->>E: outputs / exception + E->>E: set the future result + E->>E: add the done callback, which runs right away + E-)W: ⚡ succeeded / failed + E-)W: ⚡ finished + E-->>W: TaskFuture (already done) Thread ------- +^^^^^^ -The task runs in a worker thread, which emits the executor signals and the -progress signal itself. Qt queues them to the GUI thread in the order they -were emitted from that single thread, so the order holds without any extra -synchronization. +The task runs in a worker thread, which emits ``aboutToStart``, ``started`` and the +progress values. Qt delivers them in the order they were emitted, since they come from a +single thread. .. mermaid:: @@ -70,42 +281,56 @@ synchronization. end W->>E: submit_task(task, progress?) - E->>E: _handle_progess_arg: keep or drop progress + E->>E: keep or drop progress E->>R: pool.submit(runner) - R->>R: _wait_ready() - E-->>W: submitted + R->>R: wait for ready_event (blocks) + E-)W: ⚡ submitted E->>R: ready_event.set() - R-)W: aboutToStart (queued) + R-)W: 📨 aboutToStart R->>T: create task - R-)W: started (queued) + R-)W: 📨 started R->>T: execute - T->>T: [0..n] progress = v, QProgress emits sigProgressChanged(v) (queued) - T-->>R: result / exception - R->>R: raw_future.set_result, done callback (worker thread) - R->>R: _done_callback: stop_watching() (no-op), then _handle_done - R-)W: succeeded / failed (queued) - R-)W: finished (queued) + T->>T: [0..n] progress = v + T-)W: [0..n] 📨 sigProgressChanged(v) of the progress object + T-->>R: outputs / exception + R->>R: the pool sets the future result, runs the done callback + R-)E: 📨 _done + E-)W: ⚡ succeeded / failed + E-)W: ⚡ finished -The ``ready_event`` keeps the worker from starting the task before -``submitted`` has been emitted. +``submit_task()`` can only emit ``submitted`` once the pool returned the future, so the +worker waits for ``ready_event`` before it emits anything. Process -------- +^^^^^^^ -The task runs in a worker process, which cannot emit Qt signals in the parent -nor update the caller's progress object (usually a ``QObject``, which is not -picklable). Instead, the child reports its events through a single -``multiprocessing`` manager queue: +The task runs in a worker process, which cannot emit Qt signals in the GUI thread's +process nor update the :term:`progress object` (usually a ``QObject``, which is not +picklable). The ``ProcessTaskController`` in the GUI process and the +``ProcessTaskRunner`` in the worker process communicate through two queues of a +:term:`manager`: -- the runner puts ``ABOUT_TO_START`` and ``STARTED``, -- the caller's progress object is replaced by a - :class:`~ewoksorange.gui.concurrency._progress.QueueProgress`, which puts - each progress value on that same queue. +- The queue to the controller carries ``ToController.ABOUT_TO_START``, + ``ToController.STARTED`` and the progress values: the progress object is replaced by a + :class:`~ewoksorange.gui.concurrency._messages.QueueProgress`. While the task executes, + a sender thread of the runner puts each value on this queue, so that an abort never + interrupts the task thread in the middle of a queue call. The task waits when it + reports progress faster than the sender thread puts it. The values an aborted task did + not send yet are dropped. The values reported while the task is created are put + directly, since an abort cannot interrupt that. The controller puts + ``ToController.END`` on the queue to end the reading. +- The queue to the runner carries the decision whether the task may start + (``ToRunner.START`` or a ``StartDenial``), then ``ToRunner.REQUEST_STOP`` and + ``ToRunner.ABORT``. While the task executes, a thread of the runner reads them, until + the runner puts ``ToRunner.END`` on it. -In the parent, a single relay thread started by -:meth:`ProcessTaskController.watch() ` -reads the queue and calls the matching callback: it emits ``aboutToStart`` or -``started``, or sets the value on the caller's progress object. +The controller has one :term:`relay` thread per execution that reads the queue to the +controller and emits the matching signals. On ``ToController.ABOUT_TO_START``, it decides +whether the task may start (it may not once it was cancelled, stopped, aborted or killed), +replies to the runner, and emits ``aboutToStart`` when allowed. The relay only starts once +``submitted`` was emitted, so the runner cannot start the task before that. The decision +is always the first message to the runner: stop requests and aborts are only put on the +queue to the runner once the start was allowed. .. mermaid:: @@ -114,73 +339,87 @@ reads the queue and calls the matching callback: it emits ``aboutToStart`` or participant W as Widget participant E as EwoksExecutor end - participant M as Manager (events_queue) - box Parent: relay thread + box Manager process + participant M as Queues + end + box GUI process: relay thread of the controller participant RL as Relay end - box Parent: pool management thread - participant PM as ProcessPool mgmt + box GUI process: pebble pool threads + participant PP as pebble.ProcessPool end - box Child process - participant C as ProcessTaskRunner + Task + box Worker process + participant R as ProcessTaskRunner + Task end W->>E: submit_task(task, progress?) - E->>M: create events_queue, ready/abort/aborted events - E->>E: progress replaced by QueueProgress(events_queue), or dropped - E->>RL: watch(on_about_to_start, on_started, on_progress) - RL->>M: get() (blocks) - E->>PM: pool.submit(runner) (pickled) - E-->>W: submitted - E->>M: ready_event.set() - PM->>C: run runner - C->>M: _wait_ready() - C->>M: put(ABOUT_TO_START) - M-->>RL: ABOUT_TO_START - RL-)W: aboutToStart (queued) - C->>C: create task - C->>M: put(STARTED) - M-->>RL: STARTED - RL-)W: started (queued) - C->>C: execute - C->>C: [0..n] progress = v, QueueProgress puts v on events_queue - RL->>RL: [0..n] get v, on_progress(v): QProgress emits sigProgressChanged(v) (queued) - C-->>PM: result / exception - PM->>PM: raw_future.set_result, then _done_callback - PM->>M: stop_watching(): put(STOP) - M-->>RL: STOP (after all events above) + E->>M: create the queue to the controller and the queue to the runner + E->>E: replace progress by QueueProgress, or drop it + E->>PP: pool.submit(runner) + E-)W: ⚡ submitted + E->>RL: start the relay + Note over E,RL: only now can the runner start the task + RL->>M: read the queue to the controller (blocks) + PP->>R: send the runner to a worker process (pickled) + R->>M: put ToController.ABOUT_TO_START, then wait for the decision + M-->>RL: ToController.ABOUT_TO_START + RL->>M: put ToRunner.START on the queue to the runner + RL-)W: 📨 aboutToStart + M-->>R: ToRunner.START + R->>R: create task + R->>M: put ToController.STARTED + M-->>RL: ToController.STARTED + RL-)W: 📨 started + R->>R: execute + R->>M: [0..n] put v + M-->>RL: [0..n] v + RL-)W: [0..n] 📨 sigProgressChanged(v) of the progress object + R-->>PP: outputs / exception (pickled) + PP->>PP: set the future result, run the done callback + PP->>M: put ToController.END + M-->>RL: ToController.END RL->>RL: exit - PM->>RL: join(timeout=5s) - PM->>PM: _handle_done - PM-)W: succeeded / failed (queued) - PM-)W: finished (queued) - -The diagram shows the relay reading each event right after it is put; in -practice both run concurrently. What matters is the ordering: - -- **One queue, one reader.** All events of a task travel through the same - first-in first-out queue and are relayed by a single thread, so callbacks - run in the order the child emitted them. -- **Everything is delivered before completion.** The child puts all its - events before returning its result, while ``STOP`` is only put by the parent - once the result is in (in ``_done_callback``). Manager proxy calls are - synchronous, so every child event is already queued ahead of ``STOP``. - Waiting for the relay to reach ``STOP`` therefore delivers all of them before - ``succeeded`` / ``failed`` are emitted. - -The relay's lifetime is tied to the task's future, not to a timeout: - -- **Cancelled while queued.** The child never runs, so the queue only holds - ``STOP`` and the relay exits immediately. Nothing else is emitted. -- **Child crash.** The pool sets ``BrokenProcessPool`` on the future, so the - done callback still runs and puts ``STOP``. + PP->>RL: join (5 s timeout) + PP-)E: 📨 _done + E-)W: ⚡ succeeded / failed + E-)W: ⚡ finished + +The diagram shows the relay reading each message right after it is put; in practice +both run concurrently. What matters is the ordering: + +- **One queue, one reader.** All messages to the controller travel through the same + first-in first-out queue and are handled by its relay thread, so the signals are + emitted in the order the runner put the messages. +- **All messages before the completion.** The runner puts all its messages, and waits + for its sender thread to put the progress values, before it returns its result, and + the done callback only puts ``ToController.END`` once the result is in. A manager queue + call returns once the message is in the queue, so all messages of the runner are ahead + of ``ToController.END``. The done callback waits for the relay to reach + ``ToController.END`` before it emits ``_done``, so the completion signals are emitted + after all signals of the relay. + +The relay's lifetime is tied to the future of the execution, not to a timeout: + +- **Prevented from starting.** The runner only puts ``ToController.ABOUT_TO_START``, is + denied and returns or raises, so the done callback runs and puts ``ToController.END``. + After ``kill()``, the future of the pool is cancelled instead (see below), so the runner + may not run at all. +- **Crashed worker.** The pool sets a ``pebble.ProcessExpired`` error on the future and + replaces the worker process, so the done callback runs and puts ``ToController.END``. +- **Killed worker.** ``kill()`` cancels the future of the pool, which runs the done callback + right away, in the thread calling ``kill()``, so that thread waits for the relay (at + most 5 s). The done callback puts ``ToController.END``, so the messages the runner puts after + that are never read. When a worker already picked up the task, the pool then + 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 - end the relay early and silently drop the events that follow. - -``stop_watching()`` itself is bounded: putting ``STOP`` is a single call to the -manager, and the relay thread is joined with a 5 s timeout. That wait only -matters if a receiver connected with ``Qt.DirectConnection`` is slow, since -such a slot runs in the relay thread. With the default (queued) connections, -the relay only posts events to the GUI thread and never blocks. +- **Long-running task.** The relay keeps waiting, since progress may still come. There is + deliberately no timeout on reading the queue: a timeout would end the relay early and + silently drop the messages that follow. + +The done callback waits at most 5 s for the relay to reach ``ToController.END``. With the +default connections, the relay only posts signals to the GUI thread, so it reaches +``ToController.END`` quickly. A slow slot connected with ``Qt.DirectConnection`` runs in the +relay thread and delays it. When the relay does not reach ``ToController.END`` in time, the +completion signals are emitted anyway, and the remaining signals of the relay follow +them. The same happens when such a slot calls ``kill()``: the done callback then runs in +the relay thread and cannot wait for it. diff --git a/doc/explanations/execution.rst b/doc/explanations/execution.rst index 87c68500..a5d1fdfd 100644 --- a/doc/explanations/execution.rst +++ b/doc/explanations/execution.rst @@ -56,8 +56,8 @@ One execution cycle BG-->>BG: task runs Note right of BG: has_pending_task() -> still True
(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. diff --git a/doc/reference.rst b/doc/reference.rst index 899d6f04..a87ce8df 100644 --- a/doc/reference.rst +++ b/doc/reference.rst @@ -5,3 +5,4 @@ Reference :maxdepth: 1 reference/api + reference/glossary diff --git a/doc/reference/glossary.rst b/doc/reference/glossary.rst new file mode 100644 index 00000000..9a091f69 --- /dev/null +++ b/doc/reference/glossary.rst @@ -0,0 +1,114 @@ +Glossary +======== + +.. glossary:: + :sorted: + + Task + An `ewoks `_ 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 ` 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 `, 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 `, :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 ` 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() `). + It has no effect on a running task. + + Stop request + Ask a :term:`task` to stop + (:meth:`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() `), + also when it ignores :term:`stop requests `. The task can clean up and + the :term:`worker` survives. + + Kill + Terminate the :term:`worker` process running a :term:`task` + (:meth:`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 `_ 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`. 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 efaf1f70..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, ), } @@ -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,8 +238,17 @@ 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 @@ -214,6 +256,7 @@ def _on_finished(self, name: str, future: Any) -> None: 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() diff --git a/src/ewoksorange/gui/concurrency/_controllers/_in_process.py b/src/ewoksorange/gui/concurrency/_controllers/_in_process.py index cb72dc53..3d3984f1 100644 --- a/src/ewoksorange/gui/concurrency/_controllers/_in_process.py +++ b/src/ewoksorange/gui/concurrency/_controllers/_in_process.py @@ -1,8 +1,8 @@ import threading -from threading import Lock - -from ewokscore.task import Task +from concurrent.futures import Future +from typing import Optional +from .abstract import StartDenial from .abstract import TaskController @@ -11,25 +11,23 @@ class _InProcessTaskController(TaskController): or a worker thread).""" def __init__(self): - self._task = None - self._abort_event = threading.Event() - self._lock = Lock() + super().__init__() + self._stop_event = threading.Event() - def set_task(self, task: Task) -> None: - with self._lock: - self._task = task + def try_start(self) -> Optional[StartDenial]: + """Mark the task as started unless it was denied. Returns the denial.""" + return self._gate.open() @property - def abort_event(self) -> threading.Event: - return self._abort_event - - def abort(self) -> bool: - self._abort_event.set() - with self._lock: - return self._task is not None - - def aborted(self) -> bool: - with self._lock: - task = self._task - - return task is not None and task.stop_requested + def stop_event(self) -> threading.Event: + return self._stop_event + + def _send_stop_request(self) -> None: + self._stop_event.set() + + def cancel(self, future: Future) -> bool: + if not super().cancel(future): + return False + # When the pool already picked up the task, the runner does not start it. + future.cancel() + return True diff --git a/src/ewoksorange/gui/concurrency/_controllers/abstract.py b/src/ewoksorange/gui/concurrency/_controllers/abstract.py index 5e24409d..45d53297 100644 --- a/src/ewoksorange/gui/concurrency/_controllers/abstract.py +++ b/src/ewoksorange/gui/concurrency/_controllers/abstract.py @@ -1,28 +1,131 @@ from abc import ABC from abc import abstractmethod +from concurrent.futures import Future +from enum import Enum +from enum import auto +from threading import Lock from typing import Callable from typing import Optional +class StartDenial(Enum): + """Why a task is not allowed to start.""" + + CANCELLED = auto() + STOPPED = auto() + ABORTED = auto() + KILLED = auto() + + +class StartGate: + """Decides once whether a task may start.""" + + def __init__(self) -> None: + self._lock = Lock() + self._started = False + self._denial: Optional[StartDenial] = None + + @property + def started(self) -> bool: + return self._started + + @property + def denial(self) -> Optional[StartDenial]: + return self._denial + + def deny(self, denial: StartDenial) -> bool: + """Prevent the task from starting. + + Returns False when the task already started or was denied before. + """ + with self._lock: + if self._started or self._denial is not None: + return False + self._denial = denial + return True + + def open(self) -> Optional[StartDenial]: + """Let the task start unless it was denied. Returns the denial.""" + with self._lock: + self._started = self._denial is None + return self._denial + + class TaskController(ABC): - """Interface for controlling a running ewoks task. + """Interface for controlling an ewoks task. - The task (instance) is expected to be executed at most once. + The task (instance) is expected to be executed at most once. Only the first + of cancel, stop, abort and kill that reaches a task before it starts has an + effect: the task never starts. """ - @abstractmethod - def abort(self) -> bool: - """Request task abortion. + def __init__(self) -> None: + self._gate = StartGate() + self._stop_sent = False + + def request_stop(self) -> bool: + """Request the task to stop, or prevent it from starting. - Returns True if the abort request reached the task. + Returns True if the stop request prevented the start or reached the task. """ - raise NotImplementedError + if self._gate.deny(StartDenial.STOPPED): + return True + if not self._gate.started: + return False + self._stop_sent = True + self._send_stop_request() + return True @abstractmethod - def aborted(self) -> bool: - """Return True if the task was actually aborted.""" + 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. + + Returns True if the task will not be executed. + """ + return not future.done() and self._gate.deny(StartDenial.CANCELLED) + + 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 5cff0ed0..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 multiprocessing.synchronize import Event +from concurrent.futures import Future from queue import Queue from typing import Callable from typing import Optional -from .. import _progress +from .._messages import ToController +from .._messages import ToRunner +from .abstract import StartDenial from .abstract import TaskController _logger = logging.getLogger(__name__) @@ -14,21 +16,22 @@ class ProcessTaskController(TaskController): """Controls a task running in another process. - Communication with the child process happens through multiprocessing - Event/Queue proxies. The child reports "about to start", "started" and - progress through a single queue, relayed by a single thread, so the - callbacks are called in the order the child emitted them. + Communication with the runner in the other process happens through two + multiprocessing queue proxies, one per direction. The runner reports "about to + 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 and abort requests. """ - def __init__( - self, - abort_event: Event, - aborted_event: Event, - events_queue: Queue, - ): - self._abort_event = abort_event - self._aborted_event = aborted_event - self._events_queue = events_queue + def __init__(self, to_controller_queue: Queue, to_runner_queue: Queue): + super().__init__() + self._to_controller_queue = to_controller_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( @@ -37,22 +40,26 @@ def watch( on_started: Callable[[], None], on_progress: Optional[Callable[[float], None]] = None, ) -> None: - """Relay the events reported by the child process to the callbacks.""" + """Relay the messages of the runner to the callbacks. + + Also decides whether the runner may start the task, so that a + denied task never starts. + """ def _relay(): while True: try: # No timeout: `stop_watching` always ends the relay. - message = self._events_queue.get() + message = self._to_controller_queue.get() except Exception: _logger.debug("events relay failed", exc_info=True) return - if message == _progress.STOP: + if message is ToController.END: return try: - if message == _progress.ABOUT_TO_START: - on_about_to_start() - elif message == _progress.STARTED: + if message is ToController.ABOUT_TO_START: + self._allow_start(on_about_to_start) + elif message is ToController.STARTED: on_started() elif on_progress is not None: on_progress(message) @@ -62,11 +69,22 @@ def _relay(): self._relay_thread = threading.Thread(target=_relay, daemon=True) self._relay_thread.start() + def _allow_start(self, on_about_to_start: Callable[[], None]) -> None: + """Reply to the runner whether it may start the task, and call + `on_about_to_start` when allowed.""" + with self._to_runner_lock: + denial = self._gate.open() + # The runner waits for the reply. + self._to_runner_queue.put(ToRunner.START if denial is None else denial) + # Still before "started": the relay handles the events in order. + if denial is None: + on_about_to_start() + def stop_watching(self) -> None: """Wait for all events to be relayed, then stop the relay. - The child puts every event before returning its result, while the - sentinel below is only put once the parent sees the task finished (or + The runner puts every event before returning its result, while the + sentinel below is only put once the controller sees the task finished (or cancelled). Manager queues are FIFO, so draining up to the sentinel delivers all of them. """ @@ -74,17 +92,59 @@ def stop_watching(self) -> None: return try: - self._events_queue.put(_progress.STOP) + self._to_controller_queue.put(ToController.END) 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") + def _send_stop_request(self) -> None: + with self._to_runner_lock: + self._to_runner_queue.put(ToRunner.REQUEST_STOP) + def abort(self) -> bool: - self._abort_event.set() + 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) -> bool: - return self._aborted_event.is_set() + 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/_controllers/sync.py b/src/ewoksorange/gui/concurrency/_controllers/sync.py index d0457d28..bd4380f1 100644 --- a/src/ewoksorange/gui/concurrency/_controllers/sync.py +++ b/src/ewoksorange/gui/concurrency/_controllers/sync.py @@ -4,6 +4,6 @@ class SyncTaskController(_InProcessTaskController): """Controls a task running on the (blocked) calling thread. - `abort()` only has an effect when called concurrently from another + `request_stop()` only has an effect when called concurrently from another thread, since the calling thread is busy running the task. """ diff --git a/src/ewoksorange/gui/concurrency/_messages.py b/src/ewoksorange/gui/concurrency/_messages.py new file mode 100644 index 00000000..32ce5064 --- /dev/null +++ b/src/ewoksorange/gui/concurrency/_messages.py @@ -0,0 +1,117 @@ +""" +Messages between the controller of an ewoks task and the runner executing it in +another process. + +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, 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 + + +class ToController(Enum): + """Messages on the queue to the controller, next to the progress values.""" + + ABOUT_TO_START = auto() + """The runner is about to create the task and waits for a + :attr:`ToRunner.START` or a `StartDenial`.""" + + STARTED = auto() + """The runner created the task.""" + + END = auto() + """End the reading of the queue. Put by the controller itself.""" + + +class ToRunner(Enum): + """Messages on the queue to the runner.""" + + START = auto() + """Reply to :attr:`ToController.ABOUT_TO_START`: the task may start. Otherwise + the reply is a `StartDenial`.""" + + 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) + # The manager providing the queue shut down. + + def __init__(self, to_controller_queue: queue.Queue): + """ + :param to_controller_queue: A `multiprocessing` manager queue proxy. + """ + 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, 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(value) + except self._TRANSIENT_ERRORS: + pass diff --git a/src/ewoksorange/gui/concurrency/_progress.py b/src/ewoksorange/gui/concurrency/_progress.py deleted file mode 100644 index 9dec0e84..00000000 --- a/src/ewoksorange/gui/concurrency/_progress.py +++ /dev/null @@ -1,43 +0,0 @@ -""" -Ewoks task events (about to start, started, progress) across a process boundary. - -They all travel through a single queue, so the parent receives them in the -order the child process emitted them. -""" - -import queue - -from ewokscore.progress import BasePercentageProgress - -ABOUT_TO_START = "__about_to_start__" -"""Put on the queue by the child process right before creating the task.""" - -STARTED = "__started__" -"""Put on the queue by the child process once the task is created.""" - -STOP = "__stop__" -"""Sentinel put on the queue by the parent process to stop the relay.""" - - -class QueueProgress(BasePercentageProgress): - """Picklable progress that forwards updates to the parent process. - - Shares its queue with the other task events (see :data:`STARTED`). - """ - - _TRANSIENT_ERRORS = (EOFError, BrokenPipeError, ConnectionError) - # The manager providing the queue shut down. - - def __init__(self, progress_queue: queue.Queue): - """ - :param progress_queue: A `multiprocessing` manager queue proxy. - """ - super().__init__() - self._progress_queue = progress_queue - - def _update(self) -> None: - """Send the current progress to the parent process.""" - try: - self._progress_queue.put(self._progress) - except self._TRANSIENT_ERRORS: - pass diff --git a/src/ewoksorange/gui/concurrency/_runners/abstract.py b/src/ewoksorange/gui/concurrency/_runners/abstract.py index 6ddc3454..9fbaa583 100644 --- a/src/ewoksorange/gui/concurrency/_runners/abstract.py +++ b/src/ewoksorange/gui/concurrency/_runners/abstract.py @@ -3,43 +3,44 @@ from abc import abstractmethod from typing import Any from typing import Dict -from typing import Tuple +from typing import Optional from typing import Type from ewokscore.task import Task from ewokscore.variable import VariableContainer +from .._controllers.abstract import StartDenial +from ..future import TaskAbortedError +from ..future import TaskStoppedError + class TaskRunner(ABC): """Base class for ewoks task execution runners.""" - _TRANSIENT_ABORT_ERRORS: Tuple[Type[Exception], ...] = () - - def __init__( - self, - task_class: Type[Task], - task_kwargs: Dict[str, Any], - abort_event: threading.Event, - ): + def __init__(self, task_class: Type[Task], task_kwargs: Dict[str, Any]): self._task_class = task_class self._task_kwargs = task_kwargs - self._abort_event = abort_event - def __call__(self) -> VariableContainer: + def __call__(self) -> Optional[VariableContainer]: self._wait_ready() + denial = self._try_start() + if denial is StartDenial.CANCELLED: + return None + if denial is not None: + raise _denial_error(self._task_class, denial) self._announce_about_to_start() task = self._create_task() - # Announce (only) once the task exists, so an abort() reacting to - # that signal always finds a task to abort. self._announce_started(task) - try: - return self._execute(task) - finally: - self._finalize(task) + return self._execute(task) def _wait_ready(self) -> None: """Block until the task may be created and executed.""" + @abstractmethod + def _try_start(self) -> Optional[StartDenial]: + """Return why the task must not be executed, `None` to execute it.""" + raise NotImplementedError + @abstractmethod def _announce_about_to_start(self) -> None: """Announce the task is about to be created and executed.""" @@ -47,21 +48,25 @@ def _announce_about_to_start(self) -> None: @abstractmethod def _announce_started(self, task: Task) -> None: - """Register `task` (if applicable) and announce it as started.""" + """Announce `task` as started.""" raise NotImplementedError - def _finalize(self, task: Task) -> None: - """Called once execution has finished, however it ended.""" - def _create_task(self) -> Task: """Instantiate task class.""" return self._task_class(**self._task_kwargs) + @abstractmethod def _execute(self, task: Task) -> VariableContainer: - """Execute `task`, aborting it for as long as `abort_event` is set.""" + """Execute `task`, requesting it to stop when requested by the caller.""" + raise NotImplementedError + + def _execute_with_stop_event( + self, task: Task, stop_event: threading.Event + ) -> VariableContainer: + """Execute `task`, requesting it to stop for as long as `stop_event` is set.""" done = threading.Event() - def _watch_abort(): + def _watch_stop_request(): """ `task.execute()` resets the task's `stop_requested` flag, which can silently undo a `request_stop()` racing with the start @@ -71,33 +76,42 @@ def _watch_abort(): until execution finishes, so a task's `run()` reliably observes it regardless of the timing. - The output of an aborted task is undefined. It could be an + The output of a task requested to stop is undefined. It could be an exception, undefined outputs, pertially defined outputs or fully defined outputs. """ - try: - self._abort_event.wait() - if done.is_set(): - # Woken up only to let this thread exit; the task - # finished without ever being aborted. - return - while not done.is_set(): - task.request_stop() - done.wait(timeout=0.01) - except self._TRANSIENT_ABORT_ERRORS: - pass - - watcher = threading.Thread(target=_watch_abort, daemon=True) + stop_event.wait() + if done.is_set(): + # Woken up only to let this thread exit; the task + # finished without ever being requested to stop. + return + while not done.is_set(): + task.request_stop() + done.wait(timeout=0.01) + + 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() - try: - # Wake the watcher if the task completed without abort. - self._abort_event.set() - except self._TRANSIENT_ABORT_ERRORS: - pass + # Wake the watcher if the task completed without stop request. + stop_event.set() watcher.join() 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 5b4bcc30..a6b82827 100644 --- a/src/ewoksorange/gui/concurrency/_runners/process.py +++ b/src/ewoksorange/gui/concurrency/_runners/process.py @@ -1,47 +1,134 @@ +import _thread import queue +import signal +import sys import threading from typing import Any from typing import Dict +from typing import Optional from typing import Type from ewokscore.task import Task +from ewokscore.variable import VariableContainer -from .. import _progress +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 class ProcessTaskRunner(TaskRunner): """Picklable callable executed inside the subprocess.""" - _TRANSIENT_ABORT_ERRORS = (EOFError, BrokenPipeError, ConnectionError) + _TRANSIENT_ERRORS = (EOFError, BrokenPipeError, ConnectionError) # The manager providing the IPC objects shut down. def __init__( self, task_class: Type[Task], task_kwargs: Dict[str, Any], - ready_event: threading.Event, - events_queue: queue.Queue, - abort_event: threading.Event, - aborted_event: threading.Event, + to_controller_queue: queue.Queue, + to_runner_queue: queue.Queue, ): - super().__init__(task_class, task_kwargs, abort_event) - self._ready_event = ready_event - self._events_queue = events_queue - self._aborted_event = aborted_event + 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 _wait_ready(self) -> None: - self._ready_event.wait() + def _try_start(self) -> Optional[StartDenial]: + # The controller decides, so that a denied task never starts. + self._to_controller_queue.put(ToController.ABOUT_TO_START) + reply = self._to_runner_queue.get() + if reply is ToRunner.START: + return None + if isinstance(reply, StartDenial): + return reply + raise RuntimeError(f"Unexpected start decision {reply!r}") def _announce_about_to_start(self) -> None: - self._events_queue.put(_progress.ABOUT_TO_START) + # The controller announces it when allowing the start. + pass def _announce_started(self, task: Task) -> None: - self._events_queue.put(_progress.STARTED) + self._to_controller_queue.put(ToController.STARTED) - def _finalize(self, task: Task) -> None: + def _execute(self, task: Task) -> VariableContainer: + """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: + while True: + 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: - if task.stop_requested: - self._aborted_event.set() - except self._TRANSIENT_ABORT_ERRORS: - pass + 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/_runners/sync.py b/src/ewoksorange/gui/concurrency/_runners/sync.py index a3caa4a7..64771fac 100644 --- a/src/ewoksorange/gui/concurrency/_runners/sync.py +++ b/src/ewoksorange/gui/concurrency/_runners/sync.py @@ -1,10 +1,13 @@ from typing import Any from typing import Callable from typing import Dict +from typing import Optional from typing import Type from ewokscore.task import Task +from ewokscore.variable import VariableContainer +from .._controllers.abstract import StartDenial from .._controllers.sync import SyncTaskController from .abstract import TaskRunner @@ -20,14 +23,19 @@ def __init__( on_about_to_start: Callable[[], None], on_started: Callable[[], None], ): - super().__init__(task_class, task_kwargs, controller.abort_event) + super().__init__(task_class, task_kwargs) self._controller = controller self._on_about_to_start = on_about_to_start self._on_started = on_started + def _try_start(self) -> Optional[StartDenial]: + return self._controller.try_start() + def _announce_about_to_start(self) -> None: self._on_about_to_start() def _announce_started(self, task: Task) -> None: - self._controller.set_task(task) self._on_started() + + def _execute(self, task: Task) -> VariableContainer: + return self._execute_with_stop_event(task, self._controller.stop_event) diff --git a/src/ewoksorange/gui/concurrency/_runners/thread.py b/src/ewoksorange/gui/concurrency/_runners/thread.py index e72deafc..8f60e1a3 100644 --- a/src/ewoksorange/gui/concurrency/_runners/thread.py +++ b/src/ewoksorange/gui/concurrency/_runners/thread.py @@ -2,10 +2,13 @@ from typing import Any from typing import Callable from typing import Dict +from typing import Optional from typing import Type from ewokscore.task import Task +from ewokscore.variable import VariableContainer +from .._controllers.abstract import StartDenial from .._controllers.thread import ThreadTaskController from .abstract import TaskRunner @@ -22,7 +25,7 @@ def __init__( on_about_to_start: Callable[[], None], on_started: Callable[[], None], ): - super().__init__(task_class, task_kwargs, controller.abort_event) + super().__init__(task_class, task_kwargs) self._controller = controller self._ready_event = ready_event self._on_about_to_start = on_about_to_start @@ -31,9 +34,14 @@ def __init__( def _wait_ready(self) -> None: self._ready_event.wait() + def _try_start(self) -> Optional[StartDenial]: + return self._controller.try_start() + def _announce_about_to_start(self) -> None: self._on_about_to_start() def _announce_started(self, task: Task) -> None: - self._controller.set_task(task) self._on_started() + + def _execute(self, task: Task) -> VariableContainer: + return self._execute_with_stop_event(task, self._controller.stop_event) diff --git a/src/ewoksorange/gui/concurrency/executor.py b/src/ewoksorange/gui/concurrency/executor.py index 5229261c..6d1fa4e1 100644 --- a/src/ewoksorange/gui/concurrency/executor.py +++ b/src/ewoksorange/gui/concurrency/executor.py @@ -17,15 +17,18 @@ 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 from ewokscore import TaskWithProgress from ewokscore.task import Task from ewokscore.variable import VariableContainer from . import _controllers -from . import _progress +from . import _messages from . import _runners from .future import TaskFuture @@ -54,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. @@ -66,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") @@ -77,10 +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 TaskFuture objects that support - both native future cancellation (cancel()) and ewoks task abort (abort()). - 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. """ @@ -89,7 +92,7 @@ class EwoksExecutor(QObject): aboutToStart = Signal(TaskFuture) """Emitted when a worker picks the task up, right before it is created and - executed. Not emitted for a task cancelled while queued.""" + executed. Not emitted for a task prevented from starting.""" started = Signal(TaskFuture) """Emitted when a task starts executing.""" @@ -98,25 +101,40 @@ class EwoksExecutor(QObject): """Emitted when a task finishes successfully.""" failed = Signal(TaskFuture) - """Emitted when a task raises an exception.""" + """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.""" + cancelled = Signal(TaskFuture) + """Emitted when :meth:`TaskFuture.cancel` prevented a queued task from starting.""" + + 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 was aborted.""" + """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 when a task finishes (success, failure, or abort).""" + """Emitted last, for every submitted task.""" + + _done = Signal(object, object, object) + # Carries a done task to the thread of the executor, behind the events + # already queued for that thread. 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 @@ -124,13 +142,18 @@ 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) self._manager: Optional[multiprocessing.managers.SyncManager] = None @@ -146,13 +169,18 @@ def submit_task( self._running += 1 - if self._executor is None: - return self._submit_sync(task_class, task_kwargs) + try: + if self._executor is None: + return self._submit_sync(task_class, task_kwargs) - if self._as_process: - return self._submit_process(task_class, task_kwargs) + if self._as_process: + return self._submit_process(task_class, task_kwargs) - return self._submit_thread(task_class, task_kwargs) + return self._submit_thread(task_class, task_kwargs) + except BaseException: + with self._lock: + self._running -= 1 + raise def _submit_sync( self, task_class: Type[Task], task_kwargs: Dict[str, Any] @@ -160,7 +188,7 @@ def _submit_sync( controller = _controllers.SyncTaskController() self_ref = weakref.ref(self) holder: List[Optional[TaskFuture]] = [None] - task_kwargs = self._handle_progess_arg(task_class, task_kwargs) + task_kwargs = self._handle_progress_arg(task_class, task_kwargs) runner = _runners.SyncTaskRunner( task_class, task_kwargs, @@ -186,7 +214,7 @@ def _submit_thread( ready_event = Event() self_ref = weakref.ref(self) holder: List[Optional[TaskFuture]] = [None] - task_kwargs = self._handle_progess_arg(task_class, task_kwargs) + task_kwargs = self._handle_progress_arg(task_class, task_kwargs) runner = _runners.ThreadTaskRunner( task_class, task_kwargs, @@ -207,13 +235,11 @@ def _submit_process( ) -> TaskFuture: manager = self._get_manager() - ready_event = manager.Event() - events_queue = manager.Queue() - abort_event = manager.Event() - aborted_event = manager.Event() + to_controller_queue = manager.Queue() + to_runner_queue = manager.Queue() controller = _controllers.ProcessTaskController( - abort_event, aborted_event, events_queue + to_controller_queue, to_runner_queue ) # The caller's progress object is usually unpicklable (QObject), so @@ -221,37 +247,34 @@ def _submit_process( progress = self._pop_progress_arg(task_class, task_kwargs) on_progress: Optional[Callable[[float], None]] = None if progress is not None: - task_kwargs["progress"] = _progress.QueueProgress(events_queue) + task_kwargs["progress"] = _messages.QueueProgress(to_controller_queue) def on_progress(value: float) -> None: progress.progress = value runner = _runners.ProcessTaskRunner( - task_class, - task_kwargs, - ready_event, - events_queue, - abort_event, - aborted_event, + task_class, task_kwargs, to_controller_queue, to_runner_queue ) self_ref = weakref.ref(self) holder: List[Optional[TaskFuture]] = [None] - raw_future = self._executor.submit(runner) + raw_future = self._executor.submit(runner, None) - # Watch after the submission in case 'submit' fails (make sure 'STOP' will always be sent after this point). - controller.watch( - on_about_to_start=lambda: _emit_about_to_start(self_ref, holder), - on_started=lambda: _emit_started(self_ref, holder), - on_progress=on_progress, - ) + def watch() -> None: + # Only after the submission (a failing 'submit' is never watched) and + # after "submitted" (the runner cannot start before the relay allows it). + controller.watch( + on_about_to_start=lambda: _emit_about_to_start(self_ref, holder), + on_started=lambda: _emit_started(self_ref, holder), + on_progress=on_progress, + ) return self._finalize_submission( - raw_future, controller, holder, self_ref, after_submitted=ready_event.set + raw_future, controller, holder, self_ref, after_submitted=watch ) - def _handle_progess_arg( + def _handle_progress_arg( self, task_class: Type[Task], task_kwargs: Dict[str, Any] ) -> Dict[str, Any]: """Prepare the `progress` task argument for `task_class`, for backends @@ -290,10 +313,19 @@ 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( - lambda f: _done_callback(self_ref, f, task_future, controller) + lambda f: _done_callback(self_ref, f, task_future, controller, queue_done) ) return task_future @@ -312,18 +344,21 @@ def _handle_done( task_future: TaskFuture, controller: _controllers.TaskController, ) -> None: - with self._lock: - self._running -= 1 - - if raw_future.cancelled(): + if task_future.cancelled(): + self.cancelled.emit(task_future) + self.finished.emit(task_future) return + 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: - raw_future.result() - except Exception: + task_future.result() + except BaseException: self.failed.emit(task_future) else: self.succeeded.emit(task_future) @@ -335,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() @@ -345,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() @@ -355,15 +393,27 @@ def _done_callback( raw_future: futures.Future[VariableContainer], task_future: TaskFuture, controller: _controllers.TaskController, + queue_done: bool, ) -> None: - # Before emitting anything: this delivers the events still in flight - # (process backend), so receivers of "succeeded"/"failed" cannot observe - # "started" or progress arriving after they considered the task finished. - # Done even if the executor is gone, so the relay never outlives the task. + # Before emitting anything: this queues the events still in flight + # (process backend) for the thread of the executor. Done even if the + # executor is gone, so the relay never outlives the task. controller.stop_watching() executor = executor_ref() - if executor is not None: + if executor is None: + return + + with executor._lock: + executor._running -= 1 + + 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 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) @@ -390,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 b7b6114b..f985f1c3 100644 --- a/src/ewoksorange/gui/concurrency/future.py +++ b/src/ewoksorange/gui/concurrency/future.py @@ -1,5 +1,6 @@ from __future__ import annotations +from concurrent.futures import CancelledError from concurrent.futures import Future from typing import Any from typing import Callable @@ -12,8 +13,34 @@ from . import _controllers +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: - """Wraps a concurrent.futures.Future with ewoks-specific abort support.""" + """Future of an ewoks task submitted to an `EwoksExecutor`. + + 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 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`, :meth:`stop_requested`, :meth:`aborted` and :meth:`killed` + report which of these happened. + """ def __init__( self, @@ -24,28 +51,69 @@ def __init__( self._controller = controller def cancel(self) -> bool: - """Prevent execution if the task has not started. + """Prevent a queued task from starting. Has no effect once the task started. + + The result raises `CancelledError`. - This is the native Future cancellation mechanism. It only works while - the task is still waiting in the executor queue. + :return: True if this prevented the task from starting. """ - return self._future.cancel() + return self._controller.cancel(self._future) + + def request_stop(self) -> bool: + """Ask the ewoks task to stop (cooperative). + + Supported for all backends. A queued task never starts and the result raises + :class:`TaskStoppedError`. A running task decides whether and how to stop and + the result is whatever the task returns or raises. + + :return: True if this prevented the task from starting or reached the running task. + """ + if self.done(): + return False + return self._controller.request_stop() + + 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: - """Abort a running ewoks task. + """Interrupt the ewoks task like Ctrl-C (forced, with cleanup). - This delegates to the execution controller. Depending on the execution - backend, this may communicate with another thread, process, or do - nothing for already-completed tasks. + 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 the ewoks task was aborted.""" - return self._controller.aborted() + """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 self._future.cancelled() + """Return True if :meth:`cancel` prevented the execution.""" + return self._controller.cancelled(self._future) def running(self) -> bool: return self._future.running() @@ -59,11 +127,21 @@ def result(self, timeout: Optional[float] = None) -> VariableContainer: :param timeout: Maximum number of seconds to wait for the future to complete, `None` to wait forever. :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`. """ - return 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 def output_values( self, timeout: Optional[float] = None, exclude: Collection[str] = () @@ -101,7 +179,15 @@ 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. """ - return 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 def task_exception( self, timeout: Optional[float] = None @@ -126,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 b3b216ad..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 @@ -179,11 +181,15 @@ def __init__(self, *args, **kwargs): self.__executor.started.connect(self.__on_started) self.__executor.succeeded.connect(self.__on_succeeded) self.__executor.failed.connect(self.__on_failed) + 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. @@ -926,14 +932,14 @@ def has_pending_task(self) -> bool: return bool(self.__propagate_by_future) def cancel_running_task(self) -> None: - """Abort the currently running task.""" + """Request the currently running task to stop.""" warnings.warn( - "'cancel_running_task' is deprecated since 6.0. Please cancel the task by calling the " - " `cancel` method of the future provided during task submission.", + "'cancel_running_task' is deprecated since 6.0. Please stop the task by calling the " + " `request_stop` method of the future provided during task submission.", DeprecationWarning, ) if self.__current_task_future is not None: - self.__current_task_future.abort() + self.__current_task_future.request_stop() @property def post_task_exception(self) -> Optional[Exception]: @@ -1099,3 +1105,11 @@ def __on_failed(self, task_future: TaskFuture) -> None: self.progressBarFinished() self.__propagate_by_future.pop(task_future, None) self._output_changed() + + def __on_finished(self, task_future: TaskFuture) -> None: + """ + Forget a task that ended without succeeding or failing (cancelled before it started). + + :param task_future: The future of the finished task. + """ + self.__propagate_by_future.pop(task_future, None) diff --git a/src/ewoksorange/tests/executor/signals.py b/src/ewoksorange/tests/executor/signals.py index 82e914d8..74fdbbb4 100644 --- a/src/ewoksorange/tests/executor/signals.py +++ b/src/ewoksorange/tests/executor/signals.py @@ -25,7 +25,10 @@ def __init__(self): "started": 0, "succeeded": 0, "failed": 0, + "cancelled": 0, + "stopRequested": 0, "aborted": 0, + "killed": 0, "finished": 0, "ignored": 0, } @@ -36,7 +39,10 @@ def connect(self, executor: EwoksExecutor) -> None: executor.started.connect(self._store_future("started")) executor.succeeded.connect(self._store_future("succeeded")) 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")) @@ -59,6 +65,10 @@ def callback(): def count(self, name: str) -> int: return len(self._events[name]) + def futures(self, name: str) -> List[Optional[TaskFuture]]: + """Return the futures carried by all `name` signals.""" + return list(self._events[name]) + def future(self, name: str, index: int = 0) -> TaskFuture: """Return the future carried by the `index`-th `name` signal.""" return self._events[name][index] diff --git a/src/ewoksorange/tests/executor/tasks.py b/src/ewoksorange/tests/executor/tasks.py index 254404f1..335abafb 100644 --- a/src/ewoksorange/tests/executor/tasks.py +++ b/src/ewoksorange/tests/executor/tasks.py @@ -10,7 +10,7 @@ class CustomFailure(Exception): pass -class CustomCancelled(Exception): +class CustomStopped(Exception): pass @@ -26,7 +26,7 @@ def run(self): for i in range(40): time.sleep(period) if self.stop_requested: - raise CustomCancelled(f"cancelled after {(i + 1) * period} seconds") + raise CustomStopped(f"stopped after {(i + 1) * period} seconds") if self.inputs.fail: raise CustomFailure("intentional failure") @@ -43,12 +43,12 @@ def _on_stop_requested(self): pass -class RequestCancelTask( +class EarlyStopTask( Task, input_names=["duration"], output_names=["result"], ): - """Exits early on cancel, leaving its output undefined.""" + """Exits early on a stop request, leaving its output undefined.""" def run(self): step = 0.05 @@ -65,12 +65,12 @@ def _on_stop_requested(self): pass -class PartialCancelTask( +class PartialStopTask( Task, input_names=["duration"], output_names=["first", "second"], ): - """Sets `first`, then may be cancelled before setting `second`.""" + """Sets `first`, then may be requested to stop before setting `second`.""" def run(self): self.outputs.first = "first done" @@ -90,7 +90,7 @@ def _on_stop_requested(self): pass -class IgnoreCancelTask( +class IgnoreStopTask( Task, input_names=["duration"], output_names=["result"], @@ -99,14 +99,111 @@ class IgnoreCancelTask( def run(self): time.sleep(self.inputs.duration) - self.outputs.result = "completed despite abort" + self.outputs.result = "completed despite stop request" def _on_stop_requested(self): """Nothing to do: `run()` polls `self.stop_requested`.""" pass -class InterruptCancelTask( +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"], + output_names=["result"], +): + """Never checks `self.stop_requested` and completes once the file `path` exists.""" + + def run(self): + end = time.monotonic() + 30 + while not os.path.exists(self.inputs.path): + if time.monotonic() > end: + raise TimeoutError(f"{self.inputs.path} does not exist") + time.sleep(0.01) + self.outputs.result = "completed despite stop request" + + def _on_stop_requested(self): + """Nothing to do: the stop request is ignored.""" + 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"], output_names=["result"], @@ -163,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 index 4fe2c582..6f91b882 100644 --- a/src/ewoksorange/tests/executor/test_executor_abort.py +++ b/src/ewoksorange/tests/executor/test_executor_abort.py @@ -1,155 +1,304 @@ +import os +import signal +import sys import threading +import time import pytest -from ewoksutils.exceptions import TaskExecutionError +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 + ) -from .tasks import AddTask -from .tasks import IgnoreCancelTask -from .tasks import InterruptCancelTask -from .tasks import PartialCancelTask -from .tasks import RequestCancelTask + 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() -def test_abort(ewoksorange_qtapp, executor_context_factory): - """Cancellation observed by run(): it raises, so the task fails.""" - with executor_context_factory() as (kind, executor, recorder): - inputs = {"a": 1, "b": 2, "delay": 5} - thread = None - if kind == "sync": - # submit_task() blocks until the task finishes, so it must run - # on its own thread for abort() to have anything to interrupt. - thread = threading.Thread( - target=executor.submit_task, args=(AddTask,), kwargs={"inputs": inputs} - ) - thread.start() - else: - executor.submit_task(AddTask, inputs=inputs) + with pytest.raises(TaskAbortedError, match="was aborted"): + future.result(timeout=5) + assert time.monotonic() - t0 < 5 - future = recorder.wait_future("started") + 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 - assert future.abort() - match = r"cancelled after [\.0-9]+ seconds" - with pytest.raises(TaskExecutionError, match=match): - future.result(timeout=10) +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() - recorder.wait_for("finished", 1) - recorder.assert_counts( - submitted=1, aboutToStart=1, started=1, aborted=1, failed=1, finished=1 + +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 ) - recorder.assert_failed(future, TaskExecutionError, match=match) - if thread is not None: - thread.join(timeout=10) + 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() -@pytest.mark.parametrize("task_class", [RequestCancelTask, InterruptCancelTask]) -def test_abort_leaves_outputs_undefined( - ewoksorange_qtapp, executor_context_factory, task_class -): - """Cancellation observed by run(): it returns early, so no output is set. + assert blocker.result(timeout=10)["result"].value == "completed" + recorder.wait_for("finished", 3) + assert queued.cancelled() + assert not marker.exists() - Covers both ways of reacting to a stop request (RequestCancelTask: - polls `Task.stop_requested`, InterruptCancelTask: `Task._on_stop_requested`). - """ - with executor_context_factory() as (kind, executor, recorder): - inputs = {"duration": 2} - thread = None - if kind == "sync": - thread = threading.Thread( - target=executor.submit_task, - args=(task_class,), - kwargs={"inputs": inputs}, - ) - thread.start() - else: - executor.submit_task(task_class, inputs=inputs) +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() - result = future.result(timeout=10) - assert not result["result"].has_value + 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 + ) - assert future.aborted() - recorder.wait_for("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=1, aboutToStart=1, started=1, succeeded=1, finished=1, aborted=1 + submitted=2, + aboutToStart=2, + started=2, + succeeded=1, + failed=1, + killed=1, + finished=2, ) - if thread is not None: - thread.join(timeout=10) + 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_abort_leaves_partial_outputs(ewoksorange_qtapp, executor_context_factory): - """Cancellation observed by run(): it returns after only some outputs - were set, leaving the rest undefined.""" - with executor_context_factory() as (kind, executor, recorder): - inputs = {"duration": 2} - thread = None - if kind == "sync": - thread = threading.Thread( - target=executor.submit_task, - args=(PartialCancelTask,), - kwargs={"inputs": inputs}, - ) - thread.start() - else: - executor.submit_task(PartialCancelTask, inputs=inputs) +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() - result = future.result(timeout=10) - assert result["first"].value == "first done" - assert not result["second"].has_value + recorder.wait_for("finished", 1) + recorder.assert_counts( + submitted=1, + aboutToStart=1, + started=1, + stopRequested=1, + killed=1, + failed=1, + finished=1, + ) - assert future.aborted() + +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, succeeded=1, finished=1, aborted=1 + submitted=1, aboutToStart=1, started=1, failed=1, killed=1, finished=1 ) - if thread is not None: - thread.join(timeout=10) -def test_abort_does_not_guarantee_cancellation( +@pytest.mark.skipif(sys.platform == "win32", reason="requires POSIX signals") +def test_abort_signal_ignored_by_idle_worker( ewoksorange_qtapp, executor_context_factory ): - """A task that never checks `self.stop_requested` always completes normally. + """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 - `aborted()` still reports True: it reflects that abort reached the task, - not what the task chose to do about it. - """ + +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=(IgnoreCancelTask,), + args=(IgnoreStopTask,), kwargs={"inputs": inputs}, ) thread.start() else: - executor.submit_task(IgnoreCancelTask, inputs=inputs) + executor.submit_task(IgnoreStopTask, inputs=inputs) future = recorder.wait_future("started") - assert future.abort() + assert not future.abort() + assert not future.kill() result = future.result(timeout=10) - assert result["result"].value == "completed despite abort" - - assert future.aborted() + 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, aborted=1 + 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 new file mode 100644 index 00000000..aac118a2 --- /dev/null +++ b/src/ewoksorange/tests/executor/test_executor_actions.py @@ -0,0 +1,172 @@ +from concurrent.futures import CancelledError +from typing import Dict + +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", "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}, +} + + +@pytest.mark.parametrize("action", _ACTIONS) +def test_queued_task_never_starts(ewoksorange_qtapp, executor_context_factory, action): + """Each action prevents a queued task from starting, with its own outcome.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + if kind == "sync": + pytest.skip("sync execution has no queue") + + blocker = executor.submit_task(AddTask, inputs={"a": 1, "delay": 1}) + recorder.wait_for("started") + queued = executor.submit_task(AddTask, inputs={"a": 2}) + + 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) + assert _flags(queued) == {name: name == action for name in _ACTIONS} + + expected = { + "submitted": 2, + "aboutToStart": 1, + "started": 1, + "succeeded": 1, + "finished": 1, + } + for name, count in _SIGNALS[action].items(): + expected[name] = expected.get(name, 0) + count + recorder.wait_for("finished", expected["finished"]) + recorder.assert_counts(**expected) + recorder.assert_started(blocker) + assert queued not in recorder.futures("aboutToStart") + assert queued not in recorder.futures("started") + + +@pytest.mark.parametrize("first", _ACTIONS) +def test_queued_task_first_action_wins( + ewoksorange_qtapp, executor_context_factory, first +): + """Only the first action that reaches a queued task has an effect.""" + with executor_context_factory(workers=1) as (kind, executor, recorder): + 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") + queued = executor.submit_task(AddTask, inputs={"a": 2}) + + assert getattr(queued, first)() + for action in _ACTIONS: + assert not getattr(queued, action)(), action + + blocker.result(timeout=10) + exc_type, match = _ERRORS[first] + with pytest.raises(exc_type, match=match): + queued.result(timeout=10) + 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): + future = executor.submit_task(AddTask, inputs={"a": 1}) + assert future.result(timeout=10)["result"].value == 1 + + for action in _ACTIONS: + assert not getattr(future, action)(), action + + assert future.result()["result"].value == 1 + assert _flags(future) == dict.fromkeys(_ACTIONS, False) + + recorder.wait_for("finished", 1) + recorder.assert_counts( + submitted=1, aboutToStart=1, started=1, succeeded=1, finished=1 + ) + + +def _flags(future: TaskFuture) -> Dict[str, bool]: + """Which action had an effect, according to the future.""" + return { + "cancel": future.cancelled(), + "request_stop": future.stop_requested(), + "abort": future.aborted(), + "kill": future.killed(), + } diff --git a/src/ewoksorange/tests/executor/test_executor_cancel.py b/src/ewoksorange/tests/executor/test_executor_cancel.py index cfb27658..b954ffef 100644 --- a/src/ewoksorange/tests/executor/test_executor_cancel.py +++ b/src/ewoksorange/tests/executor/test_executor_cancel.py @@ -2,6 +2,7 @@ from ...gui.concurrency.executor import SubmitPolicy from .tasks import AddTask +from .tasks import IgnoreStopTask def test_cancel_queued_task(ewoksorange_qtapp, executor_context_factory): @@ -25,13 +26,6 @@ def test_cancel_queued_task(ewoksorange_qtapp, executor_context_factory): blocker = recorder.wait_future("started") - # ProcessPoolExecutor feeds its worker's call queue ahead of time - # (EXTRA_QUEUED_CALLS), so with a single worker the first couple of - # queued submissions can already be uncancellable before they truly - # start. A few filler submissions guarantee the last one is still - # genuinely queued when it gets cancelled below. - fillers = [executor.submit_task(AddTask, inputs={"a": i}) for i in range(4)] - victim = executor.submit_task(AddTask, inputs={"a": 99}) cancelled = victim.cancel() @@ -41,22 +35,22 @@ def test_cancel_queued_task(ewoksorange_qtapp, executor_context_factory): # not merely unstarted. assert cancelled is False - n_ran = 1 + len(fillers) + 1 + n_ran = 2 else: assert cancelled is True - n_ran = 1 + len(fillers) + n_ran = 1 + n_cancelled = 2 - n_ran blocker.result(timeout=10) - for filler in fillers: - filler.result(timeout=10) - recorder.wait_for("finished", n_ran) + recorder.wait_for("finished", n_ran + n_cancelled) recorder.assert_counts( - submitted=1 + len(fillers) + 1, + submitted=2, aboutToStart=n_ran, started=n_ran, succeeded=n_ran, - finished=n_ran, + cancelled=n_cancelled, + finished=n_ran + n_cancelled, ) if thread is not None: thread.join(timeout=10) @@ -81,3 +75,34 @@ def test_cancel_after_completion(ewoksorange_qtapp, executor_context_factory): assert future.cancel() is False if thread is not None: thread.join(timeout=10) + + +def test_cancel_running_task(ewoksorange_qtapp, executor_context_factory): + """cancel() has no effect on a running task.""" + with executor_context_factory() as (kind, executor, recorder): + 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.cancel() + + result = future.result(timeout=10) + assert result["result"].value == "completed despite stop request" + assert not future.cancelled() + + 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_order.py b/src/ewoksorange/tests/executor/test_executor_order.py new file mode 100644 index 00000000..0c1eff34 --- /dev/null +++ b/src/ewoksorange/tests/executor/test_executor_order.py @@ -0,0 +1,84 @@ +import time +from typing import Callable + +import pytest + +from ...gui.concurrency.executor import Concurrency +from ...gui.concurrency.executor import EwoksExecutor +from ...gui.concurrency.executor import SubmitPolicy +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( + ewoksorange_qtapp, executor_context_factory +): + """The completion signals follow the start signals, also when the task + finishes while the GUI thread does not process its events.""" + with executor_context_factory() as (kind, executor, recorder): + if kind == "sync": + pytest.skip("sync execution emits all signals directly") + + future = executor.submit_task(AddTask, inputs={"a": 1}) + _wait_without_events(future.done) + time.sleep(0.5) # let the completion handling run in the background + + recorder.wait_for("finished", 1) + recorder.assert_order( + "submitted", "aboutToStart", "started", "succeeded", "finished" + ) + + +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) + if concurrency is Concurrency.PROCESS: + pool.close() + pool.join() + else: + pool.shutdown() + executor = EwoksExecutor(pool, SubmitPolicy.DROP_IF_BUSY) + recorder = SignalRecorder() + recorder.connect(executor) + try: + for _ in range(2): + with pytest.raises(RuntimeError): + executor.submit_task(AddTask, inputs={"a": 1}) + recorder.assert_counts() + finally: + executor.shutdown(wait=True) + + +def _wait_without_events(condition: Callable[[], bool], timeout: float = 10) -> None: + """Wait for `condition` without processing the events of the GUI thread.""" + t0 = time.monotonic() + while not condition(): + assert time.monotonic() - t0 < timeout, "timeout" + time.sleep(0.01) 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 ): diff --git a/src/ewoksorange/tests/executor/test_executor_stop.py b/src/ewoksorange/tests/executor/test_executor_stop.py new file mode 100644 index 00000000..1fc2d71b --- /dev/null +++ b/src/ewoksorange/tests/executor/test_executor_stop.py @@ -0,0 +1,177 @@ +import threading + +import pytest +from ewoksutils.exceptions import TaskExecutionError + +from .tasks import AddTask +from .tasks import EarlyStopTask +from .tasks import InterruptStopTask +from .tasks import PartialStopTask +from .tasks import WaitForFileTask + + +def test_request_stop(ewoksorange_qtapp, executor_context_factory): + """Stop request observed by run(): it raises, so the task fails.""" + with executor_context_factory() as (kind, executor, recorder): + inputs = {"a": 1, "b": 2, "delay": 5} + thread = None + if kind == "sync": + # submit_task() blocks until the task finishes, so it must run + # on its own thread for request_stop() to have anything to interrupt. + thread = threading.Thread( + target=executor.submit_task, args=(AddTask,), kwargs={"inputs": inputs} + ) + thread.start() + else: + executor.submit_task(AddTask, inputs=inputs) + + future = recorder.wait_future("started") + + assert future.request_stop() + + match = r"stopped after [\.0-9]+ seconds" + with pytest.raises(TaskExecutionError, match=match): + future.result(timeout=10) + + assert future.stop_requested() + + recorder.wait_for("finished", 1) + recorder.assert_counts( + submitted=1, + aboutToStart=1, + started=1, + stopRequested=1, + failed=1, + finished=1, + ) + recorder.assert_failed(future, TaskExecutionError, match=match) + if thread is not None: + thread.join(timeout=10) + + +@pytest.mark.parametrize("task_class", [EarlyStopTask, InterruptStopTask]) +def test_stop_leaves_outputs_undefined( + ewoksorange_qtapp, executor_context_factory, task_class +): + """Stop request observed by run(): it returns early, so no output is set. + + Covers polling `Task.stop_requested` (EarlyStopTask) and reacting + in `Task._on_stop_requested` (InterruptStopTask). + """ + with executor_context_factory() as (kind, executor, recorder): + inputs = {"duration": 2} + thread = None + if kind == "sync": + thread = threading.Thread( + target=executor.submit_task, + args=(task_class,), + kwargs={"inputs": inputs}, + ) + thread.start() + else: + executor.submit_task(task_class, inputs=inputs) + + future = recorder.wait_future("started") + + assert future.request_stop() + + result = future.result(timeout=10) + assert not result["result"].has_value + + assert future.stop_requested() + + recorder.wait_for("finished", 1) + recorder.assert_counts( + submitted=1, + aboutToStart=1, + started=1, + succeeded=1, + finished=1, + stopRequested=1, + ) + if thread is not None: + thread.join(timeout=10) + + +def test_stop_leaves_partial_outputs(ewoksorange_qtapp, executor_context_factory): + """Stop request observed by run(): it returns after only some outputs + were set, leaving the rest undefined.""" + with executor_context_factory() as (kind, executor, recorder): + inputs = {"duration": 2} + thread = None + if kind == "sync": + thread = threading.Thread( + target=executor.submit_task, + args=(PartialStopTask,), + kwargs={"inputs": inputs}, + ) + thread.start() + else: + executor.submit_task(PartialStopTask, inputs=inputs) + + future = recorder.wait_future("started") + + assert future.request_stop() + + result = future.result(timeout=10) + assert result["first"].value == "first done" + assert not result["second"].has_value + + assert future.stop_requested() + + recorder.wait_for("finished", 1) + recorder.assert_counts( + submitted=1, + aboutToStart=1, + started=1, + succeeded=1, + finished=1, + stopRequested=1, + ) + if thread is not None: + thread.join(timeout=10) + + +def test_request_stop_does_not_guarantee_stop( + ewoksorange_qtapp, executor_context_factory, tmp_path +): + """A task that never checks `self.stop_requested` always completes normally. + + `stop_requested()` still reports True: it reflects that the stop request reached the task, + not what the task chose to do about it. + """ + with executor_context_factory() as (kind, executor, recorder): + release = tmp_path / "release" + inputs = {"path": str(release)} + thread = None + if kind == "sync": + thread = threading.Thread( + target=executor.submit_task, + args=(WaitForFileTask,), + kwargs={"inputs": inputs}, + ) + thread.start() + else: + executor.submit_task(WaitForFileTask, inputs=inputs) + + future = recorder.wait_future("started") + + assert future.request_stop() + release.touch() + + result = future.result(timeout=10) + assert result["result"].value == "completed despite stop request" + + assert future.stop_requested() + + recorder.wait_for("finished", 1) + recorder.assert_counts( + submitted=1, + aboutToStart=1, + started=1, + succeeded=1, + finished=1, + stopRequested=1, + ) + if thread is not None: + thread.join(timeout=10) diff --git a/src/ewoksorange/tests/executor/test_task_future.py b/src/ewoksorange/tests/executor/test_task_future.py index 0f65d908..356ec43d 100644 --- a/src/ewoksorange/tests/executor/test_task_future.py +++ b/src/ewoksorange/tests/executor/test_task_future.py @@ -1,6 +1,7 @@ import threading from ...gui.qt_utils.app import QtEvent +from ...gui.qt_utils.app import wait_until from .tasks import AddTask @@ -50,11 +51,13 @@ def test_cancelled(ewoksorange_qtapp, executor_context_factory): thread.join(timeout=10) else: future = executor.submit_task(AddTask, inputs=inputs) - future.cancel() - recorder.wait_for("finished", 1) + cancelled = future.cancel() + assert wait_until(future.done, timeout=10) # Whether cancel() wins the race with the worker picking up the # task is not guaranteed; cancelled() must simply agree with it. - assert future.cancelled() == future.cancel() + assert future.cancelled() == cancelled + if not cancelled: + recorder.wait_for("finished", 1) def test_add_done_callback(ewoksorange_qtapp, executor_context_factory): diff --git a/src/ewoksorange/tests/owwidget/test_owwidget_cancel.py b/src/ewoksorange/tests/owwidget/test_owwidget_cancel.py new file mode 100644 index 00000000..f3628f7e --- /dev/null +++ b/src/ewoksorange/tests/owwidget/test_owwidget_cancel.py @@ -0,0 +1,33 @@ +import time + +from ewokscore.task import Task + +from ...gui.owwidgets.meta import ow_build_opts +from ...gui.owwidgets.threaded import OWEwoksWidgetWithTaskStack +from ...gui.qt_utils.app import wait_until + + +class Sleep(Task, input_names=["seconds"], output_names=["seconds"]): + def run(self): + time.sleep(self.inputs.seconds) + self.outputs.seconds = self.inputs.seconds + + +class OWSleep(OWEwoksWidgetWithTaskStack, **ow_build_opts, ewokstaskclass=Sleep): + name = "test_OW_sleep" + + +def test_cancelled_task_is_not_pending(ewoksorange_qtapp): + """A task cancelled before it starts does not stay pending in the widget.""" + widget = OWSleep() + try: + widget.set_dynamic_input("seconds", 0.5) + running = widget.execute_ewoks_task() + queued = widget.execute_ewoks_task() + + assert queued.cancel() + running.result(timeout=10) + + assert wait_until(lambda: not widget.has_pending_task(), timeout=10) + finally: + widget.onDeleteWidget() diff --git a/src/ewoksorange/tests/owwidget/test_owwidget_drop.py b/src/ewoksorange/tests/owwidget/test_owwidget_drop.py index ff96e889..6fca9713 100644 --- a/src/ewoksorange/tests/owwidget/test_owwidget_drop.py +++ b/src/ewoksorange/tests/owwidget/test_owwidget_drop.py @@ -6,6 +6,7 @@ from ...gui.owwidgets.meta import ow_build_opts from ...gui.owwidgets.threaded import OWEwoksWidgetOneThread +from ...gui.qt_utils.app import wait_until class Sequential( @@ -27,34 +28,31 @@ def test_owwidget_drop(ewoksorange_qtapp): widget = OWSequential() sleep_seconds = 0.5 - values = [0, 1, 2, 3] + futures = {} - futures = [] - for value in values: + def submit(value): widget.set_dynamic_input("value", value) widget.set_dynamic_input("sleep", sleep_seconds) - futures.append(widget.execute_ewoks_task()) - time.sleep(sleep_seconds * 0.6) - - results = {} - thread_ids = {} - - for future, value in zip(futures, values): - if future is None: - results[value] = None - thread_ids[value] = None - else: - outputs = future.result(timeout=10) - value = outputs["value"].value - results[value] = outputs["time"].value - thread_ids[value] = outputs["thread_id"].value - - completed = [v for v in values if results[v] is not None] - dropped = [v for v in values if results[v] is None] - times = [results[v] for v in completed] - in_threads = [thread_ids[v] != threading.get_ident() for v in completed] - + futures[value] = widget.execute_ewoks_task() + + submit(0) + submit(1) # busy with 0 + futures[0].result(timeout=10) + assert wait_until(lambda: not widget.has_pending_task(), timeout=10) + submit(2) + submit(3) # busy with 2 + futures[2].result(timeout=10) + + completed = [v for v, future in futures.items() if future is not None] + dropped = [v for v, future in futures.items() if future is None] assert completed == [0, 2], "not in order" assert dropped == [1, 3], "not in order" + + outputs = [futures[v].result(timeout=10) for v in completed] + times = [output["time"].value for output in outputs] + in_threads = [ + output["thread_id"].value != threading.get_ident() for output in outputs + ] + assert (numpy.diff(times) >= sleep_seconds).all(), "not sequential" assert all(in_threads), "in main thread" diff --git a/src/ewoksorange/tests/owwidget/test_owwidget_parallel.py b/src/ewoksorange/tests/owwidget/test_owwidget_parallel.py index fc44dae2..fea5c6f9 100644 --- a/src/ewoksorange/tests/owwidget/test_owwidget_parallel.py +++ b/src/ewoksorange/tests/owwidget/test_owwidget_parallel.py @@ -10,7 +10,7 @@ from ...gui.owwidgets.threaded import OWEwoksWidgetOneThreadPerRun -class Cancelled(Exception): +class Stopped(Exception): def __init__(self, value, tm) -> None: self.value = value self.time = tm @@ -26,14 +26,14 @@ def run(self): # Wait for all parallel task to start self.inputs.synchronize_run.wait(timeout=10) - # Provide enough time to cancel the task + # Provide enough time to request the task to stop time.sleep(self.inputs.sleep) tm = time.perf_counter() # Return value as output or exception if self.stop_requested: - raise Cancelled(self.inputs.value, tm) + raise Stopped(self.inputs.value, tm) self.outputs.time = tm self.outputs.value = self.inputs.value @@ -55,39 +55,43 @@ def test_owwidget_parallel(ewoksorange_qtapp): sleep_seconds = 0.5 values = [0, 1, 2, 3] - cancels = [False, True, False, False] - synchronize_run = Barrier(len(values)) + stops = [False, True, False, False] + # The test joins the barrier to request the stop once all tasks run: + # a task that is requested to stop before it starts never starts. + synchronize_run = Barrier(len(values) + 1) # all jobs start when the last run method is called futures = [] - for value, cancel in zip(values, cancels): + for value, stop in zip(values, stops): widget.set_dynamic_input("value", value) widget.set_dynamic_input("sleep", sleep_seconds) widget.set_dynamic_input("synchronize_run", synchronize_run) futures.append(widget.execute_ewoks_task()) - # cancel some jobs - for future, cancel in zip(futures, cancels): - if cancel: - _ = future.abort() + synchronize_run.wait(timeout=10) + + # request some jobs to stop + for future, stop in zip(futures, stops): + if stop: + _ = future.request_stop() # check results times = [] - for future, value, cancel in zip(futures, values, cancels): - if cancel: + for future, value, stop in zip(futures, values, stops): + if stop: with pytest.raises(TaskExecutionError) as exc_info: _ = future.result(timeout=10) - assert future.aborted() + assert future.stop_requested() - assert isinstance(exc_info.value.__cause__, Cancelled) + assert isinstance(exc_info.value.__cause__, Stopped) assert exc_info.value.__cause__.value == value times.append(exc_info.value.__cause__.time) else: outputs = future.result(timeout=10) - assert not future.aborted() + assert not future.stop_requested() assert outputs["value"].value == value times.append(outputs["time"].value) diff --git a/src/ewoksorange_pytest_plugin/qtapp_teardown.py b/src/ewoksorange_pytest_plugin/qtapp_teardown.py index 1923a9be..ebcfca46 100644 --- a/src/ewoksorange_pytest_plugin/qtapp_teardown.py +++ b/src/ewoksorange_pytest_plugin/qtapp_teardown.py @@ -34,6 +34,28 @@ def _collect_garbage(app) -> None: app.processEvents() +def _delete_python_owned_qobjects(app) -> None: + """Delete the remaining Qt objects that are owned by Python and have no parent.""" + from AnyQt import sip + from AnyQt.QtCore import QObject + + # Otherwise PyQt destroys them while the interpreter exits, which crashes + # randomly on Windows. + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + for obj in gc.get_objects(): + # No attribute access on `obj`: some objects do not allow it, + # e.g. `isinstance` accesses `__class__`. + if not issubclass(type(obj), QObject) or obj is app: + continue + if sip.isdeleted(obj) or not sip.ispyowned(obj): + continue + if QObject.parent(obj) is not None: + continue + sip.delete(obj) + app.processEvents() + + def _warn_qtwidgets_alive() -> None: from ewoksorange.gui.qt_utils.app import get_all_qtwidgets @@ -49,6 +71,7 @@ def ewoksorange_qtapp_teardown(app) -> None: """ # Called in reverse order, last one first. with ExitStack() as stack: + stack.callback(_delete_python_owned_qobjects, app) stack.callback(_warn_qtwidgets_alive) stack.callback(_collect_garbage, app) stack.callback(_global_cleanup_pytest)