Worker Pools & Pipelines

Reviewed & published by Brayan K

A worker pool is a fixed set of goroutines that pull jobs from a shared channel and push results to another, letting you process a large workload concurrently while capping how many run at once — and a pipeline chains stages together so data flows through them one step at a time.

Part of the free Go course at LearnCodingFast — hands-on lessons with worked examples and the output they print, plus practice exercises and a quick quiz.

What You'll Learn in This Lesson

• Build a jobs → workers → results pool
• Use a WaitGroup to know when work is finished
• Close jobs and results channels at the right moment
• Bound concurrency with a fixed worker count
• Cap parallelism with a semaphore channel
• Connect pipeline stages through channels

1️⃣ The classic worker pool

The shape is always the same: a jobs channel, N worker goroutines that range over it, and a results channel they write to. A WaitGroup tracks the workers. Note the channel directions in the worker signature: <-chan int (receive-only jobs) and chan<- int (send-only results) document intent and let the compiler catch misuse.

package main

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

// worker reads jobs off the jobs channel until it is closed, processes
// each, and sends the result to the results channel.
func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs { // ends when jobs is closed AND drained
        results <- j * j  // the "work": square the number
    }
}

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

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

    // Start a FIXED number of workers -> bounded concurrency. No matter how
    // many jobs arrive, only numWorkers run at once.
    var wg sync.WaitGroup
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

    // Send all jobs, then close so workers know no more are coming.
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs)

    // Close results once every worker has finished (in its own goroutine so
    // main can range over results below without deadlocking).
    go func() {
        wg.Wait()
        close(results)
    }()

    // Results arrive in any order; collect then sort for a stable print.
    var collected []int
    for r := range results {
        collected = append(collected, r)
    }
    sort.Ints(collected)
    fmt.Println("squares:", collected)
}

2️⃣ Bounding concurrency with a semaphore

Sometimes you don't want a pool of pre-spawned workers — you want to launch a goroutine per task but cap how many run at once (say, to respect an API rate limit). A buffered channel works as a counting semaphore: its capacity is the limit. Send an empty struct{}{} to acquire a slot before working, receive to release it. Once full, further acquires block.

package main

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

func main() {
    tasks := []int{1, 2, 3, 4, 5, 6, 7, 8}
    const maxParallel = 2 // never more than 2 running at once

    // A buffered channel used as a counting SEMAPHORE: acquire a slot
    // before starting, release it when done. Buffer size = the limit.
    sem := make(chan struct{}, maxParallel)

    var wg sync.WaitGroup
    var mu sync.Mutex
    var out []int

    for _, t := range tasks {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            sem <- struct{}{}        // acquire (blocks if 2 already running)
            defer func() { <-sem }() // release

            mu.Lock()
            out = append(out, n*10)
            mu.Unlock()
        }(t)
    }
    wg.Wait()

    sort.Ints(out)
    fmt.Println("results:", out)
}

3️⃣ The pipeline pattern

A pipeline composes stages. Each stage is a function that takes a receive-only input channel, launches a goroutine, and returns a receive-only output channel — so you wire them up by nesting calls: double(keepEven(gen(...))). Data flows one direction; closing each output channel propagates "no more values" downstream. No shared state, no mutex.

package main

import "fmt"

// Each stage takes a receive-only input channel and returns a receive-only
// output channel, running its work in its own goroutine.

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

func keepEven(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 double(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * 2
        }
    }()
    return out
}

func main() {
    // gen -> keepEven -> double. Data flows one way; each stage is decoupled.
    // The stages are single goroutines in a straight line (no fan-out), so
    // order is preserved and the output is deterministic.
    for v := range double(keepEven(gen(1, 2, 3, 4, 5, 6))) {
        fmt.Print(v, " ")
    }
    fmt.Println()
}

🎯 Your Turn

The pool is almost complete — but the workers will block forever waiting for more jobs. Add the one line that tells them no more are coming. Fill in the blank marked ___.

package main

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

func worker(jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {
        results <- j + 100 // the "work"
    }
}

func main() {
    // 🎯 YOUR TURN — fill in the blanks marked ___
    jobs := make(chan int, 5)
    results := make(chan int, 5)
    var wg sync.WaitGroup

    for w := 0; w < 2; w++ {
        wg.Add(1)
        go worker(jobs, results, &wg)
    }
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    ___ // 👉 close(jobs)   — tell workers no more jobs

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

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

    // ✅ Expected output:
    //    [101 102 103 104 105]
}

Common Errors (and the fix)

❌ Never closing jobs — every worker's range blocks forever; deadlock.

✅ close(jobs) after sending the last job.

❌ Closing results too early (before workers finish) — "panic: send on closed channel".

✅ Close it after wg.Wait(), in a goroutine so main keeps draining.

❌ Unbuffered results with no reader — workers block on send and never call Done().

✅ Drain results concurrently, or buffer it to numJobs.

❌ Expecting ordered output from concurrent workers.

✅ Collect into a slice and sort, or key results by job ID and reassemble.

🧠 Quick Recall

sem := make(chan struct{}, 3) // what does 3 mean?

At most 3 goroutines hold a slot at once. The fourth sem <- struct{}{} blocks until someone receives (releases) a slot.

who closes results? after what?

A goroutine running wg.Wait(); close(results) — i.e. once every worker has called Done(). Never close it from a worker.

🎯 Mini-Challenge: Sum-of-Digits Pool

Build a 4-worker pool where each worker sums the decimal digits of its number. Feed in 12, 305, 99, 7, 4444, then collect and sort the results.

package main

import "fmt"

// digitSum returns the sum of the decimal digits of n (n >= 0).
func digitSum(n int) int {
    sum := 0
    for n > 0 {
        sum += n % 10
        n /= 10
    }
    return sum
}

func main() {
    // 🎯 MINI-CHALLENGE: a sum-of-digits worker pool
    // Spin up 4 workers reading from a jobs channel. Each worker calls
    // digitSum on its number and sends the result to a results channel.
    // Feed in the numbers below, close jobs, close results after wg.Wait(),
    // then collect, sort, and print.
    //
    // ✅ Expected output:
    //    [3 7 8 16 18]
    //    (7->7, 12->3, 305->8, 99->18, 4444->16)

    nums := []int{12, 305, 99, 7, 4444}
    fmt.Println("inputs:", nums)
    // your code here: jobs/results channels, 4 workers, WaitGroup, sort
}

🎉 Lesson Complete!

• ✅ A worker pool = jobs channel + N workers + results channel + WaitGroup
• ✅ The producer closes jobs; close results after wg.Wait()
• ✅ A fixed worker count gives bounded concurrency
• ✅ A buffered chan struct{} works as a semaphore to cap parallelism
• ✅ Pipelines chain stages through channels; collect & sort for deterministic output

Practice quiz

Continue this course

• Previous: The context Package
• Next: HTTP Clients (net/http) — Build requests, set timeouts, read bodies, and decode JSON responses

Frequently asked questions

What is a worker pool and why bound the number of workers?

A worker pool is a fixed set of goroutines that all read from one shared jobs channel and write to one results channel. Bounding the count gives you bounded concurrency: however many jobs arrive, only N run at once. That protects shared resources — a database with a connection limit, an API with a rate limit, or simply your CPU and memory — from being overwhelmed by spawning a goroutine per job.

Who closes the jobs channel and who closes results?

The producer that sends the jobs closes the jobs channel once it has sent the last one — that lets each worker's for range over jobs end naturally. The results channel is closed only after every worker has finished, which you detect with wg.Wait(). The standard trick is go func(){ wg.Wait(); close(results) }() so main can range over results without deadlocking.

How do I get deterministic output from a worker pool?

You can't control the order workers finish, so don't try. Collect every result into a slice, then sort it (or key results by job ID and reassemble). The lesson examples all collect-then-sort precisely so the printed output is stable across runs while the work still happens concurrently.

What's the difference between a worker pool and a pipeline?

A worker pool spreads many independent jobs across N identical workers — good for embarrassingly parallel work. A pipeline is a chain of DIFFERENT stages (generate then filter then transform), each its own goroutine, connected by channels, where data flows one direction through them. Pools are about width (parallelism); pipelines are about depth (composable stages). Real systems often combine them.

Can I limit concurrency without a full worker pool?

Yes — use a buffered channel as a counting semaphore. Make it with the capacity equal to your limit, send an empty struct{}{} to 'acquire' a slot before starting work, and receive to 'release' when done. Once the buffer is full, further acquires block until a slot frees up. It's a lighter-weight pattern when you just want to cap parallelism, not pre-spawn workers.

Why use chan struct{} for a semaphore instead of chan bool?

An empty struct{} occupies zero bytes, so a channel of struct{} signals 'a slot' or 'an event' without carrying any data or wasting memory. It's the idiomatic Go choice whenever the value sent is irrelevant and only the act of sending or closing matters — semaphores, done channels, and set membership all use it.

Related lessons

• WaitGroup and Mutex