test: make the tests reliable

* fix races
* fix timeouts
* extend timeouts when race detector is enabled

Author: Stebalien

go.mod

@@ -2,4 +2,7 @@ module github.com/libp2p/go-yamux
 
 go 1.12
 
-require github.com/libp2p/go-buffer-pool v0.0.2
+require (
+	github.com/ipfs/go-detect-race v0.0.1
+	github.com/libp2p/go-buffer-pool v0.0.2
+)

go.sum

@@ -1,2 +1,4 @@
+github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ6cOk=
+github.com/ipfs/go-detect-race v0.0.1/go.mod h1:8BNT7shDZPo99Q74BpGMK+4D8Mn4j46UU0LZ723meps=
 github.com/libp2p/go-buffer-pool v0.0.2 h1:QNK2iAFa8gjAe1SPz6mHSMuCcjs+X1wlHzeOSqcmlfs=
 github.com/libp2p/go-buffer-pool v0.0.2/go.mod h1:MvaB6xw5vOrDl8rYZGLFdKAuk/hRoRZd1Vi32+RXyFM=

session_test.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"log"
 	"net"
 	"reflect"
 	"runtime"
@@ -14,24 +13,26 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/ipfs/go-detect-race"
 )
 
 type logCapture struct{ bytes.Buffer }
 
 func (l *logCapture) logs() []string {
-	return strings.Split(strings.TrimSpace(l.String()), "\n")
+	lines := strings.Split(strings.TrimSpace(l.String()), "\n")
+	for i, line := range lines {
+		// trim leading date.
+		split := strings.SplitN(line, " ", 3)
+		lines[i] = split[2]
+	}
+	return lines
 }
 
 func (l *logCapture) match(expect []string) bool {
 	return reflect.DeepEqual(l.logs(), expect)
 }
 
-func captureLogs(s *Session) *logCapture {
-	buf := new(logCapture)
-	s.logger = log.New(buf, "", 0)
-	return buf
-}
-
 type pipeConn struct {
 	net.Conn
 	writeDeadline pipeDeadline
@@ -470,6 +471,11 @@ func TestSendData_Large(t *testing.T) {
 	defer client.Close()
 	defer server.Close()
 
+	timeout := 5 * time.Second
+	if detectrace.WithRace() {
+		timeout *= 10
+	}
+
 	const (
 		sendSize = 250 * 1024 * 1024
 		recvSize = 4 * 1024
@@ -487,28 +493,34 @@ func TestSendData_Large(t *testing.T) {
 		defer wg.Done()
 		stream, err := server.AcceptStream()
 		if err != nil {
-			t.Fatalf("err: %v", err)
+			t.Errorf("err: %v", err)
+			return
 		}
+
 		var sz int
 		buf := make([]byte, recvSize)
 		for i := 0; i < sendSize/recvSize; i++ {
 			n, err := io.ReadFull(stream, buf)
 			if err != nil {
-				t.Fatalf("err: %v", err)
+				t.Errorf("err: %v", err)
+				return
 			}
 			if n != recvSize {
-				t.Fatalf("short read: %d", n)
+				t.Errorf("short read: %d", n)
+				return
 			}
 			sz += n
 			for idx := range buf {
 				if buf[idx] != byte(idx%256) {
-					t.Fatalf("bad: %v %v %v", i, idx, buf[idx])
+					t.Errorf("bad: %v %v %v", i, idx, buf[idx])
+					return
 				}
 			}
 		}
 
 		if err := stream.Close(); err != nil {
-			t.Fatalf("err: %v", err)
+			t.Errorf("err: %v", err)
+			return
 		}
 
 		t.Logf("cap=%d, n=%d\n", stream.recvBuf.Cap(), sz)
@@ -518,19 +530,23 @@ func TestSendData_Large(t *testing.T) {
 		defer wg.Done()
 		stream, err := client.Open()
 		if err != nil {
-			t.Fatalf("err: %v", err)
+			t.Errorf("err: %v", err)
+			return
 		}
 
 		n, err := stream.Write(data)
 		if err != nil {
-			t.Fatalf("err: %v", err)
+			t.Errorf("err: %v", err)
+			return
 		}
 		if n != len(data) {
-			t.Fatalf("short write %d", n)
+			t.Errorf("short write %d", n)
+			return
 		}
 
 		if err := stream.Close(); err != nil {
-			t.Fatalf("err: %v", err)
+			t.Errorf("err: %v", err)
+			return
 		}
 	}()
 
@@ -541,13 +557,20 @@ func TestSendData_Large(t *testing.T) {
 	}()
 	select {
 	case <-doneCh:
-	case <-time.After(5 * time.Second):
-		panic("timeout")
+	case <-time.After(timeout):
+		client.Close()
+		server.Close()
+		wg.Wait()
+		t.Fatal("timeout")
 	}
 }
 
 func TestGoAway(t *testing.T) {
-	client, server := testClientServer()
+	// This test is noisy.
+	conf := testConf()
+	conf.LogOutput = ioutil.Discard
+
+	client, server := testClientServerConfig(conf)
 	defer client.Close()
 	defer server.Close()
 
@@ -905,11 +928,12 @@ func TestKeepAlive_Timeout(t *testing.T) {
 	client, _ := Client(conn1, clientConf)
 	defer client.Close()
 
-	server, _ := Server(conn2, testConf())
-	defer server.Close()
+	serverLogs := new(logCapture)
+	serverConf := testConf()
+	serverConf.LogOutput = serverLogs
 
-	_ = captureLogs(client) // Client logs aren't part of the test
-	serverLogs := captureLogs(server)
+	server, _ := Server(conn2, serverConf)
+	defer server.Close()
 
 	errCh := make(chan error, 1)
 	go func() {
@@ -982,6 +1006,11 @@ func TestSendData_VeryLarge(t *testing.T) {
 	defer client.Close()
 	defer server.Close()
 
+	if detectrace.WithRace() {
+		t.Skip("disabling slow test due to race detector")
+		return
+	}
+
 	var n int64 = 1 * 1024 * 1024 * 1024
 	var workers int = 16
 
@@ -993,25 +1022,30 @@ func TestSendData_VeryLarge(t *testing.T) {
 			defer wg.Done()
 			stream, err := server.AcceptStream()
 			if err != nil {
-				t.Fatalf("err: %v", err)
+				t.Errorf("err: %v", err)
+				return
 			}
 			defer stream.Close()
 
 			buf := make([]byte, 4)
 			_, err = io.ReadFull(stream, buf)
 			if err != nil {
-				t.Fatalf("err: %v", err)
+				t.Errorf("err: %v", err)
+				return
 			}
 			if !bytes.Equal(buf, []byte{0, 1, 2, 3}) {
-				t.Fatalf("bad header")
+				t.Errorf("bad header")
+				return
 			}
 
 			recv, err := io.Copy(ioutil.Discard, stream)
 			if err != nil {
-				t.Fatalf("err: %v", err)
+				t.Errorf("err: %v", err)
+				return
 			}
 			if recv != n {
-				t.Fatalf("bad: %v", recv)
+				t.Errorf("bad: %v", recv)
+				return
 			}
 		}()
 	}
@@ -1020,22 +1054,26 @@ func TestSendData_VeryLarge(t *testing.T) {
 			defer wg.Done()
 			stream, err := client.Open()
 			if err != nil {
-				t.Fatalf("err: %v", err)
+				t.Errorf("err: %v", err)
+				return
 			}
 			defer stream.Close()
 
 			_, err = stream.Write([]byte{0, 1, 2, 3})
 			if err != nil {
-				t.Fatalf("err: %v", err)
+				t.Errorf("err: %v", err)
+				return
 			}
 
 			unlimited := &UnlimitedReader{}
 			sent, err := io.Copy(stream, io.LimitReader(unlimited, n))
 			if err != nil {
-				t.Fatalf("err: %v", err)
+				t.Errorf("err: %v", err)
+				return
 			}
 			if sent != n {
-				t.Fatalf("bad: %v", sent)
+				t.Errorf("bad: %v", sent)
+				return
 			}
 		}()
 	}
@@ -1048,21 +1086,27 @@ func TestSendData_VeryLarge(t *testing.T) {
 	select {
 	case <-doneCh:
 	case <-time.After(20 * time.Second):
-		panic("timeout")
+		server.Close()
+		client.Close()
+		wg.Wait()
+		t.Fatal("timeout")
 	}
 }
 
 func TestBacklogExceeded_Accept(t *testing.T) {
 	client, server := testClientServer()
 	defer client.Close()
-	defer server.Close()
 
 	max := 5 * client.config.AcceptBacklog
+	done := make(chan struct{})
 	go func() {
+		defer close(done)
+		defer server.Close()
 		for i := 0; i < max; i++ {
 			stream, err := server.Accept()
 			if err != nil {
-				t.Fatalf("err: %v", err)
+				t.Errorf("err: %v", err)
+				return
 			}
 			defer stream.Close()
 		}
@@ -1072,14 +1116,15 @@ func TestBacklogExceeded_Accept(t *testing.T) {
 	for i := 0; i < max; i++ {
 		stream, err := client.Open()
 		if err != nil {
-			t.Fatalf("err: %v", err)
+			t.Errorf("err: %v", err)
 		}
 		defer stream.Close()
 
 		if _, err := stream.Write([]byte("foo")); err != nil {
-			t.Fatalf("err: %v", err)
+			t.Errorf("err: %v", err)
 		}
 	}
+	<-done
 }
 
 func TestSession_WindowUpdateWriteDuringRead(t *testing.T) {
@@ -1228,34 +1273,6 @@ func TestSession_sendMsg_Timeout(t *testing.T) {
 	}
 }
 
-func TestSession_PingOfDeath(t *testing.T) {
-	client, server := testClientServerConfig(testConfNoKeepAlive())
-	defer client.Close()
-	defer server.Close()
-
-	var wg sync.WaitGroup
-	begin := make(chan struct{})
-	for i := 0; i < 10000; i++ {
-		wg.Add(2)
-		go func() {
-			defer wg.Done()
-			<-begin
-			if _, err := server.Ping(); err != nil {
-				t.Error(err)
-			}
-		}()
-		go func() {
-			defer wg.Done()
-			<-begin
-			if _, err := client.Ping(); err != nil {
-				t.Error(err)
-			}
-		}()
-	}
-	close(begin)
-	wg.Wait()
-}
-
 func TestSession_ConnectionWriteTimeout(t *testing.T) {
 	client, server := testClientServerConfig(testConfNoKeepAlive())
 	defer client.Close()
@@ -1558,3 +1575,36 @@ func TestReadDeadlineInterrupt(t *testing.T) {
 		}
 	}
 }
+
+func TestSession_PingOfDeath(t *testing.T) {
+	client, server := testClientServerConfig(testConfNoKeepAlive())
+	defer client.Close()
+	defer server.Close()
+
+	count := 10000
+	if detectrace.WithRace() {
+		count = 1000
+	}
+
+	var wg sync.WaitGroup
+	begin := make(chan struct{})
+	for i := 0; i < count; i++ {
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			<-begin
+			if _, err := server.Ping(); err != nil {
+				t.Error(err)
+			}
+		}()
+		go func() {
+			defer wg.Done()
+			<-begin
+			if _, err := client.Ping(); err != nil {
+				t.Error(err)
+			}
+		}()
+	}
+	close(begin)
+	wg.Wait()
+}