diff --git a/internal/adaptive/async_abort_791_linux_test.go b/internal/adaptive/async_abort_791_linux_test.go index 5343b8347..b9bc3d4f3 100644 --- a/internal/adaptive/async_abort_791_linux_test.go +++ b/internal/adaptive/async_abort_791_linux_test.go @@ -157,13 +157,9 @@ func abortAdaptive791(t *testing.T, adopted bool, inject func(*Engine)) { ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() - for dl := time.Now().Add(3 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - t.Fatal("adaptive engine never bound") - } + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() + awaitBoundOrFail890(t, e, cancel, done, ended) if inject != nil { inject(e) } diff --git a/internal/adaptive/detect_bound_974_linux_test.go b/internal/adaptive/detect_bound_974_linux_test.go index 5be80f0b0..09bbd4afb 100644 --- a/internal/adaptive/detect_bound_974_linux_test.go +++ b/internal/adaptive/detect_bound_974_linux_test.go @@ -156,21 +156,17 @@ func startOnIOUring974(t *testing.T, readHeaderTimeout time.Duration) string { } ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() t.Cleanup(func() { cancel() select { - case <-done: + case <-ended: // closed after done is sent on, whoever has drained it case <-time.After(10 * time.Second): t.Error("adaptive engine did not stop within 10s") } }) - for dl := time.Now().Add(10 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - t.Fatal("adaptive engine never bound") - } + awaitBoundOrFail890(t, e, cancel, done, ended) if got := getOnce(t, addr); got != 200 { t.Fatalf("GET = %d, want 200", got) } diff --git a/internal/adaptive/driver_provider_test.go b/internal/adaptive/driver_provider_test.go index c3cd526c0..ea9eceab7 100644 --- a/internal/adaptive/driver_provider_test.go +++ b/internal/adaptive/driver_provider_test.go @@ -51,18 +51,11 @@ func newBoundAdaptive(t *testing.T) (*Engine, func()) { ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() // Wait for bound. - deadline := time.Now().Add(3 * time.Second) - for e.Addr() == nil && time.Now().Before(deadline) { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - cancel() - <-done - t.Fatal("adaptive engine never bound") - } + awaitBoundOrFail890(t, e, cancel, done, ended) stop := func() { cancel() diff --git a/internal/adaptive/epoll_halves_e9_linux_test.go b/internal/adaptive/epoll_halves_e9_linux_test.go index 97c643cda..546533ee2 100644 --- a/internal/adaptive/epoll_halves_e9_linux_test.go +++ b/internal/adaptive/epoll_halves_e9_linux_test.go @@ -57,21 +57,17 @@ func startEpollAdaptive(t *testing.T, cfg resource.Config, h stream.Handler) (*E } ctx, cancel := context.WithCancel(context.Background()) errCh := make(chan error, 1) - go func() { errCh <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { errCh <- e.Listen(ctx); close(ended) }() t.Cleanup(func() { cancel() select { - case <-errCh: - case <-time.After(5 * time.Second): // 5 s: a test that waited for Listen has drained errCh already + case <-ended: // closed after errCh is sent on, whoever has drained it + case <-time.After(5 * time.Second): } _ = e.Shutdown(context.Background()) }) - for dl := time.Now().Add(15 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - t.Fatal("adaptive engine did not bind") - } + awaitBoundOrFail890(t, e, cancel, errCh, ended) if got := e.ActiveEngine().Type(); got != engine.Epoll { t.Fatalf("e9 PREMISE: the active engine is %v, want epoll: the epoll halves of #874 and #876 are not what this test would run", got) } diff --git a/internal/adaptive/flap_conns_per_ring_test.go b/internal/adaptive/flap_conns_per_ring_test.go index e82cc5f63..1038dc60a 100644 --- a/internal/adaptive/flap_conns_per_ring_test.go +++ b/internal/adaptive/flap_conns_per_ring_test.go @@ -296,6 +296,22 @@ func awaitBind890(addr func() net.Addr, ended <-chan struct{}, limit time.Durati } } +// awaitBoundOrFail890 is awaitBind890 for a test that started Listen itself +// (it keeps ctx's cancel and the done channel, to assert on the stop). The +// caller's Listen goroutine sends its result on done and then closes ended. On +// failure it cancels Listen, joins it and fails t with Listen's own error. +func awaitBoundOrFail890(t *testing.T, e *Engine, cancel context.CancelFunc, done <-chan error, ended <-chan struct{}) net.Addr { + t.Helper() + a, took, err := awaitBind890(e.Addr, ended, bindLimit890) + if err != nil { + cancel() + lerr := <-done + t.Fatalf("adaptive engine never bound: %v (waited %v; Listen: %v)", err, took, lerr) + } + t.Logf("adaptive engine bound %v after Listen started", took.Round(time.Millisecond)) + return a +} + // listenAndAwaitBind890 starts e.Listen, waits for the bound address (see // awaitBind890), logs how long the bind took, and returns the address with a // stop func that cancels Listen and joins it. It fails the test with Listen's diff --git a/internal/adaptive/pinned_threads_905_linux_test.go b/internal/adaptive/pinned_threads_905_linux_test.go index c1b855ea5..f6172d1e4 100644 --- a/internal/adaptive/pinned_threads_905_linux_test.go +++ b/internal/adaptive/pinned_threads_905_linux_test.go @@ -60,7 +60,8 @@ func startPromoted905(t *testing.T, base string) (int, func()) { e.FreezeSwitching() ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() stop := func() { cancel() select { @@ -72,13 +73,7 @@ func startPromoted905(t *testing.T, base string) (int, func()) { t.Fatal("the adaptive engine did not stop within 15s") } } - for dl := time.Now().Add(10 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(5 * time.Millisecond) - } - if e.Addr() == nil { - stop() - t.Fatal("the adaptive engine did not bind within 10s") - } + awaitBoundOrFail890(t, e, cancel, done, ended) if got := e.ActiveEngine().Type(); got != engine.Epoll { stop() t.Fatalf("this test promotes from epoll, but the engine started on %v", got) diff --git a/internal/adaptive/prebound_listener_test.go b/internal/adaptive/prebound_listener_test.go index cb1818440..720ef8d69 100644 --- a/internal/adaptive/prebound_listener_test.go +++ b/internal/adaptive/prebound_listener_test.go @@ -52,18 +52,10 @@ func preBoundAdaptive(t *testing.T, workers int) (*Engine, string, func()) { ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() - for dl := time.Now().Add(10 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - cancel() - if lerr := <-done; lerr != nil { - t.Fatalf("adaptive engine never bound; Listen: %v", lerr) - } - t.Fatal("adaptive engine never bound") - } + awaitBoundOrFail890(t, e, cancel, done, ended) return e, want, func() { cancel() <-done diff --git a/internal/adaptive/switch_idle_conns_linux_test.go b/internal/adaptive/switch_idle_conns_linux_test.go index 3d051ebfd..21984a20a 100644 --- a/internal/adaptive/switch_idle_conns_linux_test.go +++ b/internal/adaptive/switch_idle_conns_linux_test.go @@ -255,21 +255,17 @@ func startAdaptive662(t *testing.T, cfg resource.Config) (*Engine, int) { e.FreezeSwitching() ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() t.Cleanup(func() { cancel() select { - case <-done: + case <-ended: case <-time.After(15 * time.Second): t.Error("adaptive engine did not stop within 15s") } }) - for dl := time.Now().Add(5 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - t.Fatal("adaptive engine never bound") - } + awaitBoundOrFail890(t, e, cancel, done, ended) if got := e.ActiveEngine().Type(); got != engine.Epoll { t.Fatalf("these tests switch between the two sub-engines starting from epoll, started on %v", got) } diff --git a/internal/engine/epoll/adaptive_abort_844_linux_test.go b/internal/engine/epoll/adaptive_abort_844_linux_test.go index c02648554..e4e86326f 100644 --- a/internal/engine/epoll/adaptive_abort_844_linux_test.go +++ b/internal/engine/epoll/adaptive_abort_844_linux_test.go @@ -263,11 +263,12 @@ func start844(t *testing.T, h *a844, gate bool) *rig844 { t.Cleanup(cancel) r.done = make(chan error, 1) go func() { r.done <- ae.Listen(ctx) }() - for dl := time.Now().Add(5 * time.Second); ae.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if ae.Addr() == nil { - t.Fatal("adaptive engine never bound") + // A guard against a hang, not an expectation: Listen gives up on its own after + // its bind deadline (20 s) and returns its error, which awaitAdaptiveBound844 reports. + // The 5 s it replaces could be missed by a starved poll or a slow Listen + // alone (celeris#890 item 98, internal/adaptive's awaitBind890). + if err := awaitAdaptiveBound844(ae, r.done, 30*time.Second); err != nil { + t.Fatalf("adaptive engine never bound: %v", err) } r.addr = ae.Addr().String() if pkg844 == "iouring" { @@ -546,3 +547,23 @@ func TestAdaptiveAbortLogsAPanicValueThatSharesStreamMemory844(t *testing.T) { t.Errorf("celeris#844 item 3: the slog handler formatted the panic value after the conn was torn down and got %q", text) } } + +// awaitAdaptiveBound844 waits up to d for the adaptive engine ae to publish its +// address. If Listen returns first (done receives its result) it does not wait +// out d: it fails with that error, and puts the value back on done for the +// teardown. The same wait as internal/adaptive's awaitBind890 (this is another +// package's test, so it cannot import it). +func awaitAdaptiveBound844(ae interface{ Addr() net.Addr }, done chan error, d time.Duration) error { + for dl := time.Now().Add(d); ae.Addr() == nil; { + select { + case err := <-done: + done <- err + return fmt.Errorf("Listen returned before the engine bound: %v", err) + case <-time.After(10 * time.Millisecond): + } + if time.Now().After(dl) { + return fmt.Errorf("no address after %v", d) + } + } + return nil +} diff --git a/internal/engine/iouring/adaptive_abort_844_linux_test.go b/internal/engine/iouring/adaptive_abort_844_linux_test.go index 059a0dd24..69ee5071c 100644 --- a/internal/engine/iouring/adaptive_abort_844_linux_test.go +++ b/internal/engine/iouring/adaptive_abort_844_linux_test.go @@ -263,11 +263,11 @@ func start844(t *testing.T, h *a844, gate bool) *rig844 { t.Cleanup(cancel) r.done = make(chan error, 1) go func() { r.done <- ae.Listen(ctx) }() - for dl := time.Now().Add(5 * time.Second); ae.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if ae.Addr() == nil { - t.Fatal("adaptive engine never bound") + // 30 s is a guard against a hang (Listen gives up by itself after 20 s and + // awaitAdaptiveBound844 reports its error); the 5 s it replaces could be missed by a + // starved poll or a slow Listen alone (celeris#890 item 98). + if err := awaitAdaptiveBound844(ae, r.done, 30*time.Second); err != nil { + t.Fatalf("adaptive engine never bound: %v", err) } r.addr = ae.Addr().String() if pkg844 == "iouring" { @@ -546,3 +546,23 @@ func TestAdaptiveAbortLogsAPanicValueThatSharesStreamMemory844(t *testing.T) { t.Errorf("celeris#844 item 3: the slog handler formatted the panic value after the conn was torn down and got %q", text) } } + +// awaitAdaptiveBound844 waits up to d for the adaptive engine ae to publish its +// address. If Listen returns first (done receives its result) it does not wait +// out d: it fails with that error, and puts the value back on done for the +// teardown. The same wait as internal/adaptive's awaitBind890 (this is another +// package's test, so it cannot import it). +func awaitAdaptiveBound844(ae interface{ Addr() net.Addr }, done chan error, d time.Duration) error { + for dl := time.Now().Add(d); ae.Addr() == nil; { + select { + case err := <-done: + done <- err + return fmt.Errorf("Listen returned before the engine bound: %v", err) + case <-time.After(10 * time.Millisecond): + } + if time.Now().After(dl) { + return fmt.Errorf("no address after %v", d) + } + } + return nil +} diff --git a/internal/engine/iouring/adaptive_switch_848_linux_test.go b/internal/engine/iouring/adaptive_switch_848_linux_test.go index a05d97382..73ab6d524 100644 --- a/internal/engine/iouring/adaptive_switch_848_linux_test.go +++ b/internal/engine/iouring/adaptive_switch_848_linux_test.go @@ -88,11 +88,8 @@ func TestAdaptiveSwitchIsNotALoopExit848(t *testing.T) { done := make(chan error, 1) go func() { done <- ae.Listen(ctx) }() t.Cleanup(func() { cancel(); <-done }) - for dl := time.Now().Add(5 * time.Second); ae.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if ae.Addr() == nil { - t.Fatal("adaptive engine never bound") + if err := awaitAdaptiveBound844(ae, done, 30*time.Second); err != nil { + t.Fatalf("adaptive engine never bound: %v", err) } addr := ae.Addr().String() keep, err := net.DialTimeout("tcp", addr, 2*time.Second) // idle keep-alive across the switches