fix(ssh-for-commander): fix clissh goroutine leaks

Signed-off-by: Ilya Mayorov <ilya.mayorov@flant.com>

Author: faroyam

pkg/ssh/clissh/process/executor.go

@@ -18,12 +18,14 @@ import (
 	"bufio"
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"os"
 	"os/exec"
 	"reflect"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"
 
@@ -117,17 +119,24 @@ type Executor struct {
 	pipesMutex     sync.Mutex
 	stdoutPipeFile *os.File
 	stderrPipeFile *os.File
+	// Read ends of pipes must be closed to unblock goroutines stuck in Read().
+	stdoutReadPipe        *os.File
+	stderrReadPipe        *os.File
+	stdoutHandlerReadPipe *os.File
+	stderrHandlerReadPipe *os.File
 
 	StderrBuffer   *bytes.Buffer
 	StderrSplitter bufio.SplitFunc
 	StderrHandler  func(l string)
 
 	WaitHandler func(err error)
 
-	started bool
-	stop    bool
-	waitCh  chan struct{}
-	stopCh  chan struct{}
+	started  atomic.Bool
+	stop     atomic.Bool
+	waitCh   chan struct{}
+	stopCh   chan struct{}
+	doneCh   chan struct{}
+	stopOnce sync.Once
 
 	lockWaitError sync.RWMutex
 	waitError     error
@@ -258,6 +267,7 @@ func (e *Executor) SetupStreamHandlers() error {
 
 		e.pipesMutex.Lock()
 		e.stdoutPipeFile = stdoutWritePipe
+		e.stdoutReadPipe = stdoutReadPipe
 		e.pipesMutex.Unlock()
 
 		// create pipe for StdoutHandler
@@ -266,6 +276,9 @@ func (e *Executor) SetupStreamHandlers() error {
 			if err != nil {
 				return fmt.Errorf("unable to create os pipe for stdoutHandler: %s", err)
 			}
+			e.pipesMutex.Lock()
+			e.stdoutHandlerReadPipe = stdoutHandlerReadPipe
+			e.pipesMutex.Unlock()
 		}
 	}
 
@@ -283,6 +296,7 @@ func (e *Executor) SetupStreamHandlers() error {
 
 		e.pipesMutex.Lock()
 		e.stderrPipeFile = stderrWritePipe
+		e.stderrReadPipe = stderrReadPipe
 		e.pipesMutex.Unlock()
 
 		// create pipe for StderrHandler
@@ -291,6 +305,9 @@ func (e *Executor) SetupStreamHandlers() error {
 			if err != nil {
 				return fmt.Errorf("unable to create os pipe for stderrHandler: %s", err)
 			}
+			e.pipesMutex.Lock()
+			e.stderrHandlerReadPipe = stderrHandlerReadPipe
+			e.pipesMutex.Unlock()
 		}
 	}
 
@@ -307,6 +324,12 @@ func (e *Executor) SetupStreamHandlers() error {
 	// - Copy to buffer if capture is enabled
 	// - Copy to pipe if StdoutHandler is set
 	go func() {
+		defer func() {
+			if stdoutHandlerWritePipe != nil {
+				_ = stdoutHandlerWritePipe.Close()
+			}
+			e.closeStdoutReadPipe()
+		}()
 		e.readFromStreams(stdoutReadPipe, stdoutHandlerWritePipe)
 	}()
 
@@ -316,6 +339,7 @@ func (e *Executor) SetupStreamHandlers() error {
 		if e.StdoutHandler == nil {
 			return
 		}
+		defer e.closeStdoutHandlerReadPipe()
 		e.ConsumeLines(stdoutHandlerReadPipe, e.StdoutHandler)
 		logger.DebugF("stop line consumer for '%s'", e.cmd.Args[0])
 	}()
@@ -331,6 +355,12 @@ func (e *Executor) SetupStreamHandlers() error {
 
 		logger.DebugF("Start reading from stderr pipe")
 		defer logger.DebugF("Stop reading from stderr pipe")
+		defer func() {
+			if stderrHandlerWritePipe != nil {
+				_ = stderrHandlerWritePipe.Close()
+			}
+			e.closeStderrReadPipe()
+		}()
 
 		buf := make([]byte, 16)
 		for {
@@ -347,7 +377,11 @@ func (e *Executor) SetupStreamHandlers() error {
 				_, _ = stderrHandlerWritePipe.Write(buf[:n])
 			}
 
-			if err == io.EOF {
+			if err != nil {
+				if errors.Is(err, io.EOF) || errors.Is(err, os.ErrClosed) {
+					break
+				}
+				logger.DebugF("Error reading from stderr: %s\n", err)
 				break
 			}
 		}
@@ -357,6 +391,7 @@ func (e *Executor) SetupStreamHandlers() error {
 		if e.StderrHandler == nil {
 			return
 		}
+		defer e.closeStderrHandlerReadPipe()
 		e.ConsumeLines(stderrHandlerReadPipe, e.StderrHandler)
 		logger.DebugF("stop sdterr line consumer for '%s'", e.cmd.Args[0])
 	}()
@@ -384,14 +419,6 @@ func (e *Executor) readFromStreams(stdoutReadPipe io.Reader, stdoutHandlerWriteP
 	errorsCount := 0
 	for {
 		n, err := stdoutReadPipe.Read(buf)
-		if err != nil && err != io.EOF {
-			logger.DebugF("Error reading from stdout: %s\n", err)
-			errorsCount++
-			if errorsCount > 1000 {
-				panic(fmt.Errorf("readFromStreams: too many errors, last error %v", err))
-			}
-			continue
-		}
 
 		m := 0
 		if !matchersDone {
@@ -429,9 +456,18 @@ func (e *Executor) readFromStreams(stdoutReadPipe io.Reader, stdoutHandlerWriteP
 			_, _ = stdoutHandlerWritePipe.Write(buf[m:n])
 		}
 
-		if err == io.EOF {
-			logger.DebugF("readFromStreams: EOF")
-			break
+		if err != nil {
+			if errors.Is(err, io.EOF) || errors.Is(err, os.ErrClosed) {
+				logger.DebugF("readFromStreams: EOF")
+				break
+			}
+
+			logger.DebugF("Error reading from stdout: %s\n", err)
+			errorsCount++
+			if errorsCount > 1000 {
+				panic(fmt.Errorf("readFromStreams: too many errors, last error %v", err))
+			}
+			continue
 		}
 	}
 }
@@ -468,9 +504,9 @@ func (e *Executor) Start() error {
 	if err != nil {
 		return err
 	}
-	e.started = true
 
 	e.ProcessWait()
+	e.started.Store(true)
 
 	logger.DebugF("Register stoppable: '%s'", e.cmd.String())
 	e.Session.RegisterStoppable(e)
@@ -482,34 +518,31 @@ func (e *Executor) ProcessWait() {
 	waitErrCh := make(chan error, 1)
 	e.waitCh = make(chan struct{}, 1)
 	e.stopCh = make(chan struct{}, 1)
+	e.doneCh = make(chan struct{})
 
 	// wait for process in go routine
 	go func() {
 		waitErrCh <- e.cmd.Wait()
 	}()
 
-	go func() {
-		if e.timeout > 0 {
-			time.Sleep(e.timeout)
-			if e.stopCh != nil {
-				e.stopCh <- struct{}{}
-			}
-		}
-	}()
+	go e.waitForTimeout()
 
 	// watch for wait or stop
 	go func() {
 		defer func() {
+			e.closeWritePipes()
+			close(e.doneCh)
 			close(e.waitCh)
 			close(waitErrCh)
 		}()
 		// Wait until Stop() is called or/and Wait() is returning.
 		for {
 			select {
 			case err := <-waitErrCh:
-				if e.stop {
+				if e.stop.Load() {
 					// Ignore error if Stop() was called.
 					// close(e.waitCh)
+					e.killProcessGroup()
 					return
 				}
 				e.setWaitError(err)
@@ -519,26 +552,45 @@ func (e *Executor) ProcessWait() {
 				// close(e.waitCh)
 				return
 			case <-e.stopCh:
-				e.stop = true
-				// Prevent next readings from the closed channel.
-				e.stopCh = nil
 				// The usual e.cmd.Process.Kill() is not working for the process
 				// started with the new process group (Setpgid: true).
 				// Negative pid number is used to send a signal to all processes in the group.
-				err := syscall.Kill(-e.cmd.Process.Pid, syscall.SIGKILL)
-				if err != nil {
-					e.killError = err
-				}
+				e.stop.Store(true)
+				e.killProcessGroup()
+				e.forceClosePipes()
 			}
 		}
 	}()
 }
 
-func (e *Executor) closePipes() {
+func (e *Executor) waitForTimeout() {
+	if e.timeout <= 0 {
+		return
+	}
+
+	timer := time.NewTimer(e.timeout)
+	defer timer.Stop()
+
+	select {
+	case <-timer.C:
+		e.signalStop()
+	case <-e.doneCh:
+		return
+	}
+}
+
+func (e *Executor) killProcessGroup() {
+	err := syscall.Kill(-e.cmd.Process.Pid, syscall.SIGKILL)
+	if err != nil {
+		e.killError = err
+	}
+}
+
+func (e *Executor) closeWritePipes() {
 	logger := e.settings.Logger()
 
-	logger.DebugF("Starting close piped")
-	defer logger.DebugF("Stop close piped")
+	logger.DebugF("Starting close write pipes")
+	defer logger.DebugF("Stop close write pipes")
 
 	e.pipesMutex.Lock()
 	defer e.pipesMutex.Unlock()
@@ -560,30 +612,80 @@ func (e *Executor) closePipes() {
 	}
 }
 
+func (e *Executor) forceClosePipes() {
+	e.closeWritePipes()
+
+	logger := e.settings.Logger()
+
+	logger.DebugF("Starting force close pipes")
+	defer logger.DebugF("Stop force close pipes")
+
+	e.closeStdoutReadPipe()
+	e.closeStderrReadPipe()
+	e.closeStdoutHandlerReadPipe()
+	e.closeStderrHandlerReadPipe()
+}
+
+func (e *Executor) closeStdoutReadPipe() {
+	e.closePipeFile(&e.stdoutReadPipe, "stdout read pipe")
+}
+
+func (e *Executor) closeStderrReadPipe() {
+	e.closePipeFile(&e.stderrReadPipe, "stderr read pipe")
+}
+
+func (e *Executor) closeStdoutHandlerReadPipe() {
+	e.closePipeFile(&e.stdoutHandlerReadPipe, "stdout handler read pipe")
+}
+
+func (e *Executor) closeStderrHandlerReadPipe() {
+	e.closePipeFile(&e.stderrHandlerReadPipe, "stderr handler read pipe")
+}
+
+func (e *Executor) closePipeFile(pipe **os.File, name string) {
+	e.pipesMutex.Lock()
+	defer e.pipesMutex.Unlock()
+
+	if *pipe != nil {
+		err := (*pipe).Close()
+		if err != nil {
+			e.settings.Logger().DebugF("Cannot close %s: %v", name, err)
+		}
+		*pipe = nil
+	}
+}
+
 func (e *Executor) Stop() {
 	logger := e.settings.Logger()
 
-	if e.stop {
-		logger.DebugF("Stop '%s': already stopped", e.cmd.String())
+	if e.cmd == nil {
+		logger.DebugF("Possible BUG: Call Executor.Stop with Cmd==nil")
 		return
 	}
-	if !e.started {
+	if !e.started.Load() {
 		logger.DebugF("Stop '%s': not started yet", e.cmd.String())
 		return
 	}
-	if e.cmd == nil {
-		logger.DebugF("Possible BUG: Call Executor.Stop with Cmd==nil")
-		return
-	}
 
-	e.stop = true
-	logger.DebugF("Stop '%s'", e.cmd.String())
-	if e.stopCh != nil {
-		close(e.stopCh)
+	if e.stop.CompareAndSwap(false, true) {
+		logger.DebugF("Stop '%s'", e.cmd.String())
+		e.signalStop()
+	} else {
+		logger.DebugF("Stop '%s': already stopping", e.cmd.String())
 	}
+
+	e.forceClosePipes()
 	<-e.waitCh
 	logger.DebugF("Stopped '%s': %d", e.cmd.String(), e.cmd.ProcessState.ExitCode())
-	e.closePipes()
+}
+
+func (e *Executor) signalStop() {
+	e.stopOnce.Do(func() {
+		select {
+		case e.stopCh <- struct{}{}:
+		default:
+		}
+	})
 }
 
 // Run executes a command and blocks until it is finished or stopped.
@@ -597,7 +699,7 @@ func (e *Executor) Run(_ context.Context) error {
 
 	<-e.waitCh
 
-	e.closePipes()
+	e.closeWritePipes()
 
 	return e.WaitError()
 }