phoerious commented on code in PR #16658:
URL: https://github.com/apache/beam/pull/16658#discussion_r991661936
##########
sdks/python/container/boot.go:
##########
@@ -200,24 +211,112 @@ func main() {
}
}
+ workerIds := append([]string{*id}, info.GetSiblingWorkerIds()...)
+
+ // Keep track of child PIDs for clean shutdown without zombies
+ childPids := struct {
+ v []int
+ canceled bool
+ mu sync.Mutex
+ } {v: make([]int, 0, len(workerIds))}
+
+ // Forward trapped signals to child process groups in order to
terminate them gracefully and avoid zombies
+ go func() {
+ log.Printf("Received signal: %v", <-signalChannel)
+ childPids.mu.Lock()
+ childPids.canceled = true
+ for _, pid := range childPids.v {
+ go func(pid int) {
+ // This goroutine will be canceled if the main
process exits before the 5 seconds
+ // have elapsed, i.e., as soon as all
subprocesses have returned from Wait().
+ time.Sleep(5 * time.Second)
+ if err := syscall.Kill(-pid, syscall.SIGKILL);
err == nil {
+ log.Printf("Worker process %v did not
respond, killed it.", pid)
+ }
+ }(pid)
+ syscall.Kill(-pid, syscall.SIGTERM)
Review Comment:
The native PID refers to the process group, so we can SIGTERM the whole tree
in case the Python process spawns child processes. The process group is created
further down in the code. If we don't do this, the sub processes will be
orphaned, creating zombie processes if the Python process fails to wait() them.
##########
sdks/python/container/boot.go:
##########
@@ -200,24 +211,112 @@ func main() {
}
}
+ workerIds := append([]string{*id}, info.GetSiblingWorkerIds()...)
+
+ // Keep track of child PIDs for clean shutdown without zombies
+ childPids := struct {
+ v []int
+ canceled bool
+ mu sync.Mutex
+ } {v: make([]int, 0, len(workerIds))}
+
+ // Forward trapped signals to child process groups in order to
terminate them gracefully and avoid zombies
+ go func() {
+ log.Printf("Received signal: %v", <-signalChannel)
+ childPids.mu.Lock()
+ childPids.canceled = true
+ for _, pid := range childPids.v {
+ go func(pid int) {
+ // This goroutine will be canceled if the main
process exits before the 5 seconds
+ // have elapsed, i.e., as soon as all
subprocesses have returned from Wait().
+ time.Sleep(5 * time.Second)
+ if err := syscall.Kill(-pid, syscall.SIGKILL);
err == nil {
+ log.Printf("Worker process %v did not
respond, killed it.", pid)
+ }
+ }(pid)
+ syscall.Kill(-pid, syscall.SIGTERM)
Review Comment:
The negative PID refers to the process group, so we can SIGTERM the whole
tree in case the Python process spawns child processes. The process group is
created further down in the code. If we don't do this, the sub processes will
be orphaned, creating zombie processes if the Python process fails to wait()
them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
