Capstone 2: a concurrent job runner · Go: Zero to Production

The second project is the one Go is chosen for: run many independent jobs at once, bound the concurrency, collect results in order, handle per-job failures, and support cancellation. Every piece comes from module 9, but assembling them correctly is its own skill.

The specification

Given a list of jobs, run them across a fixed pool of workers and produce:

results in the original input order, regardless of completion order
per-job errors, without one failure stopping the rest
a hard limit on how many run concurrently
a context that cancels everything

Step 1: the job and result types

package main

import (
    "fmt"
    "strings"
)

type Job struct {
    ID    int
    Input string
}

type Result struct {
    JobID  int
    Output string
    Err    error
}

func process(j Job) Result {
    if strings.TrimSpace(j.Input) == "" {
        return Result{JobID: j.ID, Err: fmt.Errorf("job %d: empty input", j.ID)}
    }
    return Result{JobID: j.ID, Output: strings.ToUpper(j.Input)}
}

func main() {
    jobs := []Job{
        {ID: 0, Input: "hello"},
        {ID: 1, Input: ""},
        {ID: 2, Input: "world"},
    }

    for _, j := range jobs {
        r := process(j)
        if r.Err != nil {
            fmt.Println("error:", r.Err)
            continue
        }
        fmt.Printf("%d -> %s\n", r.JobID, r.Output)
    }
}

Start sequential. It's easier to get right, it's easier to test, and it's the version you compare against when the concurrent one gives you a different answer. process is a pure function — job in, result out, no channels — so it stays testable no matter what runs it.

Carrying Err inside Result is the key design decision: a goroutine can't return anything, so the error rides along with the value.

Step 2: a worker pool with ordered results

package main

import (
    "fmt"
    "strings"
    "sync"
)

type Job struct {
    ID    int
    Input string
}

type Result struct {
    JobID  int
    Output string
    Err    error
}

func process(j Job) Result {
    if strings.TrimSpace(j.Input) == "" {
        return Result{JobID: j.ID, Err: fmt.Errorf("job %d: empty input", j.ID)}
    }
    return Result{JobID: j.ID, Output: strings.ToUpper(j.Input)}
}

func RunAll(jobs []Job, workers int) []Result {
    if workers < 1 {
        workers = 1
    }

    in := make(chan Job)
    results := make([]Result, len(jobs))

    var wg sync.WaitGroup
    for w := 0; w < workers; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := range in {
                // Safe because ID is the input position — see the note below.
                results[j.ID] = process(j)
            }
        }()
    }

    for _, j := range jobs {
        in <- j
    }
    close(in)
    wg.Wait()

    return results
}

func main() {
    inputs := []string{"alpha", "beta", "", "delta", "epsilon", ""}
    jobs := make([]Job, len(inputs))
    for i, s := range inputs {
        jobs[i] = Job{ID: i, Input: s}
    }

    failures := 0
    for _, r := range RunAll(jobs, 3) {
        if r.Err != nil {
            failures++
            fmt.Printf("%d: ERROR %v\n", r.JobID, r.Err)
            continue
        }
        fmt.Printf("%d: %s\n", r.JobID, r.Output)
    }
    fmt.Println("failures:", failures)
}

The ordering trick: pre-size a results slice and let each worker write to its own index. results[j.ID] = ... needs no mutex, because no two jobs share an index, and the output comes back in input order for free — no sorting, no results channel to drain.

That line hides an assumption worth naming. It works only because main numbers the jobs 0, 1, 2 …, so the ID is the position in results. That's a real invariant, not a coincidence — and it holds right up until someone feeds the same pool jobs carrying database IDs. One Job{ID: 4071} against a six-element results panics with an index out of range, inside a goroutine, which is the least pleasant place to debug one.

So either state that invariant where the jobs are built, or don't rely on it: carry the position alongside the job and let the ID go back to being a label.

type indexed struct {
    pos int
    job Job
}

// feeding the pool:
for i, j := range jobs {
    in <- indexed{pos: i, job: j}
}

// in the worker:
for it := range in {
    results[it.pos] = process(it.job)
}

Same guarantee — one writer per slot, no mutex, input order preserved — with no assumption about what an ID happens to mean.

The choreography is the one from module 9: start workers, feed jobs, close the input channel, wg.Wait(). Closing is how each worker's range ends.

Step 3: adding cancellation

Now make it stoppable. Workers check the context, and so does the feeder:

package main

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

type Job struct {
    ID    int
    Input string
}

type Result struct {
    JobID  int
    Output string
    Err    error
}

func process(ctx context.Context, j Job) Result {
    select {
    case <-ctx.Done():
        return Result{JobID: j.ID, Err: ctx.Err()}
    case <-time.After(10 * time.Millisecond):
    }
    if strings.TrimSpace(j.Input) == "" {
        return Result{JobID: j.ID, Err: fmt.Errorf("job %d: empty input", j.ID)}
    }
    return Result{JobID: j.ID, Output: strings.ToUpper(j.Input)}
}

func RunAll(ctx context.Context, jobs []Job, workers int) []Result {
    in := make(chan Job)
    results := make([]Result, len(jobs))

    var wg sync.WaitGroup
    for w := 0; w < workers; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := range in {
                results[j.ID] = process(ctx, j)
            }
        }()
    }

    go func() {
        defer close(in)
        for _, j := range jobs {
            select {
            case in <- j:
            case <-ctx.Done():
                return
            }
        }
    }()

    wg.Wait()
    return results
}

func main() {
    jobs := make([]Job, 8)
    for i := range jobs {
        jobs[i] = Job{ID: i, Input: fmt.Sprintf("task-%d", i)}
    }

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

    done, cancelled := 0, 0
    for _, r := range RunAll(ctx, jobs, 2) {
        if r.Err != nil {
            cancelled++
            continue
        }
        done++
    }

    fmt.Println("some finished before the deadline:", done > 0)
    fmt.Println("the rest were cancelled:", cancelled > 0)
    fmt.Println("all accounted for:", done+cancelled == len(jobs))
}

Two changes that matter:

The feeder moved into its own goroutine with a select on ctx.Done(). Without it, in <- j would block forever if every worker had already stopped — the deadlock from module 9.

defer close(in) in the feeder. Whoever sends is the one who closes, and deferring means it happens whether the loop finished or the context cancelled.

The output is deliberately reported as booleans rather than exact counts: concurrent timings vary between runs, and a test that asserts "exactly 4 finished" would be flaky. Assert on invariants — everything is accounted for, nothing is lost — not on the schedule.

Step 4: bounding concurrency differently

The pool bounds concurrency by having a fixed number of workers. The other way, from module 9, is a semaphore — useful when you want one goroutine per job but a limit on how many are inside the expensive part:

package main

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

func main() {
    const limit = 3
    sem := make(chan struct{}, limit)

    var wg sync.WaitGroup
    var mu sync.Mutex
    peak, current := 0, 0

    results := make([]int, 12)

    for i := 0; i < 12; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()

            sem <- struct{}{}
            defer func() { <-sem }()

            mu.Lock()
            current++
            if current > peak {
                peak = current
            }
            mu.Unlock()

            time.Sleep(3 * time.Millisecond)
            results[n] = n * n

            mu.Lock()
            current--
            mu.Unlock()
        }(i)
    }

    wg.Wait()

    fmt.Println(results)
    fmt.Println("peak concurrency:", peak, "- within limit:", peak <= limit)
}

Which to pick? A worker pool when jobs are uniform and you want a fixed number of long-lived goroutines. A semaphore when goroutines are already being spawned per item (per request, per file) and you only need to gate one resource.

What makes this correct

The checklist to run over any concurrent Go code you write:

Every goroutine has an exit path — a closed channel to range over, a context to check, or a bounded loop.
Whoever creates a channel closes it, and never sends after closing.
Shared state is either per-index or behind a mutex. results[j.ID] qualifies — as long as the index really is the position; results = append(results, r) from several goroutines does not.
defer wg.Done() on the first line of every worker.
Errors travel with results, they don't panic or get logged and lost.
Tests run with -race.

Your turn

Write squareAll, which squares each number using one goroutine per element and returns the results in input order:

[1 4 9 16 25]

package main

import (
    "fmt"
    "sync"
)

func squareAll(nums []int) []int {
    // square each number in its own goroutine, preserving order
}

func main() {
    fmt.Println(squareAll([]int{1, 2, 3, 4, 5}))
}

package main

import (
    "fmt"
    "sync"
)

func squareAll(nums []int) []int {
    out := make([]int, len(nums))

    var wg sync.WaitGroup
    for i, n := range nums {
        wg.Add(1)
        go func(idx, val int) {
            defer wg.Done()
            out[idx] = val * val
        }(i, n)
    }
    wg.Wait()

    return out
}

func main() {
    fmt.Println(squareAll([]int{1, 2, 3, 4, 5}))
}

One capstone left: a small service that ties the type system together.

64. Capstone 2: a concurrent job runner