1package wayland
2
3import (
4 "context"
5 "errors"
6 "os"
7 "sync"
8 "sync/atomic"
9
10 "github.com/bnema/go-wayland-bindings/server/linuxdrmsyncobj"
11 "github.com/bnema/go-wayland-bindings/server/wayland"
12 "github.com/bnema/neferwl/internal/ports"
13 "github.com/bnema/purego-libwayland/server"
14 "github.com/bnema/zerowrap"
15 "golang.org/x/sys/unix"
16)
17
18// wp_linux_drm_syncobj_v1: explicit sync. A client names, per commit, the
19// timeline point its GPU signals when the buffer is drawn (acquire) and
20// the point we signal once nothing reads the buffer any more (release).
21// A commit waits (like a fifo barrier) until its acquire point has a
22// fence; the fence then travels with the content as a sync file the
23// renderer and the display wait on, so no CPU waits for the GPU. The
24// release point replaces wl_buffer.release.
25
26const syncobjVersion = 1
27
28// timeline is an imported client timeline.
29type timeline struct {
30 dev syncobjDevice
31 handle uint32
32 alive bool
33 // uses counts committed points still to signal or wait on.
34 uses int
35}
36
37// syncPoint is a point on a timeline.
38type syncPoint struct {
39 tl *timeline
40 point uint64
41}
42
43func (p syncPoint) set() bool { return p.tl != nil }
44
45// syncState is a surface's wp_linux_drm_syncobj_surface_v1 state.
46type syncState struct {
47 resource *linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1
48 acquire, release syncPoint // pending, for the next commit
49}
50
51func registerSyncobj(d *server.Display, s *Server) error {
52 if s.syncDev == nil {
53 return nil
54 }
55 return linuxdrmsyncobj.NewWpLinuxDrmSyncobjManagerV1Global(d, syncobjVersion, func(c server.Client, v, id uint32) {
56 _, _ = linuxdrmsyncobj.NewWpLinuxDrmSyncobjManagerV1(c, int32(v), id, syncManager{s})
57 })
58}
59
60type syncManager struct{ server *Server }
61
62func (syncManager) Destroy(*linuxdrmsyncobj.WpLinuxDrmSyncobjManagerV1) {}
63
64func (m syncManager) GetSurface(r *linuxdrmsyncobj.WpLinuxDrmSyncobjManagerV1, id uint32, surf *wayland.Surface) {
65 state := m.server.surfaceOf(surf)
66 if state == nil {
67 return
68 }
69 if state.sync != nil {
70 r.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjManagerV1ErrorSurfaceExists), "surface already has a syncobj surface")
71 return
72 }
73 st := &syncState{}
74 res, err := linuxdrmsyncobj.NewWpLinuxDrmSyncobjSurfaceV1(r.Client(), r.Version(), id, syncSurface{state, st})
75 if err != nil {
76 return
77 }
78 st.resource = res
79 state.sync = st
80 res.OnDestroy = func() {
81 if state.sync == st {
82 state.sync = nil
83 }
84 }
85}
86
87func (m syncManager) ImportTimeline(r *linuxdrmsyncobj.WpLinuxDrmSyncobjManagerV1, id uint32, fd int) {
88 defer unix.Close(fd)
89 h, err := m.server.syncDev.fdToHandle(fd)
90 if err != nil {
91 r.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjManagerV1ErrorInvalidTimeline), "invalid timeline fd")
92 return
93 }
94 tl := &timeline{dev: m.server.syncDev, handle: h, alive: true}
95 res, err := linuxdrmsyncobj.NewWpLinuxDrmSyncobjTimelineV1(r.Client(), r.Version(), id, timelineHandler{})
96 if err != nil {
97 _ = tl.dev.destroy(h)
98 return
99 }
100 m.server.timelines[res.Resource] = tl
101 res.OnDestroy = func() {
102 delete(m.server.timelines, res.Resource)
103 // Points already committed keep the syncobj: destroyed once no
104 // held buffer or waiting commit uses it (timeline.drop).
105 tl.alive = false
106 m.server.dropTimeline(tl)
107 }
108}
109
110type timelineHandler struct{}
111
112func (timelineHandler) Destroy(*linuxdrmsyncobj.WpLinuxDrmSyncobjTimelineV1) {}
113
114type syncSurface struct {
115 surf *surface
116 st *syncState
117}
118
119func (syncSurface) Destroy(*linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1) {}
120
121func (h syncSurface) point(r *linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1, tl *linuxdrmsyncobj.WpLinuxDrmSyncobjTimelineV1, hi, lo uint32) (syncPoint, bool) {
122 if h.surf.destroyed {
123 r.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorNoSurface), "surface destroyed")
124 return syncPoint{}, false
125 }
126 t := h.surf.server.timelines[tl.Resource]
127 if t == nil {
128 return syncPoint{}, false
129 }
130 return syncPoint{tl: t, point: uint64(hi)<<32 | uint64(lo)}, true
131}
132
133func (h syncSurface) SetAcquirePoint(r *linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1, tl *linuxdrmsyncobj.WpLinuxDrmSyncobjTimelineV1, hi, lo uint32) {
134 if p, ok := h.point(r, tl, hi, lo); ok {
135 h.st.acquire = p
136 }
137}
138
139func (h syncSurface) SetReleasePoint(r *linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1, tl *linuxdrmsyncobj.WpLinuxDrmSyncobjTimelineV1, hi, lo uint32) {
140 if p, ok := h.point(r, tl, hi, lo); ok {
141 h.st.release = p
142 }
143}
144
145// checkSyncCommit applies the protocol rules to a commit with sync
146// points; false means a protocol error was posted.
147func (s *surface) checkSyncCommit() bool {
148 st := s.sync
149 if st == nil {
150 return true
151 }
152 a, r := st.acquire, st.release
153 // Points belong only to a commit that attaches a non-null buffer.
154 hasBuffer := s.next.attached && s.next.buffer != nil
155 res := st.resource
156 switch {
157 case !a.set() && !r.set():
158 if hasBuffer {
159 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorNoAcquirePoint), "buffer without acquire point")
160 return false
161 }
162 return true
163 case !hasBuffer:
164 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorNoBuffer), "points without a buffer")
165 case !a.set():
166 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorNoAcquirePoint), "no acquire point")
167 case !r.set():
168 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorNoReleasePoint), "no release point")
169 case a.tl == r.tl && a.point >= r.point:
170 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorConflictingPoints), "release point not after acquire point")
171 case s.pendingIsSHM():
172 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorUnsupportedBuffer), "explicit sync on a wl_shm buffer")
173 default:
174 return true
175 }
176 return false
177}
178
179// pendingIsSHM reports whether the buffer of the next commit is wl_shm.
180func (s *surface) pendingIsSHM() bool {
181 b := s.next.buffer
182 if !s.next.attached {
183 b = s.current
184 }
185 if b == nil {
186 return false
187 }
188 _, dma := s.server.buffers[b.Resource].(*dmabufBuffer)
189 return !dma
190}
191
192// syncWaiter watches acquire points: each gets an eventfd the kernel
193// writes once the point is signalled, and the pacer is woken to apply the
194// commits that waited for it.
195type acquirePoller interface {
196 poll([]unix.PollFd, int) (int, error)
197}
198
199type unixAcquirePoller struct{}
200
201func (unixAcquirePoller) poll(fds []unix.PollFd, timeout int) (int, error) {
202 return unix.Poll(fds, timeout)
203}
204
205type syncWaiter struct {
206 poller acquirePoller
207 terminal bool // polling ended in failure; protected by mu
208 mu sync.Mutex
209 ready map[*syncWait]bool
210 wake func()
211 pipeR *os.File
212 pipeW *os.File
213 added []*syncWait // new waits for run (under mu)
214 log zerowrap.Logger
215 pollError sync.Once
216}
217
218// syncWait is one acquire point a queued commit waits for.
219type syncWait struct {
220 efd int
221 done bool // the eventfd fired or the wait was cancelled (pacer, under Do)
222 failed bool // observation failed; done must not authorize publication
223 cancelled atomic.Bool // run stops polling it and closes efd
224}
225
226func newSyncWaiter(wake func()) (*syncWaiter, error) {
227 r, w, err := os.Pipe()
228 if err != nil {
229 return nil, err
230 }
231 return &syncWaiter{poller: unixAcquirePoller{}, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
232}
233
234// watch registers an eventfd for p; the wait is ready once it fires.
235func (sw *syncWaiter) watch(p syncPoint) (*syncWait, error) {
236 if sw.stopped() {
237 return &syncWait{efd: -1, done: true, failed: true}, unix.EIO
238 }
239 ready, err := p.tl.dev.signalled(p.tl.handle, p.point)
240 if err != nil || ready {
241 return &syncWait{efd: -1, done: true, failed: err != nil}, err
242 }
243 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
244 if err != nil {
245 return &syncWait{efd: -1, done: true, failed: true}, err
246 }
247 if err := p.tl.dev.eventfd(p.tl.handle, p.point, efd); err != nil {
248 unix.Close(efd)
249 return &syncWait{efd: -1, done: true, failed: true}, err
250 }
251 w := sw.add(efd)
252 if w.failed {
253 return w, unix.EIO
254 }
255 return w, nil
256}
257
258// watchImplicit waits for a dma-buf's exclusive fences without blocking dispatch.
259func (sw *syncWaiter) watchImplicit(b *ports.DMABuf, waits *[4]*syncWait) error {
260 if sw.stopped() {
261 return unix.EIO
262 }
263 var fds [4]unix.PollFd
264 for i, p := range b.Planes {
265 if i >= len(fds) {
266 break
267 }
268 fds[i] = unix.PollFd{Fd: int32(p.File.Fd()), Events: unix.POLLIN}
269 }
270 planes := fds[:min(len(b.Planes), len(fds))]
271 _, err := unix.Poll(planes, 0)
272 for errors.Is(err, unix.EINTR) { // a signal, not a fence error
273 _, err = unix.Poll(planes, 0)
274 }
275 if err != nil {
276 return err
277 }
278 for i, fd := range planes {
279 if fd.Revents&unix.POLLIN != 0 {
280 continue
281 }
282 if fd.Revents&(unix.POLLERR|unix.POLLNVAL|unix.POLLHUP) != 0 {
283 sw.warnPollError()
284 return unix.EIO
285 }
286 dup, err := unix.FcntlInt(uintptr(fd.Fd), unix.F_DUPFD_CLOEXEC, 0)
287 if err != nil {
288 for j, w := range waits {
289 sw.cancel(w)
290 waits[j] = nil
291 }
292 return err
293 }
294 waits[i] = sw.add(dup)
295 if waits[i].failed {
296 for j, w := range waits {
297 if w != nil {
298 sw.cancel(w)
299 waits[j] = nil
300 }
301 }
302 return unix.EIO
303 }
304 }
305 return nil
306}
307
308func (sw *syncWaiter) warnPollError() {
309 sw.pollError.Do(func() {
310 sw.log.Warn().Str("component", "wayland").Msg("implicit fence poll error")
311 })
312}
313
314func (sw *syncWaiter) add(fd int) *syncWait {
315 w := &syncWait{efd: fd}
316 sw.mu.Lock()
317 if sw.terminal {
318 sw.mu.Unlock()
319 unix.Close(fd)
320 return &syncWait{efd: -1, done: true, failed: true}
321 }
322 sw.added = append(sw.added, w)
323 sw.mu.Unlock()
324 _, _ = sw.pipeW.Write([]byte{0})
325 return w
326}
327
328// fired reports whether w's point is signalled, and forgets w once it has.
329func (sw *syncWaiter) fired(w *syncWait) bool {
330 sw.mu.Lock()
331 defer sw.mu.Unlock()
332 if ready, ok := sw.ready[w]; ok {
333 delete(sw.ready, w)
334 w.done, w.failed = true, !ready
335 }
336 return w.done
337}
338
339// cancel stops watching w (the commit was dropped).
340func (sw *syncWaiter) cancel(w *syncWait) {
341 if w == nil || w.efd < 0 {
342 return
343 }
344 w.cancelled.Store(true)
345 sw.mu.Lock()
346 delete(sw.ready, w)
347 sw.mu.Unlock()
348 w.done = true
349 _, _ = sw.pipeW.Write([]byte{0}) // run closes its eventfd
350}
351
352func (sw *syncWaiter) stopped() bool {
353 sw.mu.Lock()
354 defer sw.mu.Unlock()
355 return sw.terminal
356}
357
358// fail publishes terminal observations before run closes its descriptors.
359// add checks the same mutex, so no new wait can miss this failure handoff.
360func (sw *syncWaiter) fail(waits []*syncWait) {
361 sw.mu.Lock()
362 sw.terminal = true
363 for _, group := range [][]*syncWait{waits, sw.added} {
364 for _, w := range group {
365 if !w.cancelled.Load() {
366 sw.ready[w] = false
367 }
368 }
369 }
370 sw.mu.Unlock()
371 sw.wake()
372}
373
374// run polls every watched eventfd; it owns and closes them.
375func (sw *syncWaiter) run(ctx context.Context) {
376 defer sw.pipeR.Close()
377 var waits []*syncWait
378 defer func() {
379 sw.mu.Lock()
380 waits = append(waits, sw.added...)
381 sw.added = nil
382 sw.mu.Unlock()
383 for _, w := range waits {
384 unix.Close(w.efd)
385 }
386 }()
387 for ctx.Err() == nil {
388 live := waits[:0]
389 for _, w := range waits {
390 if w.cancelled.Load() {
391 unix.Close(w.efd)
392 continue
393 }
394 live = append(live, w)
395 }
396 waits = live
397 fds := []unix.PollFd{{Fd: int32(sw.pipeR.Fd()), Events: unix.POLLIN}}
398 for _, w := range waits {
399 fds = append(fds, unix.PollFd{Fd: int32(w.efd), Events: unix.POLLIN})
400 }
401 n, err := sw.poller.poll(fds, 100)
402 if err != nil && !errors.Is(err, unix.EINTR) {
403 sw.fail(waits)
404 return
405 }
406 if n <= 0 {
407 continue
408 }
409 fired := false
410 kept := waits[:0]
411 for i, w := range waits {
412 if w.cancelled.Load() {
413 unix.Close(w.efd)
414 continue
415 }
416 revents := fds[i+1].Revents
417 if revents&unix.POLLIN == 0 && revents&(unix.POLLERR|unix.POLLNVAL|unix.POLLHUP) == 0 {
418 kept = append(kept, w)
419 continue
420 }
421 if revents&unix.POLLIN == 0 {
422 sw.warnPollError()
423 }
424 unix.Close(w.efd)
425 sw.mu.Lock()
426 if !w.cancelled.Load() {
427 sw.ready[w] = revents&unix.POLLIN != 0
428 fired = true
429 }
430 sw.mu.Unlock()
431 }
432 waits = kept
433 // New waits join the next poll.
434 if fds[0].Revents != 0 {
435 var b [64]byte
436 _, _ = sw.pipeR.Read(b[:])
437 sw.mu.Lock()
438 waits = append(waits, sw.added...)
439 sw.added = nil
440 sw.mu.Unlock()
441 }
442 if fired {
443 sw.wake()
444 }
445 }
446}
447
448// close stops the waiter's input side.
449func (sw *syncWaiter) close() { sw.pipeW.Close() }
450
451// syncHold is a buffer's explicit-sync state while it is on screen: the
452// acquire fence handed to readers and the release point to signal.
453type syncHold struct {
454 acquire *os.File
455 release syncPoint
456}
457
458// releaseSync signals the release point once nothing reads the buffer,
459// and closes the acquire fence.
460func (s *Server) releaseSync(h syncHold) {
461 if h.acquire != nil {
462 h.acquire.Close()
463 }
464 if h.release.set() {
465 if err := h.release.tl.dev.signal(h.release.tl.handle, h.release.point); err != nil {
466 s.log.Warn().Str("component", "wayland").Err(err).Msg("release point")
467 }
468 h.release.tl.uses--
469 s.dropTimeline(h.release.tl)
470 }
471}
472
473// dropTimeline destroys a timeline the client destroyed once no held
474// buffer still needs to signal it.
475func (s *Server) dropTimeline(tl *timeline) {
476 if !tl.alive && tl.uses <= 0 && tl.handle != 0 {
477 _ = tl.dev.destroy(tl.handle)
478 tl.handle = 0
479 }
480}
481
482// takeSyncPoints moves the surface's requested points to the commit and
483// starts waiting for the acquire point's fence.
484func (s *surface) takeSyncPoints() {
485 st := s.sync
486 if st == nil || !st.acquire.set() {
487 return
488 }
489 cs := &commitSync{acquire: st.acquire, release: st.release}
490 st.acquire, st.release = syncPoint{}, syncPoint{}
491 cs.acquire.tl.uses++
492 cs.release.tl.uses++
493 w, err := s.server.syncWait.watch(cs.acquire)
494 if err != nil {
495 // Keep points owned by the captured commit for safe discard. An
496 // unobservable acquire must never reach native rendering.
497 s.server.log.Warn().Str("component", "wayland").Err(err).Msg("acquire point")
498 }
499 cs.wait = w
500 s.next.sync = cs
501}
502
503// syncReady reports whether a commit's acquire point has its fence.
504func (s *surface) syncReady(cs *commitSync) bool {
505 return cs == nil || cs.wait == nil || s.server.syncWait.fired(cs.wait)
506}
507
508// dropSync forgets a commit that will never apply. Its release point is
509// signalled only once the acquire point has a fence: signalling earlier
510// could move a shared timeline past a point the client has not submitted.
511func (s *surface) dropSync(cs *commitSync) {
512 if cs == nil {
513 return
514 }
515 if cs.fence != nil {
516 cs.fence.Close()
517 cs.fence = nil
518 }
519 ready := cs.wait != nil && s.server.syncWait.fired(cs.wait) && !cs.wait.failed
520 if cs.wait != nil {
521 s.server.syncWait.cancel(cs.wait)
522 }
523 cs.acquire.tl.uses--
524 s.server.dropTimeline(cs.acquire.tl)
525 if ready {
526 s.server.releaseSync(syncHold{release: cs.release})
527 return
528 }
529 cs.release.tl.uses--
530 s.server.dropTimeline(cs.release.tl)
531}
532
533// prepareSync exports only after observed readiness, before any current state
534// changes. Export failure must not fall back to native implicit synchronization.
535func (s *surface) prepareSync(cs *commitSync) bool {
536 if cs == nil || cs.fence != nil {
537 return true
538 }
539 f, err := cs.acquire.tl.dev.exportSyncFile(cs.acquire.tl.handle, cs.acquire.point)
540 if err != nil || f == nil {
541 if f != nil {
542 f.Close()
543 }
544 s.server.log.Warn().Str("component", "wayland").Err(err).Msg("acquire fence export rejected")
545 return false
546 }
547 cs.fence = f
548 return true
549}
550
551// applySync makes a commit's points the current buffer's: its acquire
552// fence goes with the content, its release point waits for release.
553func (s *surface) applySync(cs *commitSync) {
554 if cs == nil {
555 return
556 }
557 if cs.wait != nil {
558 s.server.syncWait.cancel(cs.wait)
559 }
560 f := cs.fence
561 cs.fence = nil
562 cs.acquire.tl.uses--
563 s.server.dropTimeline(cs.acquire.tl)
564 s.hold = syncHold{acquire: f, release: cs.release}
565}