《MMO 服务器开发》6.1 指令投递与执行队列

// file: internal/combat/command.go
// tier: L1
// verify: go test ./internal/combat/ -run 'TestQueue' -v

package combat

import (
	"sort"
	"sync"
)

// EntityID identifies a runtime entity inside one scene/instance. The high 16
// bits carry the scene/instance number and the low 48 bits an in-scene serial;
// the layout itself is not used here, only the total order it gives us.
type EntityID uint64

// Combat-domain command ids, shared by every chapter of volume 2. They mirror
// the wire command numbers, so a decoded frame maps straight onto a Command.
const (
	CmdSkillCastReq  uint16 = 0x0401
	CmdSkillCastResp uint16 = 0x0402
	CmdDamageNotify  uint16 = 0x0403
	CmdBuffSync      uint16 = 0x0404
)

// Command is one gameplay command a client asked for, already decoded from the
// wire and waiting for the scene loop to execute it.
type Command struct {
	Entity  EntityID
	Cmd     uint16
	Seq     uint32
	Payload []byte
}

// Queue is a per-scene command queue. Client commands arrive on many connection
// goroutines but are executed by the single scene loop, so the queue is the one
// place where those goroutines hand work over. It keeps one FIFO per entity and
// refuses any command whose Seq is not strictly increasing for that entity, the
// cheapest possible guard against retries and reordering.
type Queue struct {
	mu        sync.Mutex
	perEntity map[EntityID][]Command
	lastSeq   map[EntityID]uint32
	executed  uint64
	dropped   uint64
}

// NewQueue returns an empty command queue.
func NewQueue() *Queue {
	return &Queue{
		perEntity: make(map[EntityID][]Command),
		lastSeq:   make(map[EntityID]uint32),
	}
}

// Submit enqueues c and reports whether it was accepted. A command is dropped
// when its Seq is not strictly greater than the last accepted Seq for the same
// entity: that rejects a duplicate retry and an out-of-order straggler with a
// single comparison, before it ever reaches the combat pipeline.
func (q *Queue) Submit(c Command) bool {
	q.mu.Lock()
	defer q.mu.Unlock()
	if c.Seq <= q.lastSeq[c.Entity] {
		q.dropped++
		return false
	}
	q.lastSeq[c.Entity] = c.Seq
	q.perEntity[c.Entity] = append(q.perEntity[c.Entity], c)
	return true
}

// DrainTick removes and returns at most budget commands in a deterministic
// order and leaves the rest queued. budget <= 0 means "no cap".
//
// The order is: entities by ascending id, then each entity's commands in
// submission order (which Submit keeps increasing in Seq). Two servers fed the
// same submissions therefore execute the exact same sequence, which is what
// makes replays and cross-server comparisons meaningful.
func (q *Queue) DrainTick(budget int) []Command {
	q.mu.Lock()
	defer q.mu.Unlock()
	ids := make([]EntityID, 0, len(q.perEntity))
	for id, qq := range q.perEntity {
		if len(qq) > 0 {
			ids = append(ids, id)
		}
	}
	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })

	var out []Command
	for _, id := range ids {
		qq := q.perEntity[id]
		for len(qq) > 0 {
			if budget > 0 && len(out) >= budget {
				q.perEntity[id] = qq
				q.executed += uint64(len(out))
				return out
			}
			out = append(out, qq[0])
			qq = qq[1:]
		}
		delete(q.perEntity, id)
	}
	q.executed += uint64(len(out))
	return out
}

// Drain is DrainTick without a per-tick budget.
func (q *Queue) Drain() []Command { return q.DrainTick(0) }

// Pending reports how many commands are still waiting.
func (q *Queue) Pending() int {
	q.mu.Lock()
	defer q.mu.Unlock()
	n := 0
	for _, qq := range q.perEntity {
		n += len(qq)
	}
	return n
}

// Stats returns the lifetime executed and dropped counters.
func (q *Queue) Stats() (executed, dropped uint64) {
	q.mu.Lock()
	defer q.mu.Unlock()
	return q.executed, q.dropped
}

// snippet: internal/combat/command_test.go（节选）

func TestQueueDrainIsDeterministic(t *testing.T) {
	build := func() *Queue {
		q := NewQueue()
		// Submit out of entity order on purpose.
		q.Submit(Command{Entity: 30, Cmd: CmdSkillCastReq, Seq: 1})
		q.Submit(Command{Entity: 10, Cmd: CmdSkillCastReq, Seq: 1})
		q.Submit(Command{Entity: 30, Cmd: CmdDamageNotify, Seq: 2})
		q.Submit(Command{Entity: 20, Cmd: CmdBuffSync, Seq: 1})
		return q
	}
	want := []struct {
		e EntityID
		s uint32
	}{{10, 1}, {20, 1}, {30, 1}, {30, 2}}
	for run := 0; run < 5; run++ {
		got := build().Drain()
		if len(got) != len(want) {
			t.Fatalf("run %d: len = %d, want %d", run, len(got), len(want))
		}
		for i, w := range want {
			if got[i].Entity != w.e || got[i].Seq != w.s {
				t.Fatalf("run %d: [%d] = (%d,%d), want (%d,%d)",
					run, i, got[i].Entity, got[i].Seq, w.e, w.s)
			}
		}
	}
}