Merge pull request #3365 from crosbymichael/exec-lk

estesp · web-flow · commit 287582585fb4 · 2019-06-25T08:59:41.000+08:00
Reserve exec id to prevent race
diff --git a/runtime/v2/runc/container.go b/runtime/v2/runc/container.go
@@ -125,10 +125,11 @@ func NewContainer(ctx context.Context, platform rproc.Platform, r *task.CreateTa
 		return nil, errdefs.ToGRPC(err)
 	}
 	container := &Container{
-		ID:        r.ID,
-		Bundle:    r.Bundle,
-		process:   process,
-		processes: make(map[string]rproc.Process),
+		ID:              r.ID,
+		Bundle:          r.Bundle,
+		process:         process,
+		processes:       make(map[string]rproc.Process),
+		reservedProcess: make(map[string]struct{}),
 	}
 	pid := process.Pid()
 	if pid > 0 {
@@ -189,9 +190,10 @@ type Container struct {
 	// Bundle path
 	Bundle string
 
-	cgroup    cgroups.Cgroup
-	process   rproc.Process
-	processes map[string]rproc.Process
+	cgroup          cgroups.Cgroup
+	process         rproc.Process
+	processes       map[string]rproc.Process
+	reservedProcess map[string]struct{}
 }
 
 // All processes in the container
@@ -256,18 +258,35 @@ func (c *Container) Process(id string) (rproc.Process, error) {
 	return p, nil
 }
 
-// ProcessExists returns true if the process by id exists
-func (c *Container) ProcessExists(id string) bool {
+// ReserveProcess checks for the existence of an id and atomically
+// reserves the process id if it does not already exist
+//
+// Returns true if the process id was sucessfully reserved and a
+// cancel func to release the reservation
+func (c *Container) ReserveProcess(id string) (bool, func()) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	_, ok := c.processes[id]
-	return ok
+
+	if _, ok := c.processes[id]; ok {
+		return false, nil
+	}
+	if _, ok := c.reservedProcess[id]; ok {
+		return false, nil
+	}
+	c.reservedProcess[id] = struct{}{}
+	return true, func() {
+		c.mu.Lock()
+		defer c.mu.Unlock()
+		delete(c.reservedProcess, id)
+	}
 }
 
 // ProcessAdd adds a new process to the container
 func (c *Container) ProcessAdd(process rproc.Process) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+
+	delete(c.reservedProcess, process.ID())
 	c.processes[process.ID()] = process
 }
 
diff --git a/runtime/v2/runc/v1/service.go b/runtime/v2/runc/v1/service.go
@@ -326,11 +326,13 @@ func (s *service) Exec(ctx context.Context, r *taskAPI.ExecProcessRequest) (*pty
 	if err != nil {
 		return nil, err
 	}
-	if container.ProcessExists(r.ExecID) {
+	ok, cancel := container.ReserveProcess(r.ExecID)
+	if !ok {
 		return nil, errdefs.ToGRPCf(errdefs.ErrAlreadyExists, "id %s", r.ExecID)
 	}
 	process, err := container.Exec(ctx, r)
 	if err != nil {
+		cancel()
 		return nil, errdefs.ToGRPC(err)
 	}
 
diff --git a/runtime/v2/runc/v2/service.go b/runtime/v2/runc/v2/service.go
@@ -375,11 +375,13 @@ func (s *service) Exec(ctx context.Context, r *taskAPI.ExecProcessRequest) (*pty
 	if err != nil {
 		return nil, err
 	}
-	if container.ProcessExists(r.ExecID) {
+	ok, cancel := container.ReserveProcess(r.ExecID)
+	if !ok {
 		return nil, errdefs.ToGRPCf(errdefs.ErrAlreadyExists, "id %s", r.ExecID)
 	}
 	process, err := container.Exec(ctx, r)
 	if err != nil {
+		cancel()
 		return nil, errdefs.ToGRPC(err)
 	}
 

Original file line number	Diff line number	Diff line change
`@@ -326,11 +326,13 @@ func (s service) Exec(ctx context.Context, r taskAPI.ExecProcessRequest) (*pty`
`326`	`326`	`if err != nil {`
`327`	`327`	`return nil, err`
`328`	`328`	`}`
`329`		`- if container.ProcessExists(r.ExecID) {`
	`329`	`+ ok, cancel := container.ReserveProcess(r.ExecID)`
	`330`	`+ if !ok {`
`330`	`331`	`return nil, errdefs.ToGRPCf(errdefs.ErrAlreadyExists, "id %s", r.ExecID)`
`331`	`332`	`}`
`332`	`333`	`process, err := container.Exec(ctx, r)`
`333`	`334`	`if err != nil {`
	`335`	`+ cancel()`
`334`	`336`	`return nil, errdefs.ToGRPC(err)`
`335`	`337`	`}`
`336`	`338`
Original file line number	Diff line number	Diff line change
`@@ -375,11 +375,13 @@ func (s service) Exec(ctx context.Context, r taskAPI.ExecProcessRequest) (*pty`
`375`	`375`	`if err != nil {`
`376`	`376`	`return nil, err`
`377`	`377`	`}`
`378`		`- if container.ProcessExists(r.ExecID) {`
	`378`	`+ ok, cancel := container.ReserveProcess(r.ExecID)`
	`379`	`+ if !ok {`
`379`	`380`	`return nil, errdefs.ToGRPCf(errdefs.ErrAlreadyExists, "id %s", r.ExecID)`
`380`	`381`	`}`
`381`	`382`	`process, err := container.Exec(ctx, r)`
`382`	`383`	`if err != nil {`
	`384`	`+ cancel()`
`383`	`385`	`return nil, errdefs.ToGRPC(err)`
`384`	`386`	`}`
`385`	`387`