1// Package screencast serves the GNOME screen cast and screenshot D-Bus
2// interfaces (org.gnome.Mutter.ScreenCast, org.gnome.Mutter.DisplayConfig,
3// org.gnome.Shell.Introspect, org.gnome.Shell.Screenshot) so that
4// xdg-desktop-portal-gnome can offer monitor and window sharing and
5// screenshots. A cast is a PipeWire Video/Source node whose frames the
6// wayland server renders into the node's buffers (ports.CastCommand), and a
7// screenshot is taken by running nefercap.
8//
9// One owner goroutine, locked to its OS thread, owns all state: the D-Bus
10// connection, the PipeWire loop, the sessions and the streams. It waits for
11// the bus, the PipeWire loop and wake-ups from other goroutines (the wayland
12// sink and screenshot jobs) in a single poll. Callers are checked against
13// the capture allowlist through a pidfd before anything is dispatched.
14package screencast
15
16import (
17 "context"
18 "errors"
19 "runtime"
20 "sync"
21 "time"
22
23 "github.com/bnema/neferwl/internal/adapters/busretry"
24 "github.com/bnema/neferwl/internal/adapters/captureallow"
25 "github.com/bnema/neferwl/internal/adapters/clock"
26 "github.com/bnema/neferwl/internal/adapters/fdwait"
27 "github.com/bnema/neferwl/internal/ports"
28 "github.com/bnema/zerobus"
29 "github.com/bnema/zerowrap"
30 "golang.org/x/sys/unix"
31)
32
33// The bus names served, and the bounds of the fixed tables.
34const (
35 nameScreenCast = "org.gnome.Mutter.ScreenCast"
36 nameDisplayConfig = "org.gnome.Mutter.DisplayConfig"
37 nameIntrospect = "org.gnome.Shell.Introspect"
38 nameScreenshot = "org.gnome.Shell.Screenshot"
39 busDaemon = "org.freedesktop.DBus"
40 busDaemonPath = "/org/freedesktop/DBus"
41
42 maxSessions = 8 // live sessions
43 maxStreams = 4 // streams per session
44 maxPending = 16 // requests waiting for a caller check
45 maxCallers = 32 // checked D-Bus peers
46 maxArg = 128
47
48 // sinkCap bounds the wayland events waiting for the owner. Damage past
49 // it is merged into a flag; nothing else is dropped.
50 sinkCap = 256
51
52 // shotTimeout bounds a nefercap run.
53 shotTimeout = 60 * time.Second
54
55 // shutdownWait bounds the wait for frames still in flight at shutdown,
56 // once the service itself is stopping.
57 shutdownWait = 2 * time.Second
58
59 // shutdownPoll is how often shutdown looks again for the events that
60 // settle a frame in flight.
61 shutdownPoll = 5 * time.Millisecond
62)
63
64var _ ports.CastSink = (*Service)(nil)
65
66// Service is the screencast adapter. Create it with New and run it once
67// with Run; Push and Inventory are the only methods other goroutines call.
68type Service struct {
69 policy func() *captureallow.Policy
70 cmds chan<- ports.CastCommand
71 log zerowrap.Logger
72
73 poller *fdwait.Poller
74 initErr error
75
76 // Seams; tests replace them before Run.
77 clock ports.Clock
78 dial dialer
79 pw pipewireOpener
80 peers peerResolver
81 runner runner
82 address string // session bus when empty
83 env []string
84
85 // Transport: filled by other goroutines, drained by the owner.
86 mu sync.Mutex
87 queue []ports.CastEvent
88 spare []ports.CastEvent
89 lost bool // damage was merged away
90 woken bool
91 invNew *ports.CastInventory
92 shotRes shotResult
93
94 // Owner state.
95 ctx context.Context // the attempt's: it ends with the attempt
96 conn bus
97 pwc videoCore
98 owns [4]bool // nameScreenCast, nameDisplayConfig, nameIntrospect, nameScreenshot
99 inv ports.CastInventory
100 haveInv bool
101 callers [maxCallers]caller
102 pending [maxPending]pending
103 // sessions is a pointer so that a shutdown at process exit that had to
104 // leave a PipeWire source behind can abandon the table: the source's
105 // handler still points into it, so it is never reset or reused.
106 sessions *[maxSessions]session
107 nextSess uint32
108 nextStrm uint32
109 nextCast uint64
110 waitFor time.Duration // time to the next paced frame, 0 when none
111 shot shotState
112 shotSeq uint64 // ID of the last screenshot job; never reset
113 fatal error
114 jobs sync.WaitGroup
115 drain time.Duration // how long shutdown waits for frames once the service stops
116}
117
118// New creates the service. policy returns the capture allowlist in force;
119// cmds carries the commands for the wayland server.
120func New(policy func() *captureallow.Policy, cmds chan<- ports.CastCommand, log zerowrap.Logger) *Service {
121 s := &Service{
122 policy: policy,
123 cmds: cmds,
124 log: log,
125 clock: clock.System{},
126 dial: zerobusDialer{},
127 pw: realPipeWire{},
128 peers: pidfdPeers{},
129 queue: make([]ports.CastEvent, 0, sinkCap),
130 spare: make([]ports.CastEvent, 0, sinkCap),
131
132 sessions: new([maxSessions]session),
133 drain: shutdownWait,
134 }
135 s.poller, s.initErr = fdwait.New()
136 return s
137}
138
139// SetEnv sets the environment nefercap runs in (WAYLAND_DISPLAY names this
140// compositor). Call it before Run.
141func (s *Service) SetEnv(env []string) { s.env = env }
142
143// Push queues an event of the wayland server for the owner. It never blocks.
144func (s *Service) Push(ev ports.CastEvent) {
145 s.mu.Lock()
146 if ev.Kind == ports.CastDamage && len(s.queue) >= sinkCap {
147 s.lost = true
148 } else {
149 s.queue = append(s.queue, ev)
150 }
151 wake := !s.woken
152 s.woken = true
153 s.mu.Unlock()
154 if wake {
155 s.wake()
156 }
157}
158
159// Inventory takes the new outputs and windows; the owner publishes them.
160func (s *Service) Inventory(inv ports.CastInventory) {
161 s.mu.Lock()
162 s.invNew = &inv
163 wake := !s.woken
164 s.woken = true
165 s.mu.Unlock()
166 if wake {
167 s.wake()
168 }
169}
170
171func (s *Service) wake() {
172 if s.poller != nil {
173 s.poller.Wake()
174 }
175}
176
177// take swaps out what other goroutines queued. The events stay valid until
178// the next take.
179func (s *Service) take() (evs []ports.CastEvent, lost bool, inv *ports.CastInventory, shot *shotResult) {
180 s.mu.Lock()
181 s.spare, s.queue = s.queue, s.spare[:0]
182 evs, lost, inv = s.spare, s.lost, s.invNew
183 s.lost, s.invNew, s.woken = false, nil, false
184 if s.shotRes.ready {
185 r := s.shotRes
186 s.shotRes = shotResult{}
187 shot = &r
188 }
189 s.mu.Unlock()
190 return evs, lost, inv, shot
191}
192
193// Run serves until ctx ends. A lost bus or a name that others own is tried
194// again after retry, growing to a minute. It returns once every job and
195// stream is released.
196func (s *Service) Run(ctx context.Context, retry time.Duration) {
197 if s.initErr != nil {
198 s.log.Warn().Err(s.initErr).Msg("screencast disabled")
199 return
200 }
201 defer s.poller.Close()
202 s.poller.StopOn(ctx)
203 if s.runner == nil {
204 s.runner = execRunner{env: s.env}
205 }
206 busretry.Run(ctx, retry, "D-Bus screencast", s.log, s.attempt)
207 s.jobs.Wait()
208}
209
210// attempt is one connection of the service to the buses.
211func (s *Service) attempt(parent context.Context) (err error) {
212 // The PipeWire loop belongs to this OS thread for its whole life.
213 runtime.LockOSThread()
214 defer runtime.UnlockOSThread()
215 conn, err := s.dial.Dial(s.address)
216 if err != nil {
217 return err
218 }
219 defer func() { _ = conn.Close() }()
220 // The screenshot job of this attempt ends with it.
221 ctx, cancel := context.WithCancel(parent)
222 defer cancel()
223 s.reset(ctx, conn)
224 if s.pwc, err = s.pw.Open(); err != nil {
225 s.pwc = nil
226 s.log.Warn().Err(err).Msg("PipeWire unavailable: screen casting is not offered")
227 }
228 defer s.shutdown(parent)
229 if err = s.acquire(); err != nil {
230 return err
231 }
232 return s.loop(ctx)
233}
234
235// reset clears the owner state of a previous attempt.
236func (s *Service) reset(ctx context.Context, conn bus) {
237 s.ctx, s.conn, s.fatal = ctx, conn, nil
238 s.owns = [4]bool{}
239 s.haveInv = false
240 s.callers = [maxCallers]caller{}
241 s.pending = [maxPending]pending{}
242 *s.sessions = [maxSessions]session{}
243 s.shot = shotState{}
244}
245
246// acquire watches departures and takes the names.
247func (s *Service) acquire() error {
248 rule := "type='signal',sender='" + busDaemon + "',interface='" + busDaemon + "',member='NameOwnerChanged',arg2=''"
249 if err := s.conn.AddMatch(rule); err != nil {
250 return err
251 }
252 names := [4]string{nameScreenCast, nameDisplayConfig, nameIntrospect, nameScreenshot}
253 got := 0
254 for i, name := range names {
255 if i == 0 && s.pwc == nil {
256 continue
257 }
258 code, err := s.conn.RequestName(name, zerobus.NameFlagDoNotQueue)
259 if err != nil {
260 return err
261 }
262 if code != zerobus.NamePrimaryOwner && code != zerobus.NameAlreadyOwner {
263 s.log.Warn().Str("name", name).Msg("name owned by another program; not served")
264 continue
265 }
266 s.owns[i] = true
267 got++
268 s.log.Info().Str("name", name).Msg("serving")
269 }
270 if got == 0 {
271 return errors.New("no D-Bus name acquired")
272 }
273 return nil
274}
275
276// loop waits for the bus, PipeWire and other goroutines and serves them.
277func (s *Service) loop(ctx context.Context) error {
278 rc, err := s.conn.SyscallConn()
279 if err != nil {
280 return err
281 }
282 busFD := int32(-1)
283 if err := rc.Control(func(fd uintptr) { busFD = int32(fd) }); err != nil {
284 return err
285 }
286 var fds [2]unix.PollFd
287 for ctx.Err() == nil {
288 for s.fatal == nil && s.conn.Buffered() {
289 s.readOne()
290 }
291 if s.pwc != nil {
292 s.pwc.Iterate()
293 }
294 s.service()
295 if s.fatal != nil {
296 return s.fatal
297 }
298 fds[0] = unix.PollFd{Fd: busFD, Events: unix.POLLIN}
299 n := 1
300 if s.pwc != nil {
301 fds[1] = unix.PollFd{Fd: int32(s.pwc.Fd()), Events: unix.POLLIN}
302 n = 2
303 }
304 if _, err := s.poller.Poll(fds[:n], s.timeout()); err != nil {
305 return err
306 }
307 if fds[0].Revents != 0 {
308 s.readOne()
309 }
310 }
311 return nil
312}
313
314// readOne reads and handles one message; the read may wait for the rest of
315// a message that arrived in part.
316func (s *Service) readOne() {
317 m, err := s.conn.ReadMessage()
318 if err != nil {
319 s.fatal = err
320 return
321 }
322 s.onMessage(m)
323}
324
325// service applies everything the other goroutines queued, then produces the
326// frames that are due.
327func (s *Service) service() {
328 evs, lost, inv, shot := s.take()
329 if inv != nil {
330 s.applyInventory(*inv)
331 }
332 if lost {
333 s.forEachStream(func(st *stream) { st.dirty = true })
334 }
335 for i := range evs {
336 s.onCastEvent(&evs[i])
337 }
338 if shot != nil {
339 s.onShotDone(shot)
340 }
341 s.pumpAll(s.clock.Now())
342 s.reap()
343}
344
345// shutdown ends the attempt: the screenshot job is stopped and joined, every
346// session is closed, and every stream is pumped until it released its
347// resources: CastClose goes to the wayland server, a frame still in flight
348// is waited for (its buffer is lent to the wayland server), then the source
349// is destroyed and its buffers released. Only then is PipeWire closed.
350// The pidfds of the cached callers are closed first. parent is the context
351// the service runs under; the attempt's own context
352// may have ended for another reason.
353//
354// While parent is alive the attempt ended by itself (the bus was lost) and a
355// new one follows. The wayland server is alive and answers every CastFrame
356// with a CastFrameDone, so the wait for the frames in flight has no
357// deadline: a source left behind would keep its slot in the process-wide
358// table of PipeWire video sources for good, and repeated reconnects would
359// exhaust it. If parent ends meanwhile, the wait is bounded as below.
360//
361// Once parent has ended, the process is going away and the wait for the
362// frames in flight is bounded by s.drain. When a frame does not come back in
363// time, the source cannot be destroyed safely. It is left as it is, with its
364// loop, to the process exit. The sessions table that its handler points into
365// is abandoned (never reset or reused) and a new one takes its place.
366func (s *Service) shutdown(parent context.Context) {
367 s.stopShot()
368 // The cached pidfds belong to this attempt's bus connection.
369 for i := range s.callers {
370 s.callers[i].release()
371 }
372 s.forEachSession(func(sess *session) { s.closeSession(sess, true) })
373 var deadline time.Time // set once parent ended
374 for {
375 evs, _, _, _ := s.take()
376 for i := range evs {
377 s.onCastEvent(&evs[i])
378 }
379 s.pumpAll(s.clock.Now())
380 s.reap()
381 if !s.streamsLive() {
382 break
383 }
384 if deadline.IsZero() && parent.Err() != nil {
385 deadline = time.Now().Add(s.drain)
386 }
387 if !deadline.IsZero() && !time.Now().Before(deadline) {
388 s.log.Warn().Int("streams", s.streamCount()).Msg("frames still in flight at shutdown; PipeWire resources are left to the process exit")
389 s.sessions = new([maxSessions]session)
390 s.pwc = nil
391 return
392 }
393 time.Sleep(shutdownPoll)
394 }
395 if s.pwc != nil {
396 s.pwc.Close()
397 s.pwc = nil
398 }
399}
400
401// timeout is the poll timeout in milliseconds: the time to the next frame
402// that only waits for its interval, or -1.
403func (s *Service) timeout() int {
404 wait := s.waitFor
405 s.waitFor = 0
406 if wait <= 0 {
407 return -1
408 }
409 return int((wait + time.Millisecond - 1) / time.Millisecond)
410}