On Stopping Dataflows

[Mivik]

The Issue

The current dora stop command (which uses ControlRequest::Stop) depends on the Event::Stop mechanism to function. While this allows developers to exit gracefully, there are numerous instances where the Event::Stop signal goes unhandled.

In these unhandled cases, the daemon is designed to kill the process after the grace_duration expires. Unfortunately, this force kill rarely works. The process simply hangs, preventing the dataflow from being properly stopped.

Root Cause Analysis

Upon investigation, we found the issue is linked to the --uv flag. We use Process::kill to terminate the process, which by default sends the SIGKILL signal. When the uv process receives this signal, it exits immediately. Consequently, the actual Python nodes it spawned become orphan processes.

I tested replacing SIGKILL with SIGTERM (the default signal used in Docker), and this resolved the problem. However, it's important to note that SIGTERM does not offer the same termination guarantee that SIGKILL does.

Proposal for Mitigation

Given this, should we introduce a three-phase stopping procedure (Event::Stop -> SIGTERM -> SIGKILL), or would alternative mitigations be more appropriate?

[drindr]

There is a crate seems useful for this issue. The process wrap crate can wrap a command with multi-process and kill the whole process group.

[phil-opp]

Interesting crate! However, the fact that it's not cross-platforms makes this more complex to integrate unfortunately...

[drindr]

Compared to manually send signal, I still think it is worthy. According to its doc, there is only one line different for Windows and Unix-like platform.

And some nodes' with sub-process can always be closed successfully without adding extra burden for users.

let mut command = CommandWrap::with_new("watch", |command| { command.arg("ls"); });
#[cfg(unix)] { command.wrap(ProcessGroup::leader()); }
#[cfg(windows)] { command.wrap(JobObject); }
let mut child = command.spawn()?;
let status = child.wait()?;
dbg!(status);

[phil-opp]

Ok, this seems simple enough. Let's try this out!

[phil-opp]

Thanks for reporting this!

Given this, should we introduce a three-phase stopping procedure (Event::Stop -> SIGTERM -> SIGKILL), or would alternative mitigations be more appropriate?

This approach sounds reasonable to me!

[haixuanTao]

But what if the user does not want to receive a SIGTERM on Event::Stop?

Could we initially just replace SIGKILL by SIGTERM and use SIGKILL if we receive like a third ctrl-c?

[phil-opp]

Right now the ctrl-c logic for dora run and dora start --attach is:

• first ctrl-c sends a stop command to the coordinator
• second ctrl-c exits the CLI process immediately, without waiting until the dataflow is stopped

The kill logic is separate from this and uses the grace duration instead. When receiving a stop command, the coordinator/daemons:

• send a stop event to all nodes
• close the event channel of all nodes
• wait until all nodes exit or the grace duration expires
• kills all remaining nodes

[phil-opp]

How about the following approach:

• Once grace_duration expires, send SIGTERM to all nodes that are still running
• Once grace_duration * 1.5 expires, send SIGKILL to the remaining procesess
  • use process_wrap crate to kill the whole process group of each node

[Mivik]

Using process_wrap with Windows JobObjects prevents us from calling sysinfo::Process::kill, as it internally uses TerminateProcess instead of the required TerminateJobObject. This forces us to use Child::kill.

This creates a new problem: since std::process::Child is not Clone (and so is process_wrap::{std,tokio}::ChildWrapper), and both the .wait() and .kill() methods require mutable access (&mut self), we cannot kill the process while another part of the code is waiting on it.

Possible solutions:

1. Fork process-wrap to provide APIs like KillHandle.
2. Implement JobObject on our own. Since there're no well-wrapped APIs other than process-wrap, we may have to do it to hard way with windows or winapi

[phil-opp]

Or we redesign the killing logic so that ProcessId::kill is able to get a &mut reference on the Child. Maybe we can move ownership of the Child into ProcessId or we use a Arc<Mutex<Option<Child>>> (or similar).

[Mivik]

.wait() is a blocking operation and using mutex won't change the fact that it'll block other operations including kill. This is also illustrated in this SO question.

[Mivik]

.wait() is invoked here to notify the daemon when children stop.

[phil-opp]

According to https://docs.rs/tokio/1.48.0/tokio/process/struct.Child.html#method.wait, the wait function is cancel-safe. So shouldn't we be able to do a future::select on both the wait future and some kill_child oneshot receiver? Then we could cancel the waiting when a kill command arrives and call kill instead.

[Mivik]

Yes you're right, using async interface can mitigate the issue.