diff --git a/pool/static_pool/destroy_test.go b/pool/static_pool/destroy_test.go new file mode 100644 index 0000000..e2583d5 --- /dev/null +++ b/pool/static_pool/destroy_test.go @@ -0,0 +1,96 @@ +package static_pool + +import ( + "context" + "log/slog" + "os/exec" + "testing" + "time" + + "github.com/roadrunner-server/pool/v2/fsm" + "github.com/roadrunner-server/pool/v2/ipc/pipe" + "github.com/roadrunner-server/pool/v2/payload" + "github.com/roadrunner-server/pool/v2/pool" + "github.com/stretchr/testify/require" +) + +// shutdownSlack covers the one-second ticker in WorkerWatcher plus process teardown. +const shutdownSlack = 3 * time.Second + +// stuckWorkerPool returns a single-worker pool whose worker is parked in a request that never +// finishes, so shutdown can only end by hitting its timeout. +func stuckWorkerPool(t *testing.T, cfg *pool.Config) *Pool { + t.Helper() + + p, err := NewPool( + t.Context(), + func(_ []string) *exec.Cmd { return exec.Command("php", "../../tests/sleep.php") }, + pipe.NewPipeFactory(slog.Default()), + cfg, + slog.Default(), + ) + require.NoError(t, err) + + go func() { + _, _ = p.Exec(context.Background(), &payload.Payload{Body: []byte("hello")}, make(chan struct{})) + }() + + require.Eventually(t, func() bool { + workers := p.Workers() + return len(workers) == 1 && workers[0].State().Compare(fsm.StateWorking) + }, 10*time.Second, 50*time.Millisecond, "worker never started the request") + + return p +} + +func Test_Destroy_KillsStuckWorkerWithinBudget(t *testing.T) { + const budget = time.Second + + for _, tc := range []struct { + name string + destroyTimeout time.Duration + ctxTimeout time.Duration + }{ + {name: "no caller deadline", destroyTimeout: budget}, + {name: "caller deadline is longer", destroyTimeout: budget, ctxTimeout: time.Minute}, + {name: "caller deadline is shorter", destroyTimeout: time.Minute, ctxTimeout: budget}, + } { + t.Run(tc.name, func(t *testing.T) { + p := stuckWorkerPool(t, &pool.Config{ + NumWorkers: 1, + AllocateTimeout: time.Minute, + DestroyTimeout: tc.destroyTimeout, + }) + + ctx := context.Background() + if tc.ctxTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, tc.ctxTimeout) + defer cancel() + } + + start := time.Now() + p.Destroy(ctx) + + require.Less(t, time.Since(start), budget+shutdownSlack) + }) + } +} + +func Test_Reset_KillsStuckWorkerWithinBudget(t *testing.T) { + const budget = time.Second + + p := stuckWorkerPool(t, &pool.Config{ + NumWorkers: 1, + AllocateTimeout: time.Minute, + DestroyTimeout: time.Minute, + ResetTimeout: budget, + }) + t.Cleanup(func() { p.Destroy(context.Background()) }) + + start := time.Now() + require.NoError(t, p.Reset(context.Background())) + + // Reset re-allocates the worker it just killed, so the bound is looser than Destroy's. + require.Less(t, time.Since(start), budget+2*shutdownSlack) +} diff --git a/pool/static_pool/pool.go b/pool/static_pool/pool.go index b8606f8..e275c1f 100644 --- a/pool/static_pool/pool.go +++ b/pool/static_pool/pool.go @@ -162,13 +162,13 @@ func (sp *Pool) RemoveWorker(ctx context.Context) error { return sp.ww.RemoveWorker(ctx) } -// ensureDeadline bounds the context with the fallback timeout when the caller set no deadline. -func ensureDeadline(ctx context.Context, fallback time.Duration) (context.Context, context.CancelFunc) { - if _, ok := ctx.Deadline(); ok { +// ensureDeadline bounds the context by the tighter of its own deadline and the given timeout. +func ensureDeadline(ctx context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { + if deadline, ok := ctx.Deadline(); ok && time.Until(deadline) <= timeout { return ctx, func() {} } - return context.WithTimeout(ctx, fallback) + return context.WithTimeout(ctx, timeout) } // AddWorker adds one worker to the pool. With a dynamic allocator configured, workers above diff --git a/worker_watcher/worker_watcher.go b/worker_watcher/worker_watcher.go index 6ebf591..d4411e4 100644 --- a/worker_watcher/worker_watcher.go +++ b/worker_watcher/worker_watcher.go @@ -245,16 +245,25 @@ func (ww *WorkerWatcher) Release(w *worker.Process) { } } -// stopWatchedWorkers stops every tracked worker concurrently and clears the workers map. -// The caller must hold ww.mu. +// stopWatchedWorkers asks every tracked worker to shut down; Process.Stop waits out its own +// grace period before killing one that does not answer. The caller must hold ww.mu. func (ww *WorkerWatcher) stopWatchedWorkers() { + ww.disposeWatchedWorkers((*worker.Process).Stop) +} + +// killWatchedWorkers terminates every tracked worker at once, for when there is no time left +// to negotiate. The caller must hold ww.mu. +func (ww *WorkerWatcher) killWatchedWorkers() { + ww.disposeWatchedWorkers((*worker.Process).Kill) +} + +func (ww *WorkerWatcher) disposeWatchedWorkers(dispose func(*worker.Process) error) { wg := &sync.WaitGroup{} ww.workers.Range(func(key, value any) bool { w := value.(*worker.Process) wg.Go(func() { w.State().Transition(fsm.StateDestroyed) - // kill the worker - _ = w.Stop() + _ = dispose(w) // remove worker from the channel w.Callback() }) @@ -292,9 +301,8 @@ func (ww *WorkerWatcher) Reset(ctx context.Context) uint64 { return ww.numWorkers.Load() case <-ctx.Done(): - // kill workers ww.mu.Lock() - ww.stopWatchedWorkers() + ww.killWatchedWorkers() ww.container.ResetDone() ww.mu.Unlock() @@ -338,10 +346,9 @@ func (ww *WorkerWatcher) Destroy(ctx context.Context) { ww.mu.Unlock() return case <-ctx.Done(): - // kill workers ww.log.Debug("destroy: context canceled", "error", ctx.Err()) ww.mu.Lock() - ww.stopWatchedWorkers() + ww.killWatchedWorkers() ww.numWorkers.Store(0) ww.destroyed.Store(true) ww.mu.Unlock()