1package wayland
2
3import (
4 "context"
5 "sync/atomic"
6 "time"
7
8 "github.com/bnema/neferwl/internal/ports"
9 "github.com/bnema/purego-libwayland/server"
10)
11
12// commandRingSize bounds the client commands waiting for the display
13// goroutine. It absorbs a few display frames of a 1 kHz mouse and a burst of
14// key/button commands while the display parses client requests; only the
15// forwarder goroutine waits when it is full, so core never blocks on the
16// display directly, only on the Commands channel (cap 32) behind the forwarder.
17const commandRingSize = 256
18
19// queuedCommand is a ClientCommand in the command ring. Pointer motion, bare
20// or wrapped in a SecurityCommand (core wraps every one in production), keeps
21// its value and envelope state inline so folding a motion into a queued one
22// allocates nothing; the command is boxed again only if a merge happened, at
23// apply time, once per merged run.
24type queuedCommand struct {
25 cmd ports.ClientCommand
26 motion ports.PointerMotionTo // valid if isMotion
27 state ports.SecurityState // envelope state, valid if enveloped
28 // isMotion: cmd is a PointerMotionTo, bare or directly inside a
29 // SecurityCommand (then enveloped is set).
30 isMotion, enveloped bool
31 // merged: motion holds the sum of several motions and cmd is stale.
32 merged bool
33 // at is when the command entered the ring path. A merge keeps the oldest
34 // value, so a folded motion reports the wait of the first motion in it.
35 at time.Time
36}
37
38func queue(cmd ports.ClientCommand) queuedCommand {
39 q := queuedCommand{cmd: cmd, at: time.Now()}
40 if env, ok := cmd.(ports.SecurityCommand); ok {
41 q.state, q.enveloped = env.State, true
42 cmd = env.Command
43 }
44 q.motion, q.isMotion = cmd.(ports.PointerMotionTo)
45 return q
46}
47
48func (q *queuedCommand) command() ports.ClientCommand {
49 switch {
50 case !q.merged:
51 return q.cmd
52 case q.enveloped:
53 return ports.SecurityCommand{State: q.state, Command: q.motion}
54 }
55 return q.motion
56}
57
58// mergeMotion folds v into last when both are pointer motion for the same
59// window, both bare or both enveloped with an identical security state, and
60// last is the most recent queued command, so no other command is ever
61// overtaken and a bare motion never takes the place of an enveloped one. The
62// position is the latest; relative deltas add up so locked pointers (games)
63// lose no movement. last.at stays: the merge reports its first motion's wait.
64func mergeMotion(last *queuedCommand, v queuedCommand) bool {
65 if !last.isMotion || !v.isMotion || last.motion.ID != v.motion.ID ||
66 last.enveloped != v.enveloped || last.state != v.state {
67 return false
68 }
69 m := v.motion
70 m.DX += last.motion.DX
71 m.DY += last.motion.DY
72 m.UnaccelDX += last.motion.UnaccelDX
73 m.UnaccelDY += last.motion.UnaccelDY
74 last.motion, last.merged = m, true
75 return true
76}
77
78// newCommandRing returns the ring whose commands apply runs on the display
79// goroutine, in arrival order. Adjacent motions for one window are coalesced
80// at push time, against the last queued command only, under the ring mutex.
81// That also coalesces while the display is busy (a drain-side batch would
82// only see what the channel held at that moment), and a merge needs no room.
83func newCommandRing(d *server.Display, apply func(queuedCommand)) *postRing[queuedCommand] {
84 return newMergingPostRing(d, commandRingSize, apply, mergeMotion)
85}
86
87// commandStats counts what the display goroutine applied. The display
88// goroutine writes and the stats logger takes, so the fields are atomic.
89type commandStats struct {
90 applied atomic.Int64
91 maxLatency atomic.Int64 // nanoseconds from push to apply
92}
93
94// observe records one applied command pushed at at.
95func (c *commandStats) observe(at time.Time, now time.Time) {
96 c.applied.Add(1)
97 d := int64(now.Sub(at))
98 for {
99 old := c.maxLatency.Load()
100 if d <= old || c.maxLatency.CompareAndSwap(old, d) {
101 return
102 }
103 }
104}
105
106// take returns the applied count and the largest latency, and resets both.
107func (c *commandStats) take() (applied int64, maxLatency time.Duration) {
108 return c.applied.Swap(0), time.Duration(c.maxLatency.Swap(0))
109}
110
111// commandStatsInterval is how often logCommandStats reports.
112const commandStatsInterval = 10 * time.Second
113
114// logCommandStats reports command path latency every interval until ctx or the
115// display ends, so a live log shows whether the display goroutine delays input.
116// Only a period with every counter at zero logs nothing: a stalled display
117// applies nothing but still fills the ring.
118func (s *Server) logCommandStats(ctx context.Context) {
119 t := time.NewTicker(commandStatsInterval)
120 defer t.Stop()
121 for {
122 select {
123 case <-ctx.Done():
124 return
125 case <-s.display.Stopped():
126 return
127 case <-t.C:
128 s.reportCommandStats()
129 }
130 }
131}
132
133func (s *Server) reportCommandStats() {
134 rs := s.commands.takeStats()
135 applied, latency := s.commandStats.take()
136 if applied == 0 && latency == 0 && rs == (ringStats{}) {
137 return
138 }
139 s.log.Info().
140 Int64("commands", applied).
141 Int("merged", rs.merged).
142 Int("max_queued", rs.maxLen).
143 Float64("max_push_block_ms", float64(rs.maxPushBlock)/float64(time.Millisecond)).
144 Float64("max_latency_ms", float64(latency)/float64(time.Millisecond)).
145 Msg("command stats")
146}
147
148// applyQueued runs on the display goroutine; it stops applying once the
149// server context is done, like the per-command check of the old batch.
150func (s *Server) applyQueued(q queuedCommand) {
151 if s.ctx != nil && s.ctx.Err() != nil {
152 return
153 }
154 s.commandStats.observe(q.at, time.Now())
155 s.apply(q.command())
156}