Bounded concurrency with cancellation in Go¶

A backend service often needs to process many independent items: enrich records, call downstream APIs, resize files, or run background jobs. Starting one goroutine per item is easy, but it gives the caller no natural limit on concurrency.

A bounded worker pool gives you an explicit concurrency limit. Passing a context.Context through the pipeline lets the caller stop queued work and signal in-flight operations when the request is canceled or one worker fails.

This guide builds a small, dependency-free worker pool and explains the production trade-offs.

The implementation¶

Runnable example¶

The complete package and tests live in runnable Go worker-pool example. From that directory, run go test ./..., go test -race ./..., and go vet ./.... The handbook validation workflow runs these checks on every handbook change.

The example has four important properties:

At most workers calls to process run concurrently.
Results retain input order, even when workers finish out of order.
The first processing error cancels the shared context.
The producer and workers observe cancellation, so they can stop instead of blocking forever.

Create workerpool.go:

package workerpool

import (
    "context"
    "errors"
    "fmt"
    "sync"
)

var ErrInvalidWorkerCount = errors.New("worker count must be positive")

type job struct {
    index int
    value int
}

// Map runs process with a fixed number of workers and preserves input order.
// process must honor ctx when it performs blocking or expensive work.
func Map(
    parent context.Context,
    inputs []int,
    workers int,
    process func(context.Context, int) (int, error),
) ([]int, error) {
    if parent == nil {
        return nil, errors.New("parent context must not be nil")
    }
    if workers <= 0 {
        return nil, ErrInvalidWorkerCount
    }
    if process == nil {
        return nil, errors.New("process function must not be nil")
    }
    if err := parent.Err(); err != nil {
        return nil, err
    }
    if len(inputs) == 0 {
        return []int{}, nil
    }

    ctx, cancel := context.WithCancelCause(parent)
    defer cancel(nil)

    jobs := make(chan job)
    results := make([]int, len(inputs))

    var wg sync.WaitGroup
    wg.Add(workers)

    for range workers {
        go func() {
            defer wg.Done()

            for {
                select {
                case <-ctx.Done():
                    return
                case item, ok := <-jobs:
                    if !ok {
                        return
                    }
                    if ctx.Err() != nil {
                        return
                    }

                    result, err := process(ctx, item.value)
                    if err != nil {
                        cancel(fmt.Errorf("process input at index %d: %w", item.index, err))
                        return
                    }
                    results[item.index] = result
                }
            }
        }()
    }

produce:
    for index, value := range inputs {
        select {
        case <-ctx.Done():
            break produce
        case jobs <- job{index: index, value: value}:
        }
    }
    close(jobs)
    wg.Wait()

    if err := context.Cause(ctx); err != nil {
        return nil, err
    }
    return results, nil
}

The for range workers integer form requires Go 1.22 or later. If a project targets an older Go version, use a conventional counted loop instead.

Tests¶

Create workerpool_test.go:

package workerpool

import (
    "context"
    "errors"
    "reflect"
    "testing"
)

func TestMapPreservesInputOrder(t *testing.T) {
    got, err := Map(context.Background(), []int{1, 2, 3, 4}, 3,
        func(_ context.Context, value int) (int, error) {
            return value * value, nil
        },
    )
    if err != nil {
        t.Fatal(err)
    }

    want := []int{1, 4, 9, 16}
    if !reflect.DeepEqual(got, want) {
        t.Fatalf("Map() = %v, want %v", got, want)
    }
}

func TestMapReturnsProcessingError(t *testing.T) {
    wantErr := errors.New("downstream unavailable")

    _, err := Map(context.Background(), []int{1, 2, 3}, 1,
        func(_ context.Context, value int) (int, error) {
            if value == 2 {
                return 0, wantErr
            }
            return value, nil
        },
    )
    if !errors.Is(err, wantErr) {
        t.Fatalf("Map() error = %v, want wrapped %v", err, wantErr)
    }
}

func TestMapRejectsInvalidWorkerCount(t *testing.T) {
    _, err := Map(context.Background(), []int{1}, 0,
        func(_ context.Context, value int) (int, error) {
            return value, nil
        },
    )
    if !errors.Is(err, ErrInvalidWorkerCount) {
        t.Fatalf("Map() error = %v, want %v", err, ErrInvalidWorkerCount)
    }
}

Run the tests and race detector:

go test ./...
go test -race ./...
go vet ./...

Why cancellation is part of the design¶

A worker pool without cancellation can continue doing work after the HTTP request that started it has already timed out. It can also leave a producer blocked while sending to a channel that nobody is receiving from.

The derived context addresses both cases:

The caller cancels the parent context when the request is abandoned.
A worker cancels the derived context when processing fails.
The producer selects on ctx.Done() while sending jobs.
Workers select on ctx.Done() while waiting for jobs.
The caller waits for all workers before returning.

Cancellation is cooperative. If process ignores its context and blocks forever, the worker pool cannot forcibly stop it. Database calls, HTTP requests, and other blocking operations should use context-aware APIs and enforce their own deadlines.

Production considerations¶

Choose the worker count deliberately¶

A worker limit is a capacity control, not a universal performance setting.

For CPU-bound work, start near the available CPU capacity and benchmark.
For I/O-bound work, a higher limit may improve throughput, but downstream connection pools and rate limits still apply.
Bound the input queue as well if jobs arrive continuously; an unbounded queue simply moves the overload problem into memory.

Preserve backpressure¶

This implementation uses an unbuffered jobs channel. The producer can hand off a job only when a worker is ready. That naturally limits queued work, but it also means the producer participates in the workers' pace.

A buffered channel can improve throughput for some workloads, but its capacity should be an explicit limit with a reason.

Be careful with shared state¶

Each worker writes to a distinct result index. No two workers write the same index, and the caller reads results only after wg.Wait() completes. If workers update shared maps, counters, or caches, add appropriate synchronization or use a single aggregation goroutine.

Understand partial work¶

When one worker fails, other workers may already have completed side effects. Cancellation does not roll back database writes, messages, or remote API calls. Use idempotency keys, transactions, compensating actions, or durable job state where the business operation requires stronger guarantees.

Interview discussion¶

Why not start one goroutine per input?

That may be fine for a small, bounded batch. For large or unbounded input, it can create excessive scheduling overhead and overwhelm downstream services. A fixed worker count gives an explicit concurrency ceiling.

Why pass context as the first parameter?

It makes cancellation and deadlines explicit at API boundaries. Go's context guidance recommends passing it to each function that needs it rather than storing it in a struct.

Does canceling a context stop a goroutine?

No. It closes the context's Done channel. The goroutine must observe that signal and return cooperatively.

Why wait for workers after cancellation?

To ensure this function does not return while its worker goroutines are still using local state or performing work. The function can only finish once those goroutines exit.

Does cancellation undo work already completed?

No. Cancellation controls future and in-flight work that cooperates with the context; it is not a transaction or rollback mechanism.