diff --git a/README.md b/README.md index e8f8ff0..88d7165 100644 --- a/README.md +++ b/README.md @@ -16,22 +16,24 @@ if err := pool.Submit(func() { doWork() }); err != nil { ``` The pool creates the minimum workers immediately and grows on demand up to the -maximum. Its task queue has a fixed capacity of `maxWorkers/2`; once all workers -and queue slots are occupied, `Submit` waits for room. `Close` stops new -submissions and waits for accepted tasks to finish. A successful `Submit` means -the task was accepted; it does not mean the task has started or finished. With -`maxWorkers == 1`, the queue has zero capacity and submissions wait for a worker. +maximum. A channel-driven scheduler owns worker lifecycle and a FIFO task queue +with fixed capacity `maxWorkers/2`; once all workers and queue slots are +occupied, `Submit` waits for room. With `maxWorkers == 1`, the queue has zero +capacity and submissions wait for a worker. A successful `Submit` means the +scheduler accepted the task; it does not mean the task has started or finished. +`Close` stops new submissions and waits for accepted tasks to finish. After a task finishes, a worker exits immediately only when `total > min` and the number of *other* idle workers is greater than `min/2`. There is no idle -timeout. A task panic is logged and that worker continues with later tasks. +timeout. Only `Close` shuts down the pool. A task panic is logged and the same worker -continues. A worker scheduling panic or `runtime.Goexit` terminates that -worker; the pool starts a replacement as needed, and accepted queued tasks -remain available even while `Close` is draining. The task that panicked or -called `runtime.Goexit` is not retried. Call `Close` from outside pool tasks so -it does not wait for itself. +continues with later tasks. A worker scheduling panic is logged and terminates +that worker; if accepted work is queued, a replacement worker takes the oldest +task. A panicking task is not retried. An exceptional worker exit with no +queued work does not proactively restore the minimum count; later submissions +can grow the pool again. During `Close`, queued accepted work is still drained. +Call `Close` from outside pool tasks so it does not wait for itself. A task may submit another task to the same pool. If all workers block submitting to a full queue, the caller must resolve that deadlock in its task logic. diff --git a/pool.go b/pool.go index 8ee2045..c6f83d2 100644 --- a/pool.go +++ b/pool.go @@ -4,7 +4,6 @@ package worker import ( "errors" "log" - "sync" ) var ( @@ -15,236 +14,433 @@ var ( // Pool runs submitted tasks using a bounded number of workers. type Pool struct { - mu sync.Mutex - submitCond *sync.Cond - closeCond *sync.Cond - min int - max int - total int - idle []*worker - queue []func() - queueHead int - queueTail int - queueLen int - queueCap int - closed bool + submit chan func() + closeReq chan struct{} + stopping chan struct{} + done chan struct{} + events chan workerEvent + min int + max int + queueCap int } type worker struct { inbox chan func() + stop chan struct{} + dead chan struct{} + first func() } -// New creates a pool with minWorkers persistent workers and at most -// maxWorkers workers. The task queue has capacity maxWorkers/2. +type eventKind uint8 + +const ( + taskFinished eventKind = iota + workerStopped + workerPanicked +) + +type workerEvent struct { + worker *worker + kind eventKind +} + +type workerState uint8 + +const ( + workerBusy workerState = iota + workerIdle + workerRetiring + workerDying +) + +// scheduler is the sole owner of worker counts, worker states, idle workers, +// queue contents, and the pool's closed state. Its fixed-capacity channels are +// queues, never locks. Worker goroutines report state changes on events. +type scheduler struct { + pool *Pool + idle chan *worker + queue chan func() + workers map[*worker]workerState + total int + retiring int + closed bool + done bool + // pending occupies the idle worker slot whose handoff failed. It is not + // an additional queue slot, so submissions pause until it is dispatched. + pending func() + hasPending bool +} + +// New starts minWorkers workers before returning. The task queue capacity is +// maxWorkers/2; no additional task buffering is performed by Submit. func New(minWorkers, maxWorkers int) (*Pool, error) { if maxWorkers <= 0 || minWorkers < 0 || minWorkers > maxWorkers { return nil, ErrInvalidConfig } p := &Pool{ + submit: make(chan func()), + closeReq: make(chan struct{}), + stopping: make(chan struct{}), + done: make(chan struct{}), + events: make(chan workerEvent, maxWorkers), min: minWorkers, max: maxWorkers, - queue: make([]func(), maxWorkers/2), queueCap: maxWorkers / 2, } - p.submitCond = sync.NewCond(&p.mu) - p.closeCond = sync.NewCond(&p.mu) + s := newScheduler(p) for range minWorkers { - p.startWorkerLocked(nil, false) + s.startWorker(nil, false) } + go s.run() return p, nil } -// Submit adds task to the pool. It blocks if all workers and the bounded -// queue are full. A successful return means the pool accepted the task. +// Submit waits until a worker or a bounded queue slot can accept task. A nil +// task is rejected before checking whether the pool has closed. func (p *Pool) Submit(task func()) error { if task == nil { return ErrNilTask } - p.mu.Lock() - defer p.mu.Unlock() - for { - if p.closed { - return ErrClosed + select { + case p.submit <- task: + return nil + case <-p.stopping: + return ErrClosed + } +} + +// Close stops accepting new tasks and waits for all accepted tasks and workers +// to finish. It is safe to call concurrently or more than once. Call it outside +// pool tasks to avoid waiting for the calling worker itself. +func (p *Pool) Close() { + select { + case <-p.done: + return + case p.closeReq <- struct{}{}: + case <-p.stopping: + } + <-p.done +} + +func newScheduler(p *Pool) *scheduler { + return &scheduler{ + pool: p, + idle: make(chan *worker, p.max), + queue: make(chan func(), p.queueCap), + workers: make(map[*worker]workerState, p.max), + } +} + +func (s *scheduler) startWorker(task func(), withTask bool) *worker { + w := &worker{inbox: make(chan func()), stop: make(chan struct{}), dead: make(chan struct{})} + s.workers[w] = workerBusy + s.total++ + if withTask { + w.first = task + } else { + s.setIdle(w) + } + go runWorker(w, s.pool.events) + return w +} + +func (s *scheduler) setIdle(w *worker) { + s.workers[w] = workerIdle + s.idle <- w +} + +func (s *scheduler) run() { + for !s.done { + s.pump() + if s.hasPending { + s.dispatchPending() } - if n := len(p.idle); n > 0 { - w := p.idle[n-1] - p.idle = p.idle[:n-1] - if trySend(w, task) { - return nil + s.finishCloseIfReady() + if s.done { + return + } + + var submit <-chan func() + if !s.closed && !s.hasPending && s.canAccept() { + submit = s.pool.submit + } + var closeReq <-chan struct{} + if !s.closed { + closeReq = s.pool.closeReq + } + select { + case task := <-submit: + if !s.accept(task) { + // The idle worker's reserved handoff is now represented by + // pending until its terminal event frees that worker slot. + s.pending = task + s.hasPending = true } - // A worker can have published itself idle immediately before its - // closed inbox is observed. Keep task and let the normal state - // machine assign it again after that worker is accounted for. + case event := <-s.pool.events: + s.handleEvent(event) + case <-closeReq: + s.beginClose() + } + } +} + +func (s *scheduler) dispatchPending() { + if !s.hasPending { + return + } + task := s.pending + if len(s.idle) > 0 { + w := <-s.idle + if !s.deliver(w, task) { + s.markWorkerDying(w) + return + } + s.workers[w] = workerBusy + s.pending = nil + s.hasPending = false + return + } + if s.total < s.pool.max { + s.startWorker(task, true) + s.pending = nil + s.hasPending = false + } +} + +func (s *scheduler) canAccept() bool { + if len(s.queue) > 0 { + return len(s.queue) < s.pool.queueCap + } + return len(s.idle) > 0 || s.total < s.pool.max || len(s.queue) < s.pool.queueCap +} + +func (s *scheduler) accept(task func()) bool { + if len(s.queue) > 0 { + s.queue <- task + return true + } + if len(s.idle) > 0 { + w := <-s.idle + if !s.deliver(w, task) { + s.markWorkerDying(w) + return false + } + s.workers[w] = workerBusy + return true + } + if s.total < s.pool.max { + s.startWorker(task, true) + return true + } + s.queue <- task + return true +} + +// pump gives accepted FIFO work to idle workers or newly available worker +// slots before accepting more submissions. It never keeps a hidden task slot. +func (s *scheduler) pump() { + for len(s.queue) > 0 { + if len(s.idle) > 0 { + w := <-s.idle + task := <-s.queue + if !s.deliver(w, task) { + s.requeueFront(task) + s.markWorkerDying(w) + continue + } + s.workers[w] = workerBusy continue } - if p.total < p.max { - p.startWorkerLocked(task, true) - return nil + if s.total < s.pool.max { + s.startWorker(<-s.queue, true) + continue } - if p.queueLen < p.queueCap { - p.enqueueLocked(task) - return nil + break + } + if s.closed && !s.hasPending { + for len(s.idle) > 0 { + s.retire(<-s.idle) } - p.submitCond.Wait() } } -// Close stops accepting tasks and waits for workers to exit. Tasks accepted -// before Close finish before it returns, even if workers exit abnormally and -// need replacement. It is safe to call Close concurrently or more than once. -func (p *Pool) Close() { - p.mu.Lock() - defer p.mu.Unlock() - p.closed = true - for _, w := range p.idle { - safeClose(w) +func (s *scheduler) beginClose() { + if s.closed { + return } - p.idle = nil - p.submitCond.Broadcast() - for p.total != 0 { - p.closeCond.Wait() + s.closed = true + close(s.pool.stopping) + for len(s.idle) > 0 { + s.retire(<-s.idle) } } -// All pool state (worker count, idle set, queue indexes and contents, and -// closed state) is guarded by mu. submitCond is only for Submit capacity -// changes; closeCond is only for Close waiting on all workers to exit. Each -// worker has a one-slot inbox representing an idle worker's reserved handoff. -func (p *Pool) startWorkerLocked(task func(), withTask bool) { - w := &worker{inbox: make(chan func(), 1)} - p.total++ - if withTask { - w.inbox <- task - } else { - p.idle = append(p.idle, w) +func (s *scheduler) handleEvent(event workerEvent) { + switch event.kind { + case taskFinished: + s.taskFinished(event.worker) + case workerStopped: + s.workerExited(event.worker, false) + case workerPanicked: + s.workerExited(event.worker, true) } - go p.run(w) + // A worker exit can make room for accepted queue work. A normal task finish + // may also expose a queue slot to a blocked Submit. + s.pump() + s.finishCloseIfReady() } -func trySend(w *worker, task func()) (sent bool) { +func (s *scheduler) taskFinished(w *worker) { + state, exists := s.workers[w] + if !exists || state != workerBusy { + return + } + if len(s.queue) > 0 { + task := <-s.queue + if !s.deliver(w, task) { + s.requeueFront(task) + s.markWorkerDying(w) + } + return + } + select { + case <-w.dead: + s.markWorkerDying(w) + return + default: + } + if s.closed { + s.retire(w) + return + } + otherIdle := len(s.idle) + if s.total-s.retiring > s.pool.min && otherIdle > s.pool.min/2 { + s.retire(w) + return + } + s.setIdle(w) +} + +// markWorkerDying removes a worker from dispatch immediately, but leaves its +// count reserved until the unique terminal event arrives after its logger and +// exit reporting have completed. +func (s *scheduler) markWorkerDying(w *worker) { + state, exists := s.workers[w] + if !exists || state == workerDying { + return + } + if state == workerIdle { + s.removeIdle(w) + } + s.workers[w] = workerDying +} + +func (s *scheduler) retire(w *worker) { + if s.workers[w] == workerRetiring { + return + } + s.workers[w] = workerRetiring + s.retiring++ + close(w.stop) +} + +func (s *scheduler) workerExited(w *worker, abnormal bool) { + state, exists := s.workers[w] + if !exists { + return + } + if state == workerIdle { + s.removeIdle(w) + } + delete(s.workers, w) + s.total-- + if state == workerRetiring { + s.retiring-- + } + if abnormal && len(s.queue) > 0 { + s.startWorker(<-s.queue, true) + } +} + +// deliver uses an unbuffered inbox: a successful send means a live worker has +// taken ownership of the task. If the worker is dead, no task can be left +// buffered behind it. The recover handles a test-injected closed inbox; callers +// preserve the task and account for the worker before continuing. +func (s *scheduler) deliver(w *worker, task func()) (sent bool) { defer func() { if recover() != nil { sent = false } }() - w.inbox <- task - return true -} - -func safeClose(w *worker) { - defer func() { _ = recover() }() - close(w.inbox) -} - -func (p *Pool) run(w *worker) { - exitAccounted := false - defer func() { - if recovered := recover(); recovered != nil { - safeLogf("worker panic: %v", recovered) - } - if !exitAccounted { - p.accountWorkerExit(w, &exitAccounted) - } - }() - for { - task, ok := <-w.inbox - if !ok { - p.accountWorkerExit(w, &exitAccounted) - return - } - runTask(task) - if p.finishTask(w, &exitAccounted) { - return - } - } -} - -// finishTask commits the post-task transition under mu. If queue work exists, -// it first hands off the head item and only then removes it from the ring, so -// a closed inbox panic cannot lose an accepted task. The deferred unlock runs -// before run's panic cleanup acquires the mutex. -func (p *Pool) finishTask(w *worker, exitAccounted *bool) bool { - p.mu.Lock() - defer p.mu.Unlock() - if p.queueLen > 0 { - next := p.queue[p.queueHead] - w.inbox <- next - p.dequeueLocked() - p.submitCond.Signal() + select { + case w.inbox <- task: + return true + case <-w.dead: return false } - otherIdle := len(p.idle) - if p.closed || (p.total > p.min && otherIdle > p.min/2) { - p.total-- - *exitAccounted = true - if !p.closed { - p.submitCond.Signal() - } else { - p.workerExitedLocked() - } - return true - } - p.idle = append(p.idle, w) - p.submitCond.Signal() - return false } -// enqueueLocked and dequeueLocked are called with mu held. The queue is a -// fixed-capacity FIFO ring; clearing the dequeued slot releases its closure. -func (p *Pool) enqueueLocked(task func()) { - p.queue[p.queueTail] = task - p.queueTail++ - if p.queueTail == p.queueCap { - p.queueTail = 0 +// requeueFront restores a task removed for a failed unbuffered handoff while +// preserving FIFO order and the original fixed cache capacity. This only runs +// on worker failure, never on the normal dispatch path. +func (s *scheduler) requeueFront(task func()) { + queue := make(chan func(), s.pool.queueCap) + queue <- task + for len(s.queue) > 0 { + queue <- <-s.queue } - p.queueLen++ + s.queue = queue } -func (p *Pool) dequeueLocked() func() { - task := p.queue[p.queueHead] - p.queue[p.queueHead] = nil - p.queueHead++ - if p.queueHead == p.queueCap { - p.queueHead = 0 - } - p.queueLen-- - return task -} - -func (p *Pool) workerExitedLocked() { - if p.closed && p.total == 0 { - p.closeCond.Broadcast() - } -} - -// accountWorkerExit decrements a worker and, while still holding mu, starts a -// replacement for the oldest accepted task or restores the minimum worker count. -// This applies to panic, Goexit, and a closed inbox. It never retries the task -// that caused the worker to exit and never discards queued tasks. -func (p *Pool) accountWorkerExit(w *worker, exitAccounted *bool) { - p.mu.Lock() - defer p.mu.Unlock() - for i, idleWorker := range p.idle { - if idleWorker == w { - copy(p.idle[i:], p.idle[i+1:]) - last := len(p.idle) - 1 - p.idle[last] = nil - p.idle = p.idle[:last] - break +func (s *scheduler) removeIdle(target *worker) { + count := len(s.idle) + for range count { + w := <-s.idle + if w != target { + s.idle <- w } } - p.total-- - *exitAccounted = true - if p.queueLen > 0 { - task := p.dequeueLocked() - p.startWorkerLocked(task, true) - } else { - if !p.closed && p.total < p.min { - p.startWorkerLocked(nil, false) +} + +func (s *scheduler) finishCloseIfReady() { + if s.closed && s.total == 0 && len(s.queue) == 0 && !s.hasPending && !s.done { + s.done = true + close(s.pool.done) + } +} + +func runWorker(w *worker, events chan<- workerEvent) { + exitKind := workerStopped + defer func() { + recovered := recover() + close(w.dead) + if recovered != nil { + exitKind = workerPanicked + safeLogf("worker panic: %v", recovered) + } + // This is the unique terminal notification and the worker's final + // potentially blocking operation. The scheduler never accounts it + // from the earlier dead signal. + events <- workerEvent{worker: w, kind: exitKind} + }() + if w.first != nil { + task := w.first + w.first = nil + runTask(task) + events <- workerEvent{worker: w, kind: taskFinished} + } + for { + select { + case <-w.stop: + return + case task, ok := <-w.inbox: + if !ok { + panic("worker inbox closed") + } + runTask(task) + events <- workerEvent{worker: w, kind: taskFinished} } } - p.submitCond.Signal() - p.workerExitedLocked() } func safeLogf(format string, args ...any) { diff --git a/pool_bench_test.go b/pool_bench_test.go index 4124dd7..6b291c8 100644 --- a/pool_bench_test.go +++ b/pool_bench_test.go @@ -1,19 +1,44 @@ package worker -import "testing" +import ( + "runtime" + "sync/atomic" + "testing" + "time" +) -func BenchmarkQueueRingSteady(b *testing.B) { - p, err := New(0, 6) +func BenchmarkSubmitNoop(b *testing.B) { + benchmarkSubmit(b, func() {}) +} + +func BenchmarkSubmit10usTask(b *testing.B) { + benchmarkSubmit(b, func() { + deadline := time.Now().Add(10 * time.Microsecond) + for time.Now().Before(deadline) { + } + }) +} + +func benchmarkSubmit(b *testing.B, task func()) { + p, err := New(6, 6) if err != nil { b.Fatal(err) } - task := func() {} + var completed atomic.Int64 + wrapped := func() { + task() + completed.Add(1) + } b.ReportAllocs() b.ResetTimer() - for i := 0; i < b.N; i++ { - p.mu.Lock() - p.enqueueLocked(task) - p.dequeueLocked() - p.mu.Unlock() + for range b.N { + if err := p.Submit(wrapped); err != nil { + b.Fatal(err) + } } + for completed.Load() != int64(b.N) { + runtime.Gosched() + } + b.StopTimer() + p.Close() } diff --git a/pool_test.go b/pool_test.go index 09e65b9..5004795 100644 --- a/pool_test.go +++ b/pool_test.go @@ -4,7 +4,6 @@ import ( "bytes" "errors" "log" - "runtime" "strings" "sync" "sync/atomic" @@ -12,10 +11,10 @@ import ( "time" ) -func TestNewAndNilTask(t *testing.T) { - for _, config := range [][2]int{{-1, 1}, {0, 0}, {2, 1}} { - if _, err := New(config[0], config[1]); !errors.Is(err, ErrInvalidConfig) { - t.Fatalf("New(%d, %d) error = %v", config[0], config[1], err) +func TestNewValidationAndNilTask(t *testing.T) { + for _, cfg := range [][2]int{{-1, 1}, {0, 0}, {2, 1}} { + if _, err := New(cfg[0], cfg[1]); !errors.Is(err, ErrInvalidConfig) { + t.Fatalf("New(%d,%d) error=%v, want ErrInvalidConfig", cfg[0], cfg[1], err) } } p, err := New(0, 2) @@ -23,870 +22,173 @@ func TestNewAndNilTask(t *testing.T) { t.Fatal(err) } if err := p.Submit(nil); !errors.Is(err, ErrNilTask) { - t.Fatalf("Submit(nil) error = %v", err) + t.Fatalf("Submit(nil) error=%v, want ErrNilTask", err) } p.Close() + if err := p.Submit(nil); !errors.Is(err, ErrNilTask) { + t.Fatalf("Submit(nil) after Close error=%v, want nil-task priority", err) + } } -func TestMinimumWorkersStartImmediately(t *testing.T) { - p, err := New(3, 5) +func TestSchedulerStartsMinimumBeforeRun(t *testing.T) { + p := newTestPool(3, 5) + s := newScheduler(p) + for range p.min { + s.startWorker(nil, false) + } + if s.total != 3 || len(s.idle) != 3 { + t.Fatalf("initial scheduler total/idle=%d/%d, want 3/3", s.total, len(s.idle)) + } + // The actors' minimum workers are real goroutines; stop this fixture cleanly. + s.closed = true + close(p.stopping) + s.pump() + for s.total > 0 { + event := waitEvent(t, p.events) + s.handleEvent(event) + } +} + +func TestConcurrentSubmitRunsExactlyOnceAndNeverExceedsMaximum(t *testing.T) { + const count, maxWorkers = 120, 4 + p, err := New(1, maxWorkers) if err != nil { t.Fatal(err) } - p.mu.Lock() - if p.total != 3 || len(p.idle) != 3 { - t.Fatalf("initial workers/idle = %d/%d, want 3/3", p.total, len(p.idle)) - } - p.mu.Unlock() - p.Close() -} - -func TestConcurrentSubmitRunsEveryTaskOnceAndRespectsMaximum(t *testing.T) { - const taskCount = 160 - p, err := New(1, 4) - if err != nil { - t.Fatal(err) - } - var running, maxRunning atomic.Int32 - var calls [taskCount]atomic.Int32 + var calls [count]atomic.Int32 + var active, maxActive atomic.Int32 var submitters sync.WaitGroup start := make(chan struct{}) - for i := range taskCount { + results := make([]error, count) + for i := range count { submitters.Add(1) go func(i int) { defer submitters.Done() - err := p.Submit(func() { - current := running.Add(1) - for old := maxRunning.Load(); current > old && !maxRunning.CompareAndSwap(old, current); old = maxRunning.Load() { + <-start + results[i] = p.Submit(func() { + n := active.Add(1) + for old := maxActive.Load(); n > old && !maxActive.CompareAndSwap(old, n); old = maxActive.Load() { } calls[i].Add(1) - <-start - running.Add(-1) + active.Add(-1) }) - if err != nil { - t.Errorf("Submit(%d): %v", i, err) - } }(i) } - doneSubmit := make(chan struct{}) - go func() { submitters.Wait(); close(doneSubmit) }() close(start) - select { - case <-doneSubmit: - case <-time.After(5 * time.Second): - t.Fatal("concurrent submitters did not finish") - } + waitGroup(t, &submitters) p.Close() - if got := maxRunning.Load(); got > 4 { - t.Fatalf("maximum concurrent tasks = %d, want <= 4", got) - } - for i := range taskCount { - if got := calls[i].Load(); got != 1 { - t.Fatalf("task %d ran %d times, want once", i, got) + for i, result := range results { + if result != nil { + t.Fatalf("Submit(%d) error=%v, want accepted", i, result) } + if got := calls[i].Load(); got != 1 { + t.Fatalf("task %d accepted but ran %d times", i, got) + } + } + if n := maxActive.Load(); n > maxWorkers { + t.Fatalf("maximum concurrent tasks=%d, worker limit=%d", n, maxWorkers) } } -func TestQueueCapacityBlocksAndUnblocksSubmit(t *testing.T) { - p, err := New(1, 2) // queue capacity is one +func TestGrowsToMaximumBeforeUsingBoundedQueue(t *testing.T) { + p, err := New(0, 3) // one cached task if err != nil { t.Fatal(err) } - firstGate, secondGate := make(chan struct{}), make(chan struct{}) - started := make(chan struct{}, 2) - if err := p.Submit(func() { started <- struct{}{}; <-firstGate }); err != nil { - t.Fatal(err) + gate := make(chan struct{}) + started := make(chan struct{}, 3) + for range 3 { + if err := p.Submit(func() { started <- struct{}{}; <-gate }); err != nil { + t.Fatal(err) + } } - if err := p.Submit(func() { started <- struct{}{}; <-secondGate }); err != nil { - t.Fatal(err) + for range 3 { + waitSignal(t, started) } - waitSignal(t, started) - waitSignal(t, started) if err := p.Submit(func() {}); err != nil { t.Fatal(err) } - accepted := make(chan error, 1) - go func() { accepted <- p.Submit(func() {}) }() - select { - case err := <-accepted: - t.Fatalf("Submit returned before queue space opened: %v", err) - case <-time.After(30 * time.Millisecond): - } - close(firstGate) - select { - case err := <-accepted: - if err != nil { - t.Fatal(err) - } - case <-time.After(2 * time.Second): - t.Fatal("blocked Submit did not resume") - } - close(secondGate) - p.Close() -} - -func TestWorkersGrowOnlyToMaximum(t *testing.T) { - p, err := New(1, 3) - if err != nil { - t.Fatal(err) - } - gate := make(chan struct{}) - started := make(chan struct{}, 3) - for range 3 { - if err := p.Submit(func() { started <- struct{}{}; <-gate }); err != nil { - t.Fatal(err) - } - } - for range 3 { - waitSignal(t, started) - } - p.mu.Lock() - if p.total != 3 { - p.mu.Unlock() - t.Fatalf("workers at maximum load = %d, want 3", p.total) - } - p.mu.Unlock() - if err := p.Submit(func() {}); err != nil { // fills the max/2 queue - t.Fatal(err) - } blocked := make(chan error, 1) - go func() { blocked <- p.Submit(func() {}) }() - select { - case err := <-blocked: - t.Fatalf("Submit beyond maximum workers and queue returned early: %v", err) - case <-time.After(30 * time.Millisecond): - } - close(gate) - select { - case err := <-blocked: - if err != nil { - t.Fatal(err) - } - case <-time.After(2 * time.Second): - t.Fatal("Submit did not resume after capacity became available") - } - p.Close() -} - -func TestQueueRingWrapsInFIFOOrderAndClearsSlots(t *testing.T) { - p, err := New(0, 4) - if err != nil { - t.Fatal(err) - } - var order []int - f1 := func() { order = append(order, 1) } - f2 := func() { order = append(order, 2) } - f3 := func() { order = append(order, 3) } - p.mu.Lock() - p.enqueueLocked(f1) - p.enqueueLocked(f2) - if p.queueLen != p.queueCap { - p.mu.Unlock() - t.Fatalf("queue length = %d, capacity = %d", p.queueLen, p.queueCap) - } - first := p.dequeueLocked() - p.enqueueLocked(f3) // tail wraps to the beginning of the fixed ring - if p.queueHead != 1 || p.queueTail != 1 || p.queueLen != 2 { - p.mu.Unlock() - t.Fatalf("ring head/tail/len = %d/%d/%d, want 1/1/2", p.queueHead, p.queueTail, p.queueLen) - } - second, third := p.dequeueLocked(), p.dequeueLocked() - if p.queueLen != 0 || p.queue[0] != nil || p.queue[1] != nil { - p.mu.Unlock() - t.Fatal("dequeued task references remain in ring slots") - } - p.mu.Unlock() - first() - second() - third() - if len(order) != 3 || order[0] != 1 || order[1] != 2 || order[2] != 3 { - t.Fatalf("dequeue order = %v, want [1 2 3]", order) - } - p.Close() -} - -func TestSubmitPreservesFIFOAcrossQueueRingWraps(t *testing.T) { - const taskCount = 12 - p, err := New(4, 4) - if err != nil { - t.Fatal(err) - } - otherGate := make(chan struct{}) - otherStarted := make(chan struct{}, 3) - for range 3 { - if err := p.Submit(func() { otherStarted <- struct{}{}; <-otherGate }); err != nil { - t.Fatal(err) - } - } - for range 3 { - waitSignal(t, otherStarted) - } - consumerGate := make(chan struct{}) - consumerStarted := make(chan struct{}) - if err := p.Submit(func() { close(consumerStarted); <-consumerGate }); err != nil { - t.Fatal(err) - } - waitSignal(t, consumerStarted) - - startTask := make(chan int, taskCount) - taskGates := make([]chan struct{}, taskCount) - order := make([]int, 0, taskCount) - newTask := func(i int) func() { - taskGates[i] = make(chan struct{}) - return func() { - order = append(order, i) - startTask <- i - <-taskGates[i] - } - } - if err := p.Submit(newTask(0)); err != nil { - t.Fatal(err) - } - if err := p.Submit(newTask(1)); err != nil { - t.Fatal(err) - } - - submitted := make(chan int, taskCount-2) - submitterDone := make(chan struct{}) - go func() { - defer close(submitterDone) - for i := 2; i < taskCount; i++ { - if err := p.Submit(newTask(i)); err != nil { - submitted <- -1 - return - } - submitted <- i - } - }() - close(consumerGate) - - for i := 0; i < taskCount; i++ { - if got := waitSignal(t, startTask); got != i { - t.Fatalf("task execution order[%d] = %d, want %d", i, got, i) - } - if i+2 < taskCount { - if got := waitSignal(t, submitted); got != i+2 { - t.Fatalf("submitted task after queue slot release = %d, want %d", got, i+2) - } - } - close(taskGates[i]) - } - waitSignal(t, submitterDone) - close(otherGate) - p.Close() - if len(order) != taskCount { - t.Fatalf("executed %d tasks, want %d", len(order), taskCount) - } - for i, got := range order { - if got != i { - t.Fatalf("execution order = %v, want FIFO [0..%d]", order, taskCount-1) - } - } -} - -func TestZeroCapacityQueueDoesNotIndexOrWrap(t *testing.T) { - p, err := New(0, 1) - if err != nil { - t.Fatal(err) - } - if p.queueCap != 0 || len(p.queue) != 0 || p.queueLen != 0 { - t.Fatalf("zero-capacity queue state = cap:%d backing:%d len:%d", p.queueCap, len(p.queue), p.queueLen) - } - p.Close() -} - -func TestMultipleBlockedSubmittersWakeOnEachCapacityChange(t *testing.T) { - p, err := New(2, 2) - if err != nil { - t.Fatal(err) - } - gateA, gateB, gateC, gateD, gateE := make(chan struct{}), make(chan struct{}), make(chan struct{}), make(chan struct{}), make(chan struct{}) - started := make(chan struct{}, 5) - for _, gate := range []chan struct{}{gateA, gateB} { - if err := p.Submit(func() { started <- struct{}{}; <-gate }); err != nil { - t.Fatal(err) - } - } - waitSignal(t, started) - waitSignal(t, started) - if err := p.Submit(func() { started <- struct{}{}; <-gateC }); err != nil { // fill queue - t.Fatal(err) - } - results := make(chan error, 2) - attempted := make(chan struct{}, 2) - for _, gate := range []chan struct{}{gateD, gateE} { - go func(gate chan struct{}) { - attempted <- struct{}{} - results <- p.Submit(func() { started <- struct{}{}; <-gate }) - }(gate) - } - waitSignal(t, attempted) - waitSignal(t, attempted) - select { - case err := <-results: - t.Fatalf("Submit returned while the workers and queue were full: %v", err) - case <-time.After(30 * time.Millisecond): - } - // Release each busy worker separately. Each dequeue frees one slot and - // wakes one blocked submitter; the admitted task then occupies that slot. - close(gateA) - if err := waitSignal(t, results); err != nil { - t.Fatal(err) - } - waitSignal(t, started) // queue task C began - close(gateB) - if err := waitSignal(t, results); err != nil { - t.Fatal(err) - } - waitSignal(t, started) // admitted task D began - close(gateC) - waitSignal(t, started) // admitted task E began - close(gateD) - close(gateE) - p.Close() -} - -func TestUnbufferedMaxOneMinZero(t *testing.T) { - p, err := New(0, 1) - if err != nil { - t.Fatal(err) - } - gate := make(chan struct{}) - started := make(chan struct{}, 2) - if err := p.Submit(func() { started <- struct{}{}; <-gate }); err != nil { - t.Fatal(err) - } - waitSignal(t, started) - second := make(chan error, 1) - go func() { second <- p.Submit(func() { started <- struct{}{} }) }() - select { - case err := <-second: - t.Fatalf("unbuffered Submit returned while busy: %v", err) - case <-time.After(30 * time.Millisecond): - } - close(gate) - select { - case err := <-second: - if err != nil { - t.Fatal(err) - } - case <-time.After(2 * time.Second): - t.Fatal("Submit did not resume") - } - waitSignal(t, started) - p.Close() -} - -func TestShrinkUsesStrictIdleThresholdAndKeepsMinimum(t *testing.T) { - p, err := New(2, 4) - if err != nil { - t.Fatal(err) - } - gates := make([]chan struct{}, 4) - started := make(chan struct{}, 4) - for i := range gates { - gates[i] = make(chan struct{}) - } - for _, gate := range gates { - if err := p.Submit(func() { started <- struct{}{}; <-gate }); err != nil { - t.Fatal(err) - } - } - for range 4 { - waitSignal(t, started) - } - close(gates[0]) - waitPoolState(t, p, 4, 1) - close(gates[1]) - waitPoolState(t, p, 4, 2) // other idle == min/2, so no exit - p.mu.Lock() - if p.total != 4 { - t.Fatalf("workers at strict threshold = %d, want 4", p.total) - } - p.mu.Unlock() - close(gates[2]) - waitPoolState(t, p, 3, 2) - close(gates[3]) - waitPoolState(t, p, 2, 2) - p.mu.Lock() - if p.total < p.min { - t.Fatalf("workers %d fell below min %d", p.total, p.min) - } - p.mu.Unlock() - p.Close() -} - -func TestPanicIsLoggedAndWorkerContinues(t *testing.T) { - p, err := New(1, 1) - if err != nil { - t.Fatal(err) - } - var output bytes.Buffer - previousOutput := log.Writer() - log.SetOutput(&output) - defer log.SetOutput(previousOutput) - p.mu.Lock() - workerBefore := p.idle[0] - p.mu.Unlock() - completed := make(chan struct{}) - if err := p.Submit(func() { panic("expected test panic") }); err != nil { - log.SetOutput(previousOutput) - t.Fatal(err) - } - if err := p.Submit(func() { panic(nil) }); err != nil { - log.SetOutput(previousOutput) - t.Fatal(err) - } - if err := p.Submit(func() { - defer func() { panic("expected deferred panic") }() - }); err != nil { - log.SetOutput(previousOutput) - t.Fatal(err) - } - if err := p.Submit(func() { close(completed) }); err != nil { - log.SetOutput(previousOutput) - t.Fatal(err) - } - waitSignal(t, completed) - waitPoolState(t, p, 1, 1) - p.mu.Lock() - if p.total != 1 || len(p.idle) != 1 || p.idle[0] != workerBefore { - p.mu.Unlock() - t.Fatal("task panic replaced or lost the original worker") - } - p.mu.Unlock() - p.Close() - log.SetOutput(previousOutput) - got := output.String() - for _, marker := range []string{"expected test panic", "expected deferred panic"} { - if !strings.Contains(got, marker) { - t.Errorf("panic log %q does not contain %q", got, marker) - } - } -} - -type panicLogWriter struct { - panicAlways bool - panicked atomic.Bool - calls atomic.Int32 -} - -func (w *panicLogWriter) Write(data []byte) (int, error) { - w.calls.Add(1) - if w.panicAlways || w.panicked.CompareAndSwap(false, true) { - panic("test logger writer panic") - } - return len(data), nil -} - -func TestLoggerWriterPanicDoesNotKillTaskWorker(t *testing.T) { - for _, panicAlways := range []bool{false, true} { - name := "once" - if panicAlways { - name = "always" - } - t.Run(name, func(t *testing.T) { - p, err := New(1, 1) - if err != nil { - t.Fatal(err) - } - previousOutput := log.Writer() - writer := &panicLogWriter{panicAlways: panicAlways} - log.SetOutput(writer) - defer log.SetOutput(previousOutput) - p.mu.Lock() - workerBefore := p.idle[0] - p.mu.Unlock() - done := make(chan struct{}) - for _, task := range []func(){ - func() { panic("first task panic") }, - func() { panic("second task panic") }, - func() { close(done) }, - } { - if err := p.Submit(task); err != nil { - t.Fatal(err) - } - } - waitSignal(t, done) - waitPoolState(t, p, 1, 1) - p.mu.Lock() - if p.idle[0] != workerBefore { - p.mu.Unlock() - t.Fatal("task panic replaced the original worker") - } - p.mu.Unlock() - p.Close() - if writer.calls.Load() == 0 { - t.Fatal("test logger writer was not called") - } - }) - } -} - -func TestWorkerPanicReplacesWorkerAndPreservesQueue(t *testing.T) { - p, err := New(2, 2) - if err != nil { - t.Fatal(err) - } - var output bytes.Buffer - previousOutput := log.Writer() - log.SetOutput(&output) - defer log.SetOutput(previousOutput) - - p.mu.Lock() - victim := p.idle[len(p.idle)-1] - p.mu.Unlock() - victimGate, victimStarted := make(chan struct{}), make(chan struct{}) - if err := p.Submit(func() { close(victimStarted); <-victimGate }); err != nil { - t.Fatal(err) - } - waitSignal(t, victimStarted) - p.mu.Lock() - survivor := p.idle[len(p.idle)-1] - p.mu.Unlock() - survivorGate, survivorStarted := make(chan struct{}), make(chan struct{}) - if err := p.Submit(func() { close(survivorStarted); <-survivorGate }); err != nil { - t.Fatal(err) - } - waitSignal(t, survivorStarted) - queuedDone := make(chan struct{}) - if err := p.Submit(func() { close(queuedDone) }); err != nil { - t.Fatal(err) - } - close(victim.inbox) // fault injection while the selected worker is gated - close(victimGate) - waitSignal(t, queuedDone) - waitPoolState(t, p, 2, 1) - if !strings.Contains(output.String(), "worker panic: send on closed channel") { - t.Fatalf("worker panic was not logged: %q", output.String()) - } - p.mu.Lock() - replacement := p.idle[0] - if replacement == victim || replacement == survivor { - p.mu.Unlock() - t.Fatal("worker panic did not create a fresh replacement") - } - p.mu.Unlock() - close(survivorGate) - waitPoolState(t, p, 2, 2) - p.Close() - if output.String() == "" { - t.Fatal("worker panic log was lost") - } -} - -func TestUnexpectedClosedIdleInboxRestoresMinimum(t *testing.T) { - p, err := New(1, 1) - if err != nil { - t.Fatal(err) - } - p.mu.Lock() - dead := p.idle[0] - p.mu.Unlock() - close(dead.inbox) - deadline := time.Now().Add(2 * time.Second) - for time.Now().Before(deadline) { - p.mu.Lock() - replaced := p.total == 1 && len(p.idle) == 1 && p.idle[0] != dead - p.mu.Unlock() - if replaced { - break - } - time.Sleep(time.Millisecond) - } - p.mu.Lock() - replaced := p.total == 1 && len(p.idle) == 1 && p.idle[0] != dead - p.mu.Unlock() - if !replaced { - t.Fatal("pool retained an idle worker with a closed inbox") - } - done := make(chan struct{}) - if err := p.Submit(func() { close(done) }); err != nil { - t.Fatal(err) - } - waitSignal(t, done) - waitPoolState(t, p, 1, 1) - p.Close() -} - -func TestAllOriginalWorkersExitPoolRemainsOpenAndDrainsQueue(t *testing.T) { - p, err := New(2, 2) - if err != nil { - t.Fatal(err) - } - p.mu.Lock() - workerA := p.idle[len(p.idle)-1] - p.mu.Unlock() - gateA, gateB := make(chan struct{}), make(chan struct{}) - started := make(chan struct{}, 2) - if err := p.Submit(func() { started <- struct{}{}; <-gateA }); err != nil { - t.Fatal(err) - } - p.mu.Lock() - workerB := p.idle[len(p.idle)-1] - p.mu.Unlock() - if err := p.Submit(func() { started <- struct{}{}; <-gateB }); err != nil { - t.Fatal(err) - } - waitSignal(t, started) - waitSignal(t, started) - var queuedRuns atomic.Int32 - queuedDone := make(chan struct{}) - if err := p.Submit(func() { queuedRuns.Add(1); close(queuedDone) }); err != nil { - t.Fatal(err) - } - close(workerA.inbox) - close(workerB.inbox) - close(gateA) - waitSignal(t, queuedDone) - waitPoolState(t, p, 2, 1) - close(gateB) - waitPoolState(t, p, 2, 2) - if got := queuedRuns.Load(); got != 1 { - t.Fatalf("queued task ran %d times, want once", got) - } - p.mu.Lock() - closed := p.closed - p.mu.Unlock() - if closed { - t.Fatal("worker panics closed the pool without Close") - } - continued := make(chan struct{}) - if err := p.Submit(func() { close(continued) }); err != nil { - t.Fatalf("Submit after worker panics = %v", err) - } - waitSignal(t, continued) - waitPoolState(t, p, 2, 2) - p.Close() -} - -func TestSubmitHandlesClosedIdleInboxBeforeWorkerCleanup(t *testing.T) { - p, err := New(1, 1) - if err != nil { - t.Fatal(err) - } - p.mu.Lock() - dead := p.idle[0] - submitted := make(chan error, 1) - done := make(chan struct{}) attempted := make(chan struct{}) - go func() { - close(attempted) - submitted <- p.Submit(func() { close(done) }) - }() + go func() { close(attempted); blocked <- p.Submit(func() {}) }() waitSignal(t, attempted) - // Keep Submit queued on mu first, then make the published idle inbox dead. - time.Sleep(10 * time.Millisecond) - close(dead.inbox) - p.mu.Unlock() - if err := waitSignal(t, submitted); err != nil { - t.Fatalf("Submit after stale idle handoff = %v", err) + assertBlocked(t, blocked, "Submit beyond max workers and queue capacity") + close(gate) + if err := waitSignal(t, blocked); err != nil { + t.Fatalf("Submit after capacity release: %v", err) } - waitSignal(t, done) - waitPoolState(t, p, 1, 1) - p.mu.Lock() - if p.closed || p.idle[0] == dead { - p.mu.Unlock() - t.Fatal("Submit closed the pool or retained the dead worker") - } - p.mu.Unlock() p.Close() } -func TestCloseHandlesClosedIdleInboxBeforeWorkerCleanup(t *testing.T) { - p, err := New(1, 1) +func TestQueueCapacityAndMultipleBlockedSubmittersWakeIncrementally(t *testing.T) { + p, err := New(0, 2) // queue capacity one if err != nil { t.Fatal(err) } - p.mu.Lock() - dead := p.idle[0] - closeStarted, closeDone := make(chan struct{}), make(chan struct{}) - go func() { - close(closeStarted) - p.Close() - close(closeDone) - }() - waitSignal(t, closeStarted) - // Queue Close on mu before waking the worker's closed-inbox cleanup. - time.Sleep(10 * time.Millisecond) - close(dead.inbox) - p.mu.Unlock() - waitSignal(t, closeDone) - p.mu.Lock() - if !p.closed || p.total != 0 || len(p.idle) != 0 { - p.mu.Unlock() - t.Fatalf("pool after Close: closed=%v total=%d idle=%d", p.closed, p.total, len(p.idle)) - } - p.mu.Unlock() - if !p.mu.TryLock() { - t.Fatal("Close returned while still holding the pool mutex") - } - p.mu.Unlock() -} - -func TestWorkerPanicDrainsQueueAndCloseWaits(t *testing.T) { - p, err := New(2, 2) - if err != nil { - t.Fatal(err) - } - var output bytes.Buffer - previousOutput := log.Writer() - log.SetOutput(&output) - defer log.SetOutput(previousOutput) - - p.mu.Lock() - workerA := p.idle[len(p.idle)-1] - p.mu.Unlock() - started := make(chan struct{}, 2) gateA, gateB := make(chan struct{}), make(chan struct{}) + started := make(chan struct{}, 2) if err := p.Submit(func() { started <- struct{}{}; <-gateA }); err != nil { t.Fatal(err) } - p.mu.Lock() - workerB := p.idle[len(p.idle)-1] - p.mu.Unlock() if err := p.Submit(func() { started <- struct{}{}; <-gateB }); err != nil { t.Fatal(err) } waitSignal(t, started) waitSignal(t, started) - queuedStarted, queuedGate, queuedDone := make(chan struct{}), make(chan struct{}), make(chan struct{}) - var queuedRuns atomic.Int32 - if err := p.Submit(func() { close(queuedStarted); <-queuedGate; queuedRuns.Add(1); close(queuedDone) }); err != nil { + baseStarted, baseGate := make(chan struct{}), make(chan struct{}) + if err := p.Submit(func() { close(baseStarted); <-baseGate }); err != nil { t.Fatal(err) } - const closeCount = 3 - closeStarted, closeDone := make(chan struct{}, closeCount), make(chan struct{}, closeCount) - for range closeCount { - go func() { closeStarted <- struct{}{}; p.Close(); closeDone <- struct{}{} }() + type submitResult struct { + index int + err error } - for range closeCount { - waitSignal(t, closeStarted) + results := make(chan submitResult, 3) + attempted := make(chan struct{}, 3) + blockedGates := []chan struct{}{make(chan struct{}), make(chan struct{}), make(chan struct{})} + blockedStarted := make(chan int, 3) + for i := range blockedGates { + index := i + go func() { + attempted <- struct{}{} + results <- submitResult{index: index, err: p.Submit(func() { blockedStarted <- index; <-blockedGates[index] })} + }() + } + for range 3 { + waitSignal(t, attempted) + } + for range 3 { + assertBlocked(t, results, "Submitter passed a full queue") } - waitPoolClosed(t, p) - close(workerA.inbox) - close(workerB.inbox) close(gateA) - waitSignal(t, queuedStarted) - select { - case <-closeDone: - t.Fatal("Close returned before the queued task finished") - case <-time.After(10 * time.Millisecond): + waitSignal(t, baseStarted) + first := waitSignal(t, results) + if first.err != nil { + t.Fatalf("first released Submit: %v", first.err) } + assertBlocked(t, results, "more than one Submit used one freed queue slot") + close(baseGate) + firstRun := waitSignal(t, blockedStarted) + second := waitSignal(t, results) + if second.err != nil { + t.Fatalf("second released Submit: %v", second.err) + } + assertBlocked(t, results, "third Submit used a still-full queue") + close(blockedGates[firstRun]) + secondRun := waitSignal(t, blockedStarted) + close(blockedGates[secondRun]) + third := waitSignal(t, results) + if third.err != nil { + t.Fatalf("third released Submit: %v", third.err) + } + thirdRun := waitSignal(t, blockedStarted) + close(blockedGates[thirdRun]) close(gateB) - close(queuedGate) - waitSignal(t, queuedDone) - for range closeCount { - waitSignal(t, closeDone) - } - if got := queuedRuns.Load(); got != 1 { - t.Fatalf("queued task ran after worker panics %d times, want once", got) - } - p.mu.Lock() - if p.total != 0 || len(p.idle) != 0 || p.queueLen != 0 || p.queueHead != 0 || p.queueTail != 0 { - p.mu.Unlock() - t.Fatalf("pool state after worker panics: total=%d idle=%d queue=%d head=%d tail=%d", p.total, len(p.idle), p.queueLen, p.queueHead, p.queueTail) - } - for i, task := range p.queue { - if task != nil { - p.mu.Unlock() - t.Fatalf("queue slot %d retains task after worker panics", i) - } - } - p.mu.Unlock() - if !strings.Contains(output.String(), "worker panic: send on closed channel") { - t.Fatalf("worker-level panics not logged: %q", output.String()) - } + p.Close() } -func TestCloseDrainsTasksRejectsAndWakesBlockedSubmit(t *testing.T) { - p, err := New(1, 2) - if err != nil { - t.Fatal(err) - } - gate := make(chan struct{}) - started := make(chan struct{}, 3) - var completed atomic.Int32 - for range 2 { - if err := p.Submit(func() { started <- struct{}{}; <-gate; completed.Add(1) }); err != nil { - t.Fatal(err) - } - } - waitSignal(t, started) - waitSignal(t, started) - if err := p.Submit(func() { started <- struct{}{}; completed.Add(1) }); err != nil { - t.Fatal(err) - } - const submitterCount = 8 - blockedSubmit := make(chan error, submitterCount) - submitAttempted := make(chan struct{}, submitterCount) - for range submitterCount { - go func() { - submitAttempted <- struct{}{} - blockedSubmit <- p.Submit(func() {}) - }() - } - for range submitterCount { - waitSignal(t, submitAttempted) - } - select { - case err := <-blockedSubmit: - t.Fatalf("Submit did not block on full queue: %v", err) - case <-time.After(30 * time.Millisecond): - } - const closeCount = 4 - closeStarted := make(chan struct{}, closeCount) - closeDone := make(chan struct{}, closeCount) - for range closeCount { - go func() { - closeStarted <- struct{}{} - p.Close() - closeDone <- struct{}{} - }() - } - for range closeCount { - waitSignal(t, closeStarted) - } - for range submitterCount { - select { - case err := <-blockedSubmit: - if !errors.Is(err, ErrClosed) { - t.Fatalf("blocked Submit error = %v", err) - } - case <-time.After(2 * time.Second): - t.Fatal("Close did not wake every blocked Submit") - } - } - if err := p.Submit(func() {}); !errors.Is(err, ErrClosed) { - t.Fatalf("Submit after Close error = %v", err) - } - select { - case <-closeDone: - t.Fatal("Close returned before running tasks finished") - case <-time.After(30 * time.Millisecond): - } - close(gate) - waitSignal(t, started) - for range closeCount { - select { - case <-closeDone: - case <-time.After(2 * time.Second): - t.Fatal("Close did not return after accepted tasks completed") - } - } - if got := completed.Load(); got != 3 { - t.Fatalf("completed tasks = %d, want 3", got) - } - p.mu.Lock() - workers := p.total - p.mu.Unlock() - if workers != 0 { - t.Fatalf("workers after concurrent Close = %d, want 0", workers) - } -} - -func TestConcurrentCloseWaitsForRunningTask(t *testing.T) { - p, err := New(1, 2) +func TestZeroQueueMinZeroAcceptsAfterWorkerFinishes(t *testing.T) { + p, err := New(0, 1) if err != nil { t.Fatal(err) } @@ -895,34 +197,538 @@ func TestConcurrentCloseWaitsForRunningTask(t *testing.T) { t.Fatal(err) } waitSignal(t, started) - const closeCount = 4 - closeStarted := make(chan struct{}, closeCount) - closeDone := make(chan struct{}, closeCount) + second := make(chan error, 1) + attempted := make(chan struct{}) + var secondRuns atomic.Int32 + go func() { + close(attempted) + second <- p.Submit(func() { secondRuns.Add(1) }) + }() + waitSignal(t, attempted) + assertBlocked(t, second, "max=1 accepted a task while its worker was busy") + close(gate) + if err := waitSignal(t, second); err != nil { + t.Fatal(err) + } + p.Close() + if got := secondRuns.Load(); got != 1 { + t.Fatalf("second task ran %d times, want once", got) + } +} + +func TestQueueTasksAreDispatchedFIFO(t *testing.T) { + p, err := New(4, 4) // two cached tasks + if err != nil { + t.Fatal(err) + } + gates := []chan struct{}{make(chan struct{}), make(chan struct{}), make(chan struct{}), make(chan struct{})} + startedWorkers := make(chan struct{}, 4) + for _, gate := range gates { + if err := p.Submit(func() { startedWorkers <- struct{}{}; <-gate }); err != nil { + t.Fatal(err) + } + } for range 4 { - go func() { - closeStarted <- struct{}{} - p.Close() - closeDone <- struct{}{} - }() + waitSignal(t, startedWorkers) + } + queueStarted := make(chan int, 2) + queueGate := make(chan struct{}) + for i := 1; i <= 2; i++ { + index := i + if err := p.Submit(func() { queueStarted <- index; <-queueGate }); err != nil { + t.Fatal(err) + } + } + close(gates[0]) + if got := waitSignal(t, queueStarted); got != 1 { + t.Fatalf("first queued task started=%d, want 1", got) + } + close(gates[1]) + if got := waitSignal(t, queueStarted); got != 2 { + t.Fatalf("second queued task started=%d, want 2", got) + } + close(queueGate) + close(gates[2]) + close(gates[3]) + p.Close() +} + +func TestStrictShrinkThresholdAndMinimumProtection(t *testing.T) { + // Exactly min/2 other idle workers is not enough to shrink. + s := newTestScheduler(2, 5) + current := s.addFakeWorker(workerBusy) + s.addFakeWorker(workerIdle) + s.addFakeWorker(workerBusy) + s.taskFinished(current) + if s.workers[current] != workerIdle || s.retiring != 0 { + t.Fatal("worker shrank when other idle count equaled min/2") + } + + // Strictly more than min/2 permits shrink, but retiring workers are + // excluded from the effective total so repeated completions stop at min. + s = newTestScheduler(2, 6) + first := s.addFakeWorker(workerBusy) + s.addFakeWorker(workerIdle) + s.addFakeWorker(workerIdle) + second := s.addFakeWorker(workerBusy) + thirdBusy := s.addFakeWorker(workerBusy) + s.taskFinished(first) + if s.workers[first] != workerRetiring || s.retiring != 1 { + t.Fatal("worker did not shrink above the strict idle threshold") + } + s.taskFinished(second) + if s.workers[second] != workerRetiring || s.retiring != 2 { + t.Fatal("second worker should shrink while effective total remains above min") + } + s.taskFinished(thirdBusy) + if s.workers[thirdBusy] != workerRetiring || s.retiring != 3 || s.total-s.retiring != 2 { + t.Fatal("normal shrink lowered effective worker count below min") + } + lastWorker := <-s.idle + s.workers[lastWorker] = workerBusy + s.taskFinished(lastWorker) + if s.workers[lastWorker] != workerIdle || s.total-s.retiring != 2 { + t.Fatal("worker shrank below min after retiring workers were excluded") + } +} + +func TestTaskPanicsAreLoggedAndSameWorkerContinues(t *testing.T) { + var output bytes.Buffer + previous := log.Writer() + log.SetOutput(&output) + defer log.SetOutput(previous) + w := &worker{inbox: make(chan func()), stop: make(chan struct{}), dead: make(chan struct{})} + events := make(chan workerEvent, 8) + go runWorker(w, events) + for _, task := range []func(){ + func() { panic("task panic marker") }, + func() { panic(nil) }, + func() { defer func() { panic("deferred panic marker") }() }, + func() {}, + } { + w.inbox <- task + event := waitEvent(t, events) + if event.kind != taskFinished || event.worker != w { + t.Fatalf("event=%+v, want completion from same worker", event) + } + } + if !strings.Contains(output.String(), "task panic marker") || !strings.Contains(output.String(), "deferred panic marker") { + t.Fatalf("task panic logs missing markers: %q", output.String()) + } + close(w.stop) + if event := waitEvent(t, events); event.kind != workerStopped || event.worker != w { + t.Fatalf("stop event=%+v, want normal stop from same worker", event) + } +} + +type panicWriter struct { + always bool + calls atomic.Int32 +} + +func (w *panicWriter) Write(p []byte) (int, error) { + if w.calls.Add(1) > 0 && (w.always || w.calls.Load() == 1) { + panic("logger writer panic") + } + return len(p), nil +} + +func TestLoggerWriterPanicDoesNotKillTaskWorker(t *testing.T) { + for _, always := range []bool{false, true} { + t.Run(map[bool]string{false: "once", true: "always"}[always], func(t *testing.T) { + previous := log.Writer() + writer := &panicWriter{always: always} + log.SetOutput(writer) + defer log.SetOutput(previous) + w := &worker{inbox: make(chan func()), stop: make(chan struct{}), dead: make(chan struct{})} + events := make(chan workerEvent, 4) + go runWorker(w, events) + w.inbox <- func() { panic("first task") } + if event := waitEvent(t, events); event.kind != taskFinished || event.worker != w { + t.Fatalf("first event=%+v", event) + } + w.inbox <- func() { panic("second task") } + if event := waitEvent(t, events); event.kind != taskFinished || event.worker != w { + t.Fatalf("second event=%+v", event) + } + w.inbox <- func() {} + if event := waitEvent(t, events); event.kind != taskFinished || event.worker != w { + t.Fatalf("final event=%+v", event) + } + if writer.calls.Load() == 0 { + t.Fatal("logger writer was not called") + } + close(w.stop) + waitEvent(t, events) + }) + } +} + +func TestWorkerOuterPanicRecoveryReportsExit(t *testing.T) { + var output bytes.Buffer + previous := log.Writer() + log.SetOutput(&output) + defer log.SetOutput(previous) + w := &worker{inbox: make(chan func()), stop: make(chan struct{}), dead: make(chan struct{})} + close(w.inbox) // test-only fault injection into the worker receive loop + events := make(chan workerEvent, 1) + go runWorker(w, events) + event := waitEvent(t, events) + if event.kind != workerPanicked || event.worker != w { + t.Fatalf("event=%+v, want worker panic exit", event) + } + if !strings.Contains(output.String(), "worker panic: worker inbox closed") { + t.Fatalf("worker panic was not logged: %q", output.String()) + } +} + +func TestAbnormalExitStateMachineKeepsQueueAndDoesNotRefillMinimum(t *testing.T) { + s := newTestScheduler(2, 3) + w := s.addFakeWorker(workerBusy) + survivor := s.addFakeWorker(workerBusy) + var runs atomic.Int32 + done := make(chan struct{}) + s.queue <- func() { runs.Add(1); close(done) } + s.handleEvent(workerEvent{worker: w, kind: workerPanicked}) + waitSignal(t, done) + event := waitEvent(t, s.pool.events) + s.handleEvent(event) + if runs.Load() != 1 || len(s.queue) != 0 { + t.Fatalf("queue task runs=%d queue=%d, want once and empty", runs.Load(), len(s.queue)) + } + if s.total != 2 || len(s.idle) != 1 || s.workers[survivor] != workerBusy { + t.Fatalf("post-replacement total/idle=%d/%d, survivor=%d", s.total, len(s.idle), s.workers[survivor]) + } + // Retire the real replacement and account the fake surviving worker. + s.closed = true + close(s.pool.stopping) + s.pump() + s.handleEvent(waitEvent(t, s.pool.events)) + s.workerExited(survivor, false) + s.finishCloseIfReady() +} + +func TestAbnormalExitWithoutQueueDoesNotRefillButLaterSubmitWorks(t *testing.T) { + s := newTestScheduler(2, 3) + w := s.addFakeWorker(workerBusy) + s.handleEvent(workerEvent{worker: w, kind: workerPanicked}) + if s.total != 0 || s.closed { + t.Fatalf("empty abnormal exit total/closed=%d/%v, want 0/open", s.total, s.closed) + } + done := make(chan struct{}) + s.accept(func() { close(done) }) + waitSignal(t, done) + s.handleEvent(waitEvent(t, s.pool.events)) + if s.total != 1 || len(s.idle) != 1 { + t.Fatalf("later Submit total/idle=%d/%d, want one worker without min refill", s.total, len(s.idle)) + } + s.closed = true + close(s.pool.stopping) + s.pump() + s.handleEvent(waitEvent(t, s.pool.events)) +} + +func TestAbnormalExitDuringCloseStillDrainsQueue(t *testing.T) { + s := newTestScheduler(0, 2) + w := s.addFakeWorker(workerBusy) + s.closed = true + close(s.pool.stopping) + started, gate, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + s.queue <- func() { close(started); <-gate; close(done) } + s.handleEvent(workerEvent{worker: w, kind: workerPanicked}) + waitSignal(t, started) + select { + case <-s.pool.done: + t.Fatal("Close completed while replacement task was running") + default: + } + close(gate) + waitSignal(t, done) + s.handleEvent(waitEvent(t, s.pool.events)) // task completion retires worker + s.handleEvent(waitEvent(t, s.pool.events)) // retirement confirmation + select { + case <-s.pool.done: + default: + t.Fatal("Close did not finish after accepted queue drained") + } + if s.total != 0 || len(s.queue) != 0 { + t.Fatalf("final total/queue=%d/%d, want 0/0", s.total, len(s.queue)) + } +} + +func TestBusyWorkerPanicPreservesQueuedTaskAndOpenPool(t *testing.T) { + p := newTestPool(0, 2) + s := newScheduler(p) + w1 := s.startWorker(nil, false) + _ = s.startWorker(nil, false) + + started := make(chan string, 2) + gates := map[string]chan struct{}{"A": make(chan struct{}), "B": make(chan struct{})} + var runs [4]atomic.Int32 + s.accept(func() { started <- "A"; <-gates["A"]; runs[0].Add(1) }) + s.accept(func() { started <- "B"; <-gates["B"]; runs[1].Add(1) }) + waitSignal(t, started) + waitSignal(t, started) + queuedDone := make(chan struct{}) + s.queue <- func() { runs[2].Add(1); close(queuedDone) } + go s.run() + + close(w1.inbox) // worker A fails after its current task reports completion + close(gates["A"]) + waitSignal(t, queuedDone) + close(gates["B"]) + postFailureDone := make(chan struct{}) + if err := p.Submit(func() { runs[3].Add(1); close(postFailureDone) }); err != nil { + t.Fatalf("Submit after worker failure: %v", err) + } + waitSignal(t, postFailureDone) + if pClosed(p) { + t.Fatal("worker panic closed the pool") + } + p.Close() + for i := range runs { + if got := runs[i].Load(); got != 1 { + t.Errorf("task %d ran %d times, want once", i, got) + } + } +} + +func TestIdleWorkerPanicDuringSubmitDoesNotLoseTask(t *testing.T) { + for range 8 { + p := newTestPool(0, 1) + s := newScheduler(p) + w := s.startWorker(nil, false) + close(w.inbox) + go s.run() + var runs atomic.Int32 + done := make(chan struct{}) + if err := p.Submit(func() { runs.Add(1); close(done) }); err != nil { + t.Fatalf("Submit during idle worker exit: %v", err) + } + waitSignal(t, done) + if pClosed(p) { + t.Fatal("worker panic closed the pool") + } + p.Close() + if got := runs.Load(); got != 1 { + t.Fatalf("task ran %d times, want once", got) + } + } +} + +func TestBusyWorkerPanicDuringCloseDrainsQueuedTask(t *testing.T) { + p := newTestPool(0, 2) + s := newScheduler(p) + w1 := s.startWorker(nil, false) + _ = s.startWorker(nil, false) + started := make(chan string, 2) + gateA, gateB := make(chan struct{}), make(chan struct{}) + var queuedRuns atomic.Int32 + s.accept(func() { started <- "A"; <-gateA }) + s.accept(func() { started <- "B"; <-gateB }) + waitSignal(t, started) + waitSignal(t, started) + queuedDone := make(chan struct{}) + s.queue <- func() { queuedRuns.Add(1); close(queuedDone) } + go s.run() + + closeStarted := make(chan struct{}) + go func() { p.Close(); close(closeStarted) }() + waitSignal(t, p.stopping) + close(w1.inbox) + close(gateA) + waitSignal(t, queuedDone) + select { + case <-closeStarted: + t.Fatal("Close returned before remaining worker finished") + default: + } + close(gateB) + waitSignal(t, closeStarted) + if got := queuedRuns.Load(); got != 1 { + t.Fatalf("queued task ran %d times, want once", got) + } +} + +type gatedLogWriter struct { + entered chan struct{} + release chan struct{} +} + +func (w *gatedLogWriter) Write(p []byte) (int, error) { + w.entered <- struct{}{} + <-w.release + return len(p), nil +} + +func TestCloseWaitsForWorkerPanicLogsAndTerminalEvents(t *testing.T) { + writer := &gatedLogWriter{entered: make(chan struct{}, 2), release: make(chan struct{}, 2)} + previous := log.Writer() + log.SetOutput(writer) + defer log.SetOutput(previous) + + p := newTestPool(0, 2) + s := newScheduler(p) + w1 := s.startWorker(nil, false) + w2 := s.startWorker(nil, false) + started := make(chan string, 2) + gateA, gateB, gateC := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var runs [3]atomic.Int32 + s.accept(func() { started <- "A"; <-gateA; runs[0].Add(1) }) + s.accept(func() { started <- "B"; <-gateB; runs[1].Add(1) }) + waitSignal(t, started) + waitSignal(t, started) + cStarted, cDone := make(chan struct{}), make(chan struct{}) + s.queue <- func() { runs[2].Add(1); close(cStarted); <-gateC; close(cDone) } + go s.run() + + close(w1.inbox) + close(gateA) + waitSignal(t, writer.entered) + close(gateB) + waitSignal(t, cStarted) // the surviving worker continues queued work + close(w2.inbox) + close(gateC) + waitSignal(t, cDone) + + closeDone := make(chan struct{}) + go func() { p.Close(); close(closeDone) }() + waitSignal(t, p.stopping) + assertBlocked(t, closeDone, "Close returned before delayed panic logging and terminal events") + + writer.release <- struct{}{} + waitSignal(t, writer.entered) + assertBlocked(t, closeDone, "Close returned before second worker terminal event") + writer.release <- struct{}{} + waitSignal(t, closeDone) + for i := range runs { + if got := runs[i].Load(); got != 1 { + t.Errorf("task %d ran %d times, want once", i, got) + } + } +} + +func TestAcceptedPendingHandoffSurvivesDelayedExitAndClose(t *testing.T) { + writer := &gatedLogWriter{entered: make(chan struct{}, 1), release: make(chan struct{}, 1)} + previous := log.Writer() + log.SetOutput(writer) + defer log.SetOutput(previous) + + p := newTestPool(0, 1) // zero queue capacity + s := newScheduler(p) + w := s.startWorker(nil, false) + close(w.inbox) + waitSignal(t, writer.entered) + + var runs atomic.Int32 + task := func() { runs.Add(1) } + if s.accept(task) { + t.Fatal("handoff to exited idle worker unexpectedly succeeded") + } + // This is the task already received by the actor: keep its idle-worker + // handoff slot reserved while the old worker finishes delayed logging. + s.pending = task + s.hasPending = true + go s.run() + + closeDone := make(chan struct{}) + go func() { p.Close(); close(closeDone) }() + waitSignal(t, p.stopping) + assertBlocked(t, closeDone, "Close returned before pending accepted task was dispatched") + writer.release <- struct{}{} + waitSignal(t, closeDone) + if got := runs.Load(); got != 1 { + t.Fatalf("pending task ran %d times, want once", got) + } +} + +func TestCloseDrainsAcceptedTasksAndWakesBlockedSubmitters(t *testing.T) { + p, err := New(2, 2) // queue capacity one + if err != nil { + t.Fatal(err) + } + gates := []chan struct{}{make(chan struct{}), make(chan struct{})} + started := make(chan struct{}, 2) + var completed atomic.Int32 + for _, gate := range gates { + if err := p.Submit(func() { started <- struct{}{}; <-gate; completed.Add(1) }); err != nil { + t.Fatal(err) + } + } + waitSignal(t, started) + waitSignal(t, started) + if err := p.Submit(func() { completed.Add(1) }); err != nil { + t.Fatal(err) + } + blocked := make(chan error, 1) + attempted := make(chan struct{}) + go func() { close(attempted); blocked <- p.Submit(func() { completed.Add(1000) }) }() + waitSignal(t, attempted) + assertBlocked(t, blocked, "saturated Submit returned before Close") + const closeCount = 4 + closes := make(chan struct{}, closeCount) + closeStarted := make(chan struct{}, closeCount) + for range closeCount { + go func() { closeStarted <- struct{}{}; p.Close(); closes <- struct{}{} }() } for range closeCount { waitSignal(t, closeStarted) } + waitSignal(t, p.stopping) + if err := waitSignal(t, blocked); !errors.Is(err, ErrClosed) { + t.Fatalf("blocked Submit error=%v, want ErrClosed", err) + } select { - case <-closeDone: - t.Fatal("Close returned while a task was still running") - case <-time.After(30 * time.Millisecond): + case <-closes: + t.Fatal("Close returned while accepted gated work remained") + default: } - close(gate) + close(gates[0]) + close(gates[1]) for range closeCount { - waitSignal(t, closeDone) + waitSignal(t, closes) } - p.mu.Lock() - if p.total != 0 { - p.mu.Unlock() - t.Fatalf("workers after concurrent Close = %d, want 0", p.total) + if got := completed.Load(); got != 3 { + t.Fatalf("accepted tasks completed=%d, want 3", got) + } + if err := p.Submit(func() {}); !errors.Is(err, ErrClosed) { + t.Fatalf("Submit after Close=%v, want ErrClosed", err) + } +} + +func TestConcurrentSubmitAndCloseMatchesEveryResult(t *testing.T) { + const count = 100 + p, err := New(2, 4) + if err != nil { + t.Fatal(err) + } + var calls [count]atomic.Int32 + results := make([]error, count) + start := make(chan struct{}) + var submits sync.WaitGroup + for i := range count { + submits.Add(1) + go func(i int) { + defer submits.Done() + <-start + results[i] = p.Submit(func() { calls[i].Add(1) }) + }(i) + } + closeDone := make(chan struct{}) + go func() { <-start; p.Close(); close(closeDone) }() + close(start) + waitGroup(t, &submits) + waitSignal(t, closeDone) + for i, err := range results { + switch { + case err == nil && calls[i].Load() != 1: + t.Errorf("task %d accepted but ran %d times", i, calls[i].Load()) + case errors.Is(err, ErrClosed) && calls[i].Load() != 0: + t.Errorf("task %d rejected but ran %d times", i, calls[i].Load()) + case err != nil && !errors.Is(err, ErrClosed): + t.Errorf("task %d unexpected error: %v", i, err) + } } - p.mu.Unlock() } func TestReentrantSubmit(t *testing.T) { @@ -930,507 +736,60 @@ func TestReentrantSubmit(t *testing.T) { if err != nil { t.Fatal(err) } - innerDone, outerDone := make(chan struct{}), make(chan struct{}) + inner, outer := make(chan struct{}), make(chan struct{}) if err := p.Submit(func() { - if err := p.Submit(func() { close(innerDone) }); err != nil { - t.Errorf("reentrant Submit: %v", err) + if err := p.Submit(func() { close(inner) }); err != nil { + t.Errorf("nested Submit: %v", err) } - <-innerDone - close(outerDone) + <-inner + close(outer) }); err != nil { t.Fatal(err) } - waitSignal(t, outerDone) + waitSignal(t, outer) p.Close() } -func TestGoexitMaxOneCloseBeforeAndAfterWorkerExit(t *testing.T) { - t.Run("close while worker is running", func(t *testing.T) { - p, err := New(0, 1) - if err != nil { - t.Fatal(err) - } - gate, started, exited := make(chan struct{}), make(chan struct{}), make(chan struct{}) - if err := p.Submit(func() { - defer close(exited) - close(started) - <-gate - runtime.Goexit() - }); err != nil { - t.Fatal(err) - } - waitSignal(t, started) - const closeCount = 3 - closeStarted := make(chan struct{}, closeCount) - closeDone := make(chan struct{}, closeCount) - for range closeCount { - go func() { closeStarted <- struct{}{}; p.Close(); closeDone <- struct{}{} }() - } - for range closeCount { - waitSignal(t, closeStarted) - } - select { - case <-closeDone: - t.Fatal("Close returned before the gated task exited") - case <-time.After(10 * time.Millisecond): - } - close(gate) - waitSignal(t, exited) - for range closeCount { - waitSignal(t, closeDone) - } - waitPoolState(t, p, 0, 0) - }) - - t.Run("close after worker exit", func(t *testing.T) { - p, err := New(0, 1) - if err != nil { - t.Fatal(err) - } - exited := make(chan struct{}) - if err := p.Submit(func() { defer close(exited); runtime.Goexit() }); err != nil { - t.Fatal(err) - } - waitSignal(t, exited) - waitPoolState(t, p, 0, 0) - const closeCount = 3 - var closes sync.WaitGroup - for range closeCount { - closes.Add(1) - go func() { defer closes.Done(); p.Close() }() - } - done := make(chan struct{}) - go func() { closes.Wait(); close(done) }() - waitSignal(t, done) - }) +func newTestPool(min, max int) *Pool { + return &Pool{ + submit: make(chan func()), closeReq: make(chan struct{}), stopping: make(chan struct{}), + done: make(chan struct{}), events: make(chan workerEvent, max), min: min, max: max, queueCap: max / 2, + } } -func TestGoexitReplacesWorkerAndKeepsQueuedTasks(t *testing.T) { - p, err := New(2, 2) - if err != nil { - t.Fatal(err) - } - goexitGate, survivorGate := make(chan struct{}), make(chan struct{}) - started := make(chan struct{}, 2) - goexitDone := make(chan struct{}) - if err := p.Submit(func() { defer close(goexitDone); started <- struct{}{}; <-goexitGate; runtime.Goexit() }); err != nil { - t.Fatal(err) - } - if err := p.Submit(func() { started <- struct{}{}; <-survivorGate }); err != nil { - t.Fatal(err) - } - waitSignal(t, started) - waitSignal(t, started) - queuedDone := make(chan struct{}) - if err := p.Submit(func() { close(queuedDone) }); err != nil { - t.Fatal(err) - } - close(goexitGate) - waitSignal(t, goexitDone) - waitSignal(t, queuedDone) - waitPoolState(t, p, 2, 1) - p.mu.Lock() - if p.queueLen != 0 { - p.mu.Unlock() - t.Fatalf("queue length after replacement task = %d, want 0", p.queueLen) - } - p.mu.Unlock() - close(survivorGate) - waitPoolState(t, p, 2, 2) - p.Close() -} +func newTestScheduler(min, max int) *scheduler { return newScheduler(newTestPool(min, max)) } -func TestCloseWaitsAndDrainsQueueAfterGoexit(t *testing.T) { - p, err := New(2, 2) // queue capacity one - if err != nil { - t.Fatal(err) - } - gateA, gateB := make(chan struct{}), make(chan struct{}) - started := make(chan struct{}, 2) - exited := make(chan struct{}, 2) - for _, gate := range []chan struct{}{gateA, gateB} { - if err := p.Submit(func() { - defer func() { exited <- struct{}{} }() - started <- struct{}{} - <-gate - runtime.Goexit() - }); err != nil { - t.Fatal(err) - } - } - waitSignal(t, started) - waitSignal(t, started) - queuedStarted, queuedGate, queuedDone := make(chan struct{}), make(chan struct{}), make(chan struct{}) - var queuedRuns atomic.Int32 - if err := p.Submit(func() { close(queuedStarted); <-queuedGate; queuedRuns.Add(1); close(queuedDone) }); err != nil { - t.Fatal(err) - } - - const closeCount = 4 - closeStarted := make(chan struct{}, closeCount) - closeDone := make(chan struct{}, closeCount) - for range closeCount { - go func() { - closeStarted <- struct{}{} - p.Close() - closeDone <- struct{}{} - }() - } - for range closeCount { - waitSignal(t, closeStarted) - } - deadline := time.Now().Add(2 * time.Second) - for { - p.mu.Lock() - closed := p.closed - p.mu.Unlock() - if closed { - break - } - if time.Now().After(deadline) { - t.Fatal("Close did not mark the pool closed") - } - time.Sleep(time.Millisecond) - } +func pClosed(p *Pool) bool { select { - case <-closeDone: - t.Fatal("Close returned while both gated tasks were still running") - case <-time.After(10 * time.Millisecond): - } - - close(gateA) - waitSignal(t, exited) - waitSignal(t, queuedStarted) - select { - case <-closeDone: - t.Fatal("Close returned while the replacement task was still gated") - case <-time.After(10 * time.Millisecond): - } - - close(gateB) - waitSignal(t, exited) - select { - case <-closeDone: - t.Fatal("Close returned before the queued task finished") - case <-time.After(10 * time.Millisecond): - } - close(queuedGate) - waitSignal(t, queuedDone) - for range closeCount { - waitSignal(t, closeDone) - } - if got := queuedRuns.Load(); got != 1 { - t.Fatalf("queued task executed after Goexit %d times, want once", got) - } - p.mu.Lock() - if p.total != 0 || len(p.idle) != 0 || p.queueLen != 0 || p.queueHead != 0 || p.queueTail != 0 { - p.mu.Unlock() - t.Fatalf("pool state after Close: total=%d idle=%d queue=%d head=%d tail=%d", p.total, len(p.idle), p.queueLen, p.queueHead, p.queueTail) - } - for i, task := range p.queue { - if task != nil { - p.mu.Unlock() - t.Fatalf("queue slot %d retains a task after Close", i) - } - } - p.mu.Unlock() - if err := p.Submit(func() {}); !errors.Is(err, ErrClosed) { - t.Fatalf("Submit after Close = %v, want ErrClosed", err) + case <-p.stopping: + return true + default: + return false } } -func TestGoexitWakesSubmitAndAllowsWorkerRecreation(t *testing.T) { - p, err := New(0, 1) // zero-capacity queue - if err != nil { - t.Fatal(err) +func (s *scheduler) addFakeWorker(state workerState) *worker { + w := &worker{inbox: make(chan func(), 1), stop: make(chan struct{}), dead: make(chan struct{})} + s.workers[w] = state + s.total++ + if state == workerIdle { + s.idle <- w } - gate, started, exited, taskDone := make(chan struct{}), make(chan struct{}), make(chan struct{}), make(chan struct{}) - if err := p.Submit(func() { defer close(exited); close(started); <-gate; runtime.Goexit() }); err != nil { - t.Fatal(err) + if state == workerRetiring { + s.retiring++ } - waitSignal(t, started) - attempted := make(chan struct{}) - submitted := make(chan error, 1) - go func() { close(attempted); submitted <- p.Submit(func() { close(taskDone) }) }() - waitSignal(t, attempted) - select { - case err := <-submitted: - t.Fatalf("Submit returned while the only worker was busy: %v", err) - case <-time.After(10 * time.Millisecond): - } - close(gate) - waitSignal(t, exited) - if err := waitSignal(t, submitted); err != nil { - t.Fatalf("Submit after Goexit = %v", err) - } - waitSignal(t, taskDone) - waitPoolState(t, p, 1, 1) - p.Close() + return w } -func TestAllGoexitWorkersAreReplacedAndPoolCanBeReused(t *testing.T) { - p, err := New(4, 4) // queue capacity two; all workers are occupied - if err != nil { - t.Fatal(err) - } - gates := make([]chan struct{}, 4) - started := make(chan struct{}, 4) - exited := make(chan struct{}, 4) - for i := range gates { - gates[i] = make(chan struct{}) - } - for _, gate := range gates { - if err := p.Submit(func() { started <- struct{}{}; <-gate; exited <- struct{}{}; runtime.Goexit() }); err != nil { - t.Fatal(err) - } - } - for range 4 { - waitSignal(t, started) - } - var queuedRuns [2]atomic.Int32 - queuedDone := make(chan struct{}, len(queuedRuns)) - for i := range queuedRuns { - if err := p.Submit(func() { queuedRuns[i].Add(1); queuedDone <- struct{}{} }); err != nil { - t.Fatal(err) - } - } - close(gates[0]) - waitSignal(t, exited) - for range queuedRuns { - waitSignal(t, queuedDone) - } - for i := 1; i < len(gates); i++ { - close(gates[i]) - waitSignal(t, exited) - } - waitPoolState(t, p, 4, 4) - p.mu.Lock() - if p.closed || p.queueLen != 0 || p.queueHead != 0 || p.queueTail != 0 { - p.mu.Unlock() - t.Fatalf("pool after worker Goexit: closed=%v queue len/head/tail=%d/%d/%d", p.closed, p.queueLen, p.queueHead, p.queueTail) - } - for i, task := range p.queue { - if task != nil { - p.mu.Unlock() - t.Fatalf("queue slot %d retains a completed task", i) - } - } - p.mu.Unlock() - for i := range queuedRuns { - if got := queuedRuns[i].Load(); got != 1 { - t.Fatalf("queued task %d ran %d times, want once", i, got) - } - } - reused := make(chan struct{}) - if err := p.Submit(func() { close(reused) }); err != nil { - t.Fatalf("Submit after queue clear = %v", err) - } - waitSignal(t, reused) - p.Close() -} - -func TestConcurrentSubmitAndClose(t *testing.T) { - const rounds = 10 - const taskCount = 64 - for round := range rounds { - p, err := New(2, 4) - if err != nil { - t.Fatal(err) - } - var calls [taskCount]atomic.Int32 - results := make([]error, taskCount) - start := make(chan struct{}) - var submitters sync.WaitGroup - for i := range taskCount { - submitters.Add(1) - go func(i int) { - defer submitters.Done() - <-start - results[i] = p.Submit(func() { calls[i].Add(1) }) - }(i) - } - closeDone := make(chan struct{}) - go func() { <-start; p.Close(); close(closeDone) }() - close(start) - allSubmitted := make(chan struct{}) - go func() { submitters.Wait(); close(allSubmitted) }() - select { - case <-allSubmitted: - case <-time.After(3 * time.Second): - t.Fatalf("round %d: Submit calls did not finish", round) - } - select { - case <-closeDone: - case <-time.After(3 * time.Second): - t.Fatalf("round %d: Close did not finish", round) - } - for i, result := range results { - got := calls[i].Load() - switch { - case result == nil && got != 1: - t.Fatalf("round %d task %d: Submit succeeded, execution count = %d", round, i, got) - case errors.Is(result, ErrClosed) && got != 0: - t.Fatalf("round %d task %d: Submit returned ErrClosed, execution count = %d", round, i, got) - case result != nil && !errors.Is(result, ErrClosed): - t.Fatalf("round %d task %d: unexpected Submit error %v", round, i, result) - } - } - p.mu.Lock() - workers := p.total - p.mu.Unlock() - if workers != 0 { - t.Fatalf("round %d: workers after Close = %d, want 0", round, workers) - } - } -} - -func TestSaturatedSubmitAndClose(t *testing.T) { - const rounds = 10 - const taskCount = 64 - for round := range rounds { - p, err := New(4, 4) - if err != nil { - t.Fatal(err) - } - gate := make(chan struct{}) - workersStarted := make(chan struct{}, 4) - for range 4 { - if err := p.Submit(func() { workersStarted <- struct{}{}; <-gate }); err != nil { - t.Fatal(err) - } - } - for range 4 { - waitSignal(t, workersStarted) - } - var calls [taskCount]atomic.Int32 - type submitResult struct { - index int - err error - } - results := make(chan submitResult, taskCount) - start := make(chan struct{}) - attempted := make(chan struct{}, taskCount) - var submitters sync.WaitGroup - for i := range taskCount { - submitters.Add(1) - go func(i int) { - defer submitters.Done() - attempted <- struct{}{} - <-start - results <- submitResult{index: i, err: p.Submit(func() { calls[i].Add(1) })} - }(i) - } - for range taskCount { - waitSignal(t, attempted) - } - close(start) - seen := make([]bool, taskCount) - acceptedTasks := make([]bool, taskCount) - accepted := 0 - for accepted < p.queueCap { - result := waitSignal(t, results) - if seen[result.index] { - t.Fatalf("round %d task %d: duplicate Submit result", round, result.index) - } - seen[result.index] = true - if result.err != nil { - t.Fatalf("round %d: Submit returned %v before Close", round, result.err) - } - acceptedTasks[result.index] = true - accepted++ - } - closeDone := make(chan struct{}) - go func() { p.Close(); close(closeDone) }() - for received := accepted; received < taskCount; received++ { - result := waitSignal(t, results) - if seen[result.index] { - t.Fatalf("round %d task %d: duplicate Submit result", round, result.index) - } - seen[result.index] = true - if !errors.Is(result.err, ErrClosed) { - t.Fatalf("round %d task %d: Submit error = %v, want ErrClosed", round, result.index, result.err) - } - } - submitters.Wait() - select { - case <-closeDone: - t.Fatalf("round %d: Close returned before blocked worker tasks finished", round) - case <-time.After(10 * time.Millisecond): - } - close(gate) - select { - case <-closeDone: - case <-time.After(3 * time.Second): - t.Fatalf("round %d: Close did not finish", round) - } - for i := range taskCount { - if !seen[i] { - t.Fatalf("round %d task %d: missing Submit result", round, i) - } - got := calls[i].Load() - switch { - case acceptedTasks[i] && got != 1: - t.Fatalf("round %d task %d: Submit succeeded, execution count = %d", round, i, got) - case !acceptedTasks[i] && got != 0: - t.Fatalf("round %d task %d: Submit returned ErrClosed, execution count = %d", round, i, got) - } - } - p.mu.Lock() - workers := p.total - p.mu.Unlock() - if workers != 0 { - t.Fatalf("round %d: workers after Close = %d, want 0", round, workers) - } - } -} - -func waitPoolState(t *testing.T, p *Pool, total, idle int) { +func waitEvent(t *testing.T, events <-chan workerEvent) workerEvent { t.Helper() - deadline := time.Now().Add(2 * time.Second) - for time.Now().Before(deadline) { - p.mu.Lock() - gotTotal, gotIdle := p.total, len(p.idle) - p.mu.Unlock() - if gotTotal == total && gotIdle == idle { - return - } - time.Sleep(time.Millisecond) + select { + case event := <-events: + return event + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for worker event") + return workerEvent{} } - p.mu.Lock() - defer p.mu.Unlock() - t.Fatalf("pool state workers/idle = %d/%d, want %d/%d", p.total, len(p.idle), total, idle) -} - -func waitPoolQueueState(t *testing.T, p *Pool, total, idle, queueLen int) { - t.Helper() - deadline := time.Now().Add(2 * time.Second) - for time.Now().Before(deadline) { - p.mu.Lock() - gotTotal, gotIdle, gotQueue := p.total, len(p.idle), p.queueLen - p.mu.Unlock() - if gotTotal == total && gotIdle == idle && gotQueue == queueLen { - return - } - time.Sleep(time.Millisecond) - } - p.mu.Lock() - defer p.mu.Unlock() - t.Fatalf("pool state workers/idle/queue = %d/%d/%d, want %d/%d/%d", p.total, len(p.idle), p.queueLen, total, idle, queueLen) -} - -func waitPoolClosed(t *testing.T, p *Pool) { - t.Helper() - deadline := time.Now().Add(2 * time.Second) - for time.Now().Before(deadline) { - p.mu.Lock() - closed := p.closed - p.mu.Unlock() - if closed { - return - } - time.Sleep(time.Millisecond) - } - t.Fatal("pool did not enter the closed state") } func waitSignal[T any](t *testing.T, ch <-chan T) T { @@ -1438,9 +797,25 @@ func waitSignal[T any](t *testing.T, ch <-chan T) T { select { case value := <-ch: return value - case <-time.After(2 * time.Second): + case <-time.After(3 * time.Second): var zero T t.Fatal("timed out waiting for signal") return zero } } + +func assertBlocked[T any](t *testing.T, ch <-chan T, message string) { + t.Helper() + select { + case value := <-ch: + t.Fatalf("%s: unexpectedly received %v", message, value) + case <-time.After(20 * time.Millisecond): + } +} + +func waitGroup(t *testing.T, group *sync.WaitGroup) { + t.Helper() + done := make(chan struct{}) + go func() { group.Wait(); close(done) }() + waitSignal(t, done) +}