1package core
2
3import (
4 "context"
5 "sync"
6 "time"
7
8 "github.com/bnema/neferwl/internal/ports"
9)
10
11// outbox carries commands from the owner goroutine to the sender goroutine
12// without ever blocking the owner. It is a transport queue, not window
13// state: the mutex guards only the slices. Commands leave in the order they
14// were queued; the only merge is of adjacent pointer motions (see push).
15//
16// The wayland display goroutine can be busy for a while. Sending from the
17// owner would stall keyboard, scenes and frame wakes behind it, so the owner
18// queues and the sender goroutine does the blocking send.
19type outbox struct {
20 // wake is signalled (never blocking) after each push.
21 wake chan struct{}
22
23 mu sync.Mutex
24 queue, spare []outCmd
25 // busy is set while the sender holds a batch it has not finished
26 // sending: a command pushed then must queue behind it.
27 busy bool
28 // pushed and sent count the entries queued and delivered (a merge adds
29 // none); progress is signalled (never blocking) at each delivery, for
30 // flush. The owner is its only waiter, so one pending signal is enough.
31 pushed, sent uint64
32 progress chan struct{}
33 // running is set while a sender (run) owns delivery.
34 running bool
35 // sendMu is held by whoever takes and delivers a batch, so batches
36 // never interleave (the sender, or flush without one).
37 sendMu sync.Mutex
38 // stats are guarded by mu, reset by takeStats. Their times come from
39 // the system clock, not core's: they measure real waits, and the sender
40 // must not step a test clock.
41 stats CommandStats
42}
43
44// CommandStats tell whether the display holds up core's commands: how many
45// queued instead of going straight out, how long they waited, and how long
46// core itself waited (flush). They are reset when read.
47type CommandStats struct {
48 // Direct and Queued count the commands sent at once and queued; Merged
49 // counts motions folded into a queued one.
50 Direct, Queued, Merged int
51 // MaxBacklog is the longest queue seen.
52 MaxBacklog uint64
53 // MaxQueueWait is the longest time a queued command took to reach the
54 // channel; MaxFlushWait the longest time core waited in flush (a spawn's
55 // SlotsPending, or a full outbox).
56 MaxQueueWait, MaxFlushWait time.Duration
57}
58
59// takeStats returns the counters and starts over.
60func (o *outbox) takeStats() CommandStats {
61 o.mu.Lock()
62 defer o.mu.Unlock()
63 s := o.stats
64 o.stats = CommandStats{}
65 return s
66}
67
68// outCmd is a queued command. A pointer motion is kept unboxed so that a
69// merge boxes the sum once, when it leaves, not at every merge.
70type outCmd struct {
71 cmd ports.ClientCommand
72 // state is the owner envelope; secured is false without a security gate.
73 state ports.SecurityState
74 secured bool
75 // motion is cmd when it is a pointer motion. merged means cmd is stale
76 // and motion carries the sum.
77 motion ports.PointerMotionTo
78 isMotion bool
79 merged bool
80 // at is when it queued; a merge keeps the first motion's.
81 at time.Time
82}
83
84// outboxLimit bounds the commands waiting for the display. Motion merges, so
85// only a display stalled for a long while reaches it; then core waits for
86// room as a full channel made it wait before, rather than grow without
87// bound or drop order-sensitive input.
88const outboxLimit = 4096
89
90// backlog is the number of queued commands not yet delivered.
91func (o *outbox) backlog() uint64 {
92 o.mu.Lock()
93 defer o.mu.Unlock()
94 return o.pushed - o.sent
95}
96
97func newOutbox() outbox {
98 return outbox{wake: make(chan struct{}, 1), progress: make(chan struct{}, 1)}
99}
100
101// command returns what the sender delivers.
102func (o *outCmd) command() ports.ClientCommand {
103 v := o.cmd
104 if o.merged {
105 v = o.motion
106 }
107 if o.secured {
108 return ports.SecurityCommand{State: o.state, Command: v}
109 }
110 return v
111}
112
113// push delivers v, wrapped in an envelope for state when secured. With
114// nothing queued or in flight and room in out it sends at once, which keeps
115// the display's latency when it keeps up; otherwise it queues v. A pointer
116// motion for the same window and envelope as the one queued last replaces
117// it: the position is the latest and the relative deltas add up, so locked
118// pointers (games) lose no movement. Nothing else is merged or reordered:
119// buttons, focus, keys and gestures depend on their order.
120//
121// It returns the entry's sequence number, for flush.
122func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, state ports.SecurityState, secured bool) uint64 {
123 e := outCmd{cmd: v, state: state, secured: secured}
124 e.motion, e.isMotion = v.(ports.PointerMotionTo)
125 o.mu.Lock()
126 // A full channel would make the attempt fail after building the
127 // envelope; queue at once instead. The length is only a hint: queued
128 // delivery keeps the order either way.
129 if !o.busy && len(o.queue) == 0 && (cap(out) == 0 || len(out) < cap(out)) {
130 // A send that cannot block is safe under the lock.
131 select {
132 case out <- e.command():
133 o.pushed++
134 o.sent = o.pushed
135 o.stats.Direct++
136 seq := o.pushed
137 o.mu.Unlock()
138 return seq
139 default:
140 }
141 }
142 // Only a queued command pays for the clock read.
143 e.at = time.Now()
144 if n := len(o.queue); n > 0 && e.isMotion {
145 last := &o.queue[n-1]
146 if last.isMotion && last.motion.ID == e.motion.ID && last.secured == secured && last.state == state {
147 e.motion.DX += last.motion.DX
148 e.motion.DY += last.motion.DY
149 e.motion.UnaccelDX += last.motion.UnaccelDX
150 e.motion.UnaccelDY += last.motion.UnaccelDY
151 e.merged = true
152 e.at = last.at
153 *last = e
154 o.stats.Merged++
155 seq := o.pushed
156 o.mu.Unlock()
157 return seq
158 }
159 }
160 o.queue = append(o.queue, e)
161 o.pushed++
162 o.stats.Queued++
163 o.stats.MaxBacklog = max(o.stats.MaxBacklog, o.pushed-o.sent)
164 seq := o.pushed
165 o.mu.Unlock()
166 select {
167 case o.wake <- struct{}{}:
168 default:
169 }
170 return seq
171}
172
173// flush waits until the entry seq is in out, or ctx ends. It is for the
174// rare command another channel depends on (SlotsPending before a spawn);
175// everything else never waits. Without a running sender (core driven
176// without Run), it delivers the queue itself, in order.
177func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error {
178 var start time.Time
179 defer func() {
180 if !start.IsZero() {
181 w := time.Since(start)
182 o.mu.Lock()
183 o.stats.MaxFlushWait = max(o.stats.MaxFlushWait, w)
184 o.mu.Unlock()
185 }
186 }()
187 for {
188 o.mu.Lock()
189 done, running := o.sent >= seq, o.running
190 o.mu.Unlock()
191 if done {
192 return nil
193 }
194 if start.IsZero() {
195 start = time.Now()
196 }
197 if !running {
198 o.sendMu.Lock()
199 ok := o.deliver(ctx, out, o.take())
200 o.sendMu.Unlock()
201 if !ok {
202 return ctx.Err()
203 }
204 continue
205 }
206 // A signal left from an earlier delivery only costs a recheck.
207 select {
208 case <-o.progress:
209 case <-ctx.Done():
210 return ctx.Err()
211 }
212 }
213}
214
215// take returns the queued commands and leaves an empty queue. The caller
216// hands the batch back with recycle once it has sent it.
217func (o *outbox) take() []outCmd {
218 o.mu.Lock()
219 batch := o.queue
220 o.queue, o.spare = o.spare[:0], nil
221 o.busy = len(batch) > 0
222 o.mu.Unlock()
223 return batch
224}
225
226// recycle keeps the batch's storage for the next take.
227func (o *outbox) recycle(batch []outCmd) {
228 // Do not retain what the commands reference in the spare batch.
229 n := uint64(len(batch))
230 clear(batch)
231 o.mu.Lock()
232 o.spare = batch[:0]
233 o.busy = false
234 if n > 0 {
235 o.sent += n
236 }
237 o.mu.Unlock()
238 if n > 0 {
239 select {
240 case o.progress <- struct{}{}:
241 default:
242 }
243 }
244}
245
246// start marks a sender as running; call it before starting run, so that a
247// flush from the owner never delivers alongside it.
248func (o *outbox) start() {
249 o.mu.Lock()
250 o.running = true
251 o.mu.Unlock()
252}
253
254// run delivers the queue to out in order until ctx ends. A command whose
255// send is waiting is not merged any more, but what queues behind it is.
256func (o *outbox) run(ctx context.Context, out chan<- ports.ClientCommand) {
257 defer func() {
258 o.mu.Lock()
259 o.running = false
260 o.mu.Unlock()
261 }()
262 for {
263 o.sendMu.Lock()
264 batch := o.take()
265 if len(batch) == 0 {
266 o.recycle(batch)
267 o.sendMu.Unlock()
268 select {
269 case <-o.wake:
270 case <-ctx.Done():
271 return
272 }
273 continue
274 }
275 ok := o.deliver(ctx, out, batch)
276 o.sendMu.Unlock()
277 if !ok {
278 return
279 }
280 }
281}
282
283// deliver sends batch to out in order and recycles it; false when ctx ended
284// first (the rest of the batch is dropped with the session).
285func (o *outbox) deliver(ctx context.Context, out chan<- ports.ClientCommand, batch []outCmd) bool {
286 // Each entry's wait ends with its own send: a send that blocks midway
287 // must not count against the entries already out.
288 var wait time.Duration
289 for i := range batch {
290 select {
291 case out <- batch[i].command():
292 case <-ctx.Done():
293 return false
294 }
295 wait = max(wait, time.Since(batch[i].at))
296 }
297 o.mu.Lock()
298 o.stats.MaxQueueWait = max(o.stats.MaxQueueWait, wait)
299 o.mu.Unlock()
300 o.recycle(batch)
301 return true
302}