1package wayland
2
3import (
4 "sync"
5 "time"
6
7 "github.com/bnema/purego-libwayland/server"
8)
9
10// postRing carries values from producer goroutines to the display goroutine
11// without allocating. It is a transport queue, not window state: the mutex
12// guards only the slices and the pending flag. Values are applied in order of
13// arrival and never dropped. A ring built with a merge function may fold a
14// pushed value into the last queued one (and only that one, under the mutex),
15// so nothing is ever reordered.
16//
17// It binds its drain once and posts it at most once per batch: Display.Post
18// is unbounded (a Post per value would let a fast producer grow the queue),
19// so one pending drain serves every value pushed before it runs.
20type postRing[T any] struct {
21 display *server.Display
22 apply func(T)
23 drainFn func()
24 // merge, when set, tries to fold v into the last queued value, which has
25 // not been drained yet. It runs under mu and must not block or allocate.
26 merge func(last *T, v T) bool
27 // space is signalled (never blocking) after each drain and after a push
28 // that left room, to wake blocked pushers one after the other.
29 space chan struct{}
30
31 mu sync.Mutex
32 ring, spare []T
33 limit int
34 pending bool // a drain is posted and has not taken the ring yet
35 stats ringStats
36}
37
38// ringStats is what a ring saw since the last takeStats. It lets a live log
39// tell a slow display goroutine (long queue, long push block) from a slow
40// producer.
41type ringStats struct {
42 maxLen int // largest queued length
43 merged int // pushes folded into a queued value
44 maxPushBlock time.Duration // longest wait of one Push on a full ring
45}
46
47// takeStats returns the counters and resets them.
48func (r *postRing[T]) takeStats() ringStats {
49 r.mu.Lock()
50 st := r.stats
51 r.stats = ringStats{}
52 r.mu.Unlock()
53 return st
54}
55
56// newPostRing returns a ring of limit values whose values apply runs on the
57// display goroutine, in order. limit must be at least 1.
58func newPostRing[T any](d *server.Display, limit int, apply func(T)) *postRing[T] {
59 return newMergingPostRing(d, limit, apply, nil)
60}
61
62// newMergingPostRing is newPostRing with a merge function (see postRing). A
63// merged push needs no room: it succeeds on a full ring.
64func newMergingPostRing[T any](d *server.Display, limit int, apply func(T), merge func(last *T, v T) bool) *postRing[T] {
65 r := &postRing[T]{
66 display: d,
67 apply: apply,
68 space: make(chan struct{}, 1),
69 ring: make([]T, 0, limit),
70 spare: make([]T, 0, limit),
71 limit: limit,
72 merge: merge,
73 }
74 r.drainFn = r.drain
75 return r
76}
77
78// Push queues v for the display goroutine and wakes it. A full ring blocks
79// the caller, which must not be the display goroutine, until a drain made
80// room; it reports false, with v dropped, once the display has stopped.
81func (r *postRing[T]) Push(v T) bool {
82 var blocked time.Duration // total wait of this push on a full ring
83 for {
84 r.mu.Lock()
85 // A non-empty ring always has its drain posted, so the last value is
86 // still queued: merging into it needs neither room nor a new post.
87 if n := len(r.ring); r.merge != nil && n > 0 && r.merge(&r.ring[n-1], v) {
88 r.stats.merged++
89 r.stats.maxPushBlock = max(r.stats.maxPushBlock, blocked)
90 r.mu.Unlock()
91 return true
92 }
93 if len(r.ring) < r.limit {
94 r.ring = append(r.ring, v)
95 r.stats.maxLen = max(r.stats.maxLen, len(r.ring))
96 r.stats.maxPushBlock = max(r.stats.maxPushBlock, blocked)
97 post := !r.pending
98 r.pending = true
99 room := len(r.ring) < r.limit
100 r.mu.Unlock()
101 if post {
102 r.display.Post(r.drainFn)
103 }
104 if room {
105 r.signal() // chain the wake to another blocked pusher
106 }
107 return true
108 }
109 r.mu.Unlock()
110 // Only the blocking path reads the clock: the fast path stays free.
111 start := time.Now()
112 select {
113 case <-r.space:
114 blocked += time.Since(start)
115 case <-r.display.Stopped():
116 return false
117 }
118 }
119}
120
121func (r *postRing[T]) signal() {
122 select {
123 case r.space <- struct{}{}:
124 default:
125 }
126}
127
128// drain runs on the display goroutine and applies the batch in order.
129func (r *postRing[T]) drain() {
130 r.mu.Lock()
131 batch := r.ring
132 r.ring, r.spare = r.spare[:0], nil
133 r.pending = false
134 r.mu.Unlock()
135 r.signal()
136 for _, v := range batch {
137 r.apply(v)
138 }
139 // Do not retain what the values reference in the spare batch.
140 clear(batch)
141 r.mu.Lock()
142 r.spare = batch[:0]
143 r.mu.Unlock()
144}