《MMO 服务器开发》5.2 分配 Broker 与容量评估

// file: internal/scheduler/capacity.go
// tier: L1
// verify: go test ./internal/scheduler/ -run TestFits
package scheduler

// NodeMetrics 是一台机器当前的可观测负载。四个维度：CPU、内存、连接数、
// 实例槽位——任何一个到顶，这台机器就再开不出实例，所以判据必须是「四
// 个都不越界」，而不是只看 CPU。
type NodeMetrics struct {
	Node     NodeID
	CPUCores float64 // 机器总核数
	CPUUsed  float64 // 已占用核数（最近窗口均值）
	MemBytes uint64  // 机器总内存
	MemUsed  uint64  // 已占用内存
	Conns    int     // 当前连接数
	MaxConns int     // 连接数上限（fd 上限与内存共同决定）
	Slots    int     // 已开实例数
	MaxSlots int     // 实例槽位上限
	Draining bool    // 是否在下线中（不再接新实例）
}

// Demand 是一个实例的资源需求。
type Demand struct {
	Kind     Kind
	CPUCores float64
	MemBytes uint64
	Conns    int
	Slots    int
}

// Fits 判断节点能否再放下一个 demand；不能时返回第一个不满足的维度名，
// 这个维度名会原样进日志，是「为什么开不出来」的第一手线索。
func (n NodeMetrics) Fits(d Demand) (bool, string) {
	switch {
	case n.Draining:
		return false, "draining"
	case n.Slots+d.Slots > n.MaxSlots:
		return false, "slots"
	case n.Conns+d.Conns > n.MaxConns:
		return false, "conns"
	case n.CPUUsed+d.CPUCores > n.CPUCores:
		return false, "cpu"
	case n.MemUsed+d.MemBytes > n.MemBytes:
		return false, "mem"
	}
	return true, ""
}

// Headroom 返回四个维度里「剩余比例」的最小值，越大越空闲。取最小值而
// 不是平均值，是为了避免把实例塞进一台 CPU 空闲、但槽位只剩一个的机器：
// 真正卡住分配的永远是那个最紧的维度。
func (n NodeMetrics) Headroom() float64 {
	cpu := (n.CPUCores - n.CPUUsed) / n.CPUCores
	mem := 1 - float64(n.MemUsed)/float64(n.MemBytes)
	conn := float64(n.MaxConns-n.Conns) / float64(n.MaxConns)
	slot := float64(n.MaxSlots-n.Slots) / float64(n.MaxSlots)
	return minOf4(cpu, mem, conn, slot)
}

func minOf4(a, b, c, d float64) float64 {
	m := a
	if b < m {
		m = b
	}
	if c < m {
		m = c
	}
	if d < m {
		m = d
	}
	return m
}

// file: internal/scheduler/broker.go
// tier: L1
// verify: go test ./internal/scheduler/ -run TestBroker
package scheduler

import (
	"context"
	"errors"
	"sync"
	"time"
)

// ErrNoCapacity 表示在等待窗口内始终没有节点放得下这个需求。
var ErrNoCapacity = errors.New("no node has capacity")

// Broker 在候选节点里挑一台放下实例；当下没有节点放得下时，请求进入等待
// 队列，直到有节点腾出容量或超时。它同时是「放哪儿」和「放不下怎么办」的答案。
type Broker struct {
	mu    sync.Mutex
	nodes map[NodeID]NodeMetrics
	queue []*waiter
}

type waiter struct {
	demand Demand
	ready  chan NodeID
}

func NewBroker() *Broker { return &Broker{nodes: make(map[NodeID]NodeMetrics)} }

// Update 上报一台节点的最新负载，然后尝试唤醒排队的请求。
func (b *Broker) Update(m NodeMetrics) {
	b.mu.Lock()
	b.nodes[m.Node] = m
	b.mu.Unlock()
	b.pump()
}

// Allocate 先立即选一台；选不到就进队列等，等到 wait 为止。
func (b *Broker) Allocate(ctx context.Context, d Demand, wait time.Duration) (NodeID, error) {
	if node, ok := b.pick(d); ok {
		return node, nil
	}
	w := &waiter{demand: d, ready: make(chan NodeID, 1)}
	b.mu.Lock()
	b.queue = append(b.queue, w)
	b.mu.Unlock()

	timer := time.NewTimer(wait)
	defer timer.Stop()
	select {
	case node := <-w.ready:
		return node, nil
	case <-timer.C:
		b.drop(w)
		return "", ErrNoCapacity
	case <-ctx.Done():
		b.drop(w)
		return "", ctx.Err()
	}
}

// pick 在能放下的节点里挑余量最大的那台，并立即预留资源——预留是关键，
// 否则两次并发分配会同时选中同一台机器，把它的槽位开爆。
func (b *Broker) pick(d Demand) (NodeID, bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	var best NodeID
	var bestH float64
	found := false
	for id, m := range b.nodes {
		if ok, _ := m.Fits(d); !ok {
			continue
		}
		if h := m.Headroom(); !found || h > bestH {
			best, bestH, found = id, h, true
		}
	}
	if !found {
		return "", false
	}
	b.reserveLocked(best, d)
	return best, true
}

func (b *Broker) reserveLocked(id NodeID, d Demand) {
	m := b.nodes[id]
	m.CPUUsed += d.CPUCores
	m.MemUsed += d.MemBytes
	m.Conns += d.Conns
	m.Slots += d.Slots
	b.nodes[id] = m
}

// pump 按 FIFO 依次满足队列里的请求；队首放不下就停在它这里，不越位——
// 先到先服务，避免后来的小请求把队首的大请求永远饿死。
func (b *Broker) pump() {
	for {
		b.mu.Lock()
		if len(b.queue) == 0 {
			b.mu.Unlock()
			return
		}
		w := b.queue[0]
		b.mu.Unlock()

		node, ok := b.pick(w.demand)
		if !ok {
			return
		}
		b.drop(w)
		w.ready <- node
	}
}

func (b *Broker) drop(w *waiter) {
	b.mu.Lock()
	defer b.mu.Unlock()
	for i, q := range b.queue {
		if q == w {
			b.queue = append(b.queue[:i], b.queue[i+1:]...)
			return
		}
	}
}

// file: internal/scheduler/bench_test.go
// tier: B
// verify: go test -run=^$ -bench . ./internal/scheduler/
package scheduler

import (
	"strconv"
	"testing"
)

func benchNodes(n int) []NodeMetrics {
	out := make([]NodeMetrics, n)
	for i := range out {
		out[i] = NodeMetrics{
			Node:     NodeID("scene-" + strconv.Itoa(i)),
			CPUCores: 16, CPUUsed: float64(i%12) + 1,
			MemBytes: 64 << 30, MemUsed: uint64(i%40) << 30,
			Conns: (i % 80) * 100, MaxConns: 10000,
			Slots: i % 6, MaxSlots: 8,
		}
	}
	return out
}

// BenchmarkFitsAndHeadroom 量「判断一个节点放不放得下」的成本。
func BenchmarkFitsAndHeadroom(b *testing.B) {
	nodes := benchNodes(1000)
	d := battleDemand()
	b.ReportAllocs()
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		m := nodes[i%len(nodes)]
		if ok, _ := m.Fits(d); ok {
			_ = m.Headroom()
		}
	}
}

// BenchmarkBrokerPick 量「在 1000 个候选节点里挑一台」的成本。
func BenchmarkBrokerPick(b *testing.B) {
	nodes := benchNodes(1000)
	d := battleDemand()
	b.ReportAllocs()
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		b.StopTimer()
		br := NewBroker()
		for _, m := range nodes {
			br.nodes[m.Node] = m
		}
		b.StartTimer()
		_, _ = br.pick(d)
	}
}