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/checkpoint.go
package daemon // import "github.com/docker/docker/daemon"

import (
	"context"
	"fmt"
	"os"
	"path/filepath"

	"github.com/docker/docker/api/types"
	"github.com/docker/docker/daemon/names"
)

var (
	validCheckpointNameChars   = names.RestrictedNameChars
	validCheckpointNamePattern = names.RestrictedNamePattern
)

// getCheckpointDir verifies checkpoint directory for create,remove, list options and checks if checkpoint already exists
func getCheckpointDir(checkDir, checkpointID, ctrName, ctrID, ctrCheckpointDir string, create bool) (string, error) {
	var checkpointDir string
	var err2 error
	if checkDir != "" {
		checkpointDir = checkDir
	} else {
		checkpointDir = ctrCheckpointDir
	}
	checkpointAbsDir := filepath.Join(checkpointDir, checkpointID)
	stat, err := os.Stat(checkpointAbsDir)
	if create {
		switch {
		case err == nil && stat.IsDir():
			err2 = fmt.Errorf("checkpoint with name %s already exists for container %s", checkpointID, ctrName)
		case err != nil && os.IsNotExist(err):
			err2 = os.MkdirAll(checkpointAbsDir, 0700)
		case err != nil:
			err2 = err
		default:
			err2 = fmt.Errorf("%s exists and is not a directory", checkpointAbsDir)
		}
	} else {
		switch {
		case err != nil:
			err2 = fmt.Errorf("checkpoint %s does not exist for container %s", checkpointID, ctrName)
		case stat.IsDir():
			err2 = nil
		default:
			err2 = fmt.Errorf("%s exists and is not a directory", checkpointAbsDir)
		}
	}
	return checkpointAbsDir, err2
}

// CheckpointCreate checkpoints the process running in a container with CRIU
func (daemon *Daemon) CheckpointCreate(name string, config types.CheckpointCreateOptions) error {
	container, err := daemon.GetContainer(name)
	if err != nil {
		return err
	}

	if !container.IsRunning() {
		return fmt.Errorf("Container %s not running", name)
	}

	if !validCheckpointNamePattern.MatchString(config.CheckpointID) {
		return fmt.Errorf("Invalid checkpoint ID (%s), only %s are allowed", config.CheckpointID, validCheckpointNameChars)
	}

	checkpointDir, err := getCheckpointDir(config.CheckpointDir, config.CheckpointID, name, container.ID, container.CheckpointDir(), true)
	if err != nil {
		return fmt.Errorf("cannot checkpoint container %s: %s", name, err)
	}

	err = daemon.containerd.CreateCheckpoint(context.Background(), container.ID, checkpointDir, config.Exit)
	if err != nil {
		os.RemoveAll(checkpointDir)
		return fmt.Errorf("Cannot checkpoint container %s: %s", name, err)
	}

	daemon.LogContainerEvent(container, "checkpoint")

	return nil
}

// CheckpointDelete deletes the specified checkpoint
func (daemon *Daemon) CheckpointDelete(name string, config types.CheckpointDeleteOptions) error {
	container, err := daemon.GetContainer(name)
	if err != nil {
		return err
	}
	checkpointDir, err := getCheckpointDir(config.CheckpointDir, config.CheckpointID, name, container.ID, container.CheckpointDir(), false)
	if err == nil {
		return os.RemoveAll(checkpointDir)
	}
	return err
}

// CheckpointList lists all checkpoints of the specified container
func (daemon *Daemon) CheckpointList(name string, config types.CheckpointListOptions) ([]types.Checkpoint, error) {
	var out []types.Checkpoint

	container, err := daemon.GetContainer(name)
	if err != nil {
		return nil, err
	}

	checkpointDir, err := getCheckpointDir(config.CheckpointDir, "", name, container.ID, container.CheckpointDir(), false)
	if err != nil {
		return nil, err
	}

	if err := os.MkdirAll(checkpointDir, 0755); err != nil {
		return nil, err
	}

	dirs, err := os.ReadDir(checkpointDir)
	if err != nil {
		return nil, err
	}

	for _, d := range dirs {
		if !d.IsDir() {
			continue
		}
		cpt := types.Checkpoint{Name: d.Name()}
		out = append(out, cpt)
	}

	return out, nil
}

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

import (
	"context"
	"fmt"
	"os"
	"path/filepath"

	"github.com/docker/docker/api/types"
	"github.com/docker/docker/daemon/names"
)

var (
	validCheckpointNameChars   = names.RestrictedNameChars
	validCheckpointNamePattern = names.RestrictedNamePattern
)

// getCheckpointDir verifies checkpoint directory for create,remove, list options and checks if checkpoint already exists
func getCheckpointDir(checkDir, checkpointID, ctrName, ctrID, ctrCheckpointDir string, create bool) (string, error) {
	var checkpointDir string
	var err2 error
	if checkDir != "" {
		checkpointDir = checkDir
	} else {
		checkpointDir = ctrCheckpointDir
	}
	checkpointAbsDir := filepath.Join(checkpointDir, checkpointID)
	stat, err := os.Stat(checkpointAbsDir)
	if create {
		switch {
		case err == nil && stat.IsDir():
			err2 = fmt.Errorf("checkpoint with name %s already exists for container %s", checkpointID, ctrName)
		case err != nil && os.IsNotExist(err):
			err2 = os.MkdirAll(checkpointAbsDir, 0700)
		case err != nil:
			err2 = err
		default:
			err2 = fmt.Errorf("%s exists and is not a directory", checkpointAbsDir)
		}
	} else {
		switch {
		case err != nil:
			err2 = fmt.Errorf("checkpoint %s does not exist for container %s", checkpointID, ctrName)
		case stat.IsDir():
			err2 = nil
		default:
			err2 = fmt.Errorf("%s exists and is not a directory", checkpointAbsDir)
		}
	}
	return checkpointAbsDir, err2
}

// CheckpointCreate checkpoints the process running in a container with CRIU
func (daemon *Daemon) CheckpointCreate(name string, config types.CheckpointCreateOptions) error {
	container, err := daemon.GetContainer(name)
	if err != nil {
		return err
	}

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

	if !validCheckpointNamePattern.MatchString(config.CheckpointID) {
		return fmt.Errorf("Invalid checkpoint ID (%s), only %s are allowed", config.CheckpointID, validCheckpointNameChars)
	}

	checkpointDir, err := getCheckpointDir(config.CheckpointDir, config.CheckpointID, name, container.ID, container.CheckpointDir(), true)
	if err != nil {
		return fmt.Errorf("cannot checkpoint container %s: %s", name, err)
	}

	err = tsk.CreateCheckpoint(context.Background(), checkpointDir, config.Exit)
	if err != nil {
		os.RemoveAll(checkpointDir)
		return fmt.Errorf("Cannot checkpoint container %s: %s", name, err)
	}

	daemon.LogContainerEvent(container, "checkpoint")

	return nil
}

// CheckpointDelete deletes the specified checkpoint
func (daemon *Daemon) CheckpointDelete(name string, config types.CheckpointDeleteOptions) error {
	container, err := daemon.GetContainer(name)
	if err != nil {
		return err
	}
	checkpointDir, err := getCheckpointDir(config.CheckpointDir, config.CheckpointID, name, container.ID, container.CheckpointDir(), false)
	if err == nil {
		return os.RemoveAll(checkpointDir)
	}
	return err
}

// CheckpointList lists all checkpoints of the specified container
func (daemon *Daemon) CheckpointList(name string, config types.CheckpointListOptions) ([]types.Checkpoint, error) {
	var out []types.Checkpoint

	container, err := daemon.GetContainer(name)
	if err != nil {
		return nil, err
	}

	checkpointDir, err := getCheckpointDir(config.CheckpointDir, "", name, container.ID, container.CheckpointDir(), false)
	if err != nil {
		return nil, err
	}

	if err := os.MkdirAll(checkpointDir, 0755); err != nil {
		return nil, err
	}

	dirs, err := os.ReadDir(checkpointDir)
	if err != nil {
		return nil, err
	}

	for _, d := range dirs {
		if !d.IsDir() {
			continue
		}
		cpt := types.Checkpoint{Name: d.Name()}
		out = append(out, cpt)
	}

	return out, nil
}

(Just thinking out loud) should we have an RWMutex instead of a Mutex?