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.pendingNotDMABuf():
172 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorUnsupportedBuffer), "explicit sync on a non-dmabuf buffer")
173 default:
174 return true
175 }
176 return false
177}
178
179// pendingNotDMABuf reports whether the buffer of the next commit is not a
180// dmabuf: wl_shm and single-pixel buffers have no syncobj support.
181func (s *surface) pendingNotDMABuf() bool {
182 b := s.next.buffer
183 if !s.next.attached {
184 b = s.current
185 }
186 if b == nil {
187 return false
188 }
189 _, dma := s.server.buffers[b.Resource].(*dmabufBuffer)
190 return !dma
191}
192
193// syncWaiter watches acquire points: each gets an eventfd the kernel
194// writes once the point is signalled, and the pacer is woken to apply the
195// commits that waited for it.
196type acquirePoller interface {
197 poll([]unix.PollFd, int) (int, error)
198}
199
200type unixAcquirePoller struct{}
201
202func (unixAcquirePoller) poll(fds []unix.PollFd, timeout int) (int, error) {
203 return unix.Poll(fds, timeout)
204}
205
206type syncWaiter struct {
207 poller acquirePoller
208 terminal bool // polling ended in failure; protected by mu
209 mu sync.Mutex
210 ready map[*syncWait]bool
211 wake func()
212 pipeR *os.File
213 pipeW *os.File
214 added []*syncWait // new waits for run (under mu)
215 log zerowrap.Logger
216 pollError sync.Once
217}
218
219// syncWait is one acquire point a queued commit waits for.
220type syncWait struct {
221 efd int
222 done bool // the eventfd fired or the wait was cancelled (pacer, under Do)
223 failed bool // observation failed; done must not authorize publication
224 cancelled atomic.Bool // run stops polling it and closes efd
225}
226
227func newSyncWaiter(wake func()) (*syncWaiter, error) {
228 r, w, err := os.Pipe()
229 if err != nil {
230 return nil, err
231 }
232 return &syncWaiter{poller: unixAcquirePoller{}, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
233}
234
235// watch registers an eventfd for p; the wait is ready once it fires.
236func (sw *syncWaiter) watch(p syncPoint) (*syncWait, error) {
237 if sw.stopped() {
238 return &syncWait{efd: -1, done: true, failed: true}, unix.EIO
239 }
240 ready, err := p.tl.dev.signalled(p.tl.handle, p.point)
241 if err != nil || ready {
242 return &syncWait{efd: -1, done: true, failed: err != nil}, err
243 }
244 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
245 if err != nil {
246 return &syncWait{efd: -1, done: true, failed: true}, err
247 }
248 if err := p.tl.dev.eventfd(p.tl.handle, p.point, efd); err != nil {
249 unix.Close(efd)
250 return &syncWait{efd: -1, done: true, failed: true}, err
251 }
252 w := sw.add(efd)
253 if w.failed {
254 return w, unix.EIO
255 }
256 return w, nil
257}
258
259// watchImplicit waits for a dma-buf's exclusive fences without blocking dispatch.
260func (sw *syncWaiter) watchImplicit(b *ports.DMABuf, waits *[4]*syncWait) error {
261 if sw.stopped() {
262 return unix.EIO
263 }
264 var fds [4]unix.PollFd
265 for i, p := range b.Planes {
266 if i >= len(fds) {
267 break
268 }
269 fds[i] = unix.PollFd{Fd: int32(p.File.Fd()), Events: unix.POLLIN}
270 }
271 planes := fds[:min(len(b.Planes), len(fds))]
272 _, err := unix.Poll(planes, 0)
273 for errors.Is(err, unix.EINTR) { // a signal, not a fence error
274 _, err = unix.Poll(planes, 0)
275 }
276 if err != nil {
277 return err
278 }
279 for i, fd := range planes {
280 if fd.Revents&unix.POLLIN != 0 {
281 continue
282 }
283 if fd.Revents&(unix.POLLERR|unix.POLLNVAL|unix.POLLHUP) != 0 {
284 sw.warnPollError()
285 return unix.EIO
286 }
287 dup, err := unix.FcntlInt(uintptr(fd.Fd), unix.F_DUPFD_CLOEXEC, 0)
288 if err != nil {
289 for j, w := range waits {
290 sw.cancel(w)
291 waits[j] = nil
292 }
293 return err
294 }
295 waits[i] = sw.add(dup)
296 if waits[i].failed {
297 for j, w := range waits {
298 if w != nil {
299 sw.cancel(w)
300 waits[j] = nil
301 }
302 }
303 return unix.EIO
304 }
305 }
306 return nil
307}
308
309func (sw *syncWaiter) warnPollError() {
310 sw.pollError.Do(func() {
311 sw.log.Warn().Msg("implicit fence poll error")
312 })
313}
314
315func (sw *syncWaiter) add(fd int) *syncWait {
316 w := &syncWait{efd: fd}
317 sw.mu.Lock()
318 if sw.terminal {
319 sw.mu.Unlock()
320 unix.Close(fd)
321 return &syncWait{efd: -1, done: true, failed: true}
322 }
323 sw.added = append(sw.added, w)
324 sw.mu.Unlock()
325 _, _ = sw.pipeW.Write([]byte{0})
326 return w
327}
328
329// fired reports whether w's point is signalled, and forgets w once it has.
330func (sw *syncWaiter) fired(w *syncWait) bool {
331 sw.mu.Lock()
332 defer sw.mu.Unlock()
333 if ready, ok := sw.ready[w]; ok {
334 delete(sw.ready, w)
335 w.done, w.failed = true, !ready
336 }
337 return w.done
338}
339
340// cancel stops watching w (the commit was dropped).
341func (sw *syncWaiter) cancel(w *syncWait) {
342 if w == nil || w.efd < 0 {
343 return
344 }
345 w.cancelled.Store(true)
346 sw.mu.Lock()
347 delete(sw.ready, w)
348 sw.mu.Unlock()
349 w.done = true
350 _, _ = sw.pipeW.Write([]byte{0}) // run closes its eventfd
351}
352
353func (sw *syncWaiter) stopped() bool {
354 sw.mu.Lock()
355 defer sw.mu.Unlock()
356 return sw.terminal
357}
358
359// fail publishes terminal observations before run closes its descriptors.
360// add checks the same mutex, so no new wait can miss this failure handoff.
361func (sw *syncWaiter) fail(waits []*syncWait) {
362 sw.mu.Lock()
363 sw.terminal = true
364 for _, group := range [][]*syncWait{waits, sw.added} {
365 for _, w := range group {
366 if !w.cancelled.Load() {
367 sw.ready[w] = false
368 }
369 }
370 }
371 sw.mu.Unlock()
372 sw.wake()
373}
374
375// run polls every watched eventfd; it owns and closes them.
376func (sw *syncWaiter) run(ctx context.Context) {
377 defer sw.pipeR.Close()
378 var waits []*syncWait
379 defer func() {
380 sw.mu.Lock()
381 waits = append(waits, sw.added...)
382 sw.added = nil
383 sw.mu.Unlock()
384 for _, w := range waits {
385 unix.Close(w.efd)
386 }
387 }()
388 for ctx.Err() == nil {
389 live := waits[:0]
390 for _, w := range waits {
391 if w.cancelled.Load() {
392 unix.Close(w.efd)
393 continue
394 }
395 live = append(live, w)
396 }
397 waits = live
398 fds := []unix.PollFd{{Fd: int32(sw.pipeR.Fd()), Events: unix.POLLIN}}
399 for _, w := range waits {
400 fds = append(fds, unix.PollFd{Fd: int32(w.efd), Events: unix.POLLIN})
401 }
402 n, err := sw.poller.poll(fds, 100)
403 if err != nil && !errors.Is(err, unix.EINTR) {
404 sw.fail(waits)
405 return
406 }
407 if n <= 0 {
408 continue
409 }
410 fired := false
411 kept := waits[:0]
412 for i, w := range waits {
413 if w.cancelled.Load() {
414 unix.Close(w.efd)
415 continue
416 }
417 revents := fds[i+1].Revents
418 if revents&unix.POLLIN == 0 && revents&(unix.POLLERR|unix.POLLNVAL|unix.POLLHUP) == 0 {
419 kept = append(kept, w)
420 continue
421 }
422 if revents&unix.POLLIN == 0 {
423 sw.warnPollError()
424 }
425 unix.Close(w.efd)
426 sw.mu.Lock()
427 if !w.cancelled.Load() {
428 sw.ready[w] = revents&unix.POLLIN != 0
429 fired = true
430 }
431 sw.mu.Unlock()
432 }
433 waits = kept
434 // New waits join the next poll.
435 if fds[0].Revents != 0 {
436 var b [64]byte
437 _, _ = sw.pipeR.Read(b[:])
438 sw.mu.Lock()
439 waits = append(waits, sw.added...)
440 sw.added = nil
441 sw.mu.Unlock()
442 }
443 if fired {
444 sw.wake()
445 }
446 }
447}
448
449// close stops the waiter's input side.
450func (sw *syncWaiter) close() { sw.pipeW.Close() }
451
452// syncHold is a buffer's explicit-sync state while it is on screen: the
453// acquire fence handed to readers and the release point to signal.
454type syncHold struct {
455 acquire *os.File
456 release syncPoint
457}
458
459// releaseSync signals the release point once nothing reads the buffer,
460// and closes the acquire fence.
461func (s *Server) releaseSync(h syncHold) {
462 if h.acquire != nil {
463 h.acquire.Close()
464 }
465 if h.release.set() {
466 if err := h.release.tl.dev.signal(h.release.tl.handle, h.release.point); err != nil {
467 s.log.Warn().Err(err).Msg("release point")
468 }
469 h.release.tl.uses--
470 s.dropTimeline(h.release.tl)
471 }
472}
473
474// dropTimeline destroys a timeline the client destroyed once no held
475// buffer still needs to signal it.
476func (s *Server) dropTimeline(tl *timeline) {
477 if !tl.alive && tl.uses <= 0 && tl.handle != 0 {
478 _ = tl.dev.destroy(tl.handle)
479 tl.handle = 0
480 }
481}
482
483// takeSyncPoints moves the surface's requested points to the commit and
484// starts waiting for the acquire point's fence.
485func (s *surface) takeSyncPoints() {
486 st := s.sync
487 if st == nil || !st.acquire.set() {
488 return
489 }
490 // The commit's points wait in the surface until queueUpdate moves them
491 // into the update: no allocation per commit.
492 cs := &s.syncScratch
493 *cs = commitSync{acquire: st.acquire, release: st.release}
494 st.acquire, st.release = syncPoint{}, syncPoint{}
495 cs.acquire.tl.uses++
496 cs.release.tl.uses++
497 w, err := s.server.syncWait.watch(cs.acquire)
498 if err != nil {
499 // Keep points owned by the captured commit for safe discard. An
500 // unobservable acquire must never reach native rendering.
501 s.server.log.Warn().Err(err).Msg("acquire point")
502 }
503 cs.wait = w
504 s.next.sync = cs
505}
506
507// syncReady reports whether a commit's acquire point has its fence.
508func (s *surface) syncReady(cs *commitSync) bool {
509 return cs == nil || cs.wait == nil || s.server.syncWait.fired(cs.wait)
510}
511
512// dropSync forgets a commit that will never apply. Its release point is
513// signalled only once the acquire point has a fence: signalling earlier
514// could move a shared timeline past a point the client has not submitted.
515func (s *surface) dropSync(cs *commitSync) {
516 if cs == nil {
517 return
518 }
519 if cs.fence != nil {
520 cs.fence.Close()
521 cs.fence = nil
522 }
523 ready := cs.wait != nil && s.server.syncWait.fired(cs.wait) && !cs.wait.failed
524 if cs.wait != nil {
525 s.server.syncWait.cancel(cs.wait)
526 }
527 cs.acquire.tl.uses--
528 s.server.dropTimeline(cs.acquire.tl)
529 if ready {
530 s.server.releaseSync(syncHold{release: cs.release})
531 return
532 }
533 cs.release.tl.uses--
534 s.server.dropTimeline(cs.release.tl)
535}
536
537// prepareSync exports only after observed readiness, before any current state
538// changes. Export failure must not fall back to native implicit synchronization.
539func (s *surface) prepareSync(cs *commitSync) bool {
540 if cs == nil || cs.fence != nil {
541 return true
542 }
543 f, err := cs.acquire.tl.dev.exportSyncFile(cs.acquire.tl.handle, cs.acquire.point)
544 if err != nil || f == nil {
545 if f != nil {
546 f.Close()
547 }
548 s.server.log.Warn().Err(err).Msg("acquire fence export rejected")
549 return false
550 }
551 cs.fence = f
552 return true
553}
554
555// applySync makes a commit's points the current buffer's: its acquire
556// fence goes with the content, its release point waits for release.
557func (s *surface) applySync(cs *commitSync) {
558 if cs == nil {
559 return
560 }
561 if cs.wait != nil {
562 s.server.syncWait.cancel(cs.wait)
563 }
564 f := cs.fence
565 cs.fence = nil
566 cs.acquire.tl.uses--
567 s.server.dropTimeline(cs.acquire.tl)
568 s.hold = syncHold{acquire: f, release: cs.release}
569}