phase 10

internal/broker/broker_bench_test.go

@@ -0,0 +1,101 @@
+package broker_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/uopi/uopi/internal/broker"
+)
+
+// BenchmarkFanOut measures the end-to-end latency and throughput of the broker
+// fan-out with varying numbers of downstream clients.
+
+func BenchmarkFanOut1Client(b *testing.B)   { benchFanOut(b, 1) }
+func BenchmarkFanOut10Clients(b *testing.B)  { benchFanOut(b, 10) }
+func BenchmarkFanOut20Clients(b *testing.B)  { benchFanOut(b, 20) }
+func BenchmarkFanOut100Clients(b *testing.B) { benchFanOut(b, 100) }
+
+func benchFanOut(b *testing.B, nClients int) {
+	b.Helper()
+
+	brk, cancel := newBroker(b)
+	defer cancel()
+
+	ref := broker.SignalRef{DS: "stub", Name: "counter_fast"}
+
+	// Subscribe N downstream clients.
+	type sub struct {
+		ch     chan broker.Update
+		cancel func()
+	}
+	subs := make([]sub, nClients)
+	for i := range subs {
+		ch := make(chan broker.Update, 512)
+		cancelFn, err := brk.Subscribe(ref, ch)
+		if err != nil {
+			b.Skip("signal not available:", err)
+		}
+		subs[i] = sub{ch, cancelFn}
+	}
+	defer func() {
+		for _, s := range subs {
+			s.cancel()
+		}
+	}()
+
+	// Warm-up: wait for the first update to arrive on every client.
+	timeout := time.After(3 * time.Second)
+	for _, s := range subs {
+		select {
+		case <-s.ch:
+		case <-timeout:
+			b.Skip("timed out waiting for first update — signal may not exist in stub")
+		}
+	}
+
+	b.ResetTimer()
+	b.ReportAllocs()
+
+	for range b.N {
+		// Drain one update from every client channel.
+		for _, s := range subs {
+			select {
+			case <-s.ch:
+			case <-time.After(2 * time.Second):
+				b.Fatal("timed out waiting for update")
+			}
+		}
+	}
+}
+
+// newBroker is defined in broker_test.go (same package broker_test) and shared
+// across test files.  It returns a broker pre-populated with the stub DS.
+
+// BenchmarkSubscribeUnsubscribe measures the overhead of adding and removing
+// a subscription (the fast path — signal is already active).
+func BenchmarkSubscribeUnsubscribe(b *testing.B) {
+	brk, cancel := newBroker(b)
+	defer cancel()
+
+	ref := broker.SignalRef{DS: "stub", Name: "sine_1hz"}
+
+	// Prime the signal so the fanOut goroutine exists.
+	ch0 := make(chan broker.Update, 256)
+	unsub0, err := brk.Subscribe(ref, ch0)
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer unsub0()
+
+	b.ResetTimer()
+	b.ReportAllocs()
+
+	for range b.N {
+		ch := make(chan broker.Update, 16)
+		unsub, err := brk.Subscribe(ref, ch)
+		if err != nil {
+			b.Fatal(err)
+		}
+		unsub()
+	}
+}