1package wayland
2
3import (
4 "context"
5 "errors"
6 "os"
7 "sync"
8 "sync/atomic"
9
10 "github.com/bnema/purego-libwayland/protocol/linuxdrmsyncobj"
11 "github.com/bnema/purego-libwayland/protocol/wayland"
12 "github.com/bnema/purego-libwayland/server"
13 "golang.org/x/sys/unix"
14)
15
16// wp_linux_drm_syncobj_v1: explicit sync. A client names, per commit, the
17// timeline point its GPU signals when the buffer is drawn (acquire) and
18// the point we signal once nothing reads the buffer any more (release).
19// A commit waits (like a fifo barrier) until its acquire point has a
20// fence; the fence then travels with the content as a sync file the
21// renderer and the display wait on, so no CPU waits for the GPU. The
22// release point replaces wl_buffer.release.
23
24const syncobjVersion = 1
25
26// timeline is an imported client timeline.
27type timeline struct {
28 dev syncobjDevice
29 handle uint32
30 alive bool
31 // uses counts committed points still to signal or wait on.
32 uses int
33}
34
35// syncPoint is a point on a timeline.
36type syncPoint struct {
37 tl *timeline
38 point uint64
39}
40
41func (p syncPoint) set() bool { return p.tl != nil }
42
43// syncState is a surface's wp_linux_drm_syncobj_surface_v1 state.
44type syncState struct {
45 resource *linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1
46 acquire, release syncPoint // pending, for the next commit
47}
48
49func registerSyncobj(d *server.Display, s *Server) error {
50 if s.syncDev == nil {
51 return nil
52 }
53 return linuxdrmsyncobj.NewWpLinuxDrmSyncobjManagerV1Global(d, syncobjVersion, func(c server.Client, v, id uint32) {
54 _, _ = linuxdrmsyncobj.NewWpLinuxDrmSyncobjManagerV1(c, int32(v), id, syncManager{s})
55 })
56}
57
58type syncManager struct{ server *Server }
59
60func (syncManager) Destroy(*linuxdrmsyncobj.WpLinuxDrmSyncobjManagerV1) {}
61
62func (m syncManager) GetSurface(r *linuxdrmsyncobj.WpLinuxDrmSyncobjManagerV1, id uint32, surf *wayland.Surface) {
63 state := m.server.surfaceOf(surf)
64 if state == nil {
65 return
66 }
67 if state.sync != nil {
68 r.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjManagerV1ErrorSurfaceExists), "surface already has a syncobj surface")
69 return
70 }
71 st := &syncState{}
72 res, err := linuxdrmsyncobj.NewWpLinuxDrmSyncobjSurfaceV1(r.Client(), r.Version(), id, syncSurface{state, st})
73 if err != nil {
74 return
75 }
76 st.resource = res
77 state.sync = st
78 res.OnDestroy = func() {
79 if state.sync == st {
80 state.sync = nil
81 }
82 }
83}
84
85func (m syncManager) ImportTimeline(r *linuxdrmsyncobj.WpLinuxDrmSyncobjManagerV1, id uint32, fd int) {
86 defer unix.Close(fd)
87 h, err := m.server.syncDev.fdToHandle(fd)
88 if err != nil {
89 r.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjManagerV1ErrorInvalidTimeline), "invalid timeline fd")
90 return
91 }
92 tl := &timeline{dev: m.server.syncDev, handle: h, alive: true}
93 res, err := linuxdrmsyncobj.NewWpLinuxDrmSyncobjTimelineV1(r.Client(), r.Version(), id, timelineHandler{})
94 if err != nil {
95 _ = tl.dev.destroy(h)
96 return
97 }
98 m.server.timelines[res.Resource] = tl
99 res.OnDestroy = func() {
100 delete(m.server.timelines, res.Resource)
101 // Points already committed keep the syncobj: destroyed once no
102 // held buffer or waiting commit uses it (timeline.drop).
103 tl.alive = false
104 m.server.dropTimeline(tl)
105 }
106}
107
108type timelineHandler struct{}
109
110func (timelineHandler) Destroy(*linuxdrmsyncobj.WpLinuxDrmSyncobjTimelineV1) {}
111
112type syncSurface struct {
113 surf *surface
114 st *syncState
115}
116
117func (syncSurface) Destroy(*linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1) {}
118
119func (h syncSurface) point(r *linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1, tl *linuxdrmsyncobj.WpLinuxDrmSyncobjTimelineV1, hi, lo uint32) (syncPoint, bool) {
120 if h.surf.destroyed {
121 r.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorNoSurface), "surface destroyed")
122 return syncPoint{}, false
123 }
124 t := h.surf.server.timelines[tl.Resource]
125 if t == nil {
126 return syncPoint{}, false
127 }
128 return syncPoint{tl: t, point: uint64(hi)<<32 | uint64(lo)}, true
129}
130
131func (h syncSurface) SetAcquirePoint(r *linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1, tl *linuxdrmsyncobj.WpLinuxDrmSyncobjTimelineV1, hi, lo uint32) {
132 if p, ok := h.point(r, tl, hi, lo); ok {
133 h.st.acquire = p
134 }
135}
136
137func (h syncSurface) SetReleasePoint(r *linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1, tl *linuxdrmsyncobj.WpLinuxDrmSyncobjTimelineV1, hi, lo uint32) {
138 if p, ok := h.point(r, tl, hi, lo); ok {
139 h.st.release = p
140 }
141}
142
143// checkSyncCommit applies the protocol rules to a commit with sync
144// points; false means a protocol error was posted.
145func (s *surface) checkSyncCommit() bool {
146 st := s.sync
147 if st == nil {
148 return true
149 }
150 a, r := st.acquire, st.release
151 // Points belong only to a commit that attaches a non-null buffer.
152 hasBuffer := s.attached && s.pending != nil
153 res := st.resource
154 switch {
155 case !a.set() && !r.set():
156 if hasBuffer {
157 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorNoAcquirePoint), "buffer without acquire point")
158 return false
159 }
160 return true
161 case !hasBuffer:
162 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorNoBuffer), "points without a buffer")
163 case !a.set():
164 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorNoAcquirePoint), "no acquire point")
165 case !r.set():
166 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorNoReleasePoint), "no release point")
167 case a.tl == r.tl && a.point >= r.point:
168 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorConflictingPoints), "release point not after acquire point")
169 case s.pendingIsSHM():
170 res.PostError(uint32(linuxdrmsyncobj.WpLinuxDrmSyncobjSurfaceV1ErrorUnsupportedBuffer), "explicit sync on a wl_shm buffer")
171 default:
172 return true
173 }
174 return false
175}
176
177// pendingIsSHM reports whether the buffer of the next commit is wl_shm.
178func (s *surface) pendingIsSHM() bool {
179 b := s.pending
180 if !s.attached {
181 b = s.current
182 }
183 if b == nil {
184 return false
185 }
186 _, dma := s.server.buffers[b.Resource].(*dmabufBuffer)
187 return !dma
188}
189
190// syncWaiter watches acquire points: each gets an eventfd the kernel
191// writes once the point has a fence, and the pacer is woken to apply the
192// commits that waited for it.
193type syncWaiter struct {
194 mu sync.Mutex
195 ready map[*syncWait]bool
196 wake func()
197 pipeR *os.File
198 pipeW *os.File
199 added []*syncWait // new waits for run (under mu)
200}
201
202// syncWait is one acquire point a queued commit waits for.
203type syncWait struct {
204 efd int
205 done bool // the eventfd fired or the wait was cancelled (pacer, under Do)
206 cancelled atomic.Bool // run stops polling it and closes efd
207}
208
209func newSyncWaiter(wake func()) (*syncWaiter, error) {
210 r, w, err := os.Pipe()
211 if err != nil {
212 return nil, err
213 }
214 return &syncWaiter{ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w}, nil
215}
216
217// watch registers an eventfd for p; the wait is ready once it fires.
218func (sw *syncWaiter) watch(p syncPoint) (*syncWait, error) {
219 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
220 if err != nil {
221 return nil, err
222 }
223 if err := p.tl.dev.eventfd(p.tl.handle, p.point, efd); err != nil {
224 unix.Close(efd)
225 return nil, err
226 }
227 w := &syncWait{efd: efd}
228 sw.mu.Lock()
229 sw.added = append(sw.added, w)
230 sw.mu.Unlock()
231 _, _ = sw.pipeW.Write([]byte{0})
232 return w, nil
233}
234
235// fired reports whether w's point has a fence, and forgets w once it has.
236func (sw *syncWaiter) fired(w *syncWait) bool {
237 sw.mu.Lock()
238 defer sw.mu.Unlock()
239 if sw.ready[w] {
240 delete(sw.ready, w)
241 w.done = true
242 }
243 return w.done
244}
245
246// cancel stops watching w (the commit was dropped).
247func (sw *syncWaiter) cancel(w *syncWait) {
248 w.cancelled.Store(true)
249 sw.mu.Lock()
250 delete(sw.ready, w)
251 sw.mu.Unlock()
252 w.done = true
253 _, _ = sw.pipeW.Write([]byte{0}) // run closes its eventfd
254}
255
256// run polls every watched eventfd; it owns and closes them.
257func (sw *syncWaiter) run(ctx context.Context) {
258 defer sw.pipeR.Close()
259 var waits []*syncWait
260 defer func() {
261 sw.mu.Lock()
262 waits = append(waits, sw.added...)
263 sw.added = nil
264 sw.mu.Unlock()
265 for _, w := range waits {
266 unix.Close(w.efd)
267 }
268 }()
269 for ctx.Err() == nil {
270 live := waits[:0]
271 for _, w := range waits {
272 if w.cancelled.Load() {
273 unix.Close(w.efd)
274 continue
275 }
276 live = append(live, w)
277 }
278 waits = live
279 fds := []unix.PollFd{{Fd: int32(sw.pipeR.Fd()), Events: unix.POLLIN}}
280 for _, w := range waits {
281 fds = append(fds, unix.PollFd{Fd: int32(w.efd), Events: unix.POLLIN})
282 }
283 n, err := unix.Poll(fds, 100)
284 if err != nil && !errors.Is(err, unix.EINTR) {
285 return
286 }
287 if n <= 0 {
288 continue
289 }
290 fired := false
291 kept := waits[:0]
292 for i, w := range waits {
293 if w.cancelled.Load() {
294 unix.Close(w.efd)
295 continue
296 }
297 if fds[i+1].Revents == 0 {
298 kept = append(kept, w)
299 continue
300 }
301 unix.Close(w.efd)
302 sw.mu.Lock()
303 sw.ready[w] = true
304 sw.mu.Unlock()
305 fired = true
306 }
307 waits = kept
308 // New waits join the next poll.
309 if fds[0].Revents != 0 {
310 var b [64]byte
311 _, _ = sw.pipeR.Read(b[:])
312 sw.mu.Lock()
313 waits = append(waits, sw.added...)
314 sw.added = nil
315 sw.mu.Unlock()
316 }
317 if fired {
318 sw.wake()
319 }
320 }
321}
322
323// close stops the waiter's input side.
324func (sw *syncWaiter) close() { sw.pipeW.Close() }
325
326// syncHold is a buffer's explicit-sync state while it is on screen: the
327// acquire fence handed to readers and the release point to signal.
328type syncHold struct {
329 acquire *os.File
330 release syncPoint
331}
332
333// releaseSync signals the release point once nothing reads the buffer,
334// and closes the acquire fence.
335func (s *Server) releaseSync(h syncHold) {
336 if h.acquire != nil {
337 h.acquire.Close()
338 }
339 if h.release.set() {
340 if err := h.release.tl.dev.signal(h.release.tl.handle, h.release.point); err != nil {
341 s.log.Warn().Str("component", "wayland").Err(err).Msg("release point")
342 }
343 h.release.tl.uses--
344 s.dropTimeline(h.release.tl)
345 }
346}
347
348// dropTimeline destroys a timeline the client destroyed once no held
349// buffer still needs to signal it.
350func (s *Server) dropTimeline(tl *timeline) {
351 if !tl.alive && tl.uses <= 0 && tl.handle != 0 {
352 _ = tl.dev.destroy(tl.handle)
353 tl.handle = 0
354 }
355}
356
357// takeSyncPoints moves the surface's requested points to the commit and
358// starts waiting for the acquire point's fence.
359func (s *surface) takeSyncPoints() {
360 st := s.sync
361 if st == nil || !st.acquire.set() {
362 return
363 }
364 cs := &commitSync{acquire: st.acquire, release: st.release}
365 st.acquire, st.release = syncPoint{}, syncPoint{}
366 cs.acquire.tl.uses++
367 cs.release.tl.uses++
368 w, err := s.server.syncWait.watch(cs.acquire)
369 if err != nil {
370 // The point cannot be watched: apply now, readers wait on the
371 // fence (or its absence) when they read the buffer.
372 s.server.log.Warn().Str("component", "wayland").Err(err).Msg("acquire point")
373 } else {
374 cs.wait = w
375 }
376 s.pendingSync = cs
377}
378
379// syncReady reports whether a commit's acquire point has its fence.
380func (s *surface) syncReady(cs *commitSync) bool {
381 return cs == nil || cs.wait == nil || s.server.syncWait.fired(cs.wait)
382}
383
384// dropSync forgets a commit that will never apply. Its release point is
385// signalled only once the acquire point has a fence: signalling earlier
386// could move a shared timeline past a point the client has not submitted.
387func (s *surface) dropSync(cs *commitSync) {
388 if cs == nil {
389 return
390 }
391 ready := cs.wait != nil && s.server.syncWait.fired(cs.wait)
392 if cs.wait != nil {
393 s.server.syncWait.cancel(cs.wait)
394 }
395 cs.acquire.tl.uses--
396 s.server.dropTimeline(cs.acquire.tl)
397 if ready {
398 s.server.releaseSync(syncHold{release: cs.release})
399 return
400 }
401 cs.release.tl.uses--
402 s.server.dropTimeline(cs.release.tl)
403}
404
405// applySync makes a commit's points the current buffer's: its acquire
406// fence goes with the content, its release point waits for release.
407func (s *surface) applySync(cs *commitSync) {
408 if cs == nil {
409 return
410 }
411 if cs.wait != nil {
412 s.server.syncWait.cancel(cs.wait)
413 }
414 f, err := cs.acquire.tl.dev.exportSyncFile(cs.acquire.tl.handle, cs.acquire.point)
415 if err != nil {
416 s.server.log.Warn().Str("component", "wayland").Err(err).Msg("acquire fence")
417 f = nil
418 }
419 cs.acquire.tl.uses--
420 s.server.dropTimeline(cs.acquire.tl)
421 s.hold = syncHold{acquire: f, release: cs.release}
422}