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