Skip to content

WIP: Implement option to force kill #288

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
agoscinski wants to merge 5 commits into from
Closed

WIP: Implement option to force kill #288

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
07c30db
WIP: Implement option to allow force kill
agoscinski Sep 27, 2024
00fe7b5
Implementation with minimal changes
khsrali Oct 2, 2024
f1f8095
Merge branch 'master' into force-kill
unkcpz Dec 15, 2024
b9dd887
Merge branch 'master' into force-kill
unkcpz Jan 8, 2025
4a462a0
Adapt after merge the message protocol PR
unkcpz Jan 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions src/plumpy/process_states.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
'Continue',
'Interruption',
'KillInterruption',
'ForceKillInterruption',
'PauseInterruption',
]

Expand All @@ -51,6 +52,10 @@ class KillInterruption(Interruption):
pass


class ForceKillInterruption(Interruption):
pass


class PauseInterruption(Interruption):
pass

Expand Down
16 changes: 12 additions & 4 deletions src/plumpy/processes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1067,7 +1067,7 @@ def _create_interrupt_action(self, exception: process_states.Interruption) -> fu
do_pause = functools.partial(self._do_pause, str(exception))
return futures.CancellableAction(do_pause, cookie=exception)

if isinstance(exception, process_states.KillInterruption):
if isinstance(exception, (process_states.KillInterruption, process_states.ForceKillInterruption)):

def do_kill(_next_state: process_states.State) -> Any:
try:
Expand Down Expand Up @@ -1131,6 +1131,8 @@ def kill(self, msg: Union[str, None] = None) -> Union[bool, asyncio.Future]:
Kill the process
:param msg: An optional kill message
"""
force_kill = isinstance(msg, str) and '-F' in msg
Copy link
Collaborator

@sphuber sphuber Oct 8, 2024

Choose a reason for hiding this comment

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

Not a big fan of this approach. Why not add a force_kill: bool = False argument to the kill method. From aiida-core you can then pass that in the Communicator.rpc_send method

Copy link
Contributor

@khsrali khsrali Oct 9, 2024

Choose a reason for hiding this comment

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

Thanks a lot @sphuber.
Yes yes, I thought about it, but then we have to modify many functions to pass force_kill hand in hand or even to be compatible with dict.
So far, apparently in plumpy the msg is expected to be a str, if I'm not mistaken.
I tried to keep changes minimal.. although maybe not the best approach

Copy link
Member

@unkcpz unkcpz Nov 27, 2024

Choose a reason for hiding this comment

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

I think @sphuber has a good point. If the message content is checked inside the implementation, there is no way for user to know how to call it from looking at the function signature.

It is the function:

    _perform_actions(processes, controller.kill_process, 'kill', 'killing', timeout, wait, msg=message)

who send the rpc message, I think it is even better to change it to, and then can also pass the force_kill

    _perform_actions(processes, functool.partial(controller.kill_process, msg=message), 'kill', 'killing', timeout, wait)

Copy link
Member

@unkcpz unkcpz Dec 1, 2024

Choose a reason for hiding this comment

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

In #291, I try to make massage not a global variable so it is able to carry more information. After that it will be straight forward to have force option for the kill method.
Discuss with @khsrali offline, he prefer go with the current change and I'll open an amend PR after #291.

@khsrali Just add unit test for this?

Copy link
Collaborator

@sphuber sphuber Dec 1, 2024

Choose a reason for hiding this comment

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

What is the point of merging this if we agree it is not a good design and we plan on changing it straight after anyway?

Copy link
Member

@unkcpz unkcpz Dec 2, 2024

Choose a reason for hiding this comment

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

I think after #291 it is straightforward to pass force option to the kill_process. I leave a port force=False for this purpose. @khsrali can you give it a look?

Copy link
Contributor

@khsrali khsrali Dec 4, 2024

Choose a reason for hiding this comment

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

What is the point of merging this if we agree it is not a good design and we plan on changing it straight after anyway?

Well, it would deliver faster to users.

Copy link
Member

@unkcpz unkcpz Dec 4, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

Well, it would deliver faster to users.

I may not agree with this.
Finding workarounds should not be the solution to get features fast deliver to users. The fast deliver is based on developers can quickly reach the agreement. In the case of this PR, the agreement is I'll do refactoring before or an amend fix after, on the message encapsulating part.


if self.state == process_states.ProcessState.KILLED:
# Already killed
return True
Expand All @@ -1139,14 +1141,20 @@ def kill(self, msg: Union[str, None] = None) -> Union[bool, asyncio.Future]:
# Can't kill
return False

if self._killing:
if self._killing and not force_kill:
# Already killing
return self._killing

if self._stepping:
if force_kill:
# Skip interrupting the state and go straight to killed
interrupt_exception = process_states.ForceKillInterruption(msg)
self._killing = self._interrupt_action
self._state.interrupt(interrupt_exception)

elif self._stepping:
# Ask the step function to pause by setting this flag and giving the
# caller back a future
interrupt_exception = process_states.KillInterruption(msg)
interrupt_exception = process_states.KillInterruption(msg) # type: ignore
self._set_interrupt_action_from_exception(interrupt_exception)
self._killing = self._interrupt_action
self._state.interrupt(interrupt_exception)
Expand Down