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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
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
cond *sync.Cond
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{}),
}
o.cond = sync.NewCond(&o.mu)
go o.loop()
return o
}
func (o *owner) loop() {
defer close(o.done)
for task := range o.tasks {
task()
// A queue slot freed up; wake one blocked submitter waiting for
// room (see enqueue).
o.cond.Signal()
}
}
// submit enqueues without blocking. It returns false when the owner has
// stopped or the queue is full; callers that need guaranteed delivery use
// enqueue instead.
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
}
}
// enqueue submits with backpressure: it blocks while the queue is full and
// returns false only when the owner has stopped. A full queue must NEVER
// bypass the owner: running a task on the caller's goroutine while the
// owner keeps executing other tasks breaks the single-owner invariant that
// serializes all live state mutation.
func (o *owner) enqueue(fn func()) bool {
o.mu.Lock()
defer o.mu.Unlock()
for !o.closed {
select {
case o.tasks <- fn:
return true
default:
o.cond.Wait()
}
}
return false
}
func (o *owner) stop() {
o.mu.Lock()
if !o.closed {
o.closed = true
close(o.tasks)
o.cond.Broadcast()
}
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.enqueue(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. A full queue never
// takes this path: enqueue applies backpressure instead.
fn()
return
}
<-task
}
// RunAdminOnOwner synchronously executes one complete editor request. Unlike
// RunOnOwner it rejects work during shutdown and NEVER falls back to executing
// on the HTTP goroutine when the owner has stopped. Call only from outside the
// owner; runtime callbacks within fn must call owner-only methods directly.
func (g *Game) RunAdminOnOwner(fn func()) bool {
if g.isShuttingDown() {
return false
}
accepted := false
task := func() {
if !g.isShuttingDown() {
fn()
accepted = true
}
}
if g.owner == nil {
// Synthetic Games only; production always has an owner.
task()
return accepted
}
done := make(chan struct{})
if !g.owner.enqueue(func() {
task()
close(done)
}) {
return false
}
<-done
return accepted
}
// 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()
}
})
}
|
