《MMO 服务器开发》1.2 机器人压测平台

// file: internal/loadtest/bot.go
// tier: L2
// verify: go test ./internal/loadtest/
// A bot is a client that walks the whole real path instead of just dialing and
// hanging up: connect, login, enter the scene, move a few times, then send
// heartbeats and time each round trip. The state machine is explicit so a
// failed step is reported as a state, not as an anonymous error. An optional
// ready/start barrier lets the pool finish connecting before any timing starts.
package loadtest

import (
	"bufio"
	"fmt"
	"net"
	"time"
)

// State names one step of the bot state machine, in order.
type State int

const (
	StateConnect State = iota
	StateLogin
	StateEnterScene
	StateMove
	StateHeartbeat
	StateDone
)

// String renders a state for logs and the report.
func (s State) String() string {
	return [...]string{"connect", "login", "enter-scene", "move", "heartbeat", "done"}[s]
}

// BotConfig is one bot's script.
type BotConfig struct {
	Addr    string
	Moves   int           // move steps before switching to heartbeats
	Reqs    int           // timed heartbeats to send
	Timeout time.Duration // per-request read timeout
	Delay   time.Duration // staggered-connect delay applied before dialing
	Hold    time.Duration // stay connected idle for this long after setup
	// Ready/Start form an optional barrier: the bot signals Ready once it has
	// connected and entered the scene, then blocks on Start before timing. This
	// keeps the connect ramp out of the measured throughput window.
	Ready chan<- struct{}
	Start <-chan struct{}
}

// BotResult records where a bot ended up.
type BotResult struct {
	ID    int
	State State
	Err   error
}

// conn is a minimal request/response client over the fixed frame format.
type conn struct {
	c       net.Conn
	r       *bufio.Reader
	seq     uint32
	timeout time.Duration
}

func dial(addr string, timeout time.Duration) (*conn, error) {
	c, err := net.DialTimeout("tcp", addr, timeout)
	if err != nil {
		return nil, err
	}
	return &conn{c: c, r: bufio.NewReader(c), timeout: timeout}, nil
}

func (c *conn) Close() error { return c.c.Close() }

// call sends one request and reads frames until the matching reply arrives. The
// server may push other frames first, so it must not assume the next frame is
// the reply.
func (c *conn) call(cmd, want uint16, payload []byte) ([]byte, error) {
	c.seq++
	seq := c.seq
	if err := c.c.SetWriteDeadline(time.Now().Add(c.timeout)); err != nil {
		return nil, err
	}
	if err := WriteFrame(c.c, Frame{Cmd: cmd, Seq: seq, Payload: payload}); err != nil {
		return nil, err
	}
	for {
		if err := c.c.SetReadDeadline(time.Now().Add(c.timeout)); err != nil {
			return nil, err
		}
		f, err := ReadFrame(c.r)
		if err != nil {
			return nil, err
		}
		if f.Cmd == CmdErrorNotify {
			return nil, fmt.Errorf("server error notify on seq %d", f.Seq)
		}
		if f.Cmd == want && f.Seq == seq {
			return f.Payload, nil
		}
	}
}

// RunBot executes one bot's state machine, recording round-trip times into col.
func RunBot(id int, cfg BotConfig, col *Collector) BotResult {
	signaled := false
	signalReady := func() {
		if !signaled && cfg.Ready != nil {
			signaled = true
			cfg.Ready <- struct{}{}
		}
	}
	defer signalReady()

	if cfg.Delay > 0 {
		time.Sleep(cfg.Delay) // staggered connect: never dial the pool at once
	}
	c, err := dial(cfg.Addr, cfg.Timeout)
	if err != nil {
		col.Fail()
		return BotResult{ID: id, State: StateConnect, Err: err}
	}
	defer c.Close()

	name := fmt.Sprintf("bot-%d", id)
	if _, err := c.call(CmdLoginReq, CmdLoginResp, []byte(name)); err != nil {
		col.Fail()
		return BotResult{ID: id, State: StateLogin, Err: err}
	}
	if _, err := c.call(CmdEnterSceneReq, CmdEnterSceneResp, nil); err != nil {
		col.Fail()
		return BotResult{ID: id, State: StateEnterScene, Err: err}
	}
	col.ConnOK() // a live, logged-in connection now exists
	for i := 0; i < cfg.Moves; i++ {
		if _, err := c.call(CmdMoveReq, CmdMoveResp, []byte{byte(i), byte(i)}); err != nil {
			col.Fail()
			return BotResult{ID: id, State: StateMove, Err: err}
		}
	}
	// Setup done: announce readiness, then wait at the barrier so every bot
	// starts timing at the same instant.
	signalReady()
	if cfg.Start != nil {
		<-cfg.Start
	}
	if cfg.Hold > 0 {
		time.Sleep(cfg.Hold) // hold the connection open without sending
	}
	for i := 0; i < cfg.Reqs; i++ {
		begin := time.Now()
		if _, err := c.call(CmdHeartbeatReq, CmdHeartbeatResp, nil); err != nil {
			col.Fail()
			return BotResult{ID: id, State: StateHeartbeat, Err: err}
		}
		col.Record(float64(time.Since(begin).Microseconds()) / 1000.0)
	}
	return BotResult{ID: id, State: StateDone}
}

// file: internal/loadtest/pool.go
// tier: L2
// verify: go test ./internal/loadtest/
// PoolConfig and RunPool are the concurrency orchestration: they launch N bots
// against one address, spread their connects across a window, optionally wait
// for every bot to connect before releasing them together, then hand back one
// report. Staggering is not optional on macOS, where the default accept backlog
// (kern.ipc.somaxconn) is only 128.
package loadtest

import (
	"sync"
	"time"
)

// PoolConfig describes one run of N bots against one address.
type PoolConfig struct {
	Label   string
	Addr    string
	Conns   int
	Moves   int
	Reqs    int
	Hold    time.Duration // per-bot idle hold, for a connection-ceiling run
	Stagger time.Duration // total window the connects are spread over
	Timeout time.Duration
	// Barrier, when true, holds every bot at a start line until all of them
	// have connected. It is what turns "QPS over the whole run" into a clean
	// steady-state measurement with no connect ramp inside the window.
	Barrier bool
}

// RunPool launches Conns bots staggered across Stagger, waits for all of them
// and returns the aggregated report. Without the stagger a thousand goroutines
// dial at once, overflow the accept queue, and the kernel resets the extras --
// the resulting "failures" would be the test harness, not the server.
func RunPool(cfg PoolConfig) Report {
	col := NewCollector(cfg.Label)
	step := cfg.Stagger / time.Duration(max(cfg.Conns, 1))

	var ready chan struct{}
	var start chan struct{}
	if cfg.Barrier {
		ready = make(chan struct{}, cfg.Conns)
		start = make(chan struct{})
	}

	var wg sync.WaitGroup
	for i := 0; i < cfg.Conns; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			RunBot(i, BotConfig{
				Addr:    cfg.Addr,
				Moves:   cfg.Moves,
				Reqs:    cfg.Reqs,
				Hold:    cfg.Hold,
				Timeout: cfg.Timeout,
				Delay:   time.Duration(i) * step,
				Ready:   ready,
				Start:   start,
			}, col)
		}(i)
	}

	if cfg.Barrier {
		// Wait for every bot to be connected, or give up after a grace period
		// so one dead connection cannot hang the whole run.
		grace := time.After(cfg.Stagger + 10*time.Second)
	wait:
		for i := 0; i < cfg.Conns; i++ {
			select {
			case <-ready:
			case <-grace:
				break wait
			}
		}
		close(start)
	}

	wg.Wait()
	return col.Report(cfg.Conns, cfg.Reqs)
}

=== RUN   TestBotPlatformAgainstLoopback
    platform_test.go:45: conns=50 qps=2132 p50=0.31ms p99=6.58ms err=0.000
--- PASS: TestBotPlatformAgainstLoopback (0.50s)
ok  	aegis/internal/loadtest	1.091s