《MMO 服务器开发》2.3 热区隔离与背压

// file: internal/profile/isolation.go
// tier: L1
// verify: go test ./internal/profile/ -run TestHotspotIsolation
package profile

// Hotspot names a CPU-heavy stage of a tick that aegis runs on its own worker.
type Hotspot string

const (
	HotspotMovement Hotspot = "movement"
	HotspotAOI      Hotspot = "aoi"
	HotspotCombat   Hotspot = "combat"
)

// IsolatedLoop runs each hotspot on its own goroutine with its own bounded
// queue. A stalled hotspot can only fill its own queue; the others keep
// draining, so a slow AOI broadcast cannot stall movement. This is the
// structural fix for the "one slow phase eats the whole tick" failure — the
// per-hotspot queue is the blast radius of that hotspot's slowness.
type IsolatedLoop struct {
	queues map[Hotspot]*BoundedQueue
}

// NewIsolatedLoop builds a loop whose hotspots each get a queue of the given
// capacity. Movement and AOI drop the oldest under pressure (latest position
// wins); combat rejects, because dropping a damage event corrupts the fight.
func NewIsolatedLoop(capacity int) *IsolatedLoop {
	return &IsolatedLoop{queues: map[Hotspot]*BoundedQueue{
		HotspotMovement: NewBoundedQueue(capacity, OverflowDropOldest),
		HotspotAOI:      NewBoundedQueue(capacity, OverflowDropOldest),
		HotspotCombat:   NewBoundedQueue(capacity, OverflowReject),
	}}
}

// Submit enqueues work for one hotspot.
func (l *IsolatedLoop) Submit(h Hotspot, v any) any { return l.queues[h].Push(v) }

// Queue returns one hotspot's queue, e.g. for metrics.
func (l *IsolatedLoop) Queue(h Hotspot) *BoundedQueue { return l.queues[h] }

// Drain runs fn over up to max items of one hotspot and returns how many it
// processed. Running this on a per-hotspot worker goroutine is what isolates
// the hotspots from each other: a slow fn only slows its own hotspot.
func (l *IsolatedLoop) Drain(h Hotspot, max int, fn func(any)) int {
	q := l.queues[h]
	n := 0
	for n < max {
		v, ok := q.Pop()
		if !ok {
			break
		}
		fn(v)
		n++
	}
	return n
}

// file: internal/profile/backpressure.go
// tier: L1
// verify: go test ./internal/profile/ -run TestBoundedQueue
package profile

import "sync/atomic"

// Overflow is the policy a bounded queue applies when it is full. The whole
// point of naming the policy is that "what happens when we are full" is a
// decision, not an accident.
type Overflow int

const (
	// OverflowReject refuses the newest item and signals the producer to slow
	// down. Correct when every item matters and the producer can shed load.
	OverflowReject Overflow = iota
	// OverflowDropOldest evicts the oldest queued item to make room. Correct
	// for "latest state wins" streams such as AOI snapshots.
	OverflowDropOldest
	// OverflowDegrade evicts the oldest but stores the new item in a cheaper
	// form. Correct when losing detail beats losing the event.
	OverflowDegrade
)

// Degradable is a payload that can be reduced to a cheaper form under pressure,
// e.g. an AOI update that keeps position but drops effects.
type Degradable interface {
	Degrade() any
}

// Degrade reduces v to a cheaper form if it implements Degradable, else returns
// v unchanged.
func Degrade(v any) any {
	if d, ok := v.(Degradable); ok {
		return d.Degrade()
	}
	return v
}

// Counters records what a bounded queue did under pressure. Every field is
// atomic so a metrics exporter can scrape it while the queue is live.
type Counters struct {
	Enqueued atomic.Uint64
	Dequeued atomic.Uint64
	Rejected atomic.Uint64
	Dropped  atomic.Uint64
	Degraded atomic.Uint64
}

// BoundedQueue is a fixed-capacity queue with an explicit overflow policy. An
// unbounded queue does not remove backpressure — it converts a throughput
// problem into a memory-exhaustion one, and moves the failure from "slow" to
// "OOM-killed". Capacity is therefore a required constructor argument.
//
// A BoundedQueue is not internally synchronized: it is owned by a single
// goroutine (the hotspot's worker), which is exactly how section 2.3 wires it.
// The Counters are atomic so a metrics scraper on another goroutine can read
// them without a lock, but Push and Pop themselves must not run concurrently.
type BoundedQueue struct {
	items  []any
	head   int
	size   int
	policy Overflow
	count  Counters
}

// NewBoundedQueue returns a queue with the given capacity and overflow policy.
func NewBoundedQueue(capacity int, policy Overflow) *BoundedQueue {
	if capacity <= 0 {
		panic("profile: queue capacity must be positive")
	}
	return &BoundedQueue{items: make([]any, capacity), policy: policy}
}

// Len is the number of queued items.
func (q *BoundedQueue) Len() int { return q.size }

// Cap is the fixed capacity.
func (q *BoundedQueue) Cap() int { return len(q.items) }

// Counters exposes the queue's live counters.
func (q *BoundedQueue) Counters() *Counters { return &q.count }

// Push enqueues v, applying the overflow policy when full. It returns the value
// actually stored, or nil when the item was rejected.
func (q *BoundedQueue) Push(v any) any {
	if q.size < len(q.items) {
		q.items[(q.head+q.size)%len(q.items)] = v
		q.size++
		q.count.Enqueued.Add(1)
		return v
	}
	switch q.policy {
	case OverflowDropOldest:
		q.items[q.head] = v // overwrite the oldest slot with the newest value
		q.head = (q.head + 1) % len(q.items)
		q.count.Dropped.Add(1)
		q.count.Enqueued.Add(1)
		return v
	case OverflowDegrade:
		q.items[q.head] = Degrade(v)
		q.head = (q.head + 1) % len(q.items)
		q.count.Degraded.Add(1)
		q.count.Enqueued.Add(1)
		return q.items[(q.head+q.size-1)%len(q.items)]
	default: // OverflowReject
		q.count.Rejected.Add(1)
		return nil
	}
}

// Pop removes and returns the oldest item.
func (q *BoundedQueue) Pop() (any, bool) {
	if q.size == 0 {
		return nil, false
	}
	v := q.items[q.head]
	q.items[q.head] = nil
	q.head = (q.head + 1) % len(q.items)
	q.size--
	q.count.Dequeued.Add(1)
	return v, true
}

=== RUN   TestHotspotIsolation
--- PASS: TestHotspotIsolation (0.00s)
=== RUN   TestHotspotQueueOverflowIsLocal
--- PASS: TestHotspotQueueOverflowIsLocal (0.00s)
=== RUN   TestCombatRejectsRatherThanDrops
--- PASS: TestCombatRejectsRatherThanDrops (0.00s)
PASS
ok  	aegis/internal/profile	0.569s

=== RUN   TestBoundedQueueReject
--- PASS: TestBoundedQueueReject (0.00s)
=== RUN   TestBoundedQueueDropOldest
--- PASS: TestBoundedQueueDropOldest (0.00s)
=== RUN   TestBoundedQueueDegrade
--- PASS: TestBoundedQueueDegrade (0.00s)
=== RUN   TestBoundedQueueCapacityIsMandatory
--- PASS: TestBoundedQueueCapacityIsMandatory (0.00s)
=== RUN   TestCountersAccumulate
--- PASS: TestCountersAccumulate (0.00s)
PASS
ok  	aegis/internal/profile	0.550s