05 - Worker Pools

🎁 Spawning one goroutine per job works great until you spawn 50,000 of them and your program runs out of file descriptors. Worker pools give you controlled parallelism.

A worker pool is a fixed number of goroutines pulling jobs from a shared queue — the defining idea is bounded concurrency: you cap how many things run at once, regardless of how much work arrives. Pool workers are often long-lived, staying alive to wait for more jobs, though that's a configuration choice, not a rule (the batch example below exits once jobs run out).

This is one of the most common patterns in production Go.

Why Worker Pools

You could spin up a goroutine for every job. With 10,000 jobs, that's 10,000 goroutines. Goroutines are cheap, but each one might hold a database connection, an HTTP client, or a file handle. Those aren't cheap. A worker pool gives you controlled parallelism — fixed resource usage, predictable behavior, and backpressure when work piles up.

Use worker pools when:

• You're processing a queue of jobs (image resizing, email sending, API calls)
• You need to limit concurrent access to a resource (DB connections, file handles)
• Jobs arrive continuously and workers should stay alive between jobs

Why Worker Pools Over Fan-Out/Fan-In?

These aren't really competing patterns — a worker pool is one way to implement a fan-out stage. The difference is what each one is about. Fan-out/fan-in describes a dataflow shape: how work is routed to workers and recombined. A worker pool is a resource-control construct: how you bound concurrency so a burst of jobs doesn't exhaust connections, memory, or file handles.

In practice they show up differently. A worker pool keeps things simple — all workers share one input and one output channel, no merge step. Fan-out/fan-in gives you composability — each worker can have its own output channel, so you can chain the stage into a larger pipeline, or let workers produce different result types (like the Google Search example — web, image, video).

Reach for a plain worker pool when the goal is "process these jobs with bounded concurrency." Reach for explicit fan-out/fan-in when you're building a multi-stage pipeline where parallel steps chain together.

Basic Worker Pool

package main

func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("worker %d processing job %d\n", id, job)
        time.Sleep(500 * time.Millisecond) // simulate work
        results <- job * 2
    }
}

func main() {
    const numJobs = 10
    const numWorkers = 3

    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)
    var wg sync.WaitGroup

    // Start workers
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // Send jobs
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs)

    // Wait for workers, then close results
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    for r := range results {
        fmt.Println("result:", r)
    }
}

Three workers, ten jobs. Workers pull from the jobs channel until it's closed. Results flow into a buffered channel. The closing goroutine waits for all workers to finish before closing results.

Notice who closes what: main closes jobs because it's the producer of jobs. But workers are the producers of results — so who closes results? No single worker can, because they all share it. The wg.Wait() goroutine acts as a coordinator: it waits for all producers to finish, then closes the channel. This is the standard pattern when multiple goroutines write to one channel.

Worker Pool with Context

Add cancellation so the pool can be stopped mid-flight.

func worker(ctx context.Context, id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("worker %d cancelled\n", id)
            return
        case job, ok := <-jobs:
            if !ok {
                return // channel closed, no more jobs
            }
            fmt.Printf("worker %d processing job %d\n", id, job)
            time.Sleep(500 * time.Millisecond) // simulate work

            select {
            case results <- job * 2:
            case <-ctx.Done():
                return
            }
        }
    }
}

The double select is important.

1. First select: worker is idle, waiting for a job or cancellation. If the context is cancelled while waiting, the worker exits instead of picking up more work.

2. Second select: worker finished the job and wants to send the result. But if nobody is reading from results anymore (context was cancelled), results <- job * 2 would block forever. The second select prevents that deadlock.

Graceful Shutdown

There are two ways to stop a worker pool:

Graceful — close the jobs channel. Workers finish whatever they're processing, drain remaining jobs, then exit.

func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)
    var wg sync.WaitGroup

    for i := 0; i < 3; i++ {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    for j := 0; j < 20; j++ {
        jobs <- j
    }
    close(jobs) // no more jobs — workers drain what's left and exit

    wg.Wait()      // wait for all workers to finish
    close(results)  // safe to close — all producers are done

    for r := range results {
        fmt.Println("result:", r)
    }
}

Emergency — cancel the context. Workers drop everything and exit immediately, even if jobs remain in the queue.

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    jobs := make(chan int, 100)
    results := make(chan int, 100)
    var wg sync.WaitGroup

    for i := 0; i < 3; i++ {
        wg.Add(1)
        go worker(ctx, i, jobs, results, &wg)
    }

    for j := 0; j < 20; j++ {
        jobs <- j
    }

    cancel() // stop now — workers exit even if jobs remain

    wg.Wait()
    close(results)

    for r := range results {
        fmt.Println("result:", r)
    }
}

Use graceful shutdown by default. Use context cancellation when something goes wrong — a timeout, a signal from the OS, or an unrecoverable error.

Failure Mode: Closing results Too Early

The ordering of wg.Wait() and close(results) is not stylistic — get it wrong and the program panics. A tempting mistake is to close results right after sending the jobs:

func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)
    var wg sync.WaitGroup

    for i := 0; i < 3; i++ {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    for j := 0; j < 20; j++ {
        jobs <- j
    }
    close(jobs)
    close(results) // ❌ workers are still running and sending!

    wg.Wait()

    for r := range results {
        fmt.Println("result:", r)
    }
}

The workers are still draining jobs and calling results <- job * 2. Sending on a closed channel is a panic (send on closed channel), not a silent no-op — and because it happens inside a worker goroutine, it crashes the whole program. The rule: close a channel only after every goroutine that sends on it has finished. That's why the correct version waits first (wg.Wait()), then closes — and puts the close in a coordinating goroutine when the consumer needs to read concurrently:

go func() {
    wg.Wait()      // all senders done
    close(results) // now safe
}()

Struct-Based Worker Pool

The previous examples wire up channels, WaitGroups, and goroutines every time. When you need a pool in multiple places, wrap it in a struct — callers just use Submit and Shutdown without managing internals.

type Pool struct {
    jobs    chan func()
    wg      sync.WaitGroup
}

func NewPool(size int) *Pool {
    p := &Pool{
        jobs: make(chan func(), size*2),
    }
    for i := 0; i < size; i++ {
        p.wg.Add(1)
        go p.run()
    }
    return p
}

func (p *Pool) run() {
    defer p.wg.Done()
    for fn := range p.jobs {
        fn()
    }
}

func (p *Pool) Submit(fn func()) {
    p.jobs <- fn
}

func (p *Pool) Shutdown() {
    close(p.jobs)
    p.wg.Wait()
}

Usage:

func main() {
    pool := NewPool(4)

    for i := 0; i < 20; i++ {
        i := i // not needed in Go 1.22+ (loop vars are per-iteration)
        // before 1.22, all closures shared the same i — without this, every goroutine would print 20
        pool.Submit(func() {
            fmt.Printf("processing %d\n", i)
            time.Sleep(100 * time.Millisecond) // simulate work
        })
    }

    pool.Shutdown()
    fmt.Println("all done")
}

Submit sends a closure to the pool. Shutdown closes the channel and waits. Clean, reusable, and generic — the pool doesn't care what work it's doing.

Output order is non-deterministic — workers run concurrently, so numbers appear in arbitrary order. If it looks ordered on the Go Playground, that's because the Playground runs with GOMAXPROCS=1 and a deterministic scheduler. On a real machine with multiple cores, you'll see interleaving.

Buffered vs Unbuffered Job Channel

Use buffered for bursty workloads. Use unbuffered when you want the caller to slow down if workers can't keep up.

Practical Example: Concurrent URL Fetcher

type FetchResult struct {
    URL    string
    Status int
    Err    error
}

func fetchWorker(ctx context.Context, urls <-chan string, results chan<- FetchResult, wg *sync.WaitGroup) {
    defer wg.Done()
    client := &http.Client{Timeout: 10 * time.Second}

    for url := range urls {
        select {
        case <-ctx.Done():
            return
        default:
        }

        req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
        if err != nil {
            results <- FetchResult{URL: url, Err: err}
            continue
        }

        resp, err := client.Do(req)
        if err != nil {
            results <- FetchResult{URL: url, Err: err}
            continue
        }
        resp.Body.Close()
        results <- FetchResult{URL: url, Status: resp.StatusCode}
    }
}

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

    urls := []string{
        "https://go.dev",
        "https://pkg.go.dev",
        "https://github.com",
        "https://example.com",
        "https://httpbin.org/get",
    }

    urlCh := make(chan string, len(urls))
    results := make(chan FetchResult, len(urls))
    var wg sync.WaitGroup

    // Start 3 workers
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go fetchWorker(ctx, urlCh, results, &wg)
    }

    // Send URLs
    for _, u := range urls {
        urlCh <- u
    }
    close(urlCh)

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

    for r := range results {
        if r.Err != nil {
            fmt.Printf("FAIL %s: %v\n", r.URL, r.Err)
        } else {
            fmt.Printf("OK   %s: %d\n", r.URL, r.Status)
        }
    }
}

Three workers fetch URLs concurrently. Context handles timeout. Results are collected through a single channel. This is the worker pool pattern in action.

When NOT to Use a Worker Pool

A worker pool is the right default for job processing, but it's not always the simplest tool:

• You just need to bound concurrency, not process a queue. If you only want "run these N tasks, but no more than K at a time," a semaphore is less code than wiring up a job channel, results channel, and WaitGroup. Lesson 07 covers this — including errgroup.Group.SetLimit, which gives you bounded concurrency and error propagation in a few lines.
• The work is a fixed, small set of tasks. Launching a handful of goroutines and waiting on a sync.WaitGroup (or errgroup) is clearer than a pool built for a continuous stream.
• Tasks have dependencies or need per-task errors/results. A pool that runs bare func() closures hides which task failed. Prefer errgroup or a pipeline when you need to collect errors or chain stages.
• You need per-time-unit limits, not per-moment limits. A pool caps how many run at once, but not how many run per second. For "100 requests/second" you need rate limiting (lesson 06), which is a different constraint than pool size.

Rule of thumb: reach for a worker pool when work arrives continuously and you want long-lived, reusable workers. For one-off bounded parallelism, a semaphore or errgroup is usually simpler.

Key Takeaways

• Worker pool: fixed goroutines pulling from a shared job channel
• Close the job channel for graceful shutdown — workers drain remaining work
• Use context cancellation for emergency stops
• Double select pattern: check context on both receive and send
• Struct-based pools are reusable — accept func() for generic work
• Buffer the job channel for bursty workloads, unbuffered for backpressure
• A worker pool is about bounded concurrency (resource control); fan-out/fan-in is about dataflow shape — a pool is one way to implement a fan-out stage, not a competing pattern

🎁 Your worker pool can process 10,000 jobs per second — but the downstream API you're calling only allows 100 per second. How do you enforce that limit without blocking everything?