Repository navigation
Execute subprocess tasks with pebble, add TaskFuture.abort and TaskFuture.kill - #501
Conversation
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
Task.request_stop and Task.stop_requestedbd36fbe to
b30ecf5
Compare
625bc0f to
3d4fbd7
Compare
f91d853 to
f720b36
Compare
3ef0155 to
94942ee
Compare
|
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. 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: |
4ba8ca5 to
459b111
Compare
|
@payno Thanks, fair point. I split it into a stack of PRs, each based on the previous one. Merge them in order:
🤖 Generated with Claude Code |
0245b75 to
1a84e82
Compare
459b111 to
566e9c2
Compare
payno
left a comment
There was a problem hiding this comment.
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: |
There was a problem hiding this comment.
_execute_task seems more appropriate as we are doing the distinction.
| def _run_task(self, task: Task) -> None: | |
| def _execute_task(self, task: Task) -> None: |
There was a problem hiding this comment.
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") |
There was a problem hiding this comment.
I would expect we create a dedicated TaskKilledAborted
There was a problem hiding this comment.
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(): |
There was a problem hiding this comment.
| 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(): |
| 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 |
There was a problem hiding this comment.
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)
raiseThis is a bit better IMO having clear name. But not sure this is a game changer.
There was a problem hiding this comment.
Done, split into _sigint_raises_keyboard_interrupt and _raise_missed_abort.
| 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) |
There was a problem hiding this comment.
Maybe this could clarify:
| 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) |
There was a problem hiding this comment.
Swapped the branches and named the case in the comment. Testing self._pending directly avoids the variable and the assert.
| pass | ||
|
|
||
|
|
||
| class StubbornTask( |
There was a problem hiding this comment.
I had too look for Stubborn 😆 maybe obstinate would be better for people who aren’t perfectly bilingual
| class StubbornTask( | |
| class ObstinateTask( |
There was a problem hiding this comment.
I know stubborn but not obstinate. Is that french?
There was a problem hiding this comment.
Renamed to SurviveAbortTask, like the other test tasks named after what they do (IgnoreStopTask, SurviveAbortProgressTask).
There was a problem hiding this comment.
AhAh indeed for obstinate we have the same world in french. There was a bias. 👍 for SurviveAbortProgressTask
566e9c2 to
4c28df0
Compare
1a84e82 to
04e56ab
Compare
4c28df0 to
baced81
Compare
04e56ab to
d8ae60b
Compare
…kFuture.kill Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
baced81 to
3a7c641
Compare
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.pySo to me the scope was clear and fully defined but yes, still complex to implement. |
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>
Stack (split from this PR, merge in order):
PR summary
pebble.ProcessPool(new dependency) instead ofProcessPoolExecutor: a killed worker no longer breaks the pool.EwoksExecutorraisesTypeErrorfor aconcurrent.futures.ProcessPoolExecutor.OWEwoksBaseWidget._create_pool_executormay return apebble.ProcessPool.pebble.ProcessPoolworker.TaskFuture.abort(): interrupt like Ctrl-C. The task can clean up and the worker survives. Subprocess only.TaskFuture.kill(): terminate the worker. No cleanup, the pool replaces the worker. Subprocess only.kill()on a queued task completes its future right away.cancelandrequest_stop).result()raisesTaskAbortedError.abortedandkilledsignals, andTaskFuture.aborted()/killed().cancel,request_stopandabortdo not cancel thepebblefuture: cancelling it kills the worker if one just picked up the task, which would turn them into akill.cancelledandfinished.Try-out cancel vs. stop vs. abort vs. kill with
Click Submit x5 (Process parallel) followed by cancel, stop, abort or kill
AI Disclosure
🤖 Generated with Claude Code