diff --git a/daemon/config/config.go b/daemon/config/config.go index 3c11ea03fd..aa90c882b1 100644 --- a/daemon/config/config.go +++ b/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 diff --git a/daemon/containerd/image_pull.go b/daemon/containerd/image_pull.go index 4adc499ffd..e560f57184 100644 --- a/daemon/containerd/image_pull.go +++ b/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) { diff --git a/daemon/containerd/image_push.go b/daemon/containerd/image_push.go index 03396abb5b..2d0ee30b04 100644 --- a/daemon/containerd/image_push.go +++ b/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 { diff --git a/daemon/containerd/service.go b/daemon/containerd/service.go index 8d4ee14630..10b1979058 100644 --- a/daemon/containerd/service.go +++ b/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. diff --git a/daemon/daemon.go b/daemon/daemon.go index 1bc6f4ac76..07cf316082 100644 --- a/daemon/daemon.go +++ b/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 { diff --git a/integration-cli/docker_cli_daemon_test.go b/integration-cli/docker_cli_daemon_test.go index a0074e4597..3314f5abd6 100644 --- a/integration-cli/docker_cli_daemon_test.go +++ b/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