window update unit test for partial read; benchmark large buffer

bench_test.go

@@ -79,3 +79,46 @@ func BenchmarkSendRecv(b *testing.B) {
 	}
 	<-doneCh
 }
+
+func BenchmarkSendRecvLarge(b *testing.B) {
+	client, server := testClientServer()
+	defer client.Close()
+	defer server.Close()
+
+	const sendSize = 100 * 1024 * 1024
+	const recvSize = 4 * 1024
+
+	sendBuf := make([]byte, sendSize)
+	recvBuf := make([]byte, recvSize)
+
+	b.ResetTimer()
+	recvDone := make(chan struct{})
+
+	go func() {
+		stream, err := server.AcceptStream()
+		if err != nil {
+			return
+		}
+		defer stream.Close()
+		for i := 0; i < b.N; i++ {
+			for j := 0; j < sendSize/recvSize; j++ {
+				if _, err := stream.Read(recvBuf); err != nil {
+					b.Fatalf("err: %v", err)
+				}
+			}
+		}
+		close(recvDone)
+	}()
+
+	stream, err := client.Open()
+	if err != nil {
+		b.Fatalf("err: %v", err)
+	}
+	defer stream.Close()
+	for i := 0; i < b.N; i++ {
+		if _, err := stream.Write(sendBuf); err != nil {
+			b.Fatalf("err: %v", err)
+		}
+	}
+	<-recvDone
+}

session_test.go

@@ -1026,6 +1026,60 @@ func TestSession_WindowUpdateWriteDuringRead(t *testing.T) {
 	wg.Wait()
 }
 
+func TestSession_PartialReadWindowUpdate(t *testing.T) {
+	client, server := testClientServerConfig(testConfNoKeepAlive())
+	defer client.Close()
+	defer server.Close()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	// Choose a huge flood size that we know will result in a window update.
+	flood := int64(client.config.MaxStreamWindowSize)
+	var wr *Stream
+
+	// The server will accept a new stream and then flood data to it.
+	go func() {
+		defer wg.Done()
+
+		var err error
+		wr, err = server.AcceptStream()
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		defer wr.Close()
+
+		if wr.sendWindow != client.config.MaxStreamWindowSize {
+			t.Fatalf("sendWindow: exp=%d, got=%d", client.config.MaxStreamWindowSize, wr.sendWindow)
+		}
+
+		n, err := wr.Write(make([]byte, flood))
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		if int64(n) != flood {
+			t.Fatalf("short write: %d", n)
+		}
+		if wr.sendWindow != 0 {
+			t.Fatalf("sendWindow: exp=%d, got=%d", 0, wr.sendWindow)
+		}
+	}()
+
+	stream, err := client.OpenStream()
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	defer stream.Close()
+
+	wg.Wait()
+
+	_, err = stream.Read(make([]byte, flood/2+1))
+
+	if exp := uint32(flood/2 + 1); wr.sendWindow != exp {
+		t.Errorf("sendWindow: exp=%d, got=%d", exp, wr.sendWindow)
+	}
+}
+
 func TestSession_sendNoWait_Timeout(t *testing.T) {
 	client, server := testClientServerConfig(testConfNoKeepAlive())
 	defer client.Close()