1package server
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "sync/atomic"
8 "testing"
9 "time"
10
11 "golang.org/x/sys/unix"
12)
13
14// runDisplay starts a display and stops it at test cleanup.
15func runDisplay(t *testing.T) *Display {
16 t.Helper()
17 d, err := NewDisplay()
18 if err != nil {
19 t.Fatal(err)
20 }
21 ctx, cancel := context.WithCancel(context.Background())
22 done := make(chan error, 1)
23 go func() { done <- d.Run(ctx) }()
24 t.Cleanup(func() {
25 cancel()
26 if err := <-done; err != nil {
27 t.Error(err)
28 }
29 })
30 return d
31}
32
33func waitFor(t *testing.T, ch <-chan struct{}, what string) {
34 t.Helper()
35 select {
36 case <-ch:
37 case <-time.After(3 * time.Second):
38 t.Fatalf("timeout waiting for %s", what)
39 }
40}
41
42// displayTID reports the thread the display goroutine is locked to.
43func displayTID(t *testing.T, d *Display) int {
44 t.Helper()
45 tid := 0
46 if !d.Do(func() { tid = unix.Gettid() }) {
47 t.Fatal("display stopped")
48 }
49 return tid
50}
51
52func TestPostRunsOnDisplayGoroutine(t *testing.T) {
53 d := runDisplay(t)
54 want := displayTID(t, d)
55 got := make(chan int, 1)
56 go d.Post(func() { got <- unix.Gettid() })
57 select {
58 case tid := <-got:
59 if tid != want {
60 t.Fatalf("post ran on thread %d, display thread is %d", tid, want)
61 }
62 case <-time.After(3 * time.Second):
63 t.Fatal("post did not run")
64 }
65}
66
67func TestPostFIFOWithDo(t *testing.T) {
68 d := runDisplay(t)
69 var order []int
70 for i := range 200 {
71 d.Post(func() { order = append(order, 2*i) })
72 if !d.Do(func() { order = append(order, 2*i+1) }) {
73 t.Fatal("display stopped")
74 }
75 }
76 // Do returned, so the display goroutine is done with order.
77 for i, v := range order {
78 if v != i {
79 t.Fatalf("order[%d] = %d, want %d (%v)", i, v, i, order[:i+1])
80 }
81 }
82 if len(order) != 400 {
83 t.Fatalf("%d callbacks ran, want 400", len(order))
84 }
85}
86
87func TestPostFIFOOrder(t *testing.T) {
88 d := runDisplay(t)
89 var order []int
90 for i := range 1000 {
91 d.Post(func() { order = append(order, i) })
92 }
93 d.Do(func() {})
94 for i, v := range order {
95 if v != i {
96 t.Fatalf("order[%d] = %d", i, v)
97 }
98 }
99 if len(order) != 1000 {
100 t.Fatalf("%d posts ran, want 1000", len(order))
101 }
102}
103
104func TestPostCoalescesWakes(t *testing.T) {
105 d := runDisplay(t)
106 entered, release := make(chan struct{}), make(chan struct{})
107 doDone := make(chan struct{})
108 // Hold the display goroutine so every Post lands while one wake is pending.
109 go func() { d.Do(func() { close(entered); <-release }); close(doDone) }()
110 waitFor(t, entered, "display busy")
111 before := d.wakeWrites.Load()
112 var mu sync.Mutex
113 ran := 0
114 for range 1000 {
115 d.Post(func() { mu.Lock(); ran++; mu.Unlock() })
116 }
117 if w := d.wakeWrites.Load() - before; w > 2 {
118 t.Fatalf("1000 posts wrote the eventfd %d times", w)
119 }
120 close(release)
121 waitFor(t, doDone, "Do")
122 d.Do(func() {})
123 mu.Lock()
124 defer mu.Unlock()
125 if ran != 1000 {
126 t.Fatalf("%d posts ran, want 1000", ran)
127 }
128}
129
130func TestPostAfterStopDropped(t *testing.T) {
131 d, err := NewDisplay()
132 if err != nil {
133 t.Fatal(err)
134 }
135 ctx, cancel := context.WithCancel(context.Background())
136 done := make(chan error, 1)
137 go func() { done <- d.Run(ctx) }()
138 d.Do(func() {})
139 cancel()
140 if err := <-done; err != nil {
141 t.Fatal(err)
142 }
143 <-d.Stopped()
144 d.Post(func() { t.Error("post ran after stop") })
145 time.Sleep(20 * time.Millisecond)
146
147 // A display closed before Run drops posts too.
148 c, err := NewDisplay()
149 if err != nil {
150 t.Fatal(err)
151 }
152 c.Close()
153 c.Post(func() { t.Error("post ran after close") })
154 c.Post(nil)
155}
156
157func TestPostFromPostedFunc(t *testing.T) {
158 d := runDisplay(t)
159 var order []string
160 done := make(chan struct{})
161 d.Post(func() {
162 order = append(order, "outer")
163 d.Post(func() {
164 order = append(order, "inner")
165 close(done)
166 })
167 order = append(order, "outer-end")
168 })
169 waitFor(t, done, "nested post")
170 d.Do(func() {})
171 want := []string{"outer", "outer-end", "inner"}
172 if len(order) != len(want) {
173 t.Fatalf("order %v", order)
174 }
175 for i := range want {
176 if order[i] != want[i] {
177 t.Fatalf("order %v, want %v", order, want)
178 }
179 }
180}
181
182// A callback that keeps posting must not stop libwayland from dispatching, and
183// the capped rounds must be resumed by a wake instead of waiting for the
184// dispatch timeout.
185func TestPostSelfRepostingIsCappedAndResumes(t *testing.T) {
186 d := runDisplay(t)
187 const total = 100 // > maxDrainRounds, so the cap is hit
188 var n int
189 done := make(chan struct{})
190 var step func()
191 step = func() {
192 n++
193 if n == total {
194 close(done)
195 return
196 }
197 d.Post(step)
198 }
199 start := time.Now()
200 d.Post(step)
201 waitFor(t, done, "self-reposting chain")
202 if e := time.Since(start); e > 500*time.Millisecond {
203 t.Fatalf("chain of %d posts took %v; wake not re-armed at the round cap", total, e)
204 }
205 // Do still gets served.
206 if !d.Do(func() {}) {
207 t.Fatal("Do failed")
208 }
209}
210
211func TestPostAllocations(t *testing.T) {
212 d, err := NewDisplay()
213 if err != nil {
214 t.Fatal(err)
215 }
216 defer d.Close()
217 n := 0
218 fn := func() { n++ }
219 // Warm up both batch buffers, then drain on this goroutine (not running).
220 for range 4 {
221 d.Post(fn)
222 d.Post(fn)
223 d.drain()
224 }
225 if allocs := testing.AllocsPerRun(1000, func() {
226 d.Post(fn)
227 d.Post(fn)
228 d.drain()
229 }); allocs > 0 {
230 t.Fatalf("Post+drain allocates %v per run", allocs)
231 }
232 if n == 0 {
233 t.Fatal("posts did not run")
234 }
235}
236
237func TestAddFDEventfdOnDisplayGoroutine(t *testing.T) {
238 d := runDisplay(t)
239 want := displayTID(t, d)
240 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
241 if err != nil {
242 t.Fatal(err)
243 }
244 defer unix.Close(efd)
245 got := make(chan int, 4)
246 var src *Source
247 if !d.Do(func() {
248 src, err = d.AddFD(efd, func() {
249 var b [8]byte
250 _, _ = unix.Read(efd, b[:])
251 got <- unix.Gettid()
252 })
253 }) || err != nil {
254 t.Fatalf("AddFD: %v", err)
255 }
256 one := [8]byte{1}
257 if _, err := unix.Write(efd, one[:]); err != nil {
258 t.Fatal(err)
259 }
260 select {
261 case tid := <-got:
262 if tid != want {
263 t.Fatalf("callback on thread %d, display thread is %d", tid, want)
264 }
265 case <-time.After(3 * time.Second):
266 t.Fatal("callback did not fire")
267 }
268
269 // Remove stops callbacks; it is idempotent.
270 d.Do(func() { src.Remove(); src.Remove() })
271 if _, err := unix.Write(efd, one[:]); err != nil {
272 t.Fatal(err)
273 }
274 d.Do(func() {})
275 time.Sleep(30 * time.Millisecond)
276 select {
277 case <-got:
278 t.Fatal("callback fired after Remove")
279 default:
280 }
281 d.Do(func() {
282 if len(live.fds) != 0 {
283 t.Errorf("%d fd callbacks leaked", len(live.fds))
284 }
285 })
286}
287
288func TestAddFDTimerfdFiresOnce(t *testing.T) {
289 d := runDisplay(t)
290 tfd, err := unix.TimerfdCreate(unix.CLOCK_MONOTONIC, unix.TFD_NONBLOCK|unix.TFD_CLOEXEC)
291 if err != nil {
292 t.Skip(err)
293 }
294 defer unix.Close(tfd)
295 fired := make(chan struct{}, 8)
296 if !d.Do(func() {
297 _, err = d.AddFD(tfd, func() {
298 var b [8]byte
299 _, _ = unix.Read(tfd, b[:])
300 fired <- struct{}{}
301 })
302 }) || err != nil {
303 t.Fatalf("AddFD: %v", err)
304 }
305 start := time.Now()
306 if err := unix.TimerfdSettime(tfd, 0, &unix.ItimerSpec{Value: unix.NsecToTimespec(int64(time.Millisecond))}, nil); err != nil {
307 t.Fatal(err)
308 }
309 waitFor(t, fired, "timerfd")
310 if e := time.Since(start); e > 500*time.Millisecond {
311 t.Fatalf("1 ms timer fired after %v", e)
312 }
313 time.Sleep(50 * time.Millisecond)
314 if len(fired) != 0 {
315 t.Fatalf("one-shot timer fired %d more times", len(fired))
316 }
317}
318
319func TestAddFDBeforeRunAndRemoveInCallback(t *testing.T) {
320 d, err := NewDisplay()
321 if err != nil {
322 t.Fatal(err)
323 }
324 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
325 if err != nil {
326 t.Fatal(err)
327 }
328 defer unix.Close(efd)
329 fired := make(chan struct{}, 4)
330 var src *Source
331 src, err = d.AddFD(efd, func() {
332 var b [8]byte
333 _, _ = unix.Read(efd, b[:])
334 src.Remove() // from its own callback
335 fired <- struct{}{}
336 })
337 if err != nil {
338 t.Fatal(err)
339 }
340 ctx, cancel := context.WithCancel(context.Background())
341 done := make(chan error, 1)
342 go func() { done <- d.Run(ctx) }()
343 defer func() { cancel(); <-done }()
344 one := [8]byte{1}
345 for range 2 {
346 if _, err := unix.Write(efd, one[:]); err != nil {
347 t.Fatal(err)
348 }
349 time.Sleep(20 * time.Millisecond)
350 }
351 waitFor(t, fired, "callback")
352 if len(fired) != 0 {
353 t.Fatal("callback fired again after removing itself")
354 }
355}
356
357func TestAddFDErrors(t *testing.T) {
358 d, err := NewDisplay()
359 if err != nil {
360 t.Fatal(err)
361 }
362 defer d.Close()
363 if _, err := d.AddFD(-1, func() {}); err == nil {
364 t.Error("negative fd accepted")
365 }
366 if _, err := d.AddFD(0, nil); err == nil {
367 t.Error("nil callback accepted")
368 }
369 // A source left registered is freed with the display; Remove is then a no-op.
370 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
371 if err != nil {
372 t.Fatal(err)
373 }
374 defer unix.Close(efd)
375 src, err := d.AddFD(efd, func() {})
376 if err != nil {
377 t.Fatal(err)
378 }
379 d.Close()
380 src.Remove()
381 (*Source)(nil).Remove()
382 if _, err := d.AddFD(efd, func() {}); err == nil {
383 t.Error("AddFD accepted on a closed display")
384 }
385 if len(live.fds) != 0 {
386 t.Errorf("%d fd callbacks leaked", len(live.fds))
387 }
388}
389
390// Posts from many goroutines must wake the loop without any Do to rescue a
391// lost wake. The dispatch timeout is raised far past waitFor's limit, so a
392// lost wake hangs the round instead of only slowing it.
393func TestPostMultiProducerNoLostWake(t *testing.T) {
394 d, err := NewDisplay()
395 if err != nil {
396 t.Fatal(err)
397 }
398 d.dispatchMS = 60_000
399 ctx, cancel := context.WithCancel(context.Background())
400 runDone := make(chan error, 1)
401 go func() { runDone <- d.Run(ctx) }()
402 t.Cleanup(func() {
403 cancel()
404 // Wake the loop so it sees the cancellation before the long timeout.
405 d.Post(func() {})
406 if err := <-runDone; err != nil {
407 t.Error(err)
408 }
409 })
410 const producers, perProducer, rounds = 8, 250, 10
411 for range rounds {
412 // Let the loop go idle in epoll_wait.
413 time.Sleep(5 * time.Millisecond)
414 var left atomic.Int32
415 left.Store(producers * perProducer)
416 done := make(chan struct{})
417 var wg sync.WaitGroup
418 for range producers {
419 wg.Add(1)
420 go func() {
421 defer wg.Done()
422 for range perProducer {
423 d.Post(func() {
424 if left.Add(-1) == 0 {
425 close(done)
426 }
427 })
428 }
429 }()
430 }
431 waitFor(t, done, "all posts (a lost wake hangs here)")
432 wg.Wait()
433 }
434}
435
436// Do and Post racing with shutdown must neither panic nor run anything after
437// Stopped, and Do's wake must not write to the closed eventfd's number once it
438// is reused.
439func TestDoPostRaceWithCancel(t *testing.T) {
440 for range 20 {
441 d, err := NewDisplay()
442 if err != nil {
443 t.Fatal(err)
444 }
445 ctx, cancel := context.WithCancel(context.Background())
446 done := make(chan error, 1)
447 go func() { done <- d.Run(ctx) }()
448
449 var stopped atomic.Bool
450 var wg sync.WaitGroup
451 for range 4 {
452 wg.Add(2)
453 go func() {
454 defer wg.Done()
455 for range 200 {
456 d.Do(func() {
457 if stopped.Load() {
458 t.Error("Do function ran after Stopped")
459 }
460 })
461 }
462 }()
463 go func() {
464 defer wg.Done()
465 for range 200 {
466 d.Post(func() {
467 if stopped.Load() {
468 t.Error("post ran after Stopped")
469 }
470 })
471 }
472 }()
473 }
474 time.Sleep(time.Millisecond)
475 cancel()
476 if err := <-done; err != nil {
477 t.Fatal(err)
478 }
479 <-d.Stopped()
480 stopped.Store(true)
481
482 // The closed wake fd's number is free for reuse now. Late Do and Post
483 // calls must not write to whoever gets it.
484 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
485 if err != nil {
486 t.Fatal(err)
487 }
488 // Force the reuse: make the old wake fd number refer to efd, unless
489 // efd already got that number. Never clobber an fd in use.
490 if efd != d.wake {
491 if _, err := unix.FcntlInt(uintptr(d.wake), unix.F_GETFD, 0); !errors.Is(err, unix.EBADF) {
492 t.Fatalf("old wake fd %d is still in use: %v", d.wake, err)
493 }
494 if err := unix.Dup3(efd, d.wake, unix.O_CLOEXEC); err != nil {
495 t.Fatal(err)
496 }
497 }
498 for range 50 {
499 d.Do(func() { t.Error("Do function ran after Stopped") })
500 d.Post(func() { t.Error("post ran after Stopped") })
501 }
502 wg.Wait()
503 var b [8]byte
504 if _, err := unix.Read(efd, b[:]); !errors.Is(err, unix.EAGAIN) {
505 t.Fatalf("eventfd %d (display wake fd was %d) was written after stop: %v", efd, d.wake, err)
506 }
507 unix.Close(efd)
508 if efd != d.wake {
509 unix.Close(d.wake)
510 }
511 }
512}
513
514// Sources still registered when Run exits are released with the display.
515func TestRunExitReleasesSources(t *testing.T) {
516 d, err := NewDisplay()
517 if err != nil {
518 t.Fatal(err)
519 }
520 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
521 if err != nil {
522 t.Fatal(err)
523 }
524 defer unix.Close(efd)
525 ctx, cancel := context.WithCancel(context.Background())
526 done := make(chan error, 1)
527 go func() { done <- d.Run(ctx) }()
528 var addErr error
529 if !d.Do(func() { _, addErr = d.AddFD(efd, func() {}) }) || addErr != nil {
530 t.Fatalf("AddFD: %v", addErr)
531 }
532 cancel()
533 if err := <-done; err != nil {
534 t.Fatal(err)
535 }
536 <-d.Stopped()
537 if len(live.fds) != 0 {
538 t.Errorf("%d fd callbacks leaked after Run exited", len(live.fds))
539 }
540 if len(d.sources) != 0 {
541 t.Errorf("%d sources still tracked after Run exited", len(d.sources))
542 }
543}
544
545func TestPostFromAddFDCallback(t *testing.T) {
546 d := runDisplay(t)
547 want := displayTID(t, d)
548 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
549 if err != nil {
550 t.Fatal(err)
551 }
552 defer unix.Close(efd)
553 got := make(chan int, 1)
554 var src *Source
555 if !d.Do(func() {
556 src, err = d.AddFD(efd, func() {
557 var b [8]byte
558 _, _ = unix.Read(efd, b[:])
559 d.Post(func() { got <- unix.Gettid() })
560 })
561 }) || err != nil {
562 t.Fatalf("AddFD: %v", err)
563 }
564 one := [8]byte{1}
565 if _, err := unix.Write(efd, one[:]); err != nil {
566 t.Fatal(err)
567 }
568 select {
569 case tid := <-got:
570 if tid != want {
571 t.Fatalf("post ran on thread %d, display thread is %d", tid, want)
572 }
573 case <-time.After(3 * time.Second):
574 t.Fatal("post from AddFD callback did not run")
575 }
576 d.Do(func() { src.Remove() })
577}