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 ctx, cancel := context.WithCancel(context.Background())
399 runDone := make(chan error, 1)
400 go func() { runDone <- d.Run(ctx) }()
401 t.Cleanup(func() {
402 cancel()
403 // Wake the loop so it sees the cancellation before the long timeout.
404 d.Post(func() {})
405 if err := <-runDone; err != nil {
406 t.Error(err)
407 }
408 })
409 const producers, perProducer, rounds = 8, 250, 10
410 for range rounds {
411 // Let the loop go idle in epoll_wait.
412 time.Sleep(5 * time.Millisecond)
413 var left atomic.Int32
414 left.Store(producers * perProducer)
415 done := make(chan struct{})
416 var wg sync.WaitGroup
417 for range producers {
418 wg.Add(1)
419 go func() {
420 defer wg.Done()
421 for range perProducer {
422 d.Post(func() {
423 if left.Add(-1) == 0 {
424 close(done)
425 }
426 })
427 }
428 }()
429 }
430 waitFor(t, done, "all posts (a lost wake hangs here)")
431 wg.Wait()
432 }
433}
434
435// Do and Post racing with shutdown must neither panic nor run anything after
436// Stopped, and Do's wake must not write to the closed eventfd's number once it
437// is reused.
438func TestDoPostRaceWithCancel(t *testing.T) {
439 for range 20 {
440 d, err := NewDisplay()
441 if err != nil {
442 t.Fatal(err)
443 }
444 ctx, cancel := context.WithCancel(context.Background())
445 done := make(chan error, 1)
446 go func() { done <- d.Run(ctx) }()
447
448 var stopped atomic.Bool
449 var wg sync.WaitGroup
450 for range 4 {
451 wg.Add(2)
452 go func() {
453 defer wg.Done()
454 for range 200 {
455 d.Do(func() {
456 if stopped.Load() {
457 t.Error("Do function ran after Stopped")
458 }
459 })
460 }
461 }()
462 go func() {
463 defer wg.Done()
464 for range 200 {
465 d.Post(func() {
466 if stopped.Load() {
467 t.Error("post ran after Stopped")
468 }
469 })
470 }
471 }()
472 }
473 time.Sleep(time.Millisecond)
474 cancel()
475 if err := <-done; err != nil {
476 t.Fatal(err)
477 }
478 <-d.Stopped()
479 stopped.Store(true)
480
481 // The closed wake fd's number is free for reuse now. Late Do and Post
482 // calls must not write to whoever gets it.
483 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
484 if err != nil {
485 t.Fatal(err)
486 }
487 // Force the reuse: make the old wake fd number refer to efd, unless
488 // efd already got that number. Never clobber an fd in use.
489 if efd != d.wake {
490 if _, err := unix.FcntlInt(uintptr(d.wake), unix.F_GETFD, 0); !errors.Is(err, unix.EBADF) {
491 t.Fatalf("old wake fd %d is still in use: %v", d.wake, err)
492 }
493 if err := unix.Dup3(efd, d.wake, unix.O_CLOEXEC); err != nil {
494 t.Fatal(err)
495 }
496 }
497 for range 50 {
498 d.Do(func() { t.Error("Do function ran after Stopped") })
499 d.Post(func() { t.Error("post ran after Stopped") })
500 }
501 wg.Wait()
502 var b [8]byte
503 if _, err := unix.Read(efd, b[:]); !errors.Is(err, unix.EAGAIN) {
504 t.Fatalf("eventfd %d (display wake fd was %d) was written after stop: %v", efd, d.wake, err)
505 }
506 unix.Close(efd)
507 if efd != d.wake {
508 unix.Close(d.wake)
509 }
510 }
511}
512
513// Sources still registered when Run exits are released with the display.
514func TestRunExitReleasesSources(t *testing.T) {
515 d, err := NewDisplay()
516 if err != nil {
517 t.Fatal(err)
518 }
519 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
520 if err != nil {
521 t.Fatal(err)
522 }
523 defer unix.Close(efd)
524 ctx, cancel := context.WithCancel(context.Background())
525 done := make(chan error, 1)
526 go func() { done <- d.Run(ctx) }()
527 var addErr error
528 if !d.Do(func() { _, addErr = d.AddFD(efd, func() {}) }) || addErr != nil {
529 t.Fatalf("AddFD: %v", addErr)
530 }
531 cancel()
532 if err := <-done; err != nil {
533 t.Fatal(err)
534 }
535 <-d.Stopped()
536 if len(live.fds) != 0 {
537 t.Errorf("%d fd callbacks leaked after Run exited", len(live.fds))
538 }
539 if len(d.sources) != 0 {
540 t.Errorf("%d sources still tracked after Run exited", len(d.sources))
541 }
542}
543
544func TestPostFromAddFDCallback(t *testing.T) {
545 d := runDisplay(t)
546 want := displayTID(t, d)
547 efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
548 if err != nil {
549 t.Fatal(err)
550 }
551 defer unix.Close(efd)
552 got := make(chan int, 1)
553 var src *Source
554 if !d.Do(func() {
555 src, err = d.AddFD(efd, func() {
556 var b [8]byte
557 _, _ = unix.Read(efd, b[:])
558 d.Post(func() { got <- unix.Gettid() })
559 })
560 }) || err != nil {
561 t.Fatalf("AddFD: %v", err)
562 }
563 one := [8]byte{1}
564 if _, err := unix.Write(efd, one[:]); err != nil {
565 t.Fatal(err)
566 }
567 select {
568 case tid := <-got:
569 if tid != want {
570 t.Fatalf("post ran on thread %d, display thread is %d", tid, want)
571 }
572 case <-time.After(3 * time.Second):
573 t.Fatal("post from AddFD callback did not run")
574 }
575 d.Do(func() { src.Remove() })
576}