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 for outbox room.
55 MaxQueueWait, MaxFlushWait time.Duration
56}
57
58// takeStats returns the counters and starts over.
59func (o *outbox) takeStats() CommandStats {
60 o.mu.Lock()
61 defer o.mu.Unlock()
62 s := o.stats
63 s.MaxBacklog = max(s.MaxBacklog, o.pushed-o.sent)
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// bounded outbox backpressure. Without a running sender (core driven
175// without Run), it delivers the queue itself, in order.
176func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error {
177 var start time.Time
178 defer func() {
179 if !start.IsZero() {
180 w := time.Since(start)
181 o.mu.Lock()
182 o.stats.MaxFlushWait = max(o.stats.MaxFlushWait, w)
183 o.mu.Unlock()
184 }
185 }()
186 for {
187 o.mu.Lock()
188 done, running := o.sent >= seq, o.running
189 o.mu.Unlock()
190 if done {
191 return nil
192 }
193 if start.IsZero() {
194 start = time.Now()
195 }
196 if !running {
197 o.sendMu.Lock()
198 ok := o.deliver(ctx, out, o.take())
199 o.sendMu.Unlock()
200 if !ok {
201 return ctx.Err()
202 }
203 continue
204 }
205 // A signal left from an earlier delivery only costs a recheck.
206 select {
207 case <-o.progress:
208 case <-ctx.Done():
209 return ctx.Err()
210 }
211 }
212}
213
214// take returns the queued commands and leaves an empty queue. The caller
215// hands the batch back with recycle once it has sent it.
216func (o *outbox) take() []outCmd {
217 o.mu.Lock()
218 batch := o.queue
219 o.queue, o.spare = o.spare[:0], nil
220 o.busy = len(batch) > 0
221 o.mu.Unlock()
222 return batch
223}
224
225// recycle keeps the batch's storage for the next take.
226func (o *outbox) recycle(batch []outCmd) {
227 // Do not retain what the commands reference in the spare batch.
228 n := uint64(len(batch))
229 clear(batch)
230 o.mu.Lock()
231 o.spare = batch[:0]
232 o.busy = false
233 if n > 0 {
234 o.sent += n
235 }
236 o.mu.Unlock()
237 if n > 0 {
238 select {
239 case o.progress <- struct{}{}:
240 default:
241 }
242 }
243}
244
245// start marks a sender as running; call it before starting run, so that a
246// flush from the owner never delivers alongside it.
247func (o *outbox) start() {
248 o.mu.Lock()
249 o.running = true
250 o.mu.Unlock()
251}
252
253// run delivers the queue to out in order until ctx ends. A command whose
254// send is waiting is not merged any more, but what queues behind it is.
255func (o *outbox) run(ctx context.Context, out chan<- ports.ClientCommand) {
256 defer func() {
257 o.mu.Lock()
258 o.running = false
259 o.mu.Unlock()
260 }()
261 for {
262 o.sendMu.Lock()
263 batch := o.take()
264 if len(batch) == 0 {
265 o.recycle(batch)
266 o.sendMu.Unlock()
267 select {
268 case <-o.wake:
269 case <-ctx.Done():
270 return
271 }
272 continue
273 }
274 ok := o.deliver(ctx, out, batch)
275 o.sendMu.Unlock()
276 if !ok {
277 return
278 }
279 }
280}
281
282// deliver sends batch to out in order and recycles it; false when ctx ended
283// first (the rest of the batch is dropped with the session).
284func (o *outbox) deliver(ctx context.Context, out chan<- ports.ClientCommand, batch []outCmd) bool {
285 // Each entry's wait ends with its own send: a send that blocks midway
286 // must not count against the entries already out.
287 var wait time.Duration
288 for i := range batch {
289 select {
290 case out <- batch[i].command():
291 case <-ctx.Done():
292 return false
293 }
294 wait = max(wait, time.Since(batch[i].at))
295 }
296 o.mu.Lock()
297 o.stats.MaxQueueWait = max(o.stats.MaxQueueWait, wait)
298 o.mu.Unlock()
299 o.recycle(batch)
300 return true
301}