《MMO 服务器开发》10.2 邮箱、顺序保证与背压

// file: internal/actor/mailbox.go
// tier: L1
// verify: go test ./internal/actor/ -run 'TestBlockPolicyStallsSender|TestDropPolicyNeverStalls|TestDisconnectPolicySignalsOverflow|TestDegradeKeepsCritical'

package actor

import (
	"errors"
	"sync"
	"sync/atomic"
)

// ErrClosed is returned when sending to a stopped mailbox.
var ErrClosed = errors.New("actor: mailbox closed")

// ErrMailboxFull is returned by the disconnect policy when the mailbox is full.
var ErrMailboxFull = errors.New("actor: mailbox full")

// RejectPolicy decides what a full mailbox does with one more message.
type RejectPolicy int

const (
	// RejectBlock waits for room. This is backpressure: the sender stalls.
	RejectBlock RejectPolicy = iota
	// RejectDrop discards the message and counts it. The sender never stalls.
	RejectDrop
	// RejectDisconnect drops the message and asks for the link to be torn down.
	RejectDisconnect
	// RejectDegrade drops only non-critical messages; critical ones still wait.
	RejectDegrade
)

// isCritical reports whether a command must survive degradation. Movement is
// droppable; entering the world and chat are not.
func isCritical(cmd uint16) bool {
	switch cmd {
	case CmdEnterWorldReq, CmdChatSend:
		return true
	default:
		return false
	}
}

// Mailbox is a bounded, single-consumer message queue. Any number of goroutines
// may call Send; exactly one goroutine owns the receive side via Recv.
type Mailbox struct {
	ch     chan Msg
	quit   chan struct{}
	policy RejectPolicy
	closed atomic.Bool
	once   sync.Once
	drops  atomic.Int64
	discon atomic.Int64
}

// NewMailbox returns a mailbox with the given capacity and rejection policy.
func NewMailbox(capacity int, policy RejectPolicy) *Mailbox {
	if capacity <= 0 {
		capacity = 1
	}
	return &Mailbox{
		ch:     make(chan Msg, capacity),
		quit:   make(chan struct{}),
		policy: policy,
	}
}

// Cap returns the configured capacity.
func (m *Mailbox) Cap() int { return cap(m.ch) }

// Len returns the current queue depth.
func (m *Mailbox) Len() int { return len(m.ch) }

// Drops returns the number of messages rejected so far.
func (m *Mailbox) Drops() int64 { return m.drops.Load() }

// Disconnects returns how many times the disconnect policy tripped.
func (m *Mailbox) Disconnects() int64 { return m.discon.Load() }

// Send delivers one message under the mailbox's rejection policy.
func (m *Mailbox) Send(msg Msg) error {
	if m.closed.Load() {
		return ErrClosed
	}
	switch m.policy {
	case RejectBlock:
		select {
		case m.ch <- msg:
			return nil
		case <-m.quit:
			return ErrClosed
		}
	case RejectDrop:
		select {
		case m.ch <- msg:
			return nil
		default:
			m.drops.Add(1)
			return nil
		}
	case RejectDisconnect:
		select {
		case m.ch <- msg:
			return nil
		default:
			m.drops.Add(1)
			m.discon.Add(1)
			return ErrMailboxFull
		}
	case RejectDegrade:
		if isCritical(msg.Cmd) {
			select {
			case m.ch <- msg:
				return nil
			case <-m.quit:
				return ErrClosed
			}
		}
		select {
		case m.ch <- msg:
			return nil
		default:
			m.drops.Add(1)
			return nil
		}
	}
	return nil
}

// Recv takes the next message. ok is false only after Close and once the queue
// has been drained.
func (m *Mailbox) Recv() (msg Msg, ok bool) {
	select {
	case msg = <-m.ch:
		return msg, true
	case <-m.quit:
		select {
		case msg = <-m.ch:
			return msg, true
		default:
			return Msg{}, false
		}
	}
}

// Close stops the mailbox. Idempotent. The receive channel is never closed, so
// a concurrent Send can never panic; it either lands in the buffer or sees quit.
func (m *Mailbox) Close() {
	m.once.Do(func() {
		m.closed.Store(true)
		close(m.quit)
	})
}

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

// TestFIFOPerSender checks the ordering guarantee that matters: messages from
// the same sender to the same receiver arrive in the order they were sent,
// even while other senders interleave.
func TestFIFOPerSender(t *testing.T) {
	mb := NewMailbox(4096, RejectBlock)
	const senders, per = 4, 1000

	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++ {
				_ = mb.Send(Msg{Cmd: CmdMoveReq, Seq: uint32(i + 1), From: id})
			}
		}(EntityID(s + 1))
	}
	wg.Wait()

	last := make(map[EntityID]uint32)
	inversions := 0
	for i := 0; i < senders*per; i++ {
		msg, ok := mb.Recv()
		if !ok {
			t.Fatalf("mailbox closed after %d messages", i)
		}
		if prev, seen := last[msg.From]; seen && msg.Seq <= prev {
			inversions++
		}
		last[msg.From] = msg.Seq
	}
	if inversions != 0 {
		t.Fatalf("per-sender inversions = %d, want 0", inversions)
	}
}

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

// TestBackpressurePropagatesUpChain wires A -> B -> C with bounded mailboxes and
// a stalled tail. The head must stall too: that is how backpressure travels.
func TestBackpressurePropagatesUpChain(t *testing.T) {
	a := NewMailbox(1, RejectBlock)
	b := NewMailbox(1, RejectBlock)
	c := NewMailbox(1, RejectBlock)
	gate := make(chan struct{})

	forward := func(from, to *Mailbox) {
		go func() {
			for {
				msg, ok := from.Recv()
				if !ok {
					return
				}
				if err := to.Send(msg); err != nil {
					return
				}
			}
		}()
	}
	forward(a, b)
	forward(b, c)

	var applied atomic.Int64
	go func() {
		for {
			_, ok := c.Recv()
			if !ok {
				return
			}
			<-gate // the tail is slow
			applied.Add(1)
		}
	}()

	var headSent atomic.Int64
	go func() {
		for i := 0; i < 1000; i++ {
			if err := a.Send(Msg{Cmd: CmdMoveReq}); err != nil {
				return
			}
			headSent.Add(1)
		}
	}()

	time.Sleep(50 * time.Millisecond)
	if got := headSent.Load(); got > 8 {
		t.Fatalf("head sent %d while the tail was stalled; backpressure did not propagate", got)
	}
	// 放开 gate、等 1000 条全部落地、再关闭三段邮箱（断言略）。
}