Graceful: Cancel Process on monitor pages & HammerTime (#9213)

* Graceful: Create callbacks to with contexts

* Graceful: Say when Gitea is completely finished

* Graceful: Git and Process within HammerTime

Force all git commands to terminate at HammerTime
Force all process commands to terminate at HammerTime
Move almost all git processes to run as git Commands

* Graceful: Always Hammer after Shutdown

* ProcessManager: Add cancel functionality

* Fix tests

* Make sure that process.Manager.Kill() cancels

* Make threadsafe access to Processes and remove own unused Kill

* Remove cmd from the process manager as it is no longer used

* the default context is the correct context

* get rid of double till

modules/git/blame.go

@@ -6,6 +6,7 @@ package git
 
 import (
 	"bufio"
+	"context"
 	"fmt"
 	"io"
 	"os"
@@ -28,6 +29,7 @@ type BlameReader struct {
 	output  io.ReadCloser
 	scanner *bufio.Scanner
 	lastSha *string
+	cancel  context.CancelFunc
 }
 
 var shaLineRegex = regexp.MustCompile("^([a-z0-9]{40})")
@@ -76,7 +78,8 @@ func (r *BlameReader) NextPart() (*BlamePart, error) {
 
 // Close BlameReader - don't run NextPart after invoking that
 func (r *BlameReader) Close() error {
-	process.GetManager().Remove(r.pid)
+	defer process.GetManager().Remove(r.pid)
+	defer r.cancel()
 
 	if err := r.cmd.Wait(); err != nil {
 		return fmt.Errorf("Wait: %v", err)
@@ -97,20 +100,24 @@ func CreateBlameReader(repoPath, commitID, file string) (*BlameReader, error) {
 }
 
 func createBlameReader(dir string, command ...string) (*BlameReader, error) {
-	cmd := exec.Command(command[0], command[1:]...)
+	// FIXME: graceful: This should have a timeout
+	ctx, cancel := context.WithCancel(DefaultContext)
+	cmd := exec.CommandContext(ctx, command[0], command[1:]...)
 	cmd.Dir = dir
 	cmd.Stderr = os.Stderr
 
 	stdout, err := cmd.StdoutPipe()
 	if err != nil {
+		defer cancel()
 		return nil, fmt.Errorf("StdoutPipe: %v", err)
 	}
 
 	if err = cmd.Start(); err != nil {
+		defer cancel()
 		return nil, fmt.Errorf("Start: %v", err)
 	}
 
-	pid := process.GetManager().Add(fmt.Sprintf("GetBlame [repo_path: %s]", dir), cmd)
+	pid := process.GetManager().Add(fmt.Sprintf("GetBlame [repo_path: %s]", dir), cancel)
 
 	scanner := bufio.NewScanner(stdout)
 
@@ -120,5 +127,6 @@ func createBlameReader(dir string, command ...string) (*BlameReader, error) {
 		stdout,
 		scanner,
 		nil,
+		cancel,
 	}, nil
 }

modules/git/command.go

@@ -30,8 +30,10 @@ const DefaultLocale = "C"
 
 // Command represents a command with its subcommands or arguments.
 type Command struct {
-	name string
-	args []string
+	name          string
+	args          []string
+	parentContext context.Context
+	desc          string
 }
 
 func (c *Command) String() string {
@@ -47,19 +49,34 @@ func NewCommand(args ...string) *Command {
 	cargs := make([]string, len(GlobalCommandArgs))
 	copy(cargs, GlobalCommandArgs)
 	return &Command{
-		name: GitExecutable,
-		args: append(cargs, args...),
+		name:          GitExecutable,
+		args:          append(cargs, args...),
+		parentContext: DefaultContext,
 	}
 }
 
 // NewCommandNoGlobals creates and returns a new Git Command based on given command and arguments only with the specify args and don't care global command args
 func NewCommandNoGlobals(args ...string) *Command {
 	return &Command{
-		name: GitExecutable,
-		args: args,
+		name:          GitExecutable,
+		args:          args,
+		parentContext: DefaultContext,
 	}
 }
 
+// SetParentContext sets the parent context for this command
+func (c *Command) SetParentContext(ctx context.Context) *Command {
+	c.parentContext = ctx
+	return c
+}
+
+// SetDescription sets the description for this command which be returned on
+// c.String()
+func (c *Command) SetDescription(desc string) *Command {
+	c.desc = desc
+	return c
+}
+
 // AddArguments adds new argument(s) to the command.
 func (c *Command) AddArguments(args ...string) *Command {
 	c.args = append(c.args, args...)
@@ -92,7 +109,7 @@ func (c *Command) RunInDirTimeoutEnvFullPipelineFunc(env []string, timeout time.
 		log("%s: %v", dir, c)
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	ctx, cancel := context.WithTimeout(c.parentContext, timeout)
 	defer cancel()
 
 	cmd := exec.CommandContext(ctx, c.name, c.args...)
@@ -110,7 +127,11 @@ func (c *Command) RunInDirTimeoutEnvFullPipelineFunc(env []string, timeout time.
 		return err
 	}
 
-	pid := process.GetManager().Add(fmt.Sprintf("%s %s %s [repo_path: %s]", GitExecutable, c.name, strings.Join(c.args, " "), dir), cmd)
+	desc := c.desc
+	if desc == "" {
+		desc = fmt.Sprintf("%s %s %s [repo_path: %s]", GitExecutable, c.name, strings.Join(c.args, " "), dir)
+	}
+	pid := process.GetManager().Add(desc, cancel)
 	defer process.GetManager().Remove(pid)
 
 	if fn != nil {

modules/git/git.go

@@ -6,6 +6,7 @@
 package git
 
 import (
+	"context"
 	"fmt"
 	"os/exec"
 	"runtime"
@@ -35,6 +36,9 @@ var (
 	// Could be updated to an absolute path while initialization
 	GitExecutable = "git"
 
+	// DefaultContext is the default context to run git commands in
+	DefaultContext = context.Background()
+
 	gitVersion string
 )
 

modules/graceful/context.go

@@ -0,0 +1,90 @@
+// Copyright 2019 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package graceful
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// Errors for context.Err()
+var (
+	ErrShutdown  = fmt.Errorf("Graceful Manager called Shutdown")
+	ErrHammer    = fmt.Errorf("Graceful Manager called Hammer")
+	ErrTerminate = fmt.Errorf("Graceful Manager called Terminate")
+)
+
+// ChannelContext is a context that wraps a channel and error as a context
+type ChannelContext struct {
+	done <-chan struct{}
+	err  error
+}
+
+// NewChannelContext creates a ChannelContext from a channel and error
+func NewChannelContext(done <-chan struct{}, err error) *ChannelContext {
+	return &ChannelContext{
+		done: done,
+		err:  err,
+	}
+}
+
+// Deadline returns the time when work done on behalf of this context
+// should be canceled. There is no Deadline for a ChannelContext
+func (ctx *ChannelContext) Deadline() (deadline time.Time, ok bool) {
+	return
+}
+
+// Done returns the channel provided at the creation of this context.
+// When closed, work done on behalf of this context should be canceled.
+func (ctx *ChannelContext) Done() <-chan struct{} {
+	return ctx.done
+}
+
+// Err returns nil, if Done is not closed. If Done is closed,
+// Err returns the error provided at the creation of this context
+func (ctx *ChannelContext) Err() error {
+	select {
+	case <-ctx.done:
+		return ctx.err
+	default:
+		return nil
+	}
+}
+
+// Value returns nil for all calls as no values are or can be associated with this context
+func (ctx *ChannelContext) Value(key interface{}) interface{} {
+	return nil
+}
+
+// ShutdownContext returns a context.Context that is Done at shutdown
+// Callers using this context should ensure that they are registered as a running server
+// in order that they are waited for.
+func (g *gracefulManager) ShutdownContext() context.Context {
+	return &ChannelContext{
+		done: g.IsShutdown(),
+		err:  ErrShutdown,
+	}
+}
+
+// HammerContext returns a context.Context that is Done at hammer
+// Callers using this context should ensure that they are registered as a running server
+// in order that they are waited for.
+func (g *gracefulManager) HammerContext() context.Context {
+	return &ChannelContext{
+		done: g.IsHammer(),
+		err:  ErrHammer,
+	}
+}
+
+// TerminateContext returns a context.Context that is Done at terminate
+// Callers using this context should ensure that they are registered as a terminating server
+// in order that they are waited for.
+func (g *gracefulManager) TerminateContext() context.Context {
+	return &ChannelContext{
+		done: g.IsTerminate(),
+		err:  ErrTerminate,
+	}
+}

modules/graceful/manager.go

@@ -5,9 +5,12 @@
 package graceful
 
 import (
+	"context"
 	"time"
 
+	"code.gitea.io/gitea/modules/git"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/setting"
 )
 
@@ -34,9 +37,110 @@ const numberOfServersToCreate = 3
 var Manager *gracefulManager
 
 func init() {
-	Manager = newGracefulManager()
+	Manager = newGracefulManager(context.Background())
+	// Set the git default context to the HammerContext
+	git.DefaultContext = Manager.HammerContext()
+	// Set the process default context to the HammerContext
+	process.DefaultContext = Manager.HammerContext()
 }
 
+// CallbackWithContext is combined runnable and context to watch to see if the caller has finished
+type CallbackWithContext func(ctx context.Context, callback func())
+
+// RunnableWithShutdownFns is a runnable with functions to run at shutdown and terminate
+// After the callback to atShutdown is called and is complete, the main function must return.
+// Similarly the callback function provided to atTerminate must return once termination is complete.
+// Please note that use of the atShutdown and atTerminate callbacks will create go-routines that will wait till their respective signals
+// - users must therefore be careful to only call these as necessary.
+// If run is not expected to run indefinitely RunWithShutdownChan is likely to be more appropriate.
+type RunnableWithShutdownFns func(atShutdown, atTerminate func(context.Context, func()))
+
+// RunWithShutdownFns takes a function that has both atShutdown and atTerminate callbacks
+// After the callback to atShutdown is called and is complete, the main function must return.
+// Similarly the callback function provided to atTerminate must return once termination is complete.
+// Please note that use of the atShutdown and atTerminate callbacks will create go-routines that will wait till their respective signals
+// - users must therefore be careful to only call these as necessary.
+// If run is not expected to run indefinitely RunWithShutdownChan is likely to be more appropriate.
+func (g *gracefulManager) RunWithShutdownFns(run RunnableWithShutdownFns) {
+	g.runningServerWaitGroup.Add(1)
+	defer g.runningServerWaitGroup.Done()
+	run(func(ctx context.Context, atShutdown func()) {
+		go func() {
+			select {
+			case <-g.IsShutdown():
+				atShutdown()
+			case <-ctx.Done():
+				return
+			}
+		}()
+	}, func(ctx context.Context, atTerminate func()) {
+		g.RunAtTerminate(ctx, atTerminate)
+	})
+}
+
+// RunnableWithShutdownChan is a runnable with functions to run at shutdown and terminate.
+// After the atShutdown channel is closed, the main function must return once shutdown is complete.
+// (Optionally IsHammer may be waited for instead however, this should be avoided if possible.)
+// The callback function provided to atTerminate must return once termination is complete.
+// Please note that use of the atTerminate function will create a go-routine that will wait till terminate - users must therefore be careful to only call this as necessary.
+type RunnableWithShutdownChan func(atShutdown <-chan struct{}, atTerminate CallbackWithContext)
+
+// RunWithShutdownChan takes a function that has channel to watch for shutdown and atTerminate callbacks
+// After the atShutdown channel is closed, the main function must return once shutdown is complete.
+// (Optionally IsHammer may be waited for instead however, this should be avoided if possible.)
+// The callback function provided to atTerminate must return once termination is complete.
+// Please note that use of the atTerminate function will create a go-routine that will wait till terminate - users must therefore be careful to only call this as necessary.
+func (g *gracefulManager) RunWithShutdownChan(run RunnableWithShutdownChan) {
+	g.runningServerWaitGroup.Add(1)
+	defer g.runningServerWaitGroup.Done()
+	run(g.IsShutdown(), func(ctx context.Context, atTerminate func()) {
+		g.RunAtTerminate(ctx, atTerminate)
+	})
+}
+
+// RunWithShutdownContext takes a function that has a context to watch for shutdown.
+// After the provided context is Done(), the main function must return once shutdown is complete.
+// (Optionally the HammerContext may be obtained and waited for however, this should be avoided if possible.)
+func (g *gracefulManager) RunWithShutdownContext(run func(context.Context)) {
+	g.runningServerWaitGroup.Add(1)
+	defer g.runningServerWaitGroup.Done()
+	run(g.ShutdownContext())
+}
+
+// RunAtTerminate adds to the terminate wait group and creates a go-routine to run the provided function at termination
+func (g *gracefulManager) RunAtTerminate(ctx context.Context, terminate func()) {
+	g.terminateWaitGroup.Add(1)
+	go func() {
+		select {
+		case <-g.IsTerminate():
+			terminate()
+		case <-ctx.Done():
+		}
+		g.terminateWaitGroup.Done()
+	}()
+}
+
+// RunAtShutdown creates a go-routine to run the provided function at shutdown
+func (g *gracefulManager) RunAtShutdown(ctx context.Context, shutdown func()) {
+	go func() {
+		select {
+		case <-g.IsShutdown():
+			shutdown()
+		case <-ctx.Done():
+		}
+	}()
+}
+
+// RunAtHammer creates a go-routine to run the provided function at shutdown
+func (g *gracefulManager) RunAtHammer(ctx context.Context, hammer func()) {
+	go func() {
+		select {
+		case <-g.IsHammer():
+			hammer()
+		case <-ctx.Done():
+		}
+	}()
+}
 func (g *gracefulManager) doShutdown() {
 	if !g.setStateTransition(stateRunning, stateShuttingDown) {
 		return
@@ -50,6 +154,8 @@ func (g *gracefulManager) doShutdown() {
 	}
 	go func() {
 		g.WaitForServers()
+		// Mop up any remaining unclosed events.
+		g.doHammerTime(0)
 		<-time.After(1 * time.Second)
 		g.doTerminate()
 	}()

modules/graceful/manager_unix.go

@@ -7,6 +7,7 @@
 package graceful
 
 import (
+	"context"
 	"errors"
 	"os"
 	"os/signal"
@@ -31,19 +32,19 @@ type gracefulManager struct {
 	terminateWaitGroup     sync.WaitGroup
 }
 
-func newGracefulManager() *gracefulManager {
+func newGracefulManager(ctx context.Context) *gracefulManager {
 	manager := &gracefulManager{
 		isChild: len(os.Getenv(listenFDs)) > 0 && os.Getppid() > 1,
 		lock:    &sync.RWMutex{},
 	}
 	manager.createServerWaitGroup.Add(numberOfServersToCreate)
-	manager.Run()
+	manager.Run(ctx)
 	return manager
 }
 
-func (g *gracefulManager) Run() {
+func (g *gracefulManager) Run(ctx context.Context) {
 	g.setState(stateRunning)
-	go g.handleSignals()
+	go g.handleSignals(ctx)
 	c := make(chan struct{})
 	go func() {
 		defer close(c)
@@ -69,9 +70,7 @@ func (g *gracefulManager) Run() {
 	}
 }
 
-func (g *gracefulManager) handleSignals() {
-	var sig os.Signal
-
+func (g *gracefulManager) handleSignals(ctx context.Context) {
 	signalChannel := make(chan os.Signal, 1)
 
 	signal.Notify(
@@ -86,35 +85,40 @@ func (g *gracefulManager) handleSignals() {
 
 	pid := syscall.Getpid()
 	for {
-		sig = <-signalChannel
-		switch sig {
-		case syscall.SIGHUP:
-			if setting.GracefulRestartable {
-				log.Info("PID: %d. Received SIGHUP. Forking...", pid)
-				err := g.doFork()
-				if err != nil && err.Error() != "another process already forked. Ignoring this one" {
-					log.Error("Error whilst forking from PID: %d : %v", pid, err)
-				}
-			} else {
-				log.Info("PID: %d. Received SIGHUP. Not set restartable. Shutting down...", pid)
+		select {
+		case sig := <-signalChannel:
+			switch sig {
+			case syscall.SIGHUP:
+				if setting.GracefulRestartable {
+					log.Info("PID: %d. Received SIGHUP. Forking...", pid)
+					err := g.doFork()
+					if err != nil && err.Error() != "another process already forked. Ignoring this one" {
+						log.Error("Error whilst forking from PID: %d : %v", pid, err)
+					}
+				} else {
+					log.Info("PID: %d. Received SIGHUP. Not set restartable. Shutting down...", pid)
 
+					g.doShutdown()
+				}
+			case syscall.SIGUSR1:
+				log.Info("PID %d. Received SIGUSR1.", pid)
+			case syscall.SIGUSR2:
+				log.Warn("PID %d. Received SIGUSR2. Hammering...", pid)
+				g.doHammerTime(0 * time.Second)
+			case syscall.SIGINT:
+				log.Warn("PID %d. Received SIGINT. Shutting down...", pid)
 				g.doShutdown()
+			case syscall.SIGTERM:
+				log.Warn("PID %d. Received SIGTERM. Shutting down...", pid)
+				g.doShutdown()
+			case syscall.SIGTSTP:
+				log.Info("PID %d. Received SIGTSTP.", pid)
+			default:
+				log.Info("PID %d. Received %v.", pid, sig)
 			}
-		case syscall.SIGUSR1:
-			log.Info("PID %d. Received SIGUSR1.", pid)
-		case syscall.SIGUSR2:
-			log.Warn("PID %d. Received SIGUSR2. Hammering...", pid)
-			g.doHammerTime(0 * time.Second)
-		case syscall.SIGINT:
-			log.Warn("PID %d. Received SIGINT. Shutting down...", pid)
+		case <-ctx.Done():
+			log.Warn("PID: %d. Background context for manager closed - %v - Shutting down...", pid, ctx.Err())
 			g.doShutdown()
-		case syscall.SIGTERM:
-			log.Warn("PID %d. Received SIGTERM. Shutting down...", pid)
-			g.doShutdown()
-		case syscall.SIGTSTP:
-			log.Info("PID %d. Received SIGTSTP.", pid)
-		default:
-			log.Info("PID %d. Received %v.", pid, sig)
 		}
 	}
 }

modules/graceful/manager_windows.go

@@ -8,6 +8,7 @@
 package graceful
 
 import (
+	"context"
 	"os"
 	"strconv"
 	"sync"
@@ -29,6 +30,7 @@ const (
 )
 
 type gracefulManager struct {
+	ctx                    context.Context
 	isChild                bool
 	lock                   *sync.RWMutex
 	state                  state
@@ -40,10 +42,11 @@ type gracefulManager struct {
 	terminateWaitGroup     sync.WaitGroup
 }
 
-func newGracefulManager() *gracefulManager {
+func newGracefulManager(ctx context.Context) *gracefulManager {
 	manager := &gracefulManager{
 		isChild: false,
 		lock:    &sync.RWMutex{},
+		ctx:     ctx,
 	}
 	manager.createServerWaitGroup.Add(numberOfServersToCreate)
 	manager.Run()
@@ -89,23 +92,29 @@ func (g *gracefulManager) Execute(args []string, changes <-chan svc.ChangeReques
 	waitTime := 30 * time.Second
 
 loop:
-	for change := range changes {
-		switch change.Cmd {
-		case svc.Interrogate:
-			status <- change.CurrentStatus
-		case svc.Stop, svc.Shutdown:
+	for {
+		select {
+		case <-g.ctx.Done():
 			g.doShutdown()
 			waitTime += setting.GracefulHammerTime
 			break loop
-		case hammerCode:
-			g.doShutdown()
-			g.doHammerTime(0 * time.Second)
-			break loop
-		default:
-			log.Debug("Unexpected control request: %v", change.Cmd)
+		case change := <-changes:
+			switch change.Cmd {
+			case svc.Interrogate:
+				status <- change.CurrentStatus
+			case svc.Stop, svc.Shutdown:
+				g.doShutdown()
+				waitTime += setting.GracefulHammerTime
+				break loop
+			case hammerCode:
+				g.doShutdown()
+				g.doHammerTime(0 * time.Second)
+				break loop
+			default:
+				log.Debug("Unexpected control request: %v", change.Cmd)
+			}
 		}
 	}
-
 	status <- svc.Status{
 		State:    svc.StopPending,
 		WaitHint: uint32(waitTime / time.Millisecond),

modules/process/manager.go

@@ -12,6 +12,7 @@ import (
 	"fmt"
 	"io"
 	"os/exec"
+	"sort"
 	"sync"
 	"time"
 )
@@ -24,14 +25,17 @@ var (
 	// ErrExecTimeout represent a timeout error
 	ErrExecTimeout = errors.New("Process execution timeout")
 	manager        *Manager
+
+	// DefaultContext is the default context to run processing commands in
+	DefaultContext = context.Background()
 )
 
-// Process represents a working process inherit from Gogs.
+// Process represents a working process inheriting from Gitea.
 type Process struct {
 	PID         int64 // Process ID, not system one.
 	Description string
 	Start       time.Time
-	Cmd         *exec.Cmd
+	Cancel      context.CancelFunc
 }
 
 // Manager knows about all processes and counts PIDs.
@@ -39,28 +43,28 @@ type Manager struct {
 	mutex sync.Mutex
 
 	counter   int64
-	Processes map[int64]*Process
+	processes map[int64]*Process
 }
 
 // GetManager returns a Manager and initializes one as singleton if there's none yet
 func GetManager() *Manager {
 	if manager == nil {
 		manager = &Manager{
-			Processes: make(map[int64]*Process),
+			processes: make(map[int64]*Process),
 		}
 	}
 	return manager
 }
 
 // Add a process to the ProcessManager and returns its PID.
-func (pm *Manager) Add(description string, cmd *exec.Cmd) int64 {
+func (pm *Manager) Add(description string, cancel context.CancelFunc) int64 {
 	pm.mutex.Lock()
 	pid := pm.counter + 1
-	pm.Processes[pid] = &Process{
+	pm.processes[pid] = &Process{
 		PID:         pid,
 		Description: description,
 		Start:       time.Now(),
-		Cmd:         cmd,
+		Cancel:      cancel,
 	}
 	pm.counter = pid
 	pm.mutex.Unlock()
@@ -71,10 +75,32 @@ func (pm *Manager) Add(description string, cmd *exec.Cmd) int64 {
 // Remove a process from the ProcessManager.
 func (pm *Manager) Remove(pid int64) {
 	pm.mutex.Lock()
-	delete(pm.Processes, pid)
+	delete(pm.processes, pid)
 	pm.mutex.Unlock()
 }
 
+// Cancel a process in the ProcessManager.
+func (pm *Manager) Cancel(pid int64) {
+	pm.mutex.Lock()
+	process, ok := pm.processes[pid]
+	pm.mutex.Unlock()
+	if ok {
+		process.Cancel()
+	}
+}
+
+// Processes gets the processes in a thread safe manner
+func (pm *Manager) Processes() []*Process {
+	pm.mutex.Lock()
+	processes := make([]*Process, 0, len(pm.processes))
+	for _, process := range pm.processes {
+		processes = append(processes, process)
+	}
+	pm.mutex.Unlock()
+	sort.Sort(processList(processes))
+	return processes
+}
+
 // Exec a command and use the default timeout.
 func (pm *Manager) Exec(desc, cmdName string, args ...string) (string, string, error) {
 	return pm.ExecDir(-1, "", desc, cmdName, args...)
@@ -110,7 +136,7 @@ func (pm *Manager) ExecDirEnvStdIn(timeout time.Duration, dir, desc string, env
 	stdOut := new(bytes.Buffer)
 	stdErr := new(bytes.Buffer)
 
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	ctx, cancel := context.WithTimeout(DefaultContext, timeout)
 	defer cancel()
 
 	cmd := exec.CommandContext(ctx, cmdName, args...)
@@ -126,7 +152,7 @@ func (pm *Manager) ExecDirEnvStdIn(timeout time.Duration, dir, desc string, env
 		return "", "", err
 	}
 
-	pid := pm.Add(desc, cmd)
+	pid := pm.Add(desc, cancel)
 	err := cmd.Wait()
 	pm.Remove(pid)
 
@@ -137,21 +163,16 @@ func (pm *Manager) ExecDirEnvStdIn(timeout time.Duration, dir, desc string, env
 	return stdOut.String(), stdErr.String(), err
 }
 
-// Kill and remove a process from list.
-func (pm *Manager) Kill(pid int64) error {
-	if proc, exists := pm.Processes[pid]; exists {
-		pm.mutex.Lock()
-		if proc.Cmd != nil &&
-			proc.Cmd.Process != nil &&
-			proc.Cmd.ProcessState != nil &&
-			!proc.Cmd.ProcessState.Exited() {
-			if err := proc.Cmd.Process.Kill(); err != nil {
-				return fmt.Errorf("failed to kill process(%d/%s): %v", pid, proc.Description, err)
-			}
-		}
-		delete(pm.Processes, pid)
-		pm.mutex.Unlock()
-	}
+type processList []*Process
 
-	return nil
+func (l processList) Len() int {
+	return len(l)
+}
+
+func (l processList) Less(i, j int) bool {
+	return l[i].PID < l[j].PID
+}
+
+func (l processList) Swap(i, j int) {
+	l[i], l[j] = l[j], l[i]
 }

modules/process/manager_test.go

@@ -1,7 +1,7 @@
 package process
 
 import (
-	"os/exec"
+	"context"
 	"testing"
 	"time"
 
@@ -9,27 +9,42 @@ import (
 )
 
 func TestManager_Add(t *testing.T) {
-	pm := Manager{Processes: make(map[int64]*Process)}
+	pm := Manager{processes: make(map[int64]*Process)}
 
-	pid := pm.Add("foo", exec.Command("foo"))
+	pid := pm.Add("foo", nil)
 	assert.Equal(t, int64(1), pid, "expected to get pid 1 got %d", pid)
 
-	pid = pm.Add("bar", exec.Command("bar"))
+	pid = pm.Add("bar", nil)
 	assert.Equal(t, int64(2), pid, "expected to get pid 2 got %d", pid)
 }
 
-func TestManager_Remove(t *testing.T) {
-	pm := Manager{Processes: make(map[int64]*Process)}
+func TestManager_Cancel(t *testing.T) {
+	pm := Manager{processes: make(map[int64]*Process)}
 
-	pid1 := pm.Add("foo", exec.Command("foo"))
+	ctx, cancel := context.WithCancel(context.Background())
+	pid := pm.Add("foo", cancel)
+
+	pm.Cancel(pid)
+
+	select {
+	case <-ctx.Done():
+	default:
+		assert.Fail(t, "Cancel should cancel the provided context")
+	}
+}
+
+func TestManager_Remove(t *testing.T) {
+	pm := Manager{processes: make(map[int64]*Process)}
+
+	pid1 := pm.Add("foo", nil)
 	assert.Equal(t, int64(1), pid1, "expected to get pid 1 got %d", pid1)
 
-	pid2 := pm.Add("bar", exec.Command("bar"))
+	pid2 := pm.Add("bar", nil)
 	assert.Equal(t, int64(2), pid2, "expected to get pid 2 got %d", pid2)
 
 	pm.Remove(pid2)
 
-	_, exists := pm.Processes[pid2]
+	_, exists := pm.processes[pid2]
 	assert.False(t, exists, "PID %d is in the list but shouldn't", pid2)
 }