《MMO 服务器开发》2.3 运行时所有权与迁移

// file: internal/entity/owner.go
// tier: L1
// verify: go test ./internal/entity/
package entity

import "sync"

// Owner 标识一个「运行时的所有者」——通常是一个逻辑线程/worker，
// 或一个场景实例。核心不变式：实体在任意时刻只被一个 Owner 独占，
// 只有 Owner 能写它的状态，别的 Owner 想改只能发消息。
type Owner uint32

// Ownership 记录「每个实体归谁」。它只回答两个问题：谁在写这个实体、
// 一条消息该路由给谁。它不持有实体本身。
type Ownership struct {
	mu     sync.RWMutex
	home   map[EntityID]Owner
	worker map[Owner][]EntityID // 反向索引，便于按 Owner 批量枚举
}

// NewOwnership 创建空的归属表。
func NewOwnership() *Ownership {
	return &Ownership{
		home:   make(map[EntityID]Owner),
		worker: make(map[Owner][]EntityID),
	}
}

// Assign 把实体 id 的归属设为 ow。重复 Assign 会先摘掉旧归属。
func (o *Ownership) Assign(id EntityID, ow Owner) {
	o.mu.Lock()
	defer o.mu.Unlock()
	if old, ok := o.home[id]; ok {
		o.detach(old, id)
	}
	o.home[id] = ow
	o.worker[ow] = append(o.worker[ow], id)
}

// OwnerOf 返回实体当前的归属。
func (o *Ownership) OwnerOf(id EntityID) (Owner, bool) {
	o.mu.RLock()
	defer o.mu.RUnlock()
	ow, ok := o.home[id]
	return ow, ok
}

// Retire 从归属表里摘除实体（销毁时调用）。
func (o *Ownership) Retire(id EntityID) {
	o.mu.Lock()
	defer o.mu.Unlock()
	if old, ok := o.home[id]; ok {
		o.detach(old, id)
		delete(o.home, id)
	}
}

// Count 返回某个 Owner 名下当前持有的实体数。
func (o *Ownership) Count(ow Owner) int {
	o.mu.RLock()
	defer o.mu.RUnlock()
	return len(o.worker[ow])
}

// detach 从 ow 的反向列表里摘掉 id（调用方必须已持写锁）。
func (o *Ownership) detach(ow Owner, id EntityID) {
	list := o.worker[ow]
	for i, v := range list {
		if v == id {
			list[i] = list[len(list)-1]
			o.worker[ow] = list[:len(list)-1]
			break
		}
	}
	if len(o.worker[ow]) == 0 {
		delete(o.worker, ow)
	}
}

① 冻结   源侧停止处理该实体的新消息，状态定格
② 传输   把实体状态序列化、搬到目标侧
③ 接管   目标侧加载完成，开始处理消息

// file: internal/entity/migrate.go
// tier: L1
// verify: go test ./internal/entity/
package entity

import (
	"errors"
	"sync"
)

var (
	// ErrNotOwner 表示发起迁移的 from 并不是实体当前的归属方。
	ErrNotOwner = errors.New("entity: not owned by source")
	// ErrMigrating 表示该实体已有一笔迁移在进行中。
	ErrMigrating = errors.New("entity: migration already in progress")
	// ErrNoMigration 表示提交了一个不存在的迁移。
	ErrNoMigration = errors.New("entity: no migration in progress")
)

// MigrationPhase 描述一次迁移走到哪一步。
type MigrationPhase uint8

const (
	MigrateFrozen MigrationPhase = iota // 源侧已冻结，不再接受新写
	MigrateDone                         // 目标侧已接管
)

// Message 是迁移期间被暂存的玩法消息。
type Message struct {
	Cmd     uint16
	Payload []byte
}

// Migration 是一次实体迁移的账本。
type Migration struct {
	ID    EntityID
	From  Owner
	To    Owner
	Phase MigrationPhase

	buffer []Message // 冻结期间到达的消息，落地后按序重放
}

// Migrator 实现「冻结—传输—接管」三段式迁移，全程不让两个 Owner
// 同时写同一个实体。它是 2.3 里跨场景迁移的服务端骨架。
type Migrator struct {
	own    *Ownership
	mu     sync.Mutex
	active map[EntityID]*Migration
}

// NewMigrator 基于一张归属表创建迁移器。
func NewMigrator(own *Ownership) *Migrator {
	return &Migrator{own: own, active: make(map[EntityID]*Migration)}
}

// Begin 从 from 冻结实体 id 并向 to 发起迁移。它先把归属切到 to
// （让新消息直接路由到目标侧），再登记一笔冻结中的迁移。
func (m *Migrator) Begin(id EntityID, from, to Owner) (*Migration, error) {
	cur, ok := m.own.OwnerOf(id)
	if !ok || cur != from {
		return nil, ErrNotOwner
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if _, busy := m.active[id]; busy {
		return nil, ErrMigrating
	}
	mig := &Migration{ID: id, From: from, To: to, Phase: MigrateFrozen}
	m.active[id] = mig
	m.own.Assign(id, to) // 归属即刻切到目标侧
	return mig, nil
}

// Deliver 是迁移期间的消息入口：迁移中暂存并返回 true；没有迁移
// 在跑则返回 false，调用方应直接投递。暂存保证「冻结那一刻在飞的
// 消息」不会丢，也不会被源侧处理两遍。
func (m *Migrator) Deliver(id EntityID, msg Message) bool {
	m.mu.Lock()
	defer m.mu.Unlock()
	mig, ok := m.active[id]
	if !ok {
		return false
	}
	mig.buffer = append(mig.buffer, msg)
	return true
}

// Commit 标记传输完成、目标侧接管，并返回需要重放的缓冲消息。
func (m *Migrator) Commit(id EntityID) ([]Message, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	mig, ok := m.active[id]
	if !ok {
		return nil, ErrNoMigration
	}
	mig.Phase = MigrateDone
	delete(m.active, id)
	replay := mig.buffer
	mig.buffer = nil
	return replay, nil
}

// Active 返回当前进行中的迁移笔数。
func (m *Migrator) Active() int {
	m.mu.Lock()
	defer m.mu.Unlock()
	return len(m.active)
}

// file: internal/entity/route.go
// tier: L1
// verify: go test ./internal/entity/
package entity

// Router 是消息入口处的分流器：迁移中的实体把消息交给 Migrator 暂存，
// 其余消息按归属路由到对应 Owner。它把「迁移期间消息不丢不重」这条
// 规则固化成一个可测试的函数，是 2.3.3 的落地形态。
type Router struct {
	own *Ownership
	mig *Migrator

	delivered map[Owner][]Message // 已直投出去的消息，便于测试断言
}

// NewRouter 绑定归属表与迁移器。
func NewRouter(own *Ownership, mig *Migrator) *Router {
	return &Router{own: own, mig: mig, delivered: make(map[Owner][]Message)}
}

// Route 把一条消息投给实体 id 的当前归属方。若该实体正在迁移，消息
// 被暂存并返回 (0, true)；否则按归属直投并返回 (owner, false)。实体
// 已不在归属表里时返回 (0, false) 且不投递。
func (rt *Router) Route(id EntityID, msg Message) (Owner, bool) {
	if rt.mig.Deliver(id, msg) {
		return 0, true
	}
	ow, ok := rt.own.OwnerOf(id)
	if !ok {
		return 0, false
	}
	rt.delivered[ow] = append(rt.delivered[ow], msg)
	return ow, false
}

// Delivered 返回直接路由给 ow 的消息（测试用）。
func (rt *Router) Delivered(ow Owner) []Message { return rt.delivered[ow] }

=== RUN   TestOwnershipAssignRetire
--- PASS: TestOwnershipAssignRetire (0.00s)
=== RUN   TestMigrationHandshake
--- PASS: TestMigrationHandshake (0.00s)
=== RUN   TestBeginRequiresOwnership
--- PASS: TestBeginRequiresOwnership (0.00s)
=== RUN   TestBeginTwiceFails
--- PASS: TestBeginTwiceFails (0.00s)
=== RUN   TestRouterDuringMigration
--- PASS: TestRouterDuringMigration (0.00s)
=== RUN   TestRouterUnknownEntity
--- PASS: TestRouterUnknownEntity (0.00s)
PASS
ok  	aegis/internal/entity	0.220s