fix(server): guard Do wake against a closed eventfd and harden Post tests
- commit
- 046f9f020608ccbbb8d4e67f0c4b00ef126c929a
- author
- brice · · 2026-10-13 15:38 UTC
- parent
- 58529e2
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +206 -8
M server/display.go +32 -6
| @@ -54,11 +54,12 @@ var live = struct { | |||
| 54 | 54 | nextFD uintptr | |
| 55 | 55 | }{resources: map[uintptr]*Resource{}, binds: map[uintptr]BindFunc{}, clientGone: map[uintptr]func(){}, fds: map[uintptr]func(){}} | |
| 56 | 56 | ||
| 57 | - | // Display owns a wl_display. All methods except Do must run on the goroutine | |
| 58 | - | // that calls Run, or before Run starts. | |
| 57 | + | // Display owns a wl_display. All methods except Do and Post must run on the | |
| 58 | + | // goroutine that calls Run, or before Run starts. | |
| 59 | 59 | // | |
| 60 | 60 | // Resource, bind and client-listener state is shared by the whole process, so | |
| 61 | 61 | // only one Display may be running at a time per process. | |
| 62 | + | // | |
| 62 | 63 | // Do and Post wake the loop through an eventfd so calls run at once instead of | |
| 63 | 64 | // after the dispatch timeout. | |
| 64 | 65 | type Display struct { | |
| @@ -198,6 +199,7 @@ func (d *Display) destroy() { | |||
| 198 | 199 | for s := range d.sources { | |
| 199 | 200 | s.remove() | |
| 200 | 201 | } | |
| 202 | + | clear(d.sources) | |
| 201 | 203 | wlDisplayDestroyClients(d.c) | |
| 202 | 204 | wlEventSourceRemove(d.wakeSrc) | |
| 203 | 205 | wlDisplayDestroy(d.c) | |
| @@ -218,7 +220,9 @@ func (d *Display) Run(ctx context.Context) error { | |||
| 218 | 220 | if ctx.Err() != nil { | |
| 219 | 221 | return nil | |
| 220 | 222 | } | |
| 221 | - | // Run every queued post and call; each also wrote the eventfd. | |
| 223 | + | // Run every queued post and call. Wakes are coalesced, so the eventfd | |
| 224 | + | // is not written once per item; it only makes the dispatch below | |
| 225 | + | // return at once when work arrives while the loop is asleep. | |
| 222 | 226 | d.drain() | |
| 223 | 227 | wlDisplayFlushClients(d.c) | |
| 224 | 228 | // The timeout only bounds how late ctx cancellation is seen. | |
| @@ -284,12 +288,26 @@ func (d *Display) runPosts() bool { | |||
| 284 | 288 | // goroutines is safe and avoids a per-call buffer (which escapes under -race). | |
| 285 | 289 | var wakeOne = [8]byte{1} | |
| 286 | 290 | ||
| 287 | - | // wakeLoop writes the eventfd on behalf of posts. | |
| 291 | + | // wakeLoop writes the eventfd on behalf of posts. The caller must hold postMu | |
| 292 | + | // (or be the display goroutine, which destroy cannot overlap) so the fd cannot | |
| 293 | + | // be closed, and its number reused, underneath the write. | |
| 288 | 294 | func (d *Display) wakeLoop() { | |
| 289 | 295 | d.wakeWrites.Add(1) | |
| 290 | 296 | _, _ = unix.Write(d.wake, wakeOne[:]) | |
| 291 | 297 | } | |
| 292 | 298 | ||
| 299 | + | // wakeGuarded wakes the loop from any goroutine: it takes postMu and skips the | |
| 300 | + | // write once the display is closed, so a late caller never writes to a reused | |
| 301 | + | // fd number. | |
| 302 | + | func (d *Display) wakeGuarded() { | |
| 303 | + | d.postMu.Lock() | |
| 304 | + | defer d.postMu.Unlock() | |
| 305 | + | if d.closed.Load() { | |
| 306 | + | return | |
| 307 | + | } | |
| 308 | + | _, _ = unix.Write(d.wake, wakeOne[:]) | |
| 309 | + | } | |
| 310 | + | ||
| 293 | 311 | // Post queues fn to run on the display goroutine and returns at once. Unlike | |
| 294 | 312 | // Do it never waits, and unlike Do it may be called from any goroutine, | |
| 295 | 313 | // including the display goroutine itself. | |
| @@ -305,6 +323,10 @@ func (d *Display) wakeLoop() { | |||
| 305 | 323 | // the display has stopped or been closed, Post silently drops fn, and posts | |
| 306 | 324 | // that had not run when the display stopped are dropped too. In steady state | |
| 307 | 325 | // Post does not allocate. | |
| 326 | + | // | |
| 327 | + | // The queue is unbounded: Post applies no backpressure, so a producer that | |
| 328 | + | // outpaces the display goroutine grows memory without limit. Callers that can | |
| 329 | + | // produce faster than the loop runs must bound or coalesce their own posts. | |
| 308 | 330 | func (d *Display) Post(fn func()) { | |
| 309 | 331 | if fn == nil { | |
| 310 | 332 | return | |
| @@ -333,6 +355,11 @@ type Source struct { | |||
| 333 | 355 | // the eventfd or timerfd, accept, ...) or it is called again at once. It | |
| 334 | 356 | // should not block. | |
| 335 | 357 | // | |
| 358 | + | // onReadable is also called when the fd hangs up or errors (EPOLLHUP, | |
| 359 | + | // EPOLLERR), which level-triggered polling keeps reporting. The callback should | |
| 360 | + | // read the fd, and on EOF or error call Remove (it may do so from within the | |
| 361 | + | // callback), or it is called again at once. | |
| 362 | + | // | |
| 336 | 363 | // libwayland duplicates fd; the caller still owns and must close its own | |
| 337 | 364 | // descriptor, and should do so after Remove. Call AddFD only on the display | |
| 338 | 365 | // goroutine or before Run starts. Sources not removed are released when the | |
| @@ -387,8 +414,7 @@ func (d *Display) Do(fn func()) bool { | |||
| 387 | 414 | case <-d.stopped: | |
| 388 | 415 | return false | |
| 389 | 416 | } | |
| 390 | - | var one = [8]byte{1} | |
| 391 | - | _, _ = unix.Write(d.wake, one[:]) | |
| 417 | + | d.wakeGuarded() | |
| 392 | 418 | select { | |
| 393 | 419 | case <-done: | |
| 394 | 420 | return true | |
| @@ -54,11 +54,12 @@ var live = struct { | |||||
| 54 | nextFD uintptr | 54 | nextFD uintptr | ||
| 55 | }{resources: map[uintptr]*Resource{}, binds: map[uintptr]BindFunc{}, clientGone: map[uintptr]func(){}, fds: map[uintptr]func(){}} | 55 | }{resources: map[uintptr]*Resource{}, binds: map[uintptr]BindFunc{}, clientGone: map[uintptr]func(){}, fds: map[uintptr]func(){}} | ||
| 56 | 56 | ||||
| 57 | - | // Display owns a wl_display. All methods except Do must run on the goroutine | 57 | + | // Display owns a wl_display. All methods except Do and Post must run on the |
| 58 | - | // that calls Run, or before Run starts. | 58 | + | // goroutine that calls Run, or before Run starts. |
| 59 | // | 59 | // | ||
| 60 | // Resource, bind and client-listener state is shared by the whole process, so | 60 | // Resource, bind and client-listener state is shared by the whole process, so | ||
| 61 | // only one Display may be running at a time per process. | 61 | // only one Display may be running at a time per process. | ||
| 62 | + | // | |||
| 62 | // Do and Post wake the loop through an eventfd so calls run at once instead of | 63 | // Do and Post wake the loop through an eventfd so calls run at once instead of | ||
| 63 | // after the dispatch timeout. | 64 | // after the dispatch timeout. | ||
| 64 | type Display struct { | 65 | type Display struct { | ||
| @@ -198,6 +199,7 @@ func (d *Display) destroy() { | |||||
| 198 | for s := range d.sources { | 199 | for s := range d.sources { | ||
| 199 | s.remove() | 200 | s.remove() | ||
| 200 | } | 201 | } | ||
| 202 | + | clear(d.sources) | |||
| 201 | wlDisplayDestroyClients(d.c) | 203 | wlDisplayDestroyClients(d.c) | ||
| 202 | wlEventSourceRemove(d.wakeSrc) | 204 | wlEventSourceRemove(d.wakeSrc) | ||
| 203 | wlDisplayDestroy(d.c) | 205 | wlDisplayDestroy(d.c) | ||
| @@ -218,7 +220,9 @@ func (d *Display) Run(ctx context.Context) error { | |||||
| 218 | if ctx.Err() != nil { | 220 | if ctx.Err() != nil { | ||
| 219 | return nil | 221 | return nil | ||
| 220 | } | 222 | } | ||
| 221 | - | // Run every queued post and call; each also wrote the eventfd. | 223 | + | // Run every queued post and call. Wakes are coalesced, so the eventfd |
| 224 | + | // is not written once per item; it only makes the dispatch below | |||
| 225 | + | // return at once when work arrives while the loop is asleep. | |||
| 222 | d.drain() | 226 | d.drain() | ||
| 223 | wlDisplayFlushClients(d.c) | 227 | wlDisplayFlushClients(d.c) | ||
| 224 | // The timeout only bounds how late ctx cancellation is seen. | 228 | // The timeout only bounds how late ctx cancellation is seen. | ||
| @@ -284,12 +288,26 @@ func (d *Display) runPosts() bool { | |||||
| 284 | // goroutines is safe and avoids a per-call buffer (which escapes under -race). | 288 | // goroutines is safe and avoids a per-call buffer (which escapes under -race). | ||
| 285 | var wakeOne = [8]byte{1} | 289 | var wakeOne = [8]byte{1} | ||
| 286 | 290 | ||||
| 287 | - | // wakeLoop writes the eventfd on behalf of posts. | 291 | + | // wakeLoop writes the eventfd on behalf of posts. The caller must hold postMu |
| 292 | + | // (or be the display goroutine, which destroy cannot overlap) so the fd cannot | |||
| 293 | + | // be closed, and its number reused, underneath the write. | |||
| 288 | func (d *Display) wakeLoop() { | 294 | func (d *Display) wakeLoop() { | ||
| 289 | d.wakeWrites.Add(1) | 295 | d.wakeWrites.Add(1) | ||
| 290 | _, _ = unix.Write(d.wake, wakeOne[:]) | 296 | _, _ = unix.Write(d.wake, wakeOne[:]) | ||
| 291 | } | 297 | } | ||
| 292 | 298 | ||||
| 299 | + | // wakeGuarded wakes the loop from any goroutine: it takes postMu and skips the | |||
| 300 | + | // write once the display is closed, so a late caller never writes to a reused | |||
| 301 | + | // fd number. | |||
| 302 | + | func (d *Display) wakeGuarded() { | |||
| 303 | + | d.postMu.Lock() | |||
| 304 | + | defer d.postMu.Unlock() | |||
| 305 | + | if d.closed.Load() { | |||
| 306 | + | return | |||
| 307 | + | } | |||
| 308 | + | _, _ = unix.Write(d.wake, wakeOne[:]) | |||
| 309 | + | } | |||
| 310 | + | ||||
| 293 | // Post queues fn to run on the display goroutine and returns at once. Unlike | 311 | // Post queues fn to run on the display goroutine and returns at once. Unlike | ||
| 294 | // Do it never waits, and unlike Do it may be called from any goroutine, | 312 | // Do it never waits, and unlike Do it may be called from any goroutine, | ||
| 295 | // including the display goroutine itself. | 313 | // including the display goroutine itself. | ||
| @@ -305,6 +323,10 @@ func (d *Display) wakeLoop() { | |||||
| 305 | // the display has stopped or been closed, Post silently drops fn, and posts | 323 | // the display has stopped or been closed, Post silently drops fn, and posts | ||
| 306 | // that had not run when the display stopped are dropped too. In steady state | 324 | // that had not run when the display stopped are dropped too. In steady state | ||
| 307 | // Post does not allocate. | 325 | // Post does not allocate. | ||
| 326 | + | // | |||
| 327 | + | // The queue is unbounded: Post applies no backpressure, so a producer that | |||
| 328 | + | // outpaces the display goroutine grows memory without limit. Callers that can | |||
| 329 | + | // produce faster than the loop runs must bound or coalesce their own posts. | |||
| 308 | func (d *Display) Post(fn func()) { | 330 | func (d *Display) Post(fn func()) { | ||
| 309 | if fn == nil { | 331 | if fn == nil { | ||
| 310 | return | 332 | return | ||
| @@ -333,6 +355,11 @@ type Source struct { | |||||
| 333 | // the eventfd or timerfd, accept, ...) or it is called again at once. It | 355 | // the eventfd or timerfd, accept, ...) or it is called again at once. It | ||
| 334 | // should not block. | 356 | // should not block. | ||
| 335 | // | 357 | // | ||
| 358 | + | // onReadable is also called when the fd hangs up or errors (EPOLLHUP, | |||
| 359 | + | // EPOLLERR), which level-triggered polling keeps reporting. The callback should | |||
| 360 | + | // read the fd, and on EOF or error call Remove (it may do so from within the | |||
| 361 | + | // callback), or it is called again at once. | |||
| 362 | + | // | |||
| 336 | // libwayland duplicates fd; the caller still owns and must close its own | 363 | // libwayland duplicates fd; the caller still owns and must close its own | ||
| 337 | // descriptor, and should do so after Remove. Call AddFD only on the display | 364 | // descriptor, and should do so after Remove. Call AddFD only on the display | ||
| 338 | // goroutine or before Run starts. Sources not removed are released when the | 365 | // goroutine or before Run starts. Sources not removed are released when the | ||
| @@ -387,8 +414,7 @@ func (d *Display) Do(fn func()) bool { | |||||
| 387 | case <-d.stopped: | 414 | case <-d.stopped: | ||
| 388 | return false | 415 | return false | ||
| 389 | } | 416 | } | ||
| 390 | - | var one = [8]byte{1} | 417 | + | d.wakeGuarded() |
| 391 | - | _, _ = unix.Write(d.wake, one[:]) | |||
| 392 | select { | 418 | select { | ||
| 393 | case <-done: | 419 | case <-done: | ||
| 394 | return true | 420 | return true | ||
M server/post_test.go +174 -2
| @@ -3,6 +3,7 @@ package server | |||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | 5 | "sync" | |
| 6 | + | "sync/atomic" | |
| 6 | 7 | "testing" | |
| 7 | 8 | "time" | |
| 8 | 9 | ||
| @@ -197,7 +198,7 @@ func TestPostSelfRepostingIsCappedAndResumes(t *testing.T) { | |||
| 197 | 198 | start := time.Now() | |
| 198 | 199 | d.Post(step) | |
| 199 | 200 | waitFor(t, done, "self-reposting chain") | |
| 200 | - | if e := time.Since(start); e > 50*time.Millisecond { | |
| 201 | + | if e := time.Since(start); e > 500*time.Millisecond { | |
| 201 | 202 | t.Fatalf("chain of %d posts took %v; wake not re-armed at the round cap", total, e) | |
| 202 | 203 | } | |
| 203 | 204 | // Do still gets served. | |
| @@ -305,7 +306,7 @@ func TestAddFDTimerfdFiresOnce(t *testing.T) { | |||
| 305 | 306 | t.Fatal(err) | |
| 306 | 307 | } | |
| 307 | 308 | waitFor(t, fired, "timerfd") | |
| 308 | - | if e := time.Since(start); e > 50*time.Millisecond { | |
| 309 | + | if e := time.Since(start); e > 500*time.Millisecond { | |
| 309 | 310 | t.Fatalf("1 ms timer fired after %v", e) | |
| 310 | 311 | } | |
| 311 | 312 | time.Sleep(50 * time.Millisecond) | |
| @@ -357,6 +358,7 @@ func TestAddFDErrors(t *testing.T) { | |||
| 357 | 358 | if err != nil { | |
| 358 | 359 | t.Fatal(err) | |
| 359 | 360 | } | |
| 361 | + | defer d.Close() | |
| 360 | 362 | if _, err := d.AddFD(-1, func() {}); err == nil { | |
| 361 | 363 | t.Error("negative fd accepted") | |
| 362 | 364 | } | |
| @@ -383,3 +385,173 @@ func TestAddFDErrors(t *testing.T) { | |||
| 383 | 385 | t.Errorf("%d fd callbacks leaked", len(live.fds)) | |
| 384 | 386 | } | |
| 385 | 387 | } | |
| 388 | + | ||
| 389 | + | // 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. | |
| 392 | + | func TestPostMultiProducerNoLostWake(t *testing.T) { | |
| 393 | + | d := runDisplay(t) | |
| 394 | + | const producers, perProducer, rounds = 8, 250, 10 | |
| 395 | + | for round := range rounds { | |
| 396 | + | // Let the loop go idle in epoll_wait. | |
| 397 | + | time.Sleep(5 * time.Millisecond) | |
| 398 | + | var left atomic.Int32 | |
| 399 | + | left.Store(producers * perProducer) | |
| 400 | + | done := make(chan struct{}) | |
| 401 | + | start := time.Now() | |
| 402 | + | var wg sync.WaitGroup | |
| 403 | + | for range producers { | |
| 404 | + | wg.Add(1) | |
| 405 | + | go func() { | |
| 406 | + | defer wg.Done() | |
| 407 | + | for range perProducer { | |
| 408 | + | d.Post(func() { | |
| 409 | + | if left.Add(-1) == 0 { | |
| 410 | + | close(done) | |
| 411 | + | } | |
| 412 | + | }) | |
| 413 | + | } | |
| 414 | + | }() | |
| 415 | + | } | |
| 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 | + | } | |
| 420 | + | wg.Wait() | |
| 421 | + | } | |
| 422 | + | } | |
| 423 | + | ||
| 424 | + | // Do and Post racing with shutdown must neither panic nor run anything after | |
| 425 | + | // Stopped, and Do's wake must not write to the closed eventfd's number once it | |
| 426 | + | // is reused. | |
| 427 | + | func TestDoPostRaceWithCancel(t *testing.T) { | |
| 428 | + | for range 20 { | |
| 429 | + | d, err := NewDisplay() | |
| 430 | + | if err != nil { | |
| 431 | + | t.Fatal(err) | |
| 432 | + | } | |
| 433 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 434 | + | done := make(chan error, 1) | |
| 435 | + | go func() { done <- d.Run(ctx) }() | |
| 436 | + | ||
| 437 | + | var stopped atomic.Bool | |
| 438 | + | var wg sync.WaitGroup | |
| 439 | + | for range 4 { | |
| 440 | + | wg.Add(2) | |
| 441 | + | go func() { | |
| 442 | + | defer wg.Done() | |
| 443 | + | for range 200 { | |
| 444 | + | d.Do(func() { | |
| 445 | + | if stopped.Load() { | |
| 446 | + | t.Error("Do function ran after Stopped") | |
| 447 | + | } | |
| 448 | + | }) | |
| 449 | + | } | |
| 450 | + | }() | |
| 451 | + | go func() { | |
| 452 | + | defer wg.Done() | |
| 453 | + | for range 200 { | |
| 454 | + | d.Post(func() { | |
| 455 | + | if stopped.Load() { | |
| 456 | + | t.Error("post ran after Stopped") | |
| 457 | + | } | |
| 458 | + | }) | |
| 459 | + | } | |
| 460 | + | }() | |
| 461 | + | } | |
| 462 | + | time.Sleep(time.Millisecond) | |
| 463 | + | cancel() | |
| 464 | + | if err := <-done; err != nil { | |
| 465 | + | t.Fatal(err) | |
| 466 | + | } | |
| 467 | + | <-d.Stopped() | |
| 468 | + | stopped.Store(true) | |
| 469 | + | ||
| 470 | + | // The closed wake fd's number is free for reuse now. Late Do and Post | |
| 471 | + | // calls must not write to whoever gets it. | |
| 472 | + | efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK) | |
| 473 | + | if err != nil { | |
| 474 | + | t.Fatal(err) | |
| 475 | + | } | |
| 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) | |
| 479 | + | } | |
| 480 | + | for range 50 { | |
| 481 | + | d.Do(func() { t.Error("Do function ran after Stopped") }) | |
| 482 | + | d.Post(func() { t.Error("post ran after Stopped") }) | |
| 483 | + | } | |
| 484 | + | wg.Wait() | |
| 485 | + | var b [8]byte | |
| 486 | + | if _, err := unix.Read(efd, b[:]); err != unix.EAGAIN { | |
| 487 | + | t.Fatalf("eventfd %d (display wake fd was %d) was written after stop: %v", efd, d.wake, err) | |
| 488 | + | } | |
| 489 | + | unix.Close(efd) | |
| 490 | + | unix.Close(d.wake) | |
| 491 | + | } | |
| 492 | + | } | |
| 493 | + | ||
| 494 | + | // Sources still registered when Run exits are released with the display. | |
| 495 | + | func TestRunExitReleasesSources(t *testing.T) { | |
| 496 | + | d, err := NewDisplay() | |
| 497 | + | if err != nil { | |
| 498 | + | t.Fatal(err) | |
| 499 | + | } | |
| 500 | + | efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK) | |
| 501 | + | if err != nil { | |
| 502 | + | t.Fatal(err) | |
| 503 | + | } | |
| 504 | + | defer unix.Close(efd) | |
| 505 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 506 | + | done := make(chan error, 1) | |
| 507 | + | go func() { done <- d.Run(ctx) }() | |
| 508 | + | var addErr error | |
| 509 | + | if !d.Do(func() { _, addErr = d.AddFD(efd, func() {}) }) || addErr != nil { | |
| 510 | + | t.Fatalf("AddFD: %v", addErr) | |
| 511 | + | } | |
| 512 | + | cancel() | |
| 513 | + | if err := <-done; err != nil { | |
| 514 | + | t.Fatal(err) | |
| 515 | + | } | |
| 516 | + | <-d.Stopped() | |
| 517 | + | if len(live.fds) != 0 { | |
| 518 | + | t.Errorf("%d fd callbacks leaked after Run exited", len(live.fds)) | |
| 519 | + | } | |
| 520 | + | if len(d.sources) != 0 { | |
| 521 | + | t.Errorf("%d sources still tracked after Run exited", len(d.sources)) | |
| 522 | + | } | |
| 523 | + | } | |
| 524 | + | ||
| 525 | + | func TestPostFromAddFDCallback(t *testing.T) { | |
| 526 | + | d := runDisplay(t) | |
| 527 | + | want := displayTID(t, d) | |
| 528 | + | efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK) | |
| 529 | + | if err != nil { | |
| 530 | + | t.Fatal(err) | |
| 531 | + | } | |
| 532 | + | defer unix.Close(efd) | |
| 533 | + | got := make(chan int, 1) | |
| 534 | + | var src *Source | |
| 535 | + | if !d.Do(func() { | |
| 536 | + | src, err = d.AddFD(efd, func() { | |
| 537 | + | var b [8]byte | |
| 538 | + | _, _ = unix.Read(efd, b[:]) | |
| 539 | + | d.Post(func() { got <- unix.Gettid() }) | |
| 540 | + | }) | |
| 541 | + | }) || err != nil { | |
| 542 | + | t.Fatalf("AddFD: %v", err) | |
| 543 | + | } | |
| 544 | + | one := [8]byte{1} | |
| 545 | + | if _, err := unix.Write(efd, one[:]); err != nil { | |
| 546 | + | t.Fatal(err) | |
| 547 | + | } | |
| 548 | + | select { | |
| 549 | + | case tid := <-got: | |
| 550 | + | if tid != want { | |
| 551 | + | t.Fatalf("post ran on thread %d, display thread is %d", tid, want) | |
| 552 | + | } | |
| 553 | + | case <-time.After(3 * time.Second): | |
| 554 | + | t.Fatal("post from AddFD callback did not run") | |
| 555 | + | } | |
| 556 | + | d.Do(func() { src.Remove() }) | |
| 557 | + | } | |
| @@ -3,6 +3,7 @@ package server | |||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | "sync" | 5 | "sync" | ||
| 6 | + | "sync/atomic" | |||
| 6 | "testing" | 7 | "testing" | ||
| 7 | "time" | 8 | "time" | ||
| 8 | 9 | ||||
| @@ -197,7 +198,7 @@ func TestPostSelfRepostingIsCappedAndResumes(t *testing.T) { | |||||
| 197 | start := time.Now() | 198 | start := time.Now() | ||
| 198 | d.Post(step) | 199 | d.Post(step) | ||
| 199 | waitFor(t, done, "self-reposting chain") | 200 | waitFor(t, done, "self-reposting chain") | ||
| 200 | - | if e := time.Since(start); e > 50*time.Millisecond { | 201 | + | if e := time.Since(start); e > 500*time.Millisecond { |
| 201 | t.Fatalf("chain of %d posts took %v; wake not re-armed at the round cap", total, e) | 202 | t.Fatalf("chain of %d posts took %v; wake not re-armed at the round cap", total, e) | ||
| 202 | } | 203 | } | ||
| 203 | // Do still gets served. | 204 | // Do still gets served. | ||
| @@ -305,7 +306,7 @@ func TestAddFDTimerfdFiresOnce(t *testing.T) { | |||||
| 305 | t.Fatal(err) | 306 | t.Fatal(err) | ||
| 306 | } | 307 | } | ||
| 307 | waitFor(t, fired, "timerfd") | 308 | waitFor(t, fired, "timerfd") | ||
| 308 | - | if e := time.Since(start); e > 50*time.Millisecond { | 309 | + | if e := time.Since(start); e > 500*time.Millisecond { |
| 309 | t.Fatalf("1 ms timer fired after %v", e) | 310 | t.Fatalf("1 ms timer fired after %v", e) | ||
| 310 | } | 311 | } | ||
| 311 | time.Sleep(50 * time.Millisecond) | 312 | time.Sleep(50 * time.Millisecond) | ||
| @@ -357,6 +358,7 @@ func TestAddFDErrors(t *testing.T) { | |||||
| 357 | if err != nil { | 358 | if err != nil { | ||
| 358 | t.Fatal(err) | 359 | t.Fatal(err) | ||
| 359 | } | 360 | } | ||
| 361 | + | defer d.Close() | |||
| 360 | if _, err := d.AddFD(-1, func() {}); err == nil { | 362 | if _, err := d.AddFD(-1, func() {}); err == nil { | ||
| 361 | t.Error("negative fd accepted") | 363 | t.Error("negative fd accepted") | ||
| 362 | } | 364 | } | ||
| @@ -383,3 +385,173 @@ func TestAddFDErrors(t *testing.T) { | |||||
| 383 | t.Errorf("%d fd callbacks leaked", len(live.fds)) | 385 | t.Errorf("%d fd callbacks leaked", len(live.fds)) | ||
| 384 | } | 386 | } | ||
| 385 | } | 387 | } | ||
| 388 | + | ||||
| 389 | + | // 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. | |||
| 392 | + | func TestPostMultiProducerNoLostWake(t *testing.T) { | |||
| 393 | + | d := runDisplay(t) | |||
| 394 | + | const producers, perProducer, rounds = 8, 250, 10 | |||
| 395 | + | for round := range rounds { | |||
| 396 | + | // Let the loop go idle in epoll_wait. | |||
| 397 | + | time.Sleep(5 * time.Millisecond) | |||
| 398 | + | var left atomic.Int32 | |||
| 399 | + | left.Store(producers * perProducer) | |||
| 400 | + | done := make(chan struct{}) | |||
| 401 | + | start := time.Now() | |||
| 402 | + | var wg sync.WaitGroup | |||
| 403 | + | for range producers { | |||
| 404 | + | wg.Add(1) | |||
| 405 | + | go func() { | |||
| 406 | + | defer wg.Done() | |||
| 407 | + | for range perProducer { | |||
| 408 | + | d.Post(func() { | |||
| 409 | + | if left.Add(-1) == 0 { | |||
| 410 | + | close(done) | |||
| 411 | + | } | |||
| 412 | + | }) | |||
| 413 | + | } | |||
| 414 | + | }() | |||
| 415 | + | } | |||
| 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 | + | } | |||
| 420 | + | wg.Wait() | |||
| 421 | + | } | |||
| 422 | + | } | |||
| 423 | + | ||||
| 424 | + | // Do and Post racing with shutdown must neither panic nor run anything after | |||
| 425 | + | // Stopped, and Do's wake must not write to the closed eventfd's number once it | |||
| 426 | + | // is reused. | |||
| 427 | + | func TestDoPostRaceWithCancel(t *testing.T) { | |||
| 428 | + | for range 20 { | |||
| 429 | + | d, err := NewDisplay() | |||
| 430 | + | if err != nil { | |||
| 431 | + | t.Fatal(err) | |||
| 432 | + | } | |||
| 433 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 434 | + | done := make(chan error, 1) | |||
| 435 | + | go func() { done <- d.Run(ctx) }() | |||
| 436 | + | ||||
| 437 | + | var stopped atomic.Bool | |||
| 438 | + | var wg sync.WaitGroup | |||
| 439 | + | for range 4 { | |||
| 440 | + | wg.Add(2) | |||
| 441 | + | go func() { | |||
| 442 | + | defer wg.Done() | |||
| 443 | + | for range 200 { | |||
| 444 | + | d.Do(func() { | |||
| 445 | + | if stopped.Load() { | |||
| 446 | + | t.Error("Do function ran after Stopped") | |||
| 447 | + | } | |||
| 448 | + | }) | |||
| 449 | + | } | |||
| 450 | + | }() | |||
| 451 | + | go func() { | |||
| 452 | + | defer wg.Done() | |||
| 453 | + | for range 200 { | |||
| 454 | + | d.Post(func() { | |||
| 455 | + | if stopped.Load() { | |||
| 456 | + | t.Error("post ran after Stopped") | |||
| 457 | + | } | |||
| 458 | + | }) | |||
| 459 | + | } | |||
| 460 | + | }() | |||
| 461 | + | } | |||
| 462 | + | time.Sleep(time.Millisecond) | |||
| 463 | + | cancel() | |||
| 464 | + | if err := <-done; err != nil { | |||
| 465 | + | t.Fatal(err) | |||
| 466 | + | } | |||
| 467 | + | <-d.Stopped() | |||
| 468 | + | stopped.Store(true) | |||
| 469 | + | ||||
| 470 | + | // The closed wake fd's number is free for reuse now. Late Do and Post | |||
| 471 | + | // calls must not write to whoever gets it. | |||
| 472 | + | efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK) | |||
| 473 | + | if err != nil { | |||
| 474 | + | t.Fatal(err) | |||
| 475 | + | } | |||
| 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) | |||
| 479 | + | } | |||
| 480 | + | for range 50 { | |||
| 481 | + | d.Do(func() { t.Error("Do function ran after Stopped") }) | |||
| 482 | + | d.Post(func() { t.Error("post ran after Stopped") }) | |||
| 483 | + | } | |||
| 484 | + | wg.Wait() | |||
| 485 | + | var b [8]byte | |||
| 486 | + | if _, err := unix.Read(efd, b[:]); err != unix.EAGAIN { | |||
| 487 | + | t.Fatalf("eventfd %d (display wake fd was %d) was written after stop: %v", efd, d.wake, err) | |||
| 488 | + | } | |||
| 489 | + | unix.Close(efd) | |||
| 490 | + | unix.Close(d.wake) | |||
| 491 | + | } | |||
| 492 | + | } | |||
| 493 | + | ||||
| 494 | + | // Sources still registered when Run exits are released with the display. | |||
| 495 | + | func TestRunExitReleasesSources(t *testing.T) { | |||
| 496 | + | d, err := NewDisplay() | |||
| 497 | + | if err != nil { | |||
| 498 | + | t.Fatal(err) | |||
| 499 | + | } | |||
| 500 | + | efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK) | |||
| 501 | + | if err != nil { | |||
| 502 | + | t.Fatal(err) | |||
| 503 | + | } | |||
| 504 | + | defer unix.Close(efd) | |||
| 505 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 506 | + | done := make(chan error, 1) | |||
| 507 | + | go func() { done <- d.Run(ctx) }() | |||
| 508 | + | var addErr error | |||
| 509 | + | if !d.Do(func() { _, addErr = d.AddFD(efd, func() {}) }) || addErr != nil { | |||
| 510 | + | t.Fatalf("AddFD: %v", addErr) | |||
| 511 | + | } | |||
| 512 | + | cancel() | |||
| 513 | + | if err := <-done; err != nil { | |||
| 514 | + | t.Fatal(err) | |||
| 515 | + | } | |||
| 516 | + | <-d.Stopped() | |||
| 517 | + | if len(live.fds) != 0 { | |||
| 518 | + | t.Errorf("%d fd callbacks leaked after Run exited", len(live.fds)) | |||
| 519 | + | } | |||
| 520 | + | if len(d.sources) != 0 { | |||
| 521 | + | t.Errorf("%d sources still tracked after Run exited", len(d.sources)) | |||
| 522 | + | } | |||
| 523 | + | } | |||
| 524 | + | ||||
| 525 | + | func TestPostFromAddFDCallback(t *testing.T) { | |||
| 526 | + | d := runDisplay(t) | |||
| 527 | + | want := displayTID(t, d) | |||
| 528 | + | efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK) | |||
| 529 | + | if err != nil { | |||
| 530 | + | t.Fatal(err) | |||
| 531 | + | } | |||
| 532 | + | defer unix.Close(efd) | |||
| 533 | + | got := make(chan int, 1) | |||
| 534 | + | var src *Source | |||
| 535 | + | if !d.Do(func() { | |||
| 536 | + | src, err = d.AddFD(efd, func() { | |||
| 537 | + | var b [8]byte | |||
| 538 | + | _, _ = unix.Read(efd, b[:]) | |||
| 539 | + | d.Post(func() { got <- unix.Gettid() }) | |||
| 540 | + | }) | |||
| 541 | + | }) || err != nil { | |||
| 542 | + | t.Fatalf("AddFD: %v", err) | |||
| 543 | + | } | |||
| 544 | + | one := [8]byte{1} | |||
| 545 | + | if _, err := unix.Write(efd, one[:]); err != nil { | |||
| 546 | + | t.Fatal(err) | |||
| 547 | + | } | |||
| 548 | + | select { | |||
| 549 | + | case tid := <-got: | |||
| 550 | + | if tid != want { | |||
| 551 | + | t.Fatalf("post ran on thread %d, display thread is %d", tid, want) | |||
| 552 | + | } | |||
| 553 | + | case <-time.After(3 * time.Second): | |||
| 554 | + | t.Fatal("post from AddFD callback did not run") | |||
| 555 | + | } | |||
| 556 | + | d.Do(func() { src.Remove() }) | |||
| 557 | + | } | |||