mirror of
https://github.com/kemko/nomad.git
synced 2026-01-04 17:35:43 +03:00
executor: synchronize exitState accesses
exitState is set in `wait()` goroutine but accessed in a different `Wait()` goroutine, so accesses must be synchronized by a lock.
This commit is contained in:
@@ -10,6 +10,7 @@ import (
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -212,6 +213,8 @@ type UniversalExecutor struct {
|
||||
commandCfg *ExecCommand
|
||||
|
||||
exitState *ProcessState
|
||||
exitStateLock sync.Mutex
|
||||
|
||||
processExited chan interface{}
|
||||
|
||||
// resConCtx is used to track and cleanup additional resources created by
|
||||
@@ -364,6 +367,8 @@ func (e *UniversalExecutor) Wait(ctx context.Context) (*ProcessState, error) {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case <-e.processExited:
|
||||
e.exitStateLock.Lock()
|
||||
defer e.exitStateLock.Unlock()
|
||||
return e.exitState, nil
|
||||
}
|
||||
}
|
||||
@@ -378,7 +383,9 @@ func (e *UniversalExecutor) wait() {
|
||||
pid := e.childCmd.Process.Pid
|
||||
err := e.childCmd.Wait()
|
||||
if err == nil {
|
||||
e.exitStateLock.Lock()
|
||||
e.exitState = &ProcessState{Pid: pid, ExitCode: 0, Time: time.Now()}
|
||||
e.exitStateLock.Unlock()
|
||||
return
|
||||
}
|
||||
|
||||
@@ -404,7 +411,9 @@ func (e *UniversalExecutor) wait() {
|
||||
e.logger.Warn("unexpected Cmd.Wait() error type", "error", err)
|
||||
}
|
||||
|
||||
e.exitStateLock.Lock()
|
||||
e.exitState = &ProcessState{Pid: pid, ExitCode: exitCode, Signal: signal, Time: time.Now()}
|
||||
e.exitStateLock.Unlock()
|
||||
}
|
||||
|
||||
var (
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"path"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -78,7 +79,9 @@ type LibcontainerExecutor struct {
|
||||
container libcontainer.Container
|
||||
userProc *libcontainer.Process
|
||||
userProcExited chan interface{}
|
||||
exitState *ProcessState
|
||||
|
||||
exitState *ProcessState
|
||||
exitStateLock sync.Mutex
|
||||
}
|
||||
|
||||
func NewExecutorWithIsolation(logger hclog.Logger) Executor {
|
||||
@@ -248,6 +251,8 @@ func (l *LibcontainerExecutor) Wait(ctx context.Context) (*ProcessState, error)
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case <-l.userProcExited:
|
||||
l.exitStateLock.Lock()
|
||||
defer l.exitStateLock.Unlock()
|
||||
return l.exitState, nil
|
||||
}
|
||||
}
|
||||
@@ -263,7 +268,10 @@ func (l *LibcontainerExecutor) wait() {
|
||||
ps = exitErr.ProcessState
|
||||
} else {
|
||||
l.logger.Error("failed to call wait on user process", "error", err)
|
||||
l.exitStateLock.Lock()
|
||||
l.exitState = &ProcessState{Pid: 0, ExitCode: 1, Time: time.Now()}
|
||||
l.exitStateLock.Unlock()
|
||||
|
||||
return
|
||||
}
|
||||
}
|
||||
@@ -281,12 +289,14 @@ func (l *LibcontainerExecutor) wait() {
|
||||
}
|
||||
}
|
||||
|
||||
l.exitStateLock.Lock()
|
||||
l.exitState = &ProcessState{
|
||||
Pid: ps.Pid(),
|
||||
ExitCode: exitCode,
|
||||
Signal: signal,
|
||||
Time: time.Now(),
|
||||
}
|
||||
l.exitStateLock.Unlock()
|
||||
}
|
||||
|
||||
// Shutdown stops all processes started and cleans up any resources
|
||||
|
||||
Reference in New Issue
Block a user