gammazero/workerpool — Middle Level¶

Table of Contents¶

1. Introduction
2. What Changes at Middle Level
3. Prerequisites
4. Glossary
5. SubmitWait in Depth
6. Stopped and the Shutdown Lifecycle
7. StopWait Internals From the Outside
8. The Idle-Worker Timeout
9. Panic Recovery
10. Queueing Strategies
11. Pause and Resume
12. Context Integration
13. Coding Patterns
14. Real-World Scenarios
15. Pool Composition
16. Backpressure Strategies
17. Drainage Patterns
18. Observability Hooks
19. Testing Pools
20. Pros and Cons Re-Examined
21. Edge Cases and Pitfalls
22. Common Mistakes
23. Common Misconceptions
24. Tricky Points
25. Test
26. Tricky Questions
27. Cheat Sheet
28. Self-Assessment Checklist
29. Summary
30. What You Can Build
31. Further Reading
32. Related Topics
33. Diagrams and Visual Aids

Introduction¶

The junior file gave you Submit and StopWait. That is enough for most batch-processing scripts. The middle file is about the operational corners — the situations where the basic API is not enough and you need to understand the next layer of the library.

Concretely, this file covers:

• SubmitWait — when you need a single task to finish before continuing, and what blocking on it implies for the pool's internals.
• Stopped and the full shutdown lifecycle, including the moment between "Stop called" and "all workers exited".
• The 2-second idle-worker timeout and what to do when your workload's burst cadence does not match it.
• Panic recovery — what the library does on your behalf and what it does not.
• Queue depth, queue inspection, and bounded variants of the queue.
• Pause/Resume (where available) for temporarily halting work without tearing down the pool.
• Context integration: getting cancellation into tasks even though func() cannot accept one.
• Backpressure patterns: how to slow down a producer that outruns the pool.

By the end of this file you will be able to:

• Pick between Submit and SubmitWait for any given producer pattern.
• Reason about what Stop does in the middle of a busy pool.
• Use pool.WaitingQueueSize() for telemetry without misinterpreting it.
• Handle panics from tasks without crashing the program.
• Add a context-aware cancellation layer on top of workerpool.
• Decide when to use one pool, two pools, or a custom pool altogether.

You will not yet read library internals (that is the senior file) or production sizing and incident stories (that is the professional file). The middle level is about competent use: knowing all the buttons on the dashboard and what each does.

What Changes at Middle Level¶

In one sentence: at middle level you stop treating the pool as a black box.

The junior file used the pool as if it were a fancy go f(). You wrote Submit, you wrote StopWait, you trusted the library to do the right thing. At middle level, you start asking:

• "What does SubmitWait actually do to the pool's internal state?"
• "Can a task that I SubmitWait from another task deadlock?"
• "If I stop the pool while a slow task is running, when does Stop return?"
• "Where does the queue live? How do I see its size?"
• "What happens if my task panics inside a select?"

These questions all have answers. None of them require reading source code (yet). They require understanding the public contract more deeply, which is what this file is for.

Prerequisites¶

• All junior material. You should be comfortable typing the basic submit-and-stop skeleton from memory.
• Familiarity with context.Context, context.WithCancel, context.WithTimeout. Without this, the context-integration section will make no sense.
• Comfort with select blocks on multiple channels.
• A basic understanding of how recover() works, including why it must run from a deferred function.
• Exposure to runtime.Stack for goroutine debugging — useful when you want to see "what is my pool doing?" in a stuck program.

If select { case <-ctx.Done(): return case x := <-ch: process(x) } is not yet idiomatic to you, write a few small programs first.

Glossary¶

SubmitWait in Depth¶

SubmitWait is the synchronous sibling of Submit. Its signature:

func (p *WorkerPool) SubmitWait(task func())

Like Submit, it returns nothing. Unlike Submit, it does not return until task has finished. Internally, the library wraps your task in a small helper that closes a "done" channel after the task runs, then blocks on that channel.

Pseudocode for what SubmitWait does:

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

This pseudocode is close enough to the real implementation that you can reason about behaviour with it.

When to use SubmitWait¶

Four legitimate use cases:

1. Backpressure on the producer. A loop of SubmitWait calls runs at exactly the workers' pace. Submissions cannot pile up; the queue never grows. Useful when memory is precious.
2. Ordering barriers. "Run this one task and make sure it is done before I proceed." For example, a "warm up the cache" task before launching parallel work.
3. Finalisers. The last task before shutdown, where you need to know it ran before doing something else.
4. Mixing pool with sequential logic. "Five parallel tasks, then one task that summarises, then another five parallel tasks." The summary in the middle is a SubmitWait.

When NOT to use SubmitWait¶

• Inside a busy loop submitting many tasks. The serialisation defeats the purpose of the pool.
• From inside another task on the same pool, especially if maxWorkers is small. You can deadlock.

The deadlock you must avoid¶

pool := workerpool.New(1) // pool of size 1!
pool.Submit(func() {
    fmt.Println("outer task started")
    pool.SubmitWait(func() {
        fmt.Println("inner task")
    })
    // never reached
})
pool.StopWait() // hangs forever

The outer task holds the only worker slot. The inner task is queued and will never start, because there is no free worker. The outer task is blocked on <-done. Classic deadlock.

This generalises: never SubmitWait from inside a task on a pool when worker availability is in doubt. If maxWorkers is large and your usage patterns guarantee at least one free slot, you may get away with it. But the safe rule is: do not.

Real-world variant — recursion:

func walk(node *Node, pool *workerpool.WorkerPool) {
    pool.SubmitWait(func() {
        for _, child := range node.Children {
            walk(child, pool) // recursive SubmitWait through the same pool
        }
    })
}

If the tree depth exceeds maxWorkers, you deadlock. For tree traversal, use a different design: a sync.WaitGroup plus Submit, where each task spawns its children without waiting:

func walk(node *Node, pool *workerpool.WorkerPool, wg *sync.WaitGroup) {
    wg.Add(1)
    pool.Submit(func() {
        defer wg.Done()
        for _, child := range node.Children {
            walk(child, pool, wg)
        }
    })
}

The pool will spawn children even when all maxWorkers are occupied — they go on the queue. Recursion stays correct, and you wg.Wait() at the end.

Submit-then-wait versus SubmitWait¶

These look similar but mean different things:

// Option A
done := make(chan struct{})
pool.Submit(func() {
    work()
    close(done)
})
<-done

// Option B
pool.SubmitWait(func() { work() })

Both block the caller until work is done. Option B is a touch more compact and avoids a hand-rolled channel. Behaviourally identical. Use B unless you want to give up on the wait via a select on <-ctx.Done() (in which case A wins, because B does not take a context).

Returning a value from SubmitWait¶

SubmitWait returns nothing. To get a value out, use a captured variable or a channel:

var result int
pool.SubmitWait(func() {
    result = compute()
})
fmt.Println(result)

This is safe because the wait guarantees the write happened before the read. But your linter will still scold you about a captured-by-reference variable; consider *int:

result := new(int)
pool.SubmitWait(func() {
    *result = compute()
})
fmt.Println(*result)

Or a channel of size 1:

ch := make(chan int, 1)
pool.SubmitWait(func() {
    ch <- compute()
})
fmt.Println(<-ch)

These are all equivalent. Pick whichever your team finds clearest.

Stopped and the Shutdown Lifecycle¶

Stopped() is a one-line method:

func (p *WorkerPool) Stopped() bool

It returns true once Stop or StopWait has been called. The flag becomes true at the moment of the call, before any workers have actually exited. So Stopped() == true does not mean "all tasks have run" or "all goroutines have exited". It means "shutdown has been initiated".

Why is this distinction important? Because a common loop checks Stopped before submitting:

for !pool.Stopped() {
    pool.Submit(produceOne())
}

This is correct in the sense that no submission is wasted after Stop. But there is still a race: between the Stopped() check and the Submit() call, another goroutine could call Stop. In that case, the Submit is silently dropped. That is usually fine — the task will be lost, but no error or panic. If silent dropping is unacceptable, you need a heavier-weight pattern with explicit acknowledgements; see the senior file.

Full shutdown timeline¶

A picture of what happens when you call StopWait:

1. Time T=0: pool.StopWait() is called. The internal stopped flag is set to true. From this moment, Stopped() returns true and Submit is a no-op.
2. T=0+: The dispatcher closes the input channel and starts draining its internal queue, forwarding remaining tasks to workers as they become free.
3. T=k: Some worker finishes the last task. Workers exit because the input channel is closed.
4. T=k+: The dispatcher itself exits.
5. T=k+: StopWait returns.

For Stop (not StopWait):

1. T=0: pool.Stop() is called. The internal stopped flag is set.
2. T=0+: The dispatcher discards any unstarted queued tasks. Workers currently running their task continue.
3. T=k: All currently-running tasks finish. Workers exit.
4. T=k+: The dispatcher exits.
5. T=k+: Stop returns.

The crucial point: both methods wait for running tasks to finish. There is no "kill running tasks" option in workerpool. If a task is stuck (deadlocked, blocked on I/O), Stop and StopWait hang waiting for it. You must build cancellation into the task itself with a context.

Inspecting state during shutdown¶

pool := workerpool.New(4)
// ... lots of submits ...

go func() {
    // run from another goroutine to watch shutdown progress
    for {
        time.Sleep(100 * time.Millisecond)
        fmt.Println("queue size:", pool.WaitingQueueSize(), "stopped:", pool.Stopped())
        if pool.Stopped() && pool.WaitingQueueSize() == 0 {
            // not a real "all done" signal, but close
            return
        }
    }
}()

pool.StopWait()

This is a debug-only pattern. For real telemetry, expose a metrics.Gauge on WaitingQueueSize() and your monitoring system will draw the curve for you.

Idempotent shutdown¶

Both Stop and StopWait are idempotent — calling them twice is fine. So the pattern:

defer pool.StopWait()
// ... 
pool.StopWait() // explicit

is safe; the deferred call is a no-op.

The other direction:

pool.Stop()
pool.StopWait() // also a no-op (or close to it)

Calling StopWait after Stop does not somehow drain the queue (the queue was already cleared). It just waits for the running tasks if there are any.

StopWait Internals From the Outside¶

Without yet reading the source, here is what you can deduce about StopWait from observed behaviour.

1. It does not block on a fixed timeout. It waits as long as needed.
2. It waits for tasks to finish, not for goroutines to be reaped. Workers may exit a moment after StopWait returns; the goroutine count drops to 1 (then 0 after the dispatcher).
3. It is goroutine-safe. Multiple goroutines can call StopWait concurrently; only one shutdown happens; all callers return at the same time.
4. It does not call runtime.Gosched() or yield the CPU. It is just a <-doneChan style wait.

What you cannot deduce without reading the source — and what we will spell out in the senior file:

• How the dispatcher signals shutdown to workers.
• How worker exit propagates back to StopWait.
• Whether the channel signalling shutdown is buffered.

For middle-level usage, the takeaway is: StopWait always returns eventually, unless a task is stuck. If your StopWait hangs forever, the culprit is a task that does not return — almost always a missing context cancellation or a deadlock on a channel.

The Idle-Worker Timeout¶

Internally, workerpool runs a 2-second timer per worker. If a worker sits idle for 2 seconds with no new task, it exits. The next time a task arrives, the dispatcher spins up a fresh worker for it. This keeps the resident goroutine count low during quiet periods.

The 2-second value is hard-coded; you cannot change it via the public API.

Why this matters¶

For a bursty workload — say, traffic arrives once every 30 seconds in small bursts — the pool effectively has zero workers most of the time. Every burst pays the cost of spawning fresh workers. That cost is small (a goroutine and a goroutine-stack ~2KB allocation) but measurable. For typical web workloads, you will not notice; for nanosecond-sensitive code, it could add up.

For a steady workload — tasks arriving every millisecond — workers never go idle long enough to be reaped. The pool stabilises with however many workers it needs (up to maxWorkers) and keeps them all busy.

For a workload that has gaps just above 2 seconds — say, 2.5 seconds between bursts — you get the worst of both worlds: workers exit just before the next burst, and you spend each burst spinning up new ones. If profiling shows this is a hot path, options:

1. Submit a no-op heartbeat task every 1.9 seconds. Crude but effective.
2. Switch to a different library that exposes the idle timeout (ants.WithExpiryDuration).
3. Roll your own pool with no idle reaping.

Observing it¶

You can convince yourself the reaping is real:

package main

import (
    "fmt"
    "runtime"
    "time"

    "github.com/gammazero/workerpool"
)

func main() {
    pool := workerpool.New(10)
    defer pool.StopWait()

    fmt.Println("immediately after New:", runtime.NumGoroutine())

    for i := 0; i < 10; i++ {
        pool.Submit(func() { time.Sleep(50 * time.Millisecond) })
    }
    time.Sleep(100 * time.Millisecond)
    fmt.Println("during work:", runtime.NumGoroutine())

    time.Sleep(3 * time.Second) // longer than idle timeout
    fmt.Println("after idle reap:", runtime.NumGoroutine())
}

Typical output:

immediately after New: 2
during work: 12
after idle reap: 2

NumGoroutine reports the main goroutine and the dispatcher (= 2) at idle, plus up to 10 workers while busy.

A diagnostic: 2 seconds is just enough to confuse you¶

A user reports: "I see periodic CPU spikes every 2 seconds even when I am not doing anything!"

Possible explanation: somewhere, code submits a task every ~2.5 seconds. Every submit spawns a worker (because the previous one was just reaped). The fresh worker runs a fast task, then idles for 2 seconds, then exits. The cycle repeats.

The fix is either to consolidate the submission cadence to <2 seconds (no reaping happens) or to skip the pool for this workload entirely (just go f() for a one-off).

Panic Recovery¶

Modern versions of workerpool install a defer recover() around each task. If your task panics, the worker survives and the pool keeps running. The panic value, however, is not propagated to you. It is logged via log.Print (or printed to stderr — check the exact version) and discarded.

This is sometimes good (a buggy task does not kill the program) and sometimes bad (the bug is silent).

Catching panics yourself¶

pool.Submit(func() {
    defer func() {
        if r := recover(); r != nil {
            log.Printf("task %s panic: %v\n%s", taskName, r, debug.Stack())
            // ... your custom handling, e.g. Sentry, metrics, etc.
        }
    }()
    risky()
})

If you wrap every task this way, you get the panic value, the stack trace, and a place to send it to your monitoring stack. The library's outer recover then sees nothing to handle (because you already handled it).

A helper makes this less tedious:

func instrumented(f func()) func() {
    return func() {
        defer func() {
            if r := recover(); r != nil {
                log.Printf("panic: %v\n%s", r, debug.Stack())
            }
        }()
        f()
    }
}

pool.Submit(instrumented(func() { risky() }))

Panics that the library's recover cannot catch¶

recover() only catches panics in the same goroutine. If your task spawns a child goroutine that panics, the library's defer does not catch it; the program crashes. Example:

pool.Submit(func() {
    go func() {
        panic("child")  // crashes the whole program
    }()
})

Wrap children explicitly:

pool.Submit(func() {
    go func() {
        defer func() { recover() }()
        risky()
    }()
})

This is a general Go gotcha, not specific to workerpool, but worth mentioning since pools sometimes hide the parent/child structure of goroutines.

A panicking task that holds a mutex¶

If your panicking task holds a mutex when it panics, and recover swallows the panic, the mutex stays locked. Always release locks with defer mu.Unlock():

pool.Submit(func() {
    mu.Lock()
    defer mu.Unlock()
    risky() // even if it panics, mu is released
})

Without defer, a panic leaks the lock and the next caller of mu.Lock() blocks forever.

Queueing Strategies¶

The internal queue inside workerpool is a linked list of slices (deque-like). You do not interact with it directly; you only see WaitingQueueSize(). But your queueing strategy — how producers push into the pool — is entirely your choice.

Strategy 1: Naive Submit loop¶

for _, item := range items {
    item := item
    pool.Submit(func() { process(item) })
}

Pros: simplest possible code. Producer never blocks. Cons: if the producer is fast and workers are slow, the queue grows without bound. OOM risk.

When to use: bounded input, trusted producer.

Strategy 2: SubmitWait loop¶

for _, item := range items {
    item := item
    pool.SubmitWait(func() { process(item) })
}

Pros: serialised submission, no queue growth. Cons: producer is now serialised. Throughput is limited by min(workers, producer rate).

When to use: very large input, memory-bounded environment.

Strategy 3: Semaphore-bounded submit¶

sem := make(chan struct{}, 1000)
for _, item := range items {
    sem <- struct{}{}
    item := item
    pool.Submit(func() {
        defer func() { <-sem }()
        process(item)
    })
}

Pros: producer is concurrent with workers, but blocks if queue depth exceeds 1000. Cons: extra goroutine state to manage. The 1000 is now a magic number.

When to use: untrusted producer or production code where you want a hard cap on queued work.

Strategy 4: Conditional submit¶

for _, item := range items {
    item := item
    if pool.WaitingQueueSize() > 1000 {
        // drop this item, or block, or log, or whatever your policy is
        droppedCount++
        continue
    }
    pool.Submit(func() { process(item) })
}

Pros: lets you implement custom backpressure policy. Cons: WaitingQueueSize is a snapshot, so this is racy — the actual queue size at the moment of Submit may differ. Good for "approximate" throttling.

When to use: when dropping is acceptable.

Strategy 5: Channel of items, single dispatcher goroutine¶

ch := make(chan Item, 100)
go func() {
    for it := range ch {
        it := it
        pool.Submit(func() { process(it) })
    }
}()

for _, item := range items {
    ch <- item // blocks when channel is full
}
close(ch)

Pros: producer-side backpressure via the bounded channel. Cons: extra moving part.

When to use: when the producer side is in a different goroutine entirely and you want a clean handoff point.

The right strategy depends on your tolerance for queue growth and your producer's behaviour. The library does not pick for you; that is the price of its minimal API.

Pause and Resume¶

Some versions of workerpool expose a Pause(ctx context.Context) method. Its purpose: temporarily prevent the dispatcher from sending queued tasks to workers, without losing them. Resume happens when the context is cancelled.

ctx, cancel := context.WithCancel(context.Background())
pool.Pause(ctx)
// queued tasks sit. New submissions queue too. Running tasks finish normally.

// ... some seconds later ...
cancel() // pool resumes

Use cases:

• A circuit breaker — pause the pool when a downstream service is failing.
• A scheduled maintenance window — quiet the pool during a deploy.
• Coordinated shutdown across multiple pools — pause all, then StopWait each.

Check the version of workerpool you depend on; the method may not exist in older releases. The specification file has the version history.

A simple pattern:

func breaker(pool *workerpool.WorkerPool, downstreamHealthy func() bool) {
    var pauseCancel context.CancelFunc
    for {
        if !downstreamHealthy() && pauseCancel == nil {
            ctx, cancel := context.WithCancel(context.Background())
            pool.Pause(ctx)
            pauseCancel = cancel
            log.Println("pool paused, downstream is unhealthy")
        }
        if downstreamHealthy() && pauseCancel != nil {
            pauseCancel()
            pauseCancel = nil
            log.Println("pool resumed")
        }
        time.Sleep(5 * time.Second)
    }
}

A note on subtlety: Pause does not pause running tasks. Tasks already executing continue. Pause only prevents new tasks from leaving the queue and reaching workers. If a "task" is a long-running I/O call to a failing service, pausing the pool will not help you avoid that call — the task that already started will keep failing. You need either circuit-breaker logic inside the task itself, or a shorter task lifetime.

Context Integration¶

workerpool predates the widespread use of context.Context in Go libraries, and its task type is func() with no arguments. This means there is no direct way to pass a context into a task. You have to do it yourself with a closure.

Pattern: pass context via closure¶

func process(ctx context.Context, item Item) {
    select {
    case <-ctx.Done():
        return
    default:
    }
    // ... do work, respecting ctx ...
}

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

for _, item := range items {
    item := item
    pool.Submit(func() {
        process(ctx, item)
    })
}
pool.StopWait()

The context is captured by reference, shared across all tasks. Cancelling it makes every task check ctx.Done() and bail early. The pool itself does not know about the context; it just keeps running tasks. The tasks are responsible for honouring cancellation.

Pattern: context per task¶

for _, item := range items {
    item := item
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    pool.Submit(func() {
        defer cancel()
        process(ctx, item)
    })
}

Each task has its own 5-second deadline. If a task is queued for 4 seconds and then runs for 4 seconds, the 5-second context will fire mid-execution. That is usually what you want.

A subtle issue: the context starts ticking the moment you create it, even before the task runs. If queue dwell time is significant, the context could be cancelled before the task starts. The task then runs and immediately returns. To avoid this, create the context inside the task:

for _, item := range items {
    item := item
    pool.Submit(func() {
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()
        process(ctx, item)
    })
}

Now the 5-second timer starts when the task starts running.

Pattern: parent context cancels everything¶

parentCtx, parentCancel := context.WithCancel(context.Background())
defer parentCancel()

for _, item := range items {
    item := item
    pool.Submit(func() {
        ctx, cancel := context.WithTimeout(parentCtx, 5*time.Second)
        defer cancel()
        process(ctx, item)
    })
}

// elsewhere: parentCancel() will short-circuit every task

context.WithTimeout(parentCtx, ...) creates a child that respects both timeouts: the parent's cancellation and the 5-second limit. Cancelling the parent kills every child instantly.

What happens to queued tasks when the context is cancelled?¶

They still run. The pool does not consult the context. They just run fast, because the first thing they do is check ctx.Done() and return.

That is a notable cost: if you queue 100,000 tasks and then cancel the context, the pool still has to process 100,000 returns from ctx.Done(). That is fast but not free. Better:

pool.Stop() // discard all queued tasks

If your goal is "stop everything now", Stop is sometimes faster than relying on context to short-circuit each queued task.

Hybrid: cancel via context plus Stop¶

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
defer pool.Stop() // discards what's left after context cancels

for _, item := range items {
    item := item
    pool.Submit(func() { process(ctx, item) })
}

The deferred order is Stop first (since it was deferred last), then cancel. Wait — that is backwards. Defers run LIFO, so cancel runs first, then Stop. The cancel signals tasks; the stop drains the rest (which return fast).

Honestly, mixing these patterns is enough complexity that you should write a comment explaining what you intended.

Coding Patterns¶

Pattern: WaitGroup + Pool¶

Already mentioned in junior; worth restating because it is the most common middle-level idiom.

var wg sync.WaitGroup
for _, item := range items {
    item := item
    wg.Add(1)
    pool.Submit(func() {
        defer wg.Done()
        process(item)
    })
}
wg.Wait()
// pool still alive, can submit more

Use this when you need a "batch boundary" without killing the pool.

Pattern: Channel-based result collection with bounded pool¶

results := make(chan Result, 0)
go func() {
    var wg sync.WaitGroup
    for _, item := range items {
        item := item
        wg.Add(1)
        pool.Submit(func() {
            defer wg.Done()
            results <- compute(item)
        })
    }
    wg.Wait()
    close(results)
}()

for r := range results {
    // consume
}

Producers feed the pool, the pool feeds the channel, the consumer drains. close(results) happens only when wg.Wait returns, so no extra-send-after-close race.

Pattern: Map-reduce¶

type partial struct {
    sum, count int
}

partials := make(chan partial, 0)
var wg sync.WaitGroup

for _, chunk := range chunks {
    chunk := chunk
    wg.Add(1)
    pool.Submit(func() {
        defer wg.Done()
        s, c := 0, 0
        for _, v := range chunk {
            s += v
            c++
        }
        partials <- partial{sum: s, count: c}
    })
}

go func() {
    wg.Wait()
    close(partials)
}()

totalSum, totalCount := 0, 0
for p := range partials {
    totalSum += p.sum
    totalCount += p.count
}

Map and reduce phases use the same pool. The intermediate channel collects partial sums. Note the goroutine that closes the channel — without it, the reduce loop would never terminate.

Pattern: Pipeline with multiple pools¶

parse := workerpool.New(2)
enrich := workerpool.New(8)
write := workerpool.New(4)

go func() {
    defer parse.StopWait()
    for _, raw := range inputs {
        raw := raw
        parse.Submit(func() {
            p := parseOne(raw)
            enrich.Submit(func() {
                e := enrichOne(p)
                write.Submit(func() { writeOne(e) })
            })
        })
    }
}()

// wait somehow ... a wg counting writes is one option

Each stage has its own bound. A slow stage will accumulate work in front of it (the next pool's queue). Watch queue sizes; whichever queue is growing is your bottleneck.

Pattern: Two-phase shutdown¶

// phase 1: stop accepting new work
close(producerStopCh)
producerWg.Wait()

// phase 2: drain the pool
pool.StopWait()

This separates "stop the producers" from "drain the pool". In a complex service, this is the only sane way to ensure no work is dropped.

Pattern: Per-task instrumentation wrapper¶

type Pool struct {
    inner *workerpool.WorkerPool
    sub   prometheus.Counter
    done  prometheus.Counter
    dur   prometheus.Histogram
}

func (p *Pool) Submit(name string, f func()) {
    p.sub.Inc()
    start := time.Now()
    p.inner.Submit(func() {
        defer p.done.Inc()
        defer func() { p.dur.Observe(time.Since(start).Seconds()) }()
        f()
    })
}

This is the start of a proper instrumentation layer; we will expand it in the professional file.

Real-World Scenarios¶

A walkthrough of three plausible middle-level uses.

Scenario 1: Concurrent database upsert¶

You have 10,000 rows to write to a database. The database can comfortably handle 20 concurrent writes. Each write takes ~5ms.

type Row struct {
    ID   int
    Data string
}

func UpsertAll(ctx context.Context, db *sql.DB, rows []Row) error {
    pool := workerpool.New(20)
    defer pool.StopWait()

    var firstErr error
    var mu sync.Mutex

    for _, r := range rows {
        if ctx.Err() != nil {
            break
        }
        r := r
        pool.Submit(func() {
            if ctx.Err() != nil {
                return
            }
            if err := upsert(ctx, db, r); err != nil {
                mu.Lock()
                if firstErr == nil {
                    firstErr = err
                }
                mu.Unlock()
            }
        })
    }
    pool.StopWait()
    return firstErr
}

func upsert(ctx context.Context, db *sql.DB, r Row) error {
    _, err := db.ExecContext(ctx, `INSERT INTO t (id, data) VALUES ($1, $2)
        ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data`, r.ID, r.Data)
    return err
}

Things to notice:

• The pool's concurrency cap matches the DB's capacity.
• Each task respects the context; if the caller cancels, tasks bail.
• We capture the first error and keep going, rather than aborting on every error. Adjust to taste.
• The wall-clock time is 10000 * 5ms / 20 = 2.5 seconds — vs. ~50 seconds serial.

Scenario 2: Concurrent log-line enrichment¶

A log shipper reads lines from a file, enriches each line with a service lookup, then writes to another file.

type LineWithMeta struct {
    Line string
    Meta map[string]string
}

func EnrichLog(ctx context.Context, in io.Reader, out io.Writer) error {
    pool := workerpool.New(50) // I/O bound, more workers ok
    defer pool.StopWait()

    results := make(chan LineWithMeta, 1000)
    var producerWg sync.WaitGroup

    scanner := bufio.NewScanner(in)
    for scanner.Scan() {
        if ctx.Err() != nil {
            break
        }
        line := scanner.Text()
        producerWg.Add(1)
        pool.Submit(func() {
            defer producerWg.Done()
            meta, _ := lookup(ctx, line)
            results <- LineWithMeta{Line: line, Meta: meta}
        })
    }

    go func() {
        producerWg.Wait()
        close(results)
    }()

    writer := bufio.NewWriter(out)
    defer writer.Flush()
    for r := range results {
        if _, err := fmt.Fprintf(writer, "%s\t%v\n", r.Line, r.Meta); err != nil {
            return err
        }
    }
    return scanner.Err()
}

Why 50 workers? Because the bottleneck is the lookup service call, which is network-bound, and that service can handle 50 concurrent requests. Pool size mirrors downstream capacity.

The pattern is the classic "producer → pool → channel → consumer". The pool decouples reading from writing.

Scenario 3: Periodic batch job¶

A cron-style task that runs every 5 minutes, processes whatever pending work is in a queue, and exits. A small pool here avoids holding goroutines forever.

func RunBatch() {
    pool := workerpool.New(8)
    defer pool.StopWait()

    rows, err := loadPending()
    if err != nil {
        log.Fatal(err)
    }

    log.Printf("processing %d rows", len(rows))
    for _, r := range rows {
        r := r
        pool.Submit(func() {
            if err := process(r); err != nil {
                log.Printf("row %d: %v", r.ID, err)
            }
        })
    }
}

func main() {
    ticker := time.NewTicker(5 * time.Minute)
    for range ticker.C {
        RunBatch()
    }
}

A new pool per batch is fine here because a batch run is a discrete unit. The pool exists during the run and is gone afterward. Goroutines do not leak. The next batch starts fresh.

Pool Composition¶

Sometimes one pool is not enough. Composition strategies:

Two pools, one process¶

var cpuPool = workerpool.New(runtime.NumCPU())
var ioPool = workerpool.New(64)

Use cpuPool for tight number-crunching tasks, ioPool for network/disk. Why? Because mixing them in one pool means the I/O wait time of one task can starve the CPU work of another. Two pools, two queues, no cross-contention.

Pool inside a worker¶

mainPool := workerpool.New(4)
subPool := workerpool.New(8)

mainPool.Submit(func() {
    // do some prep
    for _, x := range subTasks {
        x := x
        subPool.Submit(func() { work(x) })
    }
})

The outer task submits to an inner pool. Watch for deadlocks — if you subPool.SubmitWait from inside mainPool, and the inner pool's capacity is small, you can starve.

Pipeline of pools¶

Covered in Scenario 2 above. Each stage gets its own pool.

Per-tenant pools (caution)¶

A common request: "I want a pool per tenant, so a noisy tenant cannot starve a quiet one." This works for a handful of tenants. For 10,000 tenants, you have 10,000 dispatchers, each holding a goroutine and some state. That is not great. Better:

• A single shared pool with priority queueing (the library does not support this; you would build it).
• A token bucket per tenant in front of one shared pool.
• A sharded design where N tenants map to one of K pools (tenantID % K).

workerpool's minimal design is not the right tool for per-tenant isolation at scale. Recognise the limit.

Backpressure Strategies¶

Push-back from the pool: not built in¶

workerpool.Submit never blocks, never errors. So you must add backpressure yourself.

Token bucket / leaky bucket¶

limiter := rate.NewLimiter(rate.Every(time.Millisecond), 100) // 1000/s

for _, item := range items {
    if err := limiter.Wait(ctx); err != nil {
        return err
    }
    item := item
    pool.Submit(func() { process(item) })
}

This caps the submission rate, regardless of worker speed. Useful for rate-limited downstreams.

Counting semaphore¶

Already shown above; the simplest and most common pattern.

Drop-on-overload¶

const overloadThreshold = 5000

for _, item := range items {
    if pool.WaitingQueueSize() > overloadThreshold {
        // drop and log
        metrics.Drops.Inc()
        continue
    }
    item := item
    pool.Submit(func() { process(item) })
}

Be careful: dropping may be unacceptable for your use case. Make sure the rest of the system tolerates lost work.

Block-on-overload¶

for pool.WaitingQueueSize() > overloadThreshold {
    time.Sleep(10 * time.Millisecond)
}
pool.Submit(func() { process(item) })

Crude but works. A semaphore is cleaner because it has no polling.

Hybrid: warning, then drop¶

qs := pool.WaitingQueueSize()
if qs > overloadThreshold {
    metrics.Drops.Inc()
    continue
}
if qs > warningThreshold {
    log.Printf("queue depth high: %d", qs)
}
pool.Submit(func() { process(item) })

Real production code looks like this. Logs flag impending overload; drops kick in at the breaking point.

Drainage Patterns¶

"Drainage" = orderly shutdown without losing in-flight work.

Pattern: defer StopWait¶

pool := workerpool.New(8)
defer pool.StopWait()
// ... work ...

Already covered. The simplest drainage; safe for short-lived programs.

Pattern: signal-driven drain¶

sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT)
<-sigs
log.Println("draining pool...")
pool.StopWait()
log.Println("done")
os.Exit(0)

For long-running services. SIGTERM arrives, drain, exit. Kubernetes gives you terminationGracePeriodSeconds to do this; default is 30 seconds. If your drain might exceed that, configure a longer grace period or use a deadline:

Pattern: deadline drain¶

done := make(chan struct{})
go func() {
    pool.StopWait()
    close(done)
}()

select {
case <-done:
    log.Println("drained cleanly")
case <-time.After(25 * time.Second):
    log.Println("drain deadline exceeded, hard stop")
    pool.Stop()
    <-done
}

StopWait runs in a goroutine. We wait up to 25 seconds. If it does not finish, we call Stop to discard the remaining queue, and wait for StopWait to return (it will return quickly since the queue is now empty). This pattern bounds shutdown time at the cost of losing some work.

Pattern: pause-then-drain¶

ctx, cancel := context.WithCancel(context.Background())
pool.Pause(ctx) // pause new dispatches

// ... wait for in-flight tasks to drain naturally ...

cancel() // resume (so StopWait can finish remaining queued)
pool.StopWait()

For coordinated shutdowns across multiple pools. Pause all, wait for in-flight, then drain.

Observability Hooks¶

The library exposes only WaitingQueueSize() and Stopped() for observability. The rest you bolt on.

Metric 1: queue depth¶

var (
    queueDepth = prometheus.NewGauge(prometheus.GaugeOpts{Name: "wp_queue_depth"})
)

go func() {
    ticker := time.NewTicker(5 * time.Second)
    for range ticker.C {
        queueDepth.Set(float64(pool.WaitingQueueSize()))
    }
}()

Track over time. Growing depth = capacity problem.

Metric 2: submit rate, complete rate¶

var (
    submits = prometheus.NewCounter(prometheus.CounterOpts{Name: "wp_submits_total"})
    done    = prometheus.NewCounter(prometheus.CounterOpts{Name: "wp_completions_total"})
)

wrap := func(f func()) func() {
    return func() {
        defer done.Inc()
        f()
    }
}

submits.Inc()
pool.Submit(wrap(func() { process(...) }))

The difference submits - done is "tasks in the pool right now" (queued + running).

Metric 3: task duration histogram¶

var taskDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
    Name: "wp_task_duration_seconds",
})

pool.Submit(func() {
    start := time.Now()
    defer func() { taskDuration.Observe(time.Since(start).Seconds()) }()
    process(...)
})

P95 and P99 task duration tell you whether the bottleneck is "everything is slow" or "some tasks are pathological".

Logging slow tasks¶

pool.Submit(func() {
    start := time.Now()
    process(...)
    if d := time.Since(start); d > 500*time.Millisecond {
        log.Printf("slow task: %s", d)
    }
})

Cheap, effective. Pairs nicely with the duration histogram.

Stack-dump on stuck pool¶

sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGUSR1)
go func() {
    for range sigs {
        buf := make([]byte, 1<<20)
        n := runtime.Stack(buf, true)
        log.Printf("goroutine dump:\n%s", buf[:n])
    }
}()

When StopWait hangs in production, sending SIGUSR1 dumps every goroutine's stack to logs. You can then see exactly which workers are stuck and where.

Testing Pools¶

Pools are concurrency primitives; tests for them should focus on:

1. All submitted tasks run. Counter equals submission count after StopWait.
2. No tasks run after Stop. Stop before any execution, then assert counter is zero.
3. The pool respects maxWorkers. Track peak concurrent tasks.
4. SubmitWait blocks. A task that increments a counter, then SubmitWait, then check the counter.
5. StopWait is idempotent. Call it twice; should be a no-op.
6. The race detector finds no races. Always run go test -race.

A peak-concurrency test:

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

    var inflight int64
    var peak int64

    for i := 0; i < 100; i++ {
        pool.Submit(func() {
            n := atomic.AddInt64(&inflight, 1)
            for {
                p := atomic.LoadInt64(&peak)
                if n <= p || atomic.CompareAndSwapInt64(&peak, p, n) {
                    break
                }
            }
            time.Sleep(10 * time.Millisecond)
            atomic.AddInt64(&inflight, -1)
        })
    }
    pool.StopWait()
    if peak > int64(maxW) {
        t.Fatalf("peak concurrent tasks = %d, want <= %d", peak, maxW)
    }
}

This test verifies the library's contract experimentally. It is also a nice baseline for any custom pool you build.

Pros and Cons Re-Examined¶

The junior file gave a pros/cons list. At middle level the trade-offs sharpen.

Pros (deeper)¶

• Submission contention is minimal. The dispatcher's input channel is unbuffered but the dispatcher is a single goroutine, so contention is between submitters and one reader. For most workloads this is fine.
• Idle reaping is automatic. You do not have to think about it. Most users do not even know it exists.
• SubmitWait is a one-liner. Synchronous calls are easy without any extra plumbing.
• The API is stable. Code written against v1.0 still compiles five years later.

Cons (deeper)¶

• No per-task timeout. Built-in. You must wrap.
• No per-task error. Built-in. You must collect.
• No context propagation. Tasks are func(). You must capture.
• No prioritisation. Tasks run FIFO, period.
• No "drain to a deadline". You must build it.
• No introspection beyond queue size. No "how many workers active right now?".
• Hard-coded idle timeout. Bursty workloads with 2.5s gaps pay extra.
• Per-task closure allocation. Sub-microsecond overhead but not free.

At middle level you start feeling these cons in real services. The professional file is where you decide when they justify reaching for ants instead.

Edge Cases and Pitfalls¶

SubmitWait from inside SubmitWait¶

pool := workerpool.New(2)
pool.SubmitWait(func() {
    pool.SubmitWait(func() {
        // ...
    })
})

With maxWorkers = 2, the outer occupies one slot. The inner takes another. Both finish. Works. With maxWorkers = 1, the outer occupies the only slot, the inner cannot start, the outer blocks on the inner's done channel. Deadlock.

Task that hangs¶

pool.Submit(func() {
    select{} // infinite block
})
pool.StopWait() // never returns

StopWait cannot kill a running task. If you want bounded shutdown time, give every task a context with a deadline.

Forgetting defer cancel() on contexts¶

for _, item := range items {
    item := item
    ctx, _ := context.WithTimeout(context.Background(), 5*time.Second) // missing cancel
    pool.Submit(func() {
        process(ctx, item)
    })
}

This leaks context.cancelCtx objects. Lint catches it (govet contextcheck). Always defer cancel() — but be careful where: outside the closure, the cancel fires immediately, cancelling the context before the task runs.

Correct:

pool.Submit(func() {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    process(ctx, item)
})

StopWait while Submit is racing¶

If one goroutine is Submit-ing and another calls StopWait, the racing Submit is probably silently dropped if it arrives after StopWait started. But behaviour around the exact instant of the call is subtle. To be safe, ensure submitters are stopped before draining.

Unbounded queue on a server endpoint¶

A pool that accepts requests via an HTTP handler with no rate limit is a DoS vector. We covered this in junior; the advice is the same here, more emphatic. Always bound the queue if input is untrusted.

Idle workers and runtime.NumGoroutine reporting¶

A debugger that checks runtime.NumGoroutine() to assert "I should have 8 workers" will be confused. The pool has 0 to 8 workers depending on load. Use pool.WaitingQueueSize() and your own metrics for assertions.

Pause does not affect running tasks¶

A paused pool still has its workers running whatever they were running. Pause only stops new dispatches. Do not assume "pool paused" = "system quiet".

Closing a result channel from inside a task¶

results := make(chan int)
for i := 0; i < 10; i++ {
    i := i
    pool.Submit(func() {
        results <- i
        if i == 9 {
            close(results)
        }
    })
}

Wrong. Workers may finish in any order. The task with i == 9 may close while others still try to send. Close after StopWait:

pool.StopWait()
close(results)

But then you cannot range over results in the same goroutine that calls StopWait. Use:

go func() {
    pool.StopWait()
    close(results)
}()
for r := range results {
    // ...
}

Or use a WaitGroup.

Common Mistakes¶

1. Forgetting that SubmitWait can deadlock¶

The classic small-pool self-deadlock. Audit any SubmitWait inside a task.

2. Misusing Stopped as "all done"¶

Stopped() == true means shutdown started, not finished. Use StopWait to know all tasks have run.

3. Per-task contexts shared across goroutines¶

ctx, cancel := context.WithTimeout(...)
defer cancel()
for ... {
    pool.Submit(func() { use(ctx) })
}

This is correct — one shared context for the whole batch — but a mistake is to expect per-task timeouts. Each task shares the same deadline. If you want a fresh deadline per task, create the context inside the closure.

4. Submitting without checking Stopped¶

In long-running producers, missing the if pool.Stopped() return means the loop spins after shutdown, accomplishing nothing.

5. Trusting that panic recovery is enough¶

The library recovers panics in the worker goroutine. It does not recover panics in goroutines you spawn from inside a task. Audit your tasks; wrap child goroutines.

6. Believing WaitingQueueSize is precise¶

It is a snapshot. By the time you read it, it may have changed. For metrics: fine. For synchronisation: do not.

7. Mixing Submit and SubmitWait haphazardly¶

A loop that calls one or the other based on a condition can be hard to reason about. Pick one style per loop.

8. Calling StopWait from inside a task¶

pool.Submit(func() {
    pool.StopWait() // wait for self to finish?
})

Deadlock. StopWait waits for all tasks, including this one. The task waits for StopWait. Forever.

9. Using Pause to handle errors¶

Pausing the pool is not error handling. It is suppression. If a downstream is failing, you eventually need to either retry, drop, or surface the error — not just sit on a paused queue forever.

10. Forgetting runtime.GOMAXPROCS¶

If you have set GOMAXPROCS=1 (some test runners do, some embedded targets force it), a CPU-bound pool with maxWorkers > 1 gains you nothing. The OS scheduler will serialise the goroutines anyway. Always know your GOMAXPROCS.

Common Misconceptions¶

"Pause stops running tasks"¶

No. Pause only prevents new dispatches.

"SubmitWait is faster than Submit because it skips the queue"¶

No. SubmitWait still goes through the same dispatcher and queue. It just blocks the caller until the task finishes. It is slower per call, not faster.

"The pool retries failed tasks"¶

No. Tasks run once. Retries are your responsibility.

"If I create a pool with maxWorkers=1, my tasks run in order"¶

In practice, yes — there is only one worker, and the queue is FIFO. But the library does not guarantee FIFO across all versions and edge cases. If strict ordering matters, use a single goroutine reading from a channel.

"Workers are reused so my init code runs once"¶

No. The pool does not let you initialise workers. Each task is independent. If you need per-worker state, use tunny.

"StopWait kills running tasks after a timeout"¶

No. StopWait waits forever for running tasks. Add your own deadline.

"Submitting nil is okay because the library handles it"¶

Some versions silently drop nil, some recover the resulting panic, some panic before recovery. Do not submit nil.

"Pause is the same as Stop"¶

No. Pause is reversible. Stop is not.

Tricky Points¶

SubmitWait blocks even if you do not care about the result¶

If you only want backpressure, SubmitWait is fine. If you have nothing to wait for and just used it out of habit, you have made your submitter into a single-task-at-a-time loop. Consider Submit + semaphore for better throughput.

The recover wraps your task in defer¶

This means a panic in a deferred function inside your task still gets caught by the library's outer recover. That can hide bugs in deferred cleanup.

Pause semantics around in-flight tasks¶

When you call pool.Pause(ctx), tasks already dispatched to workers continue. Tasks in the queue stay there. Until ctx is cancelled, new submits queue but never dispatch. After cancel, the dispatcher resumes from where it stopped.

Stop with WaitingQueueSize() == 0¶

If the queue is empty and you call Stop, the only thing remaining is running tasks. Workers finish, exit, dispatcher exits, Stop returns. Same as StopWait would do in that case.

Order of Submit and Pause¶

If you submit immediately after Pause, the task queues but does not run until resume. Useful for "queue work without doing it yet".

Pool with all workers busy¶

When maxWorkers workers are all running tasks, the dispatcher queues more. Workers free up one by one and grab queued tasks. The WaitingQueueSize is exactly the queued count, not including the maxWorkers running.

Submitting after cancel() of the pause context¶

If you cancel the pause context (resume), tasks queued during the pause dispatch as soon as workers are free. Then new submits flow normally.

Submit from many goroutines¶

The library is fine with this. Internally, all submits funnel through a single channel into the dispatcher. The dispatcher serialises them. So submission contention is bounded by the channel, not by user code.

Test¶

package main

import (
    "errors"
    "fmt"
    "sync"
    "sync/atomic"
    "testing"
    "time"

    "github.com/gammazero/workerpool"
)

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

    var done int64
    pool.SubmitWait(func() {
        time.Sleep(50 * time.Millisecond)
        atomic.StoreInt64(&done, 1)
    })
    if atomic.LoadInt64(&done) != 1 {
        t.Fatal("SubmitWait did not wait for task")
    }
}

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

    var ran int64
    pool.Submit(func() { atomic.AddInt64(&ran, 1) })

    time.Sleep(10 * time.Millisecond)
    if atomic.LoadInt64(&ran) != 0 {
        t.Fatal("submit after StopWait should be no-op")
    }
}

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

func TestStopDiscardsUnstartedRoughly(t *testing.T) {
    pool := workerpool.New(1)
    var done int64
    for i := 0; i < 100; i++ {
        pool.Submit(func() {
            time.Sleep(10 * time.Millisecond)
            atomic.AddInt64(&done, 1)
        })
    }
    pool.Stop()
    // With 100 tasks sleeping 10ms each on 1 worker, Stop should drop most.
    if atomic.LoadInt64(&done) == 100 {
        t.Log("edge case: all tasks finished before Stop; test is racy by design")
    }
}

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

    pool.Submit(func() { panic("boom") })

    // After the panic, the pool should still process new tasks.
    var ran int64
    pool.Submit(func() { atomic.AddInt64(&ran, 1) })
    pool.StopWait()

    if atomic.LoadInt64(&ran) != 1 {
        t.Fatal("pool did not recover from task panic")
    }
}

func TestRecoverableErrorCollection(t *testing.T) {
    pool := workerpool.New(4)

    var mu sync.Mutex
    var errs []error
    for i := 0; i < 10; i++ {
        i := i
        pool.Submit(func() {
            if i%2 == 0 {
                mu.Lock()
                errs = append(errs, fmt.Errorf("err %d", i))
                mu.Unlock()
            }
        })
    }
    pool.StopWait()

    if len(errs) != 5 {
        t.Fatalf("expected 5 errors, got %d", len(errs))
    }
}

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

    var ran int64
    ctx, cancel := context.WithCancel(context.Background())
    pool.Submit(func() {
        select {
        case <-ctx.Done():
            return
        case <-time.After(time.Hour):
            atomic.AddInt64(&ran, 1)
        }
    })
    time.Sleep(10 * time.Millisecond)
    cancel()
    pool.StopWait()
    if atomic.LoadInt64(&ran) != 0 {
        t.Fatal("task should have honoured context cancellation")
    }
}

var _ = errors.New // keep imports tidy