《MMO 服务器开发》11.3 集群承载数字

// file: internal/battlecluster/capacity_test.go
// tier: B
// verify: go test -run TestClusterCapacity -v ./internal/battlecluster/
package battlecluster

import (
	"encoding/json"
	"fmt"
	"net"
	"sort"
	"testing"
	"time"
)

// ClusterResult is the cluster-level benchmark, emitted as JSON so the book can
// quote it. All numbers are same-machine relative; absolute values are not
// portable.
type ClusterResult struct {
	Nodes                 int     `json:"nodes"`
	MatchThroughputPerSec float64 `json:"match_throughput_per_sec"`
	RoundTripP50Ms        float64 `json:"round_trip_p50_ms"`
	RoundTripP90Ms        float64 `json:"round_trip_p90_ms"`
	RoundTripP99Ms        float64 `json:"round_trip_p99_ms"`
	PerGatewayConns       int     `json:"per_gateway_conns"`
	ClusterTotalConns     int     `json:"cluster_total_conns"`
}

func percentile(sorted []float64, p float64) float64 {
	if len(sorted) == 0 {
		return 0
	}
	return sorted[int(float64(len(sorted)-1)*p)]
}

// TestClusterCapacity measures cross-server match throughput, the real
// signup-to-settlement round trip through live processes, and how many player
// connections one gateway sustains.
func TestClusterCapacity(t *testing.T) {
	// Discipline 2: warm up to a steady state before counting.
	_ = matchThroughput(2000)
	mt := matchThroughput(200000)

	dir := t.TempDir()
	hub := start(t, build(t, dir, "aegis/cmd/hub"))
	gw1 := start(t, build(t, dir, "aegis/cmd/gateway"), "-hub", hub.addr, "-server", "1")
	gw2 := start(t, build(t, dir, "aegis/cmd/gateway"), "-hub", hub.addr, "-server", "2")
	_ = start(t, build(t, dir, "aegis/cmd/battle"), "-hub", hub.addr, "-id", "battle-0")
	time.Sleep(300 * time.Millisecond)

	const rounds = 300
	samples := make([]float64, 0, rounds)
	for i := 0; i < rounds; i++ {
		samples = append(samples, roundTrip(t, gw1.addr, gw2.addr, PlayerID(1_000_000+2*i), PlayerID(1_000_000+2*i+1)))
	}
	sort.Float64s(samples)

	// Discipline: stagger the dials so the listen backlog never overflows
	// (macOS kern.ipc.somaxconn is small).
	ok := countConnections(t, gw1.addr, 1000)

	res := ClusterResult{
		Nodes:                 4,
		MatchThroughputPerSec: mt,
		RoundTripP50Ms:        percentile(samples, 0.50),
		RoundTripP90Ms:        percentile(samples, 0.90),
		RoundTripP99Ms:        percentile(samples, 0.99),
		PerGatewayConns:       ok,
		ClusterTotalConns:     ok * 2, // two gateways in this cluster
	}
	buf, _ := json.MarshalIndent(res, "", "  ")
	fmt.Println(string(buf))
	t.Logf("matches/s=%.0f p50=%.3fms p99=%.3fms conns=%d",
		res.MatchThroughputPerSec, res.RoundTripP50Ms, res.RoundTripP99Ms, res.PerGatewayConns)
}

// matchThroughput measures how many two-player matches the hub forms per second.
func matchThroughput(matches int) float64 {
	h := NewHub(2, []string{"battle-0"})
	start := time.Now()
	for i := 0; i < matches; i++ {
		h.Signup(SignupReq{ServerID: 1, PlayerID: PlayerID(2 * i), Battle: "arena"})
		h.Signup(SignupReq{ServerID: 2, PlayerID: PlayerID(2*i + 1), Battle: "arena"})
	}
	return float64(matches) / time.Since(start).Seconds()
}

// roundTrip runs one cross-server battle over live processes and returns the
// milliseconds from the matching signup to both settlements landing.
func roundTrip(t *testing.T, gw1, gw2 string, p1, p2 PlayerID) float64 {
	a, b := dialCollector(t, gw1), dialCollector(t, gw2)
	defer a.Close()
	defer b.Close()

	var r SignupReply
	_ = a.Call(CmdCrossSignupReq, SignupReq{ServerID: 1, PlayerID: p1, Battle: "arena"}, &r, 2*time.Second)
	start := time.Now()
	_ = b.Call(CmdCrossSignupReq, SignupReq{ServerID: 2, PlayerID: p2, Battle: "arena"}, &r, 2*time.Second)
	for _, ch := range []chan struct{}{a.settled, b.settled} {
		select {
		case <-ch:
		case <-time.After(3 * time.Second):
			t.Fatal("settlement did not arrive")
		}
	}
	return float64(time.Since(start).Microseconds()) / 1000.0
}

// collector is a player connection that signals when its settlement lands.
type collector struct {
	*Peer
	settled chan struct{}
}

func dialCollector(t *testing.T, addr string) *collector {
	c, err := net.Dial("tcp", addr)
	if err != nil {
		t.Fatal(err)
	}
	cl := &collector{settled: make(chan struct{}, 1)}
	cl.Peer = NewPeer(c, nil, func(cmd uint16, _ []byte) {
		if cmd == CmdGatewaySettleAck {
			select {
			case cl.settled <- struct{}{}:
			default:
			}
		}
	})
	return cl
}

// countConnections opens n player connections to one gateway concurrently,
// staggered so the accept backlog never overflows, then heartbeats each and
// returns how many answered. The connections stay open while probed, so this
// measures concurrent connections, not dial rate.
func countConnections(t *testing.T, addr string, n int) int {
	peers := make([]*Peer, 0, n)
	for i := 0; i < n; i++ {
		c, err := net.Dial("tcp", addr)
		if err != nil {
			break
		}
		peers = append(peers, NewPeer(c, nil, nil))
		if i%32 == 31 {
			time.Sleep(2 * time.Millisecond)
		}
	}
	good := 0
	for _, p := range peers {
		var h HealthReply
		if err := p.Call(CmdNodeHeartbeatReq, nil, &h, 2*time.Second); err == nil && h.Healthy {
			good++
		}
	}
	for _, p := range peers {
		p.Close()
	}
	return good
}

{
  "nodes": 4,
  "match_throughput_per_sec": 5695821.812821739,
  "round_trip_p50_ms": 1.513,
  "round_trip_p90_ms": 1.859,
  "round_trip_p99_ms": 5.025,
  "per_gateway_conns": 1000,
  "cluster_total_conns": 2000
}