gammazero/workerpool — Senior Level¶

Table of Contents¶

1. Introduction
2. Library Layout
3. The WorkerPool Struct
4. New: Construction Walkthrough
5. The Dispatcher Loop
6. Submit: From User to Worker
7. SubmitWait: The Done-Channel Trick
8. The Worker Goroutine
9. Idle-Worker Reaping in Detail
10. The Internal Queue Structure
11. Stop and StopWait Implementation
12. Pause Semantics
13. Synchronisation Primitives Used
14. The Cost of Each Operation
15. Comparison with a Hand-Rolled chan func() Pool
16. Comparison with ants
17. Comparison with tunny
18. Dynamic Scaling Considerations
19. Custom Pool Patterns
20. Adapting the Library for Your Needs
21. Memory Layout and Allocations
22. Race Conditions That Could Exist
23. Tricky Internals
24. Edge Cases at the Library Level
25. Debugging Pool Internals
26. Common Senior-Level Mistakes
27. Test
28. Tricky Questions
29. Cheat Sheet
30. Self-Assessment Checklist
31. Summary
32. What You Can Build
33. Further Reading
34. Related Topics
35. Diagrams and Visual Aids

Introduction¶

The junior file taught the API. The middle file taught the operations. The senior file teaches the code.

gammazero/workerpool is roughly 300 lines. By the end of this file you should be able to:

• Sketch the pool struct from memory.
• Explain the dispatcher loop's select block.
• Trace a Submit from the caller's goroutine to the worker's goroutine.
• Explain how idle workers are reaped, including the timer mechanism.
• Compare the design against ants, tunny, and a hand-rolled pool.
• Identify which design decisions are intentional and which are accidental.
• Decide whether to fork the library, wrap it, or replace it for a given need.

This is not about API mastery anymore. It is about understanding why the API behaves the way it does, with enough source-level knowledge that you can answer "what would happen if I changed X" with confidence.

This file uses an approximation of the library's code where helpful. Versions of workerpool differ in small ways; we describe the v1.x design. Read the actual source alongside if precision matters. The code we present is faithful to the public behaviour but may simplify naming or layout.

Library Layout¶

The repository has, in essence, two source files:

workerpool.go              # the library
workerpool_test.go         # tests

No subpackages. No internal directories. The public type is WorkerPool. The constructor is New. Methods are Submit, SubmitWait, Stop, StopWait, Stopped, WaitingQueueSize, and (in newer versions) Pause.

This is intentionally small. The author has resisted feature creep for years. Open issues that request "metrics support" or "context-aware submit" tend to be closed with "wrap it yourself" — and they're right.

The WorkerPool Struct¶

The struct, simplified:

type WorkerPool struct {
    maxWorkers   int
    taskQueue    chan func()
    workerQueue  chan func()
    stoppedChan  chan struct{}
    stopSignal   chan struct{}
    waitingQueue deque.Deque[func()]
    stopLock     sync.Mutex
    waiting      int32
    stopped      bool
    wait         bool
}

Let us walk each field:

• maxWorkers — the cap on concurrent workers. Set at construction; read-only.
• taskQueue — an unbuffered channel of func(). This is what Submit writes to.
• workerQueue — an unbuffered channel of func(). This is what workers read from.
• stoppedChan — closed when shutdown completes. Used by Stop and StopWait to know "we're done".
• stopSignal — closed when shutdown begins. The dispatcher and workers watch this.
• waitingQueue — the deque (double-ended queue) of tasks waiting for a free worker.
• stopLock — protects stopped and wait (and the act of closing stopSignal).
• waiting — an atomic counter of tasks currently in waitingQueue. Exposed as WaitingQueueSize.
• stopped — bool, true after Stop or StopWait is called.
• wait — bool, true if StopWait (drain queue) vs Stop (discard).

The exact field names and types vary by version, but the shape is constant: two channels (input + output of dispatcher), a deque for queued tasks, a lock, and shutdown bookkeeping.

The two-channel design (taskQueue and workerQueue) is the key insight. Why two channels? Because the dispatcher needs to choose between feeding workers and accepting new submissions. With one channel, the choice is fixed by who reaches the channel first. With two channels and a select, the dispatcher has full control.

New: Construction Walkthrough¶

New(maxWorkers int) does roughly:

func New(maxWorkers int) *WorkerPool {
    if maxWorkers < 1 {
        maxWorkers = 1
    }
    pool := &WorkerPool{
        maxWorkers:  maxWorkers,
        taskQueue:   make(chan func()),
        workerQueue: make(chan func()),
        stoppedChan: make(chan struct{}),
        stopSignal:  make(chan struct{}),
    }
    go pool.dispatch()
    return pool
}

Things to note:

1. maxWorkers < 1 is silently clamped to 1. This is friendly but ambiguous; some users would prefer a panic. The library's choice prevents New(0) from creating a pool that never runs anything.
2. Both channels are unbuffered. That is deliberate — buffered channels would let the dispatcher get "ahead" of itself, complicating the back-and-forth between submitter, dispatcher, and worker.
3. The dispatcher goroutine starts immediately. So New always leaks at least one goroutine until Stop/StopWait.
4. No worker goroutines are created here. Workers spin up lazily inside dispatch when there is work to do.

The constructor is intentionally synchronous and very short. There is no error to return — the only failure mode would be running out of goroutines, which would manifest as a runtime panic on go pool.dispatch().

The Dispatcher Loop¶

The heart of the library. Simplified:

func (p *WorkerPool) dispatch() {
    defer close(p.stoppedChan)

    timeout := time.NewTimer(idleTimeout)
    var workerCount int
    var idle bool

Loop:
    for {
        if p.waitingQueue.Len() != 0 {
            if !p.processWaitingQueue() {
                break Loop
            }
            continue
        }

        select {
        case task, ok := <-p.taskQueue:
            if !ok {
                break Loop
            }
            select {
            case p.workerQueue <- task:
            default:
                if workerCount < p.maxWorkers {
                    go worker(task, p.workerQueue)
                    workerCount++
                } else {
                    p.waitingQueue.PushBack(task)
                    atomic.StoreInt32(&p.waiting, int32(p.waitingQueue.Len()))
                }
            }
            idle = false
        case <-timeout.C:
            if idle && workerCount > 0 {
                if p.killIdleWorker() {
                    workerCount--
                }
            }
            idle = true
            timeout.Reset(idleTimeout)
        }
    }

    // shutdown
    if p.wait {
        p.runQueuedTasks()
    }
    for workerCount > 0 {
        p.workerQueue <- nil // poison pill
        workerCount--
    }
    timeout.Stop()
}

This is the most important block of code in the library. Let us read it carefully.

Outer loop structure. The dispatcher loops until told to stop. Each iteration does one of two things: process a queued task (if any), or select on incoming work and idle timeout.

The queued-task fast path. If waitingQueue is non-empty, the dispatcher tries to forward its head to a worker before considering new submissions. This gives queued tasks priority over new ones — first-come-first-served behaviour.

The select block. - case task, ok := <-p.taskQueue: — a Submit arrived. The dispatcher then tries to hand it directly to an idle worker via case p.workerQueue <- task:. If no worker is ready (the inner select hits default), the dispatcher either spawns a new worker (if workerCount < maxWorkers) or queues the task. - case <-timeout.C: — the idle timer fired. If the pool has been idle since the last timeout and has workers, reap one. The timer resets and idle is set to true.

The idle flag. A subtle bookkeeping variable. It is false whenever a task arrives. It becomes true only after a timeout fires with no work. So two consecutive timer fires with no submits in between trigger reaping. This makes the reaping window roughly 2-4 seconds, not exactly 2.

Shutdown. When taskQueue closes (ok = false), the loop breaks. Then: - If wait is true (i.e., StopWait was called), run the remaining queued tasks. - Send nil to each worker as a poison pill — workers exit on nil. - Stop the timer. - The deferred close(p.stoppedChan) signals "all done" to anyone waiting.

This dispatcher design is clean and small. Notice what it does not do: - No mutex around the queue (it is goroutine-local to the dispatcher). - No worker goroutine bookkeeping beyond a count. - No retries, no priorities, no rate limiting.

The simplicity is the point.

Submit: From User to Worker¶

A trace of one Submit:

func (p *WorkerPool) Submit(task func()) {
    if task != nil {
        p.taskQueue <- task
    }
}

That is the entire method. (Some versions check Stopped first; we'll see how shutdown coordination works below.)

So Submit simply blocks on a channel send to taskQueue. The dispatcher is the receiver. If the dispatcher is in the middle of doing other work (e.g., flushing queued tasks), the submitter waits — but only for microseconds, because the dispatcher loops back to the select quickly.

Steps:

1. Caller calls Submit(myFunc).
2. Send myFunc on taskQueue. Block here until dispatcher reads.
3. Dispatcher's case task := <-p.taskQueue fires.
4. Dispatcher decides: hand to worker, spawn worker, or queue.
5. Submit returns to caller.

The whole round trip is ~100-300 nanoseconds for a non-contended dispatcher.

Why an unbuffered taskQueue? Because the dispatcher must make decisions about each incoming task immediately. A buffer would let many submissions pile up before the dispatcher saw them, which is exactly what waitingQueue is for — but waitingQueue is inside the dispatcher, not a buffer between submitter and dispatcher. The unbuffered design forces tight coordination.

Why Submit "never blocks" then? Because the dispatcher's select loop is fast. The dispatcher reads taskQueue in a tight loop; the submitter's send completes as soon as the dispatcher's next iteration. Unless the dispatcher is in shutdown or paused, the send completes in nanoseconds.

This is also why Submit is not truly non-blocking in the sense of select { default: ... }. It will block if the dispatcher is busy. But the dispatcher is rarely busy for long, so practically the block is invisible.

SubmitWait: The Done-Channel Trick¶

func (p *WorkerPool) SubmitWait(task func()) {
    if task == nil {
        return
    }
    doneChan := make(chan struct{})
    p.Submit(func() {
        defer close(doneChan)
        task()
    })
    <-doneChan
}

This is the entire implementation. Three observations:

1. SubmitWait is built on Submit, not the other way around. The library exposes both because users want both, but conceptually SubmitWait is sugar.
2. The done channel is unbuffered, size 0. The wrapper closes it after the task runs.
3. The wrapper uses defer close(doneChan), which means even if task() panics, the close fires and SubmitWait returns. The library's outer recover then catches the panic.

The cost of SubmitWait over Submit is: - One channel allocation (make(chan struct{})). - One closure allocation (the wrapper). - One <-doneChan wait.

For a synchronous-style API on top of an async primitive, this is essentially free. The major cost is conceptual: a SubmitWait from inside a busy task can deadlock, as we saw.

The Worker Goroutine¶

func worker(task func(), workerQueue chan func()) {
    for task != nil {
        task()
        task = <-workerQueue
    }
}

A worker is one of the smallest non-trivial Go functions you will read. It:

1. Runs its initial task.
2. Waits for the next task on workerQueue.
3. If the next task is nil, exit.
4. Otherwise, run it and repeat.

That is it. The worker has no state. It does not know about maxWorkers, waitingQueue, or stopSignal. It just pumps from a channel.

This minimalism makes worker reaping clean. To reap a worker, the dispatcher just sends nil once on workerQueue. The worker exits. The dispatcher updates its workerCount. Done.

Where is the panic recovery? In some versions, it is wrapped around task() inside the worker. Pseudocode:

func worker(task func(), workerQueue chan func()) {
    for task != nil {
        runWithRecover(task)
        task = <-workerQueue
    }
}

func runWithRecover(task func()) {
    defer func() {
        if r := recover(); r != nil {
            // log or ignore
        }
    }()
    task()
}

The recover lives outside the user's view but inside the worker. So a panicking task does not kill the worker; the worker proceeds to fetch the next task. This is why a single panic does not destroy the pool.

(Earlier versions of the library did not have this recover. If you depend on a very old version, check; you might be one panic away from a crashed program.)

Idle-Worker Reaping in Detail¶

The reaping mechanism uses a single time.Timer shared across all workers. The dispatcher creates one timer in its initialisation and resets it on each iteration of the select block.

The logic:

on each select iteration:
  if dispatched a new task:
    idle = false
  elif timer fired:
    if idle and workerCount > 0:
      reap one worker
      workerCount--
    idle = true
    timer.Reset(idleTimeout)

The two-state machine means one reaping per double-timeout window. The first timeout sets idle = true. If no task arrives before the second timeout, one worker is reaped. If a task arrives between, idle flips back to false and the cycle restarts.

So the effective reaping interval is between idleTimeout and 2 * idleTimeout, depending on phase. With idleTimeout = 2s, that is 2 to 4 seconds.

Why reap only one worker per timeout? Because aggressive reaping would mean: if you have 100 idle workers and a 2-second pause, you would lose all 100 at once and pay the spawn cost for the next 100 on the next burst. Reaping one at a time is gentler — the pool shrinks gradually, and a single late-arriving task only re-spawns one worker.

This is a subtle design choice. A different library might reap differently: - ants reaps all workers older than the expiry duration in one sweep. - A custom pool might keep a "minimum residency" floor that never reaps below.

workerpool's gradual one-at-a-time approach is a reasonable middle ground for unpredictable workloads.

The Internal Queue Structure¶

waitingQueue is a deque.Deque[func()] from github.com/gammazero/deque (same author). A deque is a double-ended queue, implemented as a ring buffer that grows as needed.

Why a deque and not a slice or a linked list?

• Slice: appending is fast, but removing from the front (q = q[1:]) holds the original backing array hostage. With many tasks, you would accumulate garbage. The library wants FIFO with cheap pop-front.
• Linked list: allocates per node. Cache-unfriendly. Each Push is an allocation.
• Deque (ring buffer): amortised O(1) push and pop on both ends; reuses memory; cache-friendly.

The deque from gammazero/deque doubles its capacity when full (like a slice) and halves when nearly empty (unlike a slice). So queue memory does not stick around forever after a burst.

Push appends to the back. Pop removes from the front. The dispatcher uses both: it pushes new tasks that could not find an idle worker, and it pops from the front when ready to dispatch a queued task.

WaitingQueueSize reads p.waiting, an atomic counter that mirrors waitingQueue.Len(). It is updated atomically inside the dispatcher whenever the deque size changes. So reading from outside is cheap.

Stop and StopWait Implementation¶

func (p *WorkerPool) Stop() {
    p.stop(false)
}

func (p *WorkerPool) StopWait() {
    p.stop(true)
}

func (p *WorkerPool) stop(wait bool) {
    p.stopLock.Lock()
    if p.stopped {
        p.stopLock.Unlock()
        <-p.stoppedChan
        return
    }
    p.stopped = true
    p.wait = wait
    close(p.taskQueue) // signal dispatcher
    p.stopLock.Unlock()

    <-p.stoppedChan
}

Things to notice:

1. Lock-protected idempotency. Both Stop and StopWait go through stop(bool). The lock ensures only the first caller actually triggers shutdown; later callers see stopped == true and just wait on stoppedChan.
2. Closing taskQueue is the shutdown signal. Once the dispatcher reads from a closed channel with ok = false, it breaks out of its loop.
3. wait flag carries intent. The dispatcher reads p.wait after the loop break to decide whether to drain queued tasks or skip them.
4. <-p.stoppedChan is the synchronisation. Both methods block until the dispatcher's deferred close(p.stoppedChan) runs. So when Stop or StopWait returns, the pool is provably done.

Why close taskQueue instead of a separate channel? Because the dispatcher already has taskQueue in its select. Closing it gives the dispatcher a clean signal in the existing code path: case task, ok := <-p.taskQueue: if !ok { break Loop }. No extra channel needed.

The subtle ordering for Submit after Stop. Imagine:

Goroutine A: pool.Submit(f) — blocked on p.taskQueue <- f
Goroutine B: pool.Stop() — about to close(p.taskQueue)

If B closes taskQueue while A is mid-send, Go panics with "send on closed channel". The library handles this by:

1. Setting stopped = true while holding stopLock.
2. Having Submit check stopped before sending (in some versions).
3. Or accepting the rare panic and recovering in the dispatcher.

The exact handling varies. In any case, the visible behaviour is "submits after Stop are dropped". The library makes this safe under concurrent access.

(Side note: this is one of the subtle areas where reading the actual source pays off, because the safety story is not obvious from the API contract alone.)

Pause Semantics¶

Pause(ctx) is the newest addition to the API. Simplified:

func (p *WorkerPool) Pause(ctx context.Context) {
    p.Submit(func() {
        <-ctx.Done()
    })
    for i := 0; i < p.maxWorkers-1; i++ {
        p.Submit(func() {
            <-ctx.Done()
        })
    }
}

Yes — the implementation is "submit maxWorkers blocking tasks that wait for the context". This consumes every worker slot. New submissions queue up behind them. When ctx cancels, all workers return, the queue drains.

This is a beautifully elegant hack. No new internal state, no new fields, no API explosion. The pause is implemented entirely with what was already there.

The downside: Pause consumes worker slots. If your maxWorkers is 4 and you call Pause, you have 4 tasks running (each waiting on the ctx). Other queued tasks cannot dispatch until those 4 unblock. The "pause" is really "saturate the pool with blocking sentinels".

This has implications:

• If you Pause(ctx) and then Stop(), the dispatcher needs to deliver poison pills, but workers are stuck waiting on ctx.Done(). Workers do not run the next iteration of their for task := <-workerQueue loop because they are inside the sentinel function. The library typically handles this by canceling the context internally on shutdown.
• WaitingQueueSize during a pause includes the sentinel tasks if they haven't been dispatched yet — confusing for observability.

These rough edges are why Pause is sometimes considered "use sparingly". For most use cases it works fine; for high-stakes coordinated shutdowns, you might prefer to manage pause-equivalents at the application layer.

Synchronisation Primitives Used¶

A list of all sync primitives in the library:

1. Channels (taskQueue, workerQueue, stoppedChan, stopSignal). The dominant coordination mechanism.
2. sync.Mutex (stopLock). Protects the shutdown flags.
3. atomic.Int32 (or int32 with sync/atomic) for waiting. Exposed via WaitingQueueSize.

That is it. No sync.WaitGroup, no sync.Cond, no sync.Once. The whole design is channel-first, with a tiny lock and a tiny atomic.

This is consistent with the idiom "share memory by communicating, not communicate by sharing memory". The pool's shared state lives in channels; the only truly shared variable is the atomic counter, which is read-only from outside.

Compare to ants, which uses a lot more locking — sync.Mutex around worker allocation, condition variables for blocking pool waits. workerpool's channel-centric design is cleaner but limits some kinds of optimisation.

The Cost of Each Operation¶

A breakdown of the runtime cost of each public method, in nanoseconds, on a modern x86 machine.

These numbers are rough but realistic. The dominant per-task cost is in Submit; everything else is either constant (instant) or proportional to work.

A pool can sustain roughly 5-10 million Submits per second from a single goroutine before the dispatcher becomes a bottleneck. Beyond that, you would batch or use a different library.

Comparison with a Hand-Rolled chan func() Pool¶

The minimal hand-rolled pool:

func RunPool(maxW int, tasks []func()) {
    ch := make(chan func())
    var wg sync.WaitGroup
    for i := 0; i < maxW; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for f := range ch {
                f()
            }
        }()
    }
    for _, t := range tasks {
        ch <- t
    }
    close(ch)
    wg.Wait()
}

This is 12 lines, all goroutine-creation and a channel. Pros:

• Zero dependencies.
• Easy to read.
• One less channel hop (no dispatcher).

Cons compared to workerpool:

• Producer blocks when workers busy. If you submit faster than workers drain, you block. That can be a feature (backpressure) or a bug (deadlock with self-submitting tasks).
• Workers live forever. All maxW workers stay around until close(ch). With no idle reaping, the pool holds maxW * stackSize bytes of memory permanently.
• No SubmitWait for free. You write the done-channel pattern yourself.
• No shutdown drain vs discard distinction. Closing ch always drains.
• No queue depth observability. You would have to add an external counter.
• No panic recovery. A panicking task kills its worker; you have one fewer worker for the rest of the run.

workerpool makes all of these work out of the box. The price is one extra channel hop per submit (~50ns) and one extra dispatcher goroutine.

Use the hand-rolled pool when: you need zero dependencies, your workload is bounded and predictable, you don't need idle reaping, and you're comfortable rewriting it as needs evolve.

Use workerpool when: you want the standard behaviour without writing it; you want SubmitWait for free; you value the stable API.

Comparison with ants¶

panjf2000/ants is the other popular pool library. It is more configurable and (often) faster. The differences:

Both libraries are battle-tested. The choice is mostly about what features you need.

Choose ants when: you need runtime resize, typed task args, faster throughput, or per-pool tuning of the idle reaper.

Choose workerpool when: you want minimal code, stable API, and the configurability is irrelevant.

A common migration path: start with workerpool for its simplicity, switch to ants when you start writing too many work-arounds.

Comparison with tunny¶

Jeffail/tunny solves a different problem: request-response with stateful workers.

pool := tunny.NewFunc(4, func(payload interface{}) interface{} {
    // process payload, return result
    return result(payload)
})

result := pool.Process(input)

Tunny's submit returns a result. Workers can have per-worker state via tunny.NewCallback. Workers block on input — there is no internal queue; the pool blocks the caller until a worker is free.

Choose tunny when: - Each worker needs initialised state (compiled regex, DB connection, ML model). - You want request-response semantics and a bounded synchronous interface.

Choose workerpool when: - Tasks are stateless func() calls. - You want async submission with an unbounded queue.

Many services use both: tunny for stateful CPU-bound work (image encoding, video transcoding) and workerpool for stateless I/O fan-out.

Dynamic Scaling Considerations¶

workerpool does not support changing maxWorkers after construction. This is a limitation; we'll list the trade-offs.

Why not allow resize?¶

The dispatcher logic depends on maxWorkers. Changing it mid-flight would require: - Adding a lock around workerCount < maxWorkers (currently lock-free). - Handling the case where the new max is lower than current workers (drain them gracefully?). - API surface area for the resize operation (timing, blocking vs non-blocking).

The library's author chose simplicity over flexibility. If you need dynamic resize, you use ants or build a wrapper.

A simple wrapper for dynamic resize¶

type ResizablePool struct {
    mu      sync.Mutex
    current *workerpool.WorkerPool
    maxW    int
}

func NewResizablePool(max int) *ResizablePool {
    return &ResizablePool{
        current: workerpool.New(max),
        maxW:    max,
    }
}

func (rp *ResizablePool) Resize(newMax int) {
    rp.mu.Lock()
    defer rp.mu.Unlock()
    old := rp.current
    rp.current = workerpool.New(newMax)
    rp.maxW = newMax
    go old.StopWait() // drain the old pool in the background
}

func (rp *ResizablePool) Submit(f func()) {
    rp.mu.Lock()
    p := rp.current
    rp.mu.Unlock()
    p.Submit(f)
}

This creates a new pool on resize, drains the old one in the background, and switches Submit to use the new pool. Caveats:

• Tasks already queued on the old pool keep running on the old pool's workers until it drains.
• The lock around current adds tiny overhead to every Submit.
• Two pools exist transiently during resize.

For occasional resize (every few minutes), this works fine. For high-frequency resize (every second), the cost adds up.

Adaptive sizing¶

For services that want pool size to track load, an outer controller adjusts maxWorkers periodically:

ticker := time.NewTicker(30 * time.Second)
for range ticker.C {
    target := computeTarget()
    if abs(target - rp.maxW) > threshold {
        rp.Resize(target)
    }
}

Where computeTarget looks at queue depth, downstream latency, CPU utilisation, etc. We will detail this in the professional file.

Custom Pool Patterns¶

When workerpool is not enough, here are five patterns for rolling your own.

Pattern 1: Bounded queue with explicit backpressure¶

type BoundedPool struct {
    sem chan struct{}
    pool *workerpool.WorkerPool
}

func NewBoundedPool(maxW, queueCap int) *BoundedPool {
    return &BoundedPool{
        sem: make(chan struct{}, queueCap),
        pool: workerpool.New(maxW),
    }
}

func (bp *BoundedPool) Submit(ctx context.Context, f func()) error {
    select {
    case bp.sem <- struct{}{}:
        bp.pool.Submit(func() {
            defer func() { <-bp.sem }()
            f()
        })
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

This wraps workerpool with a hard cap on queued tasks. Submitters block (or context-cancel) when the cap is reached.

Pattern 2: Priority pool¶

Two underlying pools, one for high-priority tasks, one for low.

type PriorityPool struct {
    high, low *workerpool.WorkerPool
}

func (pp *PriorityPool) SubmitHigh(f func()) {
    pp.high.Submit(f)
}

func (pp *PriorityPool) SubmitLow(f func()) {
    pp.low.Submit(f)
}

The high pool has, say, 16 workers; the low pool has 4. High-priority work cannot be starved. The trade-off: total worker count is the sum, not shared.

For real priority (single queue with priority order), you would need a heap-based queue and a custom dispatcher.

Pattern 3: Per-key serialisation¶

Some workloads need "all tasks for the same key run sequentially". Example: per-user actions where two writes for the same user must not race.

type KeyedPool struct {
    pool *workerpool.WorkerPool
    mu   sync.Mutex
    locks map[string]*sync.Mutex
}

func (kp *KeyedPool) Submit(key string, f func()) {
    kp.pool.Submit(func() {
        kp.mu.Lock()
        m, ok := kp.locks[key]
        if !ok {
            m = &sync.Mutex{}
            kp.locks[key] = m
        }
        kp.mu.Unlock()
        m.Lock()
        defer m.Unlock()
        f()
    })
}

A task acquires a key-specific mutex before running. Tasks with different keys run in parallel; same-key tasks serialise. Memory leak warning: the locks map grows without bound; you'd want a periodic cleanup or sharded design.

Pattern 4: Result-bearing tasks¶

type Result struct {
    Value interface{}
    Err   error
}

type ResultPool struct {
    pool *workerpool.WorkerPool
}

func (rp *ResultPool) Submit(f func() (interface{}, error)) <-chan Result {
    ch := make(chan Result, 1)
    rp.pool.Submit(func() {
        v, err := f()
        ch <- Result{Value: v, Err: err}
    })
    return ch
}

The caller gets a channel back; reading from it gives the task's result. Pattern: result := <-pool.Submit(myFunc). This is mid-way between fire-and-forget and SubmitWait.

Pattern 5: Pool with metrics¶

type MetricsPool struct {
    pool    *workerpool.WorkerPool
    submits prometheus.Counter
    runs    prometheus.Counter
    panics  prometheus.Counter
    dur     prometheus.Histogram
}

func (mp *MetricsPool) Submit(f func()) {
    mp.submits.Inc()
    mp.pool.Submit(func() {
        start := time.Now()
        defer func() {
            mp.runs.Inc()
            mp.dur.Observe(time.Since(start).Seconds())
            if r := recover(); r != nil {
                mp.panics.Inc()
                log.Printf("task panic: %v", r)
            }
        }()
        f()
    })
}

This is the foundation of every production wrapper. Every submit gets counted, every run gets timed, every panic gets logged and counted.

Adapting the Library for Your Needs¶

Three levels of adaptation:

1. Wrap¶

Build a struct that contains a *workerpool.WorkerPool and adds methods. All the patterns above are wraps. This is the right approach 95% of the time.

2. Fork¶

Copy the library into your repo and edit. Useful if you need to change the idle timeout, add prometheus hooks directly, or remove panic recovery.

When forking, mark the file clearly:

// Forked from github.com/gammazero/workerpool v1.1.3
// Modifications:
// - idle timeout changed from 2s to 10s for our bursty workload
// - panic recovery removed (we handle panics at the call site)

Diff your fork against the upstream periodically to pick up bug fixes.

3. Replace¶

Write your own from scratch. Cost: 200-500 lines of well-tested Go and a 1-2 week project. Benefit: total control. Worth it for libraries that ship to many users (e.g., a database driver bundling its own pool), or for performance-critical paths where every cycle matters.

Memory Layout and Allocations¶

A breakdown of what allocates and where.

Pool creation: - One WorkerPool struct (~100 bytes). - Two unbuffered chan func() (~96 bytes each). - One chan struct{} (stoppedChan, ~96 bytes). - One deque.Deque[func()] (lazy: ~0 bytes initially). - One time.Timer (~96 bytes). - One dispatcher goroutine (~2 KB stack).

Total at creation: roughly 2.5 KB.

Per-task: - The user's closure (varies; ~50-200 bytes depending on captures). - For SubmitWait, a wrapper closure (~50 bytes) and a done channel (~96 bytes).

Per-worker (active): - A worker goroutine (~2 KB stack, can grow).

Queue growth: - Deque starts at a small capacity (typically 16) and doubles as needed. - Memory is reclaimed when the queue shrinks past a threshold.

For a pool with maxWorkers = 100 and a steady-state queue of 1,000 tasks: roughly 2.5 KB (pool) + 100 × 2 KB (workers) + 16 KB (deque slots) + 1,000 × 100 bytes (closures) = ~310 KB. Trivial for a single pool; nontrivial for hundreds of pools.

Allocation rate: each Submit allocates one closure (the user's func()) and possibly bumps the deque. The closure allocation is on the user side; the library itself does not allocate per Submit (after dispatcher startup). So the GC pressure of using workerpool is dominated by your closures, not the library's overhead.

This is why you sometimes see advice like "use ants.PoolWithFunc" for high-throughput pools: it avoids the closure allocation by binding the function once and accepting an interface{} per task.

Race Conditions That Could Exist¶

The library is well-tested. Running go test -race ./... on the repo passes cleanly. But here are the kinds of races a worker pool design must guard against, and how workerpool does it:

Race 1: Submit during Stop¶

A goroutine calls Submit(f) while another calls Stop. Possible outcomes: - The submitter's send completes before Stop closes taskQueue: task runs (or queues then runs / discards based on Stop/StopWait). - The submitter's send happens during close: would panic "send on closed channel" in naive code.

workerpool handles this by: (a) the stopLock ensures only one shutdown closes the channel; (b) some versions check Stopped inside Submit before sending. In the worst case, a race could lead to a panic that the library's recover or defer cleanup catches.

In practice, you mitigate by stopping producers before stopping the pool — the application-level lifecycle ordering.

Race 2: Multiple Stop callers¶

Goroutines A and B both call Stop at the same time. Without stopLock, both would try to close taskQueue, causing a "close of closed channel" panic.

The library protects with if p.stopped { ... return } inside stop.

Race 3: WaitingQueueSize returns stale value¶

The atomic counter is updated by the dispatcher when it pushes/pops the deque. An external reader sees a snapshot. If a Submit and a dispatcher iteration happen between read and use, the value is stale.

This is not a race in the strict sense — there is no data race — but it is a logical race: the value cannot be used for synchronisation. The library docs (implicitly) acknowledge this; treat it as telemetry only.

Race 4: Pause and Stop concurrently¶

A goroutine calls Pause(ctx); another calls Stop. The pause submits sentinel tasks; Stop closes the task queue. If Stop wins, the sentinel tasks may or may not have made it onto the queue. The visible behaviour is: pause is irrelevant because the pool is shutting down anyway.

The library handles this gracefully by virtue of Stop overriding everything.

Race 5: Reading the pool through a stale pointer¶

If you swap pool references (as in the ResizablePool example), readers may have the old pointer cached. Lock-free reads must use atomic loads. If you use sync.Mutex around the pointer, lock around every access. Mixing styles is dangerous.

This is not a workerpool issue per se; it is a general Go concurrency issue when you wrap mutable state.

Tricky Internals¶

The dispatcher prioritises the queue¶

Look at the loop:

if p.waitingQueue.Len() != 0 {
    if !p.processWaitingQueue() {
        break Loop
    }
    continue
}
select {
    case task := <-p.taskQueue: ...
    case <-timeout.C: ...
}

The queue check is outside the select. So if there are tasks queued, the dispatcher does not even consider new submissions — it drains the queue first, one at a time per iteration.

Practical effect: under heavy load, the queue grows and new submits are blocked (they cannot reach the dispatcher because it is processing queue). That sounds like backpressure, but it is not — the submitter is blocked on chan send, not by a queue cap.

Once the queue clears, new submits flow again.

Idle timer fires even with work¶

The timer fires regardless of whether the dispatcher has work to do. The case <-timeout.C branch is reached only when the select chooses it — typically when the dispatcher is in the select block (idle). If the dispatcher is busy processing queued tasks, the timer is still running but not consulted. When the queue clears and the dispatcher hits the select, the timer may already have fired and the channel has a buffered value.

The idle flag mechanism handles this: it gets set to false whenever a task arrives, so a timer-fire after recent activity does not trigger reaping.

Worker count can briefly exceed maxWorkers¶

There is no atomic check on workerCount. The dispatcher is single-threaded (it is the only goroutine writing workerCount), so this is safe. But if you fork the library and add multi-threaded dispatcher work, you must guard workerCount.

Poison pills are nil¶

The dispatcher signals worker exit by sending nil on workerQueue. The worker's for task != nil loop exits. No special "stop" signal needed.

This is clever, but it means you must never Submit(nil) — the worker would exit unexpectedly. Modern versions check for nil in Submit and return early. Older versions might let it through, causing a worker to vanish.

Stop with running tasks waits for them¶

When you call Stop, the dispatcher closes taskQueue and breaks the loop. It then sends poison pills to workers. But workers may be in the middle of running a task; they do not poll workerQueue until the current task returns. So Stop effectively waits for all running tasks.

There is no way to interrupt a running task from outside. You must use a context.Context inside the task.

Edge Cases at the Library Level¶

New(0) and New(-1)¶

Both are normalised to New(1). No error, no panic. Friendly but ambiguous; some users would prefer strictness.

Submit(nil)¶

Modern versions: silently dropped (early return in Submit). Older versions: queued; worker tries nil(); library's recover catches the panic; worker survives.

Either way, do not submit nil.

SubmitWait(nil)¶

Modern versions: early return; the done channel never closes. But since you never <-done, you do not block. Older versions: same as Submit(nil) — wrapped, queued, panics, recovered. The wrapper still closes done, so SubmitWait returns.

Either way: do not submit nil.

Stop followed by New on the same pool¶

Stop does not magically transform the pool back into a fresh one. A stopped pool is dead. You need a new workerpool.New(N) to get a fresh one.

Pool sized larger than goroutine limit¶

Go's default goroutine limit is effectively unlimited (constrained only by memory). But runtime.SetMaxThreads and similar could limit you in pathological setups. A pool with maxWorkers = 1_000_000 would happily try to create a million worker goroutines if needed. You would OOM or hit thread limits long before any "pool limit".

The library does not warn you. Pick reasonable values.

Calling Stop while paused¶

Already covered. Stop overrides pause.

StopWait with an empty queue¶

Fast — workers (if any) finish their current task, exit, and StopWait returns. Effectively the same as Stop in this case.

Goroutine leak from forgotten pool¶

If you pool := workerpool.New(N) and never call Stop/StopWait, the dispatcher goroutine lives forever. With idle reaping, worker goroutines come and go, but the dispatcher persists.

For long-running processes, a forgotten pool is a steady drip of goroutines (1 per forgotten pool). With pools per request, this accumulates dangerously.

Lint rules: some linters can catch missing StopWait. Consider one.

Debugging Pool Internals¶

Several techniques for inspecting a pool in trouble.

Goroutine dump¶

buf := make([]byte, 1<<20)
n := runtime.Stack(buf, true)
fmt.Println(string(buf[:n]))

You will see something like:

goroutine 1 [running]:
main.main()
    /path/to/main.go:42

goroutine 17 [select]:
github.com/gammazero/workerpool.(*WorkerPool).dispatch(0xc000010100)
    /path/to/workerpool.go:120
created by github.com/gammazero/workerpool.New
    /path/to/workerpool.go:38

goroutine 18 [chan receive]:
github.com/gammazero/workerpool.worker(0x0, 0xc000018180)
    /path/to/workerpool.go:200
created by github.com/gammazero/workerpool.(*WorkerPool).dispatch
    /path/to/workerpool.go:160

The dispatcher is in [select] state — waiting in its main loop. Workers are in [chan receive] — waiting for the next task. A pool with no work shows exactly this pattern.

If you see workers in [running] or any other state, that is a task currently executing. If the program is hung, the workers' stack traces tell you exactly where each task is stuck.

pprof goroutine profile¶

In a service with net/http/pprof enabled:

curl http://localhost:6060/debug/pprof/goroutine?debug=2

Same data as runtime.Stack but accessible remotely. Aggregate counts in the debug=0 mode for big services.

Custom pool.Stats() wrapper¶

type Pool struct {
    p *workerpool.WorkerPool
    submitted int64
    completed int64
}

func (p *Pool) Stats() string {
    return fmt.Sprintf("queue=%d submitted=%d completed=%d stopped=%v",
        p.p.WaitingQueueSize(),
        atomic.LoadInt64(&p.submitted),
        atomic.LoadInt64(&p.completed),
        p.p.Stopped(),
    )
}

Expose this on an admin endpoint or log it periodically.

Stuck StopWait¶

If StopWait hangs forever: 1. Dump goroutines. 2. Find the workers. Look at where they are running. 3. If they are stuck on I/O, a missing context cancellation is the culprit. 4. If they are stuck on a channel or mutex, you have a deadlock in user code.

The library itself does not deadlock unless user tasks misbehave.

Reproducing pool behaviour locally¶

A standalone tool:

func main() {
    pool := workerpool.New(*flag.Int("c", 4, "concurrency"))
    defer pool.StopWait()

    for i := 0; i < 10; i++ {
        i := i
        pool.Submit(func() {
            time.Sleep(time.Second)
            fmt.Println(i)
        })
        fmt.Println("after submit", i, "queue:", pool.WaitingQueueSize())
    }
}

Run with -c 2, -c 4, -c 10. Observe queue size right after each submit. You will see the queue grow as workers cannot keep up, then drain.

Common Senior-Level Mistakes¶

1. Trusting WaitingQueueSize for synchronisation¶

It is a counter, not a barrier. Use it for metrics; for "all done" semantics, use StopWait or a WaitGroup.

2. Forking without diffing¶

Forking the library is fine. Failing to track upstream changes is not. Periodically diff and merge bug fixes.

3. Reusing closures across submissions¶

f := func() { use(state) }
for _, s := range states {
    state = s   // RACE — modify captured var while previous task runs
    pool.Submit(f)
}

Each submission's f reads state at execution time, by which point state may have changed. Always create the closure per iteration with a captured value.

4. Forgetting that nil submissions reach workers¶

In old versions of workerpool, Submit(nil) is queued. The worker runs nil(), panics, library recovers. The user's intent is lost. Validate inputs.

5. Misusing Pause¶

Pause consumes worker slots with sentinel tasks. If you Pause and then submit a task expecting "queued tasks dispatch when I resume", you are right — but you also tied up maxWorkers slots in pause sentinels. The pool is fully utilised during pause; new submits queue.

6. Building intricate two-pool dependencies¶

Two pools where one's tasks submit to the other are easy to deadlock. Always ensure that one pool's queue cannot block on the other's.

7. Tuning maxWorkers based on aspirational thinking¶

Setting maxWorkers = 1000 "in case we need it" is a code smell. It implies you do not know the actual concurrency requirement. Measure, then set.

8. Treating the library as feature-rich¶

Looking for "how do I do X with workerpool" where X is "retry / priorities / cancel a task". The answer is usually "wrap it" or "use a different library". Stop fighting the library's minimalism.

9. Holding the pool reference past program lifetime¶

In tests, a static pool that lives across test cases is a footgun: state leaks between tests. Prefer per-test pools.

10. Believing Submit is wait-free¶

It is fast, but it sends on an unbuffered channel. If the dispatcher is in a tight loop processing queue, Submit blocks until the next iteration. Microseconds at worst, but not "wait-free".

Test¶

A senior-level test suite exercises internals:

package main

import (
    "context"
    "sync"
    "sync/atomic"
    "testing"
    "time"

    "github.com/gammazero/workerpool"
)

func TestIdleWorkerReaping(t *testing.T) {
    pool := workerpool.New(10)
    defer pool.StopWait()

    // Saturate the pool.
    for i := 0; i < 10; i++ {
        pool.Submit(func() { time.Sleep(50 * time.Millisecond) })
    }

    n0 := runtime.NumGoroutine()
    t.Logf("during work: %d goroutines", n0)

    // Wait longer than idle timeout.
    time.Sleep(3 * time.Second)

    n1 := runtime.NumGoroutine()
    t.Logf("after idle: %d goroutines", n1)

    if n1 >= n0 {
        t.Errorf("expected workers to be reaped, got %d -> %d", n0, n1)
    }
}

func TestSubmitDoesNotBlockUnboundedly(t *testing.T) {
    pool := workerpool.New(1)
    defer pool.StopWait()

    // Block the worker forever.
    pool.Submit(func() { time.Sleep(time.Hour) })

    // Try to submit 1000 more in 1 second.
    done := make(chan struct{})
    go func() {
        for i := 0; i < 1000; i++ {
            pool.Submit(func() {})
        }
        close(done)
    }()

    select {
    case <-done:
        t.Log("all 1000 submits completed despite worker blocked")
    case <-time.After(time.Second):
        t.Fatal("submits did not complete within 1s")
    }
}

func TestStopDiscardsQueuedTasks(t *testing.T) {
    pool := workerpool.New(1)
    var done int64

    // Queue 100 tasks; only some will run.
    pool.Submit(func() { time.Sleep(100 * time.Millisecond) }) // saturate
    for i := 0; i < 100; i++ {
        pool.Submit(func() { atomic.AddInt64(&done, 1) })
    }

    pool.Stop()

    final := atomic.LoadInt64(&done)
    if final == 100 {
        t.Log("edge case: all tasks finished before Stop")
    } else if final > 0 {
        t.Logf("%d/100 tasks ran before Stop discarded the rest", final)
    } else {
        t.Log("Stop discarded all queued tasks")
    }
}

func TestStopWaitWaitsForAllTasks(t *testing.T) {
    pool := workerpool.New(2)
    var done int64
    for i := 0; i < 100; i++ {
        pool.Submit(func() {
            time.Sleep(10 * time.Millisecond)
            atomic.AddInt64(&done, 1)
        })
    }
    pool.StopWait()
    if got := atomic.LoadInt64(&done); got != 100 {
        t.Fatalf("StopWait did not drain: %d/100", got)
    }
}

func TestPauseAndResume(t *testing.T) {
    pool := workerpool.New(4)
    defer pool.StopWait()

    ctx, cancel := context.WithCancel(context.Background())
    pool.Pause(ctx)

    var done int64
    for i := 0; i < 4; i++ {
        pool.Submit(func() {
            atomic.AddInt64(&done, 1)
        })
    }

    time.Sleep(50 * time.Millisecond)
    if atomic.LoadInt64(&done) > 0 {
        t.Fatal("task ran during pause")
    }

    cancel()
    time.Sleep(100 * time.Millisecond)
    if atomic.LoadInt64(&done) != 4 {
        t.Fatalf("tasks did not run after resume: %d/4", atomic.LoadInt64(&done))
    }
}

func TestConcurrentSubmits(t *testing.T) {
    pool := workerpool.New(8)
    defer pool.StopWait()

    var done int64
    const submitters = 100
    const perSubmitter = 1000

    var wg sync.WaitGroup
    for s := 0; s < submitters; s++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for i := 0; i < perSubmitter; i++ {
                pool.Submit(func() { atomic.AddInt64(&done, 1) })
            }
        }()
    }
    wg.Wait()
    pool.StopWait()

    if got := atomic.LoadInt64(&done); got != submitters*perSubmitter {
        t.Fatalf("expected %d, got %d", submitters*perSubmitter, got)
    }
}

func TestSubmitAfterStop(t *testing.T) {
    pool := workerpool.New(2)
    pool.StopWait()

    var ran int64
    pool.Submit(func() { atomic.AddInt64(&ran, 1) })
    if ran != 0 {
        t.Fatal("task ran after StopWait")
    }
}

Run with go test -race -timeout 30s ./....

Tricky Questions¶

Q: What does the dispatcher do during a Submit from a closed pool?¶

If the user calls Submit after Stop, the library typically checks Stopped() first and returns. If somehow a Submit reaches the closed taskQueue, the send would panic. The library's defer (in some versions) catches this.

Q: How is workerCount updated?¶

Only inside the dispatcher loop. The dispatcher increments on spawn and decrements on reap. There is no lock — the dispatcher is the only goroutine writing.

Q: What if a worker panics and the dispatcher does not know?¶

In modern versions, a panic in task() is recovered inside the worker's wrapper. The worker stays alive. The dispatcher does not need to know.

In hypothetical older versions without that recover, a panic would kill the worker without notifying the dispatcher. The dispatcher would try to send the next task on workerQueue, but the worker is gone. The send would block (no receiver). Deadlock.

The library's recover is therefore not just a niceness — it is critical for correctness.

Q: Why an unbuffered workerQueue?¶

So the dispatcher gets immediate feedback whether a worker is ready. With select { case workerQueue <- task: default: }, the dispatcher knows in one nanosecond whether a worker is waiting. With a buffer, the dispatcher would just enqueue regardless of worker state.

Q: Could you replace waitingQueue with a channel?¶

Conceptually yes — a buffered channel of arbitrary size could serve as the queue. But Go channels do not auto-grow; you must pick a fixed buffer size. An unbounded deque is more memory-efficient over time (shrinking when sparse).

Q: What's the GC impact of long-lived pools?¶

The pool itself is small. The queue's deque keeps growing then shrinking; each growth allocates a larger backing array, each shrink frees it. So there is some GC churn under varying loads.

For pools with very stable load, GC impact is minimal — once the deque has reached its working size, it stays there.

Q: Could workerCount go negative on aggressive reaping?¶

No. The dispatcher decrements only after confirming a reap, and killIdleWorker is a synchronous send of nil to a worker that is waiting on workerQueue. The worker reads the nil and exits; the dispatcher knows the worker is gone.

Q: What if the timer fires during shutdown?¶

The shutdown path breaks out of the loop and ignores further timer fires. The deferred timeout.Stop() cleans up.

Q: How would you add per-pool metrics without forking?¶

Wrap. Every Submit increments a counter before calling pool.Submit(wrapped). The wrapped closure decrements (or increments a "completed" counter) on exit. The pool itself is unaware.

Q: How would you add priority?¶

You cannot, without forking or replacing. The library's deque is FIFO. To support priority, you would need a heap-based queue inside the dispatcher.

Cheat Sheet¶

WorkerPool struct (essentials):
  maxWorkers    int
  taskQueue     chan func()       // submitters write here
  workerQueue   chan func()       // workers read here
  waitingQueue  Deque[func()]     // overflow when no worker ready
  stoppedChan   chan struct{}     // closed at end of dispatcher
  stopped       bool              // shutdown initiated?
  wait          bool              // drain (true) or discard (false)?

Dispatcher loop:
  if queue non-empty: forward head to a worker, repeat
  else select:
    case task := <-taskQueue:
      try workerQueue <- task else spawn or enqueue
    case <-timer.C:
      if idle and workers>0: reap one
      reset timer; idle = true

Worker loop:
  for task != nil:
    runWithRecover(task)
    task = <-workerQueue

Shutdown:
  close(taskQueue) — dispatcher exits its loop
  if wait: drain queue
  send nil to each worker — they exit
  close(stoppedChan) — Stop/StopWait return

Memorising this picture is the goal of the senior file.

Self-Assessment Checklist¶

• I can sketch the WorkerPool struct fields from memory.
• I can trace a Submit from caller to worker, step by step.
• I can explain why there are two channels (taskQueue and workerQueue).
• I know how the dispatcher prioritises the waiting queue over new submits.
• I understand the two-state idle flag mechanism for reaping.
• I can compare workerpool to ants and tunny with specific feature differences.
• I can implement a BoundedPool wrapper.
• I can implement a PriorityPool with two underlying pools.
• I can debug a stuck pool via runtime.Stack.
• I can explain why Pause consumes worker slots.
• I can decide whether to fork, wrap, or replace for a given need.

If all are yes, you are ready for the professional file.

Summary¶

The senior file is the source code tour. The library is small enough that you can hold it all in your head: one dispatcher goroutine, N worker goroutines, a deque for overflow, and channels for coordination. The dispatcher's select loop is the only really intricate piece, and even that is just three cases.

Three things to internalise from this level:

1. Two channels, one dispatcher. The design's whole personality is here. The dispatcher mediates between submitters and workers; the two channels give it the select choice it needs.
2. Workers are stateless. This makes reaping cheap and panic recovery clean.
3. The simplicity is intentional. Every feature you wish were here was deliberately omitted to keep the API minimal.

When the simplicity becomes a constraint, you wrap, fork, or replace. All three are senior-level skills. The professional file picks up here to talk about when to do each, and what production wisdom looks like in real services.

What You Can Build¶

At senior level you can build:

• A custom pool from scratch in a few hundred lines that matches workerpool's contract.
• A wrapped pool with metrics, priorities, per-key serialisation, or bounded queues.
• A forked variant tuned for a specific workload (different idle timeout, different recover strategy).
• A multi-pool architecture for complex pipelines.

You can also read other Go libraries with the same level of comprehension — every networking library, every cache, every queue eventually uses similar patterns.

Further Reading¶

• The library's source: clone and read workerpool.go.
• The ants source for comparison: panjf2000/ants is a great study.
• The tunny source for a different architecture.
• "Concurrency in Go" by Katherine Cox-Buday — the chapters on patterns are foundational.
• The Go runtime's source for chan operations (runtime/chan.go).

Related Topics¶

• Goroutine schedulers — the GMP model that makes workerpool viable.
• Channels in depth — buffered vs unbuffered semantics.
• sync.Mutex vs atomics — when the library uses which.
• time.Timer — the reaper's backbone.

Diagrams and Visual Aids¶

Two-channel dispatcher¶

            +---------+         +-----------+         +----------+
  Submit -> taskQueue +--------->+ dispatch +--------->+ worker(s)|
            +---------+         | select   |         +----------+
                                | block    |              ^
                                +-----+----+              |
                                      |                   |
                                      v                   |
                                +-----+----+              |
                                | spawn or |              |
                                | enqueue  +--------------+
                                | workerQ  |  workerQueue
                                +----------+

Dispatcher state machine¶

                     +-------+
                     | start |
                     +---+---+
                         |
                         v
                  +------+-------+
                  | check waiting |
                  +--+---------+--+
                non-empty|       |empty
                         v       v
                +--------+--+ +--+----------------+
                | dispatch  | | select:           |
                | queued    | |   taskQueue ->    |
                | -> worker | |     route        |
                +--+--------+ |   timer -> reap   |
                   |          +--+----------------+
                   |             |
                   +------+------+
                          |
                          v
                       repeat
                          |
                          | (taskQueue closed)
                          v
                  +-------+-------+
                  | shutdown      |
                  | drain or skip |
                  | poison        |
                  | close stopChan|
                  +---------------+

Worker exit via nil¶

worker: for task != nil:
          runWithRecover(task)
          task = <-workerQueue   // dispatcher sends nil here at shutdown
        return  // goroutine exits

Memory shape at steady state¶

WorkerPool
├── maxWorkers (int)
├── taskQueue (chan func())
├── workerQueue (chan func())
├── waitingQueue [Deque ring buffer]
│       size grows/shrinks dynamically
├── stoppedChan (chan struct{})
├── stopLock (sync.Mutex)
├── stopped (bool)
└── wait (bool)

dispatcher goroutine ── 1 always alive
worker goroutines ── 0..maxWorkers, reaped after ~2s idle

These pictures are what to keep in your head when reading or writing senior-level pool code.

Appendix A: Line-by-Line Walkthrough of dispatch¶

Let us read the dispatcher in detail, line by line. The actual library may differ slightly; we present a faithful approximation.

func (p *WorkerPool) dispatch() {
    defer close(p.stoppedChan)

The deferred close happens when the dispatcher exits, signalling completion to whoever called Stop or StopWait. This is the only way the outside world learns shutdown is done.

    timeout := time.NewTimer(idleTimeout)

Create the idle timer once at startup. It will be reset on every iteration to ensure exactly one tick fires per idleTimeout (~2s) of dispatcher inactivity.

    var workerCount int
    var idle bool

Two local variables. workerCount is the current number of worker goroutines alive. idle tracks whether we've been idle since the last timer tick.

Loop:
    for {

Label Loop is for break Loop later. Go does not have multi-level break by index; named labels are the idiom.

        if p.waitingQueue.Len() != 0 {
            if !p.processWaitingQueue() {
                break Loop
            }
            continue
        }

If there are queued tasks, process one. processWaitingQueue returns false if shutdown is in progress (the task queue is closed and we should exit). Otherwise it pops the front, hands it to a worker, decrements the atomic counter, and returns true. continue restarts the loop — which re-checks the queue. This effectively gives queued tasks priority over new submissions.

        select {
        case task, ok := <-p.taskQueue:
            if !ok {
                break Loop
            }

Read from taskQueue. The ok is false when the channel is closed (by Stop or StopWait). On close, break out and head to shutdown.

            select {
            case p.workerQueue <- task:
            default:
                if workerCount < p.maxWorkers {
                    go worker(task, p.workerQueue)
                    workerCount++
                } else {
                    p.waitingQueue.PushBack(task)
                    atomic.StoreInt32(&p.waiting, int32(p.waitingQueue.Len()))
                }
            }

The inner select is the heart of dispatching: 1. Try to send the task to a waiting worker. If a worker is currently waiting on workerQueue (i.e., its task = <-workerQueue is blocked), this succeeds and the worker runs the task. 2. If no worker is waiting (default): - If we can spawn more workers, do so. The new worker starts with this task. - Otherwise, queue the task in waitingQueue and update the atomic counter.

The default case is the magic. It lets the dispatcher decide immediately whether a worker is available. No timeout, no polling.

            idle = false

A task arrived; we are not idle. This resets the reaping cycle.

        case <-timeout.C:
            if idle && workerCount > 0 {
                if p.killIdleWorker() {
                    workerCount--
                }
            }
            idle = true
            timeout.Reset(idleTimeout)
        }

The timer fired. If we have been idle since the last fire (two consecutive timer ticks with no task), and we have workers, reap one. Then set idle = true (until a task arrives) and reset the timer.

killIdleWorker sends nil to workerQueue. If a worker is waiting, it reads the nil and exits. The send is non-blocking with select { case ... <- nil: default: return false } semantics — if no worker is waiting, do nothing.

    }

End of for loop. We exit here after break Loop.

    // shutdown sequence
    if p.wait {
        p.runQueuedTasks()
    }

If wait was set (i.e., StopWait), drain the remaining queued tasks before tearing down workers. Otherwise (Stop), skip — those tasks are dropped.

runQueuedTasks is a loop over the waiting queue, sending each to a worker (waiting for one to be ready, or spawning one if under cap).

    for workerCount > 0 {
        p.workerQueue <- nil
        workerCount--
    }

Send a nil poison pill to each worker. Workers exit on nil, decrementing the count.

    timeout.Stop()
}

Stop the timer to avoid a goroutine leak from the timer's internal goroutine.

That's the entire dispatcher. About 40 lines. Read it three times.

Appendix B: Line-by-Line Walkthrough of worker¶

func worker(initialTask func(), workerQueue chan func()) {
    for task := initialTask; task != nil; task = <-workerQueue {
        runTask(task)
    }
}

If you want the recovery inside:

func worker(initialTask func(), workerQueue chan func()) {
    for task := initialTask; task != nil; task = <-workerQueue {
        runTask(task)
    }
}

func runTask(task func()) {
    defer func() {
        if r := recover(); r != nil {
            log.Println("worker recovered:", r)
        }
    }()
    task()
}

The worker is barely a function. Its loop:

1. Start with initialTask. Run it (with recovery).
2. Block on workerQueue.
3. If the received value is nil, the loop condition fails and the function returns. Goroutine exits.
4. Otherwise, run the task and repeat.

That is all. Workers do not know about maxWorkers, waitingQueue, stoppedChan, the timer, or even the pool. They just pump from a channel.

This minimalism is the secret to clean teardown: to stop a worker, send nil. To stop all workers, send nil workerCount times.

Appendix C: The processWaitingQueue Helper¶

A simplified version:

func (p *WorkerPool) processWaitingQueue() bool {
    select {
    case task, ok := <-p.taskQueue:
        if !ok {
            return false
        }
        p.waitingQueue.PushBack(task)
    case p.workerQueue <- p.waitingQueue.Front():
        p.waitingQueue.PopFront()
    }
    atomic.StoreInt32(&p.waiting, int32(p.waitingQueue.Len()))
    return true
}

What this does: while waiting tasks exist, the dispatcher hopes to send the front of the queue to a worker. Two outcomes:

• A worker accepts: pop the front, update counter, return true.
• A new task arrives on taskQueue: push it to the back of the queue. (Or close: return false to break out.)

This is the "drain the queue while accepting more" inner loop. It maintains FIFO ordering: the front task always gets dispatched first, but if new tasks arrive while waiting for a worker, they go to the back.

The select blocks until either a worker is ready or a new task arrives. So the dispatcher does not busy-wait; it parks on the channels.

A subtle point: p.waitingQueue.Front() is evaluated to obtain the value to send. The send operation is p.workerQueue <- task. If the send is selected, only then do we pop. If the other case is selected (a new submit), we leave the queue's front intact for the next iteration.

This is one of the trickier patterns in Go. The select with one side being a send of a peeked value works because Go evaluates the case expressions before selecting.

Appendix D: Building a Pool From First Principles¶

To really understand workerpool, build one yourself. We'll do it step by step.

Step 1: The simplest possible pool¶

type Pool struct {
    ch chan func()
}

func NewPool(n int) *Pool {
    p := &Pool{ch: make(chan func())}
    for i := 0; i < n; i++ {
        go func() {
            for f := range p.ch {
                f()
            }
        }()
    }
    return p
}

func (p *Pool) Submit(f func()) {
    p.ch <- f
}

func (p *Pool) Stop() {
    close(p.ch)
}

12 lines. Works. Differences from workerpool:

• All workers spawn immediately (no lazy startup).
• Workers live forever until Stop (no idle reaping).
• Submit blocks when all workers are busy (no queue).
• Stop does not wait for in-flight tasks to finish.
• No SubmitWait, no Stopped, no WaitingQueueSize.

Each of these gaps motivates a feature workerpool provides.

Step 2: Add a queue¶

type Pool struct {
    in    chan func()
    out   chan func()
    queue []func()
    done  chan struct{}
}

func NewPool(n int) *Pool {
    p := &Pool{
        in:   make(chan func()),
        out:  make(chan func()),
        done: make(chan struct{}),
    }
    go p.dispatch()
    for i := 0; i < n; i++ {
        go p.worker()
    }
    return p
}

func (p *Pool) dispatch() {
    for {
        var sendCh chan func()
        var next func()
        if len(p.queue) > 0 {
            sendCh = p.out
            next = p.queue[0]
        }
        select {
        case f, ok := <-p.in:
            if !ok {
                close(p.done)
                return
            }
            p.queue = append(p.queue, f)
        case sendCh <- next:
            p.queue = p.queue[1:]
        }
    }
}

func (p *Pool) worker() {
    for f := range p.out {
        f()
    }
}

func (p *Pool) Submit(f func()) { p.in <- f }
func (p *Pool) Stop()           { close(p.in); <-p.done }

Now we have a queue (via the slice queue) and a dispatcher. Submit never blocks for long — even if all workers are busy, the dispatcher just appends to the queue.

The select trick here is identical to what workerpool does: when the queue is empty, sendCh is nil, and a select on a nil channel blocks forever — so the dispatcher only tries to send when there is something to send.

Step 3: Add lazy workers and idle reaping¶

// (skipping for brevity, but the pattern is: track worker count, spawn lazily,
//  send nil to a worker after N seconds of pool inactivity)

This is roughly 50 more lines. At this point you have re-built workerpool.

Step 4: Add panic recovery¶

func (p *Pool) worker() {
    for f := range p.out {
        func() {
            defer func() { _ = recover() }()
            f()
        }()
    }
}

A few more lines. Now your pool survives a panicking task.

Step 5: Add SubmitWait¶

func (p *Pool) SubmitWait(f func()) {
    done := make(chan struct{})
    p.Submit(func() {
        defer close(done)
        f()
    })
    <-done
}

10 more lines.

Total: about 100-150 lines of code to roughly reproduce workerpool's feature set. The library itself is denser because it handles edge cases, but the core algorithm is the same.

This exercise is worth doing in a notepad even if you do not run it. Building it once gives you a mental model that no amount of reading can replace.

Appendix E: When the Library's Choices Bite¶

A list of specific decisions in workerpool that cause problems in certain workloads.

Choice 1: Hard-coded 2-second idle timeout¶

Bursty workloads with 2.5-second gaps reap workers between bursts and respawn on each one. Workaround: keep-alive heartbeat tasks.

Choice 2: Unbounded queue¶

Under load spike, memory grows. Workaround: semaphore in front of Submit.

Choice 3: No per-task error¶

Errors must be captured by closure. Workaround: build a typed wrapper that returns (value, error) via channels.

Choice 4: func() only¶

No typed arguments. Workaround: closures capture them. Cost: closure allocation per submit.

Choice 5: No resize¶

maxWorkers is fixed for the pool's lifetime. Workaround: create new pool, drain old. Cost: complexity.

Choice 6: No priority¶

FIFO only. Workaround: two pools, or a priority queue + dispatcher of your own.

Choice 7: Single dispatcher¶

If submissions are dominated by a single goroutine, the dispatcher hops fine. At extreme rates (millions of submits per second from many goroutines), the dispatcher's select could become a bottleneck. Workaround: shard into multiple pools.

Choice 8: No introspection of running tasks¶

You cannot ask "what are my workers running right now?". Workaround: per-task instrumentation.

Each workaround adds code. When the workarounds outweigh the library's value, switch.

Appendix F: Performance Microbenchmarks¶

A few representative benchmarks. Numbers from a recent x86 laptop with Go 1.22; your mileage will vary.

func BenchmarkSubmitNoop(b *testing.B) {
    pool := workerpool.New(8)
    defer pool.StopWait()

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        pool.Submit(func() {})
    }
}

func BenchmarkSubmitWaitNoop(b *testing.B) {
    pool := workerpool.New(8)
    defer pool.StopWait()

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        pool.SubmitWait(func() {})
    }
}

func BenchmarkRawGoroutineNoop(b *testing.B) {
    var wg sync.WaitGroup
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        wg.Add(1)
        go func() {
            wg.Done()
        }()
    }
    wg.Wait()
}

func BenchmarkBufferedChannelNoop(b *testing.B) {
    ch := make(chan func(), 1024)
    var wg sync.WaitGroup
    for w := 0; w < 8; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for f := range ch {
                f()
            }
        }()
    }

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        ch <- func() {}
    }
    close(ch)
    wg.Wait()
}