bugfixes and small improvements to bench and orchestrate

Author: crimist

cmd/bench/http.go

@@ -2,10 +2,10 @@ package main
 
 import (
 	"context"
-	"fmt"
 	"io"
 	mrand "math/rand"
 	"net"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -55,15 +55,15 @@ func buildHTTPAnnounce(hash, peer, host string, port int, left int64, numwant in
 	b.WriteString("&peer_id=")
 	b.WriteString(peer)
 	b.WriteString("&port=")
-	b.WriteString(fmt.Sprintf("%d", port))
+	b.WriteString(strconv.Itoa(port))
 	b.WriteString("&uploaded=0&downloaded=0&left=")
-	b.WriteString(fmt.Sprintf("%d", left))
+	b.WriteString(strconv.FormatInt(left, 10))
 	if compact {
 		b.WriteString("&compact=1")
 	}
 	if numwant >= 0 {
 		b.WriteString("&numwant=")
-		b.WriteString(fmt.Sprintf("%d", numwant))
+		b.WriteString(strconv.Itoa(numwant))
 	}
 	b.WriteString(" HTTP/1.1\r\nHost: ")
 	b.WriteString(host)
@@ -95,7 +95,7 @@ type httpBenchWorker struct {
 
 func (w *httpBenchWorker) run(ctx context.Context, cfg config, ds *dataset, limiter *rateLimiter) *workerMetrics {
 	metrics := newWorkerMetrics()
-	rng := mrand.New(mrand.NewSource(cfg.rngSeed + int64(w.id*3571)))
+	rng := mrand.New(mrand.NewSource(cfg.rngSeed + int64(w.id*rngOffsetHTTP)))
 	peerIdx := w.id % len(ds.peers)
 	peer := ds.encodedPeers[peerIdx]
 

cmd/bench/main.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
+	"io"
 	"os"
 	"runtime"
 	"time"
@@ -32,6 +33,11 @@ const (
 	defaultSeedFraction = 0.2 // 20% seeders
 
 	defaultPort = 6881
+
+	// RNG seed offsets for reproducible but distinct sequences per worker type.
+	rngOffsetUDP  = 7919
+	rngOffsetHTTP = 3571
+	rngOffsetSeed = 11
 )
 
 // Hardcoded numwant distribution: 20 (90%), 30 (8%), 50 (2%)
@@ -193,18 +199,18 @@ func main() {
 		Warnings: warnings,
 	}
 
-	enc := json.NewEncoder(os.Stdout)
-	enc.SetIndent("", "  ")
+	var w io.Writer = os.Stdout
 	if cfg.outPath != "" {
 		f, err := os.Create(cfg.outPath)
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "failed to open output file: %v\n", err)
 			os.Exit(1)
 		}
 		defer f.Close()
-		enc = json.NewEncoder(f)
-		enc.SetIndent("", "  ")
+		w = f
 	}
+	enc := json.NewEncoder(w)
+	enc.SetIndent("", "  ")
 	if err := enc.Encode(out); err != nil {
 		fmt.Fprintf(os.Stderr, "failed to encode results: %v\n", err)
 		os.Exit(1)

cmd/bench/metrics.go

@@ -6,8 +6,6 @@ import (
 	"time"
 )
 
-const maxInt64 = int64(^uint64(0) >> 1)
-
 // Latency histogram.
 
 type latencyHistogram struct {
@@ -31,7 +29,7 @@ func newLatencyHistogram(start time.Duration, buckets int) *latencyHistogram {
 	return &latencyHistogram{
 		bounds: bounds,
 		counts: make([]int64, buckets+1),
-		min:    maxInt64,
+		min:    math.MaxInt64,
 	}
 }
 
@@ -74,7 +72,7 @@ func (h *latencyHistogram) merge(other *latencyHistogram) {
 
 func (h *latencyHistogram) quantile(q float64) time.Duration {
 	if q <= 0 {
-		if h.min == maxInt64 {
+		if h.min == math.MaxInt64 {
 			return 0
 		}
 		return time.Duration(h.min)
@@ -112,7 +110,7 @@ func (h *latencyHistogram) mean() time.Duration {
 }
 
 func formatDuration(d time.Duration) string {
-	if int64(d) == maxInt64 {
+	if int64(d) == math.MaxInt64 {
 		return "0s"
 	}
 	return d.String()

cmd/bench/seed.go

@@ -11,7 +11,7 @@ func seedPhase(ctx context.Context, cfg config, ds *dataset) error {
 		return nil
 	}
 
-	rng := mrand.New(mrand.NewSource(cfg.rngSeed + 11))
+	rng := mrand.New(mrand.NewSource(cfg.rngSeed + rngOffsetSeed))
 
 	seedHashes := ds.torrents
 	if cfg.seed > 0 && cfg.seed < len(seedHashes) {

cmd/bench/udp.go

@@ -162,7 +162,7 @@ type udpBenchWorker struct {
 
 func (w *udpBenchWorker) run(ctx context.Context, cfg config, ds *dataset, limiter *rateLimiter) *workerMetrics {
 	metrics := newWorkerMetrics()
-	rng := mrand.New(mrand.NewSource(cfg.rngSeed + int64(w.id*7919)))
+	rng := mrand.New(mrand.NewSource(cfg.rngSeed + int64(w.id*rngOffsetUDP)))
 	peerIdx := w.id % len(ds.peers)
 	peerID := ds.peers[peerIdx]
 

cmd/bench/util.go

@@ -14,7 +14,7 @@ import (
 // Random number generation.
 
 func cryptoSeed() int64 {
-	n, err := rand.Int(rand.Reader, big.NewInt(maxInt64))
+	n, err := rand.Int(rand.Reader, big.NewInt(math.MaxInt64))
 	if err != nil {
 		return time.Now().UnixNano()
 	}
@@ -48,10 +48,10 @@ func chooseWeighted(rng *mrand.Rand, choices []weightedChoice, fallback int) int
 // Statistical distributions.
 
 func clampInt(val, minVal, maxVal int) int {
-	if minVal > 0 && val < minVal {
+	if val < minVal {
 		return minVal
 	}
-	if maxVal > 0 && val > maxVal {
+	if val > maxVal {
 		return maxVal
 	}
 	return val

cmd/orchestrate/config.go

@@ -392,7 +392,7 @@ type RunResult struct {
 	RunID       string                 `json:"run_id"`
 	Mode        string                 `json:"mode"`
 	Routines    int                    `json:"routines"`
-	OtherRoutes int                    `json:"other_routines"`
+	OtherRoutines int                    `json:"other_routines"`
 	ResultFile  string                 `json:"result_file"`
 	Status      string                 `json:"status"`
 	Error       string                 `json:"error,omitempty"`
@@ -478,6 +478,26 @@ func (rc *ResultCollector) GetResults() []RunResult {
 	return results
 }
 
+// benchResultFile represents the structure of a bench result JSON file.
+type benchResultFile struct {
+	Metrics map[string]benchMetric `json:"metrics"`
+}
+
+type benchMetric struct {
+	RatePerSec float64      `json:"rate_per_sec"`
+	Success    float64      `json:"success"`
+	Errors     float64      `json:"errors"`
+	Timeouts   float64      `json:"timeouts"`
+	Latency    benchLatency `json:"latency"`
+}
+
+type benchLatency struct {
+	P50 string `json:"p50"`
+	P90 string `json:"p90"`
+	P95 string `json:"p95"`
+	P99 string `json:"p99"`
+}
+
 // LoadResultFile loads and parses a bench result file.
 func LoadResultFile(path string) (*ResultSummary, map[string]interface{}, error) {
 	data, err := os.ReadFile(path)
@@ -490,40 +510,22 @@ func LoadResultFile(path string) (*ResultSummary, map[string]interface{}, error)
 		return nil, nil, err
 	}
 
-	summary := &ResultSummary{}
+	var result benchResultFile
+	if err := json.Unmarshal(data, &result); err != nil {
+		return nil, nil, err
+	}
 
-	if metrics, ok := raw["metrics"].(map[string]interface{}); ok {
-		for _, v := range metrics {
-			if m, ok := v.(map[string]interface{}); ok {
-				if rate, ok := m["rate_per_sec"].(float64); ok {
-					summary.Throughput = rate
-				}
-				if success, ok := m["success"].(float64); ok {
-					summary.Success = int64(success)
-				}
-				if errors, ok := m["errors"].(float64); ok {
-					summary.Errors = int64(errors)
-				}
-				if timeouts, ok := m["timeouts"].(float64); ok {
-					summary.Timeouts = int64(timeouts)
-				}
-				if lat, ok := m["latency"].(map[string]interface{}); ok {
-					if p50, ok := lat["p50"].(string); ok {
-						summary.P50 = p50
-					}
-					if p90, ok := lat["p90"].(string); ok {
-						summary.P90 = p90
-					}
-					if p95, ok := lat["p95"].(string); ok {
-						summary.P95 = p95
-					}
-					if p99, ok := lat["p99"].(string); ok {
-						summary.P99 = p99
-					}
-				}
-				break
-			}
-		}
+	summary := &ResultSummary{}
+	for _, m := range result.Metrics {
+		summary.Throughput = m.RatePerSec
+		summary.Success = int64(m.Success)
+		summary.Errors = int64(m.Errors)
+		summary.Timeouts = int64(m.Timeouts)
+		summary.P50 = m.Latency.P50
+		summary.P90 = m.Latency.P90
+		summary.P95 = m.Latency.P95
+		summary.P99 = m.Latency.P99
+		break // Only need the first metric
 	}
 
 	return summary, raw, nil

cmd/orchestrate/main.go

@@ -131,36 +131,7 @@ func runLocal(args []string) error {
 	executor := NewLocalExecutor(tracker, benchBin, *target, *outputDir)
 	collector := NewResultCollector(*outputDir, scenario.Name, *mode)
 
-	slog.Info(fmt.Sprintf("\nRunning %d benchmark(s)...\n", len(runs)))
-
-	for i, run := range runs {
-		slog.Info(fmt.Sprintf("[%d/%d] %s (routines=%d)", i+1, len(runs), run.Name, run.Routines))
-
-		if *pause && i > 0 {
-			fmt.Fprint(os.Stderr, "Press Enter to continue...")
-			if err := WaitForInput(); err != nil {
-				return fmt.Errorf("wait for input: %w", err)
-			}
-		}
-
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		default:
-		}
-
-		result := executor.Run(ctx, run)
-		collector.AddResult(result)
-
-		if result.Summary != nil {
-			slog.Info(fmt.Sprintf("  -> %.0f req/s | p50: %s | p90: %s | p95: %s | p99: %s",
-				result.Summary.Throughput, result.Summary.P50, result.Summary.P90, result.Summary.P95, result.Summary.P99))
-		} else if result.Error != "" {
-			slog.Error(fmt.Sprintf("  -> %s", result.Error))
-		}
-	}
-
-	return FinalizeResults(collector)
+	return runBenchmarkLoop(ctx, runs, executor, collector, *pause)
 }
 
 func runClient(args []string) error {
@@ -258,12 +229,19 @@ func runClient(args []string) error {
 	executor := NewClientExecutor(conn, benchBin, *target, *statsURL, *outputDir)
 	collector := NewResultCollector(*outputDir, scenario.Name, *mode)
 
-	slog.Info(fmt.Sprintf("Connected. Running %d benchmark(s)...\n", len(runs)))
+	slog.Info(fmt.Sprintf("Connected. Running %d benchmark(s)...", len(runs)))
+
+	return runBenchmarkLoop(ctx, runs, executor, collector, *pause)
+}
+
+// runBenchmarkLoop executes the benchmark runs and collects results.
+func runBenchmarkLoop(ctx context.Context, runs []BenchmarkRun, executor *Executor, collector *ResultCollector, pause bool) error {
+	slog.Info(fmt.Sprintf("Running %d benchmark(s)...\n", len(runs)))
 
 	for i, run := range runs {
 		slog.Info(fmt.Sprintf("[%d/%d] %s (routines=%d)", i+1, len(runs), run.Name, run.Routines))
 
-		if *pause && i > 0 {
+		if pause && i > 0 {
 			fmt.Fprint(os.Stderr, "Press Enter to continue...")
 			if err := WaitForInput(); err != nil {
 				return fmt.Errorf("wait for input: %w", err)

cmd/orchestrate/run.go

@@ -55,7 +55,7 @@ func (e *Executor) Run(ctx context.Context, run BenchmarkRun) RunResult {
 		RunID:       run.Name,
 		Mode:        run.Mode,
 		Routines:    run.Routines,
-		OtherRoutes: run.OtherRoutines,
+		OtherRoutines: run.OtherRoutines,
 	}
 
 	udpRoutines, httpRoutines := DetermineRoutines(run)
@@ -128,23 +128,10 @@ func (e *Executor) startTracker(ctx context.Context, udpRoutines, httpRoutines i
 
 // startTrackerLocal handles local tracker startup.
 func (e *Executor) startTrackerLocal(ctx context.Context, udpRoutines, httpRoutines int, run BenchmarkRun) error {
-	if e.tracker.IsAlive() {
-		slog.Info("  Stopping existing tracker...")
-		if err := e.tracker.Stop(ctx); err != nil {
-			slog.Warn("stop tracker", "error", err)
-		}
-		time.Sleep(time.Second)
-	}
-
-	slog.Info("  Clearing cache...")
-	if err := e.tracker.ClearCache(); err != nil {
-		slog.Warn("failed to clear cache", "error", err)
-	}
-
 	slog.Info(fmt.Sprintf("  Starting tracker (%s_routines=%d, other=%d)...",
 		run.Mode, run.Routines, run.OtherRoutines))
 
-	return e.tracker.Start(ctx, udpRoutines, httpRoutines)
+	return e.tracker.Restart(ctx, udpRoutines, httpRoutines)
 }
 
 // startTrackerRemote handles remote tracker startup via protocol.
@@ -169,7 +156,9 @@ func (e *Executor) startTrackerRemote(udpRoutines, httpRoutines int) error {
 		return nil
 	case MsgTypeError:
 		var errMsg ErrorMessage
-		json.Unmarshal(data, &errMsg)
+		if err := json.Unmarshal(data, &errMsg); err != nil {
+			return fmt.Errorf("server error (failed to parse: %v)", err)
+		}
 		return fmt.Errorf("server error: %s", errMsg.ErrorMsg)
 	default:
 		return fmt.Errorf("unexpected response: %s", msgType)
@@ -207,7 +196,9 @@ func (e *Executor) notifyDone(udpRoutines, httpRoutines int, resultFile string,
 		return nil
 	case MsgTypeError:
 		var errMsg ErrorMessage
-		json.Unmarshal(data, &errMsg)
+		if err := json.Unmarshal(data, &errMsg); err != nil {
+			return fmt.Errorf("server error on stop (failed to parse: %v)", err)
+		}
 		return fmt.Errorf("server error on stop: %s", errMsg.ErrorMsg)
 	default:
 		return fmt.Errorf("unexpected response: %s", msgType)