feat(benchmarks/rps): add latency output

Signed-off-by: René Jochum <jochumdev@gmail.com>

Author: jochumdev

benchmarks/rps/cmd/orb-rps-client/main.go

@@ -7,8 +7,10 @@ import (
 	"crypto/rand"
 	"errors"
 	"fmt"
+	"math"
 	"os"
 	"runtime"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -38,6 +40,49 @@ import (
 
 const serverName = "benchmarks.rps.server"
 
+type latencyStats struct {
+	latencies []time.Duration
+	mu        sync.Mutex
+}
+
+func (l *latencyStats) add(d time.Duration) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.latencies = append(l.latencies, d)
+}
+
+func (l *latencyStats) calculate() (time.Duration, time.Duration) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	if len(l.latencies) == 0 {
+		return 0, 0
+	}
+
+	// Sort latencies for P99 calculation
+	sort.Slice(l.latencies, func(i, j int) bool {
+		return l.latencies[i] < l.latencies[j]
+	})
+
+	// Calculate average
+	var total time.Duration
+	for _, d := range l.latencies {
+		total += d
+	}
+
+	avg := total / time.Duration(len(l.latencies))
+
+	// Calculate P99
+	p99Index := int(math.Floor(float64(len(l.latencies)) * 0.99))
+	if p99Index >= len(l.latencies) {
+		p99Index = len(l.latencies) - 1
+	}
+
+	p99 := l.latencies[p99Index]
+
+	return avg, p99
+}
+
 type benchStats struct {
 	ok    atomic.Uint64
 	error atomic.Uint64
@@ -52,6 +97,7 @@ func runConnection(
 	msg []byte,
 	opts []client.CallOption,
 	stats *benchStats,
+	latencyStats *latencyStats,
 ) {
 	defer wg.Done()
 
@@ -63,6 +109,7 @@ func runConnection(
 		case <-ctx.Done():
 			return
 		default:
+			start := time.Now()
 			// Run the query
 			resp, err := client.Echo(
 				ctx,
@@ -88,6 +135,7 @@ func runConnection(
 				continue
 			}
 
+			latencyStats.add(time.Since(start))
 			stats.ok.Add(1)
 		}
 	}
@@ -113,10 +161,11 @@ func runBenchmark(
 	logger log.Logger,
 	msg []byte,
 	opts []client.CallOption,
-) (uint64, uint64) {
+) (uint64, uint64, time.Duration, time.Duration) {
 	var wg sync.WaitGroup
 
 	stats := &benchStats{}
+	latencyStats := &latencyStats{latencies: make([]time.Duration, 0, 2000000)}
 
 	// Create benchmark context with timeout
 	benchCtx, cancel := context.WithTimeout(ctx, time.Duration(duration)*time.Second)
@@ -126,7 +175,7 @@ func runBenchmark(
 	for i := 0; i < connections; i++ {
 		wg.Add(1)
 
-		go runConnection(benchCtx, &wg, cli, logger, msg, opts, stats)
+		go runConnection(benchCtx, &wg, cli, logger, msg, opts, stats, latencyStats)
 	}
 
 	// Wait for timeout or cancellation
@@ -135,11 +184,13 @@ func runBenchmark(
 	// Wait for all goroutines to finish
 	wg.Wait()
 
-	return stats.ok.Load(), stats.error.Load()
+	avgLatency, p99Latency := latencyStats.calculate()
+	return stats.ok.Load(), stats.error.Load(), avgLatency, p99Latency
 }
 
 // bench is the main benchmark function.
 func bench(ctx context.Context, cfg *clientConfig, logger log.Logger, cli client.Type) error {
+
 	// Log configuration
 	logger.Info("Configuration",
 		"connections", cfg.Connections,
@@ -168,18 +219,20 @@ func bench(ctx context.Context, cfg *clientConfig, logger log.Logger, cli client
 
 	// Run warmup phase
 	logger.Info("Warming up...")
-	warmupOk, warmupErr := runBenchmark(ctx, 5, cfg.Connections, cli, logger, msg, opts)
+	warmupOk, warmupErr, _, _ := runBenchmark(ctx, 5, cfg.Connections, cli, logger, msg, opts)
 	logger.Debug("Warmup complete", "requests_ok", warmupOk, "requests_error", warmupErr)
 
 	// Run benchmark phase
 	logger.Info("Running benchmark...")
-	reqsOk, reqsError := runBenchmark(ctx, cfg.Duration, cfg.Connections, cli, logger, msg, opts)
+	reqsOk, reqsError, avg, p99 := runBenchmark(ctx, cfg.Duration, cfg.Connections, cli, logger, msg, opts)
 
 	// Log results
 	logger.Info("Summary",
 		"requests_ok", reqsOk,
 		"requests_error", reqsError,
 		"qps", float64(reqsOk)/float64(cfg.Duration),
+		"avg_latency_us", avg.Microseconds(),
+		"p99_latency_us", p99.Microseconds(),
 		"connections", cfg.Connections,
 		"duration_seconds", cfg.Duration,
 		"package_size", cfg.PackageSize,