1package screencast
2
3import (
4 "time"
5
6 "github.com/bnema/neferwl/internal/ports"
7 "github.com/bnema/purego-pipewire/pipewire"
8)
9
10// The producer turns wayland's damage into PipeWire frames, one stream at a
11// time and without allocating in steady state:
12//
13// CastDamage -> dirty
14// pump: STREAMING, dirty or first frame, no frame in flight, interval
15// elapsed, a buffer free -> Dequeue, CastFrame{Data: buffer}
16// CastFrameDone ok -> Queue, then Trigger (the order PipeWire needs)
17// CastFrameDone with an error -> Release, retry paced
18//
19// The buffer's mapping is lent to the wayland server until the CastFrameDone
20// of that buffer, so the source is never resized or destroyed while a frame
21// is in flight.
22//
23// A cast ends (the session signals Closed) on Stop, when the owner leaves
24// the bus (NameOwnerChanged), when wayland stops the cast (output or window
25// gone, lock), when the PipeWire stream goes to ERROR or back to
26// UNCONNECTED, and when the service shuts down. It does not end when the
27// consumer leaves: as in Mutter, the node then goes to PAUSED, no frame is
28// requested while it is not STREAMING (a frame already in flight is
29// released instead of queued), and the first frame after it is STREAMING
30// again is forced. The D-Bus session stays up for the next consumer.
31
32// wl_shm formats of the supported video formats.
33const (
34 shmARGB8888 = 0
35 shmXRGB8888 = 1
36)
37
38// maxFrameErrors is how many frames in a row may fail before the stream
39// stops.
40const maxFrameErrors = 3
41
42// retryOpen is how long a command waits when the wayland channel is full.
43const retryOpen = 10 * time.Millisecond
44
45func shmFormat(f pipewire.VideoFormat) uint32 {
46 if f == pipewire.VideoFormatBGRA {
47 return shmARGB8888
48 }
49 return shmXRGB8888
50}
51
52// streamOfCast finds the stream of a cast ID.
53func (s *Service) streamOfCast(id uint64) *stream {
54 for i := range s.sessions {
55 sess := &s.sessions[i]
56 if !sess.used {
57 continue
58 }
59 for j := range sess.streams {
60 if st := &sess.streams[j]; st.used && st.cast == id {
61 return st
62 }
63 }
64 }
65 return nil
66}
67
68// onCastEvent applies one event of the wayland server.
69func (s *Service) onCastEvent(ev *ports.CastEvent) {
70 st := s.streamOfCast(ev.ID)
71 if st == nil {
72 return
73 }
74 switch ev.Kind {
75 case ports.CastDamage:
76 st.dirty = true
77 case ports.CastSize:
78 if st.closing {
79 return
80 }
81 if st.haveSize && st.w == ev.Width && st.h == ev.Height {
82 return
83 }
84 st.w, st.h, st.haveSize = ev.Width, ev.Height, true
85 if st.src != nil {
86 st.needResize = true
87 }
88 case ports.CastFrameDone:
89 s.frameDone(st, ev)
90 case ports.CastStopped:
91 st.stopped, st.needClose = true, false
92 if st.inflight {
93 // Wayland orders CastStopped after the frame's CastFrameDone.
94 s.log.Warn().Str("stream", st.path).Msg("cast stopped with a frame in flight")
95 st.inflight = false
96 s.giveBack(st)
97 }
98 if !st.closing {
99 s.log.Info().Str("stream", st.path).Str("reason", string(ev.Reason)).Err(ev.Err).Msg("cast stopped")
100 s.closeSession(st.sess, true)
101 }
102 }
103}
104
105// giveBack releases the buffer a frame borrowed.
106func (s *Service) giveBack(st *stream) {
107 if st.buf != nil && st.src != nil {
108 st.src.Release(st.buf)
109 }
110 st.buf = nil
111}
112
113// frameDone settles the frame in flight: a complete image is queued and
114// triggered, anything else gives the buffer back.
115func (s *Service) frameDone(st *stream, ev *ports.CastEvent) {
116 b := s.settleFrame(st, ev)
117 if b == nil {
118 return
119 }
120 if st.pwState != pipewire.StreamStateStreaming {
121 // The consumer left while the frame was drawn: nobody wants it.
122 // The picture is kept for the consumer that comes next.
123 st.src.Release(b)
124 st.dirty = true
125 return
126 }
127 st.src.Queue(b, st.seq, st.pts, false)
128 st.src.Trigger()
129}
130
131// settleFrame updates the stream for the CastFrameDone of its frame and
132// returns the buffer to queue, nil when there is none. It touches no
133// PipeWire object but Release on the failure path.
134func (s *Service) settleFrame(st *stream, ev *ports.CastEvent) *pipewire.VideoBuffer {
135 if !st.inflight || st.buf == nil || st.buf.Index != ev.Buf {
136 s.log.Warn().Str("stream", st.path).Int("buf", ev.Buf).Msg("frame done for no frame in flight")
137 return nil
138 }
139 st.inflight = false
140 if ev.Err != nil || st.closing {
141 s.giveBack(st)
142 if ev.Err == nil {
143 return nil
144 }
145 st.errs++
146 st.dirty = true
147 if !st.closing {
148 s.log.Info().Str("stream", st.path).Err(ev.Err).Int("errors", st.errs).Msg("cast frame failed")
149 if st.errs >= maxFrameErrors {
150 s.closeSession(st.sess, true)
151 }
152 }
153 return nil
154 }
155 st.errs = 0
156 st.seq++
157 st.pts = 0
158 if !ev.Time.IsZero() {
159 st.pts = time.Duration(ev.Time.UnixNano())
160 }
161 b := st.buf
162 st.buf = nil
163 return b
164}
165
166// pumpAll runs every stream: commands, negotiation and frames.
167func (s *Service) pumpAll(now time.Time) {
168 s.waitFor = 0
169 for i := range s.sessions {
170 sess := &s.sessions[i]
171 if !sess.used {
172 continue
173 }
174 for j := range sess.streams {
175 if st := &sess.streams[j]; st.used {
176 s.pump(st, now)
177 }
178 }
179 }
180}
181
182// wait asks for a poll timeout no longer than d.
183func (s *Service) wait(d time.Duration) {
184 if d <= 0 {
185 d = time.Millisecond
186 }
187 if s.waitFor == 0 || d < s.waitFor {
188 s.waitFor = d
189 }
190}
191
192// send puts a command on the wayland channel without blocking.
193func (s *Service) sendCmd(cmd ports.CastCommand) bool {
194 select {
195 case s.cmds <- cmd:
196 return true
197 default:
198 return false
199 }
200}
201
202func (s *Service) pump(st *stream, now time.Time) {
203 if st.closing {
204 s.pumpClosing(st)
205 return
206 }
207 if st.needOpen {
208 cmd := ports.CastCommand{Kind: ports.CastOpen, ID: st.cast, Output: st.output, Window: st.window, Cursor: st.cursor}
209 if !s.sendCmd(cmd) {
210 s.wait(retryOpen)
211 return
212 }
213 st.needOpen, st.opened = false, true
214 }
215 if !st.haveSize {
216 return
217 }
218 if st.src == nil {
219 s.openSource(st)
220 if st.src == nil {
221 return
222 }
223 }
224 if st.pwState == pipewire.StreamStateError || st.pwState == pipewire.StreamStateUnconnected && st.seenState {
225 s.log.Info().Str("stream", st.path).Int32("state", st.pwState).Msg("PipeWire stream ended")
226 s.closeSession(st.sess, true)
227 return
228 }
229 if st.needResize && !st.inflight {
230 st.needResize = false
231 st.fw, st.fh = 0, 0
232 if err := st.src.Resize(st.w, st.h); err != nil {
233 s.log.Warn().Err(err).Str("stream", st.path).Msg("PipeWire resize failed")
234 s.closeSession(st.sess, true)
235 return
236 }
237 }
238 if !st.announced && st.pwState >= pipewire.StreamStatePaused && st.nodeID != 0 {
239 st.announced = true
240 s.conn.NewSignalTo(st.sess.owner, st.path, ifaceStream, "PipeWireStreamAdded", "u").Uint32(st.nodeID)
241 s.send()
242 }
243 s.pumpFrame(st, now)
244}
245
246// openSource creates the PipeWire node once the target size is known.
247func (s *Service) openSource(st *stream) {
248 if s.pwc == nil {
249 s.closeSession(st.sess, true)
250 return
251 }
252 cfg := pipewire.VideoConfig{Name: "neferwl-screencast", Width: st.w, Height: st.h, MaxFPS: st.maxFPS, Formats: []pipewire.VideoFormat{pipewire.VideoFormatBGRx, pipewire.VideoFormatBGRA}}
253 src, err := s.pwc.NewVideoSource(cfg, &st.h2)
254 if err != nil {
255 s.log.Warn().Err(err).Str("stream", st.path).Msg("PipeWire stream not created")
256 s.closeSession(st.sess, true)
257 return
258 }
259 st.src = src
260}
261
262// pumpFrame requests the next frame when one is due.
263func (s *Service) pumpFrame(st *stream, now time.Time) {
264 if !s.frameDue(st, now) {
265 return
266 }
267 b := st.src.Dequeue()
268 if b == nil {
269 // Every buffer is with the consumer: look again shortly.
270 s.wait(st.interval())
271 return
272 }
273 if !s.issueFrame(st, b, now) {
274 st.src.Release(b)
275 }
276}
277
278// interval is the shortest time between two frames.
279func (st *stream) interval() time.Duration {
280 return time.Second / time.Duration(max(st.maxFPS, 1))
281}
282
283// frameDue reports whether a frame may be requested now: streaming, the
284// negotiated size is the target's, something changed (or it is the first
285// frame), none in flight and the interval has passed. When only the interval
286// is missing, it asks for a poll timeout.
287func (s *Service) frameDue(st *stream, now time.Time) bool {
288 if st.inflight || st.pwState != pipewire.StreamStateStreaming || st.needResize ||
289 st.fw != st.w || st.fh != st.h || st.fw == 0 {
290 return false
291 }
292 if !st.dirty && !st.forced {
293 return false
294 }
295 if since := now.Sub(st.lastReq); !st.lastReq.IsZero() && since < st.interval() {
296 s.wait(st.interval() - since)
297 return false
298 }
299 return true
300}
301
302// issueFrame lends buffer b to the wayland server. False: the buffer is not
303// taken and stays the caller's.
304func (s *Service) issueFrame(st *stream, b *pipewire.VideoBuffer, now time.Time) bool {
305 if b.Width != st.w || b.Height != st.h || len(b.Data) < b.Stride*b.Height {
306 s.wait(st.interval())
307 return false
308 }
309 cmd := ports.CastCommand{Kind: ports.CastFrame, ID: st.cast, Buf: b.Index, Data: b.Data, Stride: b.Stride, Width: b.Width, Height: b.Height, Format: shmFormat(b.Format)}
310 if !s.sendCmd(cmd) {
311 s.wait(st.interval())
312 return false
313 }
314 st.inflight, st.buf = true, b
315 st.lastReq, st.dirty, st.forced = now, false, false
316 return true
317}
318
319// pumpClosing finishes a stream of an ended session: CastClose, then the
320// PipeWire node once no frame is in flight. Destroying the node does not
321// wait for CastClose to be sent (only a frame in flight lends its buffer),
322// but the slot is kept until it is.
323func (s *Service) pumpClosing(st *stream) {
324 if st.needClose {
325 if s.sendCmd(ports.CastCommand{Kind: ports.CastClose, ID: st.cast}) {
326 st.needClose = false
327 } else {
328 s.wait(retryOpen)
329 }
330 }
331 if st.inflight {
332 return
333 }
334 if st.src != nil {
335 s.giveBack(st)
336 st.src.Destroy()
337 st.src = nil
338 }
339 if !st.needClose {
340 *st = stream{}
341 }
342}
343
344// reap frees the dying sessions once every stream released its resources.
345func (s *Service) reap() {
346 for i := range s.sessions {
347 sess := &s.sessions[i]
348 if !sess.used || !sess.dying {
349 continue
350 }
351 live := false
352 for j := range sess.streams {
353 live = live || sess.streams[j].used
354 }
355 if !live {
356 *sess = session{}
357 }
358 }
359}
360
361// streamCount is the number of streams that still hold something: a
362// PipeWire source, a lent buffer or an unsent CastClose.
363func (s *Service) streamCount() int {
364 n := 0
365 s.forEachStream(func(*stream) { n++ })
366 return n
367}
368
369// streamsLive reports whether a stream still holds something to release.
370func (s *Service) streamsLive() bool { return s.streamCount() > 0 }