From 467fa8ccc995688510f3c7858397ab575f8a8e57 Mon Sep 17 00:00:00 2001 From: Guillaume Lours Date: Wed, 30 Sep 2026 19:01:41 +0200 Subject: [PATCH] fix: share --parallel budget across down's ops and up's log-attach down's outer ops dispatch bounded only how many ops ran concurrently, separate from the semaphore already shared by image removal -- a single network plus --rmi could spend up to 2x the configured budget. Network and volume removal now share that same semaphore. up's post-attach log streaming called ContainerInspect directly and passed no limiter to doLogContainer, unlike `compose logs --follow`'s equivalent path, so it never shared --parallel's budget either. The plan executor also held a node's slot for its whole dependency wait instead of just its own work, and internal/sync/tar.go's file sync used a hardcoded cap of 16 instead of --parallel. Signed-off-by: Guillaume Lours Co-Authored-By: Claude Sonnet 5 --- internal/sync/tar.go | 15 +++- internal/sync/tar_test.go | 110 +++++++++++++++++++++-- pkg/compose/compose.go | 14 +-- pkg/compose/containers.go | 2 +- pkg/compose/down.go | 81 ++++++++++------- pkg/compose/down_test.go | 148 +++++++++++++++++++++++++++---- pkg/compose/executor.go | 21 +++-- pkg/compose/executor_test.go | 105 +++++++++++++++++++--- pkg/compose/image_pruner.go | 103 ++++++++++----------- pkg/compose/image_pruner_test.go | 2 +- pkg/compose/restart.go | 2 +- pkg/compose/up.go | 13 ++- pkg/compose/up_test.go | 49 ++++++++++ pkg/compose/watch.go | 2 +- 14 files changed, 522 insertions(+), 145 deletions(-) diff --git a/internal/sync/tar.go b/internal/sync/tar.go index 7e5446881..e2851a285 100644 --- a/internal/sync/tar.go +++ b/internal/sync/tar.go @@ -53,14 +53,19 @@ type Tar struct { client LowLevelClient projectName string + // maxConcurrency bounds concurrent per-container sync calls. <=0 + // (including the Go zero-value) leaves it unlimited, since + // errgroup.SetLimit(0) means "allow zero goroutines", not "unlimited". + maxConcurrency int } var _ Syncer = &Tar{} -func NewTar(projectName string, client LowLevelClient) *Tar { +func NewTar(projectName string, client LowLevelClient, maxConcurrency int) *Tar { return &Tar{ - projectName: projectName, - client: client, + projectName: projectName, + client: client, + maxConcurrency: maxConcurrency, } } @@ -93,7 +98,9 @@ func (t *Tar) Sync(ctx context.Context, service string, paths []*PathMapping) er errs = make([]error, 0, len(containers)*2) // max 2 errs per container ) - eg.SetLimit(16) // arbitrary limit, adjust to taste :D + if t.maxConcurrency > 0 { + eg.SetLimit(t.maxConcurrency) + } for i := range containers { containerID := containers[i].ID tarReader := tarArchive(pathsToCopy, keepImpliedDirectories) diff --git a/internal/sync/tar_test.go b/internal/sync/tar_test.go index 2dafeb16c..2149c012e 100644 --- a/internal/sync/tar_test.go +++ b/internal/sync/tar_test.go @@ -22,7 +22,9 @@ import ( "os" "path/filepath" "runtime" + "sync" "testing" + "time" "github.com/moby/moby/api/types/container" "gotest.tools/v3/assert" @@ -31,6 +33,7 @@ import ( // fakeLowLevelClient records calls made to it for test assertions. type fakeLowLevelClient struct { + mu sync.Mutex containers []container.Summary execCmds [][]string untarCount int @@ -38,6 +41,9 @@ type fakeLowLevelClient struct { untarErrs []error // untarHeaders[i] holds the headers the i-th Untar call received, by entry name untarHeaders []map[string]tar.Header + // onUntar, when set, is invoked at the start of every Untar call -- + // used to observe how many run concurrently. + onUntar func() } func (f *fakeLowLevelClient) ContainersForService(_ context.Context, _ string, _ string) ([]container.Summary, error) { @@ -50,6 +56,9 @@ func (f *fakeLowLevelClient) Exec(_ context.Context, _ string, cmd []string, _ i } func (f *fakeLowLevelClient) Untar(_ context.Context, _ string, reader io.ReadCloser) error { + if f.onUntar != nil { + f.onUntar() + } headers := map[string]tar.Header{} tr := tar.NewReader(reader) for { @@ -62,6 +71,9 @@ func (f *fakeLowLevelClient) Untar(_ context.Context, _ string, reader io.ReadCl } headers[header.Name] = *header } + + f.mu.Lock() + defer f.mu.Unlock() f.untarHeaders = append(f.untarHeaders, headers) f.untarCount++ @@ -79,7 +91,7 @@ func TestSync_ExistingPath(t *testing.T) { client := &fakeLowLevelClient{ containers: []container.Summary{{ID: "ctr1"}}, } - syncer := NewTar("proj", client) + syncer := NewTar("proj", client, 0) err := syncer.Sync(t.Context(), "svc", []*PathMapping{ {HostPath: existingFile, ContainerPath: "/app/exists.txt"}, @@ -94,7 +106,7 @@ func TestSync_NonExistentPath(t *testing.T) { client := &fakeLowLevelClient{ containers: []container.Summary{{ID: "ctr1"}}, } - syncer := NewTar("proj", client) + syncer := NewTar("proj", client, 0) err := syncer.Sync(t.Context(), "svc", []*PathMapping{ {HostPath: "/no/such/file", ContainerPath: "/app/gone.txt"}, @@ -128,7 +140,7 @@ func TestSync_StatPermissionError(t *testing.T) { client := &fakeLowLevelClient{ containers: []container.Summary{{ID: "ctr1"}}, } - syncer := NewTar("proj", client) + syncer := NewTar("proj", client, 0) err := syncer.Sync(t.Context(), "svc", []*PathMapping{ {HostPath: targetFile, ContainerPath: "/app/secret.txt"}, @@ -140,6 +152,88 @@ func TestSync_StatPermissionError(t *testing.T) { assert.Equal(t, len(client.execCmds), 0, "should not attempt delete on stat error") } +// syncConcurrencyTracker records the highest number of overlapping +// enter()/leave() pairs seen, used to assert Sync's concurrency bound. +type syncConcurrencyTracker struct { + mu sync.Mutex + current int + peak int +} + +func (t *syncConcurrencyTracker) enter() { + t.mu.Lock() + t.current++ + if t.current > t.peak { + t.peak = t.current + } + t.mu.Unlock() +} + +func (t *syncConcurrencyTracker) leave() { + t.mu.Lock() + t.current-- + t.mu.Unlock() +} + +func (t *syncConcurrencyTracker) Peak() int { + t.mu.Lock() + defer t.mu.Unlock() + return t.peak +} + +// TestSync_ConcurrencyIsBounded guards internal/sync/tar.go's own fan-out: +// it used to hardcode eg.SetLimit(16) regardless of --parallel, the same +// class of gap fixed everywhere else in this repo for bulk engine calls. +func TestSync_ConcurrencyIsBounded(t *testing.T) { + tracker := &syncConcurrencyTracker{} + client := &fakeLowLevelClient{ + containers: []container.Summary{{ID: "ctr1"}, {ID: "ctr2"}, {ID: "ctr3"}}, + onUntar: func() { + tracker.enter() + time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up + tracker.leave() + }, + } + syncer := NewTar("proj", client, 1) + + tmpDir := t.TempDir() + existingFile := filepath.Join(tmpDir, "exists.txt") + assert.NilError(t, os.WriteFile(existingFile, []byte("data"), 0o644)) + + err := syncer.Sync(t.Context(), "svc", []*PathMapping{ + {HostPath: existingFile, ContainerPath: "/app/exists.txt"}, + }) + assert.NilError(t, err) + assert.Equal(t, tracker.Peak(), 1, "Sync must never run more than maxConcurrency Untar calls at once") +} + +// TestSync_ZeroMaxConcurrencyIsUnlimited guards the errgroup.SetLimit(0) +// footgun: 0 is maxConcurrency's Go zero-value, and SetLimit(0) means "allow +// zero goroutines", not "unlimited" -- which would deadlock Sync forever. +func TestSync_ZeroMaxConcurrencyIsUnlimited(t *testing.T) { + client := &fakeLowLevelClient{ + containers: []container.Summary{{ID: "ctr1"}, {ID: "ctr2"}}, + } + syncer := NewTar("proj", client, 0) + + tmpDir := t.TempDir() + existingFile := filepath.Join(tmpDir, "exists.txt") + assert.NilError(t, os.WriteFile(existingFile, []byte("data"), 0o644)) + + done := make(chan error, 1) + go func() { + done <- syncer.Sync(t.Context(), "svc", []*PathMapping{ + {HostPath: existingFile, ContainerPath: "/app/exists.txt"}, + }) + }() + select { + case err := <-done: + assert.NilError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("Sync deadlocked with maxConcurrency=0") + } +} + func TestSync_MixedPaths(t *testing.T) { tmpDir := t.TempDir() existingFile := filepath.Join(tmpDir, "keep.txt") @@ -148,7 +242,7 @@ func TestSync_MixedPaths(t *testing.T) { client := &fakeLowLevelClient{ containers: []container.Summary{{ID: "ctr1"}}, } - syncer := NewTar("proj", client) + syncer := NewTar("proj", client, 0) err := syncer.Sync(t.Context(), "svc", []*PathMapping{ {HostPath: existingFile, ContainerPath: "/app/keep.txt"}, @@ -170,7 +264,7 @@ func TestSync_PopulatedDirectoryKeepsItsHeader(t *testing.T) { client := &fakeLowLevelClient{ containers: []container.Summary{{ID: "ctr1"}}, } - syncer := NewTar("proj", client) + syncer := NewTar("proj", client, 0) err := syncer.Sync(t.Context(), "svc", []*PathMapping{ {HostPath: filepath.Join(tmpDir, "sub"), ContainerPath: "/app/sub"}, @@ -198,7 +292,7 @@ func TestSync_RetriesWithoutImpliedDirectoriesWhenCopyFails(t *testing.T) { containers: []container.Summary{{ID: "ctr1"}}, untarErrs: []error{errors.New(`cannot overwrite non-directory "/app/sub" with directory "/"`)}, } - syncer := NewTar("proj", client) + syncer := NewTar("proj", client, 0) err := syncer.Sync(t.Context(), "svc", []*PathMapping{ {HostPath: filepath.Join(tmpDir, "sub"), ContainerPath: "/app/sub"}, @@ -224,7 +318,7 @@ func TestSync_ReportsTheCopyErrorWhenTheRetryFailsToo(t *testing.T) { errors.New("no such container"), }, } - syncer := NewTar("proj", client) + syncer := NewTar("proj", client, 0) err := syncer.Sync(t.Context(), "svc", []*PathMapping{ {HostPath: filepath.Join(tmpDir, "sub"), ContainerPath: "/app/sub"}, @@ -243,7 +337,7 @@ func TestSync_DoesNotRetryACopyThatFailedForAnotherReason(t *testing.T) { containers: []container.Summary{{ID: "ctr1"}}, untarErrs: []error{errors.New("no such container")}, } - syncer := NewTar("proj", client) + syncer := NewTar("proj", client, 0) err := syncer.Sync(t.Context(), "svc", []*PathMapping{ {HostPath: filepath.Join(tmpDir, "sub"), ContainerPath: "/app/sub"}, diff --git a/pkg/compose/compose.go b/pkg/compose/compose.go index d97702de4..54929dcab 100644 --- a/pkg/compose/compose.go +++ b/pkg/compose/compose.go @@ -212,22 +212,22 @@ func panicSafeReleaseSlot(limiter *semaphore.Weighted) { } } -// forEachContainerWithLimiter runs fn concurrently for each container, -// bounded by limiter. Unlike newLimitedErrgroup, limiter is built by the -// caller and can be shared across several concurrently-dispatched calls -// (e.g. one per service visited by InDependencyOrder), so the combined +// forEachWithLimiter runs fn concurrently for each item, bounded by limiter. +// Unlike newLimitedErrgroup, limiter is built by the caller and can be +// shared across several concurrently-dispatched calls (e.g. one per service +// visited by InDependencyOrder, or one per sibling down op), so the combined // concurrency across all of them never exceeds the configured budget. // Use forEachContainerConcurrent (containers.go) instead when the call is // standalone and doesn't need to share its budget with any other call. -func forEachContainerWithLimiter(ctx context.Context, limiter *semaphore.Weighted, containers []container.Summary, fn func(context.Context, container.Summary) error) error { +func forEachWithLimiter[T any](ctx context.Context, limiter *semaphore.Weighted, items []T, fn func(context.Context, T) error) error { eg, ctx := errgroup.WithContext(ctx) - for _, ctr := range containers { + for _, item := range items { eg.Go(func() error { if err := acquireSlot(ctx, limiter); err != nil { return err } defer releaseSlot(limiter) - return fn(ctx, ctr) + return fn(ctx, item) }) } return eg.Wait() diff --git a/pkg/compose/containers.go b/pkg/compose/containers.go index 7f94b14ce..01ea9d272 100644 --- a/pkg/compose/containers.go +++ b/pkg/compose/containers.go @@ -210,7 +210,7 @@ func (containers Containers) filter(predicates ...containerPredicate) Containers } // forEachContainerConcurrent runs fn for every container concurrently and -// waits for all goroutines. Use forEachContainerWithLimiter (compose.go) +// waits for all goroutines. Use forEachWithLimiter (compose.go) // instead when the concurrency budget must be shared across several // concurrently-dispatched calls, e.g. one per service visited by // InDependencyOrder. diff --git a/pkg/compose/down.go b/pkg/compose/down.go index a87617601..d10ee50ca 100644 --- a/pkg/compose/down.go +++ b/pkg/compose/down.go @@ -30,6 +30,7 @@ import ( "github.com/moby/moby/api/types/image" "github.com/moby/moby/client" "github.com/sirupsen/logrus" + "golang.org/x/sync/errgroup" "golang.org/x/sync/semaphore" "github.com/docker/compose/v5/pkg/api" @@ -120,28 +121,35 @@ func (s *composeService) down(ctx context.Context, projectName string, options a return err } - ops := s.ensureNetworksDown(ctx, project) + ops := s.ensureNetworksDown(ctx, project, limiter) if options.Images != "" { - ops = append(ops, s.ensureImagesDown(ctx, project, options)...) + ops = append(ops, s.ensureImagesDown(ctx, project, options, limiter)...) } if options.Volumes { - ops = append(ops, s.ensureVolumesDown(ctx, project)...) + ops = append(ops, s.ensureVolumesDown(ctx, project, limiter)...) } if !resourceToRemove && len(ops) == 0 { logrus.Warnf("Warning: No resource found to remove for project %q.", projectName) } - eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) + // ops dispatch itself is intentionally unbounded: each op already gates + // its own engine call(s) on the shared limiter (network/volume ops + // directly, image ops internally via their own forEachWithLimiter call), + // so bounding this dispatch too on the same limiter would have an op + // hold a slot for its own fan-out to acquire from -- deadlocking once + // enough ops are in flight to exhaust it (see + // TestDown_NetworkAndImageRemovalShareConcurrencyBudget). + eg, ctx := errgroup.WithContext(ctx) for _, op := range ops { eg.Go(op) } return eg.Wait() } -func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.Project) []downOp { +func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.Project, limiter *semaphore.Weighted) []downOp { var ops []downOp for _, vol := range project.Volumes { if vol.External { @@ -149,6 +157,10 @@ func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.P } volumeName := vol.Name ops = append(ops, func() error { + if err := acquireSlot(ctx, limiter); err != nil { + return err + } + defer releaseSlot(limiter) return s.removeVolume(ctx, volumeName) }) } @@ -156,7 +168,7 @@ func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.P return ops } -func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Project, options api.DownOptions) []downOp { +func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Project, options api.DownOptions, limiter *semaphore.Weighted) []downOp { pruneOpts := ImagePruneOptions{ Mode: ImagePruneMode(options.Images), RemoveOrphans: options.RemoveOrphans, @@ -165,7 +177,7 @@ func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Pr var ops []downOp if pruneOpts.Mode != ImagePruneNone { ops = append(ops, func() error { - return s.removeTaggedImagesOp(ctx, project, pruneOpts) + return s.removeTaggedImagesOp(ctx, project, pruneOpts, limiter) }) // a dangling image from a service no longer in the project must be @@ -176,7 +188,7 @@ func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Pr } projectName := project.Name ops = append(ops, func() error { - return s.removeDanglingImagesOp(ctx, projectName, keep) + return s.removeDanglingImagesOp(ctx, projectName, keep, limiter) }) } return ops @@ -189,8 +201,12 @@ func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Pr // sibling down ops (network, volumes, dangling images), since — unlike // this op — they've already been scheduled onto the same errgroup by the // time `ensureImagesDown` used to fail synchronously. -func (s *composeService) removeTaggedImagesOp(ctx context.Context, project *types.Project, pruneOpts ImagePruneOptions) error { - images, err := NewImagePruner(s.apiClient(), project, s.maxConcurrency).ImagesToPrune(ctx, pruneOpts) +// +// limiter is shared with the rest of down's ops (see down()'s comment) so +// this op's fan-out doesn't spend its own full maxConcurrency budget on top +// of removeDanglingImagesOp's, which runs concurrently with it. +func (s *composeService) removeTaggedImagesOp(ctx context.Context, project *types.Project, pruneOpts ImagePruneOptions, limiter *semaphore.Weighted) error { + images, err := NewImagePruner(s.apiClient(), project, limiter).ImagesToPrune(ctx, pruneOpts) if err != nil { s.events.On(errorEvent("Tagged images", err.Error())) return err @@ -198,22 +214,21 @@ func (s *composeService) removeTaggedImagesOp(ctx context.Context, project *type var mu sync.Mutex var errs []error - eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) - for i := range images { - img := images[i] - eg.Go(func() error { - if err := s.removeResource("Image "+img, func() error { - _, err := s.apiClient().ImageRemove(ctx, img, client.ImageRemoveOptions{}) - return err - }); err != nil { - mu.Lock() - errs = append(errs, err) - mu.Unlock() - } - return nil - }) - } - _ = eg.Wait() // errgroup is only used for fan-out here; goroutines never return an error + // forEachWithLimiter's own error handling isn't used here since a single + // image failure must not cancel the shared errgroup ctx (see + // TestRemoveTaggedImagesOp_ContinuesAfterOneImageFails) -- every failure + // is collected instead, so fn itself always returns nil. + _ = forEachWithLimiter(ctx, limiter, images, func(ctx context.Context, img string) error { + if err := s.removeResource("Image "+img, func() error { + _, err := s.apiClient().ImageRemove(ctx, img, client.ImageRemoveOptions{}) + return err + }); err != nil { + mu.Lock() + errs = append(errs, err) + mu.Unlock() + } + return nil + }) return errors.Join(errs...) } @@ -237,7 +252,7 @@ func (s *composeService) removeTaggedImagesOp(ctx context.Context, project *type // removed either: stillInUse (a real, not-yet-achieved goal) is checked // separately, the same distinction removeResource already makes per tagged // image (an already-gone image simply isn't counted in either bucket). -func (s *composeService) removeDanglingImagesOp(ctx context.Context, projectName string, keep func(image.Summary) bool) error { +func (s *composeService) removeDanglingImagesOp(ctx context.Context, projectName string, keep func(image.Summary) bool, limiter *semaphore.Weighted) error { eventID := "Dangling images" eligible, err := s.eligibleDanglingImages(ctx, projectName, keep) if err != nil { @@ -249,7 +264,7 @@ func (s *composeService) removeDanglingImagesOp(ctx context.Context, projectName } s.events.On(removingEvent(eventID)) - removed, stillInUse, err := s.removeImages(ctx, eligible) + removed, stillInUse, err := s.removeImages(ctx, eligible, limiter) var stillInUseNote string if len(stillInUse) > 0 { stillInUseNote = fmt.Sprintf("%d image(s) still in use", len(stillInUse)) @@ -281,7 +296,7 @@ func (s *composeService) removeDanglingImagesOp(ctx context.Context, projectName return err } -func (s *composeService) ensureNetworksDown(ctx context.Context, project *types.Project) []downOp { +func (s *composeService) ensureNetworksDown(ctx context.Context, project *types.Project, limiter *semaphore.Weighted) []downOp { var ops []downOp for key, n := range project.Networks { if n.External { @@ -291,6 +306,10 @@ func (s *composeService) ensureNetworksDown(ctx context.Context, project *types. networkKey := key idOrName := n.Name ops = append(ops, func() error { + if err := acquireSlot(ctx, limiter); err != nil { + return err + } + defer releaseSlot(limiter) return s.removeNetwork(ctx, networkKey, project.Name, idOrName) }) } @@ -437,7 +456,7 @@ func (s *composeService) stopContainers( listener api.ContainerEventListener, limiter *semaphore.Weighted, ) error { - return forEachContainerWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error { + return forEachWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error { return s.stopContainer(ctx, serv, ctr, timeout, listener) }) } @@ -450,7 +469,7 @@ func (s *composeService) removeContainers( volumes bool, limiter *semaphore.Weighted, ) error { - return forEachContainerWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error { + return forEachWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error { return s.stopAndRemoveContainer(ctx, ctr, service, timeout, volumes) }) } diff --git a/pkg/compose/down_test.go b/pkg/compose/down_test.go index 7c179231c..6c9d228e3 100644 --- a/pkg/compose/down_test.go +++ b/pkg/compose/down_test.go @@ -21,6 +21,7 @@ import ( "errors" "fmt" "os" + "sort" "strings" "testing" "time" @@ -135,6 +136,113 @@ func TestDown_ConcurrencyIsBoundedAcrossServices(t *testing.T) { assert.Equal(t, tracker.Peak(), 1, "down must never run more than maxConcurrency ContainerRemove calls at once") } +// TestDown_ImagePruningSharesConcurrencyBudgetAcrossOps guards against a +// regression flagged in review: removeTaggedImagesOp and +// removeDanglingImagesOp each opened their own bounded errgroup sized to the +// full maxConcurrency budget, but run concurrently as two independent down +// ops -- so combined image-removal calls could reach 2x maxConcurrency +// instead of sharing one budget, the same class of leak +// TestDown_ConcurrencyIsBoundedAcrossServices guards for container removal. +func TestDown_ImagePruningSharesConcurrencyBudgetAcrossOps(t *testing.T) { + svc, apiClient := newTestService(t, WithMaxConcurrency(2)) + + project := &types.Project{Name: "prj"} + + apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()). + Return(client.ContainerListResult{}, nil) + apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()). + Return(client.ContainerListResult{}, nil) // removePreStartHookContainers lookup + + apiClient.EXPECT().ImageList(gomock.Any(), client.ImageListOptions{ + Filters: projectFilter("prj").Add("dangling", "false"), + }).Return(client.ImageListResult{Items: []image.Summary{ + {ID: "sha256:tag1", RepoTags: []string{"repo/tag1:latest"}}, + {ID: "sha256:tag2", RepoTags: []string{"repo/tag2:latest"}}, + }}, nil) + apiClient.EXPECT().ImageList(gomock.Any(), client.ImageListOptions{ + Filters: projectFilter("prj").Add("dangling", "true"), + }).Return(client.ImageListResult{Items: []image.Summary{ + {ID: "sha256:dangling1"}, + {ID: "sha256:dangling2"}, + }}, nil) + + tracker := &peakConcurrencyTracker{} + apiClient.EXPECT().ImageRemove(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(context.Context, string, client.ImageRemoveOptions) (client.ImageRemoveResult, error) { + tracker.enter() + time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up + tracker.leave() + return client.ImageRemoveResult{}, nil + }). + Times(4) + + err := svc.down(t.Context(), "prj", compose.DownOptions{Project: project, Images: "local", RemoveOrphans: true}) + assert.NilError(t, err) + assert.Assert(t, tracker.Peak() <= 2, "tagged- and dangling-image removal must share the same budget, got peak %d", tracker.Peak()) +} + +// TestDown_NetworkAndImageRemovalShareConcurrencyBudget guards against a +// regression flagged in review: the outer ops dispatch bounded only the +// *count* of concurrently-running down ops, independent of the shared +// limiter each op's own engine calls are gated on -- so a single ordinary +// project network plus --rmi could spend up to 2x maxConcurrency, the same +// class of leak TestDown_ImagePruningSharesConcurrencyBudgetAcrossOps guards +// between the two image ops. +func TestDown_NetworkAndImageRemovalShareConcurrencyBudget(t *testing.T) { + svc, apiClient := newTestService(t, WithMaxConcurrency(2)) + + project := &types.Project{ + Name: "prj", + Networks: types.Networks{ + "default": {Name: "prj_default"}, + }, + } + + apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()). + Return(client.ContainerListResult{}, nil) + apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()). + Return(client.ContainerListResult{}, nil) // removePreStartHookContainers lookup + + apiClient.EXPECT().ImageList(gomock.Any(), client.ImageListOptions{ + Filters: projectFilter("prj").Add("dangling", "false"), + }).Return(client.ImageListResult{Items: []image.Summary{ + {ID: "sha256:tag1", RepoTags: []string{"repo/tag1:latest"}}, + {ID: "sha256:tag2", RepoTags: []string{"repo/tag2:latest"}}, + }}, nil) + apiClient.EXPECT().ImageList(gomock.Any(), client.ImageListOptions{ + Filters: projectFilter("prj").Add("dangling", "true"), + }).Return(client.ImageListResult{}, nil) + + tracker := &peakConcurrencyTracker{} + apiClient.EXPECT().ImageRemove(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(context.Context, string, client.ImageRemoveOptions) (client.ImageRemoveResult, error) { + tracker.enter() + time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up + tracker.leave() + return client.ImageRemoveResult{}, nil + }). + Times(2) + + apiClient.EXPECT().NetworkList(gomock.Any(), client.NetworkListOptions{ + Filters: projectFilter("prj").Add("label", networkFilter("default")), + }).Return(client.NetworkListResult{Items: []network.Summary{ + {Network: network.Network{ID: "net1", Name: "prj_default"}}, + }}, nil) + apiClient.EXPECT().NetworkInspect(gomock.Any(), "net1", gomock.Any()). + DoAndReturn(func(context.Context, string, client.NetworkInspectOptions) (client.NetworkInspectResult, error) { + tracker.enter() + time.Sleep(20 * time.Millisecond) + tracker.leave() + return client.NetworkInspectResult{Network: network.Inspect{Network: network.Network{ID: "net1"}}}, nil + }) + apiClient.EXPECT().NetworkRemove(gomock.Any(), "net1", gomock.Any()). + Return(client.NetworkRemoveResult{}, nil) + + err := svc.down(t.Context(), "prj", compose.DownOptions{Project: project, Images: "local", RemoveOrphans: true}) + assert.NilError(t, err) + assert.Assert(t, tracker.Peak() <= 2, "network- and image-removal ops must share the same concurrency budget, got peak %d", tracker.Peak()) +} + func TestDownWithGivenServices(t *testing.T) { mockCtrl := gomock.NewController(t) defer mockCtrl.Finish() @@ -534,7 +642,7 @@ func TestEnsureImagesDown_TaggedImageListingFailureStaysVisibleWithoutAbortingDo Filters: projectFilter("prj").Add("dangling", "true"), }).Return(client.ImageListResult{}, nil) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op opErr := ops[0]() @@ -595,19 +703,25 @@ func TestRemoveTaggedImagesOp_ContinuesAfterOneImageFails(t *testing.T) { apiClient.EXPECT().ImageRemove(liveCtx{}, "repo/c:latest", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, nil) - opErr := svc.removeTaggedImagesOp(t.Context(), project, ImagePruneOptions{Mode: ImagePruneLocal, RemoveOrphans: true}) + opErr := svc.removeTaggedImagesOp(t.Context(), project, ImagePruneOptions{Mode: ImagePruneLocal, RemoveOrphans: true}, newOptionalLimiter(1)) assert.ErrorContains(t, opErr, daemonErr.Error()) + // dispatch order across images is no longer deterministic now that the + // concurrency bound comes from a shared limiter acquired inside each + // goroutine (see removeImages) rather than errgroup.SetLimit blocking the + // submitting loop itself -- only the per-image Removing->Removed sequence + // (enforced by removeResource) still holds, so compare as a set. events := make([]string, len(rec.resources)) for i, e := range rec.resources { events[i] = e.ID + ": " + e.Text } + sort.Strings(events) assert.DeepEqual(t, events, []string{ "Image repo/a:latest: Removing", - "Image repo/b:latest: Removing", "Image repo/b:latest: Removed", - "Image repo/c:latest: Removing", + "Image repo/b:latest: Removing", "Image repo/c:latest: Removed", + "Image repo/c:latest: Removing", }) } @@ -641,7 +755,7 @@ func TestEnsureImagesDown_ReportsDanglingImagesAsOneGroupedEvent(t *testing.T) { // RemoveOrphans:true bypasses the per-service keep check so this test // stays focused on the single-grouped-event behavior. - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) for _, op := range ops { assert.NilError(t, op()) } @@ -690,7 +804,7 @@ func TestEnsureImagesDown_SparesDanglingImagesOfOrphanedServices(t *testing.T) { apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:web-dangling", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, nil) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local"}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local"}, nil) for _, op := range ops { assert.NilError(t, op()) } @@ -725,7 +839,7 @@ func TestEnsureImagesDown_RemoveOrphansAlsoTakesDanglingImages(t *testing.T) { apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:orphan-dangling", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, nil) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) for _, op := range ops { assert.NilError(t, op()) } @@ -762,7 +876,7 @@ func TestEnsureImagesDown_NoDanglingImagesToRemove(t *testing.T) { Filters: projectFilter("prj").Add("dangling", "true"), }).Return(client.ImageListResult{}, nil) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op for _, op := range ops { assert.NilError(t, op()) @@ -785,7 +899,7 @@ func TestEnsureImagesDown_AllDanglingImagesSparedStaysSilent(t *testing.T) { }}, nil) // no ImageRemove expectation — a call for the spared image fails the test - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local"}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local"}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op for _, op := range ops { assert.NilError(t, op()) @@ -812,7 +926,7 @@ func TestEnsureImagesDown_DanglingListFailureStaysVisibleWithoutAbortingDown(t * // ensureImagesDown itself must not fail: the listing error is only // surfaced when the returned op actually runs, same as every other // down op, so it can't block ops collected/scheduled around it. - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds @@ -848,7 +962,7 @@ func TestEnsureImagesDown_PartialRemovalFailureStaysVisibleAlongsideRemoved(t *t apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:fails", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, errdefs.ErrPermissionDenied.WithMessage("permission denied")) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds @@ -879,7 +993,7 @@ func TestEnsureImagesDown_TotalRemovalFailureReportsPlainError(t *testing.T) { apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:fails", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, errdefs.ErrPermissionDenied.WithMessage("permission denied")) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds @@ -915,7 +1029,7 @@ func TestEnsureImagesDown_TotalFailureAlsoMentionsStillInUse(t *testing.T) { apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:fails", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, errdefs.ErrPermissionDenied.WithMessage("permission denied")) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds @@ -949,7 +1063,7 @@ func TestEnsureImagesDown_AllEligibleStillInUseDoesNotClaimRemoved(t *testing.T) apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:in-use", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, errdefs.ErrConflict.WithMessage("image is being used by a container")) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds @@ -983,7 +1097,7 @@ func TestEnsureImagesDown_MixedRemovedAndStillInUseKeepsBothVisible(t *testing.T apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:in-use", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, errdefs.ErrConflict.WithMessage("image is being used by a container")) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds @@ -1020,7 +1134,7 @@ func TestEnsureImagesDown_PartialFailureAlsoMentionsStillInUse(t *testing.T) { apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:fails", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, errdefs.ErrPermissionDenied.WithMessage("permission denied")) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds @@ -1052,7 +1166,7 @@ func TestEnsureImagesDown_AllEligibleAlreadyGoneReportsNoResourceFound(t *testin apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:already-gone", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, errdefs.ErrNotFound.WithMessage("already removed")) - ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}) + ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil) assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds diff --git a/pkg/compose/executor.go b/pkg/compose/executor.go index 8c7ace721..f69cd155c 100644 --- a/pkg/compose/executor.go +++ b/pkg/compose/executor.go @@ -22,6 +22,7 @@ import ( "sync" "github.com/compose-spec/compose-go/v2/types" + "golang.org/x/sync/errgroup" ) // planExecutor executes a reconciliation Plan by walking the DAG and performing @@ -101,13 +102,14 @@ func (exec *planExecutor) run(ctx context.Context, plan *Plan) error { groups := exec.buildGroupTracker(plan) events := exec.compose.events - // Each node's goroutine occupies its concurrency slot for the entire wait - // below, not just its own work, so a small maxConcurrency can serialize - // more than a caller might expect on a wide/shallow DAG. Forward progress - // is still guaranteed: plan.Nodes is topologically sorted, so a node's - // dependencies were always already dispatched to eg.Go by the time this - // loop reaches it. - eg, ctx := newLimitedErrgroup(ctx, exec.compose.maxConcurrency) + // Every node's goroutine is dispatched unconditionally, so one waiting on + // a dependency never occupies a concurrency slot -- only the actual + // executeNode call does, via the semaphore below. This also means + // deadlock-freedom no longer depends on plan.Nodes staying topologically + // sorted: a goroutine blocked on <-done[dep.ID] holds no slot for a + // still-pending dependency to be starved on. + eg, ctx := errgroup.WithContext(ctx) + limiter := newOptionalLimiter(exec.compose.maxConcurrency) for _, node := range plan.Nodes { eg.Go(func() error { // Wait for all dependencies @@ -119,6 +121,11 @@ func (exec *planExecutor) run(ctx context.Context, plan *Plan) error { } } + if err := acquireSlot(ctx, limiter); err != nil { + return err + } + defer releaseSlot(limiter) + // Emit group start event if this is the first node of a group groups.onNodeStart(node, events) diff --git a/pkg/compose/executor_test.go b/pkg/compose/executor_test.go index 1440ee796..e8da0c64f 100644 --- a/pkg/compose/executor_test.go +++ b/pkg/compose/executor_test.go @@ -370,17 +370,14 @@ func TestExecutePlanConcurrentRemovesCacheCoherence(t *testing.T) { "all removed containers should be dropped from the live view") } -// TestExecutePlanRespectsMaxConcurrencyAcrossDependencyChain guards the -// invariant run() relies on to stay deadlock-free once maxConcurrency bounds -// the errgroup (see the comment on newLimitedErrgroup's call in run()): -// plan.Nodes must stay topologically sorted, so a node's dependencies are -// always already dispatched to eg.Go by the time the dispatch loop reaches -// a dependent. With maxConcurrency=1, a multi-level dependency chain forces -// strictly serial execution; if that invariant were ever broken (a node -// added before a dependency it references), the dispatch loop would stall -// forever waiting for a slot held by a goroutine itself waiting on a -// not-yet-dispatched dependency — so this test would hang instead of -// completing. +// TestExecutePlanRespectsMaxConcurrencyAcrossDependencyChain guards that a +// multi-level dependency chain still executes serially under +// maxConcurrency=1 and completes rather than deadlocking. run() dispatches +// every node's goroutine unconditionally and only acquires a concurrency +// slot around the actual executeNode call (see run()'s comment), so this no +// longer depends on plan.Nodes staying topologically sorted -- a goroutine +// blocked on <-done[dep.ID] holds no slot for its dependency to starve on, +// however plan.Nodes is ordered. func TestExecutePlanRespectsMaxConcurrencyAcrossDependencyChain(t *testing.T) { svc, apiClient := newTestService(t, WithMaxConcurrency(1)) @@ -424,7 +421,91 @@ func TestExecutePlanRespectsMaxConcurrencyAcrossDependencyChain(t *testing.T) { case err := <-done: assert.NilError(t, err) case <-time.After(5 * time.Second): - t.Fatal("run() deadlocked: a bounded errgroup requires plan.Nodes to stay topologically sorted") + t.Fatal("run() deadlocked on a serial dependency chain under maxConcurrency=1") + } +} + +// TestExecutePlanIndependentNodeNotSerializedBehindADependencyWait guards +// that a node with no unmet dependencies can run concurrently with an +// unrelated node that's still waiting on its own dependency, instead of +// being starved behind it. Before run() acquired its concurrency slot only +// around executeNode, a waiting node's goroutine held its slot for the whole +// wait, so on a wide/shallow DAG an independent, ready node could be blocked +// from even dispatching until the waiter's chain finished -- the two +// ContainerStop calls below could never overlap under the old behavior, +// since the independent node's slot only freed once the dependency chain's +// own goroutine (holding the other slot for its entire wait) had returned. +func TestExecutePlanIndependentNodeNotSerializedBehindADependencyWait(t *testing.T) { + svc, apiClient := newTestService(t, WithMaxConcurrency(2)) + + slow := container.Summary{ID: "c-slow", Names: []string{"/test-web-1"}, Labels: map[string]string{ + api.ServiceLabel: "web", api.ContainerNumberLabel: "1", + }} + waiter := container.Summary{ID: "c-waiter", Names: []string{"/test-web-2"}, Labels: map[string]string{ + api.ServiceLabel: "web", api.ContainerNumberLabel: "2", + }} + independent := container.Summary{ID: "c-independent", Names: []string{"/test-app-1"}, Labels: map[string]string{ + api.ServiceLabel: "app", api.ContainerNumberLabel: "1", + }} + + // slow blocks mid-call until the test releases it, so the assertion below + // is a deterministic happens-before check, not a timing race. + slowStarted := make(chan struct{}) + slowRelease := make(chan struct{}) + apiClient.EXPECT().ContainerStop(gomock.Any(), slow.ID, gomock.Any()). + DoAndReturn(func(context.Context, string, client.ContainerStopOptions) (client.ContainerStopResult, error) { + close(slowStarted) + <-slowRelease + return client.ContainerStopResult{}, nil + }) + apiClient.EXPECT().ContainerStop(gomock.Any(), waiter.ID, gomock.Any()). + Return(client.ContainerStopResult{}, nil) + + independentStarted := make(chan struct{}) + apiClient.EXPECT().ContainerStop(gomock.Any(), independent.ID, gomock.Any()). + DoAndReturn(func(context.Context, string, client.ContainerStopOptions) (client.ContainerStopResult, error) { + close(independentStarted) + return client.ContainerStopResult{}, nil + }) + + // slow -> waiter (waiter depends on slow); independent has no deps and + // shares no resource with either. + plan := &Plan{} + slowNode := plan.addNode(Operation{ + Type: OpStopContainer, ResourceID: "service:web:1", Cause: "chain", Container: &slow, + }, "") + plan.addNode(Operation{ + Type: OpStopContainer, ResourceID: "service:web:2", Cause: "chain", Container: &waiter, + }, "", slowNode) + plan.addNode(Operation{ + Type: OpStopContainer, ResourceID: "service:app:1", Cause: "unrelated", Container: &independent, + }, "") + + exec := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test")) + + done := make(chan error, 1) + go func() { done <- exec.run(t.Context(), plan) }() + + select { + case <-slowStarted: + case <-time.After(2 * time.Second): + t.Fatal("slow node never started") + } + + select { + case <-independentStarted: + // the independent node ran while slow is still blocked -- it wasn't + // serialized behind the dependency chain's wait. + case <-time.After(2 * time.Second): + t.Fatal("independent node never started while the slow node (holding the other slot) was still running -- it was serialized behind the dependency chain") + } + close(slowRelease) + + select { + case err := <-done: + assert.NilError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("run() did not complete after releasing the slow node") } } diff --git a/pkg/compose/image_pruner.go b/pkg/compose/image_pruner.go index 0b1e861fe..13461c072 100644 --- a/pkg/compose/image_pruner.go +++ b/pkg/compose/image_pruner.go @@ -29,6 +29,7 @@ import ( "github.com/moby/moby/api/types/image" "github.com/moby/moby/client" "github.com/sirupsen/logrus" + "golang.org/x/sync/semaphore" "github.com/docker/compose/v5/pkg/api" ) @@ -59,17 +60,22 @@ type ImagePruneOptions struct { // ImagePruner handles image removal during Compose `down` operations. type ImagePruner struct { - client client.ImageAPIClient - project *types.Project - maxConcurrency int + client client.ImageAPIClient + project *types.Project + // limiter bounds concurrent engine calls, shared with the rest of the + // down op set so tagged- and dangling-image removal (which run as two + // independent, concurrently-dispatched ops) don't each spend their own + // full budget on top of one another. + limiter *semaphore.Weighted } -// NewImagePruner creates an ImagePruner object for a project. -func NewImagePruner(imageClient client.ImageAPIClient, project *types.Project, maxConcurrency int) *ImagePruner { +// NewImagePruner creates an ImagePruner object for a project. limiter may be +// nil, meaning unbounded. +func NewImagePruner(imageClient client.ImageAPIClient, project *types.Project, limiter *semaphore.Weighted) *ImagePruner { return &ImagePruner{ - client: imageClient, - project: project, - maxConcurrency: maxConcurrency, + client: imageClient, + project: project, + limiter: limiter, } } @@ -175,37 +181,36 @@ func (p *ImagePruner) labeledLocalImages(ctx context.Context) ([]image.Summary, // removal, so it's simply not counted against either bucket. Any other // removal error is joined into err so callers reporting failures to the // user keep the actual daemon error instead of just an image ID. -func (s *composeService) removeImages(ctx context.Context, images []image.Summary) (removed, stillInUse []string, err error) { +func (s *composeService) removeImages(ctx context.Context, images []image.Summary, limiter *semaphore.Weighted) (removed, stillInUse []string, err error) { var mu sync.Mutex var errs []error - eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) - for _, img := range images { - eg.Go(func() error { - if _, err := s.apiClient().ImageRemove(ctx, img.ID, client.ImageRemoveOptions{}); err != nil { - mu.Lock() - defer mu.Unlock() - imgStillInUse, alreadyGone := classifyRemovalError(err) - switch { - case alreadyGone: - // e.g. removed concurrently by something else - logrus.Debugf("dangling image %s already removed: %v", img.ID, err) - case imgStillInUse: - // the same benign skip the tagged-image path reports via - // removeResource's conflict branch - logrus.Debugf("dangling image %s still in use: %v", img.ID, err) - stillInUse = append(stillInUse, img.ID) - default: - errs = append(errs, fmt.Errorf("image %s: %w", img.ID, err)) - } - return nil - } + // as in removeTaggedImagesOp, a single image failure must not cancel the + // shared errgroup ctx, so fn always returns nil and collects into the + // buckets below instead. + _ = forEachWithLimiter(ctx, limiter, images, func(ctx context.Context, img image.Summary) error { + if _, err := s.apiClient().ImageRemove(ctx, img.ID, client.ImageRemoveOptions{}); err != nil { mu.Lock() defer mu.Unlock() - removed = append(removed, img.ID) + imgStillInUse, alreadyGone := classifyRemovalError(err) + switch { + case alreadyGone: + // e.g. removed concurrently by something else + logrus.Debugf("dangling image %s already removed: %v", img.ID, err) + case imgStillInUse: + // the same benign skip the tagged-image path reports via + // removeResource's conflict branch + logrus.Debugf("dangling image %s still in use: %v", img.ID, err) + stillInUse = append(stillInUse, img.ID) + default: + errs = append(errs, fmt.Errorf("image %s: %w", img.ID, err)) + } return nil - }) - } - _ = eg.Wait() // errgroup is only used for fan-out here; goroutines never return an error + } + mu.Lock() + defer mu.Unlock() + removed = append(removed, img.ID) + return nil + }) return removed, stillInUse, errors.Join(errs...) } @@ -238,7 +243,7 @@ func (s *composeService) removeDanglingImages(ctx context.Context, projectName s if err != nil { return nil, err } - removed, _, err := s.removeImages(ctx, eligible) + removed, _, err := s.removeImages(ctx, eligible, newOptionalLimiter(s.maxConcurrency)) return removed, err } @@ -273,23 +278,19 @@ func (p *ImagePruner) filterImagesByExistence(ctx context.Context, imageNames [] var mu sync.Mutex var ret []string - eg, ctx := newLimitedErrgroup(ctx, p.maxConcurrency) - for _, img := range imageNames { - eg.Go(func() error { - _, err := p.client.ImageInspect(ctx, img) - if errdefs.IsNotFound(err) { - // err on the side of caution: only skip if we successfully - // queried the API and got back a definitive "not exists" - return nil - } - mu.Lock() - defer mu.Unlock() - ret = append(ret, img) + err := forEachWithLimiter(ctx, p.limiter, imageNames, func(ctx context.Context, img string) error { + _, err := p.client.ImageInspect(ctx, img) + if errdefs.IsNotFound(err) { + // err on the side of caution: only skip if we successfully + // queried the API and got back a definitive "not exists" return nil - }) - } - - if err := eg.Wait(); err != nil { + } + mu.Lock() + defer mu.Unlock() + ret = append(ret, img) + return nil + }) + if err != nil { return nil, err } diff --git a/pkg/compose/image_pruner_test.go b/pkg/compose/image_pruner_test.go index a4c5062e8..379a23259 100644 --- a/pkg/compose/image_pruner_test.go +++ b/pkg/compose/image_pruner_test.go @@ -101,7 +101,7 @@ func TestRemoveImages_BucketsStillInUseSeparatelyFromRemoved(t *testing.T) { apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:fails", client.ImageRemoveOptions{}). Return(client.ImageRemoveResult{}, errdefs.ErrPermissionDenied.WithMessage("permission denied")) - removed, stillInUse, err := svc.removeImages(t.Context(), images) + removed, stillInUse, err := svc.removeImages(t.Context(), images, nil) assert.DeepEqual(t, removed, []string{"sha256:removed"}) assert.DeepEqual(t, stillInUse, []string{"sha256:in-use"}) assert.ErrorContains(t, err, "sha256:fails") diff --git a/pkg/compose/restart.go b/pkg/compose/restart.go index ccbfc0685..d9a6cf8b3 100644 --- a/pkg/compose/restart.go +++ b/pkg/compose/restart.go @@ -60,7 +60,7 @@ func (s *composeService) restart(ctx context.Context, projectName string, option return err } - return forEachContainerWithLimiter(ctx, limiter, containers.filter(isService(service)), func(ctx context.Context, ctr container.Summary) error { + return forEachWithLimiter(ctx, limiter, containers.filter(isService(service)), func(ctx context.Context, ctr container.Summary) error { return s.restartContainer(ctx, project.Services[service], ctr, options) }) }) diff --git a/pkg/compose/up.go b/pkg/compose/up.go index 93a2e80d1..2c7998a46 100644 --- a/pkg/compose/up.go +++ b/pkg/compose/up.go @@ -32,9 +32,9 @@ import ( "github.com/containerd/errdefs" "github.com/docker/cli/cli" "github.com/eiannone/keyboard" - "github.com/moby/moby/client" "github.com/sirupsen/logrus" "golang.org/x/sync/errgroup" + "golang.org/x/sync/semaphore" "github.com/docker/compose/v5/cmd/formatter" "github.com/docker/compose/v5/internal/desktop" @@ -82,6 +82,10 @@ type upSession struct { menu *formatter.LogKeyboard globalCtx context.Context cancel context.CancelFunc + // logOpenLimiter bounds concurrent log-attach opens from + // followStartedContainers, the same footgun --parallel closes for + // `compose logs --follow` (see doLogContainer). + logOpenLimiter *semaphore.Weighted signalChan chan os.Signal isTerminated atomic.Bool @@ -136,6 +140,7 @@ func (s *composeService) runInteractiveUp(ctx context.Context, project *types.Pr menu: navigationMenu, globalCtx: globalCtx, cancel: cancel, + logOpenLimiter: newOptionalLimiter(s.maxConcurrency), signalChan: signalChan, } @@ -453,15 +458,15 @@ func (u *upSession) followStartedContainers(attached []string) api.ContainerEven } func (u *upSession) streamContainerLogs(event api.ContainerEvent, since string) error { - res, err := u.apiClient().ContainerInspect(u.globalCtx, event.ID, client.ContainerInspectOptions{}) + ctr, err := u.inspectWithSlot(u.globalCtx, u.logOpenLimiter, event.ID) if err != nil { return err } if since == "" { - since = logsSinceLastRun(res.Container) + since = logsSinceLastRun(ctr) } - err = u.doLogContainer(u.globalCtx, nil, u.options.Start.Attach, event.Source, res.Container, api.LogOptions{ + err = u.doLogContainer(u.globalCtx, u.logOpenLimiter, u.options.Start.Attach, event.Source, ctr, api.LogOptions{ Follow: true, Since: since, }) diff --git a/pkg/compose/up_test.go b/pkg/compose/up_test.go index 10902463b..23a5b0868 100644 --- a/pkg/compose/up_test.go +++ b/pkg/compose/up_test.go @@ -20,8 +20,15 @@ import ( "context" "errors" "fmt" + "io" + "strings" "testing" + "time" + containerType "github.com/moby/moby/api/types/container" + "github.com/moby/moby/client" + "go.uber.org/mock/gomock" + "golang.org/x/sync/errgroup" "gotest.tools/v3/assert" "github.com/docker/compose/v5/pkg/api" @@ -179,3 +186,45 @@ func TestAppendErrKeepsCancellationBeforeShutdown(t *testing.T) { u.appendErr(context.Canceled) assert.Equal(t, len(u.errs), 1) } + +// TestStreamContainerLogs_ConcurrencyIsBounded guards a regression flagged +// in review: streamContainerLogs (attaches logs for containers that start +// after up's initial attach, e.g. via depends_on/restart/watch rebuilds) +// used to call ContainerInspect directly and pass nil to doLogContainer, so +// unlike `compose logs --follow` it never shared --parallel's budget +// bounding concurrent log-open calls. +func TestStreamContainerLogs_ConcurrencyIsBounded(t *testing.T) { + svc, apiClient := newTestService(t, WithMaxConcurrency(1)) + + tracker := &peakConcurrencyTracker{} + for _, id := range []string{"c1", "c2"} { + apiClient.EXPECT().ContainerInspect(gomock.Any(), id, gomock.Any()). + DoAndReturn(func(context.Context, string, client.ContainerInspectOptions) (client.ContainerInspectResult, error) { + tracker.enter() + time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up + tracker.leave() + return client.ContainerInspectResult{ + Container: containerType.InspectResponse{ + ID: id, + Config: &containerType.Config{Tty: false}, + State: &containerType.State{}, + }, + }, nil + }) + apiClient.EXPECT().ContainerLogs(gomock.Any(), id, gomock.Any()). + Return(io.NopCloser(strings.NewReader("")), nil) + } + + u := &upSession{ + composeService: svc, + globalCtx: t.Context(), + logOpenLimiter: newOptionalLimiter(1), + options: api.UpOptions{Start: api.StartOptions{Attach: &testLogConsumer{}}}, + } + + var eg errgroup.Group + eg.Go(func() error { return u.streamContainerLogs(api.ContainerEvent{ID: "c1", Service: "web"}, "") }) + eg.Go(func() error { return u.streamContainerLogs(api.ContainerEvent{ID: "c2", Service: "web"}, "") }) + assert.NilError(t, eg.Wait()) + assert.Equal(t, tracker.Peak(), 1, "streamContainerLogs must share --parallel's budget for concurrent log-attach opens") +} diff --git a/pkg/compose/watch.go b/pkg/compose/watch.go index 50a73f67d..220d4ca29 100644 --- a/pkg/compose/watch.go +++ b/pkg/compose/watch.go @@ -129,7 +129,7 @@ func (s *composeService) getSyncImplementation(project *types.Project) (sync.Syn return nil, errors.New("no available sync implementation") } - return sync.NewTar(project.Name, tarDockerClient{s: s}), nil + return sync.NewTar(project.Name, tarDockerClient{s: s}, s.maxConcurrency), nil } func (s *composeService) Watch(ctx context.Context, project *types.Project, options api.WatchOptions) error {