1package app
2
3import (
4 "context"
5 "fmt"
6 "sync/atomic"
7 "time"
8
9 "github.com/bnema/zerowrap"
10
11 "github.com/bnema/neferwl/internal/adapters/logging"
12 "github.com/bnema/neferwl/internal/adapters/outputkit/capture"
13 "github.com/bnema/neferwl/internal/ports"
14)
15
16// outputRouter feeds the running outputs their scenes, cursor and capture
17// requests. It runs in its own goroutine and never blocks: every mailbox is
18// latest-wins (scenes, cursor) or fails fast when full (captures). Output
19// lifecycle (scan, modeset, start and stop, security, leases) runs on the
20// backend loop, so a hotplug or a configuration on one output never delays
21// another output's frames or cursor.
22//
23// The backend loop registers an output before it runs and unregisters it
24// once it stopped, over ops; registration is lossless and ordered, and an
25// output's instance keys it so a late unregister never removes its
26// successor.
27type outputRouter struct {
28 outs map[string]*mailbox
29 // off is the latest Scene.Off per output name, kept after the output
30 // goes: it reconnects off until core's first scene (drm StartOff).
31 off map[string]bool
32 cursor ports.CursorChange
33 captured chan<- ports.CaptureDone
34 ctx context.Context
35
36 in outputRoutes
37 ops chan routerOp
38 done chan struct{} // closed when run returns
39
40 log zerowrap.Logger
41 // stats counts since the last report; only run's goroutine touches it.
42 stats routerStats
43 // eventWait is the longest time (ns) the backend loop blocked sending an
44 // OutputEvent to core since the last report. The backend loop writes it,
45 // so it is atomic.
46 eventWait atomic.Int64
47}
48
49// routerStatsEvery is the period of the "routing stats" log entry.
50const routerStatsEvery = 10 * time.Second
51
52// routerStats tells whether outputs keep up: a scene or cursor replaced
53// unread means an output did not consume the previous one before the next.
54type routerStats struct {
55 scenes, scenesReplaced uint64
56 cursor, cursorReplaced uint64
57 capturesFull uint64 // captures failed because the queue was full
58}
59
60// outputRoutes are the router's inputs: core's scenes, wayland's cursor and
61// capture requests.
62type outputRoutes struct {
63 scenes <-chan []ports.Scene
64 cursor <-chan ports.CursorChange
65 captures <-chan ports.CaptureRequest
66}
67
68// mailbox is what the router feeds one output lifetime.
69type mailbox struct {
70 instance ports.OutputInstance
71 scenes chan ports.Scene
72 cursor chan ports.CursorChange
73 captures chan ports.CaptureRequest
74 // ctx ends and done closes with the output; stop cancels it.
75 ctx context.Context
76 stop context.CancelFunc
77 done chan struct{}
78}
79
80func newMailbox(instance ports.OutputInstance, ctx context.Context, stop context.CancelFunc, done chan struct{}) *mailbox {
81 return &mailbox{instance: instance, scenes: make(chan ports.Scene, 1), cursor: make(chan ports.CursorChange, 1), captures: make(chan ports.CaptureRequest, 16), ctx: ctx, stop: stop, done: done}
82}
83
84// routerOp registers (add set) or unregisters an output. reply carries the
85// registration's start-off state; it is closed after an unregister.
86type routerOp struct {
87 name string
88 add *mailbox
89 gone ports.OutputInstance
90 reply chan bool
91}
92
93func newOutputRouter(ctx context.Context, captured chan<- ports.CaptureDone, in outputRoutes) *outputRouter {
94 return &outputRouter{outs: map[string]*mailbox{}, off: map[string]bool{}, captured: captured, ctx: ctx, in: in, ops: make(chan routerOp), done: make(chan struct{}), log: logging.For(ctx, "router")}
95}
96
97// run routes until ctx ends.
98func (r *outputRouter) run(ctx context.Context) {
99 defer close(r.done)
100 report := time.NewTicker(routerStatsEvery)
101 defer report.Stop()
102 for {
103 select {
104 case <-ctx.Done():
105 return
106 case <-report.C:
107 r.report()
108 case op := <-r.ops:
109 r.apply(op)
110 case set := <-r.in.scenes:
111 r.scenes(set)
112 case c := <-r.in.cursor:
113 r.setCursor(c)
114 case q := <-r.in.captures:
115 r.routeCapture(q)
116 }
117 }
118}
119
120// report logs and resets the counters; it stays quiet when nothing happened.
121func (r *outputRouter) report() {
122 st, wait := r.stats, time.Duration(r.eventWait.Swap(0))
123 r.stats = routerStats{}
124 if st == (routerStats{}) && wait == 0 {
125 return
126 }
127 r.log.Info().
128 Uint64("scenes", st.scenes).
129 Uint64("scenes_replaced", st.scenesReplaced).
130 Uint64("cursor", st.cursor).
131 Uint64("cursor_replaced", st.cursorReplaced).
132 Uint64("captures_full", st.capturesFull).
133 Float64("max_event_send_ms", float64(wait)/float64(time.Millisecond)).
134 Msg("routing stats")
135}
136
137// noteEventSend records how long the backend loop waited for core to take an
138// event. Backend loop only; lock-free so it never adds to the wait.
139func (r *outputRouter) noteEventSend(d time.Duration) {
140 n := int64(d)
141 for {
142 cur := r.eventWait.Load()
143 if n <= cur || r.eventWait.CompareAndSwap(cur, n) {
144 return
145 }
146 }
147}
148
149func (r *outputRouter) apply(op routerOp) {
150 if op.add != nil {
151 op.reply <- r.register(op.name, op.add)
152 return
153 }
154 r.unregister(op.name, op.gone)
155 close(op.reply)
156}
157
158// register feeds m from now on, starting with the current cursor, and
159// reports whether the output's latest scene turned it off.
160func (r *outputRouter) register(name string, m *mailbox) bool {
161 r.outs[name] = m
162 m.cursor <- r.cursor
163 return r.off[name]
164}
165
166// unregister stops feeding the output lifetime instance of name.
167func (r *outputRouter) unregister(name string, instance ports.OutputInstance) {
168 if m := r.outs[name]; m != nil && m.instance == instance {
169 delete(r.outs, name)
170 }
171}
172
173// setCursor gives every output the latest cursor, replacing an unread one.
174func (r *outputRouter) setCursor(c ports.CursorChange) {
175 r.cursor = c
176 r.stats.cursor++
177 for _, m := range r.outs {
178 if latest(m.cursor, c) {
179 r.stats.cursorReplaced++
180 }
181 }
182}
183
184// scenes routes each scene to its output, replacing an unread one.
185func (r *outputRouter) scenes(set []ports.Scene) {
186 for _, sc := range set {
187 if sc.Off {
188 r.off[sc.Output] = true
189 } else {
190 delete(r.off, sc.Output)
191 }
192 if m := r.outs[sc.Output]; m != nil {
193 r.stats.scenes++
194 if latest(m.scenes, sc) {
195 r.stats.scenesReplaced++
196 }
197 }
198 }
199}
200
201// routeCapture hands the request to its output or fails it at once: a full
202// or stopped output never holds up the router.
203func (r *outputRouter) routeCapture(req ports.CaptureRequest) {
204 m := r.outs[req.Output]
205 if m == nil {
206 capture.Fail(r.ctx, req, fmt.Errorf("output %q unavailable", req.Output), r.captured)
207 return
208 }
209 select {
210 case m.captures <- req:
211 case <-m.done:
212 m.stop()
213 capture.Fail(r.ctx, req, fmt.Errorf("output %q stopped", req.Output), r.captured)
214 case <-m.ctx.Done():
215 capture.Fail(r.ctx, req, fmt.Errorf("output %q stopped", req.Output), r.captured)
216 default:
217 r.stats.capturesFull++
218 capture.Fail(r.ctx, req, fmt.Errorf("output %q capture queue full", req.Output), r.captured)
219 }
220}
221
222// add registers m for name and returns its start-off state. Backend loop
223// only; false once the router has stopped.
224func (r *outputRouter) add(name string, m *mailbox) (off, ok bool) {
225 reply := make(chan bool, 1)
226 select {
227 case r.ops <- routerOp{name: name, add: m, reply: reply}:
228 return <-reply, true
229 case <-r.done:
230 return false, false
231 }
232}
233
234// remove unregisters the lifetime instance of name and returns once the
235// router no longer feeds it, so its queued captures can be failed. Backend
236// loop only.
237func (r *outputRouter) remove(name string, instance ports.OutputInstance) {
238 reply := make(chan bool)
239 select {
240 case r.ops <- routerOp{name: name, gone: instance, reply: reply}:
241 <-reply
242 case <-r.done:
243 }
244}