Skip to content

Execute subprocess tasks with pebble, add TaskFuture.abort and TaskFuture.kill - #501

Merged
woutdenolf merged 7 commits into
mainfrom
use_task_request_stop
Oct 24, 2026
Merged

woutdenolf merged 7 commits into
mainfrom
use_task_request_stop

Conversation

@woutdenolf

@woutdenolf woutdenolf commented Oct 14, 2026 •

Copy link
Copy Markdown
Member

Stack (split from this PR, merge in order):

  1. Qt test fixes: drop test_owwidget_drop timing dependency, fix Windows crash at exit #506 Qt test fixes: drop test_owwidget_drop timing dependency, fix Windows crash at exit
  2. Rename TaskFuture.abort/aborted to request_stop/stop_requested #507 Rename TaskFuture.abort/aborted to request_stop/stop_requested
  3. Never start a queued task that is cancelled or stopped; two queues for the process backend #508 Never start a queued task that is cancelled or stopped; two queues for the process backend
  4. EwoksExecutor: completion signals after start signals, finished for every execution #509 EwoksExecutor: completion signals after start signals, finished for every execution
  5. Execute subprocess tasks with pebble, add TaskFuture.abort and TaskFuture.kill #501 Execute subprocess tasks with pebble, add TaskFuture.abort and TaskFuture.kill 👈 this PR
  6. Doc: executor contract, backend diagrams and glossary #510 Doc: executor contract, backend diagrams and glossary

PR summary

  • Execute subprocess tasks with pebble.ProcessPool (new dependency) instead of ProcessPoolExecutor: a killed worker no longer breaks the pool.
    • EwoksExecutor raises TypeError for a concurrent.futures.ProcessPoolExecutor.
    • OWEwoksBaseWidget._create_pool_executor may return a pebble.ProcessPool.
    • Orange-Spectroscopy uses pebble for the same reason: Peak Fit kills a running fit in a pebble.ProcessPool worker.
  • Add TaskFuture.abort(): interrupt like Ctrl-C. The task can clean up and the worker survives. Subprocess only.
    • The worker installs its SIGINT handler only while the task executes. Elsewhere, pebble's workers ignore SIGINT.
    • While the task executes, its progress goes through a sender thread in the worker (at most 100 pending values), so an abort never interrupts a queue call. The progress an aborted task did not send yet is dropped.
  • Add TaskFuture.kill(): terminate the worker. No cleanup, the pool replaces the worker. Subprocess only.
    • kill() on a queued task completes its future right away.
  • Both also prevent a queued task from starting (same start gate as cancel and request_stop). result() raises TaskAbortedError.
  • New aborted and killed signals, and TaskFuture.aborted() / killed().
  • cancel, request_stop and abort do not cancel the pebble future: cancelling it kills the worker if one just picked up the task, which would turn them into a kill.
  • A submission that fails after the pool accepted the task never starts it and ends with cancelled and finished.
  • Executor demo script: cancel, request stop, abort and kill buttons.

Try-out cancel vs. stop vs. abort vs. kill with

python scripts/ewoks_executor_demo.py --task-duration 5
Click Submit x5 (Process parallel) followed by cancel, stop, abort or kill
# Submit x5 (Process parallel) + cancel
[   4.458s] [P PARALLEL] Submitted -> 140536643910208
[   4.469s] [P PARALLEL] Submitted -> 140536643993728
[   4.481s] [P PARALLEL] Submitted -> 140536643995120
[   4.492s] [P PARALLEL] Submitted -> 140536643996608
[   4.502s] [P PARALLEL] Submitted -> 140536643998240
[   4.502s] [P PARALLEL] ▶ Started -> 140536643910208
[   4.503s] [P PARALLEL] ▶ Started -> 140536643993728
[   4.503s] [P PARALLEL] ▶ Started -> 140536643995120
[   4.503s] [P PARALLEL] ▶ Started -> 140536643996608
[   5.680s] Cancel (queued only): accepted by 1/5 tasks
[   9.486s] [P PARALLEL] ✅ 140536643910208 -> {'result': 2}
[   9.536s] [P PARALLEL] ■ Finished -> 140536643910208
[   9.539s] [P PARALLEL] ✅ 140536643993728 -> {'result': 4}
[   9.540s] [P PARALLEL] ■ Finished -> 140536643993728
[   9.541s] [P PARALLEL] ✅ 140536643995120 -> {'result': 6}
[   9.541s] [P PARALLEL] ■ Finished -> 140536643995120
[   9.543s] [P PARALLEL] ✅ 140536643996608 -> {'result': 8}
[   9.544s] [P PARALLEL] ■ Finished -> 140536643996608
[   9.544s] [P PARALLEL] 🚫 Cancelled -> 140536643998240
[   9.646s] [P PARALLEL] ■ Finished -> 140536643998240

# Submit x5 (Process parallel) + stop
[  12.566s] [P PARALLEL] Submitted -> 140536643997520
[  12.580s] [P PARALLEL] Submitted -> 140536644000592
[  12.597s] [P PARALLEL] Submitted -> 140536644002080
[  12.611s] [P PARALLEL] Submitted -> 140536644003568
[  12.622s] [P PARALLEL] Submitted -> 140536644005008
[  12.623s] [P PARALLEL] ▶ Started -> 140536643997520
[  12.625s] [P PARALLEL] ▶ Started -> 140536644000592
[  12.626s] [P PARALLEL] ▶ Started -> 140536644002080
[  12.626s] [P PARALLEL] ▶ Started -> 140536644003568
[  14.051s] Request stop: accepted by 5/5 tasks
[  14.078s] [P PARALLEL] ⏹ Stop requested -> 140536644002080
[  14.105s] [P PARALLEL] ❌ 140536644002080 -> Execution failed for ewoks task (id: None, task: '__main__.SumTask'): Stopped on request
[  14.106s] [P PARALLEL] ■ Finished -> 140536644002080
[  14.108s] [P PARALLEL] ⏹ Stop requested -> 140536644003568
[  14.109s] [P PARALLEL] ❌ 140536644003568 -> Execution failed for ewoks task (id: None, task: '__main__.SumTask'): Stopped on request
[  14.109s] [P PARALLEL] ■ Finished -> 140536644003568
[  14.110s] [P PARALLEL] ⏹ Stop requested -> 140536643997520
[  14.110s] [P PARALLEL] ❌ 140536643997520 -> Execution failed for ewoks task (id: None, task: '__main__.SumTask'): Stopped on request
[  14.110s] [P PARALLEL] ■ Finished -> 140536643997520
[  14.111s] [P PARALLEL] ⏹ Stop requested -> 140536644005008
[  14.111s] [P PARALLEL] ❌ 140536644005008 -> Task SumTask was stopped before it started
[  14.111s] [P PARALLEL] ■ Finished -> 140536644005008
[  14.114s] [P PARALLEL] ⏹ Stop requested -> 140536644000592
[  14.114s] [P PARALLEL] ❌ 140536644000592 -> Execution failed for ewoks task (id: None, task: '__main__.SumTask'): Stopped on request
[  14.114s] [P PARALLEL] ■ Finished -> 140536644000592

# Submit x5 (Process parallel) + abort
[  18.367s] [P PARALLEL] Submitted -> 140536644177840
[  18.379s] [P PARALLEL] Submitted -> 140536644179616
[  18.401s] [P PARALLEL] Submitted -> 140536644181152
[  18.411s] [P PARALLEL] Submitted -> 140536644182736
[  18.423s] [P PARALLEL] Submitted -> 140536644184320
[  18.424s] [P PARALLEL] ▶ Started -> 140536644177840
[  18.425s] [P PARALLEL] ▶ Started -> 140536644179616
[  18.425s] [P PARALLEL] ▶ Started -> 140536644181152
[  18.425s] [P PARALLEL] ▶ Started -> 140536644182736
[  20.026s] Abort (process only): accepted by 5/5 tasks
[  20.030s] [P PARALLEL] ⛔ Aborted -> 140536644179616
[  20.031s] [P PARALLEL] ❌ 140536644179616 -> Task SumTask was aborted
[  20.032s] [P PARALLEL] ■ Finished -> 140536644179616
[  20.041s] [P PARALLEL] ⛔ Aborted -> 140536644182736
[  20.042s] [P PARALLEL] ❌ 140536644182736 -> Task SumTask was aborted
[  20.044s] [P PARALLEL] ■ Finished -> 140536644182736
[  20.045s] [P PARALLEL] ⛔ Aborted -> 140536644177840
[  20.045s] [P PARALLEL] ❌ 140536644177840 -> Task SumTask was aborted
[  20.049s] [P PARALLEL] ■ Finished -> 140536644177840
[  20.052s] [P PARALLEL] ⛔ Aborted -> 140536644181152
[  20.052s] [P PARALLEL] ❌ 140536644181152 -> Task SumTask was aborted
[  20.053s] [P PARALLEL] ■ Finished -> 140536644181152
[  20.054s] [P PARALLEL] ⛔ Aborted -> 140536644184320
[  20.054s] [P PARALLEL] ❌ 140536644184320 -> Task SumTask was aborted before it started
[  20.055s] [P PARALLEL] ■ Finished -> 140536644184320

# Submit x5 (Process parallel) + kill
[  23.359s] [P PARALLEL] Submitted -> 140536644181488
[  23.377s] [P PARALLEL] Submitted -> 140536644185376
[  23.390s] [P PARALLEL] Submitted -> 140536644006208
[  23.413s] [P PARALLEL] Submitted -> 140536644183696
[  23.425s] [P PARALLEL] Submitted -> 140536644181056
[  23.426s] [P PARALLEL] ▶ Started -> 140536644181488
[  23.427s] [P PARALLEL] ▶ Started -> 140536644185376
[  23.427s] [P PARALLEL] ▶ Started -> 140536644006208
[  23.429s] [P PARALLEL] ▶ Started -> 140536644183696
[  25.652s] Kill (process only): accepted by 5/5 tasks
[  25.653s] [P PARALLEL] 💀 Killed -> 140536644181488
[  25.654s] [P PARALLEL] ❌ 140536644181488 -> The task was killed
[  25.654s] [P PARALLEL] ■ Finished -> 140536644181488
[  25.655s] [P PARALLEL] 💀 Killed -> 140536644185376
[  25.655s] [P PARALLEL] ❌ 140536644185376 -> The task was killed
[  25.656s] [P PARALLEL] ■ Finished -> 140536644185376
[  25.656s] [P PARALLEL] 💀 Killed -> 140536644006208
[  25.657s] [P PARALLEL] ❌ 140536644006208 -> The task was killed
[  25.657s] [P PARALLEL] ■ Finished -> 140536644006208
[  25.658s] [P PARALLEL] 💀 Killed -> 140536644183696
[  25.658s] [P PARALLEL] ❌ 140536644183696 -> The task was killed
[  25.659s] [P PARALLEL] ■ Finished -> 140536644183696
[  25.659s] [P PARALLEL] 💀 Killed -> 140536644181056
[  25.660s] [P PARALLEL] ❌ 140536644181056 -> The task was killed before it started
[  25.660s] [P PARALLEL] ■ Finished -> 140536644181056
image

AI Disclosure

  • AI tools claude used for generate the code

🤖 Generated with Claude Code

@codecov

codecov Bot commented Oct 14, 2026 •

Copy link
Copy Markdown

@woutdenolf woutdenolf changed the title Use Task.request_stop and Task.stop_requested Use the ewokscore stop API and rename abort to stop Oct 14, 2026
@woutdenolf woutdenolf changed the title Use the ewokscore stop API and rename abort to stop Use the ewokscore stop API, rename abort to request_stop and add a real abort Oct 14, 2026
@woutdenolf
woutdenolf force-pushed the use_task_request_stop branch 3 times, most recently from bd36fbe to b30ecf5 Compare October 14, 2026 04:49
@woutdenolf
woutdenolf marked this pull request as draft October 14, 2026 05:24
@woutdenolf woutdenolf changed the title Use the ewokscore stop API, rename abort to request_stop and add a real abort Use the ewokscore stop API, execute subprocess tasks with pebble, add abort and kill Oct 14, 2026
@woutdenolf
woutdenolf force-pushed the use_task_request_stop branch 6 times, most recently from 625bc0f to 3d4fbd7 Compare October 14, 2026 06:10
@woutdenolf
woutdenolf force-pushed the use_task_request_stop branch 2 times, most recently from f91d853 to f720b36 Compare October 14, 2026 15:06
@woutdenolf woutdenolf changed the title Use the ewokscore stop API, execute subprocess tasks with pebble, add abort and kill Rename TaskFuture.abort to request_stop, execute subprocess tasks with pebble, add abort and kill Oct 14, 2026
@woutdenolf
woutdenolf force-pushed the use_task_request_stop branch 11 times, most recently from 3ef0155 to 94942ee Compare October 15, 2026 06:26
@payno

payno commented Oct 19, 2026

Copy link
Copy Markdown
Member

I have trouble reviewing the PR to be honest (I am a human 🤣 ). Even with the tip.

This is too massive for me. I tried twice and even if a lot of it is tests I fail to review.
I understand the idea of moving to pebble but there is a lot of details I can't really properly follow because the changes touch several areas at once.

the first commit already reworks several things. Could we split this PR into several smaller ones (that can be on top of each others) - AI can probably do it ?

Here is a suggestion:

  • One with qt test fixes stuff (8b5f185, 62a026d)
  • One with renaiming: 329efdf
  • One for adding the startGate and the two queues (controller→runner and runner→controller.).
  • Moving to pebble (and potentially the kill)

@woutdenolf woutdenolf added this to the v6.0.0 milestone Oct 19, 2026
@woutdenolf
woutdenolf force-pushed the use_task_request_stop branch from 4ba8ca5 to 459b111 Compare October 21, 2026 13:03
@woutdenolf woutdenolf changed the title Rename TaskFuture.abort to request_stop, execute subprocess tasks with pebble, add abort and kill Execute subprocess tasks with pebble, add TaskFuture.abort and TaskFuture.kill Oct 21, 2026
@woutdenolf
woutdenolf changed the base branch from main to executor_signal_order October 21, 2026 13:04
@woutdenolf

woutdenolf commented Oct 21, 2026 •

Copy link
Copy Markdown
Member Author

@payno Thanks, fair point. I split it into a stack of PRs, each based on the previous one. Merge them in order:

  1. Qt test fixes: drop test_owwidget_drop timing dependency, fix Windows crash at exit #506 Qt test fixes (test_owwidget_drop, Windows crash at exit)
  2. Rename TaskFuture.abort/aborted to request_stop/stop_requested #507 Rename abort/aborted → request_stop/stop_requested (rename only)
  3. Never start a queued task that is cancelled or stopped; two queues for the process backend #508 Start gate + the two queues (controller → runner, runner → controller), still with ProcessPoolExecutor
  4. EwoksExecutor: completion signals after start signals, finished for every execution #509 Executor signal order (cancelled, stopRequested, finished for every execution)
  5. Execute subprocess tasks with pebble, add TaskFuture.abort and TaskFuture.kill #501 (👈 this PR, now based on EwoksExecutor: completion signals after start signals, finished for every execution #509) Move to pebble + abort + kill
  6. Doc: executor contract, backend diagrams and glossary #510 Docs: executor contract, diagrams and glossary

🤖 Generated with Claude Code

@woutdenolf
woutdenolf force-pushed the executor_signal_order branch from 0245b75 to 1a84e82 Compare October 22, 2026 08:55
@woutdenolf
woutdenolf force-pushed the use_task_request_stop branch from 459b111 to 566e9c2 Compare October 22, 2026 08:55

@payno payno left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Personally, this is about the limit of complexity I steel feel comfortable maintaining. Without AI assistance, it would already be challenging. But I am also part of the 'experienced' developer. Not sure it would be the case of everyone.

I'm still following the principle "Don't build what you can't maintain without AI," which is why I want to flag this. Going forward, any additional features in this area should be discussed upstream first.

To be clear, this feature is fine: we agreed on it during the workflow meeting. But I'd like to avoid adding more features of this complexity, as it would make maintenance very difficult in the years to come.


return task.output_variables

def _run_task(self, task: Task) -> None:

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_execute_task seems more appropriate as we are doing the distinction.

Suggested change
def _run_task(self, task: Task) -> None:
def _execute_task(self, task: Task) -> None:

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Renamed to _execute_task.

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")

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would expect we create a dedicated TaskKilledAborted

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added TaskKilledError, a subclass of TaskAbortedError, so catching TaskAbortedError still catches both.

if message is ToRunner.REQUEST_STOP:
stop_event.set()
elif message is ToRunner.ABORT:
if not abort_sent.is_set():

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if not abort_sent.is_set():
# `abort_sent` cannot be set twice because send SIGINT is send to the main thread (raises `KeyboardInterrupt` for python). Raising it twice would be unpredictable.
if not abort_sent.is_set():

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added, reworded.

Comment on lines +106 to +121
def _run_task(self, task: Task) -> None:
# SIGINT only interrupts the task, not the runner. Outside the task, the
# pebble worker ignores SIGINT.
previous = signal.signal(signal.SIGINT, _raise_keyboard_interrupt)
try:
try:
if self._abort_sent is not None and self._abort_sent.is_set():
# The SIGINT arrived before the handler was installed and was ignored.
raise KeyboardInterrupt
super()._run_task(task)
finally:
signal.signal(signal.SIGINT, previous)
except KeyboardInterrupt:
# The SIGINT can also interrupt the restore above.
signal.signal(signal.SIGINT, previous)
raise

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This function was hard to understand for me. And this is kind of key. But not sure to have something better to propose. AI propose context manager like:

    def _run_task(self, task: Task) -> None:
        with _sigint_raises_keyboard_interrupt():
            self._raise_missed_abort()
            super()._run_task(task)

    def _raise_missed_abort(self) -> None:
        """Interrupt the task when the SIGINT of an abort arrived before it was
        handled: the pebble worker ignored it."""
        if self._abort_sent is not None and self._abort_sent.is_set():
            raise KeyboardInterrupt


@contextmanager
def _sigint_raises_keyboard_interrupt() -> Generator[None, None, None]:
    """SIGINT raises `KeyboardInterrupt` inside the context only. Outside, the
    pebble worker ignores SIGINT."""
    previous = signal.signal(signal.SIGINT, _raise_keyboard_interrupt)
    try:
        try:
            yield
        finally:
            signal.signal(signal.SIGINT, previous)
    except KeyboardInterrupt:
        # A pending SIGINT can interrupt the restore above: restore again.
        signal.signal(signal.SIGINT, previous)
        raise

This is a bit better IMO having clear name. But not sure this is a game changer.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done, split into _sigint_raises_keyboard_interrupt and _raise_missed_abort.

Comment on lines +97 to +102
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)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe this could clarify:

Suggested change
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)
sender_thread_running = self._pending is not None
if sender_thread_running:
assert self._pending is not None
# 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)
else:
self._put(self._progress)

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Swapped the branches and named the case in the comment. Testing self._pending directly avoids the variable and the assert.

Comment thread src/ewoksorange/tests/executor/tasks.py Outdated
pass


class StubbornTask(

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had too look for Stubborn 😆 maybe obstinate would be better for people who aren’t perfectly bilingual

Suggested change
class StubbornTask(
class ObstinateTask(

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I know stubborn but not obstinate. Is that french?

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Renamed to SurviveAbortTask, like the other test tasks named after what they do (IgnoreStopTask, SurviveAbortProgressTask).

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AhAh indeed for obstinate we have the same world in french. There was a bias. 👍 for SurviveAbortProgressTask

@woutdenolf
woutdenolf force-pushed the use_task_request_stop branch from 566e9c2 to 4c28df0 Compare October 24, 2026 08:21
@woutdenolf
woutdenolf force-pushed the executor_signal_order branch from 1a84e82 to 04e56ab Compare October 24, 2026 08:21
@woutdenolf
woutdenolf force-pushed the use_task_request_stop branch from 4c28df0 to baced81 Compare October 24, 2026 08:23
@woutdenolf
woutdenolf force-pushed the executor_signal_order branch from 04e56ab to d8ae60b Compare October 24, 2026 08:23
…kFuture.kill

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@woutdenolf
woutdenolf force-pushed the use_task_request_stop branch from baced81 to 3a7c641 Compare October 24, 2026 08:31
@woutdenolf

Copy link
Copy Markdown
Member Author

But I'd like to avoid adding more features of this complexity

Yes the "executor with Qt signals" and cancel/stop/abort/kill support is complex.

The demo script covers the full extend of the features

python scripts/ewoks_executor_demo.py

So to me the scope was clear and fully defined but yes, still complex to implement.

@woutdenolf
woutdenolf changed the base branch from executor_signal_order to main October 24, 2026 08:47
woutdenolf and others added 6 commits October 24, 2026 10:50
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@woutdenolf
woutdenolf enabled auto-merge October 24, 2026 09:01
@woutdenolf
woutdenolf merged commit 0d76732 into main Oct 24, 2026
15 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants