Make one-shot stats faster
<!--
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:
-->

**- What I did**

Kuberenets CRI interface `ListContainerStats` requires to collect every pod container's stats. Currently `daemon/stats/collector.go` runs in a serial way and sleep `1s` at the end of `for` loop. This causes `ListContainerStats` to suffer long delays if there are many pod containers on the node. 

This PR is to address this issue.

**- How I did it**

- Make the stats collector parallel with the number of CPUs.
- ~`Sleep` is not needed in a normal loop since a condition variable is used. Only sleep when an error happens at system level, then retry in next loop~. `time.Sleep(s.interval)` is needed for `one-short=0` or `stream=1`, otherwise the caller would receive successive stats without any interval.
- Move `bufReader` out of `Collector`, since it cannot be shared by multiple goroutines.
- Get container stats directly (i.e. without using publishers) for `one-shot` requests. 

**- How to verify it**

Run the PR build with `cri-dockerd`, observe the time taken for `ListContainerStats`. Without this code change, it could take over 3 seconds when about 40 pod containers on the node. ~With this fix, it takes less than 1 second~. 
- Making stats collection parallel reduces the `ListContainerStats` time to 2 seconds on the 4-CPU node. 
- Getting container stats directly for `one-shot` requests reduces the `ListContainerStats` time to less than 200ms. (The requests sent by `cri-dockerd` are `one-shot` requests)  

**- Description for the changelog**
<!--
Write a short (one line) summary that describes the changes in this
pull request for inclusion in the changelog:
-->
Improves the performance of the stats collector

**- A picture of a cute animal (not mandatory but encouraged)**


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

import (
	"context"
	"encoding/json"
	"errors"
	"runtime"
	"time"

	"github.com/docker/docker/api/types"
	"github.com/docker/docker/api/types/backend"
	"github.com/docker/docker/api/types/versions"
	"github.com/docker/docker/api/types/versions/v1p20"
	"github.com/docker/docker/container"
	"github.com/docker/docker/errdefs"
	"github.com/docker/docker/pkg/ioutils"
)

// ContainerStats writes information about the container to the stream
// given in the config object.
func (daemon *Daemon) ContainerStats(ctx context.Context, prefixOrName string, config *backend.ContainerStatsConfig) error {
	// Engine API version (used for backwards compatibility)
	apiVersion := config.Version

	if isWindows && versions.LessThan(apiVersion, "1.21") {
		return errors.New("API versions pre v1.21 do not support stats on Windows")
	}

	ctr, err := daemon.GetContainer(prefixOrName)
	if err != nil {
		return err
	}

	if config.Stream && config.OneShot {
		return errdefs.InvalidParameter(errors.New("cannot have stream=true and one-shot=true"))
	}

	// If the container is either not running or restarting and requires no stream, return an empty stats.
	if (!ctr.IsRunning() || ctr.IsRestarting()) && !config.Stream {
		return json.NewEncoder(config.OutStream).Encode(&types.StatsJSON{
			Name: ctr.Name,
			ID:   ctr.ID,
		})
	}

	outStream := config.OutStream
	if config.Stream {
		wf := ioutils.NewWriteFlusher(outStream)
		defer wf.Close()
		wf.Flush()
		outStream = wf
	}

	var preCPUStats types.CPUStats
	var preRead time.Time
	getStatJSON := func(v interface{}) *types.StatsJSON {
		ss := v.(types.StatsJSON)
		ss.Name = ctr.Name
		ss.ID = ctr.ID
		ss.PreCPUStats = preCPUStats
		ss.PreRead = preRead
		preCPUStats = ss.CPUStats
		preRead = ss.Read
		return &ss
	}

	enc := json.NewEncoder(outStream)

	updates := daemon.subscribeToContainerStats(ctr)
	defer daemon.unsubscribeToContainerStats(ctr, updates)

	noStreamFirstFrame := !config.OneShot
	for {
		select {
		case v, ok := <-updates:
			if !ok {
				return nil
			}

			var statsJSON interface{}
			statsJSONPost120 := getStatJSON(v)
			if versions.LessThan(apiVersion, "1.21") {
				var (
					rxBytes   uint64
					rxPackets uint64
					rxErrors  uint64
					rxDropped uint64
					txBytes   uint64
					txPackets uint64
					txErrors  uint64
					txDropped uint64
				)
				for _, v := range statsJSONPost120.Networks {
					rxBytes += v.RxBytes
					rxPackets += v.RxPackets
					rxErrors += v.RxErrors
					rxDropped += v.RxDropped
					txBytes += v.TxBytes
					txPackets += v.TxPackets
					txErrors += v.TxErrors
					txDropped += v.TxDropped
				}
				statsJSON = &v1p20.StatsJSON{
					Stats: statsJSONPost120.Stats,
					Network: types.NetworkStats{
						RxBytes:   rxBytes,
						RxPackets: rxPackets,
						RxErrors:  rxErrors,
						RxDropped: rxDropped,
						TxBytes:   txBytes,
						TxPackets: txPackets,
						TxErrors:  txErrors,
						TxDropped: txDropped,
					},
				}
			} else {
				statsJSON = statsJSONPost120
			}

			if !config.Stream && noStreamFirstFrame {
				// prime the cpu stats so they aren't 0 in the final output
				noStreamFirstFrame = false
				continue
			}

			if err := enc.Encode(statsJSON); err != nil {
				return err
			}

			if !config.Stream {
				return nil
			}
		case <-ctx.Done():
			return nil
		}
	}
}

func (daemon *Daemon) subscribeToContainerStats(c *container.Container) chan interface{} {
	return daemon.statsCollector.Collect(c)
}

func (daemon *Daemon) unsubscribeToContainerStats(c *container.Container, ch chan interface{}) {
	daemon.statsCollector.Unsubscribe(c, ch)
}

// GetContainerStats collects all the stats published by a container
func (daemon *Daemon) GetContainerStats(container *container.Container) (*types.StatsJSON, error) {
	stats, err := daemon.stats(container)
	if err != nil {
		return nil, err
	}

	// We already have the network stats on Windows directly from HCS.
	if !container.Config.NetworkDisabled && runtime.GOOS != "windows" {
		if stats.Networks, err = daemon.getNetworkStats(container); err != nil {
			return nil, err
		}
	}

	return stats, nil
}

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

import (
	"context"
	"encoding/json"
	"errors"
	"runtime"
	"time"

	"github.com/containerd/containerd/log"
	"github.com/docker/docker/api/types"
	"github.com/docker/docker/api/types/backend"
	"github.com/docker/docker/api/types/versions"
	"github.com/docker/docker/api/types/versions/v1p20"
	"github.com/docker/docker/container"
	"github.com/docker/docker/errdefs"
	"github.com/docker/docker/pkg/ioutils"
)

// ContainerStats writes information about the container to the stream
// given in the config object.
func (daemon *Daemon) ContainerStats(ctx context.Context, prefixOrName string, config *backend.ContainerStatsConfig) error {
	// Engine API version (used for backwards compatibility)
	apiVersion := config.Version

	if isWindows && versions.LessThan(apiVersion, "1.21") {
		return errors.New("API versions pre v1.21 do not support stats on Windows")
	}

	ctr, err := daemon.GetContainer(prefixOrName)
	if err != nil {
		return err
	}

	if config.Stream && config.OneShot {
		return errdefs.InvalidParameter(errors.New("cannot have stream=true and one-shot=true"))
	}

	// If the container is either not running or restarting and requires no stream, return an empty stats.
	if (!ctr.IsRunning() || ctr.IsRestarting()) && !config.Stream {
		return json.NewEncoder(config.OutStream).Encode(&types.StatsJSON{
			Name: ctr.Name,
			ID:   ctr.ID,
		})
	}

	// Get container stats directly if OneShot is set
	if config.OneShot {
		stats, err := daemon.GetContainerStats(ctr)
		if err != nil {
			return err
		}
		return json.NewEncoder(config.OutStream).Encode(stats)
	}

	outStream := config.OutStream
	if config.Stream {
		wf := ioutils.NewWriteFlusher(outStream)
		defer wf.Close()
		wf.Flush()
		outStream = wf
	}

	var preCPUStats types.CPUStats
	var preRead time.Time
	getStatJSON := func(v interface{}) *types.StatsJSON {
		ss := v.(types.StatsJSON)
		ss.Name = ctr.Name
		ss.ID = ctr.ID
		ss.PreCPUStats = preCPUStats
		ss.PreRead = preRead
		preCPUStats = ss.CPUStats
		preRead = ss.Read
		return &ss
	}

	enc := json.NewEncoder(outStream)

	updates := daemon.subscribeToContainerStats(ctr)
	defer daemon.unsubscribeToContainerStats(ctr, updates)

	noStreamFirstFrame := !config.OneShot
	for {
		select {
		case v, ok := <-updates:
			if !ok {
				return nil
			}

			var statsJSON interface{}
			statsJSONPost120 := getStatJSON(v)
			if versions.LessThan(apiVersion, "1.21") {
				var (
					rxBytes   uint64
					rxPackets uint64
					rxErrors  uint64
					rxDropped uint64
					txBytes   uint64
					txPackets uint64
					txErrors  uint64
					txDropped uint64
				)
				for _, v := range statsJSONPost120.Networks {
					rxBytes += v.RxBytes
					rxPackets += v.RxPackets
					rxErrors += v.RxErrors
					rxDropped += v.RxDropped
					txBytes += v.TxBytes
					txPackets += v.TxPackets
					txErrors += v.TxErrors
					txDropped += v.TxDropped
				}
				statsJSON = &v1p20.StatsJSON{
					Stats: statsJSONPost120.Stats,
					Network: types.NetworkStats{
						RxBytes:   rxBytes,
						RxPackets: rxPackets,
						RxErrors:  rxErrors,
						RxDropped: rxDropped,
						TxBytes:   txBytes,
						TxPackets: txPackets,
						TxErrors:  txErrors,
						TxDropped: txDropped,
					},
				}
			} else {
				statsJSON = statsJSONPost120
			}

			if !config.Stream && noStreamFirstFrame {
				// prime the cpu stats so they aren't 0 in the final output
				noStreamFirstFrame = false
				continue
			}

			if err := enc.Encode(statsJSON); err != nil {
				return err
			}

			if !config.Stream {
				return nil
			}
		case <-ctx.Done():
			return nil
		}
	}
}

func (daemon *Daemon) subscribeToContainerStats(c *container.Container) chan interface{} {
	return daemon.statsCollector.Collect(c)
}

func (daemon *Daemon) unsubscribeToContainerStats(c *container.Container, ch chan interface{}) {
	daemon.statsCollector.Unsubscribe(c, ch)
}

// GetContainerStats collects all the stats published by a container
func (daemon *Daemon) GetContainerStats(container *container.Container) (*types.StatsJSON, error) {
	stats, err := daemon.stats(container)
	if err != nil {
		goto done
	}

	// Sample system CPU usage close to container usage to avoid
	// noise in metric calculations.
	// FIXME: move to containerd on Linux (not Windows)
	stats.CPUStats.SystemUsage, stats.CPUStats.OnlineCPUs, err = getSystemCPUUsage()
	if err != nil {
		goto done
	}

	// We already have the network stats on Windows directly from HCS.
	if !container.Config.NetworkDisabled && runtime.GOOS != "windows" {
		stats.Networks, err = daemon.getNetworkStats(container)
	}

done:
	switch err.(type) {
	case nil:
		return stats, nil
	case errdefs.ErrConflict, errdefs.ErrNotFound:
		// return empty stats containing only name and ID if not running or not found
		return &types.StatsJSON{
			Name: container.Name,
			ID:   container.ID,
		}, nil
	default:
		log.G(context.TODO()).Errorf("collecting stats for container %s: %v", container.Name, err)
		return nil, err
	}
}

@corhere I pushed a new [commit](https://github.com/moby/moby/commit/3d70af4ee46874fbc859878f3f22ee4f6aa5f5f3) today for this minor change. Please check if it is what you want. Thanks.