Worker pools, errgroup, and the race detector

50 min · lesson 4 of 4 · #concurrency · #testing

Worker pool: a channel, N goroutines, two closers

package main

import (
	"fmt"
	"sync"
)

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

	for w := 0; w < 3; w++ { // 3 workers, fixed
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := range jobs { // ends when jobs is closed and drained
				results <- j * j
			}
		}()
	}

	go func() {
		for i := 1; i <= 6; i++ {
			jobs <- i
		}
		close(jobs) // producer closes jobs
	}()

	go func() {
		wg.Wait()      // closer waits for workers
		close(results) // then closes results
	}()

	sum := 0
	for r := range results {
		sum += r
	}
	fmt.Println("sum of squares 1..6 =", sum) // 91
}

// Java: the pool owns the threads, you never write the loop
var pool = Executors.newFixedThreadPool(3);
var futures = IntStream.rangeClosed(1, 6).mapToObj(i -> pool.submit(() -> i * i)).toList();
int sum = 0;
for (var f : futures) sum += f.get();
pool.shutdown();

// Node: concurrency is a promise pool (p-limit), because there are no threads
import pLimit from 'p-limit';
const limit = pLimit(3);
const sum = (await Promise.all([1,2,3,4,5,6].map(i => limit(() => i * i)))).reduce((a, b) => a + b, 0);

This is the shape for every bounded-concurrency job: N workers, one jobs channel, one results channel, two closers. Any deviation you see in a codebase is usually a leak.

errgroup: the pool you should write instead

golang.org/x/sync/errgroup is the standard answer for "run these N things, fail fast, propagate the first error":

g, ctx := errgroup.WithContext(ctx) // ctx is cancelled when any fn returns an error
g.SetLimit(8)                       // bounded concurrency, no manual semaphore

for _, url := range urls {
	url := url
	g.Go(func() error {
		req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
		if err != nil {
			return err
		}
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			return fmt.Errorf("fetch %s: %w", url, err)
		}
		defer resp.Body.Close()
		return nil
	})
}
if err := g.Wait(); err != nil { // first error wins; others are dropped
	return err
}

SetLimit + WithContext replaces the hand-rolled semaphore-and-WaitGroup pattern that Go code used to be full of. The Java equivalent is StructuredTaskScope.ShutdownOnFailure — same contract, different syntax.

Mutexes are not a failure

The proverb is "share memory by communicating", but the follow-up sentence matters: channels orchestrate, mutexes protect. A counter, a cache, a map of connections — those want a lock, not a goroutine with a channel.

package main

import (
	"fmt"
	"sync"
)

func main() {
	var mu sync.Mutex
	count := 0
	var wg sync.WaitGroup

	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			mu.Lock()
			count++ // critical section: read-modify-write
			mu.Unlock()
		}()
	}
	wg.Wait()
	fmt.Println(count) // 1000, deterministically
}

Remove the mu.Lock/Unlock and this still prints a number in the 900s — that is the whole problem with concurrency bugs. It fails silently, intermittently, and only in production.

The race detector is your compiler for concurrency

go test -race ./...        # instrumented build; catches unsynchronised access at runtime
go run -race .             # same for a binary
go test -race -count=10 ./pkg  # repeat: races are probabilistic, one run is not proof

When it fires, it prints both stacks and the memory address — the actual evidence, not a guess. It only reports races it observes, so: run it in CI on every PR, with -count for the flaky ones, and keep it enabled in integration tests. It costs ~2–10× CPU, which is nothing compared to a Friday-night incident.

Two things -race does not catch: deadlocks (use timeouts everywhere) and logical races (mutex-protected but wrong order of operations). go vet's copylocks check catches the other classic: copying a sync.Mutex by value.

The checks that belong in CI

go vet ./...                    # copylocks, printf args, lost cancels
staticcheck ./...               # the real linter: unused, ineffective, unsafe conversions
go test -race -shuffle=on ./...  # order-independent tests, race-instrumented
goleak.VerifyTestMain(m)        # in TestMain: fails the suite if goroutines outlive it

Trap: -shuffle=on is not cosmetic. Test order dependence is the most common way a package that "always passed" breaks the moment someone adds a test that shares a global.

Trap: sync.Mutex embedded in a struct is promoted to the public API (Lock/Unlock become callable by callers) and it makes the struct non-copyable. Use a named field: mu sync.Mutex.

Proof checkpoint

• Run the pool example with 3 workers vs 20, measure wall time with a sleep job — then explain the result in terms of buffering and backpressure.
• Deliberately race the counter, run go test -race -count=5, and paste the detector output.
• Rewrite a WaitGroup + error-channel loop as errgroup with SetLimit, and say what cancellation buys you.