Merge pull request #52042 from thaJeztah/waitgroup_go

rewrite some code to use use WaitGroup.Go

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()
 

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()
 

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 {

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()
 

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()

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)

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 {