diff --git a/daemon/daemon.go b/daemon/daemon.go index 20d824b359..619acb3223 100644 --- a/daemon/daemon.go +++ b/daemon/daemon.go @@ -233,17 +233,14 @@ func (daemon *Daemon) loadContainers(ctx context.Context) (map[string]map[string sem := semaphore.NewWeighted(int64(parallelLimit)) for _, v := range dir { - group.Add(1) - go func(id string) { - defer group.Done() - _ = sem.Acquire(context.Background(), 1) + id := v.Name() + group.Go(func() { + _ = sem.Acquire(context.WithoutCancel(ctx), 1) defer sem.Release(1) - logger := log.G(ctx).WithField("container", id) - c, err := daemon.load(id) if err != nil { - logger.WithError(err).Error("failed to load container") + log.G(ctx).WithFields(log.Fields{"error": err, "container": id}).Error("Failed to load container") return } @@ -256,7 +253,7 @@ func (daemon *Daemon) loadContainers(ctx context.Context) (map[string]map[string containers[c.ID] = c } mapLock.Unlock() - }(v.Name()) + }) } group.Wait() @@ -284,10 +281,11 @@ func (daemon *Daemon) restore(ctx context.Context, cfg *configStore, containers activeSandboxes := make(map[string]any) for _, c := range containers { - group.Add(1) - go func(c *container.Container) { - defer group.Done() - _ = sem.Acquire(context.Background(), 1) + group.Go(func() { + if err := sem.Acquire(context.WithoutCancel(ctx), 1); err != nil { + // ctx is done; should never happen. + return + } defer sem.Release(1) logger := log.G(ctx).WithField("container", c.ID) @@ -317,7 +315,7 @@ func (daemon *Daemon) restore(ctx context.Context, cfg *configStore, containers mapLock.Unlock() return } - }(c) + }) } group.Wait() diff --git a/daemon/internal/metrics/plugin_unix.go b/daemon/internal/metrics/plugin_unix.go index 89f6abfbd9..cd1c7cf1a9 100644 --- a/daemon/internal/metrics/plugin_unix.go +++ b/daemon/internal/metrics/plugin_unix.go @@ -79,14 +79,9 @@ func RegisterPlugin(store *plugin.Store, path string) error { // CleanupPlugin stops metrics collection for all plugins func CleanupPlugin(store plugingetter.PluginGetter) { - plugins := store.GetAllManagedPluginsByCap(pluginType) var wg sync.WaitGroup - wg.Add(len(plugins)) - - for _, p := range plugins { - go func() { - defer wg.Done() - + for _, p := range store.GetAllManagedPluginsByCap(pluginType) { + wg.Go(func() { adapter, err := makePluginAdapter(p) if err != nil { log.G(context.TODO()).WithFields(log.Fields{ @@ -101,7 +96,7 @@ func CleanupPlugin(store plugingetter.PluginGetter) { "plugin": p.Name(), }).Error("Error stopping plugin metrics collection") } - }() + }) } wg.Wait() diff --git a/daemon/internal/stream/streams.go b/daemon/internal/stream/streams.go index 1483b20eac..928774f759 100644 --- a/daemon/internal/stream/streams.go +++ b/daemon/internal/stream/streams.go @@ -126,9 +126,7 @@ func (c *Config) CopyToPipe(iop *cio.DirectIO) { c.dio = iop copyFunc := func(name string, w io.Writer, r io.ReadCloser) { - c.wg.Add(1) - go func() { - defer c.wg.Done() + c.wg.Go(func() { if _, err := pools.Copy(w, r); err != nil { if c.closed.Load() { return @@ -138,7 +136,7 @@ func (c *Config) CopyToPipe(iop *cio.DirectIO) { if err := r.Close(); err != nil && !c.closed.Load() { log.G(ctx).WithFields(log.Fields{"stream": name, "error": err}).Warn("close stream failed") } - }() + }) } if iop.Stdout != nil { diff --git a/daemon/libnetwork/osl/interface_linux_test.go b/daemon/libnetwork/osl/interface_linux_test.go index 051ddcc77a..d9781a4d64 100644 --- a/daemon/libnetwork/osl/interface_linux_test.go +++ b/daemon/libnetwork/osl/interface_linux_test.go @@ -72,12 +72,10 @@ func TestAddInterfaceInParallel(t *testing.T) { wg := sync.WaitGroup{} for i := range 10 { src := fmt.Sprintf("dummy%d", i) - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { err := ns.AddInterface(context.Background(), src, "eth", "", WithCreatedInContainer(true)) assert.NilError(t, err) - }() + }) } wg.Wait() diff --git a/integration-cli/docker_api_exec_resize_test.go b/integration-cli/docker_api_exec_resize_test.go index 1667c0b9cd..2e7b6fe773 100644 --- a/integration-cli/docker_api_exec_resize_test.go +++ b/integration-cli/docker_api_exec_resize_test.go @@ -93,13 +93,11 @@ func (s *DockerAPISuite) TestExecResizeImmediatelyAfterExecStart(c *testing.T) { wg sync.WaitGroup ) for range n { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { if err := testExecResize(); err != nil { ch <- err } - }() + }) } wg.Wait() diff --git a/integration-cli/docker_cli_exec_test.go b/integration-cli/docker_cli_exec_test.go index eb5d7d1244..a6389068c3 100644 --- a/integration-cli/docker_cli_exec_test.go +++ b/integration-cli/docker_cli_exec_test.go @@ -284,9 +284,7 @@ func (s *DockerCLIExecSuite) TestExecCgroup(c *testing.T) { errChan := make(chan error, 5) // exec a few times concurrently to get consistent failure for range 5 { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { out, _, err := dockerCmdWithError("exec", "testing", "cat", "/proc/self/cgroup") if err != nil { errChan <- err @@ -297,7 +295,7 @@ func (s *DockerCLIExecSuite) TestExecCgroup(c *testing.T) { mu.Lock() execCgroups = append(execCgroups, cg) mu.Unlock() - }() + }) } wg.Wait() close(errChan) diff --git a/integration/internal/container/container.go b/integration/internal/container/container.go index a82c9242d8..2408de9781 100644 --- a/integration/internal/container/container.go +++ b/integration/internal/container/container.go @@ -145,12 +145,10 @@ func demultiplexStreams(ctx context.Context, resp client.HijackedResponse) (stre outputDone := make(chan error, 1) var wg sync.WaitGroup - wg.Add(1) - go func() { + wg.Go(func() { _, err := stdcopy.StdCopy(&s.stdout, &s.stderr, resp.Reader) outputDone <- err - wg.Done() - }() + }) var err error select {