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
194type acquirePoller interface {
195 poll([]unix.PollFd, int) (int, error)
196}
197
198type unixAcquirePoller struct{ waiter *fdwait.Poller }
199
200func (p unixAcquirePoller) poll(fds []unix.PollFd, timeout int) (int, error) {
201 return p.waiter.Poll(fds, timeout)
202}
203
204// acquireSource registers an acquire eventfd with the display loop: onReadable
205// runs on the display goroutine while fd is readable. Display goroutine only.
206type acquireSource interface {
207 add(fd int, onReadable func()) (acquireRemover, error)
208}
209
210// acquireRemover unregisters a source added by acquireSource.
211type acquireRemover interface {
212 Remove()
213}
214
215type displayAcquireSource struct{ d *server.Display }
216
217func (s displayAcquireSource) add(fd int, onReadable func()) (acquireRemover, error) {
218 src, err := s.d.AddFD(fd, onReadable)
219 if err != nil {
220 return nil, err
221 }
222 return src, nil
223}
224
225// syncWaiter watches acquire points: each gets an eventfd the kernel
226// writes once the point is signalled, and the pacer is woken to apply the
227// commits that waited for it. Explicit waits are display-loop sources; the
228// goroutine (run) with its pipe polls implicit dma-buf fences only.
229type syncWaiter struct {
230 // src and pace serve explicit waits, which run on the display loop:
231 // src registers the eventfd there, pace applies what the fired wait
232 // unblocked (display goroutine only).
233 src acquireSource
234 pace func()
235 pending map[*syncWait]struct{} // explicit waits still registered (display goroutine, then Run's exit)
236 poller acquirePoller
237 waiter *fdwait.Poller
238 terminal bool // polling ended in failure; protected by mu
239 mu sync.Mutex
240 ready map[*syncWait]bool
241 wake func()
242 pipeR *os.File
243 pipeW *os.File
244 added []*syncWait // new waits for run (under mu)
245 log zerowrap.Logger
246 pollError sync.Once
247}
248
249// syncWait is one acquire point a queued commit waits for. An explicit wait
250// lives on the display loop only; an implicit one is polled by run.
251type syncWait struct {
252 efd int
253 src acquireRemover // explicit: the display-loop registration, nil once removed
254 explicit bool
255 done bool // the eventfd fired or the wait was cancelled (display goroutine)
256 failed bool // observation failed; done must not authorize publication
257 cancelled atomic.Bool // run stops polling it and closes efd
258}
259
260func newSyncWaiter(wake func()) (*syncWaiter, error) {
261 r, w, err := os.Pipe()
262 if err != nil {
263 return nil, err
264 }
265 p, err := fdwait.New()
266 if err != nil {
267 r.Close()
268 w.Close()
269 return nil, err
270 }
271 return &syncWaiter{poller: unixAcquirePoller{p}, waiter: p, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
272}
273
274// watch registers an eventfd for p on the display loop; the wait is ready once
275// it fires. Display goroutine only.
276func (sw *syncWaiter) watch(p syncPoint) (*syncWait, error) {
277 ready, err := p.tl.dev.signalled(p.tl.handle, p.point)
278 if err != nil || ready {
279 return &syncWait{efd: -1, explicit: true, done: true, failed: err != nil}, err
280 }
281 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
282 if err != nil {
283 return &syncWait{efd: -1, explicit: true, done: true, failed: true}, err
284 }
285 if err := p.tl.dev.eventfd(p.tl.handle, p.point, efd); err != nil {
286 unix.Close(efd)
287 return &syncWait{efd: -1, explicit: true, done: true, failed: true}, err
288 }
289 w := &syncWait{efd: efd, explicit: true}
290 src, err := sw.src.add(efd, func() { sw.onAcquire(w) })
291 if err != nil {
292 unix.Close(efd)
293 return &syncWait{efd: -1, explicit: true, done: true, failed: true}, err
294 }
295 w.src = src
296 if sw.pending == nil {
297 sw.pending = map[*syncWait]struct{}{}
298 }
299 sw.pending[w] = struct{}{}
300 return w, nil
301}
302
303// onAcquire is the eventfd's callback, on the display goroutine.
304func (sw *syncWaiter) onAcquire(w *syncWait) {
305 if w.done {
306 return
307 }
308 var b [8]byte
309 if _, err := unix.Read(w.efd, b[:]); err != nil {
310 if errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EINTR) {
311 return // not signalled yet: keep waiting
312 }
313 w.failed = true
314 }
315 sw.finish(w)
316 sw.pace()
317}
318
319// finish unregisters then closes an explicit wait's eventfd and marks it done.
320func (sw *syncWaiter) finish(w *syncWait) {
321 delete(sw.pending, w)
322 if w.src != nil {
323 w.src.Remove()
324 w.src = nil
325 }
326 if w.efd >= 0 {
327 unix.Close(w.efd)
328 w.efd = -1
329 }
330 w.done = true
331}
332
333// watchImplicit waits for a dma-buf's exclusive fences without blocking dispatch.
334func (sw *syncWaiter) watchImplicit(b *ports.DMABuf, waits *[4]*syncWait) error {
335 if sw.stopped() {
336 return unix.EIO
337 }
338 var fds [4]unix.PollFd
339 for i, p := range b.Planes {
340 if i >= len(fds) {
341 break
342 }
343 fds[i] = unix.PollFd{Fd: int32(p.File.Fd()), Events: unix.POLLIN}
344 }
345 planes := fds[:min(len(b.Planes), len(fds))]
346 _, err := unix.Poll(planes, 0)
347 for errors.Is(err, unix.EINTR) { // a signal, not a fence error
348 _, err = unix.Poll(planes, 0)
349 }
350 if err != nil {
351 return err
352 }
353 for i, fd := range planes {
354 if fd.Revents&unix.POLLIN != 0 {
355 continue
356 }
357 if fd.Revents&(unix.POLLERR|unix.POLLNVAL|unix.POLLHUP) != 0 {
358 sw.warnPollError()
359 return unix.EIO
360 }
361 dup, err := unix.FcntlInt(uintptr(fd.Fd), unix.F_DUPFD_CLOEXEC, 0)
362 if err != nil {
363 for j, w := range waits {
364 sw.cancel(w)
365 waits[j] = nil
366 }
367 return err
368 }
369 waits[i] = sw.add(dup)
370 if waits[i].failed {
371 for j, w := range waits {
372 if w != nil {
373 sw.cancel(w)
374 waits[j] = nil
375 }
376 }
377 return unix.EIO
378 }
379 }
380 return nil
381}
382
383func (sw *syncWaiter) warnPollError() {
384 sw.pollError.Do(func() {
385 sw.log.Warn().Msg("implicit fence poll error")
386 })
387}
388
389func (sw *syncWaiter) add(fd int) *syncWait {
390 w := &syncWait{efd: fd}
391 sw.mu.Lock()
392 if sw.terminal {
393 sw.mu.Unlock()
394 unix.Close(fd)
395 return &syncWait{efd: -1, done: true, failed: true}
396 }
397 sw.added = append(sw.added, w)
398 sw.mu.Unlock()
399 _, _ = sw.pipeW.Write([]byte{0})
400 return w
401}
402
403// fired reports whether w's point is signalled, and forgets w once it has.
404func (sw *syncWaiter) fired(w *syncWait) bool {
405 if w.explicit {
406 return w.done // display goroutine only
407 }
408 sw.mu.Lock()
409 defer sw.mu.Unlock()
410 if ready, ok := sw.ready[w]; ok {
411 delete(sw.ready, w)
412 w.done, w.failed = true, !ready
413 }
414 return w.done
415}
416
417// cancel stops watching w (the commit was dropped).
418func (sw *syncWaiter) cancel(w *syncWait) {
419 if w == nil {
420 return
421 }
422 if w.explicit {
423 sw.finish(w)
424 return
425 }
426 if w.efd < 0 {
427 return
428 }
429 w.cancelled.Store(true)
430 sw.mu.Lock()
431 delete(sw.ready, w)
432 sw.mu.Unlock()
433 w.done = true
434 _, _ = sw.pipeW.Write([]byte{0}) // run closes its fence fd
435}
436
437func (sw *syncWaiter) stopped() bool {
438 sw.mu.Lock()
439 defer sw.mu.Unlock()
440 return sw.terminal
441}
442
443// fail publishes terminal observations before run closes its descriptors.
444// add checks the same mutex, so no new wait can miss this failure handoff.
445func (sw *syncWaiter) fail(waits []*syncWait) {
446 sw.mu.Lock()
447 sw.terminal = true
448 for _, group := range [][]*syncWait{waits, sw.added} {
449 for _, w := range group {
450 if !w.cancelled.Load() {
451 sw.ready[w] = false
452 }
453 }
454 }
455 sw.mu.Unlock()
456 sw.wake()
457}
458
459// run polls the implicit dma-buf fence descriptors; it owns and closes them.
460func (sw *syncWaiter) run(ctx context.Context) {
461 defer sw.pipeR.Close()
462 defer sw.waiter.Close()
463 pipeFD := int(sw.pipeR.Fd())
464 // The pipe lives as long as the poller; the fence fds come and go. The
465 // poll sleeps until a fence, a new or cancelled wait (the pipe), or the
466 // end of ctx.
467 var waits []*syncWait
468 defer func() {
469 sw.mu.Lock()
470 waits = append(waits, sw.added...)
471 sw.added = nil
472 sw.mu.Unlock()
473 for _, w := range waits {
474 unix.Close(w.efd)
475 }
476 }()
477 sw.waiter.StopOn(ctx)
478 if err := sw.waiter.Pin(int32(pipeFD)); err != nil {
479 sw.log.Warn().Err(err).Msg("sync waiter")
480 sw.fail(nil)
481 return
482 }
483 for ctx.Err() == nil {
484 live := waits[:0]
485 for _, w := range waits {
486 if w.cancelled.Load() {
487 unix.Close(w.efd)
488 continue
489 }
490 live = append(live, w)
491 }
492 waits = live
493 fds := []unix.PollFd{{Fd: int32(pipeFD), Events: unix.POLLIN}}
494 for _, w := range waits {
495 fds = append(fds, unix.PollFd{Fd: int32(w.efd), Events: unix.POLLIN})
496 }
497 n, err := sw.poller.poll(fds, -1)
498 if err != nil && !errors.Is(err, unix.EINTR) {
499 sw.fail(waits)
500 return
501 }
502 if n <= 0 {
503 continue
504 }
505 fired := false
506 kept := waits[:0]
507 for i, w := range waits {
508 if w.cancelled.Load() {
509 unix.Close(w.efd)
510 continue
511 }
512 revents := fds[i+1].Revents
513 if revents&unix.POLLIN == 0 && revents&(unix.POLLERR|unix.POLLNVAL|unix.POLLHUP) == 0 {
514 kept = append(kept, w)
515 continue
516 }
517 if revents&unix.POLLIN == 0 {
518 sw.warnPollError()
519 }
520 unix.Close(w.efd)
521 sw.mu.Lock()
522 if !w.cancelled.Load() {
523 sw.ready[w] = revents&unix.POLLIN != 0
524 fired = true
525 }
526 sw.mu.Unlock()
527 }
528 waits = kept
529 // New waits join the next poll.
530 if fds[0].Revents != 0 {
531 var b [64]byte
532 _, _ = sw.pipeR.Read(b[:])
533 sw.mu.Lock()
534 waits = append(waits, sw.added...)
535 sw.added = nil
536 sw.mu.Unlock()
537 }
538 if fired {
539 sw.wake()
540 }
541 }
542}
543
544// close stops the waiter's input side.
545func (sw *syncWaiter) close() { sw.pipeW.Close() }
546
547// discard releases what run would have closed, for a waiter whose run
548// never starts.
549func (sw *syncWaiter) discard() {
550 sw.pipeR.Close()
551 sw.waiter.Close()
552}
553
554// syncHold is a buffer's explicit-sync state while it is on screen: the
555// acquire fence handed to readers and the release point to signal.
556type syncHold struct {
557 acquire *os.File
558 release syncPoint
559}
560
561// releaseSync signals the release point once nothing reads the buffer,
562// and closes the acquire fence.
563func (s *Server) releaseSync(h syncHold) {
564 if h.acquire != nil {
565 h.acquire.Close()
566 }
567 if h.release.set() {
568 if err := h.release.tl.dev.signal(h.release.tl.handle, h.release.point); err != nil {
569 s.log.Warn().Err(err).Msg("release point")
570 }
571 h.release.tl.uses--
572 s.dropTimeline(h.release.tl)
573 }
574}
575
576// dropTimeline destroys a timeline the client destroyed once no held
577// buffer still needs to signal it.
578func (s *Server) dropTimeline(tl *timeline) {
579 if !tl.alive && tl.uses <= 0 && tl.handle != 0 {
580 _ = tl.dev.destroy(tl.handle)
581 tl.handle = 0
582 }
583}
584
585// takeSyncPoints moves the surface's requested points to the commit and
586// starts waiting for the acquire point's fence.
587func (s *surface) takeSyncPoints() {
588 st := s.sync
589 if st == nil || !st.acquire.set() {
590 return
591 }
592 // The commit's points wait in the surface until queueUpdate moves them
593 // into the update: no allocation per commit.
594 cs := &s.syncScratch
595 *cs = commitSync{acquire: st.acquire, release: st.release}
596 st.acquire, st.release = syncPoint{}, syncPoint{}
597 cs.acquire.tl.uses++
598 cs.release.tl.uses++
599 w, err := s.server.syncWait.watch(cs.acquire)
600 if err != nil {
601 // Keep points owned by the captured commit for safe discard. An
602 // unobservable acquire must never reach native rendering.
603 s.server.log.Warn().Err(err).Msg("acquire point")
604 }
605 cs.wait = w
606 s.next.sync = cs
607}
608
609// syncReady reports whether a commit's acquire point has its fence.
610func (s *surface) syncReady(cs *commitSync) bool {
611 return cs == nil || cs.wait == nil || s.server.syncWait.fired(cs.wait)
612}
613
614// dropSync forgets a commit that will never apply. Its release point is
615// signalled only once the acquire point has a fence: signalling earlier
616// could move a shared timeline past a point the client has not submitted.
617func (s *surface) dropSync(cs *commitSync) {
618 if cs == nil {
619 return
620 }
621 if cs.fence != nil {
622 cs.fence.Close()
623 cs.fence = nil
624 }
625 ready := cs.wait != nil && s.server.syncWait.fired(cs.wait) && !cs.wait.failed
626 if cs.wait != nil {
627 s.server.syncWait.cancel(cs.wait)
628 }
629 cs.acquire.tl.uses--
630 s.server.dropTimeline(cs.acquire.tl)
631 if ready {
632 s.server.releaseSync(syncHold{release: cs.release})
633 return
634 }
635 cs.release.tl.uses--
636 s.server.dropTimeline(cs.release.tl)
637}
638
639// prepareSync exports only after observed readiness, before any current state
640// changes. Export failure must not fall back to native implicit synchronization.
641func (s *surface) prepareSync(cs *commitSync) bool {
642 if cs == nil || cs.fence != nil {
643 return true
644 }
645 f, err := cs.acquire.tl.dev.exportSyncFile(cs.acquire.tl.handle, cs.acquire.point)
646 if err != nil || f == nil {
647 if f != nil {
648 f.Close()
649 }
650 s.server.log.Warn().Err(err).Msg("acquire fence export rejected")
651 return false
652 }
653 cs.fence = f
654 return true
655}
656
657// applySync makes a commit's points the current buffer's: its acquire
658// fence goes with the content, its release point waits for release.
659func (s *surface) applySync(cs *commitSync) {
660 if cs == nil {
661 return
662 }
663 if cs.wait != nil {
664 s.server.syncWait.cancel(cs.wait)
665 }
666 f := cs.fence
667 cs.fence = nil
668 cs.acquire.tl.uses--
669 s.server.dropTimeline(cs.acquire.tl)
670 s.hold = syncHold{acquire: f, release: cs.release}
671}
672
673// acquireSignalled applies what a fired acquire point unblocked: like a
674// commit, it ticks inline instead of posting a tick to the display loop.
675func (s *Server) acquireSignalled() {
676 s.paceDepth++
677 s.wakePacer()
678 s.paceDepth--
679 if s.paceDepth == 0 && s.paceDirty {
680 s.paceDirty = false
681 s.paceNow()
682 }
683}
684
685// releasePending unregisters and closes the explicit waits still pending at
686// shutdown. Called once the display loop has stopped.
687func (sw *syncWaiter) releasePending() {
688 for w := range sw.pending {
689 sw.finish(w)
690 }
691}