Day 23 — Worker pools & pipelines

Updated

July 30, 2026

Day 23 — Worker pools & pipelines

Stage III · ~3h (theory-heavy)
Goal: Bound concurrency with worker pools, design multi-stage pipelines with backpressure, and build a race-clean concurrent “fetcher” that does not spawn unbounded goroutines.

Note

Unbounded go per item is the most common production footgun. Pools turn “as much as possible” into “as much as safe.”

Why this day exists

You can start 10k goroutines (Day 19). Doing that per HTTP request or per file in a million-file walk:

  • Exhausts memory and file descriptors
  • Overwhelms remote APIs
  • Makes cancellation (Day 24) painful

Worker pools and pipelines are the default industrial shape of concurrent Go.

Theory 1 — Bound concurrency, not just spawn

Three common bounders:

Technique Mechanism
Fixed workers N goroutines ranging over a jobs channel
Semaphore channel Buffered chan struct{} of size N
errgroup.SetLimit Day 27 — structured concurrency with limits

Fixed worker pool skeleton

jobs := make(chan Job)
results := make(chan Result)

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

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

// sender:
go func() {
    defer close(jobs)
    for _, j := range all {
        jobs <- j
    }
}()

for r := range results {
    handle(r)
}

Properties:

  • At most workers concurrent process calls
  • Backpressure: slow process fills jobs buffer and blocks senders
  • Clean shutdown when jobs closed and workers exit

Theory 2 — Semaphore style (ad-hoc pool)

sem := make(chan struct{}, maxConcurrent)

for _, item := range items {
    sem <- struct{}{} // acquire
    go func(item Item) {
        defer func() { <-sem }() // release
        work(item)
    }(item)
}
// still need WaitGroup to join!

This allows up to maxConcurrent goroutines alive for work, but still creates one goroutine per item over time (each exits after work). Fixed workers reuse goroutines—often clearer for long pipelines.

Theory 3 — Pipeline stages

A pipeline is a series of stages connected by channels:

source → stage1 → stage2 → sink

Each stage:

  1. Receives until input closed
  2. Sends outputs
  3. Closes its output when done (usually via defer after WaitGroup if fan-out inside stage)

Stage template

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

Compose:

out := square(gen(1, 2, 3, 4))
for v := range out {
    fmt.Println(v)
}

Theory 4 — Errors in pools (preview)

Naive pools only pass happy-path results. Real pools need:

  • Result{Value, Err} structs
  • Or a separate error channel (harder—ordering)
  • Or errgroup (Day 27) for first-error cancel

For today, use a result struct:

type Result struct {
    URL string
    Err error
    N   int
}

Collect all results; decide fail-fast vs best-effort at the aggregator.

Theory 5 — Choosing buffer sizes

Buffer Effect
0 Strong coupling; good for low latency handoff
small (workers) Smooths jitter
huge Hides backpressure; can OOM

Heuristic: buffer ≈ worker count or small multiple—not “millions.”

Theory 6 — Graceful drain vs cancel

Two shutdown modes:

  1. Drain: stop accepting new jobs; finish in-flight; close
  2. Cancel: abandon in-flight soon (needs context + cooperative work—Day 24)

Pools without cancel only support drain. That is OK for batch CLIs; services need cancel.

Worked example — bounded concurrent “fetch”

Simulated fetch with sleep and optional failure:

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

type Job struct{ URL string }
type Result struct {
    URL string
    OK  bool
    Ms  int
}

func fetch(url string) Result {
    ms := 20 + rand.Intn(80)
    time.Sleep(time.Duration(ms) * time.Millisecond)
    return Result{URL: url, OK: rand.Float32() > 0.1, Ms: ms}
}

func runPool(urls []string, workers int) []Result {
    jobs := make(chan Job)
    results := make(chan Result)

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

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

    go func() {
        defer close(jobs)
        for _, u := range urls {
            jobs <- Job{URL: u}
        }
    }()

    var out []Result
    for r := range results {
        out = append(out, r)
    }
    return out
}

func main() {
    urls := make([]string, 50)
    for i := range urls {
        urls[i] = fmt.Sprintf("https://example.com/%d", i)
    }
    start := time.Now()
    res := runPool(urls, 8)
    fmt.Printf("got %d results in %s\n", len(res), time.Since(start))
    fail := 0
    for _, r := range res {
        if !r.OK {
            fail++
        }
    }
    fmt.Println("failures", fail)
}

Lab 1 — Implement a generic-shaped pool (concrete type OK)

Workspace: ~/lab/90daysofx/01-go/day23

mkdir -p ~/lab/90daysofx/01-go/day23 && cd ~/lab/90daysofx/01-go/day23
go mod init example.com/day23

Requirements:

  • workers flag or const (try 1, 4, 16)
  • Process at least 100 jobs
  • Each job: hash a string or sleep + counter
  • Print throughput (jobs/sec)
  • go run -race .

Lab 2 — Three-stage pipeline

Stages:

  1. gen — emit lines / ints
  2. transform — uppercase or square
  3. filter — drop some values

Each stage is a function <-chan T → <-chan U. Main only ranges the final channel. Race-clean.

Lab 3 — Unbounded vs bounded comparison

Version A: for _, item := range items { go work(item) } + WaitGroup
Version B: worker pool with workers=8

For items=5000 with 5–20ms simulated I/O:

  • Compare max goroutines roughly via runtime.NumGoroutine() sampling
  • Compare elapsed time
  • Journal memory feel (Activity Monitor optional)
// sample
go func() {
    for range time.Tick(50 * time.Millisecond) {
        fmt.Println("g", runtime.NumGoroutine())
    }
}()

Stop the sampler cleanly (done channel)—don’t leak the ticker goroutine.

Lab 4 — Result aggregation with errors

Return []Result including errors. Exit non-zero if any job failed after draining the pool (do not abandon workers mid-flight yet—that is Day 24/27).

go test -race ./...

Common gotchas

Gotcha Fix
Close jobs before send finished Defer close in sender only
Workers blocked on full results + main not reading Buffer results or read concurrently
Starting workers after closing jobs Start workers first
One huge results slice race Single aggregator goroutine
Pool without join WaitGroup workers → close results
workers > len(jobs) worry Fine; idle workers exit on close

Checkpoint

  • Explain backpressure via full channels
  • Fixed-size pool processes N jobs with W workers
  • Multi-stage pipeline closes correctly end-to-end
  • Compared unbounded vs bounded goroutine counts
  • Aggregated errors without losing workers
  • go run -race / go test -race clean

Commit

git add .
git commit -m "day23: worker pool + pipeline + bounded fetcher"

Write three personal gotchas before continuing.

Tomorrow

Day 24 — context: cancellation, deadlines, values (sparingly), and cancelling the fetcher cleanly mid-flight.