Bound goroutines with a worker pool

Run a fixed number of goroutines over a channel of jobs, so work goes in parallel with a ceiling on how much runs at once.

Every worker ranges over the same channel. Go delivers each job to exactly one of them, so no dispatch logic is needed.

Closing jobs ends every range loop, and wg.Wait blocks until the last worker returns.

jobs := make(chan string)
var wg sync.WaitGroup

for i := 0; i < 8; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for job := range jobs {
            process(job)
        }
    }()
}

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

A goroutine is cheap, and a hundred thousand of them still exhaust file descriptors, connection limits, and memory. Pools cap the resource the work consumes, not the goroutines themselves.

Size the pool to the bottleneck. Network calls tolerate hundreds, CPU-bound work wants roughly runtime.NumCPU().

// Unbounded: one goroutine per item.
for _, job := range allJobs {
    go process(job) // 100k jobs, 100k goroutines
}

Collecting results

Closing a results channel from a worker panics the others still sending on it. A separate goroutine that waits and then closes is the safe place for it.

Buffering the channel to the job count keeps a worker from blocking on a send while the collector is busy.

results := make(chan Result, len(allJobs))

for i := 0; i < workers; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for job := range jobs {
            results <- process(job)
        }
    }()
}

go func() {
    wg.Wait()
    close(results) // after every worker is done, never before
}()

for r := range results {
    // ...
}

errgroup from golang.org/x/sync does the same job in less code. SetLimit bounds the concurrency, Wait returns the first error, and the shared context cancels the rest.

It is an external module. The channel version above stays inside the standard library.

g, ctx := errgroup.WithContext(ctx)
g.SetLimit(8)

for _, job := range allJobs {
    g.Go(func() error {
        return process(ctx, job)
    })
}

if err := g.Wait(); err != nil {
    // the first failure, with every other job cancelled
}

Working program

A complete program that runs eight 100ms jobs through four workers. Serially it would take 800ms.

package main

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

type result struct {
    url  string
    size int
}

// check stands in for a slow network call.
func check(url string) result {
    time.Sleep(100 * time.Millisecond)
    return result{url: url, size: len(url) * 10}
}

func main() {
    urls := []string{
        "/a", "/bb", "/ccc", "/dddd",
        "/eeeee", "/ffffff", "/ggggggg", "/hhhhhhhh",
    }

    const workers = 4
    jobs := make(chan string)
    results := make(chan result, len(urls))

    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for url := range jobs {
                results <- check(url)
            }
        }()
    }

    start := time.Now()
    for _, u := range urls {
        jobs <- u
    }
    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].url < all[j].url })

    for _, r := range all {
        fmt.Printf("%-10s %d\n", r.url, r.size)
    }
    fmt.Printf("%d urls, %d workers, %v\n", len(urls), workers, time.Since(start).Round(10*time.Millisecond))
}

Run it.

Terminal

$ go run main.go
/a         20
/bb        30
/ccc       40
/dddd      50
/eeeee     60
/ffffff    70
/ggggggg   80
/hhhhhhhh  90
8 urls, 4 workers, 200ms

Worker Pools The pattern itself, worked through step by step.Channels Sending, ranging, and what closing does to a receiver.sync WaitGroup, Mutex, and Once.context Cancelling in-flight workers when one fails.

Previous Decode a JSON request body Next Topic Format a date and time

Bound goroutines with a worker pool

Collecting results

Working program

Related