Lab B · A worker pool that cancels cleanly

Objectives: Build a small fan-out worker pool from scratch, wire it to a context.Context so it stops promptly on cancellation, and size its results channel so it can never deadlock — then verify it under go test -race. Estimated time: 35 minutes.

This is a hands-on lab. You will write the pool in stages, break it deliberately, then fix it using the same reasoning the safeheaders-go team applied when they fixed real deadlocks in jsmn-go/parallel.go and stb-image-go/stb_image.go.

---

What this actually means (plain English)

No jargon — here's what the ideas in this lesson actually mean, and why they matter.

• Worker pool = "a team of cashiers sharing one queue, not one cashier per customer." A fixed number of goroutines drain a shared jobs channel; no matter how large the input, only that many goroutines ever run concurrently.
• Fan-out = "the manager deals every card face-down on the table before sitting back." The coordinator pushes all job indices into the channel up front, closes it, and blocks at wg.Wait() while workers pick items up.
• Context cancellation = "a fire alarm that every worker can hear from any floor." A context.Context carries a Done channel; when cancel fires, workers detect it via ctx.Err() or a ctx.Done() case and exit early rather than finishing unneeded work.
• Channel buffering = "a mail slot: if the slot is full, the letter carrier is stuck in the hallway until someone clears it." A buffered channel holds values without a receiver ready; if the buffer is smaller than the maximum possible sends, a sender goroutine blocks permanently and never calls wg.Done().
• sync.WaitGroup = "a departure board that flips to LANDED only when every flight has checked in." The coordinator calls wg.Add(1) per goroutine; each goroutine calls wg.Done() on exit; wg.Wait() blocks until the counter returns to zero.

Why it matters: the deadlock bug in jsmn-go (parallel.go) was caused by an under-sized channel buffer — safe to fix by understanding exactly how many sends can happen in the worst case.

See it — worker pool fan-out and the buffer sizing that prevents deadlock.

---

The bug template: why under-sizing the results channel deadlocks

Before building your own pool, look at the comment in jsmn-go/parallel.go (lines 51-55) that documents the real deadlock the team hit (a deadlock is a situation where two or more parts of a program are each waiting on the other to move first, so nothing ever proceeds and the program simply freezes):

// Buffer for the worst case so no worker can block on send (and thus never
// reach wg.Done): every job produces one result (numJobs) and, on context
// cancellation, each worker may emit one extra cancel result (numWorkers).
// An under-sized buffer here deadlocks wg.Wait on mid-parse cancellation.
resultsCh := make(chan chunkResult, numJobs+numWorkers)

In plain terms: this line reserves a mail-slot-like "channel" (a pipe goroutines use to pass values to each other) sized to hold numJobs+numWorkers items at once, so nobody sending a result ever gets stuck waiting for space.

The same pattern — for the same reason — appears in stb-image-go/stb_image.go (lines 106-109):

// Buffer the worst case so no worker blocks on send: up to len(datas) decode
// failures plus up to numWorkers cancellation sends. An under-sized buffer
// deadlocks wg.Wait when cancellation coincides with decode failures.
errs := make(chan error, len(datas)+numWorkers)

In plain terms: same idea as above, sized so every worker can report an error or a cancellation without ever getting stuck.

The rule is:

buffer size = (max sends on the happy path) + (max sends on the cancel path)

For a pool with numJobs jobs and numWorkers workers where each worker sends exactly one result per job and one extra send if the context fires:

buffer = numJobs + numWorkers

That's the formula you will use in this lab.

The silent failure mode

An under-sized resultsCh does not panic (in plain terms: the program doesn't crash with a visible error). The program just hangs at wg.Wait() forever, with no error message. This is why the jsmn-go team added a watchdog test (a small piece of code written specifically to check that another piece of code behaves correctly) that cancels mid-parse and asserts the function returns promptly (in plain terms: it checks that the function finishes and hands its result back quickly, rather than hanging) — the deadlock was invisible without it.

---

Step 1 — scaffold the package

Create a new directory outside the workspace so you can iterate freely:

mkdir -p /tmp/pool-lab && cd /tmp/pool-lab
go mod init pool-lab

In plain terms: the first line makes a new folder and moves into it; the second line turns that folder into a Go "package" — a named, importable unit of code (a package is just a folder of Go source files that other code can pull in with an import statement) — and names it pool-lab.

Create pool.go:

package poollab

import (
    "context"
    "fmt"
    "runtime"
    "sync"
)

// Result carries the outcome of processing one item.
type Result struct {
    Index int
    Value string
    Err   error
}

// Process runs fn on each item in parallel, using up to runtime.NumCPU()
// workers. It respects ctx: if the context is cancelled, workers stop early
// and Process returns ctx.Err().
func Process(ctx context.Context, items []string, fn func(string) (string, error)) ([]Result, error) {
    numWorkers := runtime.NumCPU()
    if len(items) < numWorkers {
        numWorkers = len(items)
    }
    if numWorkers == 0 {
        return nil, nil
    }

    // Fan out all job indices into a closed buffered channel so workers can
    // drain it without a coordinator goroutine.
    jobs := make(chan int, len(items))
    for i := range items {
        jobs <- i
    }
    close(jobs)

    // BUG (intentional): buffer is too small — fix in Step 3.
    out := make(chan Result, len(items)) // <-- will deadlock on cancel

    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for {
                if err := ctx.Err(); err != nil {
                    out <- Result{Err: err} // cancel send
                    return
                }
                select {
                case idx, ok := <-jobs:
                    if !ok {
                        return
                    }
                    v, err := fn(items[idx])
                    out <- Result{Index: idx, Value: v, Err: err} // job send
                case <-ctx.Done():
                    out <- Result{Err: ctx.Err()} // cancel send
                    return
                }
            }
        }()
    }

    wg.Wait()
    close(out)

    // Collect.
    var results []Result
    var firstErr error
    for r := range out {
        if r.Err != nil {
            if firstErr == nil {
                firstErr = r.Err
            }
            continue
        }
        results = append(results, r)
    }
    if firstErr != nil {
        return nil, firstErr
    }
    return results, nil
}

// identity is a trivial fn for tests.
func identity(s string) (string, error) {
    return fmt.Sprintf("done:%s", s), nil
}

In plain terms: this file defines a "struct" called Result — a bundle of labeled values (here, an index number, a text value, and an error) grouped under one name, the way a form has labeled boxes (each labeled box is called a "field"). It also defines a function named Process — a named, reusable block of instructions that takes inputs (here ctx, items, and fn) and returns (hands back to whoever ran it) a list of Results and an error. items []string means "a slice of strings" — a slice is a resizable list of values sitting in memory; []string is the type "list of text values." fn func(string) (string, error) means fn is itself a function that the caller (the code that runs Process) hands in as an input, to be run once per item.

Inside Process: make(chan int, len(items)) creates a channel — a pipe that goroutines use to send values to each other safely — sized to hold len(items) values without anyone waiting. A goroutine is a lightweight, independently-running stream of instructions; Go can run many of them at once, and this pattern of running many goroutines side by side to get work done faster is called concurrency. The line jobs <- i sends the number i into that channel; close(jobs) marks the channel as "no more values coming," which lets a receiver detect when it's empty. go func() { ... }() is how you start a new goroutine — the code inside the curly braces begins running independently, at the same time as everything after it.

var wg sync.WaitGroup sets up a "departure board" object (explained above in plain English) that the coordinator uses to know when every goroutine has finished. wg.Add(1) marks one more goroutine to wait for; defer wg.Done() schedules "mark this one goroutine as finished" to run automatically right before that goroutine exits, however it exits. wg.Wait() is where the coordinator blocks — meaning that line simply waits, doing nothing else, until the counter of unfinished goroutines drops back to zero.

The select { case ...: ... } block lets a goroutine wait on two channels at once — it proceeds with whichever one produces a value first: either a new job arriving on jobs, or the ctx.Done() signal firing because the context was cancelled. ctx.Err() asks the context "has cancellation already happened?" and hands back an error if so. out <- Result{...} sends a finished Result value into the out channel so the coordinator can collect it later; for r := range out loops over every value that arrives on out until it is closed. This whole block is the "BUG (intentional)" this lab exists to find and fix: out is sized only for len(items) sends, but as the comments show, a cancelling worker can also send — so in the worst case there are more sends than the channel has room for, and a worker gets stuck forever trying to send, which means it never reaches wg.Done().

Try it (happy path, should pass)

cd /tmp/pool-lab
go test -v -run TestHappyPath ./...

You need a test file first — create it in Step 2. (A test is a small piece of code written specifically to check that another piece of code behaves correctly; go test finds and runs all the tests in a package and reports which passed or failed.)

---

Step 2 — write the tests (happy path + cancel path)

Create pool_test.go:

package poollab

import (
    "context"
    "errors"
    "testing"
    "time"
)

func TestHappyPath(t *testing.T) {
    items := []string{"a", "b", "c", "d", "e"}
    results, err := Process(context.Background(), items, identity)
    if err != nil {
        t.Fatalf("unexpected error: %v", err)
    }
    if len(results) != len(items) {
        t.Fatalf("got %d results, want %d", len(results), len(items))
    }
}

// TestCancelDeadlock deliberately cancels the context mid-pool.
// With the buggy buffer size it hangs forever; the test has a 2-second timeout
// to surface the deadlock as a failure instead of a mystery hang.
func TestCancelDeadlock(t *testing.T) {
    t.Parallel()
    items := make([]string, 100)
    for i := range items {
        items[i] = "x"
    }

    ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
    defer cancel()

    done := make(chan struct{})
    go func() {
        _, _ = Process(ctx, items, func(s string) (string, error) {
            time.Sleep(10 * time.Millisecond) // slow enough for the timeout to fire
            return identity(s)
        })
        close(done)
    }()

    select {
    case <-done:
        // good — returned promptly after cancellation
    case <-time.After(2 * time.Second):
        t.Fatal("Process did not return after context cancellation — likely deadlock (buffer too small)")
    }
}

func TestErrPropagation(t *testing.T) {
    boom := errors.New("boom")
    _, err := Process(context.Background(), []string{"a", "b"}, func(s string) (string, error) {
        return "", boom
    })
    if !errors.Is(err, boom) {
        t.Fatalf("expected boom, got %v", err)
    }
}

In plain terms: each func TestXxx(t *testing.T) { ... } is one independent test; t.Fatalf records a failure with a message and stops that test immediately, while t.Fatal does the same without the message formatting. t.Parallel() tells Go's testing tool it's safe to run this test at the same time as other tests, rather than one after another. TestCancelDeadlock builds a context that automatically cancels itself after a timeout (a fixed amount of time to wait before giving up) using context.WithTimeout, starts Process running in its own goroutine, and then races two things in a select: either that goroutine finishes and signals done, or 2 real seconds pass. If the 2-second branch wins, the test declares a likely deadlock — proof that Process never returned.

Run the happy path:

go test -v -run TestHappyPath ./...

Expected output:

--- PASS: TestHappyPath (0.00s)
PASS

Now run the cancel test — it should fail (hang and time out) because the buffer is too small:

go test -v -run TestCancelDeadlock -timeout 5s ./...

Expected output (the test catches the bug):

--- FAIL: TestCancelDeadlock (2.00s)
    pool_test.go:45: Process did not return after context cancellation — likely deadlock (buffer too small)
FAIL

Predict-then-run

Before running the cancel test, answer for yourself: how many sends can land in out in the worst case? Count: up to len(items) job sends, plus up to numWorkers cancel sends. The current buffer holds only len(items). When cancellation fires before all jobs are drained, one or more workers blocks on out <- Result{Err: ctx.Err()} and never reaches wg.Done(). The coordinator hangs at wg.Wait() forever.

---

Step 3 — apply the fix

The fix mirrors the formula used in both jsmn-go/parallel.go and stb-image-go/stb_image.go. In pool.go, change one line:

// Before (buggy):
out := make(chan Result, len(items))

// After (correct):
out := make(chan Result, len(items)+numWorkers)

The reasoning: - Every item produces at most one send (out <- Result{...} inside the select case). - Every worker produces at most one cancel send (out <- Result{Err: ctx.Err()} in the cancel branch). - Total worst-case sends: len(items) + numWorkers. - A buffer of that size guarantees every send completes immediately, so every worker reaches wg.Done(), and wg.Wait() unblocks.

Run the full suite again:

go test -v -timeout 5s ./...

Expected output:

--- PASS: TestHappyPath (0.00s)
--- PASS: TestCancelDeadlock (0.06s)
--- PASS: TestErrPropagation (0.00s)
PASS

---

Step 4 — prove there is no data race

A data race is a bug where two goroutines read and write the same piece of memory at the same time with no coordination, so the result depends on unpredictable timing — one of the hardest categories of bug to spot by eye. The pool writes results from multiple goroutines — the only shared mutable state (memory that more than one goroutine can change) is the out channel, which Go's channel implementation protects internally. But let the race detector (a tool built into Go that watches a running program and flags exactly this kind of unsafe simultaneous access) confirm it:

Try it

go test -race -count=3 ./...

Expected output: PASS with no DATA RACE report.

If you accidentally wrote to a shared slice without a lock you would see something like:

WARNING: DATA RACE
Write at 0x... by goroutine N:
...

The approach used in both safeheaders-go pools — write results into an index-keyed slice allocated up front rather than appending to a shared slice — eliminates the need for a mutex over the output. Each worker writes only to results[idx], its own slot, which no other worker touches. In this lab the out channel serves the same isolation purpose.

---

Step 5 — optional: a pre-allocated output slice variant

The jsmn-go parallel path allocates jobResults := make([]chunkResult, numJobs) once and lets each worker write to its own slot. That removes the channel entirely for the output, at the cost of one allocation up front. The stb-image-go pool does the same thing (line 105 in stb_image.go):

results := make([]image.Image, len(datas))
// ...
results[idx] = img   // worker writes to its own slot — no lock needed

You can refactor your pool to this pattern if you want zero channel sends for the happy path. The errs channel then only carries errors and cancel signals, so its buffer is numWorkers (one per worker, worst case), not len(items) + numWorkers. The trade-off: you need to allocate the output slice up front and accept that failed slots remain their zero value.

Which pattern to use?

Use the pre-allocated slot pattern when items are independent and you always want all results regardless of partial errors. Use the channel collection pattern (this lab) when you want to stop at the first error or stream results as they arrive.

---

How the real pools compare

Detail	jsmn-go/parallel.go	stb-image-go/stb_image.go	This lab
Jobs channel	pre-filled, closed	pre-filled, closed	pre-filled, closed
Results channel buffer	numJobs + numWorkers	len(datas) + numWorkers	len(items) + numWorkers
Output storage	pre-allocated []chunkResult	pre-allocated []image.Image	collected from channel
Cancel check	bare select (ctx.Done() case)	ctx.Err() before select	ctx.Err() before select
Race-safe?	yes (-race in CI)	yes (-race in CI)	verified in Step 4

The channel buffer formula is identical in both production modules and this lab. Once you see the shape — (one send per job) + (one send per worker on cancel) — it becomes mechanical to apply.

---

Key takeaways

• Size your results channel for the worst case: numJobs + numWorkers, where the extra numWorkers slots absorb the cancel-path sends that each worker may emit before exiting.
• An under-sized channel deadlocks silently at wg.Wait() — no panic, no error, just a hang. A timeout-gated test is the only reliable detector.
• Check ctx.Err() before the select so an already-cancelled context is honored immediately; a bare select between a ready job and ctx.Done() is non-deterministic.
• -race is not optional for concurrent code — the Go race detector catches categories of bugs that are invisible to functional tests and code review.
• Pre-allocate the output slice when possible — writing to results[idx] from each worker needs no lock because each worker owns its own slot.