《MMO 服务器开发》2.1 阻塞、非阻塞与 Reactor

// snippet: internal/netio/reactor.go（数据结构与注册入口）
type Reactor struct {
	ready chan Event   // 就绪事件队列，等价于 epoll_wait 的返回数组
	wake  chan struct{} // 叫醒信号，等价于 netpollBreak 的自管道

	mu      sync.Mutex
	regs    map[int]map[EventType][]registration // fd -> type -> 按注册顺序的回调
	timers  map[int]timerArm
	nextSeq uint64

	now func() time.Time

	dispatched int
	timedOut   int
}

// Add 为 (fd, type) 注册回调；同一 (fd, type) 上按注册先后顺序依次触发。
func (r *Reactor) Add(fd int, typ EventType, cb Callback) {
	r.mu.Lock()
	defer r.mu.Unlock()
	m := r.regs[fd]
	if m == nil {
		m = make(map[EventType][]registration)
		r.regs[fd] = m
	}
	r.nextSeq++
	m[typ] = append(m[typ], registration{cb: cb, seq: r.nextSeq})
}

// file: internal/netio/reactor_loop.go
// tier: L1
// verify: go test ./internal/netio/
// 事件循环本体：一个 select 同时等四件事。
//
// 与 reactor.go 同属 package netio，拆成两个文件只是为了正文里
// 每块代码都能完整贴出来，不影响编译。
package netio

import (
	"context"
	"sort"
	"time"
)

// nextWait 返回距离最近一个超时还有多久；没有挂载定时器时 ok=false。
func (r *Reactor) nextWait() (time.Duration, bool) {
	r.mu.Lock()
	defer r.mu.Unlock()
	var best time.Time
	for _, ta := range r.timers {
		if best.IsZero() || ta.deadline.Before(best) {
			best = ta.deadline
		}
	}
	if best.IsZero() {
		return 0, false
	}
	if d := best.Sub(r.now()); d > 0 {
		return d, true
	}
	return 0, true
}

// fireTimeouts 把所有已到期的定时器一次分发掉，
// 同一轮内按 Arm 的先后顺序（seq）分发，保证「先挂的先响」。
func (r *Reactor) fireTimeouts() {
	r.mu.Lock()
	now := r.now()
	var due []timerArm
	for fd, ta := range r.timers {
		if !ta.deadline.After(now) {
			ta.fd = fd
			due = append(due, ta)
			delete(r.timers, fd)
		}
	}
	r.mu.Unlock()

	sort.Slice(due, func(i, j int) bool { return due[i].seq < due[j].seq })
	for _, ta := range due {
		r.dispatch(Event{FD: ta.fd, Type: EventTimeout, At: now})
	}
}

// dispatch 把事件交给该 (fd, type) 上注册的回调，按注册顺序串行执行。
func (r *Reactor) dispatch(ev Event) {
	r.mu.Lock()
	m := r.regs[ev.FD]
	cbs := append([]registration(nil), m[ev.Type]...)
	r.dispatched++
	if ev.Type == EventTimeout {
		r.timedOut++
	}
	r.mu.Unlock()

	for _, reg := range cbs {
		reg.cb(ev)
	}
}

// Run 跑事件循环直到 ctx 取消。一个 select 同时等四件事：
// 就绪事件、最近一个定时器、外部改表后的唤醒、退出信号。
// 真实 netpoller 里「最近定时器」由 epoll_wait 的 timeout 参数承担、
// 「唤醒」由 netpollBreak 的自管道承担，语义完全一致。
func (r *Reactor) Run(ctx context.Context) (int, int) {
	for {
		wait, hasTimer := r.nextWait()
		var (
			timerC <-chan time.Time
			timer  *time.Timer
		)
		if hasTimer {
			timer = time.NewTimer(wait)
			timerC = timer.C
		}
		stopTimer := func() {
			if timer != nil {
				timer.Stop()
			}
		}
		select {
		case ev := <-r.ready:
			stopTimer()
			r.dispatch(ev)
		case <-timerC:
			r.fireTimeouts()
		case <-r.wake:
			stopTimer() // 定时器表被改动，下一轮重算 nextWait
		case <-ctx.Done():
			stopTimer()
			return r.Stats()
		}
	}
}

        ┌─────────────────────────────────────────────────────┐
        │  Run: for { select { ... } }                        │
        └─────────────────────────────────────────────────────┘
             ▲           ▲             ▲             ▲
             │           │             │             │
         ready 通道    timer.C      wake 通道     ctx.Done()
      （epoll_wait  （epoll_wait  （netpollBreak  （退出信号）
        的返回值）    的 timeout）    的自管道）
             │
             ▼
      dispatch(ev) ──► regs[fd][type] ──► cb1 → cb2 → cb3
                                          （按注册顺序，串行执行）

=== RUN   TestDispatchInRegistrationOrder
--- PASS: TestDispatchInRegistrationOrder (0.00s)
=== RUN   TestFDAndTypeIsolation
--- PASS: TestFDAndTypeIsolation (0.00s)
=== RUN   TestTimeoutFiredOnce
--- PASS: TestTimeoutFiredOnce (0.19s)
=== RUN   TestArmOverwrites
--- PASS: TestArmOverwrites (0.25s)
=== RUN   TestDisarmCancels
--- PASS: TestDisarmCancels (0.25s)
=== RUN   TestRemoveDropsEverything
--- PASS: TestRemoveDropsEverything (0.20s)
PASS
ok  	aegis/internal/netio	1.449s

// snippet: src/internal/poll/fd_unix.go:163（FD.Read 主循环，Go 1.26.0）
for {
	n, err := ignoringEINTRIO(syscall.Read, fd.Sysfd, p)
	if err != nil {
		n = 0
		if err == syscall.EAGAIN && fd.pd.pollable() {
			if err = fd.pd.waitRead(fd.isFile); err == nil {
				continue
			}
		}
	}
	err = fd.eofError(n, err)
	return n, err
}

// snippet: src/runtime/netpoll.go:575（netpollblock 的挂起动作，Go 1.26.0）
if waitio || netpollcheckerr(pd, mode) == pollNoError {
	gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceBlockNet, 5)
}

// snippet: src/runtime/netpoll_kqueue.go:90（netpoll 的签名与返回值，Go 1.26.0）
// netpoll checks for ready network connections.
// Returns a list of goroutines that become runnable,
// and a delta to add to netpollWaiters.
func netpoll(delay int64) (gList, int32) {

1765:		list, delta := netpoll(0) // non-blocking
3496:		list, delta := netpoll(0)
3754:		list, delta := netpoll(delay) // block until new work is available
3813:		if list, delta := netpoll(0); !list.empty() {
6571:			list, delta := netpoll(0) // non-blocking - returns list of goroutines