Implementing a Bounded Worker Pool for Concurrency Control

Hello. In the previous lesson, you made blocked channel operations responsive to cancellation. Now you can use those same lifecycle rules to build a worker pool: a fixed number of goroutines share a stream of jobs, so no more than that number can process work at once.

This lesson takes about 40 minutes. You’ll implement a pool, check that its concurrency limit holds, and distinguish limiting active work from merely creating a queue.

What the limit actually limits

Suppose you have 1,000 independent jobs, but each job uses a connection or a substantial amount of memory. Starting 1,000 goroutines may start too much work at once. With three workers reading from one jobs channel, at most three calls to the processing function can be in progress.

Start with Go by Example: Worker Pools. Read the worker and main code as one small program.

The example buffers both channels for every job. That is convenient for a five-job demonstration, but it does not bound the number of jobs waiting in memory. The worker count caps simultaneous processing; the jobs-channel capacity controls how many jobs can wait in the channel. An unbuffered jobs channel provides a simple form of backpressure: the sender must wait until a worker accepts each job.

Implement a pool with a clear lifetime

The following pool takes a slice of inputs and returns results in input order. Its processing function is synchronous: each worker calls it once per job and does not accept another job until that call returns. Save this as main.go.

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

type job struct {
	index int
	value int
}

func runPool(
	ctx context.Context,
	inputs []int,
	limit int,
	process func(context.Context, int) (int, error),
) ([]int, error) {
	if limit < 1 {
		return nil, fmt.Errorf("worker limit must be positive")
	}

	jobs := make(chan job) // No queued jobs in the channel.
	results := make([]int, len(inputs))
	jobErrors := make([]error, len(inputs))

	var workers sync.WaitGroup
	workers.Add(limit)

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

			for {
				select {
				case <-ctx.Done():
					return
				case j, ok := <-jobs:
					if !ok {
						return
					}
					results[j.index], jobErrors[j.index] =
						process(ctx, j.value)
				}
			}
		}()
	}

dispatch:
	for i, value := range inputs {
		select {
		case <-ctx.Done():
			break dispatch
		case jobs <- job{index: i, value: value}:
		}
	}

	close(jobs)  // This function is the sole sender.
	workers.Wait() // All writes finish before results are inspected.

	if err := ctx.Err(); err != nil {
		return nil, err
	}
	for _, err := range jobErrors {
		if err != nil {
			return nil, err
		}
	}
	return results, nil
}

func doubleAfterDelay(ctx context.Context, n int) (int, error) {
	timer := time.NewTimer(25 * time.Millisecond)
	defer timer.Stop()

	select {
	case <-ctx.Done():
		return 0, ctx.Err()
	case <-timer.C:
		return n * 2, nil
	}
}

func main() {
	got, err := runPool(
		context.Background(),
		[]int{1, 2, 3, 4, 5},
		2,
		doubleAfterDelay,
	)
	fmt.Println(got, err)
}

Run go run main.go. The result is [2 4 6 8 10] <nil>, regardless of which worker finishes a particular job first. Because there are two workers, the five delayed jobs finish in roughly three rounds; elapsed time is not itself a reliable proof of the concurrency bound.

Several details make this more than a loop that launches goroutines:

• workers.Add(limit) happens before the goroutines start and before Wait. Each worker calls Done exactly once when it exits.
• The caller is the only sender to jobs, so it closes jobs after dispatch ends. A worker receives ok == false when no more jobs are coming.
• A canceled context can release the caller from a blocked job send and workers from blocked receives. The processing function also checks the context; a select around channel operations cannot interrupt a processing function that ignores cancellation.
• Each job has a distinct index. Workers write to separate, already-allocated slice elements; they do not append to the slices. Wait completes before the caller reads those elements. There is no shared slice element being modified concurrently.

This pool’s error policy is deliberately straightforward: a processing error is recorded, but other submitted jobs continue. After every worker exits, the function returns the first error in input order. Cancellation takes precedence and may leave some result slots unwritten; in that case, the function returns no results. A select can still accept a ready job when cancellation is also ready, so cancellation means stop promptly, not guarantee that no further job starts.

Verify the bound rather than inferring it from speed

A test can count calls currently inside process. The following one makes the first three calls wait until all three have started. That checks that the pool can use its configured capacity; its peak count checks that it does not exceed it. Save it as main_test.go.

package main

import (
	"context"
	"reflect"
	"sync"
	"testing"
	"time"
)

func TestRunPoolBoundsWorkAndPreservesOrder(t *testing.T) {
	ctx, cancel := context.WithTimeout(
		context.Background(), 3*time.Second,
	)
	defer cancel()

	var mu sync.Mutex
	active, peak := 0, 0
	released := false
	threeStarted := make(chan struct{})

	process := func(ctx context.Context, n int) (int, error) {
		mu.Lock()
		active++
		if active > peak {
			peak = active
		}
		if active == 3 && !released {
			released = true
			close(threeStarted)
		}
		mu.Unlock()

		defer func() {
			mu.Lock()
			active--
			mu.Unlock()
		}()

		select {
		case <-ctx.Done():
			return 0, ctx.Err()
		case <-threeStarted:
			return n * 2, nil
		}
	}

	got, err := runPool(
		ctx, []int{1, 2, 3, 4, 5, 6}, 3, process,
	)
	if err != nil {
		t.Fatal(err)
	}

	want := []int{2, 4, 6, 8, 10, 12}
	if !reflect.DeepEqual(got, want) {
		t.Fatalf("results = %v, want %v", got, want)
	}
	if peak != 3 {
		t.Fatalf("peak active calls = %d, want 3", peak)
	}
}

Run go test -race main.go main_test.go. The timeout is a failure bound: if fewer than three workers can enter process, the test returns an error rather than waiting forever. The mutex protects the test’s counters, while the race detector also checks the pool’s concurrent result writes.

The function does not return until workers.Wait() completes. That matters when using the pool inside a request or command: the caller receives a result and knows that this invocation left no worker goroutines running.

A fixed pool is a choice, not the only limiter

Watch the worker-pool portion of Bryan C. Mills’s Rethinking Classical Concurrency Patterns. It places the fixed-worker design alongside another way to limit active work.

An on-demand design can acquire a slot before launching each job goroutine, avoiding idle workers. A fixed pool is useful when jobs arrive through a shared channel and a stable group of workers should consume them. Neither design is safe merely because it has a concurrency limit: you still need to decide how submissions stop, how in-progress work responds to cancellation, and how completion is observed.

Also keep the two bounds separate. This lesson’s unbuffered channel prevents a backlog inside the jobs channel; the caller still holds the entire inputs slice. Streaming an arbitrarily large input without retaining it all requires an input source that produces jobs incrementally.

Takeaways

A bounded worker pool starts a fixed number of workers, feeds them jobs, and waits for them to exit. The worker count limits simultaneous processing; channel capacity separately controls queued jobs. Keep the jobs channel’s closer on the sending side, make blocking operations cancellable, and wait before reading worker-written results or claiming shutdown is complete.

This finishes the module’s concurrency and lifecycle patterns. In the next module, you’ll measure concurrent behavior—including whether a chosen concurrency limit improves throughput without unacceptable latency.