《MMO 服务器开发》10.1 消息驱动的实体

// file: internal/actor/entity.go
// tier: L1
// verify: go test ./internal/actor/ -run TestSingleSenderFIFO

package actor

// Command numbers reused verbatim from the protocol chapter; the actor layer
// never invents its own wire values.
const (
	CmdEnterWorldReq uint16 = 0x0101
	CmdMoveReq       uint16 = 0x0201
	CmdChatSend      uint16 = 0x0301
)

// EntityID is a world-unique entity identifier.
type EntityID uint64

// Msg is one message handed to an entity. It carries values only: never a
// pointer into another entity's state.
type Msg struct {
	Cmd     uint16
	Seq     uint32
	From    EntityID
	Payload []byte
}

// Entity is a player or NPC. Every field below is touched by exactly one
// goroutine — the one running this entity's loop — so none of them needs a lock.
type Entity struct {
	ID      EntityID
	X       int32
	Y       int32
	HP      int
	Applied int

	lastSeq    map[EntityID]uint32
	outOfOrder int
}

func newEntity(id EntityID) *Entity {
	return &Entity{ID: id, HP: 100, lastSeq: make(map[EntityID]uint32)}
}

// OutOfOrder reports how many messages arrived with a non-increasing sequence
// number from a sender that had already been seen.
func (e *Entity) OutOfOrder() int { return e.outOfOrder }

func (e *Entity) apply(msg Msg) {
	if msg.From != 0 {
		if last, ok := e.lastSeq[msg.From]; ok && msg.Seq <= last {
			e.outOfOrder++
		}
		e.lastSeq[msg.From] = msg.Seq
	}
	if msg.Cmd == CmdMoveReq && len(msg.Payload) >= 4 {
		dx := int32(int16(uint16(msg.Payload[0]) | uint16(msg.Payload[1])<<8))
		dy := int32(int16(uint16(msg.Payload[2]) | uint16(msg.Payload[3])<<8))
		e.X += dx
		e.Y += dy
	}
	e.Applied++
}

// file: internal/actor/handle.go
// tier: L1
// verify: go test ./internal/actor/ -run 'TestSingleSenderFIFO|TestRouteAfterStopIsClosed'

package actor

import "sync"

// Handle is the only reference other goroutines hold to an entity. It exposes a
// Send method and nothing else, so nobody outside can touch Entity fields.
type Handle struct {
	id   EntityID
	mb   *Mailbox
	e    *Entity
	done chan struct{}
}

// ID returns the entity id behind the handle.
func (h *Handle) ID() EntityID { return h.id }

// Send routes a message into the entity's mailbox.
func (h *Handle) Send(msg Msg) error { return h.mb.Send(msg) }

// Done is closed after the entity loop has drained and exited.
func (h *Handle) Done() <-chan struct{} { return h.done }

// Stop asks the loop to finish; wait on Done for it to drain.
func (h *Handle) Stop() { h.mb.Close() }

// Snapshot returns a copy of the entity state. It is safe only after Done has
// closed, when the loop is guaranteed to have stopped and the happens-before
// edge through Done protects the read.
func (h *Handle) Snapshot() Entity { return *h.e }

// Registry is the router: it maps an entity id to its handle. It is the only
// shared, lock-protected structure in the runtime, and it is read-mostly.
type Registry struct {
	mu     sync.RWMutex
	m      map[EntityID]*Handle
	cap    int
	policy RejectPolicy
}

// NewRegistry returns a registry that spawns entities with the given mailbox
// capacity and rejection policy.
func NewRegistry(capacity int, policy RejectPolicy) *Registry {
	return &Registry{m: make(map[EntityID]*Handle), cap: capacity, policy: policy}
}

// Spawn starts an entity with its own goroutine and returns its handle.
func (r *Registry) Spawn(id EntityID) *Handle {
	e := newEntity(id)
	h := &Handle{id: id, mb: NewMailbox(r.cap, r.policy), e: e, done: make(chan struct{})}
	r.mu.Lock()
	r.m[id] = h
	r.mu.Unlock()
	go func() {
		defer close(h.done)
		for {
			msg, ok := h.mb.Recv()
			if !ok {
				return
			}
			e.apply(msg)
		}
	}()
	return h
}

// Lookup returns the handle for id, if the entity is alive.
func (r *Registry) Lookup(id EntityID) (*Handle, bool) {
	r.mu.RLock()
	h, ok := r.m[id]
	r.mu.RUnlock()
	return h, ok
}

// Route resolves the handle through the table and sends msg to it.
func (r *Registry) Route(id EntityID, msg Msg) error {
	h, ok := r.Lookup(id)
	if !ok {
		return ErrClosed
	}
	return h.Send(msg)
}

// Stop stops the entity, waits for it to drain, and removes it from the table.
func (r *Registry) Stop(id EntityID) {
	r.mu.Lock()
	h, ok := r.m[id]
	delete(r.m, id)
	r.mu.Unlock()
	if ok {
		h.Stop()
		<-h.done
	}
}

// snippet: internal/actor/entity_test.go（节选）

// TestConcurrentSendersNoRace hammers one entity from many goroutines. Run with
// -race it proves that the entity fields need no lock: only the loop touches
// them. Each sender's own order is still preserved.
func TestConcurrentSendersNoRace(t *testing.T) {
	r := NewRegistry(64, RejectBlock)
	h := r.Spawn(7)
	const senders, per = 8, 2000

	var wg sync.WaitGroup
	for s := 0; s < senders; s++ {
		wg.Add(1)
		go func(id EntityID) {
			defer wg.Done()
			for i := 0; i < per; i++ {
				_ = h.Send(Msg{Cmd: CmdMoveReq, Seq: uint32(i + 1), From: id, Payload: movePayload(1, 1)})
			}
		}(EntityID(s + 1))
	}
	wg.Wait()
	r.Stop(7)

	s := h.Snapshot()
	if s.Applied != senders*per {
		t.Fatalf("applied = %d, want %d", s.Applied, senders*per)
	}
	if s.OutOfOrder() != 0 {
		t.Fatalf("out-of-order = %d, want 0", s.OutOfOrder())
	}
	if s.X != senders*per || s.Y != senders*per {
		t.Fatalf("position = (%d,%d), want (%d,%d)", s.X, s.Y, senders*per, senders*per)
	}
}