test(server): make the lost-wake test hang instead of timing it
- commit
- 6ec2cf940624d449fef70149961b172e0bf07c57
- author
- brice · · 2026-10-13 18:08 UTC
- parent
- 046f9f0
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +40 -16
M server/display.go +6 -2
| @@ -88,6 +88,10 @@ type Display struct { | |||
| 88 | 88 | ||
| 89 | 89 | // sources are the AddFD sources not yet removed (display goroutine only). | |
| 90 | 90 | sources map[*Source]struct{} | |
| 91 | + | ||
| 92 | + | // dispatchMS bounds one event loop wait; it only bounds how late ctx | |
| 93 | + | // cancellation is seen. Tests raise it so a lost wake hangs. | |
| 94 | + | dispatchMS int32 | |
| 91 | 95 | } | |
| 92 | 96 | ||
| 93 | 97 | // maxDrainRounds bounds how many times one loop iteration re-drains calls and | |
| @@ -108,7 +112,7 @@ func NewDisplay() (*Display, error) { | |||
| 108 | 112 | wlDisplayDestroy(c) | |
| 109 | 113 | return nil, fmt.Errorf("purego-libwayland: eventfd: %w", err) | |
| 110 | 114 | } | |
| 111 | - | d := &Display{c: c, loop: wlDisplayGetEventLoop(c), calls: make(chan func(), 64), wake: wake, stopped: make(chan struct{}), sources: map[*Source]struct{}{}} | |
| 115 | + | d := &Display{c: c, loop: wlDisplayGetEventLoop(c), calls: make(chan func(), 64), wake: wake, stopped: make(chan struct{}), sources: map[*Source]struct{}{}, dispatchMS: 100} | |
| 112 | 116 | // The callback only drains the counter; Run then executes queued calls. | |
| 113 | 117 | if d.wakeSrc = wlEventLoopAddFD(d.loop, int32(wake), 1 /* WL_EVENT_READABLE */, cbWake, 0); d.wakeSrc == 0 { | |
| 114 | 118 | unix.Close(wake) | |
| @@ -226,7 +230,7 @@ func (d *Display) Run(ctx context.Context) error { | |||
| 226 | 230 | d.drain() | |
| 227 | 231 | wlDisplayFlushClients(d.c) | |
| 228 | 232 | // The timeout only bounds how late ctx cancellation is seen. | |
| 229 | - | if wlEventLoopDispatch(d.loop, 100) < 0 { | |
| 233 | + | if wlEventLoopDispatch(d.loop, d.dispatchMS) < 0 { | |
| 230 | 234 | errno := lastErrno() | |
| 231 | 235 | consecutiveFailures++ | |
| 232 | 236 | // epoll_wait is interrupted by Go's preemption signals (EINTR); a persistent failure (EBADF, EINVAL) repeats. | |
| @@ -88,6 +88,10 @@ type Display struct { | |||||
| 88 | 88 | ||||
| 89 | // sources are the AddFD sources not yet removed (display goroutine only). | 89 | // sources are the AddFD sources not yet removed (display goroutine only). | ||
| 90 | sources map[*Source]struct{} | 90 | sources map[*Source]struct{} | ||
| 91 | + | ||||
| 92 | + | // dispatchMS bounds one event loop wait; it only bounds how late ctx | |||
| 93 | + | // cancellation is seen. Tests raise it so a lost wake hangs. | |||
| 94 | + | dispatchMS int32 | |||
| 91 | } | 95 | } | ||
| 92 | 96 | ||||
| 93 | // maxDrainRounds bounds how many times one loop iteration re-drains calls and | 97 | // maxDrainRounds bounds how many times one loop iteration re-drains calls and | ||
| @@ -108,7 +112,7 @@ func NewDisplay() (*Display, error) { | |||||
| 108 | wlDisplayDestroy(c) | 112 | wlDisplayDestroy(c) | ||
| 109 | return nil, fmt.Errorf("purego-libwayland: eventfd: %w", err) | 113 | return nil, fmt.Errorf("purego-libwayland: eventfd: %w", err) | ||
| 110 | } | 114 | } | ||
| 111 | - | d := &Display{c: c, loop: wlDisplayGetEventLoop(c), calls: make(chan func(), 64), wake: wake, stopped: make(chan struct{}), sources: map[*Source]struct{}{}} | 115 | + | d := &Display{c: c, loop: wlDisplayGetEventLoop(c), calls: make(chan func(), 64), wake: wake, stopped: make(chan struct{}), sources: map[*Source]struct{}{}, dispatchMS: 100} |
| 112 | // The callback only drains the counter; Run then executes queued calls. | 116 | // The callback only drains the counter; Run then executes queued calls. | ||
| 113 | if d.wakeSrc = wlEventLoopAddFD(d.loop, int32(wake), 1 /* WL_EVENT_READABLE */, cbWake, 0); d.wakeSrc == 0 { | 117 | if d.wakeSrc = wlEventLoopAddFD(d.loop, int32(wake), 1 /* WL_EVENT_READABLE */, cbWake, 0); d.wakeSrc == 0 { | ||
| 114 | unix.Close(wake) | 118 | unix.Close(wake) | ||
| @@ -226,7 +230,7 @@ func (d *Display) Run(ctx context.Context) error { | |||||
| 226 | d.drain() | 230 | d.drain() | ||
| 227 | wlDisplayFlushClients(d.c) | 231 | wlDisplayFlushClients(d.c) | ||
| 228 | // The timeout only bounds how late ctx cancellation is seen. | 232 | // The timeout only bounds how late ctx cancellation is seen. | ||
| 229 | - | if wlEventLoopDispatch(d.loop, 100) < 0 { | 233 | + | if wlEventLoopDispatch(d.loop, d.dispatchMS) < 0 { |
| 230 | errno := lastErrno() | 234 | errno := lastErrno() | ||
| 231 | consecutiveFailures++ | 235 | consecutiveFailures++ | ||
| 232 | // epoll_wait is interrupted by Go's preemption signals (EINTR); a persistent failure (EBADF, EINVAL) repeats. | 236 | // epoll_wait is interrupted by Go's preemption signals (EINTR); a persistent failure (EBADF, EINVAL) repeats. | ||
M server/post_test.go +34 -14
| @@ -2,6 +2,7 @@ package server | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "errors" | |
| 5 | 6 | "sync" | |
| 6 | 7 | "sync/atomic" | |
| 7 | 8 | "testing" | |
| @@ -387,18 +388,32 @@ func TestAddFDErrors(t *testing.T) { | |||
| 387 | 388 | } | |
| 388 | 389 | ||
| 389 | 390 | // Posts from many goroutines must wake the loop without any Do to rescue a | |
| 390 | - | // lost wake: a lost wake would leave the posts waiting for the 100 ms dispatch | |
| 391 | - | // timeout. | |
| 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. | |
| 392 | 393 | func TestPostMultiProducerNoLostWake(t *testing.T) { | |
| 393 | - | d := runDisplay(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 | + | }) | |
| 394 | 410 | const producers, perProducer, rounds = 8, 250, 10 | |
| 395 | - | for round := range rounds { | |
| 411 | + | for range rounds { | |
| 396 | 412 | // Let the loop go idle in epoll_wait. | |
| 397 | 413 | time.Sleep(5 * time.Millisecond) | |
| 398 | 414 | var left atomic.Int32 | |
| 399 | 415 | left.Store(producers * perProducer) | |
| 400 | 416 | done := make(chan struct{}) | |
| 401 | - | start := time.Now() | |
| 402 | 417 | var wg sync.WaitGroup | |
| 403 | 418 | for range producers { | |
| 404 | 419 | wg.Add(1) | |
| @@ -413,10 +428,7 @@ func TestPostMultiProducerNoLostWake(t *testing.T) { | |||
| 413 | 428 | } | |
| 414 | 429 | }() | |
| 415 | 430 | } | |
| 416 | - | waitFor(t, done, "all posts") | |
| 417 | - | if e := time.Since(start); e > 90*time.Millisecond { | |
| 418 | - | t.Fatalf("round %d: %d posts took %v; a wake was lost", round, producers*perProducer, e) | |
| 419 | - | } | |
| 431 | + | waitFor(t, done, "all posts (a lost wake hangs here)") | |
| 420 | 432 | wg.Wait() | |
| 421 | 433 | } | |
| 422 | 434 | } | |
| @@ -473,9 +485,15 @@ func TestDoPostRaceWithCancel(t *testing.T) { | |||
| 473 | 485 | if err != nil { | |
| 474 | 486 | t.Fatal(err) | |
| 475 | 487 | } | |
| 476 | - | // Force the reuse: make the old wake fd number refer to efd. | |
| 477 | - | if err := unix.Dup3(efd, d.wake, unix.O_CLOEXEC); err != nil { | |
| 478 | - | t.Fatal(err) | |
| 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 | + | } | |
| 479 | 497 | } | |
| 480 | 498 | for range 50 { | |
| 481 | 499 | d.Do(func() { t.Error("Do function ran after Stopped") }) | |
| @@ -483,11 +501,13 @@ func TestDoPostRaceWithCancel(t *testing.T) { | |||
| 483 | 501 | } | |
| 484 | 502 | wg.Wait() | |
| 485 | 503 | var b [8]byte | |
| 486 | - | if _, err := unix.Read(efd, b[:]); err != unix.EAGAIN { | |
| 504 | + | if _, err := unix.Read(efd, b[:]); !errors.Is(err, unix.EAGAIN) { | |
| 487 | 505 | t.Fatalf("eventfd %d (display wake fd was %d) was written after stop: %v", efd, d.wake, err) | |
| 488 | 506 | } | |
| 489 | 507 | unix.Close(efd) | |
| 490 | - | unix.Close(d.wake) | |
| 508 | + | if efd != d.wake { | |
| 509 | + | unix.Close(d.wake) | |
| 510 | + | } | |
| 491 | 511 | } | |
| 492 | 512 | } | |
| 493 | 513 | ||
| @@ -2,6 +2,7 @@ package server | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "errors" | |||
| 5 | "sync" | 6 | "sync" | ||
| 6 | "sync/atomic" | 7 | "sync/atomic" | ||
| 7 | "testing" | 8 | "testing" | ||
| @@ -387,18 +388,32 @@ func TestAddFDErrors(t *testing.T) { | |||||
| 387 | } | 388 | } | ||
| 388 | 389 | ||||
| 389 | // Posts from many goroutines must wake the loop without any Do to rescue a | 390 | // Posts from many goroutines must wake the loop without any Do to rescue a | ||
| 390 | - | // lost wake: a lost wake would leave the posts waiting for the 100 ms dispatch | 391 | + | // lost wake. The dispatch timeout is raised far past waitFor's limit, so a |
| 391 | - | // timeout. | 392 | + | // lost wake hangs the round instead of only slowing it. |
| 392 | func TestPostMultiProducerNoLostWake(t *testing.T) { | 393 | func TestPostMultiProducerNoLostWake(t *testing.T) { | ||
| 393 | - | d := runDisplay(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 | + | }) | |||
| 394 | const producers, perProducer, rounds = 8, 250, 10 | 410 | const producers, perProducer, rounds = 8, 250, 10 | ||
| 395 | - | for round := range rounds { | 411 | + | for range rounds { |
| 396 | // Let the loop go idle in epoll_wait. | 412 | // Let the loop go idle in epoll_wait. | ||
| 397 | time.Sleep(5 * time.Millisecond) | 413 | time.Sleep(5 * time.Millisecond) | ||
| 398 | var left atomic.Int32 | 414 | var left atomic.Int32 | ||
| 399 | left.Store(producers * perProducer) | 415 | left.Store(producers * perProducer) | ||
| 400 | done := make(chan struct{}) | 416 | done := make(chan struct{}) | ||
| 401 | - | start := time.Now() | |||
| 402 | var wg sync.WaitGroup | 417 | var wg sync.WaitGroup | ||
| 403 | for range producers { | 418 | for range producers { | ||
| 404 | wg.Add(1) | 419 | wg.Add(1) | ||
| @@ -413,10 +428,7 @@ func TestPostMultiProducerNoLostWake(t *testing.T) { | |||||
| 413 | } | 428 | } | ||
| 414 | }() | 429 | }() | ||
| 415 | } | 430 | } | ||
| 416 | - | waitFor(t, done, "all posts") | 431 | + | waitFor(t, done, "all posts (a lost wake hangs here)") |
| 417 | - | if e := time.Since(start); e > 90*time.Millisecond { | |||
| 418 | - | t.Fatalf("round %d: %d posts took %v; a wake was lost", round, producers*perProducer, e) | |||
| 419 | - | } | |||
| 420 | wg.Wait() | 432 | wg.Wait() | ||
| 421 | } | 433 | } | ||
| 422 | } | 434 | } | ||
| @@ -473,9 +485,15 @@ func TestDoPostRaceWithCancel(t *testing.T) { | |||||
| 473 | if err != nil { | 485 | if err != nil { | ||
| 474 | t.Fatal(err) | 486 | t.Fatal(err) | ||
| 475 | } | 487 | } | ||
| 476 | - | // Force the reuse: make the old wake fd number refer to efd. | 488 | + | // Force the reuse: make the old wake fd number refer to efd, unless |
| 477 | - | if err := unix.Dup3(efd, d.wake, unix.O_CLOEXEC); err != nil { | 489 | + | // efd already got that number. Never clobber an fd in use. |
| 478 | - | t.Fatal(err) | 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 | + | } | |||
| 479 | } | 497 | } | ||
| 480 | for range 50 { | 498 | for range 50 { | ||
| 481 | d.Do(func() { t.Error("Do function ran after Stopped") }) | 499 | d.Do(func() { t.Error("Do function ran after Stopped") }) | ||
| @@ -483,11 +501,13 @@ func TestDoPostRaceWithCancel(t *testing.T) { | |||||
| 483 | } | 501 | } | ||
| 484 | wg.Wait() | 502 | wg.Wait() | ||
| 485 | var b [8]byte | 503 | var b [8]byte | ||
| 486 | - | if _, err := unix.Read(efd, b[:]); err != unix.EAGAIN { | 504 | + | if _, err := unix.Read(efd, b[:]); !errors.Is(err, unix.EAGAIN) { |
| 487 | t.Fatalf("eventfd %d (display wake fd was %d) was written after stop: %v", efd, d.wake, err) | 505 | t.Fatalf("eventfd %d (display wake fd was %d) was written after stop: %v", efd, d.wake, err) | ||
| 488 | } | 506 | } | ||
| 489 | unix.Close(efd) | 507 | unix.Close(efd) | ||
| 490 | - | unix.Close(d.wake) | 508 | + | if efd != d.wake { |
| 509 | + | unix.Close(d.wake) | |||
| 510 | + | } | |||
| 491 | } | 511 | } | ||
| 492 | } | 512 | } | ||
| 493 | 513 | ||||