🐛 Fix hanging processes and subprocess termination (#444)

<!--
Copyright (C) 2020-2022 Arm Limited or its affiliates and Contributors.
All rights reserved.
SPDX-License-Identifier: Apache-2.0
-->
### Description

- Trying to solve how subprocesses are spawned so that their termination
is done gracefully and no orphans are left hanging



### Test Coverage

<!--
Please put an `x` in the correct box e.g. `[x]` to indicate the testing
coverage of this change.
-->

- [x]  This change is covered by existing or additional automated tests.
- [ ] Manual testing has been performed (and evidence provided) as
automated testing was not feasible.
- [ ] Additional tests are not required for this change (e.g.
documentation update).

Author: acabarbaye

changes/20240403170926.bugfix

@@ -0,0 +1 @@
+:bug: Fix sub processes termination and command process setup as described in [(go issue)](https://github.com/golang/go/issues/24050).

utils/subprocess/command_wrapper.go

@@ -2,12 +2,12 @@
  * Copyright (C) 2020-2022 Arm Limited or its affiliates and Contributors. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
+
 package subprocess
 
 import (
 	"context"
 	"fmt"
-	"os"
 	"os/exec"
 	"time"
 
@@ -16,6 +16,7 @@ import (
 	"github.com/ARM-software/golang-utils/utils/commonerrors"
 	"github.com/ARM-software/golang-utils/utils/logs"
 	"github.com/ARM-software/golang-utils/utils/parallelisation"
+	"github.com/ARM-software/golang-utils/utils/proc"
 	commandUtils "github.com/ARM-software/golang-utils/utils/subprocess/command"
 )
 
@@ -46,7 +47,7 @@ func (c *cmdWrapper) Start() error {
 	if c.cmd == nil {
 		return fmt.Errorf("%w:undefined command", commonerrors.ErrUndefined)
 	}
-	return commonerrors.ConvertContextError(c.cmd.Start())
+	return ConvertCommandError(c.cmd.Start())
 }
 
 func (c *cmdWrapper) Run() error {
@@ -55,7 +56,7 @@ func (c *cmdWrapper) Run() error {
 	if c.cmd == nil {
 		return fmt.Errorf("%w:undefined command", commonerrors.ErrUndefined)
 	}
-	return commonerrors.ConvertContextError(c.cmd.Run())
+	return ConvertCommandError(c.cmd.Run())
 }
 
 func (c *cmdWrapper) Stop() error {
@@ -70,11 +71,11 @@ func (c *cmdWrapper) Stop() error {
 	if subprocess != nil {
 		pid := subprocess.Pid
 		parallelisation.ScheduleAfter(ctx, 10*time.Millisecond, func(time.Time) {
-			process, err := os.FindProcess(pid)
+			process, err := proc.FindProcess(ctx, pid)
 			if process == nil || err != nil {
 				return
 			}
-			_ = process.Kill()
+			_ = process.KillWithChildren(ctx)
 		})
 	}
 	_ = c.cmd.Wait()
@@ -110,10 +111,11 @@ type command struct {
 func (c *command) createCommand(cmdCtx context.Context) *exec.Cmd {
 	newCmd, newArgs := c.as.Redefine(c.cmd, c.args...)
 	cmd := exec.CommandContext(cmdCtx, newCmd, newArgs...) //nolint:gosec
-	cmd.Stdout = newOutStreamer(c.loggers)
-	cmd.Stderr = newErrLogStreamer(c.loggers)
+	cmd.Stdout = newOutStreamer(cmdCtx, c.loggers)
+	cmd.Stderr = newErrLogStreamer(cmdCtx, c.loggers)
 	cmd.Env = cmd.Environ()
 	cmd.Env = append(cmd.Env, c.env...)
+	setGroupAttrToCmd(cmd)
 	return cmd
 }
 
@@ -157,3 +159,31 @@ func newCommand(loggers logs.Loggers, as *commandUtils.CommandAsDifferentUser, e
 	}
 	return
 }
+
+func ConvertCommandError(err error) error {
+	return proc.ConvertProcessError(err)
+}
+
+func CleanKillOfCommand(ctx context.Context, cmd *exec.Cmd) (err error) {
+	if cmd == nil {
+		return
+	}
+	defer func() {
+		if cmd.Process != nil {
+			_ = cmd.Process.Kill()
+		}
+	}()
+
+	thisP := cmd.Process
+	if thisP == nil {
+		return
+	} else {
+		p, subErr := proc.FindProcess(ctx, thisP.Pid)
+		if subErr != nil {
+			err = subErr
+			return
+		}
+		err = p.KillWithChildren(ctx)
+	}
+	return
+}

utils/subprocess/command_wrapper_darwin.go

@@ -0,0 +1,16 @@
+//go:build darwin
+// +build darwin
+
+package subprocess
+
+import (
+	"os/exec"
+	"syscall"
+)
+
+// See https://github.com/tgulacsi/go/blob/master/proc/
+func setGroupAttrToCmd(c *exec.Cmd) {
+	c.SysProcAttr = &syscall.SysProcAttr{
+		Setpgid: true, // to be able to kill all children, too
+	}
+}

utils/subprocess/command_wrapper_linux.go

@@ -0,0 +1,17 @@
+//go:build linux
+// +build linux
+
+package subprocess
+
+import (
+	"os/exec"
+	"syscall"
+)
+
+// See https://github.com/tgulacsi/go/blob/master/proc/
+func setGroupAttrToCmd(c *exec.Cmd) {
+	c.SysProcAttr = &syscall.SysProcAttr{
+		Setpgid:   true, // to be able to kill all children, too
+		Pdeathsig: syscall.SIGKILL,
+	}
+}

utils/subprocess/command_wrapper_windows.go

@@ -0,0 +1,17 @@
+//go:build windows
+// +build windows
+
+package subprocess
+
+import (
+	"os/exec"
+	"syscall"
+)
+
+func setGroupAttrToCmd(c *exec.Cmd) {
+	c.SysProcAttr = &syscall.SysProcAttr{
+		HideWindow: true,
+		// Windows Process Creation Flags:  https://learn.microsoft.com/en-us/windows/win32/procthread/process-creation-flags
+		CreationFlags: syscall.CREATE_UNICODE_ENVIRONMENT | syscall.CREATE_NEW_PROCESS_GROUP,
+	}
+}

utils/subprocess/executor.go

@@ -18,13 +18,13 @@ import (
 	commandUtils "github.com/ARM-software/golang-utils/utils/subprocess/command"
 )
 
-// Subprocess describes what a subproccess is as well as any monitoring it may need.
+// Subprocess describes what a subprocess is as well as any monitoring it may need.
 type Subprocess struct {
 	mu                deadlock.RWMutex
 	isRunning         atomic.Bool
 	command           *command
 	processMonitoring *subprocessMonitoring
-	messsaging        *subprocessMessaging
+	messaging         *subprocessMessaging
 }
 
 // New creates a subprocess description.
@@ -156,7 +156,7 @@ func (s *Subprocess) setup(ctx context.Context, loggers logs.Loggers, env []stri
 	s.isRunning.Store(false)
 	s.processMonitoring = newSubprocessMonitoring(ctx)
 	s.command = newCommand(loggers, as, env, cmd, args...)
-	s.messsaging = newSubprocessMessaging(loggers, withAdditionalMessages, messageOnSuccess, messageOnFailure, messageOnStart, s.command.GetPath())
+	s.messaging = newSubprocessMessaging(loggers, withAdditionalMessages, messageOnSuccess, messageOnFailure, messageOnStart, s.command.GetPath())
 	s.reset()
 	return s.check()
 }
@@ -172,11 +172,11 @@ func (s *Subprocess) check() (err error) {
 	if err != nil {
 		return
 	}
-	if s.messsaging == nil {
+	if s.messaging == nil {
 		err = commonerrors.ErrNoLogger
 		return
 	}
-	err = s.messsaging.Check()
+	err = s.messaging.Check()
 	return
 }
 
@@ -212,26 +212,26 @@ func (s *Subprocess) Start() (err error) {
 	cmd := s.getCmd()
 	err = cmd.Start()
 	if err != nil {
-		s.messsaging.LogFailedStart(err)
+		s.messaging.LogFailedStart(err)
 		s.isRunning.Store(false)
 		s.Cancel()
 		return
 	}
 	pid, err := cmd.Pid()
 	if err != nil {
-		s.messsaging.LogFailedStart(err)
+		s.messaging.LogFailedStart(err)
 		s.isRunning.Store(false)
 		s.Cancel()
 		return
 	}
 
 	s.isRunning.Store(true)
-	s.messsaging.SetPid(pid)
-	s.messsaging.LogStarted()
+	s.messaging.SetPid(pid)
+	s.messaging.LogStarted()
 	return
 }
 
-// Cancel interrupts an on-going process. This method is idempotent.
+// Cancel interrupts an ongoing process. This method is idempotent.
 func (s *Subprocess) Cancel() {
 	s.processMonitoring.CancelSubprocess()
 }
@@ -251,13 +251,13 @@ func (s *Subprocess) Execute() (err error) {
 	}
 	s.processMonitoring.Reset()
 	s.command.Reset()
-	s.messsaging.LogStart()
+	s.messaging.LogStart()
 	s.runProcessMonitoring()
 	cmd := s.getCmd()
 	s.isRunning.Store(true)
 	err = cmd.Run()
 	s.isRunning.Store(false)
-	s.messsaging.LogEnd(err)
+	s.messaging.LogEnd(err)
 	return
 }
 
@@ -307,10 +307,10 @@ func (s *Subprocess) stop(cancel bool) (err error) {
 	if !s.IsOn() {
 		return
 	}
-	s.messsaging.LogStopping()
+	s.messaging.LogStopping()
 	err = s.getCmd().Stop()
 	s.command.Reset()
 	s.isRunning.Store(false)
-	s.messsaging.LogEnd(nil)
+	s.messaging.LogEnd(nil)
 	return
 }

utils/subprocess/logging.go

@@ -5,19 +5,20 @@
 package subprocess
 
 import (
+	"context"
 	"io"
 	"strings"
 
 	"github.com/ARM-software/golang-utils/utils/logs"
 	"github.com/ARM-software/golang-utils/utils/platform"
+	"github.com/ARM-software/golang-utils/utils/safeio"
 )
 
 var lineSep = platform.UnixLineSeparator()
 
 // INTERNAL
 // Way of redirecting process output to a logger.
 type logStreamer struct {
-	io.Writer
 	IsStdErr bool
 	Loggers  logs.Loggers
 }
@@ -37,17 +38,17 @@ func (l *logStreamer) Write(p []byte) (n int, err error) {
 	return len(p), nil
 }
 
-func newLogStreamer(isStdErr bool, loggers logs.Loggers) *logStreamer {
-	return &logStreamer{
+func newLogStreamer(ctx context.Context, isStdErr bool, loggers logs.Loggers) io.Writer {
+	return safeio.ContextualWriter(ctx, &logStreamer{
 		IsStdErr: isStdErr,
 		Loggers:  loggers,
-	}
+	})
 }
 
-func newOutStreamer(loggers logs.Loggers) *logStreamer {
-	return newLogStreamer(false, loggers)
+func newOutStreamer(ctx context.Context, loggers logs.Loggers) io.Writer {
+	return newLogStreamer(ctx, false, loggers)
 }
 
-func newErrLogStreamer(loggers logs.Loggers) *logStreamer {
-	return newLogStreamer(true, loggers)
+func newErrLogStreamer(ctx context.Context, loggers logs.Loggers) io.Writer {
+	return newLogStreamer(ctx, true, loggers)
 }

utils/subprocess/messaging.go

@@ -27,35 +27,35 @@ type subprocessMessaging struct {
 	pid                    atomic.Int64
 }
 
-// Logs subprocess start.
+// LogStart logs subprocess start.
 func (s *subprocessMessaging) LogStart() {
 	if s.withAdditionalMessages {
 		s.loggers.Log(s.messageOnProcessStart)
 	}
 }
 
-// Logs when subprocess failed to start.
+// LogFailedStart logs when subprocess failed to start.
 func (s *subprocessMessaging) LogFailedStart(err error) {
 	if s.withAdditionalMessages {
 		s.loggers.LogError(fmt.Sprintf("Failed starting process `%v`: %v", s.commandPath, err))
 	}
 }
 
-// Logs when subprocess has started.
+// LogStarted logs when subprocess has started.
 func (s *subprocessMessaging) LogStarted() {
 	if s.withAdditionalMessages {
 		s.loggers.Log(fmt.Sprintf("Started process [%v]", s.pid.Load()))
 	}
 }
 
-// Logs when subprocess is asked to stop.
+// LogStopping logs when subprocess is asked to stop.
 func (s *subprocessMessaging) LogStopping() {
 	if s.withAdditionalMessages {
 		s.loggers.Log(fmt.Sprintf("Stopping process [%v]", s.pid.Load()))
 	}
 }
 
-// Logs subprocess end with err if an error occurred.
+// LogEnd logs subprocess end with err if an error occurred.
 func (s *subprocessMessaging) LogEnd(err error) {
 	if !s.withAdditionalMessages {
 		return
@@ -67,7 +67,7 @@ func (s *subprocessMessaging) LogEnd(err error) {
 	}
 }
 
-// Sets the process PID.
+// SetPid sets the process PID.
 func (s *subprocessMessaging) SetPid(pid int) {
 	s.pid.Store(int64(pid))
 }

utils/subprocess/monitoring.go

@@ -32,7 +32,7 @@ func newSubprocessMonitoring(parentCtx context.Context) *subprocessMonitoring {
 	return m
 }
 
-// Interrupts an on-going process.
+// CancelSubprocess interrupts an on-going process.
 func (s *subprocessMonitoring) CancelSubprocess() {
 	s.monitoringStopping.Store(true)
 	s.cancelStore.Cancel()