feat(perf): continuosly measure on single conn (iperf-style)

perf/README.md

@@ -79,11 +79,33 @@ Given you have provisioned your infrastructure, you can now build and run the li
             - `--download-bytes` number of bytes to download per stream.
           - Output
             - Logging MUST go to `stderr`.
-            - Measurement output is printed to `stdout` as JSON in the form of:
-              ```json
-              {"latency": 0.246442851}
-              ```
-              Note that the measurement includes the time to (1) establish the
-              connection, (2) upload the bytes and (3) download the bytes.
+            - Measurement output is printed to `stdout` as JSON.
+            - The output schema is:
+               ``` typescript
+               interface Data {
+                 type: "intermediary" | "final";
+                 timeSeconds: number;
+                 uploadBytes: number;
+                 downloadBytes: number;
+               }
+               ```
+            - Every second the client must print the current progress to stdout. See example below. Note the `type: "intermediary"`.
+               ``` json
+               {
+                 "type": "intermediary",
+                 "timeSeconds": 1.004957645,
+                 "uploadBytes": 73039872,
+                 "downloadBytes": 0
+               },
+               ```
+            - Before terminating the client must print a final summary. See example below. Note the `type: "final"`. Also note that the measurement includes the time to (1) establish the connection, (2) upload the bytes and (3) download the bytes.
+               ``` json
+               {
+                 "type": "final",
+                 "timeSeconds": 60.127230659,
+                 "uploadBytes": 4382392320,
+                 "downloadBytes": 0
+               }
+               ```
 2. For a new implementation, in [`impl/Makefile` include your implementation in the `all` target.](./impl/Makefile#L7)
 3. For a new version, reference version in [`runner/src/versions.ts`](./runner/src/versions.ts#L7-L43).

perf/impl/go-libp2p/v0.29/main.go

@@ -89,7 +89,10 @@ func main() {
 	}
 
 	jsonB, err := json.Marshal(Result{
-		Latency: time.Since(start).Seconds(),
+		TimeSeconds: time.Since(start).Seconds(),
+		UploadBytes: uint(*uploadBytes),
+		DownloadBytes: uint(*downloadBytes),
+		Type: "final",
 	})
 	if err != nil {
 		log.Fatalf("failed to marshal perf result: %s", err)
@@ -99,7 +102,10 @@ func main() {
 }
 
 type Result struct {
-	Latency float64 `json:"latency"`
+	Type          string  `json:"type"`
+	TimeSeconds   float64 `json:"timeSeconds"`
+	UploadBytes   uint    `json:"uploadBytes"`
+	DownloadBytes uint    `json:"downloadBytes"`
 }
 
 type simpleReader struct {

perf/impl/go-libp2p/v0.29/perf.go

@@ -3,8 +3,10 @@ package main
 import (
 	"context"
 	"encoding/binary"
+	"encoding/json"
 	"fmt"
 	"io"
+	"time"
 
 	logging "github.com/ipfs/go-log/v2"
 	pool "github.com/libp2p/go-buffer-pool"
@@ -89,7 +91,26 @@ func sendBytes(s io.Writer, bytesToSend uint64) error {
 	buf := pool.Get(blockSize)
 	defer pool.Put(buf)
 
+	lastReportTime := time.Now()
+	lastReportWrite := 0
+
 	for bytesToSend > 0 {
+		now := time.Now()
+		if now.Sub(lastReportTime) > time.Second {
+			jsonB, err := json.Marshal(Result{
+				TimeSeconds: now.Sub(lastReportTime).Seconds(),
+				UploadBytes: uint(lastReportWrite),
+				Type: "intermediary",
+			})
+			if err != nil {
+				log.Fatalf("failed to marshal perf result: %s", err)
+			}
+			fmt.Println(string(jsonB))
+
+			lastReportTime = now
+			lastReportWrite = 0
+		}
+
 		toSend := buf
 		if bytesToSend < blockSize {
 			toSend = buf[:bytesToSend]
@@ -100,15 +121,49 @@ func sendBytes(s io.Writer, bytesToSend uint64) error {
 			return err
 		}
 		bytesToSend -= uint64(n)
+		lastReportWrite += n
 	}
 	return nil
 }
 
 func drainStream(s io.Reader) (uint64, error) {
 	var recvd int64
-	recvd, err := io.Copy(io.Discard, s)
+	recvd, err := io.Copy(io.Discard, & reportingReader { orig: s, LastReportTime: time.Now() })
 	if err != nil && err != io.EOF {
 		return uint64(recvd), err
 	}
 	return uint64(recvd), nil
 }
+
+type reportingReader struct {
+	orig             io.Reader
+	LastReportTime   time.Time
+	lastReportRead   uint64
+}
+
+var _ io.Reader = &reportingReader{}
+
+func (r *reportingReader) Read(b []byte) (int, error) {
+	n, err := r.orig.Read(b)
+	r.lastReportRead += uint64(n)
+
+	now := time.Now()
+	if now.Sub(r.LastReportTime) > time.Second {
+		result := Result{
+			TimeSeconds: now.Sub(r.LastReportTime).Seconds(),
+			Type: "intermediary",
+			DownloadBytes: uint(r.lastReportRead),
+		}
+
+		jsonB, err := json.Marshal(result)
+		if err != nil {
+			log.Fatalf("failed to marshal perf result: %s", err)
+		}
+		fmt.Println(string(jsonB))
+
+		r.LastReportTime = now
+		r.lastReportRead = 0
+	}
+
+	return n, err
+}

perf/impl/https/v0.1/main.go

@@ -49,26 +49,6 @@ func handleRequest(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-type nullReader struct {
-	N    uint64
-	read uint64
-}
-
-var _ io.Reader = &nullReader{}
-
-func (r *nullReader) Read(b []byte) (int, error) {
-	remaining := r.N - r.read
-	l := uint64(len(b))
-	if uint64(len(b)) > remaining {
-		l = remaining
-	}
-	r.read += l
-	if r.read == r.N {
-		return int(l), io.EOF
-	}
-	return int(l), nil
-}
-
 func runClient(serverAddr string, uploadBytes, downloadBytes uint64) (time.Duration, error) {
 	client := &http.Client{
 		Transport: &http.Transport{
@@ -84,7 +64,7 @@ func runClient(serverAddr string, uploadBytes, downloadBytes uint64) (time.Durat
 		fmt.Sprintf("https://%s/", serverAddr),
 		io.MultiReader(
 			bytes.NewReader(b),
-			&nullReader{N: uploadBytes},
+			&reportingReader { orig: &nullReader{ N: uploadBytes }, LastReportTime: time.Now(), isUpload: true },
 		),
 	)
 	if err != nil {
@@ -167,7 +147,10 @@ func generateEphemeralCertificate() (tls.Certificate, error) {
 }
 
 type Result struct {
-	Latency float64 `json:"latency"`
+	Type          string  `json:"type"`
+	TimeSeconds   float64 `json:"timeSeconds"`
+	UploadBytes   uint    `json:"uploadBytes"`
+	DownloadBytes uint    `json:"downloadBytes"`
 }
 
 func main() {
@@ -220,7 +203,10 @@ func main() {
 		}
 
 		jsonB, err := json.Marshal(Result{
-			Latency: latency.Seconds(),
+			TimeSeconds: latency.Seconds(),
+			UploadBytes: uint(*uploadBytes),
+			DownloadBytes: uint(*downloadBytes),
+			Type: "final",
 		})
 		if err != nil {
 			log.Fatalf("failed to marshal perf result: %s", err)
@@ -249,9 +235,72 @@ func sendBytes(s io.Writer, bytesToSend uint64) error {
 
 func drainStream(s io.Reader) (uint64, error) {
 	var recvd int64
-	recvd, err := io.Copy(io.Discard, s)
+	recvd, err := io.Copy(io.Discard, & reportingReader { orig: s, LastReportTime: time.Now(), isUpload: false })
 	if err != nil && err != io.EOF {
 		return uint64(recvd), err
 	}
 	return uint64(recvd), nil
 }
+
+type reportingReader struct {
+	orig             io.Reader
+	LastReportTime   time.Time
+	lastReportRead   uint64
+	isUpload bool
+}
+
+var _ io.Reader = &reportingReader{}
+
+func (r *reportingReader) Read(b []byte) (int, error) {
+	n, err := r.orig.Read(b)
+	r.lastReportRead += uint64(n)
+
+	now := time.Now()
+	if now.Sub(r.LastReportTime) > time.Second {
+		// This section is analogous to your Read implementation
+		result := Result{
+			TimeSeconds: now.Sub(r.LastReportTime).Seconds(),
+			Type: "intermediary",
+		}
+		if r.isUpload {
+			result.UploadBytes = uint(r.lastReportRead)
+		} else {
+			result.DownloadBytes = uint(r.lastReportRead)
+		}
+
+		jsonB, err := json.Marshal(result)
+		if err != nil {
+			log.Fatalf("failed to marshal perf result: %s", err)
+		}
+		fmt.Println(string(jsonB))
+
+		r.LastReportTime = now
+		r.lastReportRead = 0
+	}
+
+	return n, err
+}
+
+type nullReader struct {
+	N    uint64
+	read uint64
+	LastReportTime time.Time
+	lastReportRead uint64
+}
+
+var _ io.Reader = &nullReader{}
+
+func (r *nullReader) Read(b []byte) (int, error) {
+	remaining := r.N - r.read
+	l := uint64(len(b))
+	if uint64(len(b)) > remaining {
+		l = remaining
+	}
+	r.read += l
+
+	if r.read == r.N {
+		return int(l), io.EOF
+	}
+
+	return int(l), nil
+}

perf/impl/quic-go/v0.34/Makefile

@@ -1,4 +1,4 @@
-commitSha := a5cd126c97b6d8d8328141bfa84cc57e74ebc57c
+commitSha := 3a12ea9207e40de20533b0a6aa2e40e3727aa796
 
 all: perf
 
@@ -10,7 +10,8 @@ perf-${commitSha}: perf-${commitSha}.zip
 	unzip -o perf-${commitSha}.zip
 
 perf-${commitSha}.zip:
-	wget -O $@ "https://github.com/quic-go/perf/archive/${commitSha}.zip"
+	# TODO: revert
+	wget -O $@ "https://github.com/mxinden/perf/archive/${commitSha}.zip"
 
 clean:
 	rm perf-*.zip

perf/impl/rust-libp2p/v0.52/Makefile

@@ -1,4 +1,4 @@
-commitSha := e8759c85c278006f5fc94e823c2a3620abaaf697
+commitSha := 0f99dd444f2b9b8e903c5be2db5b420086114c16
 
 all: perf
 
@@ -12,7 +12,8 @@ rust-libp2p-${commitSha}: rust-libp2p-${commitSha}.zip
 	unzip -o rust-libp2p-${commitSha}.zip
 
 rust-libp2p-${commitSha}.zip:
-	wget -O $@ "https://github.com/libp2p/rust-libp2p/archive/${commitSha}.zip"
+	# TODO revert
+	wget -O $@ "https://github.com/mxinden/rust-libp2p/archive/${commitSha}.zip"
 
 clean:
 	rm rust-libp2p-*.zip