Worker Pools in Go: Bounded Concurrency Without Goroutine Leaks
go doSomething(item) inside a for loop is the fastest way to turn a bounded workload into an unbounded one. One goroutine per file, per row, per HTTP request. It works fine on your laptop with a test fixture of forty items, and then someone points it at a directory with 400,000 files and the process gets OOM-killed.
A worker pool fixes that by limiting the number of goroutines and passing work through a channel. The Go blog’s pipelines article calls this “bounded parallelism”, and its bounded.go example is a useful reference. The loop is short; the important details are who closes each channel, how errors get out, how cancellation propagates, and how you know the pool has actually stopped.
The three-stage shape of a worker pool
A common Go worker-pool design has three moving parts. A producer writes jobs to a channel and closes it when it runs out. N workers range over that job channel and write to a shared results channel. A separate closer goroutine waits for all the workers to finish, then closes the results channel.
That third goroutine is the part people leave out, and it’s the part that determines whether your program panics. The producer owns the jobs channel; nobody else touches close on it. The workers do not own the results channel, because several of them send on it, and sends on a closed channel panic. So the close has to happen somewhere that knows all senders are finished, which is exactly what a sync.WaitGroup tells you. The Go blog is blunt about it: “digester does not close its output channel, as multiple goroutines are sending on a shared channel.”
A useful ownership rule is that the sending side closes the channel. When there are many senders, a WaitGroup can establish when the last sender has finished. If channel close semantics still feel slippery, our guide to Go channels covers the rules.
A pool with context cancellation
The 2014 blog post uses a done chan struct{} closed via defer. For request-scoped work, context.Context adds deadlines and carries cancellation across API boundaries. Selecting on ctx.Done() gives workers the same kind of broadcast cancellation signal.
package main
import (
"context"
"errors"
"fmt"
"sync"
)
type Job struct {
ID int
URL string
}
type Result struct {
JobID int
Bytes int
Err error
}
// process is the unit of work. It must respect ctx itself; a worker pool
// can't interrupt a blocking call that ignores cancellation.
func process(ctx context.Context, j Job) (int, error) {
select {
case <-ctx.Done():
return 0, ctx.Err()
default:
}
// pretend this does real I/O keyed off ctx
return len(j.URL), nil
}
func RunPool(ctx context.Context, jobs []Job, workers int) ([]Result, error) {
if workers <= 0 {
return nil, fmt.Errorf("workers must be positive")
}
in := make(chan Job)
out := make(chan Result)
// Producer: owns `in`, closes it on every return path.
go func() {
defer close(in)
for _, j := range jobs {
select {
case in <- j:
case <-ctx.Done():
return
}
}
}()
// Workers: shared sender on `out`, so they never close it.
var wg sync.WaitGroup
wg.Add(workers)
for i := 0; i < workers; i++ {
go func() {
defer wg.Done()
for j := range in {
n, err := process(ctx, j)
select {
case out <- Result{JobID: j.ID, Bytes: n, Err: err}:
case <-ctx.Done():
return
}
}
}()
}
// Closer: the only goroutine allowed to close `out`.
go func() {
wg.Wait()
close(out)
}()
var results []Result
for r := range out {
results = append(results, r)
}
if err := ctx.Err(); err != nil {
return results, err
}
return results, nil
}
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
jobs := []Job{{1, "a"}, {2, "bb"}, {3, "ccc"}}
res, err := RunPool(ctx, jobs, 2)
if err != nil && !errors.Is(err, context.Canceled) {
fmt.Println("pool error:", err)
return
}
for _, r := range res {
fmt.Printf("job %d -> %d bytes (err=%v)\n", r.JobID, r.Bytes, r.Err)
}
}
Three details in there carry more weight than the rest of the code.
Every potentially blocking send sits inside a select with ctx.Done(). That is one essential part of leak prevention; the work performed by process must also observe the context. A worker blocked on out <- result with nobody reading cannot exit, and the runtime will not clean it up for you. As the Go blog explains, goroutine stacks can also keep referenced data alive. We walk through more shapes of this bug in common goroutine leaks in Go.
The consumer drains out all the way to close. Even when you cancel early, ranging until the channel closes guarantees each worker’s pending send either lands or falls through to ctx.Done(). Break out of that range loop without cancelling first and you’ve reintroduced the leak you just spent a select avoiding.
And wg.Add(workers) happens before the loop, not inside each goroutine. The sync documentation requires a positive Add that starts from zero to happen before Wait; putting it inside the goroutine lets the closer observe a zero counter and close out too early.
Backpressure: unbuffered vs buffered job channels
make(chan Job) is unbuffered, so the producer blocks until a worker is ready to receive. That gives you backpressure: the channel itself cannot accumulate queued jobs, and the producer paces itself against actual consumption.
Add a buffer when the producer is bursty and you want to smooth the gaps:
in := make(chan Job, workers*2)
A small buffer can smooth short gaps in production. A large buffer can simply relocate the memory problem into the channel. If it is large enough to hold the entire input, a slice plus an explicit work index may be easier to reason about.
The Go blog flags the same trap for buffered fan-in: “The choice of buffer size of 1 here depends on knowing the number of values merge will receive and the number of values downstream stages will consume. This is fragile.”
Error propagation: fail fast with errgroup
Stuffing errors into Result.Err is right when you want every job attempted and a report at the end. When the first failure should stop the world, golang.org/x/sync/errgroup deletes most of the plumbing. errgroup.WithContext cancels the derived context on the first non-nil error, and SetLimit bounds concurrency with no worker loop at all.
package main
import (
"context"
"fmt"
"golang.org/x/sync/errgroup"
)
func FetchAll(ctx context.Context, urls []string, limit int) ([]string, error) {
if limit <= 0 {
return nil, fmt.Errorf("limit must be positive")
}
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(limit) // bounded concurrency, no manual worker loop
// Index-addressed results: no mutex, no results channel.
out := make([]string, len(urls))
for i, u := range urls {
i, u := i, u // pre-Go 1.22 loop variable capture
g.Go(func() error {
body, err := fetch(ctx, u)
if err != nil {
return fmt.Errorf("fetch %s: %w", u, err)
}
out[i] = body
return nil
})
}
if err := g.Wait(); err != nil {
return nil, err
}
return out, nil
}
func fetch(ctx context.Context, url string) (string, error) {
select {
case <-ctx.Done():
return "", ctx.Err()
default:
return "body of " + url, nil
}
}
Two things you get for free here. g.SetLimit(limit) makes g.Go block once limit goroutines are already running, so the for loop becomes the backpressure mechanism. And writing out[i] from separate goroutines needs no lock, because each goroutine owns a distinct element and g.Wait() supplies the happens-before edge the reader needs.
The cost: errgroup.Wait returns the first non-nil error. If you need all of them for a report, go back to the results-channel design and aggregate with errors.Join. Our post on error handling in Go covers wrapping and joining.
Shutdown: knowing the pool is actually done
“Cancelled” and “stopped” are different states, and conflating them is how you get a use-after-close on a database handle. cancel() returns instantly; the workers are still unwinding somewhere behind you. If your pool holds a connection or an open file, you have to wait for the goroutines to exit before you release anything.
For a long-lived pool, make that wait part of the API:
type Pool struct {
jobs chan Job
wg sync.WaitGroup
ctx context.Context
cancel context.CancelFunc
mu sync.RWMutex
closed bool
}
var ErrPoolClosed = errors.New("pool is closed")
func NewPool(ctx context.Context, workers int) *Pool {
if workers <= 0 {
panic("workers must be positive")
}
ctx, cancel := context.WithCancel(ctx)
p := &Pool{
jobs: make(chan Job),
ctx: ctx,
cancel: cancel,
}
p.wg.Add(workers)
for i := 0; i < workers; i++ {
go func() {
defer p.wg.Done()
for {
select {
case j, ok := <-p.jobs:
if !ok {
return // graceful: queue drained
}
_, _ = process(ctx, j)
case <-ctx.Done():
return // abrupt: abandon queued work
}
}
}()
}
return p
}
// Submit blocks until a worker is free or the pool shuts down.
func (p *Pool) Submit(ctx context.Context, j Job) error {
p.mu.RLock()
defer p.mu.RUnlock()
if p.closed {
return ErrPoolClosed
}
select {
case p.jobs <- j:
return nil
case <-ctx.Done():
return ctx.Err()
case <-p.ctx.Done():
return p.ctx.Err()
}
}
func (p *Pool) closeJobs() {
p.mu.Lock()
defer p.mu.Unlock()
if !p.closed {
p.closed = true
close(p.jobs)
}
}
// Shutdown stops accepting work, drains what's queued, and waits.
func (p *Pool) Shutdown() {
p.closeJobs()
p.wg.Wait()
}
// Abort cancels in-flight work and waits for workers to exit.
func (p *Pool) Abort() {
p.cancel()
p.closeJobs()
p.wg.Wait()
}
Shutdown is drain-then-stop; Abort signals cancellation before closing the queue. Both end in p.wg.Wait(), which is the honest signal that no worker goroutine is still running. http.Server draws a similar line between Shutdown and Close; see graceful shutdown for Go HTTP servers for that version of the argument.
The mutex matters because checking an atomic “closed” flag is not enough on its own: shutdown could close the channel between that check and the send. Holding a read lock across Submit prevents closeJobs from closing the channel until an in-progress send has completed or been cancelled, and later submissions return ErrPoolClosed.
Testing that the pool doesn’t leak
This is the bug class that passes your test suite. A leaked goroutine fails no assertion and prints no error; the test goes green and the memory graph goes up. go.uber.org/goleak turns it into a test failure.
package pool
import (
"context"
"errors"
"testing"
"time"
"go.uber.org/goleak"
)
func TestMain(m *testing.M) {
goleak.VerifyTestMain(m)
}
func TestPoolExitsOnCancel(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
cancel()
jobs := make([]Job, 1000)
for i := range jobs {
jobs[i] = Job{ID: i, URL: "x"}
}
done := make(chan error, 1)
go func() {
_, err := RunPool(ctx, jobs, 4)
done <- err
}()
select {
case err := <-done:
if !errors.Is(err, context.Canceled) {
t.Fatalf("RunPool error = %v, want context.Canceled", err)
}
case <-time.After(time.Second):
t.Fatal("pool did not exit after cancel")
}
}
Run it with -race too. The race detector only finds races on executed code paths, so pair it with tests that exercise cancellation, shutdown, and concurrent submissions.
Choosing the worker count
There’s no universal number, and anyone who gives you one hasn’t seen your workload. The Go blog’s bounded.go hardcodes numDigesters = 20 for file hashing, which is a defensible I/O-bound guess and nothing more.
For CPU-bound work, runtime.GOMAXPROCS(0) is a sensible starting point. More workers often add scheduling overhead without adding CPU throughput, so measure before increasing it.
For I/O-bound work, the practical ceiling is often the downstream service. Respect its connection and concurrency limits, and add a rate limiter when it publishes a request rate. Two hundred goroutines pointed at a database with twenty connections still leave up to 180 goroutines waiting for a connection; that extra concurrency may add no throughput.
Mixed workloads deserve two pools with two limits, not one compromise number that’s wrong for both halves.
The number you pick matters less than whether you measured it. A pool of 8 that saturates the remote API beats a pool of 500 that trips its throttle and spends the afternoon retrying.