Skip to content

A worker pool

Level: 201 · you have a list of jobs and want a fixed number of goroutines working through it

One line: Three goroutines ranging over one jobs channel are a pool of three. A job that carries its index lets main put each result back in its place, so the report comes out in input order however the jobs were scheduled. The pool shuts down in one direction only: close jobs, the workers return, then close results.

The output

Verified output of a_worker_pool_go.go — regenerated by tools/run_examples.py, never hand-typed.

workers:                    3
jobs sent:                  6
results received:           6
results closed:             after all 3 workers returned
most jobs in progress <= 3: true
report, in input order:
  1. 7 words  a goroutine is a function running concurrently
  2. 6 words  a channel carries values between goroutines
  3. 9 words  close the jobs channel once every job is sent
  4. 8 words  each worker returns when its range loop ends
  5. 6 words  results are stored by job index
  6. 9 words  so the report keeps the order of the input
a_worker_pool_go.go

a_worker_pool_go.go in full — pasted here by tools/run_examples.py from the file CI runs.

// A worker pool: three workers take jobs from one channel and send results on
// another. Each job carries its index, and main stores each result at that
// index, so the report comes out in input order however the jobs were
// scheduled. The pool shuts down in three steps: the feeder closes jobs, each
// worker's range loop ends and it returns, and once all three have returned a
// WaitGroup lets the closer close results, which ends main's range loop.
//
//  go build a_worker_pool_go.go && ./a_worker_pool_go
package main

import (
    "fmt"
    "strings"
    "sync"
    "sync/atomic"
)

const workers = 3

type job struct {
    index int
    line  string
}

type result struct {
    index int
    words int
}

func main() {
    lines := []string{
        "a goroutine is a function running concurrently",
        "a channel carries values between goroutines",
        "close the jobs channel once every job is sent",
        "each worker returns when its range loop ends",
        "results are stored by job index",
        "so the report keeps the order of the input",
    }

    jobs := make(chan job)
    results := make(chan result)

    // The feeder sends from its own goroutine, so that main is free to receive
    // results while jobs are still going out.
    go func() {
        defer close(jobs) // step 1: no more jobs
        for i, line := range lines {
            jobs <- job{index: i, line: line}
        }
    }()

    var pool sync.WaitGroup
    var busy, mostBusy atomic.Int32
    for range workers {
        pool.Go(func() {
            for j := range jobs { // step 2: ends when jobs is closed and drained
                n := busy.Add(1)
                for m := mostBusy.Load(); n > m && !mostBusy.CompareAndSwap(m, n); m = mostBusy.Load() {
                }
                words := len(strings.Fields(j.line))
                busy.Add(-1)
                results <- result{index: j.index, words: words}
            }
        })
    }

    go func() {
        pool.Wait()    // every worker has returned, so none will send again
        close(results) // step 3
    }()

    report := make([]int, len(lines))
    received := 0
    for r := range results { // arrival order is the scheduler's
        report[r.index] = r.words
        received++
    }

    fmt.Printf("workers:                    %d\n", workers)
    fmt.Printf("jobs sent:                  %d\n", len(lines))
    fmt.Printf("results received:           %d\n", received)
    fmt.Printf("results closed:             after all %d workers returned\n", workers)
    fmt.Printf("most jobs in progress <= %d: %t\n", workers, mostBusy.Load() <= workers)
    fmt.Println("report, in input order:")
    for i, line := range lines {
        fmt.Printf("  %d. %d words  %s\n", i+1, report[i], line)
    }
}

Reading the output

The report is in input order. The results were not. Results reach main in whatever order the workers finish, and the program does nothing to change that. Instead, the feeder sends each line together with its index, the worker sends the index back with its count, and main stores the count at report[r.index]. Only main writes to report, so there is no lock and no sort, and the printed order is the order of lines. Fan-out, fan-in sorts instead, which works when the results have a natural key. An index works for any job.

Shutting down takes three steps, and each step has one owner.

Step Who does it What it ends
1. close(jobs) the feeder, the only sender on jobs, after its last send nothing yet: the workers first receive what was already sent
2. each worker returns the worker, when its for j := range jobs loop ends its place in the pool WaitGroup, which pool.Go gave it
3. close(results) the closer goroutine, once pool.Wait() returns main's for r := range results loop

The workers are the senders on results, but there are three of them, so none of them can close it: after the first one closed it, the next one's send would panic (spec: Close ↗). The goroutine that waits for all three closes it instead. The Go blog's Pipelines and cancellation ↗ does the same in its section "Bounded parallelism", where a fixed number of digesters send on one shared channel that is closed after a WaitGroup sees all of them finish. The steps cannot happen in another order, and the output is the proof that step 3 happened: main prints nothing until its loop over results has ended.

most jobs in progress <= 3: true. Each worker counts itself busy with an atomic counter and records the highest value it sees (Atomic counters). Three goroutines cannot run more than three jobs at once. How many actually overlapped is up to the scheduler, so the key records only the bound. A buffered channel as a semaphore forces the maximum to be exactly its limit.

Why the feeder is a goroutine of its own

main could send the jobs itself, but not all of them before it starts reading results. This driver does exactly that:

Verified output of a_worker_pool_deadlock_sh.sh — regenerated by tools/run_examples.py, never hand-typed.

$ go build deadlock.go
$ ./deadlock
main: sent job 1
main: sent job 2
main: sent job 3
fatal error: all goroutines are asleep - deadlock!
exit status 2

Three jobs go out and the fourth never does, on every run, and the scheduler has nothing to do with it. jobs and results are unbuffered, so each worker takes one job and then blocks sending its result to main, which is not receiving. main blocks sending job 4, because no worker is receiving (An unbuffered send waits for a receiver). Giving results a buffer of len(lines) would also fix it, when the number of jobs is known before any are sent (A buffered channel is a bounded queue).

a_worker_pool_deadlock_sh.sh

a_worker_pool_deadlock_sh.sh in full — pasted here by tools/run_examples.py from the file CI runs.

#!/usr/bin/env bash
# The worker pool with one change: main sends every job itself before it reads
# any result. Each of the three workers takes one job, then blocks sending its
# result, because main is not receiving; main blocks sending the fourth job,
# because no worker is receiving. Once every goroutine is blocked, Go's runtime
# ends the program. The goroutine dump after its first line names goroutine
# numbers, so this script keeps the first line and the exit status.
set -u

dir=$(mktemp -d)
trap 'rm -rf "$dir"' EXIT
export GOTOOLCHAIN=local

cat >"$dir/deadlock.go" <<'GO'
package main

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

func main() {
    lines := []string{
        "a goroutine is a function running concurrently",
        "a channel carries values between goroutines",
        "close the jobs channel once every job is sent",
        "each worker returns when its range loop ends",
        "results are stored by job index",
        "so the report keeps the order of the input",
    }
    jobs := make(chan string)
    results := make(chan int)

    var pool sync.WaitGroup
    for range 3 {
        pool.Go(func() {
            for line := range jobs {
                results <- len(strings.Fields(line))
            }
        })
    }

    for i, line := range lines { // every job first...
        jobs <- line
        fmt.Printf("main: sent job %d\n", i+1)
    }
    close(jobs)
    for words := range results { // ...then the results: never reached
        fmt.Println(words)
    }
}
GO

say() { printf '$ %s\n' "$*"; }

say 'go build deadlock.go'
(cd "$dir" && go build -trimpath deadlock.go) || exit 1

say './deadlock'
(cd "$dir" && ./deadlock) 2>"$dir/stderr"
status=$?
grep '^fatal error' "$dir/stderr"
echo "exit status $status"

What to do

  • Send the jobs from a goroutine of their own, and have it defer close(jobs).
  • Send an index with each job and back with each result, store the results by index, and print once the results channel has closed.
  • Close results from a goroutine that waits on the workers' WaitGroup, never from a worker.
  • Size the pool for the resource the jobs share: CPUs for computation, connections for a database, the other side's limit for calls to a service.

In other languages

  • The Concurrency library's Getting a result back ↗: Java's ExecutorService and Python's ThreadPoolExecutor are those languages' worker pools. submit returns a future for each task, so a result stays attached to its task instead of to an index, and Java's ExecutorService.invokeAll ↗ returns its futures in the order of the tasks. In Go the pool is goroutines and channels you write yourself; the Go section of that page uses a result channel, as this one does.
  • The Rust library's Channels ↗ lists a work queue among the jobs of a channel, but its Receiver is not Clone, so Rust workers cannot each hold the queue the way three Go goroutines each range over jobs. They share the one receiver, typically behind a mutex. (Not machine-checked here.)
  • The Concurrency library's chapters that will compare this across languages, 05 Message passing and 07 Parallelism, are planned.

Sources

  • The Go blog, Pipelines and cancellation ↗, section "Bounded parallelism".
  • The Go specification: Close ↗, Channel types ↗.
  • sync.WaitGroup ↗ and sync/atomic ↗.
  • Burak Serdar, Effective Concurrency in Go (Packt, 2023), chapter 5, "Worker Pools and Pipelines".
  • James Cutajar, Learn Concurrent Programming with Go (Manning, 2024), chapter 10, "Concurrency patterns": "Using worker pools".
  • Jon Bodner, Learning Go, 2nd ed. (O'Reilly, 2024), chapter 12, "Concurrency in Go": "Use WaitGroups", which recommends a WaitGroup for exactly this kind of cleanup, closing a channel that several workers write to.