Refactor libcontainerd to minimize containerd RPCs
<!--
Please make sure you've read and understood our contributing guidelines;
https://github.com/moby/moby/blob/master/CONTRIBUTING.md

** Make sure all your commits include a signature generated with `git commit -s` **

For additional information on our contributing process, read our contributing
guide https://docs.docker.com/opensource/code/

If this is a bug fix, make sure your description includes "fixes #xxxx", or
"closes #xxxx"

Please provide the following information:
-->

- Relates to #33933

**- What I did**
While investigating a report of slow `docker exec`s and failing health checks on heavily-loaded systems, I instrumented `dockerd` with OpenTelemetry tracing (https://github.com/corhere/moby/tree/otel-trace) and captured traces of various container operations. I was surprised to see RPCs such as `containerd.services.containers.v1.Container/Get` being called multiple times in a row for nearly every operation. In the common case of a local containerd process each RPC costs at least two context switches to complete, so we want to keep the number of RPCs per operation to a minimum to reduce the overhead of dockerd and containerd on the system. I overhauled libcontainerd and refactored how the daemon uses libcontainerd to do just that.

I modified the behavior of the `Container.State.OOMKilled` flag to update immediately (rather than on container exit) to simplify the code and reduce the amount of state libcontainerd needs to track, with the side benefit of the new behavior being more intuitive to users.

I benchmarked these changes against the base commit (bb88ff4ab44c0e030bcd02ee364c66cdb5cef5a4) and measured a reduction in wallclock time for `docker exec` of 6.8% in my test environment.

<details>
<summary>Benchmarking details</summary>

```console
$ id=$(docker run -d alpine sleep infinity)
$ for x in {1..5}; do time for i in {1..500}; do docker exec $id true; done; done
```

| | master | libcontainerd-overhaul |
| --- | --- | --- |
| Run 1 | 45.768s | 43.029s |
| Run 2 | 47.123s | 42.953s |
| Run 3 | 46.916s | 43.299s |
| Run 4 | 47.735s | 43.245s |
| Run 5 | 43.926s | 43.223s |
| **Average** | **46.294s** | **43.150s** |

```
(43.150 - 46.294) / 46.294 = -6.79%
```

</details>

**- How I did it**
The containerd client library is very chatty, generally [erring on the side of refreshing metadata](https://github.com/containerd/containerd/blob/6067aeb6fde8afb9357652c09330c7fec6515eab/container.go#L105-L137) over referencing a locally-cached copy. (The containerd gRPC API has no concurrency control so multiple clients could race to mutate the same resources. Aggressively refreshing metadata only shortens the race window; it does not eliminate it. Perhaps a future version of the containerd API could improve upon the situation so it can be less aggressive about refreshing. But I digress...) And further exacerbating the situation, the libcontainerd implementation is effectively forced to reload the containerd resources at the start of every operation because containers and processes are referenced by ID strings. To illustrate, consider the libcontainerd `Start()` method. Its implementation is, in broad strokes:
```go
func (c *client) Start(ctx context.Context, id string, /* ... */) {
	ctr, _ := c.client.LoadContainer(ctx, id) // v1.Container/Get
	spec, _ := ctr.Spec(ctx) // v1.Container/Get
	labels, _ := ctr.Labels(ctx) // v1.Container/Get
	t, _ := ctr.NewTask(ctx, /* options derived from spec, labels */) // v1.Container/Get, v1.Task/Create
	_ = t.Start(ctx) // v1.Task/Start
}
```
That's four calls to `containerd.services.containers.v1.Container/Get` in a row, all to get the exact same metadata! Two of those redundant calls can be easily eliminated by reusing the container metadata initially loaded by `c.client.LoadContainer`. And the `LoadContainer` call could also be elided if the container object returned by `c.client.NewContainer` from when the container was created was retained. The number of RPCs to start a container can be _cut in half_ without any changes to the containerd client! The containerd client resource objects need to be retained and reused to pull this off, which means more state for the Docker daemon to manage.

The Docker daemon already keeps track of state for every container and exec, with mappings from ID strings to the respective state objects. Holding persistent state in the libcontainerd layer with mappings from ID strings to state objects would be redundant, so I overhauled the libcontainerd interfaces to allow for implementations which hold no shared mutable state. Not coincidentally, this shape mirrors the containerd client's interfaces. I delegated retaining references to the libcontainerd objects to the daemon's container state and exec config structs.

The local Windows (HCSv1) libcontainerd implementation was a bit of a challenge to refactor to fit the new shape of the libcontainerd interface, but it also benefited from eliminating shared mutable state.

**- How to verify it**
CI ✅ 

**- Description for the changelog**
- Reduced the number of RPCs required to perform operations with containerd-based runtimes
- A container's `State.OOMKilled` flag is now set to true immediately upon any container process getting OOM-killed by the kernel, and cleared to false when the container is restarted

**- A picture of a cute animal (not mandatory but encouraged)**
<a href="https://xkcd.com/292"><img src="https://imgs.xkcd.com/comics/goto.png"><a>

daemon/exec.go
package daemon // import "github.com/docker/docker/daemon"

import (
	"context"
	"fmt"
	"io"
	"runtime"
	"strings"
	"time"

	"github.com/docker/docker/api/types"
	containertypes "github.com/docker/docker/api/types/container"
	"github.com/docker/docker/api/types/strslice"
	"github.com/docker/docker/container"
	"github.com/docker/docker/container/stream"
	"github.com/docker/docker/daemon/exec"
	"github.com/docker/docker/errdefs"
	"github.com/docker/docker/pkg/pools"
	"github.com/moby/sys/signal"
	"github.com/moby/term"
	specs "github.com/opencontainers/runtime-spec/specs-go"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
)

func (daemon *Daemon) registerExecCommand(container *container.Container, config *exec.Config) {
	// Storing execs in container in order to kill them gracefully whenever the container is stopped or removed.
	container.ExecCommands.Add(config.ID, config)
	// Storing execs in daemon for easy access via Engine API.
	daemon.execCommands.Add(config.ID, config)
}

// ExecExists looks up the exec instance and returns a bool if it exists or not.
// It will also return the error produced by `getConfig`
func (daemon *Daemon) ExecExists(name string) (bool, error) {
	if _, err := daemon.getExecConfig(name); err != nil {
		return false, err
	}
	return true, nil
}

// getExecConfig looks up the exec instance by name. If the container associated
// with the exec instance is stopped or paused, it will return an error.
func (daemon *Daemon) getExecConfig(name string) (*exec.Config, error) {
	ec := daemon.execCommands.Get(name)
	if ec == nil {
		return nil, errExecNotFound(name)
	}

	// If the exec is found but its container is not in the daemon's list of
	// containers then it must have been deleted, in which case instead of
	// saying the container isn't running, we should return a 404 so that
	// the user sees the same error now that they will after the
	// 5 minute clean-up loop is run which erases old/dead execs.
	ctr := daemon.containers.Get(ec.ContainerID)
	if ctr == nil {
		return nil, containerNotFound(name)
	}
	if !ctr.IsRunning() {
		return nil, errNotRunning(ctr.ID)
	}
	if ctr.IsPaused() {
		return nil, errExecPaused(ctr.ID)
	}
	if ctr.IsRestarting() {
		return nil, errContainerIsRestarting(ctr.ID)
	}
	return ec, nil
}

func (daemon *Daemon) unregisterExecCommand(container *container.Container, execConfig *exec.Config) {
	container.ExecCommands.Delete(execConfig.ID, execConfig.Pid)
	daemon.execCommands.Delete(execConfig.ID, execConfig.Pid)
}

func (daemon *Daemon) getActiveContainer(name string) (*container.Container, error) {
	ctr, err := daemon.GetContainer(name)
	if err != nil {
		return nil, err
	}

	if !ctr.IsRunning() {
		return nil, errNotRunning(ctr.ID)
	}
	if ctr.IsPaused() {
		return nil, errExecPaused(name)
	}
	if ctr.IsRestarting() {
		return nil, errContainerIsRestarting(ctr.ID)
	}
	return ctr, nil
}

// ContainerExecCreate sets up an exec in a running container.
func (daemon *Daemon) ContainerExecCreate(name string, config *types.ExecConfig) (string, error) {
	cntr, err := daemon.getActiveContainer(name)
	if err != nil {
		return "", err
	}

	cmd := strslice.StrSlice(config.Cmd)
	entrypoint, args := daemon.getEntrypointAndArgs(strslice.StrSlice{}, cmd)

	keys := []byte{}
	if config.DetachKeys != "" {
		keys, err = term.ToBytes(config.DetachKeys)
		if err != nil {
			err = fmt.Errorf("Invalid escape keys (%s) provided", config.DetachKeys)
			return "", err
		}
	}

	execConfig := exec.NewConfig()
	execConfig.OpenStdin = config.AttachStdin
	execConfig.OpenStdout = config.AttachStdout
	execConfig.OpenStderr = config.AttachStderr
	execConfig.ContainerID = cntr.ID
	execConfig.DetachKeys = keys
	execConfig.Entrypoint = entrypoint
	execConfig.Args = args
	execConfig.Tty = config.Tty
	execConfig.ConsoleSize = config.ConsoleSize
	execConfig.Privileged = config.Privileged
	execConfig.User = config.User
	execConfig.WorkingDir = config.WorkingDir

	linkedEnv, err := daemon.setupLinkedContainers(cntr)
	if err != nil {
		return "", err
	}
	execConfig.Env = container.ReplaceOrAppendEnvValues(cntr.CreateDaemonEnvironment(config.Tty, linkedEnv), config.Env)
	if len(execConfig.User) == 0 {
		execConfig.User = cntr.Config.User
	}
	if len(execConfig.WorkingDir) == 0 {
		execConfig.WorkingDir = cntr.Config.WorkingDir
	}

	daemon.registerExecCommand(cntr, execConfig)

	attributes := map[string]string{
		"execID": execConfig.ID,
	}
	daemon.LogContainerEventWithAttributes(cntr, "exec_create: "+execConfig.Entrypoint+" "+strings.Join(execConfig.Args, " "), attributes)

	return execConfig.ID, nil
}

// ContainerExecStart starts a previously set up exec instance. The
// std streams are set up.
// If ctx is cancelled, the process is terminated.
func (daemon *Daemon) ContainerExecStart(ctx context.Context, name string, options containertypes.ExecStartOptions) (err error) {
	var (
		cStdin           io.ReadCloser
		cStdout, cStderr io.Writer
	)

	ec, err := daemon.getExecConfig(name)
	if err != nil {
		return err
	}

	ec.Lock()
	if ec.ExitCode != nil {
		ec.Unlock()
		err := fmt.Errorf("Error: Exec command %s has already run", ec.ID)
		return errdefs.Conflict(err)
	}

	if ec.Running {
		ec.Unlock()
		return errdefs.Conflict(fmt.Errorf("Error: Exec command %s is already running", ec.ID))
	}
	ec.Running = true
	ec.Unlock()

	c := daemon.containers.Get(ec.ContainerID)
	if c == nil {
		return containerNotFound(ec.ContainerID)
	}
	logrus.Debugf("starting exec command %s in container %s", ec.ID, c.ID)
	attributes := map[string]string{
		"execID": ec.ID,
	}
	daemon.LogContainerEventWithAttributes(c, "exec_start: "+ec.Entrypoint+" "+strings.Join(ec.Args, " "), attributes)

	defer func() {
		if err != nil {
			ec.Lock()
			ec.Running = false
			exitCode := 126
			ec.ExitCode = &exitCode
			if err := ec.CloseStreams(); err != nil {
				logrus.Errorf("failed to cleanup exec %s streams: %s", c.ID, err)
			}
			ec.Unlock()
			c.ExecCommands.Delete(ec.ID, ec.Pid)
		}
	}()

	if ec.OpenStdin && options.Stdin != nil {
		r, w := io.Pipe()
		go func() {
			defer w.Close()
			defer logrus.Debug("Closing buffered stdin pipe")
			pools.Copy(w, options.Stdin)
		}()
		cStdin = r
	}
	if ec.OpenStdout {
		cStdout = options.Stdout
	}
	if ec.OpenStderr {
		cStderr = options.Stderr
	}

	if ec.OpenStdin {
		ec.StreamConfig.NewInputPipes()
	} else {
		ec.StreamConfig.NewNopInputPipe()
	}

	p := &specs.Process{}
	if runtime.GOOS != "windows" {
		ctr, err := daemon.containerdCli.LoadContainer(ctx, ec.ContainerID)
		if err != nil {
			return err
		}
		spec, err := ctr.Spec(ctx)
		if err != nil {
			return err
		}
		p = spec.Process
	}
	p.Args = append([]string{ec.Entrypoint}, ec.Args...)
	p.Env = ec.Env
	p.Cwd = ec.WorkingDir
	p.Terminal = ec.Tty

	consoleSize := options.ConsoleSize
	// If size isn't specified for start, use the one provided for create
	if consoleSize == nil {
		consoleSize = ec.ConsoleSize
	}
	if p.Terminal && consoleSize != nil {
		p.ConsoleSize = &specs.Box{
			Height: consoleSize[0],
			Width:  consoleSize[1],
		}
	}

	if p.Cwd == "" {
		p.Cwd = "/"
	}

	if err := daemon.execSetPlatformOpt(c, ec, p); err != nil {
		return err
	}

	attachConfig := stream.AttachConfig{
		TTY:        ec.Tty,
		UseStdin:   cStdin != nil,
		UseStdout:  cStdout != nil,
		UseStderr:  cStderr != nil,
		Stdin:      cStdin,
		Stdout:     cStdout,
		Stderr:     cStderr,
		DetachKeys: ec.DetachKeys,
		CloseStdin: true,
	}
	ec.StreamConfig.AttachStreams(&attachConfig)
	// using context.Background() so that attachErr does not race ctx.Done().
	copyCtx, cancel := context.WithCancel(context.Background())
	defer cancel()
	attachErr := ec.StreamConfig.CopyStreams(copyCtx, &attachConfig)

	// Synchronize with libcontainerd event loop
	ec.Lock()
	c.ExecCommands.Lock()
	systemPid, err := daemon.containerd.Exec(ctx, c.ID, ec.ID, p, cStdin != nil, ec.InitializeStdio)
	// the exec context should be ready, or error happened.
	// close the chan to notify readiness
	close(ec.Started)
	if err != nil {
		c.ExecCommands.Unlock()
		ec.Unlock()
		return translateContainerdStartErr(ec.Entrypoint, ec.SetExitCode, err)
	}
	ec.Pid = systemPid
	c.ExecCommands.Unlock()
	ec.Unlock()

	select {
	case <-ctx.Done():
		log := logrus.
			WithField("container", c.ID).
			WithField("exec", name)
		log.Debug("Sending KILL signal to container process")
		sigCtx, cancelFunc := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancelFunc()
		err := daemon.containerd.SignalProcess(sigCtx, c.ID, name, signal.SignalMap["KILL"])
		if err != nil {
			log.WithError(err).Error("Could not send KILL signal to container process")
		}
		return ctx.Err()
	case err := <-attachErr:
		if err != nil {
			if _, ok := err.(term.EscapeError); !ok {
				return errdefs.System(errors.Wrap(err, "exec attach failed"))
			}
			attributes := map[string]string{
				"execID": ec.ID,
			}
			daemon.LogContainerEventWithAttributes(c, "exec_detach", attributes)
		}
	}
	return nil
}

// execCommandGC runs a ticker to clean up the daemon references
// of exec configs that are no longer part of the container.
func (daemon *Daemon) execCommandGC() {
	for range time.Tick(5 * time.Minute) {
		var (
			cleaned          int
			liveExecCommands = daemon.containerExecIds()
		)
		for id, config := range daemon.execCommands.Commands() {
			if config.CanRemove {
				cleaned++
				daemon.execCommands.Delete(id, config.Pid)
			} else {
				if _, exists := liveExecCommands[id]; !exists {
					config.CanRemove = true
				}
			}
		}
		if cleaned > 0 {
			logrus.Debugf("clean %d unused exec commands", cleaned)
		}
	}
}

// containerExecIds returns a list of all the current exec ids that are in use
// and running inside a container.
func (daemon *Daemon) containerExecIds() map[string]struct{} {
	ids := map[string]struct{}{}
	for _, c := range daemon.containers.List() {
		for _, id := range c.ExecCommands.List() {
			ids[id] = struct{}{}
		}
	}
	return ids
}

package daemon // import "github.com/docker/docker/daemon"

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"runtime"
	"strings"
	"time"

	"github.com/containerd/containerd"
	"github.com/docker/docker/api/types"
	containertypes "github.com/docker/docker/api/types/container"
	"github.com/docker/docker/api/types/strslice"
	"github.com/docker/docker/container"
	"github.com/docker/docker/container/stream"
	"github.com/docker/docker/errdefs"
	"github.com/docker/docker/pkg/pools"
	"github.com/moby/sys/signal"
	"github.com/moby/term"
	specs "github.com/opencontainers/runtime-spec/specs-go"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
)

func (daemon *Daemon) registerExecCommand(container *container.Container, config *container.ExecConfig) {
	// Storing execs in container in order to kill them gracefully whenever the container is stopped or removed.
	container.ExecCommands.Add(config.ID, config)
	// Storing execs in daemon for easy access via Engine API.
	daemon.execCommands.Add(config.ID, config)
}

// ExecExists looks up the exec instance and returns a bool if it exists or not.
// It will also return the error produced by `getConfig`
func (daemon *Daemon) ExecExists(name string) (bool, error) {
	if _, err := daemon.getExecConfig(name); err != nil {
		return false, err
	}
	return true, nil
}

// getExecConfig looks up the exec instance by name. If the container associated
// with the exec instance is stopped or paused, it will return an error.
func (daemon *Daemon) getExecConfig(name string) (*container.ExecConfig, error) {
	ec := daemon.execCommands.Get(name)
	if ec == nil {
		return nil, errExecNotFound(name)
	}

	// If the exec is found but its container is not in the daemon's list of
	// containers then it must have been deleted, in which case instead of
	// saying the container isn't running, we should return a 404 so that
	// the user sees the same error now that they will after the
	// 5 minute clean-up loop is run which erases old/dead execs.
	ctr := daemon.containers.Get(ec.Container.ID)
	if ctr == nil {
		return nil, containerNotFound(name)
	}
	if !ctr.IsRunning() {
		return nil, errNotRunning(ctr.ID)
	}
	if ctr.IsPaused() {
		return nil, errExecPaused(ctr.ID)
	}
	if ctr.IsRestarting() {
		return nil, errContainerIsRestarting(ctr.ID)
	}
	return ec, nil
}

func (daemon *Daemon) unregisterExecCommand(container *container.Container, execConfig *container.ExecConfig) {
	container.ExecCommands.Delete(execConfig.ID)
	daemon.execCommands.Delete(execConfig.ID)
}

func (daemon *Daemon) getActiveContainer(name string) (*container.Container, error) {
	ctr, err := daemon.GetContainer(name)
	if err != nil {
		return nil, err
	}

	if !ctr.IsRunning() {
		return nil, errNotRunning(ctr.ID)
	}
	if ctr.IsPaused() {
		return nil, errExecPaused(name)
	}
	if ctr.IsRestarting() {
		return nil, errContainerIsRestarting(ctr.ID)
	}
	return ctr, nil
}

// ContainerExecCreate sets up an exec in a running container.
func (daemon *Daemon) ContainerExecCreate(name string, config *types.ExecConfig) (string, error) {
	cntr, err := daemon.getActiveContainer(name)
	if err != nil {
		return "", err
	}

	cmd := strslice.StrSlice(config.Cmd)
	entrypoint, args := daemon.getEntrypointAndArgs(strslice.StrSlice{}, cmd)

	keys := []byte{}
	if config.DetachKeys != "" {
		keys, err = term.ToBytes(config.DetachKeys)
		if err != nil {
			err = fmt.Errorf("Invalid escape keys (%s) provided", config.DetachKeys)
			return "", err
		}
	}

	execConfig := container.NewExecConfig(cntr)
	execConfig.OpenStdin = config.AttachStdin
	execConfig.OpenStdout = config.AttachStdout
	execConfig.OpenStderr = config.AttachStderr
	execConfig.DetachKeys = keys
	execConfig.Entrypoint = entrypoint
	execConfig.Args = args
	execConfig.Tty = config.Tty
	execConfig.ConsoleSize = config.ConsoleSize
	execConfig.Privileged = config.Privileged
	execConfig.User = config.User
	execConfig.WorkingDir = config.WorkingDir

	linkedEnv, err := daemon.setupLinkedContainers(cntr)
	if err != nil {
		return "", err
	}
	execConfig.Env = container.ReplaceOrAppendEnvValues(cntr.CreateDaemonEnvironment(config.Tty, linkedEnv), config.Env)
	if len(execConfig.User) == 0 {
		execConfig.User = cntr.Config.User
	}
	if len(execConfig.WorkingDir) == 0 {
		execConfig.WorkingDir = cntr.Config.WorkingDir
	}

	daemon.registerExecCommand(cntr, execConfig)

	attributes := map[string]string{
		"execID": execConfig.ID,
	}
	daemon.LogContainerEventWithAttributes(cntr, "exec_create: "+execConfig.Entrypoint+" "+strings.Join(execConfig.Args, " "), attributes)

	return execConfig.ID, nil
}

// ContainerExecStart starts a previously set up exec instance. The
// std streams are set up.
// If ctx is cancelled, the process is terminated.
func (daemon *Daemon) ContainerExecStart(ctx context.Context, name string, options containertypes.ExecStartOptions) (err error) {
	var (
		cStdin           io.ReadCloser
		cStdout, cStderr io.Writer
	)

	ec, err := daemon.getExecConfig(name)
	if err != nil {
		return err
	}

	ec.Lock()
	if ec.ExitCode != nil {
		ec.Unlock()
		err := fmt.Errorf("Error: Exec command %s has already run", ec.ID)
		return errdefs.Conflict(err)
	}

	if ec.Running {
		ec.Unlock()
		return errdefs.Conflict(fmt.Errorf("Error: Exec command %s is already running", ec.ID))
	}
	ec.Running = true
	ec.Unlock()

	logrus.Debugf("starting exec command %s in container %s", ec.ID, ec.Container.ID)
	attributes := map[string]string{
		"execID": ec.ID,
	}
	daemon.LogContainerEventWithAttributes(ec.Container, "exec_start: "+ec.Entrypoint+" "+strings.Join(ec.Args, " "), attributes)

	defer func() {
		if err != nil {
			ec.Lock()
			ec.Container.ExecCommands.Delete(ec.ID)
			ec.Running = false
			exitCode := 126
			ec.ExitCode = &exitCode
			if err := ec.CloseStreams(); err != nil {
				logrus.Errorf("failed to cleanup exec %s streams: %s", ec.Container.ID, err)
			}
			ec.Unlock()
		}
	}()

	if ec.OpenStdin && options.Stdin != nil {
		r, w := io.Pipe()
		go func() {
			defer w.Close()
			defer logrus.Debug("Closing buffered stdin pipe")
			pools.Copy(w, options.Stdin)
		}()
		cStdin = r
	}
	if ec.OpenStdout {
		cStdout = options.Stdout
	}
	if ec.OpenStderr {
		cStderr = options.Stderr
	}

	if ec.OpenStdin {
		ec.StreamConfig.NewInputPipes()
	} else {
		ec.StreamConfig.NewNopInputPipe()
	}

	p := &specs.Process{}
	if runtime.GOOS != "windows" {
		ctr, err := daemon.containerdCli.LoadContainer(ctx, ec.Container.ID)
		if err != nil {
			return err
		}
		md, err := ctr.Info(ctx, containerd.WithoutRefreshedMetadata)
		if err != nil {
			return err
		}
		spec := specs.Spec{Process: p}
		if err := json.Unmarshal(md.Spec.GetValue(), &spec); err != nil {
			return err
		}
	}
	p.Args = append([]string{ec.Entrypoint}, ec.Args...)
	p.Env = ec.Env
	p.Cwd = ec.WorkingDir
	p.Terminal = ec.Tty

	consoleSize := options.ConsoleSize
	// If size isn't specified for start, use the one provided for create
	if consoleSize == nil {
		consoleSize = ec.ConsoleSize
	}
	if p.Terminal && consoleSize != nil {
		p.ConsoleSize = &specs.Box{
			Height: consoleSize[0],
			Width:  consoleSize[1],
		}
	}

	if p.Cwd == "" {
		p.Cwd = "/"
	}

	if err := daemon.execSetPlatformOpt(ctx, ec, p); err != nil {
		return err
	}

	attachConfig := stream.AttachConfig{
		TTY:        ec.Tty,
		UseStdin:   cStdin != nil,
		UseStdout:  cStdout != nil,
		UseStderr:  cStderr != nil,
		Stdin:      cStdin,
		Stdout:     cStdout,
		Stderr:     cStderr,
		DetachKeys: ec.DetachKeys,
		CloseStdin: true,
	}
	ec.StreamConfig.AttachStreams(&attachConfig)
	// using context.Background() so that attachErr does not race ctx.Done().
	copyCtx, cancel := context.WithCancel(context.Background())
	defer cancel()
	attachErr := ec.StreamConfig.CopyStreams(copyCtx, &attachConfig)

	ec.Container.Lock()
	tsk, err := ec.Container.GetRunningTask()
	ec.Container.Unlock()
	if err != nil {
		return err
	}

	// Synchronize with libcontainerd event loop
	ec.Lock()
	ec.Process, err = tsk.Exec(ctx, ec.ID, p, cStdin != nil, ec.InitializeStdio)
	// the exec context should be ready, or error happened.
	// close the chan to notify readiness
	close(ec.Started)
	if err != nil {
		defer ec.Unlock()
		return translateContainerdStartErr(ec.Entrypoint, ec.SetExitCode, err)
	}
	ec.Unlock()

	select {
	case <-ctx.Done():
		log := logrus.
			WithField("container", ec.Container.ID).
			WithField("exec", ec.ID)
		log.Debug("Sending KILL signal to container process")
		sigCtx, cancelFunc := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancelFunc()
		err := ec.Process.Kill(sigCtx, signal.SignalMap["KILL"])
		if err != nil {
			log.WithError(err).Error("Could not send KILL signal to container process")
		}
		return ctx.Err()
	case err := <-attachErr:
		if err != nil {
			if _, ok := err.(term.EscapeError); !ok {
				return errdefs.System(errors.Wrap(err, "exec attach failed"))
			}
			attributes := map[string]string{
				"execID": ec.ID,
			}
			daemon.LogContainerEventWithAttributes(ec.Container, "exec_detach", attributes)
		}
	}
	return nil
}

// execCommandGC runs a ticker to clean up the daemon references
// of exec configs that are no longer part of the container.
func (daemon *Daemon) execCommandGC() {
	for range time.Tick(5 * time.Minute) {
		var (
			cleaned          int
			liveExecCommands = daemon.containerExecIds()
		)
		for id, config := range daemon.execCommands.Commands() {
			if config.CanRemove {
				cleaned++
				daemon.execCommands.Delete(id)
			} else {
				if _, exists := liveExecCommands[id]; !exists {
					config.CanRemove = true
				}
			}
		}
		if cleaned > 0 {
			logrus.Debugf("clean %d unused exec commands", cleaned)
		}
	}
}

// containerExecIds returns a list of all the current exec ids that are in use
// and running inside a container.
func (daemon *Daemon) containerExecIds() map[string]struct{} {
	ids := map[string]struct{}{}
	for _, c := range daemon.containers.List() {
		for _, id := range c.ExecCommands.List() {
			ids[id] = struct{}{}
		}
	}
	return ids
}

It is the same bit of code!