postring.go RawHistoryFolder

1package wayland
2
3import (
4	"sync"
5
6	"github.com/bnema/purego-libwayland/server"
7)
8
9// postRing carries values from producer goroutines to the display goroutine
10// without allocating. It is a transport queue, not window state: the mutex
11// guards only the slices and the pending flag. Values are never merged or
12// dropped, in order of arrival.
13//
14// It binds its drain once and posts it at most once per batch: Display.Post
15// is unbounded (a Post per value would let a fast producer grow the queue),
16// so one pending drain serves every value pushed before it runs.
17type postRing[T any] struct {
18	display *server.Display
19	apply   func(T)
20	drainFn func()
21	// space is signalled (never blocking) after each drain and after a push
22	// that left room, to wake blocked pushers one after the other.
23	space chan struct{}
24
25	mu          sync.Mutex
26	ring, spare []T
27	limit       int
28	pending     bool // a drain is posted and has not taken the ring yet
29}
30
31// newPostRing returns a ring of limit values whose values apply runs on the
32// display goroutine, in order. limit must be at least 1.
33func newPostRing[T any](d *server.Display, limit int, apply func(T)) *postRing[T] {
34	r := &postRing[T]{
35		display: d,
36		apply:   apply,
37		space:   make(chan struct{}, 1),
38		ring:    make([]T, 0, limit),
39		spare:   make([]T, 0, limit),
40		limit:   limit,
41	}
42	r.drainFn = r.drain
43	return r
44}
45
46// Push queues v for the display goroutine and wakes it. A full ring blocks
47// the caller, which must not be the display goroutine, until a drain made
48// room; it reports false, with v dropped, once the display has stopped.
49func (r *postRing[T]) Push(v T) bool {
50	for {
51		r.mu.Lock()
52		if len(r.ring) < r.limit {
53			r.ring = append(r.ring, v)
54			post := !r.pending
55			r.pending = true
56			room := len(r.ring) < r.limit
57			r.mu.Unlock()
58			if post {
59				r.display.Post(r.drainFn)
60			}
61			if room {
62				r.signal() // chain the wake to another blocked pusher
63			}
64			return true
65		}
66		r.mu.Unlock()
67		select {
68		case <-r.space:
69		case <-r.display.Stopped():
70			return false
71		}
72	}
73}
74
75func (r *postRing[T]) signal() {
76	select {
77	case r.space <- struct{}{}:
78	default:
79	}
80}
81
82// drain runs on the display goroutine and applies the batch in order.
83func (r *postRing[T]) drain() {
84	r.mu.Lock()
85	batch := r.ring
86	r.ring, r.spare = r.spare[:0], nil
87	r.pending = false
88	r.mu.Unlock()
89	r.signal()
90	for _, v := range batch {
91		r.apply(v)
92	}
93	// Do not retain what the values reference in the spare batch.
94	clear(batch)
95	r.mu.Lock()
96	r.spare = batch[:0]
97	r.mu.Unlock()
98}