phoerious commented on a change in pull request #16658:
URL: https://github.com/apache/beam/pull/16658#discussion_r820078506
##########
File path: sdks/python/container/boot.go
##########
@@ -214,41 +211,105 @@ func mainError() error {
}
}
+ 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 {
+ syscall.Kill(-pid, syscall.SIGTERM)
+ go func() {
+ // 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)
+ log.Printf("Worker process did not respond,
killing it.")
+ syscall.Kill(-pid, syscall.SIGKILL)
+ }()
+ }
+ childPids.mu.Unlock()
+ }()
+
args := []string{
"-m",
sdkHarnessEntrypoint,
}
- workerIds := append([]string{*id}, info.GetSiblingWorkerIds()...)
var wg sync.WaitGroup
wg.Add(len(workerIds))
for _, workerId := range workerIds {
go func(workerId string) {
defer wg.Done()
+
+ childPids.mu.Lock()
+ if childPids.canceled {
+ childPids.mu.Unlock()
+ return
+ }
log.Printf("Executing Python (worker %v): python %v",
workerId, strings.Join(args, " "))
- log.Printf("Python (worker %v) exited with code: %v",
workerId, execx.ExecuteEnv(map[string]string{"WORKER_ID": workerId}, "python",
args...))
+ cmd := StartCommandEnv(map[string]string{"WORKER_ID":
workerId}, "python", args...)
+ childPids.v = append(childPids.v, cmd.Process.Pid)
+ childPids.mu.Unlock()
+
+ if err := cmd.Wait(); err != nil {
+ log.Printf("Python (worker %v) exited: %v",
workerId, err)
Review comment:
No, since Python workers are often terminated by SIGTERM even if they
were successful.
--
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]
