diff --git a/pool.go b/pool.go index e3b7798..03d7632 100644 --- a/pool.go +++ b/pool.go @@ -331,7 +331,8 @@ func (p *pool) subpoolWorker(task any) func() (output any, err error) { } func (p *pool) Go(task func()) error { - return p.submit(task, p.nonBlocking) + // Wrap the task so it is skipped if the pool context is canceled while it is queued + return p.submit(wrapTask[struct{}, func(error)](task, func(error) {}, p.ctx, p.panicRecovery), p.nonBlocking) } func (p *pool) Submit(task func()) Task { diff --git a/pool_test.go b/pool_test.go index 19ccecc..a36afdf 100644 --- a/pool_test.go +++ b/pool_test.go @@ -184,6 +184,43 @@ func TestPoolStopAndWaitShouldNotDeadlockWhenContextCanceledWithQueuedTasks(t *t assert.Equal(t, int64(0), pool.RunningWorkers()) } +func TestPoolGoShouldNotRunQueuedTasksWhenContextCanceled(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + pool := NewPool(1, WithContext(ctx)) + + firstTaskStarted := make(chan struct{}) + releaseFirstTask := make(chan struct{}) + + err := pool.Go(func() { + close(firstTaskStarted) + <-releaseFirstTask + }) + assert.Equal(t, nil, err) + + // This task stays queued while the only worker is blocked in the first task. + var queuedTaskExecuted atomic.Bool + err = pool.Go(func() { + queuedTaskExecuted.Store(true) + }) + assert.Equal(t, nil, err) + + <-firstTaskStarted + cancel() + close(releaseFirstTask) + + pool.StopAndWait() + + assert.Equal(t, false, queuedTaskExecuted.Load()) + + assert.Equal(t, uint64(2), pool.SubmittedTasks()) + assert.Equal(t, uint64(1), pool.CompletedTasks()) + assert.Equal(t, uint64(1), pool.CanceledTasks()) + assert.Equal(t, uint64(0), pool.WaitingTasks()) + assert.Equal(t, int64(0), pool.RunningWorkers()) +} + func TestPoolMetrics(t *testing.T) { pool := NewPool(100)