diff --git a/pkg/compose/compose.go b/pkg/compose/compose.go index 09596a933..6f4885f03 100644 --- a/pkg/compose/compose.go +++ b/pkg/compose/compose.go @@ -200,6 +200,27 @@ func releaseSlot(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 +// 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 { + eg, ctx := errgroup.WithContext(ctx) + for _, ctr := range containers { + eg.Go(func() error { + if err := acquireSlot(ctx, limiter); err != nil { + return err + } + defer releaseSlot(limiter) + return fn(ctx, ctr) + }) + } + return eg.Wait() +} + // WithDryRun configure Compose to run without actually applying changes func WithDryRun(s *composeService) error { s.dryRun = true diff --git a/pkg/compose/containers.go b/pkg/compose/containers.go index 208e0d0fa..7f94b14ce 100644 --- a/pkg/compose/containers.go +++ b/pkg/compose/containers.go @@ -209,7 +209,11 @@ func (containers Containers) filter(predicates ...containerPredicate) Containers return filtered } -// forEachContainerConcurrent runs fn for every container concurrently and waits for all goroutines. +// forEachContainerConcurrent runs fn for every container concurrently and +// waits for all goroutines. Use forEachContainerWithLimiter (compose.go) +// instead when the concurrency budget must be shared across several +// concurrently-dispatched calls, e.g. one per service visited by +// InDependencyOrder. func forEachContainerConcurrent(ctx context.Context, maxConcurrency int, containers Containers, fn func(context.Context, container.Summary) error) error { eg, ctx := newLimitedErrgroup(ctx, maxConcurrency) for _, ctr := range containers { diff --git a/pkg/compose/down.go b/pkg/compose/down.go index 451410073..26e127b82 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/semaphore" "github.com/docker/compose/v5/pkg/api" "github.com/docker/compose/v5/pkg/utils" @@ -96,8 +97,15 @@ func (s *composeService) down(ctx context.Context, projectName string, options a resourceToRemove = true } + // shared by every service so the dependency-order fan-out and the + // per-service container fan-out combined never exceed maxConcurrency + // concurrent container removals — a per-service bound alone allows as + // many independent services to run at once as the graph permits, each + // with its own maxConcurrency budget (same fix as restart.go) + limiter := newOptionalLimiter(s.maxConcurrency) + err = InReverseDependencyOrder(ctx, project, func(c context.Context, service string) error { - return s.downService(ctx, project, containers, options, service) + return s.downService(ctx, project, containers, options, service, limiter) }, WithRootNodesAndDown(options.Services)) if err != nil { return err @@ -105,7 +113,7 @@ func (s *composeService) down(ctx context.Context, projectName string, options a orphans := containers.filter(isOrphaned(project)) if options.RemoveOrphans && len(orphans) > 0 { - err := s.removeContainers(ctx, orphans, nil, options.Timeout, false) + err := s.removeContainers(ctx, orphans, nil, options.Timeout, false, limiter) if err != nil { return err } @@ -424,24 +432,30 @@ func (s *composeService) stopContainer(ctx context.Context, service *types.Servi return nil } -func (s *composeService) stopContainers(ctx context.Context, serv *types.ServiceConfig, containers []containerType.Summary, timeout *time.Duration, listener api.ContainerEventListener) error { - eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) - for _, ctr := range containers { - eg.Go(func() error { - return s.stopContainer(ctx, serv, ctr, timeout, listener) - }) - } - return eg.Wait() +func (s *composeService) stopContainers( + ctx context.Context, + serv *types.ServiceConfig, + containers []containerType.Summary, + timeout *time.Duration, + listener api.ContainerEventListener, + limiter *semaphore.Weighted, +) error { + return forEachContainerWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error { + return s.stopContainer(ctx, serv, ctr, timeout, listener) + }) } -func (s *composeService) removeContainers(ctx context.Context, containers []containerType.Summary, service *types.ServiceConfig, timeout *time.Duration, volumes bool) error { - eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) - for _, ctr := range containers { - eg.Go(func() error { - return s.stopAndRemoveContainer(ctx, ctr, service, timeout, volumes) - }) - } - return eg.Wait() +func (s *composeService) removeContainers( + ctx context.Context, + containers []containerType.Summary, + service *types.ServiceConfig, + timeout *time.Duration, + volumes bool, + limiter *semaphore.Weighted, +) error { + return forEachContainerWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error { + return s.stopAndRemoveContainer(ctx, ctr, service, timeout, volumes) + }) } func (s *composeService) stopAndRemoveContainer(ctx context.Context, ctr containerType.Summary, service *types.ServiceConfig, timeout *time.Duration, volumes bool) error { @@ -471,10 +485,10 @@ func (s *composeService) stopAndRemoveContainer(ctx context.Context, ctr contain // own project containers — the relay deployed when it published endpoints — // and the plugin only removes the provider's own resource, so the containers // go first, mirroring up, which provisions the resource before the relay. -func (s *composeService) downService(ctx context.Context, project *types.Project, containers Containers, options api.DownOptions, service string) error { +func (s *composeService) downService(ctx context.Context, project *types.Project, containers Containers, options api.DownOptions, service string, limiter *semaphore.Weighted) error { serv := project.Services[service] serviceContainers := containers.filter(isService(service)) - if err := s.removeContainers(ctx, serviceContainers, &serv, options.Timeout, options.Volumes); err != nil { + if err := s.removeContainers(ctx, serviceContainers, &serv, options.Timeout, options.Volumes, limiter); err != nil { return err } if serv.Provider != nil { diff --git a/pkg/compose/down_test.go b/pkg/compose/down_test.go index 24031539a..7c179231c 100644 --- a/pkg/compose/down_test.go +++ b/pkg/compose/down_test.go @@ -23,6 +23,7 @@ import ( "os" "strings" "testing" + "time" "github.com/compose-spec/compose-go/v2/types" "github.com/containerd/errdefs" @@ -99,6 +100,41 @@ func TestDown(t *testing.T) { assert.NilError(t, err) } +// TestDown_ConcurrencyIsBoundedAcrossServices guards against the same +// per-service budget leak fixed on restart.go and stop.go: +// InReverseDependencyOrder dispatches independent services concurrently, so +// a limiter created fresh inside removeContainers for each service visit +// would let each one spend its own maxConcurrency budget at the same time. +func TestDown_ConcurrencyIsBoundedAcrossServices(t *testing.T) { + svc, apiClient := newTestService(t, WithMaxConcurrency(1)) + + const numServices = 4 + project, containers := nIndependentServiceContainers(numServices) + + apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()). + Return(client.ContainerListResult{Items: containers}, nil) + apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()). + Return(client.ContainerListResult{}, nil) // removePreStartHookContainers lookup + + apiClient.EXPECT().ContainerStop(gomock.Any(), gomock.Any(), gomock.Any()). + Return(client.ContainerStopResult{}, nil). + Times(numServices) + + tracker := &peakConcurrencyTracker{} + apiClient.EXPECT().ContainerRemove(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(context.Context, string, client.ContainerRemoveOptions) (client.ContainerRemoveResult, error) { + tracker.enter() + time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up + tracker.leave() + return client.ContainerRemoveResult{}, nil + }). + Times(numServices) + + err := svc.down(t.Context(), "prj", compose.DownOptions{Project: project}) + assert.NilError(t, err) + assert.Equal(t, tracker.Peak(), 1, "down must never run more than maxConcurrency ContainerRemove calls at once") +} + func TestDownWithGivenServices(t *testing.T) { mockCtrl := gomock.NewController(t) defer mockCtrl.Finish() diff --git a/pkg/compose/executor_test.go b/pkg/compose/executor_test.go index 63b9ed64e..1440ee796 100644 --- a/pkg/compose/executor_test.go +++ b/pkg/compose/executor_test.go @@ -19,7 +19,9 @@ package compose import ( "context" "errors" + "fmt" "strconv" + "sync" "testing" "time" @@ -42,6 +44,51 @@ func (noopEventProcessor) Start(_ context.Context, _ string) {} func (noopEventProcessor) On(_ ...api.Resource) {} func (noopEventProcessor) Done(_ string, _ bool) {} +// peakConcurrencyTracker records the highest number of overlapping +// enter()/leave() pairs seen, used to assert a --parallel bound was honored. +type peakConcurrencyTracker struct { + mu sync.Mutex + current int + peak int +} + +func (t *peakConcurrencyTracker) enter() { + t.mu.Lock() + t.current++ + if t.current > t.peak { + t.peak = t.current + } + t.mu.Unlock() +} + +func (t *peakConcurrencyTracker) leave() { + t.mu.Lock() + t.current-- + t.mu.Unlock() +} + +func (t *peakConcurrencyTracker) Peak() int { + t.mu.Lock() + defer t.mu.Unlock() + return t.peak +} + +// nIndependentServiceContainers builds n services with no depends_on between +// them — so InDependencyOrder/InReverseDependencyOrder dispatches them +// concurrently — plus one non-oneoff container per service, used to assert a +// limiter shared across service visits (see TestDown/TestStop +// ConcurrencyIsBoundedAcrossServices). +func nIndependentServiceContainers(n int) (*types.Project, []container.Summary) { + project := &types.Project{Name: "prj", Services: types.Services{}} + var containers []container.Summary + for i := range n { + name := fmt.Sprintf("svc%d", i) + project.Services[name] = types.ServiceConfig{Name: name} + containers = append(containers, testContainer(name, fmt.Sprintf("c%d", i), false)) + } + return project, containers +} + func newTestService(t *testing.T, opts ...Option) (*composeService, *mocks.MockAPIClient) { t.Helper() mockCtrl := gomock.NewController(t) diff --git a/pkg/compose/logs.go b/pkg/compose/logs.go index 0fc281716..e97143a27 100644 --- a/pkg/compose/logs.go +++ b/pkg/compose/logs.go @@ -107,18 +107,29 @@ func (s *composeService) selectLogsContainers(ctx context.Context, projectName s return containers, nil } +// inspectWithSlot acquires limiter's slot, then inspects the container, +// releasing the slot on error since no caller reaches doLogContainer (which +// owns the slot from here on) in that case. +func (s *composeService) inspectWithSlot(ctx context.Context, limiter *semaphore.Weighted, id string) (container.InspectResponse, error) { + if err := acquireSlot(ctx, limiter); err != nil { + return container.InspectResponse{}, err + } + res, err := s.apiClient().ContainerInspect(ctx, id, client.ContainerInspectOptions{}) + if err != nil { + releaseSlot(limiter) + return container.InspectResponse{}, err + } + return res.Container, nil +} + // logContainer streams a container's logs, warning when its logging driver // doesn't support reading logs func (s *composeService) logContainer(ctx context.Context, limiter *semaphore.Weighted, consumer api.LogConsumer, ctr container.Summary, options api.LogOptions) error { - if err := acquireSlot(ctx, limiter); err != nil { - return err - } - res, err := s.apiClient().ContainerInspect(ctx, ctr.ID, client.ContainerInspectOptions{}) + res, err := s.inspectWithSlot(ctx, limiter, ctr.ID) if err != nil { - releaseSlot(limiter) return err } - err = s.doLogContainer(ctx, limiter, consumer, getContainerNameWithoutProject(ctr), res.Container, options) + err = s.doLogContainer(ctx, limiter, consumer, getContainerNameWithoutProject(ctr), res, options) if errdefs.IsNotImplemented(err) { logrus.Warnf("Can't retrieve logs for %q: %s", getCanonicalContainerName(ctr), err.Error()) return nil @@ -148,19 +159,15 @@ func (s *composeService) followStartedContainersLogs( // would drop the whole run. since := runEnds.Since(event.ID) eg.Go(func() error { - if err := acquireSlot(ctx, limiter); err != nil { - return err - } - res, err := s.apiClient().ContainerInspect(ctx, event.ID, client.ContainerInspectOptions{}) + res, err := s.inspectWithSlot(ctx, limiter, event.ID) if err != nil { - releaseSlot(limiter) return err } if since == "" { - since = logsSinceLastRun(res.Container) + since = logsSinceLastRun(res) } - err = s.doLogContainer(ctx, limiter, consumer, event.Source, res.Container, api.LogOptions{ + err = s.doLogContainer(ctx, limiter, consumer, event.Source, res, api.LogOptions{ Follow: options.Follow, Since: since, Until: options.Until, diff --git a/pkg/compose/logs_test.go b/pkg/compose/logs_test.go index b4b4fb5c4..becc15688 100644 --- a/pkg/compose/logs_test.go +++ b/pkg/compose/logs_test.go @@ -344,27 +344,14 @@ func TestComposeService_Logs_FollowLimitsConcurrentStreamOpens(t *testing.T) { }, nil) } - var ( - mu sync.Mutex - current int - peak int - ) + tracker := &peakConcurrencyTracker{} writers := make(chan *io.PipeWriter, len(ids)) for _, id := range ids { api.EXPECT().ContainerLogs(anyCancellableContext(), id, gomock.Any()). DoAndReturn(func(context.Context, string, client.ContainerLogsOptions) (io.ReadCloser, error) { - mu.Lock() - current++ - if current > peak { - peak = current - } - mu.Unlock() - + tracker.enter() time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up - - mu.Lock() - current-- - mu.Unlock() + tracker.leave() r, w := io.Pipe() writers <- w @@ -403,7 +390,7 @@ func TestComposeService_Logs_FollowLimitsConcurrentStreamOpens(t *testing.T) { _ = w.Close() } assert.NilError(t, <-done) - assert.Equal(t, peak, 1, "opening follow-mode log streams must be bounded by maxConcurrency") + assert.Equal(t, tracker.Peak(), 1, "opening follow-mode log streams must be bounded by maxConcurrency") } type testLogConsumer struct { diff --git a/pkg/compose/restart.go b/pkg/compose/restart.go index 5373d06af..ccbfc0685 100644 --- a/pkg/compose/restart.go +++ b/pkg/compose/restart.go @@ -23,7 +23,6 @@ import ( "github.com/compose-spec/compose-go/v2/types" "github.com/moby/moby/api/types/container" "github.com/moby/moby/client" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" "github.com/docker/compose/v5/pkg/utils" @@ -61,17 +60,9 @@ func (s *composeService) restart(ctx context.Context, projectName string, option return err } - eg, ctx := errgroup.WithContext(ctx) - for _, ctr := range containers.filter(isService(service)) { - eg.Go(func() error { - if err := acquireSlot(ctx, limiter); err != nil { - return err - } - defer releaseSlot(limiter) - return s.restartContainer(ctx, project.Services[service], ctr, options) - }) - } - return eg.Wait() + return forEachContainerWithLimiter(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/restart_test.go b/pkg/compose/restart_test.go index e7904d1fb..5469049ca 100644 --- a/pkg/compose/restart_test.go +++ b/pkg/compose/restart_test.go @@ -22,7 +22,6 @@ import ( "context" "fmt" "net" - "sync" "testing" "time" @@ -33,7 +32,6 @@ import ( "gotest.tools/v3/assert" "github.com/docker/compose/v5/pkg/api" - "github.com/docker/compose/v5/pkg/mocks" ) // These tests characterize the restart path before the lifecycle engines @@ -182,19 +180,11 @@ func TestRestartContainerSkipsHooksForRelay(t *testing.T) { // spawning up to maxConcurrency ContainerRestart calls — up to N times the // documented --parallel bound. The limiter must be shared across services. func TestRestart_ConcurrencyIsBoundedAcrossServices(t *testing.T) { - mockCtrl := gomock.NewController(t) - t.Cleanup(mockCtrl.Finish) - apiClient := mocks.NewMockAPIClient(mockCtrl) - cli := mocks.NewMockCli(mockCtrl) - cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, apiClient := newTestService(t, WithMaxConcurrency(1)) apiClient.EXPECT().Ping(gomock.Any(), client.PingOptions{NegotiateAPIVersion: true}). Return(client.PingResult{APIVersion: "1.44"}, nil).AnyTimes() apiClient.EXPECT().ClientVersion().Return("1.44").AnyTimes() - svcIface, err := NewComposeService(cli, WithMaxConcurrency(1)) - assert.NilError(t, err) - svc := svcIface.(*composeService) - const numServices = 4 project := &types.Project{Name: "prj", Services: types.Services{}} var containers Containers @@ -207,30 +197,17 @@ func TestRestart_ConcurrencyIsBoundedAcrossServices(t *testing.T) { apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()). Return(client.ContainerListResult{Items: containers}, nil) - var ( - mu sync.Mutex - current int - peak int - ) + tracker := &peakConcurrencyTracker{} apiClient.EXPECT().ContainerRestart(gomock.Any(), gomock.Any(), gomock.Any()). DoAndReturn(func(context.Context, string, client.ContainerRestartOptions) (client.ContainerRestartResult, error) { - mu.Lock() - current++ - if current > peak { - peak = current - } - mu.Unlock() - + tracker.enter() time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up - - mu.Lock() - current-- - mu.Unlock() + tracker.leave() return client.ContainerRestartResult{}, nil }). Times(numServices) - err = svc.restart(t.Context(), "prj", api.RestartOptions{Project: project}) + err := svc.restart(t.Context(), "prj", api.RestartOptions{Project: project}) assert.NilError(t, err) - assert.Equal(t, peak, 1, "restart must never run more than maxConcurrency ContainerRestart calls at once") + assert.Equal(t, tracker.Peak(), 1, "restart must never run more than maxConcurrency ContainerRestart calls at once") } diff --git a/pkg/compose/start.go b/pkg/compose/start.go index de5ac98f8..64f50ee62 100644 --- a/pkg/compose/start.go +++ b/pkg/compose/start.go @@ -61,6 +61,10 @@ func (s *composeService) start(ctx context.Context, projectName string, options return err } + // the node-level bound is a correct proxy for --parallel here, unlike + // restart/down/stop: startService starts a service's containers in a + // plain sequential loop, so one engine-call burst per node is all this + // traversal ever dispatches concurrently. err = InDependencyOrder(ctx, project, func(c context.Context, name string) error { service, err := project.GetService(name) if err != nil { @@ -68,6 +72,8 @@ func (s *composeService) start(ctx context.Context, projectName string, options } return s.startService(ctx, project, service, containers, listener, options.WaitTimeout) + }, func(traversal *graphTraversal) { + traversal.maxConcurrency = s.maxConcurrency }) if err != nil { return err diff --git a/pkg/compose/start_test.go b/pkg/compose/start_test.go index 0542148ea..c7e6aec4a 100644 --- a/pkg/compose/start_test.go +++ b/pkg/compose/start_test.go @@ -20,10 +20,12 @@ package compose import ( "context" + "fmt" "net" "strconv" "sync" "testing" + "time" "github.com/compose-spec/compose-go/v2/types" "github.com/docker/cli/cli/config/configfile" @@ -117,6 +119,53 @@ func serviceContainer(service string, num int, state container.ContainerState) c } } +// TestStart_ConcurrencyIsBoundedAcrossServices guards against the same +// per-service budget leak fixed on restart.go/down.go/stop.go: +// InDependencyOrder dispatches independent services concurrently, and +// startService's own per-container loop is sequential, so a missing +// node-level maxConcurrency option left the dispatch itself unbounded. +// ContainerStart is serialized process-wide by startMx regardless of this +// bound, so the injected-secret copy (which isn't) is used as the observable +// signal instead. +func TestStart_ConcurrencyIsBoundedAcrossServices(t *testing.T) { + svc, apiClient := newTestService(t, WithMaxConcurrency(1)) + + const numServices = 4 + project := &types.Project{Name: "prj", Services: types.Services{}, Secrets: types.Secrets{}} + var containers []container.Summary + for i := range numServices { + name := fmt.Sprintf("svc%d", i) + secretName := fmt.Sprintf("secret%d", i) + project.Secrets[secretName] = types.SecretConfig{Name: secretName, Content: "shh"} + project.Services[name] = types.ServiceConfig{ + Name: name, + ContainerSpec: types.ContainerSpec{Secrets: []types.ServiceSecretConfig{{Source: secretName}}}, + } + containers = append(containers, testContainer(name, fmt.Sprintf("c%d", i), false)) + } + + apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()). + Return(client.ContainerListResult{Items: containers}, nil) + + tracker := &peakConcurrencyTracker{} + apiClient.EXPECT().CopyToContainer(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(context.Context, string, client.CopyToContainerOptions) (client.CopyToContainerResult, error) { + tracker.enter() + time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up + tracker.leave() + return client.CopyToContainerResult{}, nil + }). + Times(numServices) + + apiClient.EXPECT().ContainerStart(gomock.Any(), gomock.Any(), gomock.Any()). + Return(client.ContainerStartResult{}, nil). + Times(numServices) + + err := svc.start(t.Context(), "prj", api.StartOptions{Project: project}, nil) + assert.NilError(t, err) + assert.Equal(t, tracker.Peak(), 1, "start must never dispatch more than maxConcurrency services concurrently") +} + func TestStartService_AlreadyRunningIsSilent(t *testing.T) { svc, _, rec := newStartTestService(t) diff --git a/pkg/compose/stop.go b/pkg/compose/stop.go index ea359401c..66e52ab39 100644 --- a/pkg/compose/stop.go +++ b/pkg/compose/stop.go @@ -48,6 +48,13 @@ func (s *composeService) stop(ctx context.Context, projectName string, options a options.Services = project.ServiceNames() } + // shared by every service so the dependency-order fan-out and the + // per-service container fan-out combined never exceed maxConcurrency + // concurrent container stops — a per-service bound alone allows as many + // independent services to run at once as the graph permits, each with + // its own maxConcurrency budget (same fix as restart.go) + limiter := newOptionalLimiter(s.maxConcurrency) + return InReverseDependencyOrder(ctx, project, func(c context.Context, service string) error { if !slices.Contains(options.Services, service) { return nil @@ -57,7 +64,7 @@ func (s *composeService) stop(ctx context.Context, projectName string, options a // deployed when it published endpoints — and the plugin's own stop // hook (if any) only concerns the provider's resource, so the // container is stopped the same way as for any other service. - if err := s.stopContainers(ctx, &serv, containers.filter(isService(service)).filter(isNotOneOff), options.Timeout, event); err != nil { + if err := s.stopContainers(ctx, &serv, containers.filter(isService(service)).filter(isNotOneOff), options.Timeout, event, limiter); err != nil { return err } if serv.Provider != nil { diff --git a/pkg/compose/stop_test.go b/pkg/compose/stop_test.go index c191e66ac..3333277cf 100644 --- a/pkg/compose/stop_test.go +++ b/pkg/compose/stop_test.go @@ -17,6 +17,7 @@ package compose import ( + "context" "strings" "testing" "time" @@ -66,3 +67,32 @@ func TestStopTimeout(t *testing.T) { }) assert.NilError(t, err) } + +// TestStop_ConcurrencyIsBoundedAcrossServices guards against the same +// per-service budget leak fixed on restart.go and down.go: InReverseDependencyOrder +// dispatches independent services concurrently, so a limiter created fresh +// inside stopContainers for each service visit would let each one spend its +// own maxConcurrency budget at the same time. +func TestStop_ConcurrencyIsBoundedAcrossServices(t *testing.T) { + svc, apiClient := newTestService(t, WithMaxConcurrency(1)) + + const numServices = 4 + project, containers := nIndependentServiceContainers(numServices) + + apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()). + Return(client.ContainerListResult{Items: containers}, nil) + + tracker := &peakConcurrencyTracker{} + apiClient.EXPECT().ContainerStop(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(context.Context, string, client.ContainerStopOptions) (client.ContainerStopResult, error) { + tracker.enter() + time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up + tracker.leave() + return client.ContainerStopResult{}, nil + }). + Times(numServices) + + err := svc.stop(t.Context(), "prj", compose.StopOptions{Project: project}, nil) + assert.NilError(t, err) + assert.Equal(t, tracker.Peak(), 1, "stop must never run more than maxConcurrency ContainerStop calls at once") +} diff --git a/pkg/compose/wait.go b/pkg/compose/wait.go index 9fb76d815..374febc24 100644 --- a/pkg/compose/wait.go +++ b/pkg/compose/wait.go @@ -22,6 +22,7 @@ import ( "fmt" "github.com/moby/moby/client" + "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" ) @@ -54,7 +55,10 @@ func (s *composeService) Wait(ctx context.Context, projectName string, options a return 0, fmt.Errorf("no containers for project %q", projectName) } - eg, waitCtx := newLimitedErrgroup(ctx, s.maxConcurrency) + // ContainerWait blocks until the container exits, so it must not be + // bound by --parallel: capping concurrency here would serialize waits + // that are meant to run together (same rationale as waitDependencies). + eg, waitCtx := errgroup.WithContext(ctx) var statusCode int64 for _, ctr := range containers { eg.Go(func() error { diff --git a/pkg/compose/watch.go b/pkg/compose/watch.go index dc62850f5..50a73f67d 100644 --- a/pkg/compose/watch.go +++ b/pkg/compose/watch.go @@ -197,7 +197,9 @@ func (s *composeService) watch(ctx context.Context, project *types.Project, opti if err != nil { return nil, err } - eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) + // exactly one goroutine is ever submitted to eg (below); the real + // --parallel bound for watch is handleWatchBatch's per-batch fan-out. + eg, ctx := errgroup.WithContext(ctx) var ( rules []watchRule