diff --git a/pool.go b/pool.go index 1ba2092..b957464 100644 --- a/pool.go +++ b/pool.go @@ -220,6 +220,9 @@ func (p *pool) SubmittedTasks() uint64 { } func (p *pool) WaitingTasks() uint64 { + p.mutex.Lock() + defer p.mutex.Unlock() + return p.tasks.Len() } diff --git a/pool_test.go b/pool_test.go index a185d6a..19ccecc 100644 --- a/pool_test.go +++ b/pool_test.go @@ -932,3 +932,29 @@ func TestTaskInvokesGoexitAndPoolRecovers(t *testing.T) { t.Fatal("StopAndWait() blocked indefinitely") } } + +func TestPoolWaitingTasksUnderLoad(t *testing.T) { + const queueSize = 64 + pool := NewPool(4, WithQueueSize(queueSize)) + defer pool.StopAndWait() + + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 100000; i++ { + pool.Go(func() {}) + } + }() + + for { + if waiting := pool.WaitingTasks(); waiting > queueSize { + <-done + t.Fatalf("waiting task count %d exceeds queue size %d", waiting, queueSize) + } + select { + case <-done: + return + default: + } + } +}