1package wayland
2
3import (
4 "context"
5 "runtime"
6 "strings"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/bnema/neferwl/internal/adapters/logging"
12 "github.com/bnema/neferwl/internal/ports"
13)
14
15// Pushing a capture reply and applying it on the display goroutine allocates
16// nothing in steady state. The count includes the display goroutine:
17// AllocsPerRun reads the global malloc counter.
18func TestCapturedForwardAllocations(t *testing.T) {
19 if raceEnabled {
20 t.Skip("the race detector makes the purego wake callback allocate")
21 }
22 s, _, _, _, _, _ := presentServer(t)
23 done := ports.CaptureDone{ID: 77, Output: "HEADLESS-1", Time: time.Unix(1, 2)}
24 // Posts run in order: once the marker ran, the drain before it did. The
25 // marker is bound once and the channel reused, so waiting allocates nothing.
26 ran := make(chan struct{}, 1)
27 marker := func() { ran <- struct{}{} }
28 for range 4 {
29 s.capturedRing.Push(done)
30 s.display.Post(marker)
31 <-ran
32 }
33 allocs := testing.AllocsPerRun(200, func() {
34 s.capturedRing.Push(done)
35 s.display.Post(marker)
36 <-ran
37 })
38 if allocs != 0 {
39 t.Errorf("push + drain: %.1f allocs, want 0", allocs)
40 }
41}
42
43// A full ring blocks the push until the display goroutine drains it, and no
44// value is lost or reordered.
45func TestPostRingFullBlocksWithoutLoss(t *testing.T) {
46 s, _, _, _, _, _ := presentServer(t)
47 const limit, total = 4, 3*4 + 1
48 var got []int // display goroutine only
49 r := newPostRing(s.display, limit, func(v int) { got = append(got, v) })
50
51 // Stall the display goroutine so nothing drains.
52 release := make(chan struct{})
53 stalled := make(chan struct{})
54 s.display.Post(func() { close(stalled); <-release })
55 <-stalled
56
57 pushed := make(chan int, total)
58 var wg sync.WaitGroup
59 wg.Add(1)
60 go func() {
61 defer wg.Done()
62 for i := range total {
63 if !r.Push(i) {
64 t.Errorf("push %d refused", i)
65 return
66 }
67 pushed <- i
68 }
69 }()
70 for range limit {
71 select {
72 case <-pushed:
73 case <-time.After(2 * time.Second):
74 t.Fatal("pushes below the limit blocked")
75 }
76 }
77 select {
78 case i := <-pushed:
79 t.Fatalf("push %d returned on a full ring", i)
80 case <-time.After(50 * time.Millisecond):
81 }
82 close(release)
83 wg.Wait()
84 // The marker runs after every drain posted before it.
85 var final []int
86 ran := make(chan struct{})
87 for {
88 s.display.Post(func() {
89 if len(got) == total {
90 final = append(final, got...)
91 }
92 ran <- struct{}{}
93 })
94 <-ran
95 if final != nil {
96 break
97 }
98 time.Sleep(time.Millisecond)
99 }
100 for i, v := range final {
101 if v != i {
102 t.Fatalf("applied %v, want 0..%d in order", final, total-1)
103 }
104 }
105}
106
107// A push blocked on a full ring resumes once the display drains it.
108func TestPostRingBlockedPushResumes(t *testing.T) {
109 s, _, _, _, _, _ := presentServer(t)
110 r := newPostRing(s.display, 1, func(int) {})
111 release := make(chan struct{})
112 stalled := make(chan struct{})
113 s.display.Post(func() { close(stalled); <-release })
114 <-stalled
115 if !r.Push(1) {
116 t.Fatal("first push refused")
117 }
118 result := make(chan bool, 1)
119 go func() { result <- r.Push(2) }()
120 select {
121 case ok := <-result:
122 t.Fatalf("push on a full ring returned %v", ok)
123 case <-time.After(50 * time.Millisecond):
124 }
125 close(release)
126 select {
127 case ok := <-result:
128 if !ok {
129 t.Fatal("push refused while the display runs")
130 }
131 case <-time.After(2 * time.Second):
132 t.Fatal("blocked push never resumed")
133 }
134}
135
136// A push blocked on a full ring returns false once the display has stopped,
137// instead of waiting for a drain that never comes.
138func TestPostRingPushStopsWithDisplay(t *testing.T) {
139 s, err := New(Options{RuntimeDir: t.TempDir()}, Channels{}, logging.For(context.Background(), "wayland"))
140 if err != nil {
141 t.Fatal(err)
142 }
143 ctx, cancel := context.WithCancel(context.Background())
144 done := make(chan error, 1)
145 go func() { done <- s.Run(ctx) }()
146 cancel()
147 <-done
148 r := newPostRing(s.display, 1, func(int) {})
149 if !r.Push(1) { // room: queued, its drain is dropped by the stopped display
150 t.Fatal("first push refused")
151 }
152 result := make(chan bool, 1)
153 go func() { result <- r.Push(2) }()
154 select {
155 case ok := <-result:
156 if ok {
157 t.Fatal("push on a stopped display reported success")
158 }
159 case <-time.After(2 * time.Second):
160 t.Fatal("push blocked after the display stopped")
161 }
162}
163
164// A push that waits on a full ring reports how long, and a drain-side push
165// that finds room does not.
166func TestPostRingStatsPushBlock(t *testing.T) {
167 s, _, _, _, _, _ := presentServer(t)
168 r := newPostRing(s.display, 1, func(int) {})
169 // A failed assertion still frees the display before server shutdown.
170 release := releaseOnce(t)
171 stalled := make(chan struct{})
172 s.display.Post(func() { close(stalled); <-release.ch })
173 <-stalled
174 r.Push(1)
175 if st := r.takeStats(); st.maxLen != 1 || st.maxPushBlock != 0 {
176 t.Fatalf("stats = %+v, want maxLen 1 and no block", st)
177 }
178 done := make(chan struct{})
179 go func() { defer close(done); r.Push(2) }() // full: waits for the drain
180 waitPushWaiting(t)
181 seen := time.Now()
182 time.Sleep(2 * time.Millisecond) // the block lasts at least this
183 held := time.Since(seen)
184 release.do()
185 <-done
186 if st := r.takeStats(); st.maxPushBlock < held {
187 t.Fatalf("stats = %+v, want a push block of at least %v", st, held)
188 }
189}
190
191// waitPushWaiting returns once a goroutine waits inside postRing.Push, read
192// from the goroutine dump, so the test needs no sleep to know the push blocks.
193func waitPushWaiting(t *testing.T) {
194 t.Helper()
195 buf := make([]byte, 1<<20)
196 deadline := time.Now().Add(5 * time.Second)
197 for {
198 if time.Now().After(deadline) {
199 t.Fatalf("no goroutine waits in postRing.Push:\n%s", buf[:runtime.Stack(buf, true)])
200 }
201 dump := string(buf[:runtime.Stack(buf, true)])
202 for _, g := range strings.Split(dump, "\n\n") {
203 if strings.Contains(g, "[select") && strings.Contains(g, "postRing") && strings.Contains(g, ").Push(") {
204 return
205 }
206 }
207 runtime.Gosched()
208 }
209}
210
211func TestPostRingStatsSampleStalledOccupancy(t *testing.T) {
212 s, _, _, _, _, _ := presentServer(t)
213 release := make(chan struct{})
214 stalled := make(chan struct{})
215 s.display.Post(func() { close(stalled); <-release })
216 <-stalled
217 // Release even if an assertion fails, so server cleanup cannot deadlock.
218 released := false
219 defer func() {
220 if !released {
221 close(release)
222 }
223 }()
224 r := newPostRing(s.display, 2, func(int) {})
225 for i := range 2 {
226 if !r.Push(i) {
227 t.Fatal("push refused")
228 }
229 }
230 for range 2 {
231 if got := r.takeStats().maxLen; got != 2 {
232 t.Fatalf("occupancy %d, want 2", got)
233 }
234 }
235 close(release)
236 released = true
237 if !s.display.Do(func() {}) {
238 t.Fatal("display stopped")
239 }
240 if got := r.takeStats().maxLen; got != 0 {
241 t.Fatalf("recovered occupancy %d", got)
242 }
243}