Merge pull request #40595 from sfzhu93/GL_2_25

[integration] add buffer to prevent goroutine leak

integration-cli/docker_api_attach_test.go

@@ -43,14 +43,14 @@ func (s *DockerSuite) TestGetContainersAttachWebsocket(c *testing.T) {
 	expected := []byte("hello")
 	actual := make([]byte, len(expected))
 
-	outChan := make(chan error)
+	outChan := make(chan error, 1)
 	go func() {
 		_, err := io.ReadFull(ws, actual)
 		outChan <- err
 		close(outChan)
 	}()
 
-	inChan := make(chan error)
+	inChan := make(chan error, 1)
 	go func() {
 		_, err := ws.Write(expected)
 		inChan <- err
@@ -279,7 +279,7 @@ func bodyIsWritable(r *http.Response) bool {
 
 // readTimeout read from io.Reader with timeout
 func readTimeout(r io.Reader, buf []byte, timeout time.Duration) (n int, err error) {
-	ch := make(chan bool)
+	ch := make(chan bool, 1)
 	go func() {
 		n, err = io.ReadFull(r, buf)
 		ch <- true

integration-cli/docker_api_containers_test.go

@@ -338,7 +338,7 @@ func (s *DockerSuite) TestGetStoppedContainerStats(c *testing.T) {
 	name := "statscontainer"
 	dockerCmd(c, "create", "--name", name, "busybox", "ps")
 
-	chResp := make(chan error)
+	chResp := make(chan error, 1)
 
 	// We expect an immediate response, but if it's not immediate, the test would hang, so put it in a goroutine
 	// below we'll check this on a timeout.

integration-cli/docker_api_logs_test.go

@@ -30,7 +30,7 @@ func (s *DockerSuite) TestLogsAPIWithStdout(c *testing.T) {
 		err error
 	}
 
-	chLog := make(chan logOut)
+	chLog := make(chan logOut, 1)
 	res, body, err := request.Get(fmt.Sprintf("/containers/%s/logs?follow=1&stdout=1&timestamps=1", id))
 	assert.NilError(c, err)
 	assert.Equal(c, res.StatusCode, http.StatusOK)
@@ -116,6 +116,8 @@ func (s *DockerSuite) TestLogsAPIUntilFutureFollow(c *testing.T) {
 	}
 
 	chLog := make(chan logOut)
+	stop := make(chan struct{})
+	defer close(stop)
 
 	go func() {
 		bufReader := bufio.NewReader(reader)
@@ -126,11 +128,20 @@ func (s *DockerSuite) TestLogsAPIUntilFutureFollow(c *testing.T) {
 				if err == io.EOF {
 					return
 				}
-				chLog <- logOut{"", err}
+				select {
+				case <-stop:
+					return
+				case chLog <- logOut{"", err}:
+				}
+
 				return
 			}
 
-			chLog <- logOut{strings.TrimSpace(string(out)), err}
+			select {
+			case <-stop:
+				return
+			case chLog <- logOut{strings.TrimSpace(string(out)), err}:
+			}
 		}
 	}()
 

integration-cli/docker_cli_attach_test.go

@@ -102,7 +102,7 @@ func (s *DockerSuite) TestAttachTTYWithoutStdin(c *testing.T) {
 	id := strings.TrimSpace(out)
 	assert.NilError(c, waitRun(id))
 
-	done := make(chan error)
+	done := make(chan error, 1)
 	go func() {
 		defer close(done)
 

integration-cli/docker_cli_attach_unix_test.go

@@ -33,7 +33,7 @@ func (s *DockerSuite) TestAttachClosedOnContainerStop(c *testing.T) {
 	err = attachCmd.Start()
 	assert.NilError(c, err)
 
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 	go func() {
 		time.Sleep(300 * time.Millisecond)
 		defer close(errChan)
@@ -68,7 +68,7 @@ func (s *DockerSuite) TestAttachAfterDetach(c *testing.T) {
 	cmd.Stdout = tty
 	cmd.Stderr = tty
 
-	cmdExit := make(chan error)
+	cmdExit := make(chan error, 1)
 	go func() {
 		cmdExit <- cmd.Run()
 		close(cmdExit)

integration-cli/docker_cli_build_test.go

@@ -490,7 +490,7 @@ func (s *DockerSuite) TestBuildAddSingleFileToWorkdir(c *testing.T) {
 		}))
 	defer ctx.Close()
 
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 	go func() {
 		errChan <- buildImage(name, build.WithExternalBuildContext(ctx)).Error
 		close(errChan)
@@ -833,7 +833,7 @@ COPY test_file .`),
 		}))
 	defer ctx.Close()
 
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 	go func() {
 		errChan <- buildImage(name, build.WithExternalBuildContext(ctx)).Error
 		close(errChan)

integration-cli/docker_cli_daemon_test.go

@@ -1246,7 +1246,7 @@ func (s *DockerDaemonSuite) TestDaemonRestartKillWait(c *testing.T) {
 
 	s.d.Restart(c)
 
-	errchan := make(chan error)
+	errchan := make(chan error, 1)
 	go func() {
 		if out, err := s.d.Cmd("wait", containerID); err != nil {
 			errchan <- fmt.Errorf("%v:\n%s", err, out)
@@ -1599,15 +1599,17 @@ func (s *DockerDaemonSuite) TestDaemonRestartWithPausedContainer(c *testing.T) {
 	}
 	s.d.Restart(c)
 
-	errchan := make(chan error)
+	errchan := make(chan error, 1)
 	go func() {
 		out, err := s.d.Cmd("start", "test")
 		if err != nil {
 			errchan <- fmt.Errorf("%v:\n%s", err, out)
+			return
 		}
 		name := strings.TrimSpace(out)
 		if name != "test" {
 			errchan <- fmt.Errorf("Paused container start error on docker daemon restart, expected 'test' but got '%s'", name)
+			return
 		}
 		close(errchan)
 	}()

integration-cli/docker_cli_events_unix_test.go

@@ -51,7 +51,7 @@ func (s *DockerSuite) TestEventsRedirectStdout(c *testing.T) {
 func (s *DockerSuite) TestEventsOOMDisableFalse(c *testing.T) {
 	testRequires(c, DaemonIsLinux, oomControl, memoryLimitSupport, swapMemorySupport, NotPpc64le)
 
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 	go func() {
 		defer close(errChan)
 		out, exitCode, _ := dockerCmdWithError("run", "--name", "oomFalse", "-m", "10MB", "busybox", "sh", "-c", "x=a; while true; do x=$x$x$x$x; done")
@@ -81,7 +81,7 @@ func (s *DockerSuite) TestEventsOOMDisableFalse(c *testing.T) {
 func (s *DockerSuite) TestEventsOOMDisableTrue(c *testing.T) {
 	testRequires(c, DaemonIsLinux, oomControl, memoryLimitSupport, NotArm, swapMemorySupport, NotPpc64le)
 
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 	observer, err := newEventObserver(c)
 	assert.NilError(c, err)
 	err = observer.Start()

integration-cli/docker_cli_exec_test.go

@@ -53,7 +53,7 @@ func (s *DockerSuite) TestExecInteractive(c *testing.T) {
 	assert.Equal(c, line, "test")
 	err = stdin.Close()
 	assert.NilError(c, err)
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 	go func() {
 		errChan <- execCmd.Wait()
 		close(errChan)
@@ -170,7 +170,7 @@ func (s *DockerSuite) TestExecTTYWithoutStdin(c *testing.T) {
 	id := strings.TrimSpace(out)
 	assert.NilError(c, waitRun(id))
 
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 	go func() {
 		defer close(errChan)
 
@@ -230,7 +230,7 @@ func (s *DockerSuite) TestExecStopNotHanging(c *testing.T) {
 		out string
 		err error
 	}
-	ch := make(chan dstop)
+	ch := make(chan dstop, 1)
 	go func() {
 		result := icmd.RunCommand(dockerBinary, "stop", "testing")
 		ch <- dstop{result.Combined(), result.Error}
@@ -256,11 +256,12 @@ func (s *DockerSuite) TestExecCgroup(c *testing.T) {
 	var wg sync.WaitGroup
 	var mu sync.Mutex
 	var execCgroups []sort.StringSlice
-	errChan := make(chan error)
+	errChan := make(chan error, 5)
 	// exec a few times concurrently to get consistent failure
 	for i := 0; i < 5; i++ {
 		wg.Add(1)
 		go func() {
+			defer wg.Done()
 			out, _, err := dockerCmdWithError("exec", "testing", "cat", "/proc/self/cgroup")
 			if err != nil {
 				errChan <- err
@@ -271,7 +272,6 @@ func (s *DockerSuite) TestExecCgroup(c *testing.T) {
 			mu.Lock()
 			execCgroups = append(execCgroups, cg)
 			mu.Unlock()
-			wg.Done()
 		}()
 	}
 	wg.Wait()

integration-cli/docker_cli_exec_unix_test.go

@@ -26,7 +26,7 @@ func (s *DockerSuite) TestExecInteractiveStdinClose(c *testing.T) {
 
 	b := bytes.NewBuffer(nil)
 
-	ch := make(chan error)
+	ch := make(chan error, 1)
 	go func() { ch <- cmd.Wait() }()
 
 	select {
@@ -56,7 +56,7 @@ func (s *DockerSuite) TestExecTTY(c *testing.T) {
 	_, err = p.Write([]byte("cat /foo && exit\n"))
 	assert.NilError(c, err)
 
-	chErr := make(chan error)
+	chErr := make(chan error, 1)
 	go func() {
 		chErr <- cmd.Wait()
 	}()

integration-cli/docker_cli_external_volume_driver_test.go

@@ -365,7 +365,7 @@ func (s *DockerExternalVolumeSuite) TestExternalVolumeDriverLookupNotBlocked(c *
 	defer os.RemoveAll(specPath)
 
 	chCmd1 := make(chan struct{})
-	chCmd2 := make(chan error)
+	chCmd2 := make(chan error, 1)
 	cmd1 := exec.Command(dockerBinary, "volume", "create", "-d", "down-driver")
 	cmd2 := exec.Command(dockerBinary, "volume", "create")
 
@@ -398,7 +398,7 @@ func (s *DockerExternalVolumeSuite) TestExternalVolumeDriverRetryNotImmediatelyE
 	s.d.StartWithBusybox(c)
 	driverName := "test-external-volume-driver-retry"
 
-	errchan := make(chan error)
+	errchan := make(chan error, 1)
 	started := make(chan struct{})
 	go func() {
 		close(started)

integration-cli/docker_cli_logs_test.go

@@ -128,7 +128,7 @@ func (s *DockerSuite) TestLogsFollowStopped(c *testing.T) {
 	logsCmd := exec.Command(dockerBinary, "logs", "-f", id)
 	assert.NilError(c, logsCmd.Start())
 
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 	go func() {
 		errChan <- logsCmd.Wait()
 		close(errChan)

integration-cli/docker_cli_pull_local_test.go

@@ -82,8 +82,8 @@ func testConcurrentPullWholeRepo(c *testing.T) {
 	dockerCmd(c, args...)
 
 	// Run multiple re-pulls concurrently
-	results := make(chan error)
 	numPulls := 3
+	results := make(chan error, numPulls)
 
 	for i := 0; i != numPulls; i++ {
 		go func() {
@@ -120,8 +120,8 @@ func testConcurrentFailingPull(c *testing.T) {
 	repoName := fmt.Sprintf("%v/dockercli/busybox", privateRegistryURL)
 
 	// Run multiple pulls concurrently
-	results := make(chan error)
 	numPulls := 3
+	results := make(chan error, numPulls)
 
 	for i := 0; i != numPulls; i++ {
 		go func() {
@@ -170,7 +170,7 @@ func testConcurrentPullMultipleTags(c *testing.T) {
 	dockerCmd(c, args...)
 
 	// Re-pull individual tags, in parallel
-	results := make(chan error)
+	results := make(chan error, len(repos))
 
 	for _, repo := range repos {
 		go func(repo string) {

integration-cli/docker_cli_push_test.go

@@ -162,7 +162,7 @@ func testConcurrentPush(c *testing.T) {
 	}
 
 	// Push tags, in parallel
-	results := make(chan error)
+	results := make(chan error, len(repos))
 
 	for _, repo := range repos {
 		go func(repo string) {

integration-cli/docker_cli_run_test.go

@@ -1765,7 +1765,7 @@ func (s *DockerSuite) TestRunExitOnStdinClose(c *testing.T) {
 	if err := stdin.Close(); err != nil {
 		c.Fatal(err)
 	}
-	finish := make(chan error)
+	finish := make(chan error, 1)
 	go func() {
 		finish <- runCmd.Wait()
 		close(finish)
@@ -2523,7 +2523,7 @@ func (s *DockerSuite) TestRunPortFromDockerRangeInUse(c *testing.T) {
 }
 
 func (s *DockerSuite) TestRunTTYWithPipe(c *testing.T) {
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 	go func() {
 		defer close(errChan)
 
@@ -2810,7 +2810,7 @@ func (s *DockerSuite) TestRunPIDHostWithChildIsKillable(c *testing.T) {
 
 	assert.Assert(c, waitRun(name) == nil)
 
-	errchan := make(chan error)
+	errchan := make(chan error, 1)
 	go func() {
 		if out, _, err := dockerCmdWithError("kill", name); err != nil {
 			errchan <- fmt.Errorf("%v:\n%s", err, out)
@@ -3622,7 +3622,7 @@ func (s *DockerSuite) TestRunStdinBlockedAfterContainerExit(c *testing.T) {
 	cmd.Stderr = stdout
 	assert.Assert(c, cmd.Start() == nil)
 
-	waitChan := make(chan error)
+	waitChan := make(chan error, 1)
 	go func() {
 		waitChan <- cmd.Wait()
 	}()

integration-cli/docker_cli_run_unix_test.go

@@ -40,7 +40,7 @@ func (s *DockerSuite) TestRunRedirectStdout(c *testing.T) {
 		cmd.Stdout = tty
 		cmd.Stderr = tty
 		assert.NilError(c, cmd.Start())
-		ch := make(chan error)
+		ch := make(chan error, 1)
 		go func() {
 			ch <- cmd.Wait()
 			close(ch)
@@ -122,7 +122,7 @@ func (s *DockerSuite) TestRunAttachDetach(c *testing.T) {
 	_, err = cpty.Write([]byte{17})
 	assert.NilError(c, err)
 
-	ch := make(chan struct{})
+	ch := make(chan struct{}, 1)
 	go func() {
 		cmd.Wait()
 		ch <- struct{}{}
@@ -188,7 +188,7 @@ func (s *DockerSuite) TestRunAttachDetachFromFlag(c *testing.T) {
 		c.Fatal(err)
 	}
 
-	ch := make(chan struct{})
+	ch := make(chan struct{}, 1)
 	go func() {
 		cmd.Wait()
 		ch <- struct{}{}
@@ -304,7 +304,7 @@ func (s *DockerSuite) TestRunAttachDetachFromConfig(c *testing.T) {
 		c.Fatal(err)
 	}
 
-	ch := make(chan struct{})
+	ch := make(chan struct{}, 1)
 	go func() {
 		cmd.Wait()
 		ch <- struct{}{}
@@ -387,7 +387,7 @@ func (s *DockerSuite) TestRunAttachDetachKeysOverrideConfig(c *testing.T) {
 		c.Fatal(err)
 	}
 
-	ch := make(chan struct{})
+	ch := make(chan struct{}, 1)
 	go func() {
 		cmd.Wait()
 		ch <- struct{}{}
@@ -615,7 +615,7 @@ func (s *DockerSuite) TestRunWithInvalidPathforBlkioDeviceWriteIOps(c *testing.T
 
 func (s *DockerSuite) TestRunOOMExitCode(c *testing.T) {
 	testRequires(c, memoryLimitSupport, swapMemorySupport, NotPpc64le)
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 	go func() {
 		defer close(errChan)
 		// memory limit lower than 8MB will raise an error of "device or resource busy" from docker-runc.

integration-cli/docker_cli_service_logs_test.go

@@ -177,6 +177,8 @@ func (s *DockerSwarmSuite) TestServiceLogsFollow(c *testing.T) {
 	// Make sure pipe is written to
 	ch := make(chan *logMessage)
 	done := make(chan struct{})
+	stop := make(chan struct{})
+	defer close(stop)
 	go func() {
 		reader := bufio.NewReader(r)
 		for {
@@ -184,6 +186,8 @@ func (s *DockerSwarmSuite) TestServiceLogsFollow(c *testing.T) {
 			msg.data, _, msg.err = reader.ReadLine()
 			select {
 			case ch <- msg:
+			case <-stop:
+				return
 			case <-done:
 				return
 			}

integration-cli/docker_cli_start_test.go

@@ -26,7 +26,7 @@ func (s *DockerSuite) TestStartAttachReturnsOnError(c *testing.T) {
 	// err shouldn't be nil because container test2 try to link to stopped container
 	assert.Assert(c, err != nil, "out: %s", out)
 
-	ch := make(chan error)
+	ch := make(chan error, 1)
 	go func() {
 		// Attempt to start attached to the container that won't start
 		// This should return an error immediately since the container can't be started

integration-cli/docker_cli_stats_test.go

@@ -26,7 +26,7 @@ func (s *DockerSuite) TestStatsNoStream(c *testing.T) {
 		err error
 	}
 
-	ch := make(chan output)
+	ch := make(chan output, 1)
 	go func() {
 		out, err := statsCmd.Output()
 		ch <- output{out, err}

integration/container/exec_test.go

@@ -53,7 +53,7 @@ func TestExecWithCloseStdin(t *testing.T) {
 		resCh  = make(chan struct {
 			content string
 			err     error
-		})
+		}, 1)
 	)
 
 	go func() {

integration/plugin/logging/logging_linux_test.go

@@ -47,7 +47,7 @@ func TestContinueAfterPluginCrash(t *testing.T) {
 	attach, err := client.ContainerAttach(context.Background(), id, types.ContainerAttachOptions{Stream: true, Stdout: true})
 	assert.NilError(t, err)
 
-	chErr := make(chan error)
+	chErr := make(chan error, 1)
 	go func() {
 		defer close(chErr)
 		rdr := bufio.NewReader(attach.Reader)

testutil/daemon/daemon.go

@@ -645,13 +645,14 @@ func (d *Daemon) ReloadConfig() error {
 		return errors.New("daemon is not running")
 	}
 
-	errCh := make(chan error)
+	errCh := make(chan error, 1)
 	started := make(chan struct{})
 	go func() {
 		_, body, err := request.Get("/events", request.Host(d.Sock()))
 		close(started)
 		if err != nil {
 			errCh <- err
+			return
 		}
 		defer body.Close()
 		dec := json.NewDecoder(body)