notify/signal_benchmark_test.go

package notify

import (
	"errors"
	"net"
	"strconv"
	"sync/atomic"
	"syscall"
	"testing"
	"time"
)

func BenchmarkSignalTCPRoundTrip(b *testing.B) {
	server, addr := startSignalRoundTripServerForBenchmark(b)
	defer func() {
		_ = server.Stop()
	}()

	client := newSignalRoundTripBenchmarkClient(b, addr)
	defer func() {
		_ = client.Stop()
	}()

	payload := []byte("ping")
	b.ReportAllocs()
	b.SetBytes(int64(len(payload) * 2))
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		reply, err := client.SendWait("signal-roundtrip", payload, 5*time.Second)
		if err != nil {
			b.Fatalf("SendWait failed at iter %d: %v", i, err)
		}
		if got, want := string(reply.Value), "ack:ping"; got != want {
			b.Fatalf("reply mismatch at iter %d: got %q want %q", i, got, want)
		}
	}
}

func BenchmarkSignalTCPRoundTripParallel(b *testing.B) {
	server, addr := startSignalRoundTripServerForBenchmark(b)
	defer func() {
		_ = server.Stop()
	}()

	client := newSignalRoundTripBenchmarkClient(b, addr)
	defer func() {
		_ = client.Stop()
	}()

	var seq atomic.Uint64
	b.ReportAllocs()
	b.SetBytes(int64(len("ping-0") * 2))
	b.ResetTimer()
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			id := seq.Add(1)
			payload := []byte("ping-" + strconv.FormatUint(id, 10))
			reply, err := client.SendWait("signal-roundtrip", payload, 5*time.Second)
			if err != nil {
				b.Fatalf("parallel SendWait failed: %v", err)
			}
			want := "ack:" + string(payload)
			if got := string(reply.Value); got != want {
				b.Fatalf("parallel reply mismatch: got %q want %q", got, want)
			}
		}
	})
}

func startSignalRoundTripServerForBenchmark(b *testing.B) (*ServerCommon, string) {
	b.Helper()
	server := NewServer().(*ServerCommon)
	if err := UseModernPSKServer(server, integrationSharedSecret, integrationModernPSKOptions()); err != nil {
		b.Fatalf("UseModernPSKServer failed: %v", err)
	}
	server.SetLink("signal-roundtrip", func(msg *Message) {
		_ = msg.Reply([]byte("ack:" + string(msg.Value)))
	})
	if err := server.Listen("tcp", "127.0.0.1:0"); err != nil {
		if benchmarkListenPermissionDenied(err) {
			b.Skipf("tcp benchmark requires local listen permission: %v", err)
		}
		b.Fatalf("server Listen failed: %v", err)
	}
	return server, signalRoundTripServerAddr(server, "")
}

func benchmarkListenPermissionDenied(err error) bool {
	if err == nil {
		return false
	}
	var opErr *net.OpError
	if errors.As(err, &opErr) {
		err = opErr.Err
	}
	return errors.Is(err, syscall.EPERM) || errors.Is(err, syscall.EACCES)
}

func newSignalRoundTripBenchmarkClient(b *testing.B, addr string) *ClientCommon {
	b.Helper()
	client := NewClient().(*ClientCommon)
	if err := UseModernPSKClient(client, integrationSharedSecret, integrationModernPSKOptions()); err != nil {
		b.Fatalf("UseModernPSKClient failed: %v", err)
	}
	if err := client.Connect("tcp", addr); err != nil {
		b.Fatalf("client Connect failed: %v", err)
	}
	return client
}