support iperf-style intermittent results (#17)

* feat(perf): support iperf-style intermittent results

Required for libp2p/test-plans#261.

* Print on read

* go fmt

* Apply code review comments

* go fmt

client.go

@@ -14,7 +14,10 @@ import (
 )
 
 type Result struct {
-	Latency float64 `json:"latency"`
+	Type          string  `json:"type"`
+	TimeSeconds   float64 `json:"timeSeconds"`
+	UploadBytes   uint64  `json:"uploadBytes"`
+	DownloadBytes uint64  `json:"downloadBytes"`
 }
 
 func RunClient(addr string, uploadBytes, downloadBytes uint64, keyLogFile io.Writer) error {
@@ -45,7 +48,10 @@ func RunClient(addr string, uploadBytes, downloadBytes uint64, keyLogFile io.Wri
 	log.Printf("uploaded %s: %.2fs (%s/s)", formatBytes(uploadBytes), uploadTook.Seconds(), formatBytes(bandwidth(uploadBytes, uploadTook)))
 	log.Printf("downloaded %s: %.2fs (%s/s)", formatBytes(downloadBytes), downloadTook.Seconds(), formatBytes(bandwidth(downloadBytes, downloadTook)))
 	json, err := json.Marshal(Result{
-		Latency: time.Since(start).Seconds(),
+		TimeSeconds:   time.Since(start).Seconds(),
+		Type:          "final",
+		UploadBytes:   uploadBytes,
+		DownloadBytes: downloadBytes,
 	})
 	if err != nil {
 		return err
@@ -64,7 +70,27 @@ func handleClientStream(str io.ReadWriteCloser, uploadBytes, downloadBytes uint6
 	// upload data
 	b = make([]byte, 16*1024)
 	uploadStart := time.Now()
+
+	lastReportTime := time.Now()
+	lastReportWrite := uint64(0)
+
 	for uploadBytes > 0 {
+		now := time.Now()
+		if now.Sub(lastReportTime) >= time.Second {
+			jsonB, err := json.Marshal(Result{
+				TimeSeconds: now.Sub(lastReportTime).Seconds(),
+				UploadBytes: lastReportWrite,
+				Type:        "intermediary",
+			})
+			if err != nil {
+				log.Fatalf("failed to marshal perf result: %s", err)
+			}
+			fmt.Println(string(jsonB))
+
+			lastReportTime = now
+			lastReportWrite = 0
+		}
+
 		if uploadBytes < uint64(len(b)) {
 			b = b[:uploadBytes]
 		}
@@ -73,7 +99,9 @@ func handleClientStream(str io.ReadWriteCloser, uploadBytes, downloadBytes uint6
 			return 0, 0, err
 		}
 		uploadBytes -= uint64(n)
+		lastReportWrite += uint64(n)
 	}
+
 	if err := str.Close(); err != nil {
 		return 0, 0, err
 	}
@@ -83,12 +111,33 @@ func handleClientStream(str io.ReadWriteCloser, uploadBytes, downloadBytes uint6
 	b = b[:cap(b)]
 	remaining := downloadBytes
 	downloadStart := time.Now()
+
+	lastReportTime = time.Now()
+	lastReportRead := uint64(0)
+
 	for remaining > 0 {
+		now := time.Now()
+		if now.Sub(lastReportTime) >= time.Second {
+			jsonB, err := json.Marshal(Result{
+				TimeSeconds:   now.Sub(lastReportTime).Seconds(),
+				DownloadBytes: lastReportRead,
+				Type:          "intermediary",
+			})
+			if err != nil {
+				log.Fatalf("failed to marshal perf result: %s", err)
+			}
+			fmt.Println(string(jsonB))
+
+			lastReportTime = now
+			lastReportRead = 0
+		}
+
 		n, err := str.Read(b)
 		if uint64(n) > remaining {
 			return 0, 0, fmt.Errorf("server sent more data than expected, expected %d, got %d", downloadBytes, remaining+uint64(n))
 		}
 		remaining -= uint64(n)
+		lastReportRead += uint64(n)
 		if err != nil {
 			if err == io.EOF {
 				if remaining == 0 {