supervisor: cancel SendTask when channel is full

This patch handles case c) in moby/moby#31487

(c) A request cannot be added to 's.tasks' because the queue is full.

case b) isn't fixable at this point case there's no way to cancel stuff
from a channel since channels aren't context aware (yet?).
Parallelizing tasks handling isn't easily doable either.

Signed-off-by: Antonio Murdaca <runcom@redhat.com>

Author: runcom

api/grpc/server/server.go

@@ -46,6 +46,7 @@ func (s *apiServer) CreateContainer(ctx context.Context, c *types.CreateContaine
 		return nil, errors.New("empty bundle path")
 	}
 	e := &supervisor.StartTask{}
+	e.WithContext(ctx)
 	e.ID = c.Id
 	e.BundlePath = c.BundlePath
 	e.Stdin = c.Stdin
@@ -56,7 +57,6 @@ func (s *apiServer) CreateContainer(ctx context.Context, c *types.CreateContaine
 	e.Runtime = c.Runtime
 	e.RuntimeArgs = c.RuntimeArgs
 	e.StartResponse = make(chan supervisor.StartResponse, 1)
-	e.Ctx = ctx
 	if c.Checkpoint != "" {
 		e.CheckpointDir = c.CheckpointDir
 		e.Checkpoint = &runtime.Checkpoint{
@@ -79,6 +79,7 @@ func (s *apiServer) CreateContainer(ctx context.Context, c *types.CreateContaine
 
 func (s *apiServer) CreateCheckpoint(ctx context.Context, r *types.CreateCheckpointRequest) (*types.CreateCheckpointResponse, error) {
 	e := &supervisor.CreateCheckpointTask{}
+	e.WithContext(ctx)
 	e.ID = r.Id
 	e.CheckpointDir = r.CheckpointDir
 	e.Checkpoint = &runtime.Checkpoint{
@@ -102,6 +103,7 @@ func (s *apiServer) DeleteCheckpoint(ctx context.Context, r *types.DeleteCheckpo
 		return nil, errors.New("checkpoint name cannot be empty")
 	}
 	e := &supervisor.DeleteCheckpointTask{}
+	e.WithContext(ctx)
 	e.ID = r.Id
 	e.CheckpointDir = r.CheckpointDir
 	e.Checkpoint = &runtime.Checkpoint{
@@ -116,6 +118,7 @@ func (s *apiServer) DeleteCheckpoint(ctx context.Context, r *types.DeleteCheckpo
 
 func (s *apiServer) ListCheckpoint(ctx context.Context, r *types.ListCheckpointRequest) (*types.ListCheckpointResponse, error) {
 	e := &supervisor.GetContainersTask{}
+	e.WithContext(ctx)
 	s.sv.SendTask(e)
 	if err := <-e.ErrorCh(); err != nil {
 		return nil, err
@@ -150,6 +153,7 @@ func (s *apiServer) ListCheckpoint(ctx context.Context, r *types.ListCheckpointR
 
 func (s *apiServer) Signal(ctx context.Context, r *types.SignalRequest) (*types.SignalResponse, error) {
 	e := &supervisor.SignalTask{}
+	e.WithContext(ctx)
 	e.ID = r.Id
 	e.PID = r.Pid
 	e.Signal = syscall.Signal(int(r.Signal))
@@ -167,6 +171,7 @@ func (s *apiServer) State(ctx context.Context, r *types.StateRequest) (*types.St
 	}
 
 	e := &supervisor.GetContainersTask{}
+	e.WithContext(ctx)
 	e.ID = r.Id
 	e.GetState = getState
 	s.sv.SendTask(e)
@@ -253,6 +258,7 @@ func toUint32(its []int) []uint32 {
 
 func (s *apiServer) UpdateContainer(ctx context.Context, r *types.UpdateContainerRequest) (*types.UpdateContainerResponse, error) {
 	e := &supervisor.UpdateTask{}
+	e.WithContext(ctx)
 	e.ID = r.Id
 	e.State = runtime.State(r.Status)
 	if r.Resources != nil {
@@ -304,6 +310,7 @@ func (s *apiServer) UpdateContainer(ctx context.Context, r *types.UpdateContaine
 
 func (s *apiServer) UpdateProcess(ctx context.Context, r *types.UpdateProcessRequest) (*types.UpdateProcessResponse, error) {
 	e := &supervisor.UpdateProcessTask{}
+	e.WithContext(ctx)
 	e.ID = r.Id
 	e.PID = r.Pid
 	e.Height = int(r.Height)
@@ -482,6 +489,7 @@ func getSystemCPUUsage() (uint64, error) {
 
 func (s *apiServer) Stats(ctx context.Context, r *types.StatsRequest) (*types.StatsResponse, error) {
 	e := &supervisor.StatsTask{}
+	e.WithContext(ctx)
 	e.ID = r.Id
 	e.Stat = make(chan *runtime.Stat, 1)
 	s.sv.SendTask(e)

api/grpc/server/server_linux.go

@@ -49,14 +49,14 @@ func (s *apiServer) AddProcess(ctx context.Context, r *types.AddProcessRequest)
 		return nil, fmt.Errorf("process id cannot be empty")
 	}
 	e := &supervisor.AddProcessTask{}
+	e.WithContext(ctx)
 	e.ID = r.Id
 	e.PID = r.Pid
 	e.ProcessSpec = process
 	e.Stdin = r.Stdin
 	e.Stdout = r.Stdout
 	e.Stderr = r.Stderr
 	e.StartResponse = make(chan supervisor.StartResponse, 1)
-	e.Ctx = ctx
 	s.sv.SendTask(e)
 	if err := <-e.ErrorCh(); err != nil {
 		return nil, err

api/grpc/server/server_solaris.go

@@ -25,6 +25,7 @@ func (s *apiServer) AddProcess(ctx context.Context, r *types.AddProcessRequest)
 		return nil, fmt.Errorf("process id cannot be empty")
 	}
 	e := &supervisor.AddProcessTask{}
+	e.WithContext(ctx)
 	e.ID = r.Id
 	e.PID = r.Pid
 	e.ProcessSpec = process

supervisor/add_process.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/containerd/containerd/runtime"
 	"github.com/containerd/containerd/specs"
-	"golang.org/x/net/context"
 )
 
 // AddProcessTask holds everything necessary to add a process to a
@@ -20,7 +19,6 @@ type AddProcessTask struct {
 	Stdin         string
 	ProcessSpec   *specs.ProcessSpec
 	StartResponse chan StartResponse
-	Ctx           context.Context
 }
 
 func (s *Supervisor) addProcess(t *AddProcessTask) error {
@@ -29,7 +27,7 @@ func (s *Supervisor) addProcess(t *AddProcessTask) error {
 	if !ok {
 		return ErrContainerNotFound
 	}
-	process, err := ci.container.Exec(t.Ctx, t.PID, *t.ProcessSpec, runtime.NewStdio(t.Stdin, t.Stdout, t.Stderr))
+	process, err := ci.container.Exec(t.Ctx(), t.PID, *t.ProcessSpec, runtime.NewStdio(t.Stdin, t.Stdout, t.Stderr))
 	if err != nil {
 		return err
 	}

supervisor/create.go

@@ -5,7 +5,6 @@ import (
 	"time"
 
 	"github.com/containerd/containerd/runtime"
-	"golang.org/x/net/context"
 )
 
 // StartTask holds needed parameters to create a new container
@@ -23,7 +22,6 @@ type StartTask struct {
 	CheckpointDir string
 	Runtime       string
 	RuntimeArgs   []string
-	Ctx           context.Context
 }
 
 func (s *Supervisor) start(t *StartTask) error {
@@ -59,7 +57,7 @@ func (s *Supervisor) start(t *StartTask) error {
 		Stdin:         t.Stdin,
 		Stdout:        t.Stdout,
 		Stderr:        t.Stderr,
-		Ctx:           t.Ctx,
+		Ctx:           t.Ctx(),
 	}
 	if t.Checkpoint != nil {
 		task.CheckpointPath = filepath.Join(t.CheckpointDir, t.Checkpoint.Name)

supervisor/exit.go

@@ -41,6 +41,7 @@ func (s *Supervisor) exit(t *ExitTask) error {
 			Status:  status,
 			Process: proc,
 		}
+		ne.WithContext(t.Ctx())
 		s.execExit(ne)
 		return nil
 	}
@@ -51,6 +52,7 @@ func (s *Supervisor) exit(t *ExitTask) error {
 		PID:     proc.ID(),
 		Process: proc,
 	}
+	ne.WithContext(t.Ctx())
 	s.delete(ne)
 
 	ExitProcessTimer.UpdateSince(start)

supervisor/supervisor.go

@@ -1,6 +1,7 @@
 package supervisor
 
 import (
+	"context"
 	"encoding/json"
 	"io"
 	"io/ioutil"
@@ -290,15 +291,21 @@ func (s *Supervisor) Machine() Machine {
 
 // SendTask sends the provided event the the supervisors main event loop
 func (s *Supervisor) SendTask(evt Task) {
-	TasksCounter.Inc(1)
-	s.tasks <- evt
+	select {
+	case <-evt.Ctx().Done():
+		evt.ErrorCh() <- evt.Ctx().Err()
+		close(evt.ErrorCh())
+	case s.tasks <- evt:
+		TasksCounter.Inc(1)
+	}
 }
 
 func (s *Supervisor) exitHandler() {
 	for p := range s.monitor.Exits() {
 		e := &ExitTask{
 			Process: p,
 		}
+		e.WithContext(context.Background())
 		s.SendTask(e)
 	}
 }
@@ -308,6 +315,7 @@ func (s *Supervisor) oomHandler() {
 		e := &OOMTask{
 			ID: id,
 		}
+		e.WithContext(context.Background())
 		s.SendTask(e)
 	}
 }
@@ -371,6 +379,7 @@ func (s *Supervisor) restore() error {
 				e := &ExitTask{
 					Process: p,
 				}
+				e.WithContext(context.Background())
 				s.SendTask(e)
 			}
 		}

supervisor/task.go

@@ -1,6 +1,7 @@
 package supervisor
 
 import (
+	"context"
 	"sync"
 
 	"github.com/containerd/containerd/runtime"
@@ -17,13 +18,28 @@ type StartResponse struct {
 type Task interface {
 	// ErrorCh returns a channel used to report and error from an async task
 	ErrorCh() chan error
+	// Ctx carries the context of a task
+	Ctx() context.Context
 }
 
 type baseTask struct {
 	errCh chan error
+	ctx   context.Context
 	mu    sync.Mutex
 }
 
+func (t *baseTask) WithContext(ctx context.Context) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.ctx = ctx
+}
+
+func (t *baseTask) Ctx() context.Context {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.ctx
+}
+
 func (t *baseTask) ErrorCh() chan error {
 	t.mu.Lock()
 	defer t.mu.Unlock()

supervisor/worker.go

@@ -55,6 +55,7 @@ func (w *worker) Start() {
 				NoEvent: true,
 				Process: process,
 			}
+			evt.WithContext(t.Ctx)
 			w.s.SendTask(evt)
 			continue
 		}
@@ -71,6 +72,7 @@ func (w *worker) Start() {
 				NoEvent: true,
 				Process: process,
 			}
+			evt.WithContext(t.Ctx)
 			w.s.SendTask(evt)
 			continue
 		}
@@ -85,6 +87,7 @@ func (w *worker) Start() {
 					NoEvent: true,
 					Process: process,
 				}
+				evt.WithContext(t.Ctx)
 				w.s.SendTask(evt)
 				continue
 			}