diff --git a/review_r62_pool_drain_test.go b/review_r62_pool_drain_test.go new file mode 100644 index 0000000..93f6cce --- /dev/null +++ b/review_r62_pool_drain_test.go @@ -0,0 +1,60 @@ +package traefikoidc + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +// TestGoroutinePoolShutdownDrainsQueuedTasks verifies that a graceful Shutdown +// runs every already-submitted task and lets Wait() return. Previously a +// worker that randomly chose the closed shutdownChan over a ready task exited, +// leaving buffered tasks silently dropped and their pendingTasks count never +// decremented, so Wait() blocked forever. +func TestGoroutinePoolShutdownDrainsQueuedTasks(t *testing.T) { + const iterations = 15 + for it := 0; it < iterations; it++ { + pool := NewGoroutinePool(1, NewLogger(DefaultLogLevel)) + + var ran int32 + gate := make(chan struct{}) + + if err := pool.Submit(func() { <-gate }); err != nil { // occupies the worker + t.Fatalf("iter %d: submit block failed: %v", it, err) + } + if err := pool.Submit(func() { atomic.AddInt32(&ran, 1) }); err != nil { + t.Fatalf("iter %d: submit counted task failed: %v", it, err) + } + + done := make(chan struct{}) + go func() { + _ = pool.Shutdown(context.Background()) + close(done) + }() + // Release the blocking task after shutdown is signaled. + go func() { + time.Sleep(5 * time.Millisecond) + close(gate) + }() + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatalf("iter %d: Shutdown did not return", it) + } + + if atomic.LoadInt32(&ran) != 1 { + t.Fatalf("iter %d: queued task %d was dropped on shutdown (want 1)", it, atomic.LoadInt32(&ran)) + } + + // Wait() must return now that every counted task completed. + waitCh := make(chan struct{}) + go func() { pool.Wait(); close(waitCh) }() + select { + case <-waitCh: + case <-time.After(1 * time.Second): + t.Fatalf("iter %d: Wait() blocked after shutdown (pending task never ran)", it) + } + } +} diff --git a/singleton_resources.go b/singleton_resources.go index 17de2e5..8d56337 100644 --- a/singleton_resources.go +++ b/singleton_resources.go @@ -415,10 +415,38 @@ func (p *GoroutinePool) worker(id int) { } } case <-p.shutdownChan: - if p.logger != nil { - p.logger.Debugf("Worker %d shutting down", id) + // Drain any tasks still queued before exiting. Without this, a + // worker that randomly selects the (now closed) shutdownChan over a + // ready task would exit, and once all workers have exited the + // remaining buffered tasks are silently dropped — and their + // pendingTasks count is never decremented, so a concurrent Wait() + // blocks forever. + for { + select { + case task := <-p.taskQueue: + if task != nil { + func() { + defer func() { + if r := recover(); r != nil { + if p.logger != nil { + p.logger.Errorf("Worker %d panic recovered: %v", id, r) + } + } + }() + task() + }() + + newCount := atomic.AddInt64(&p.pendingTasks, -1) + if newCount == 0 { + p.taskCond.L.Lock() + p.taskCond.Broadcast() + p.taskCond.L.Unlock() + } + } + default: + return + } } - return } } }