BrevFeed
Sign in

47. Worker pools and pipelines

📖 Reading · 13 min
💡 Most code boxes below are live — edit one and hit Run. Boxes without a Run button are reference-only (they can't run in your browser).

Goroutines, channels and select are the primitives. This lesson is the handful of shapes you'll actually build with them — the ones that show up in every real Go codebase.

The worker pool

Fixed number of workers, one jobs channel, one results channel. This is the single most useful concurrency pattern in Go:

package main

import (
    "fmt"
    "sort"
    "sync"
)

func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {
        results <- j * j
    }
}

func main() {
    const numJobs = 9
    const numWorkers = 3

    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)

    var wg sync.WaitGroup
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs)

    wg.Wait()
    close(results)

    var got []int
    for r := range results {
        got = append(got, r)
    }
    sort.Ints(got)
    fmt.Println(got)
}

Read the choreography, because the order matters:

  1. Start the workers. Each one ranges over jobs — the range ends when the channel is closed, which is how a worker knows to exit.
  2. Send all the jobs, then close jobs. Closing is the shutdown signal.
  3. wg.Wait() for every worker to finish, then close results. Closing results before the workers are done would panic on their next send.
  4. Drain results.

Why a pool rather than one goroutine per job? Because goroutines are cheap but the resources they use aren't — database connections, file handles, memory, an API's rate limit. A pool of N gives you a hard ceiling on concurrency, which is usually what production wants.

Sizing: for CPU-bound work, runtime.NumCPU(). For I/O-bound work (HTTP calls, database queries) you can go much higher, since workers spend most of their time blocked. Measure rather than guess.

Results that can fail

Real jobs fail. Send a struct so the error travels with the result:

package main

import (
    "fmt"
    "sort"
    "sync"
)

type Job struct {
    ID    int
    Input string
}

type Result struct {
    JobID  int
    Length int
    Err    error
}

func worker(jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {
        if j.Input == "" {
            results <- Result{JobID: j.ID, Err: fmt.Errorf("job %d: empty input", j.ID)}
            continue
        }
        results <- Result{JobID: j.ID, Length: len(j.Input)}
    }
}

func main() {
    inputs := []string{"go", "", "concurrency", "channels", ""}

    jobs := make(chan Job, len(inputs))
    results := make(chan Result, len(inputs))

    var wg sync.WaitGroup
    for w := 0; w < 3; w++ {
        wg.Add(1)
        go worker(jobs, results, &wg)
    }

    for i, in := range inputs {
        jobs <- Job{ID: i, Input: in}
    }
    close(jobs)

    wg.Wait()
    close(results)

    var all []Result
    for r := range results {
        all = append(all, r)
    }
    sort.Slice(all, func(i, j int) bool { return all[i].JobID < all[j].JobID })

    failures := 0
    for _, r := range all {
        if r.Err != nil {
            failures++
            fmt.Println("error:", r.Err)
            continue
        }
        fmt.Printf("job %d -> %d chars\n", r.JobID, r.Length)
    }
    fmt.Println("failures:", failures)
}

Note that a failing job doesn't stop the pool — the worker sends a result carrying the error and moves on. Deciding what to do with failures belongs to the coordinator, not the worker.

Pipelines

Each stage is a function that takes an input channel and returns an output channel, running its own goroutine. Stages compose by nesting:

package main

import "fmt"

func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

func filterOdd(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if n%2 != 0 {
                out <- n
            }
        }
    }()
    return out
}

func main() {
    for v := range filterOdd(square(generate(1, 2, 3, 4, 5, 6))) {
        fmt.Println(v)
    }
}

filterOdd(square(generate(...))) reads inside-out like a Unix pipe, and every stage runs concurrently: generate is already producing the fourth number while square works on the second.

The stage template, which never changes:

func stage(in <-chan T) <-chan U {
    out := make(chan U)
    go func() {
        defer close(out)   // always close what you own
        for v := range in {
            out <- transform(v)
        }
    }()
    return out             // return immediately, work happens in background
}

The discipline that makes it safe: each stage closes only the channel it created, and does it with defer so a panic can't leave the next stage hanging. Output is unbuffered, so a slow consumer naturally slows the whole pipeline — that's backpressure, and you get it for free.

Fan-out, fan-in

When one stage is the bottleneck, run several copies of it and merge the results:

package main

import (
    "fmt"
    "sort"
    "sync"
)

func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

func merge(channels []<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for i := 0; i < len(channels); i++ {
        wg.Add(1)
        go func(ch <-chan int) {
            defer wg.Done()
            for v := range ch {
                out <- v
            }
        }(channels[i])
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    src := generate(1, 2, 3, 4, 5, 6, 7, 8)

    // fan-out: three squarers reading the same source
    workers := []<-chan int{square(src), square(src), square(src)}

    var got []int
    for v := range merge(workers) {
        got = append(got, v)
    }

    sort.Ints(got)
    fmt.Println(got)
}

Fan-out is three goroutines ranging over the same input channel — Go distributes the values automatically, since each is delivered exactly once.

Fan-in is merge: one goroutine per input copying into a shared output, plus a small closer goroutine that waits for all of them and then closes out. That closer is the crux — close too early and a copier panics; never close and the consumer hangs. wg.Wait() in its own goroutine is the standard solution.

Results arrive interleaved, so sort at the end if you need order.

(Two notes on that code. Real Go usually writes merge as variadic — merge(channels ...<-chan int) — and loops with for _, c := range channels; the in-browser interpreter mishandles ranging over a slice of channels, so these boxes use a slice parameter and an index loop. The behaviour is identical, and outside the browser either spelling works.)

Avoiding goroutine leaks

Every pattern above has an exit path. Here's what happens when one doesn't:

func leaky() <-chan int {
    out := make(chan int)
    go func() {
        for i := 0; ; i++ {
            out <- i        // blocks forever once the consumer stops reading
        }
    }()
    return out
}

func main() {
    ch := leaky()
    fmt.Println(<-ch)   // take one value and walk away
    // the goroutine is now blocked on `out <- 1` for the life of the program
}

The goroutine is parked on a send nobody will ever receive. It holds its stack, its captured variables, and anything they reference — forever. The runtime won't warn you: it only detects deadlock when every goroutine is blocked.

Three defences:

  1. Always close what you produce, with defer close(out).
  2. Give every long-running goroutine a way to be told to stop — a quit channel, or a context (next lesson).
  3. Buffer by one when you abandon a goroutine deliberately, as the timeout example did, so its final send can complete.

Your turn

Build a three-worker pool that sums the lengths of the given strings. Print the total:

21

package main

import (
    "fmt"
    "sync"
)

func main() {
    words := []string{"go", "channels", "workers", "pool"}

    jobs := make(chan string, len(words))
    results := make(chan int, len(words))

    var wg sync.WaitGroup
    // start 3 workers that range over jobs and send len(word) to results

    for _, w := range words {
        jobs <- w
    }
    close(jobs)

    wg.Wait()
    close(results)

    total := 0
    for r := range results {
        total += r
    }
    fmt.Println(total)
}

package main

import (
    "fmt"
    "sync"
)

func main() {
    words := []string{"go", "channels", "workers", "pool"}

    jobs := make(chan string, len(words))
    results := make(chan int, len(words))

    var wg sync.WaitGroup
    for w := 0; w < 3; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for word := range jobs {
                results <- len(word)
            }
        }()
    }

    for _, w := range words {
        jobs <- w
    }
    close(jobs)

    wg.Wait()
    close(results)

    total := 0
    for r := range results {
        total += r
    }
    fmt.Println(total)
}

Next: cancelling all of it, from the top.

← Previous Next →