Merge pull request #2773 from crosbymichael/state-locking

Fix process locking and state management
This commit is contained in:
Michael Crosby 2018-11-16 14:15:04 -05:00 committed by GitHub
commit d48d7464ad
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 178 additions and 227 deletions

View File

@ -40,7 +40,6 @@ func (s Stdio) IsNull() bool {
// Process on a system // Process on a system
type Process interface { type Process interface {
State
// ID returns the id for the process // ID returns the id for the process
ID() string ID() string
// Pid returns the pid for the process // Pid returns the pid for the process
@ -57,10 +56,6 @@ type Process interface {
Status(context.Context) (string, error) Status(context.Context) (string, error)
// Wait blocks until the process has exited // Wait blocks until the process has exited
Wait() Wait()
}
// State of a process
type State interface {
// Resize resizes the process console // Resize resizes the process console
Resize(ws console.WinSize) error Resize(ws console.WinSize) error
// Start execution of the process // Start execution of the process

View File

@ -41,7 +41,7 @@ import (
type execProcess struct { type execProcess struct {
wg sync.WaitGroup wg sync.WaitGroup
proc.State execState execState
mu sync.Mutex mu sync.Mutex
id string id string
@ -86,6 +86,13 @@ func (e *execProcess) ExitedAt() time.Time {
return e.exited return e.exited
} }
func (e *execProcess) SetExited(status int) {
e.mu.Lock()
defer e.mu.Unlock()
e.execState.SetExited(status)
}
func (e *execProcess) setExited(status int) { func (e *execProcess) setExited(status int) {
e.status = status e.status = status
e.exited = time.Now() e.exited = time.Now()
@ -93,6 +100,13 @@ func (e *execProcess) setExited(status int) {
close(e.waitBlock) close(e.waitBlock)
} }
func (e *execProcess) Delete(ctx context.Context) error {
e.mu.Lock()
defer e.mu.Unlock()
return e.execState.Delete(ctx)
}
func (e *execProcess) delete(ctx context.Context) error { func (e *execProcess) delete(ctx context.Context) error {
e.wg.Wait() e.wg.Wait()
if e.io != nil { if e.io != nil {
@ -107,6 +121,13 @@ func (e *execProcess) delete(ctx context.Context) error {
return nil return nil
} }
func (e *execProcess) Resize(ws console.WinSize) error {
e.mu.Lock()
defer e.mu.Unlock()
return e.execState.Resize(ws)
}
func (e *execProcess) resize(ws console.WinSize) error { func (e *execProcess) resize(ws console.WinSize) error {
if e.console == nil { if e.console == nil {
return nil return nil
@ -114,6 +135,13 @@ func (e *execProcess) resize(ws console.WinSize) error {
return e.console.Resize(ws) return e.console.Resize(ws)
} }
func (e *execProcess) Kill(ctx context.Context, sig uint32, _ bool) error {
e.mu.Lock()
defer e.mu.Unlock()
return e.execState.Kill(ctx, sig, false)
}
func (e *execProcess) kill(ctx context.Context, sig uint32, _ bool) error { func (e *execProcess) kill(ctx context.Context, sig uint32, _ bool) error {
pid := e.pid pid := e.pid
if pid != 0 { if pid != 0 {
@ -132,6 +160,13 @@ func (e *execProcess) Stdio() proc.Stdio {
return e.stdio return e.stdio
} }
func (e *execProcess) Start(ctx context.Context) error {
e.mu.Lock()
defer e.mu.Unlock()
return e.execState.Start(ctx)
}
func (e *execProcess) start(ctx context.Context) (err error) { func (e *execProcess) start(ctx context.Context) (err error) {
var ( var (
socket *runc.Socket socket *runc.Socket

View File

@ -25,6 +25,14 @@ import (
"github.com/pkg/errors" "github.com/pkg/errors"
) )
type execState interface {
Resize(console.WinSize) error
Start(context.Context) error
Delete(context.Context) error
Kill(context.Context, uint32, bool) error
SetExited(int)
}
type execCreatedState struct { type execCreatedState struct {
p *execProcess p *execProcess
} }
@ -32,11 +40,11 @@ type execCreatedState struct {
func (s *execCreatedState) transition(name string) error { func (s *execCreatedState) transition(name string) error {
switch name { switch name {
case "running": case "running":
s.p.State = &execRunningState{p: s.p} s.p.execState = &execRunningState{p: s.p}
case "stopped": case "stopped":
s.p.State = &execStoppedState{p: s.p} s.p.execState = &execStoppedState{p: s.p}
case "deleted": case "deleted":
s.p.State = &deletedState{} s.p.execState = &deletedState{}
default: default:
return errors.Errorf("invalid state transition %q to %q", stateName(s), name) return errors.Errorf("invalid state transition %q to %q", stateName(s), name)
} }
@ -44,15 +52,10 @@ func (s *execCreatedState) transition(name string) error {
} }
func (s *execCreatedState) Resize(ws console.WinSize) error { func (s *execCreatedState) Resize(ws console.WinSize) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.resize(ws) return s.p.resize(ws)
} }
func (s *execCreatedState) Start(ctx context.Context) error { func (s *execCreatedState) Start(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
if err := s.p.start(ctx); err != nil { if err := s.p.start(ctx); err != nil {
return err return err
} }
@ -63,22 +66,15 @@ func (s *execCreatedState) Delete(ctx context.Context) error {
if err := s.p.delete(ctx); err != nil { if err := s.p.delete(ctx); err != nil {
return err return err
} }
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.transition("deleted") return s.transition("deleted")
} }
func (s *execCreatedState) Kill(ctx context.Context, sig uint32, all bool) error { func (s *execCreatedState) Kill(ctx context.Context, sig uint32, all bool) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.kill(ctx, sig, all) return s.p.kill(ctx, sig, all)
} }
func (s *execCreatedState) SetExited(status int) { func (s *execCreatedState) SetExited(status int) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
s.p.setExited(status) s.p.setExited(status)
if err := s.transition("stopped"); err != nil { if err := s.transition("stopped"); err != nil {
@ -93,7 +89,7 @@ type execRunningState struct {
func (s *execRunningState) transition(name string) error { func (s *execRunningState) transition(name string) error {
switch name { switch name {
case "stopped": case "stopped":
s.p.State = &execStoppedState{p: s.p} s.p.execState = &execStoppedState{p: s.p}
default: default:
return errors.Errorf("invalid state transition %q to %q", stateName(s), name) return errors.Errorf("invalid state transition %q to %q", stateName(s), name)
} }
@ -101,37 +97,22 @@ func (s *execRunningState) transition(name string) error {
} }
func (s *execRunningState) Resize(ws console.WinSize) error { func (s *execRunningState) Resize(ws console.WinSize) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.resize(ws) return s.p.resize(ws)
} }
func (s *execRunningState) Start(ctx context.Context) error { func (s *execRunningState) Start(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot start a running process") return errors.Errorf("cannot start a running process")
} }
func (s *execRunningState) Delete(ctx context.Context) error { func (s *execRunningState) Delete(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot delete a running process") return errors.Errorf("cannot delete a running process")
} }
func (s *execRunningState) Kill(ctx context.Context, sig uint32, all bool) error { func (s *execRunningState) Kill(ctx context.Context, sig uint32, all bool) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.kill(ctx, sig, all) return s.p.kill(ctx, sig, all)
} }
func (s *execRunningState) SetExited(status int) { func (s *execRunningState) SetExited(status int) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
s.p.setExited(status) s.p.setExited(status)
if err := s.transition("stopped"); err != nil { if err := s.transition("stopped"); err != nil {
@ -146,7 +127,7 @@ type execStoppedState struct {
func (s *execStoppedState) transition(name string) error { func (s *execStoppedState) transition(name string) error {
switch name { switch name {
case "deleted": case "deleted":
s.p.State = &deletedState{} s.p.execState = &deletedState{}
default: default:
return errors.Errorf("invalid state transition %q to %q", stateName(s), name) return errors.Errorf("invalid state transition %q to %q", stateName(s), name)
} }
@ -154,16 +135,10 @@ func (s *execStoppedState) transition(name string) error {
} }
func (s *execStoppedState) Resize(ws console.WinSize) error { func (s *execStoppedState) Resize(ws console.WinSize) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot resize a stopped container") return errors.Errorf("cannot resize a stopped container")
} }
func (s *execStoppedState) Start(ctx context.Context) error { func (s *execStoppedState) Start(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot start a stopped process") return errors.Errorf("cannot start a stopped process")
} }
@ -171,15 +146,11 @@ func (s *execStoppedState) Delete(ctx context.Context) error {
if err := s.p.delete(ctx); err != nil { if err := s.p.delete(ctx); err != nil {
return err return err
} }
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.transition("deleted") return s.transition("deleted")
} }
func (s *execStoppedState) Kill(ctx context.Context, sig uint32, all bool) error { func (s *execStoppedState) Kill(ctx context.Context, sig uint32, all bool) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.kill(ctx, sig, all) return s.p.kill(ctx, sig, all)
} }

View File

@ -47,7 +47,7 @@ const InitPidFile = "init.pid"
// Init represents an initial process for a container // Init represents an initial process for a container
type Init struct { type Init struct {
wg sync.WaitGroup wg sync.WaitGroup
initState initState initState
// mu is used to ensure that `Start()` and `Exited()` calls return in // mu is used to ensure that `Start()` and `Exited()` calls return in
// the right order when invoked in separate go routines. // the right order when invoked in separate go routines.
@ -212,6 +212,7 @@ func (p *Init) Pid() int {
func (p *Init) ExitStatus() int { func (p *Init) ExitStatus() int {
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
return p.status return p.status
} }
@ -219,6 +220,7 @@ func (p *Init) ExitStatus() int {
func (p *Init) ExitedAt() time.Time { func (p *Init) ExitedAt() time.Time {
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
return p.exited return p.exited
} }
@ -226,6 +228,7 @@ func (p *Init) ExitedAt() time.Time {
func (p *Init) Status(ctx context.Context) (string, error) { func (p *Init) Status(ctx context.Context) (string, error) {
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
c, err := p.runtime.State(ctx, p.id) c, err := p.runtime.State(ctx, p.id)
if err != nil { if err != nil {
if strings.Contains(err.Error(), "does not exist") { if strings.Contains(err.Error(), "does not exist") {
@ -236,11 +239,27 @@ func (p *Init) Status(ctx context.Context) (string, error) {
return c.Status, nil return c.Status, nil
} }
func (p *Init) start(context context.Context) error { // Start the init process
err := p.runtime.Start(context, p.id) func (p *Init) Start(ctx context.Context) error {
p.mu.Lock()
defer p.mu.Unlock()
return p.initState.Start(ctx)
}
func (p *Init) start(ctx context.Context) error {
err := p.runtime.Start(ctx, p.id)
return p.runtimeError(err, "OCI runtime start failed") return p.runtimeError(err, "OCI runtime start failed")
} }
// SetExited of the init process with the next status
func (p *Init) SetExited(status int) {
p.mu.Lock()
defer p.mu.Unlock()
p.initState.SetExited(status)
}
func (p *Init) setExited(status int) { func (p *Init) setExited(status int) {
p.exited = time.Now() p.exited = time.Now()
p.status = status p.status = status
@ -248,9 +267,17 @@ func (p *Init) setExited(status int) {
close(p.waitBlock) close(p.waitBlock)
} }
func (p *Init) delete(context context.Context) error { // Delete the init process
func (p *Init) Delete(ctx context.Context) error {
p.mu.Lock()
defer p.mu.Unlock()
return p.initState.Delete(ctx)
}
func (p *Init) delete(ctx context.Context) error {
p.wg.Wait() p.wg.Wait()
err := p.runtime.Delete(context, p.id, nil) err := p.runtime.Delete(ctx, p.id, nil)
// ignore errors if a runtime has already deleted the process // ignore errors if a runtime has already deleted the process
// but we still hold metadata and pipes // but we still hold metadata and pipes
// //
@ -270,7 +297,7 @@ func (p *Init) delete(context context.Context) error {
p.io.Close() p.io.Close()
} }
if err2 := mount.UnmountAll(p.Rootfs, 0); err2 != nil { if err2 := mount.UnmountAll(p.Rootfs, 0); err2 != nil {
log.G(context).WithError(err2).Warn("failed to cleanup rootfs mount") log.G(ctx).WithError(err2).Warn("failed to cleanup rootfs mount")
if err == nil { if err == nil {
err = errors.Wrap(err2, "failed rootfs umount") err = errors.Wrap(err2, "failed rootfs umount")
} }
@ -278,6 +305,17 @@ func (p *Init) delete(context context.Context) error {
return err return err
} }
// Resize the init processes console
func (p *Init) Resize(ws console.WinSize) error {
p.mu.Lock()
defer p.mu.Unlock()
if p.console == nil {
return nil
}
return p.console.Resize(ws)
}
func (p *Init) resize(ws console.WinSize) error { func (p *Init) resize(ws console.WinSize) error {
if p.console == nil { if p.console == nil {
return nil return nil
@ -285,26 +323,40 @@ func (p *Init) resize(ws console.WinSize) error {
return p.console.Resize(ws) return p.console.Resize(ws)
} }
func (p *Init) pause(context context.Context) error { // Pause the init process and all its child processes
err := p.runtime.Pause(context, p.id) func (p *Init) Pause(ctx context.Context) error {
return p.runtimeError(err, "OCI runtime pause failed") p.mu.Lock()
defer p.mu.Unlock()
return p.initState.Pause(ctx)
} }
func (p *Init) resume(context context.Context) error { // Resume the init process and all its child processes
err := p.runtime.Resume(context, p.id) func (p *Init) Resume(ctx context.Context) error {
return p.runtimeError(err, "OCI runtime resume failed") p.mu.Lock()
defer p.mu.Unlock()
return p.initState.Resume(ctx)
} }
func (p *Init) kill(context context.Context, signal uint32, all bool) error { // Kill the init process
err := p.runtime.Kill(context, p.id, int(signal), &runc.KillOpts{ func (p *Init) Kill(ctx context.Context, signal uint32, all bool) error {
p.mu.Lock()
defer p.mu.Unlock()
return p.initState.Kill(ctx, signal, all)
}
func (p *Init) kill(ctx context.Context, signal uint32, all bool) error {
err := p.runtime.Kill(ctx, p.id, int(signal), &runc.KillOpts{
All: all, All: all,
}) })
return checkKillError(err) return checkKillError(err)
} }
// KillAll processes belonging to the init process // KillAll processes belonging to the init process
func (p *Init) KillAll(context context.Context) error { func (p *Init) KillAll(ctx context.Context) error {
err := p.runtime.Kill(context, p.id, int(syscall.SIGKILL), &runc.KillOpts{ err := p.runtime.Kill(ctx, p.id, int(syscall.SIGKILL), &runc.KillOpts{
All: true, All: true,
}) })
return p.runtimeError(err, "OCI runtime killall failed") return p.runtimeError(err, "OCI runtime killall failed")
@ -320,8 +372,16 @@ func (p *Init) Runtime() *runc.Runc {
return p.runtime return p.runtime
} }
// Exec returns a new child process
func (p *Init) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) {
p.mu.Lock()
defer p.mu.Unlock()
return p.initState.Exec(ctx, path, r)
}
// exec returns a new exec'd process // exec returns a new exec'd process
func (p *Init) exec(context context.Context, path string, r *ExecConfig) (proc.Process, error) { func (p *Init) exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) {
// process exec request // process exec request
var spec specs.Process var spec specs.Process
if err := json.Unmarshal(r.Spec.Value, &spec); err != nil { if err := json.Unmarshal(r.Spec.Value, &spec); err != nil {
@ -342,18 +402,26 @@ func (p *Init) exec(context context.Context, path string, r *ExecConfig) (proc.P
}, },
waitBlock: make(chan struct{}), waitBlock: make(chan struct{}),
} }
e.State = &execCreatedState{p: e} e.execState = &execCreatedState{p: e}
return e, nil return e, nil
} }
func (p *Init) checkpoint(context context.Context, r *CheckpointConfig) error { // Checkpoint the init process
func (p *Init) Checkpoint(ctx context.Context, r *CheckpointConfig) error {
p.mu.Lock()
defer p.mu.Unlock()
return p.initState.Checkpoint(ctx, r)
}
func (p *Init) checkpoint(ctx context.Context, r *CheckpointConfig) error {
var actions []runc.CheckpointAction var actions []runc.CheckpointAction
if !r.Exit { if !r.Exit {
actions = append(actions, runc.LeaveRunning) actions = append(actions, runc.LeaveRunning)
} }
work := filepath.Join(p.WorkDir, "criu-work") work := filepath.Join(p.WorkDir, "criu-work")
defer os.RemoveAll(work) defer os.RemoveAll(work)
if err := p.runtime.Checkpoint(context, p.id, &runc.CheckpointOpts{ if err := p.runtime.Checkpoint(ctx, p.id, &runc.CheckpointOpts{
WorkDir: work, WorkDir: work,
ImagePath: r.Path, ImagePath: r.Path,
AllowOpenTCP: r.AllowOpenTCP, AllowOpenTCP: r.AllowOpenTCP,
@ -364,19 +432,27 @@ func (p *Init) checkpoint(context context.Context, r *CheckpointConfig) error {
}, actions...); err != nil { }, actions...); err != nil {
dumpLog := filepath.Join(p.Bundle, "criu-dump.log") dumpLog := filepath.Join(p.Bundle, "criu-dump.log")
if cerr := copyFile(dumpLog, filepath.Join(work, "dump.log")); cerr != nil { if cerr := copyFile(dumpLog, filepath.Join(work, "dump.log")); cerr != nil {
log.G(context).Error(err) log.G(ctx).Error(err)
} }
return fmt.Errorf("%s path= %s", criuError(err), dumpLog) return fmt.Errorf("%s path= %s", criuError(err), dumpLog)
} }
return nil return nil
} }
func (p *Init) update(context context.Context, r *google_protobuf.Any) error { // Update the processes resource configuration
func (p *Init) Update(ctx context.Context, r *google_protobuf.Any) error {
p.mu.Lock()
defer p.mu.Unlock()
return p.initState.Update(ctx, r)
}
func (p *Init) update(ctx context.Context, r *google_protobuf.Any) error {
var resources specs.LinuxResources var resources specs.LinuxResources
if err := json.Unmarshal(r.Value, &resources); err != nil { if err := json.Unmarshal(r.Value, &resources); err != nil {
return err return err
} }
return p.runtime.Update(context, p.id, &resources) return p.runtime.Update(ctx, p.id, &resources)
} }
// Stdio of the process // Stdio of the process

View File

@ -30,16 +30,20 @@ import (
runc "github.com/containerd/go-runc" runc "github.com/containerd/go-runc"
google_protobuf "github.com/gogo/protobuf/types" google_protobuf "github.com/gogo/protobuf/types"
"github.com/pkg/errors" "github.com/pkg/errors"
"github.com/sirupsen/logrus"
) )
type initState interface { type initState interface {
proc.State Resize(console.WinSize) error
Start(context.Context) error
Delete(context.Context) error
Pause(context.Context) error Pause(context.Context) error
Resume(context.Context) error Resume(context.Context) error
Update(context.Context, *google_protobuf.Any) error Update(context.Context, *google_protobuf.Any) error
Checkpoint(context.Context, *CheckpointConfig) error Checkpoint(context.Context, *CheckpointConfig) error
Exec(context.Context, string, *ExecConfig) (proc.Process, error) Exec(context.Context, string, *ExecConfig) (proc.Process, error)
Kill(context.Context, uint32, bool) error
SetExited(int)
} }
type createdState struct { type createdState struct {
@ -61,43 +65,26 @@ func (s *createdState) transition(name string) error {
} }
func (s *createdState) Pause(ctx context.Context) error { func (s *createdState) Pause(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot pause task in created state") return errors.Errorf("cannot pause task in created state")
} }
func (s *createdState) Resume(ctx context.Context) error { func (s *createdState) Resume(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot resume task in created state") return errors.Errorf("cannot resume task in created state")
} }
func (s *createdState) Update(context context.Context, r *google_protobuf.Any) error { func (s *createdState) Update(ctx context.Context, r *google_protobuf.Any) error {
s.p.mu.Lock() return s.p.update(ctx, r)
defer s.p.mu.Unlock()
return s.p.update(context, r)
} }
func (s *createdState) Checkpoint(context context.Context, r *CheckpointConfig) error { func (s *createdState) Checkpoint(ctx context.Context, r *CheckpointConfig) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot checkpoint a task in created state") return errors.Errorf("cannot checkpoint a task in created state")
} }
func (s *createdState) Resize(ws console.WinSize) error { func (s *createdState) Resize(ws console.WinSize) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.resize(ws) return s.p.resize(ws)
} }
func (s *createdState) Start(ctx context.Context) error { func (s *createdState) Start(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
if err := s.p.start(ctx); err != nil { if err := s.p.start(ctx); err != nil {
return err return err
} }
@ -105,8 +92,6 @@ func (s *createdState) Start(ctx context.Context) error {
} }
func (s *createdState) Delete(ctx context.Context) error { func (s *createdState) Delete(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
if err := s.p.delete(ctx); err != nil { if err := s.p.delete(ctx); err != nil {
return err return err
} }
@ -114,16 +99,10 @@ func (s *createdState) Delete(ctx context.Context) error {
} }
func (s *createdState) Kill(ctx context.Context, sig uint32, all bool) error { func (s *createdState) Kill(ctx context.Context, sig uint32, all bool) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.kill(ctx, sig, all) return s.p.kill(ctx, sig, all)
} }
func (s *createdState) SetExited(status int) { func (s *createdState) SetExited(status int) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
s.p.setExited(status) s.p.setExited(status)
if err := s.transition("stopped"); err != nil { if err := s.transition("stopped"); err != nil {
@ -132,8 +111,6 @@ func (s *createdState) SetExited(status int) {
} }
func (s *createdState) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) { func (s *createdState) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.exec(ctx, path, r) return s.p.exec(ctx, path, r)
} }
@ -157,43 +134,26 @@ func (s *createdCheckpointState) transition(name string) error {
} }
func (s *createdCheckpointState) Pause(ctx context.Context) error { func (s *createdCheckpointState) Pause(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot pause task in created state") return errors.Errorf("cannot pause task in created state")
} }
func (s *createdCheckpointState) Resume(ctx context.Context) error { func (s *createdCheckpointState) Resume(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot resume task in created state") return errors.Errorf("cannot resume task in created state")
} }
func (s *createdCheckpointState) Update(context context.Context, r *google_protobuf.Any) error { func (s *createdCheckpointState) Update(ctx context.Context, r *google_protobuf.Any) error {
s.p.mu.Lock() return s.p.update(ctx, r)
defer s.p.mu.Unlock()
return s.p.update(context, r)
} }
func (s *createdCheckpointState) Checkpoint(context context.Context, r *CheckpointConfig) error { func (s *createdCheckpointState) Checkpoint(ctx context.Context, r *CheckpointConfig) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot checkpoint a task in created state") return errors.Errorf("cannot checkpoint a task in created state")
} }
func (s *createdCheckpointState) Resize(ws console.WinSize) error { func (s *createdCheckpointState) Resize(ws console.WinSize) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.resize(ws) return s.p.resize(ws)
} }
func (s *createdCheckpointState) Start(ctx context.Context) error { func (s *createdCheckpointState) Start(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
p := s.p p := s.p
sio := p.stdio sio := p.stdio
@ -247,8 +207,6 @@ func (s *createdCheckpointState) Start(ctx context.Context) error {
} }
func (s *createdCheckpointState) Delete(ctx context.Context) error { func (s *createdCheckpointState) Delete(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
if err := s.p.delete(ctx); err != nil { if err := s.p.delete(ctx); err != nil {
return err return err
} }
@ -256,16 +214,10 @@ func (s *createdCheckpointState) Delete(ctx context.Context) error {
} }
func (s *createdCheckpointState) Kill(ctx context.Context, sig uint32, all bool) error { func (s *createdCheckpointState) Kill(ctx context.Context, sig uint32, all bool) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.kill(ctx, sig, all) return s.p.kill(ctx, sig, all)
} }
func (s *createdCheckpointState) SetExited(status int) { func (s *createdCheckpointState) SetExited(status int) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
s.p.setExited(status) s.p.setExited(status)
if err := s.transition("stopped"); err != nil { if err := s.transition("stopped"); err != nil {
@ -274,9 +226,6 @@ func (s *createdCheckpointState) SetExited(status int) {
} }
func (s *createdCheckpointState) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) { func (s *createdCheckpointState) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return nil, errors.Errorf("cannot exec in a created state") return nil, errors.Errorf("cannot exec in a created state")
} }
@ -297,67 +246,42 @@ func (s *runningState) transition(name string) error {
} }
func (s *runningState) Pause(ctx context.Context) error { func (s *runningState) Pause(ctx context.Context) error {
s.p.mu.Lock() if err := s.p.runtime.Pause(ctx, s.p.id); err != nil {
defer s.p.mu.Unlock() return s.p.runtimeError(err, "OCI runtime pause failed")
if err := s.p.pause(ctx); err != nil {
return err
} }
return s.transition("paused") return s.transition("paused")
} }
func (s *runningState) Resume(ctx context.Context) error { func (s *runningState) Resume(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot resume a running process") return errors.Errorf("cannot resume a running process")
} }
func (s *runningState) Update(context context.Context, r *google_protobuf.Any) error { func (s *runningState) Update(ctx context.Context, r *google_protobuf.Any) error {
s.p.mu.Lock() return s.p.update(ctx, r)
defer s.p.mu.Unlock()
return s.p.update(context, r)
} }
func (s *runningState) Checkpoint(ctx context.Context, r *CheckpointConfig) error { func (s *runningState) Checkpoint(ctx context.Context, r *CheckpointConfig) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.checkpoint(ctx, r) return s.p.checkpoint(ctx, r)
} }
func (s *runningState) Resize(ws console.WinSize) error { func (s *runningState) Resize(ws console.WinSize) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.resize(ws) return s.p.resize(ws)
} }
func (s *runningState) Start(ctx context.Context) error { func (s *runningState) Start(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot start a running process") return errors.Errorf("cannot start a running process")
} }
func (s *runningState) Delete(ctx context.Context) error { func (s *runningState) Delete(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot delete a running process") return errors.Errorf("cannot delete a running process")
} }
func (s *runningState) Kill(ctx context.Context, sig uint32, all bool) error { func (s *runningState) Kill(ctx context.Context, sig uint32, all bool) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.kill(ctx, sig, all) return s.p.kill(ctx, sig, all)
} }
func (s *runningState) SetExited(status int) { func (s *runningState) SetExited(status int) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
s.p.setExited(status) s.p.setExited(status)
if err := s.transition("stopped"); err != nil { if err := s.transition("stopped"); err != nil {
@ -366,8 +290,6 @@ func (s *runningState) SetExited(status int) {
} }
func (s *runningState) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) { func (s *runningState) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.exec(ctx, path, r) return s.p.exec(ctx, path, r)
} }
@ -388,79 +310,54 @@ func (s *pausedState) transition(name string) error {
} }
func (s *pausedState) Pause(ctx context.Context) error { func (s *pausedState) Pause(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot pause a paused container") return errors.Errorf("cannot pause a paused container")
} }
func (s *pausedState) Resume(ctx context.Context) error { func (s *pausedState) Resume(ctx context.Context) error {
s.p.mu.Lock() if err := s.p.runtime.Resume(ctx, s.p.id); err != nil {
defer s.p.mu.Unlock() return s.p.runtimeError(err, "OCI runtime resume failed")
if err := s.p.resume(ctx); err != nil {
return err
} }
return s.transition("running") return s.transition("running")
} }
func (s *pausedState) Update(context context.Context, r *google_protobuf.Any) error { func (s *pausedState) Update(ctx context.Context, r *google_protobuf.Any) error {
s.p.mu.Lock() return s.p.update(ctx, r)
defer s.p.mu.Unlock()
return s.p.update(context, r)
} }
func (s *pausedState) Checkpoint(ctx context.Context, r *CheckpointConfig) error { func (s *pausedState) Checkpoint(ctx context.Context, r *CheckpointConfig) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.checkpoint(ctx, r) return s.p.checkpoint(ctx, r)
} }
func (s *pausedState) Resize(ws console.WinSize) error { func (s *pausedState) Resize(ws console.WinSize) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.resize(ws) return s.p.resize(ws)
} }
func (s *pausedState) Start(ctx context.Context) error { func (s *pausedState) Start(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot start a paused process") return errors.Errorf("cannot start a paused process")
} }
func (s *pausedState) Delete(ctx context.Context) error { func (s *pausedState) Delete(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot delete a paused process") return errors.Errorf("cannot delete a paused process")
} }
func (s *pausedState) Kill(ctx context.Context, sig uint32, all bool) error { func (s *pausedState) Kill(ctx context.Context, sig uint32, all bool) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return s.p.kill(ctx, sig, all) return s.p.kill(ctx, sig, all)
} }
func (s *pausedState) SetExited(status int) { func (s *pausedState) SetExited(status int) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
s.p.setExited(status) s.p.setExited(status)
if err := s.p.runtime.Resume(context.Background(), s.p.id); err != nil {
logrus.WithError(err).Error("resuming exited container from paused state")
}
if err := s.transition("stopped"); err != nil { if err := s.transition("stopped"); err != nil {
panic(err) panic(err)
} }
} }
func (s *pausedState) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) { func (s *pausedState) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return nil, errors.Errorf("cannot exec in a paused state") return nil, errors.Errorf("cannot exec in a paused state")
} }
@ -479,50 +376,30 @@ func (s *stoppedState) transition(name string) error {
} }
func (s *stoppedState) Pause(ctx context.Context) error { func (s *stoppedState) Pause(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot pause a stopped container") return errors.Errorf("cannot pause a stopped container")
} }
func (s *stoppedState) Resume(ctx context.Context) error { func (s *stoppedState) Resume(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot resume a stopped container") return errors.Errorf("cannot resume a stopped container")
} }
func (s *stoppedState) Update(context context.Context, r *google_protobuf.Any) error { func (s *stoppedState) Update(ctx context.Context, r *google_protobuf.Any) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot update a stopped container") return errors.Errorf("cannot update a stopped container")
} }
func (s *stoppedState) Checkpoint(ctx context.Context, r *CheckpointConfig) error { func (s *stoppedState) Checkpoint(ctx context.Context, r *CheckpointConfig) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot checkpoint a stopped container") return errors.Errorf("cannot checkpoint a stopped container")
} }
func (s *stoppedState) Resize(ws console.WinSize) error { func (s *stoppedState) Resize(ws console.WinSize) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot resize a stopped container") return errors.Errorf("cannot resize a stopped container")
} }
func (s *stoppedState) Start(ctx context.Context) error { func (s *stoppedState) Start(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return errors.Errorf("cannot start a stopped process") return errors.Errorf("cannot start a stopped process")
} }
func (s *stoppedState) Delete(ctx context.Context) error { func (s *stoppedState) Delete(ctx context.Context) error {
s.p.mu.Lock()
defer s.p.mu.Unlock()
if err := s.p.delete(ctx); err != nil { if err := s.p.delete(ctx); err != nil {
return err return err
} }
@ -538,8 +415,5 @@ func (s *stoppedState) SetExited(status int) {
} }
func (s *stoppedState) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) { func (s *stoppedState) Exec(ctx context.Context, path string, r *ExecConfig) (proc.Process, error) {
s.p.mu.Lock()
defer s.p.mu.Unlock()
return nil, errors.Errorf("cannot exec in a stopped state") return nil, errors.Errorf("cannot exec in a stopped state")
} }