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