test: change channel operations to avoid potential goroutine leaks

In these unit tests, goroutines may leak if certain branches are chosen. This commit edits channel operations and buffer sizes, so no matter what branch is chosen, the test will end correctly. This commit doesn't change the semantics of unit tests.

pkg/transport/timeout_dialer_test.go

@@ -22,14 +22,14 @@ import (
 
 func TestReadWriteTimeoutDialer(t *testing.T) {
 	stop := make(chan struct{})
-	defer func() {
-		stop <- struct{}{}
-	}()
 
 	ln, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
 		t.Fatalf("unexpected listen error: %v", err)
 	}
+	defer func() {
+		stop <- struct{}{}
+	}()
 	ts := testBlockingServer{ln, 2, stop}
 	go ts.Start(t)
 

server/etcdserver/server_test.go

@@ -213,7 +213,7 @@ func TestApplyRepeat(t *testing.T) {
 	// wait for conf change message
 	act, err := n.Wait(1)
 	// wait for stop message (async to avoid deadlock)
-	stopc := make(chan error)
+	stopc := make(chan error, 1)
 	go func() {
 		_, werr := n.Wait(1)
 		stopc <- werr

tests/integration/clientv3/experimental/recipes/v3_double_barrier_test.go

@@ -36,6 +36,7 @@ func TestDoubleBarrier(t *testing.T) {
 
 	b := recipe.NewDoubleBarrier(session, "test-barrier", waiters)
 	donec := make(chan struct{})
+	defer close(donec)
 	for i := 0; i < waiters-1; i++ {
 		go func() {
 			session, err := concurrency.NewSession(clus.RandClient())
@@ -48,17 +49,17 @@ func TestDoubleBarrier(t *testing.T) {
 			if err := bb.Enter(); err != nil {
 				t.Errorf("could not enter on barrier (%v)", err)
 			}
-			donec <- struct{}{}
+			<-donec
 			if err := bb.Leave(); err != nil {
 				t.Errorf("could not leave on barrier (%v)", err)
 			}
-			donec <- struct{}{}
+			<-donec
 		}()
 	}
 
 	time.Sleep(10 * time.Millisecond)
 	select {
-	case <-donec:
+	case donec <- struct{}{}:
 		t.Fatalf("barrier did not enter-wait")
 	default:
 	}
@@ -72,13 +73,13 @@ func TestDoubleBarrier(t *testing.T) {
 		select {
 		case <-timerC:
 			t.Fatalf("barrier enter timed out")
-		case <-donec:
+		case donec <- struct{}{}:
 		}
 	}
 
 	time.Sleep(10 * time.Millisecond)
 	select {
-	case <-donec:
+	case donec <- struct{}{}:
 		t.Fatalf("barrier did not leave-wait")
 	default:
 	}
@@ -89,7 +90,7 @@ func TestDoubleBarrier(t *testing.T) {
 		select {
 		case <-timerC:
 			t.Fatalf("barrier leave timed out")
-		case <-donec:
+		case donec <- struct{}{}:
 		}
 	}
 }
@@ -100,6 +101,7 @@ func TestDoubleBarrierFailover(t *testing.T) {
 
 	waiters := 10
 	donec := make(chan struct{})
+	defer close(donec)
 
 	s0, err := concurrency.NewSession(clus.Client(0))
 	if err != nil {
@@ -118,7 +120,7 @@ func TestDoubleBarrierFailover(t *testing.T) {
 		if berr := b.Enter(); berr != nil {
 			t.Errorf("could not enter on barrier (%v)", berr)
 		}
-		donec <- struct{}{}
+		<-donec
 	}()
 
 	for i := 0; i < waiters-1; i++ {
@@ -127,16 +129,16 @@ func TestDoubleBarrierFailover(t *testing.T) {
 			if berr := b.Enter(); berr != nil {
 				t.Errorf("could not enter on barrier (%v)", berr)
 			}
-			donec <- struct{}{}
+			<-donec
 			b.Leave()
-			donec <- struct{}{}
+			<-donec
 		}()
 	}
 
 	// wait for barrier enter to unblock
 	for i := 0; i < waiters; i++ {
 		select {
-		case <-donec:
+		case donec <- struct{}{}:
 		case <-time.After(10 * time.Second):
 			t.Fatalf("timed out waiting for enter, %d", i)
 		}
@@ -148,7 +150,7 @@ func TestDoubleBarrierFailover(t *testing.T) {
 	// join on rest of waiters
 	for i := 0; i < waiters-1; i++ {
 		select {
-		case <-donec:
+		case donec <- struct{}{}:
 		case <-time.After(10 * time.Second):
 			t.Fatalf("timed out waiting for leave, %d", i)
 		}

tests/integration/clientv3/kv_test.go

@@ -779,7 +779,7 @@ func TestKVPutFailGetRetry(t *testing.T) {
 		t.Fatalf("got success on disconnected put, wanted error")
 	}
 
-	donec := make(chan struct{})
+	donec := make(chan struct{}, 1)
 	go func() {
 		// Get will fail, but reconnect will trigger
 		gresp, gerr := kv.Get(context.TODO(), "foo")

tests/integration/v2_http_kv_test.go

@@ -954,7 +954,7 @@ func TestV2WatchKeyInDir(t *testing.T) {
 	tc := NewTestClient()
 
 	var body map[string]interface{}
-	c := make(chan bool)
+	c := make(chan bool, 1)
 
 	// Create an expiring directory
 	v := url.Values{}

tests/integration/v3_watch_test.go

@@ -1097,7 +1097,7 @@ func TestV3WatchWithFilter(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	recv := make(chan *pb.WatchResponse)
+	recv := make(chan *pb.WatchResponse, 1)
 	go func() {
 		// check received PUT
 		resp, rerr := ws.Recv()