1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
package game
import "sync"
// owner is the single execution authority for live game state. Every
// mutation of players, sessions, the Hub membership, runtime maps, and all
// tick callbacks runs on exactly one goroutine (the loop below). Network and
// admin goroutines submit tasks; they never touch live objects directly.
//
// Rules:
// - Game.run blocks the caller until the closure has executed on the
// owner goroutine. Connection goroutines use it per input line, which
// gives natural per-client backpressure without ever blocking another
// client: the owner stays free because network I/O happens outside
// (Session writes go through outbound queues, saves serialize copies).
// - Code already executing on the owner must NOT call Game.run (it would
// deadlock waiting for itself). Owner-executed code calls functions
// directly.
// - Task ordering is FIFO: submitted input lines, disconnect callbacks,
// and admin mutations execute in submission order. Tick callbacks
// serialize against these tasks the same way (the engine's ticker
// goroutine submits each callback), preserving the documented tick
// order relative to subscriber registration IDs.
type owner struct {
mu sync.Mutex
closed bool
tasks chan func()
done chan struct{}
}
const ownerTaskQueue = 4096
func newOwner() *owner {
o := &owner{
tasks: make(chan func(), ownerTaskQueue),
done: make(chan struct{}),
}
go o.loop()
return o
}
func (o *owner) loop() {
defer close(o.done)
for task := range o.tasks {
task()
}
}
// submit enqueues a task; it returns false once the owner has stopped.
// The closed flag is guarded by the mutex so a submit racing with stop
// cannot send on a closed channel.
func (o *owner) submit(fn func()) bool {
o.mu.Lock()
defer o.mu.Unlock()
if o.closed {
return false
}
select {
case o.tasks <- fn:
return true
default:
return false
}
}
func (o *owner) stop() {
o.mu.Lock()
if !o.closed {
o.closed = true
close(o.tasks)
}
o.mu.Unlock()
<-o.done
}
// run executes fn on the owner goroutine and waits for it. When no owner
// exists (synthetic Games in unit tests) fn runs inline on the caller's
// goroutine; production Games always have one, started by New.
func (g *Game) run(fn func()) {
o := g.owner
if o == nil {
fn()
return
}
task := make(chan struct{})
if !o.submit(func() {
fn()
close(task)
}) {
// The owner loop has stopped (graceful shutdown in progress).
// Admissions are already closed, so only stragglers reach this:
// execute inline rather than block forever.
fn()
return
}
<-task
}
// tryRun executes fn on the owner if it is currently idle-draining; it is
// not currently used for correctness paths but exists for future non-blocking
// administrative reads. It returns false when the queue is full.
func (g *Game) tryRun(fn func()) bool {
o := g.owner
if o == nil {
fn()
return true
}
return o.submit(fn)
}
// stopOwner drains and stops the owner loop. Called once during graceful
// shutdown after all entry points have stopped submitting.
func (g *Game) stopOwner() {
g.ownerStop.Do(func() {
if g.owner != nil {
g.owner.stop()
}
})
}
|
