daemon/containerd: Enforce global transfer concurrency limits

Before this commit containerd-backed pulls and pushes ignored the
daemon's max concurrent transfer settings, so each operation can
independently consume the full configured concurrency.

Signed-off-by: Paweł Gronowski <pawel.gronowski@docker.com>

daemon/config/config.go

@@ -211,11 +211,11 @@ type CommonConfig struct {
 	LiveRestoreEnabled bool `json:"live-restore,omitempty"`
 
 	// MaxConcurrentDownloads is the maximum number of downloads that
-	// may take place at a time for each pull.
+	// may take place at a time across all pulls.
 	MaxConcurrentDownloads int `json:"max-concurrent-downloads,omitempty"`
 
 	// MaxConcurrentUploads is the maximum number of uploads that
-	// may take place at a time for each push.
+	// may take place at a time across all pushes.
 	MaxConcurrentUploads int `json:"max-concurrent-uploads,omitempty"`
 
 	// MaxDownloadAttempts is the maximum number of attempts that

daemon/containerd/image_pull.go

@@ -235,6 +235,14 @@ func (i *ImageService) pullTag(ctx context.Context, ref reference.Named, platfor
 	opts = append(opts, containerd.WithImageHandlerWrapper(joinHandlerWrappers(infoHandler, referrers.Handler)))
 	opts = append(opts, containerd.WithReferrersProvider(referrers))
 
+	i.transferLimitMu.Lock()
+	maxConcurrentDownloads := i.maxConcurrentDownloads
+	downloadLimiter := i.downloadLimiter
+	i.transferLimitMu.Unlock()
+	if maxConcurrentDownloads > 0 {
+		opts = append(opts, containerd.WithMaxConcurrentDownloads(maxConcurrentDownloads))
+	}
+	opts = append(opts, containerd.WithDownloadLimiter(downloadLimiter))
 	img, err := i.client.Pull(ctx, ref.String(), opts...)
 	if err != nil {
 		if errors.Is(err, docker.ErrInvalidAuthorization) {

daemon/containerd/image_push.go

@@ -140,7 +140,9 @@ func (i *ImageService) pushRef(ctx context.Context, targetRef reference.Named, p
 		}
 	}()
 
-	var limiter *semaphore.Weighted // TODO: Respect max concurrent downloads/uploads
+	i.transferLimitMu.Lock()
+	limiter := i.uploadLimiter
+	i.transferLimitMu.Unlock()
 
 	mountableBlobs, err := findMissingMountable(ctx, store, jobsQueue, target, targetRef, limiter)
 	if err != nil {

daemon/containerd/service.go

@@ -3,6 +3,7 @@ package containerd
 import (
 	"context"
 	"fmt"
+	"sync"
 	"sync/atomic"
 
 	containerd "github.com/containerd/containerd/v2/client"
@@ -26,6 +27,7 @@ import (
 	"github.com/opencontainers/image-spec/identity"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
+	"golang.org/x/sync/semaphore"
 )
 
 // ImageService implements daemon.ImageService
@@ -45,10 +47,24 @@ type ImageService struct {
 	policyVerifier      func() (*policyverifier.Verifier, error)
 	identity            imageIdentityState
 
+	// transferLimitMu keeps limiter pointers and their settings consistent while
+	// configuration reload replaces them.
+	transferLimitMu        sync.Mutex
+	maxConcurrentDownloads int
+	downloadLimiter        *semaphore.Weighted
+	uploadLimiter          *semaphore.Weighted
+
 	// defaultPlatformOverride is used in tests to override the host platform.
 	defaultPlatformOverride *ocispec.Platform
 }
 
+func newTransferLimiter(maxConcurrent int) *semaphore.Weighted {
+	if maxConcurrent <= 0 {
+		return nil
+	}
+	return semaphore.NewWeighted(int64(maxConcurrent))
+}
+
 type ImageServiceConfig struct {
 	Client                 *containerd.Client
 	Containers             container.Store
@@ -60,10 +76,15 @@ type ImageServiceConfig struct {
 	RefCountMounter        snapshotter.Mounter
 	IDMapping              user.IdentityMapping
 	PolicyVerifierProvider func() (*policyverifier.Verifier, error)
+	MaxConcurrentDownloads int
+	MaxConcurrentUploads   int
 }
 
 // NewService creates a new ImageService.
 func NewService(config ImageServiceConfig) *ImageService {
+	log.G(context.TODO()).Debugf("Max Concurrent Downloads: %d", config.MaxConcurrentDownloads)
+	log.G(context.TODO()).Debugf("Max Concurrent Uploads: %d", config.MaxConcurrentUploads)
+
 	service := &ImageService{
 		client:  config.Client,
 		images:  config.Client.ImageService(),
@@ -89,10 +110,17 @@ func NewService(config ImageServiceConfig) *ImageService {
 			}(),
 		},
 	}
+	service.setTransferLimits(config.MaxConcurrentDownloads, config.MaxConcurrentUploads)
 	service.startImageIdentityCacheRefresh()
 	return service
 }
 
+func (i *ImageService) setTransferLimits(maxDownloads, maxUploads int) {
+	i.maxConcurrentDownloads = maxDownloads
+	i.downloadLimiter = newTransferLimiter(maxDownloads)
+	i.uploadLimiter = newTransferLimiter(maxUploads)
+}
+
 func (i *ImageService) snapshotterService(snapshotter string) snapshots.Snapshotter {
 	s, ok := i.snapshotterServices[snapshotter]
 	if !ok {
@@ -289,7 +317,10 @@ func (i *ImageService) ImageDiskUsage(ctx context.Context) (int64, error) {
 //
 // called from reload.go
 func (i *ImageService) UpdateConfig(maxDownloads, maxUploads int) {
-	log.G(context.TODO()).Warn("max downloads and uploads is not yet implemented with the containerd store")
+	i.transferLimitMu.Lock()
+	defer i.transferLimitMu.Unlock()
+
+	i.setTransferLimits(maxDownloads, maxUploads)
 }
 
 // GetContainerLayerSize returns the real size & virtual size of the container.

daemon/daemon.go

@@ -1321,6 +1321,8 @@ func NewDaemon(ctx context.Context, config *config.Config, pluginStore *plugin.S
 			IDMapping:              idMapping,
 			RefCountMounter:        snapshotter.NewMounter(config.Root, driverName, idMapping),
 			PolicyVerifierProvider: verifierProvider(cfgStore.Root),
+			MaxConcurrentDownloads: config.MaxConcurrentDownloads,
+			MaxConcurrentUploads:   config.MaxConcurrentUploads,
 		})
 
 		if migrationConfig.ImageCount > 0 {

integration-cli/docker_cli_daemon_test.go

@@ -1537,8 +1537,6 @@ func (s *DockerDaemonSuite) TestDaemonLogOptions(c *testing.T) {
 
 // Test case for #20936, #22443
 func (s *DockerDaemonSuite) TestDaemonMaxConcurrency(c *testing.T) {
-	skip.If(c, testEnv.UsingSnapshotter, "max concurrency is not implemented (yet) with containerd snapshotters https://github.com/moby/moby/issues/46610")
-
 	s.d.Start(c, "--max-concurrent-uploads=6", "--max-concurrent-downloads=8")
 
 	expectedMaxConcurrentUploads := `level=debug msg="Max Concurrent Uploads: 6"`
@@ -1551,8 +1549,6 @@ func (s *DockerDaemonSuite) TestDaemonMaxConcurrency(c *testing.T) {
 
 // Test case for #20936, #22443
 func (s *DockerDaemonSuite) TestDaemonMaxConcurrencyWithConfigFile(c *testing.T) {
-	skip.If(c, testEnv.UsingSnapshotter, "max concurrency is not implemented (yet) with containerd snapshotters https://github.com/moby/moby/issues/46610")
-
 	testRequires(c, testEnv.IsLocalDaemon, DaemonIsLinux)
 
 	// daemon config file
@@ -1585,8 +1581,6 @@ func (s *DockerDaemonSuite) TestDaemonMaxConcurrencyWithConfigFile(c *testing.T)
 
 // Test case for #20936, #22443
 func (s *DockerDaemonSuite) TestDaemonMaxConcurrencyWithConfigFileReload(c *testing.T) {
-	skip.If(c, testEnv.UsingSnapshotter, "max concurrency is not implemented (yet) with containerd snapshotters https://github.com/moby/moby/issues/46610")
-
 	testRequires(c, testEnv.IsLocalDaemon, DaemonIsLinux)
 
 	// daemon config file