Module 2: Concurrency Patterns
Go Concurrency Patterns
Learning objectives
- Explain the core mental model behind Go Concurrency Patterns
- Apply Go Concurrency Patterns within Concurrency Patterns
- Identify important boundaries, trade-offs, and failure modes
- Produce concrete evidence from the practice exercise
Related: Go Channels and Select | Go Goroutines | Go Context | Go Sync Package and Mutexes | Go Index
Why Patterns Rather Than Primitives
Goroutines and channels are two primitives, and almost every real concurrency problem is one of about eight recombinations of them. Learning the recombinations is worth more than learning the primitives again, because each pattern comes with a known shutdown protocol — and shutdown is where concurrent code actually breaks.
THE CATALOGUE
pipeline stages connected by channels, each a goroutine
fan-out / fan-in N workers on one input, results merged to one output
worker pool a fixed goroutine count consuming a job channel
semaphore a buffered channel bounding concurrency
done channel broadcast cancellation via close()
or-done wrap a channel so it respects cancellation
tee duplicate one stream into two
bridge flatten a channel-of-channels
errgroup fan-out with first-error propagation and cancellationThree rules underlie all of them:
1. the SENDER closes a channel. never the receiver.
2. every goroutine must have a defined way to EXIT.
3. bound your concurrency. `go` in an unbounded loop is a resource leak.Verification analogy: these are the standard testbench topologies. A pipeline is a chain of TLM stages; fan-out/fan-in is a sequencer feeding several drivers whose responses converge on one scoreboard; the done channel is disable fork propagating through a process hierarchy. The topologies are the same; Go's version has explicit backpressure and a compiler-enforced type on every connection.
Pipeline
Each stage is a goroutine reading from an input channel and writing to an output channel it owns and closes.
func gen(nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out) // the OWNER closes
for _, n := range nums { out <- n }
}()
return out // returns a RECEIVE-ONLY channel
}
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in { out <- n * n } // exits when `in` closes
}()
return out
}
func main() {
for v := range square(gen(1, 2, 3, 4)) {
fmt.Println(v) // 1 4 9 16
}
} gen ──[chan int]──▶ square ──[chan int]──▶ main
THE SHUTDOWN CASCADE (this is the part that matters):
gen finishes → close(out)
→ square's `range in` ends
→ square's defer close(out) runs
→ main's `range` ends
closing propagates DOWNSTREAM automatically. each stage's `range`
terminating is what triggers the next close.Directional return types (<-chan int) are not decoration — they make it impossible for a consumer to close a channel it does not own, which is the source of the "close of closed channel" panic.
Fan-Out / Fan-In
func fanOut(in <-chan Job, workers int, fn func(Job) Result) <-chan Result {
out := make(chan Result)
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := range in { out <- fn(j) } // ALL workers share `in`
}()
}
go func() { wg.Wait(); close(out) }() // ← the closer goroutine
return out
} ┌── worker 1 ──┐
in ──[chan Job]──┼── worker 2 ──┼──[chan Result]──▶ out
├── worker 3 ──┤
└── worker 4 ──┘
FAN-OUT: N goroutines range over ONE channel. the runtime delivers each
value to whichever worker is free — a work queue for free, with
automatic load balancing.
FAN-IN: all N send to ONE channel.
THE CLOSER: with N senders, no single one can close. `wg.Wait(); close(out)`
in a dedicated goroutine sequences it. THIS IS THE MOST
IMPORTANT IDIOM IN GO CONCURRENCY — memorize it.There is no separate "distribute the work" code. Multiple goroutines ranging over one channel is the distribution, and it is naturally load-balanced: a worker that finishes early takes the next item.
Semaphore: Bounding Concurrency
sem := make(chan struct{}, 8) // capacity IS the limit
for _, job := range jobs {
sem <- struct{}{} // acquire — blocks when 8 are in flight
go func(j Job) {
defer func() { <-sem }() // release
process(j)
}(job)
}struct{} occupies zero bytes, so the channel is pure signalling. Note the acquire happens before go, not inside it — otherwise you spawn all N goroutines immediately and they merely queue on the acquire, which defeats the purpose.
The standard-library-adjacent version, which also handles waiting and errors:
import "golang.org/x/sync/errgroup"
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(8) // the semaphore, built in
for _, job := range jobs {
job := job
g.Go(func() error { return process(ctx, job) })
}
if err := g.Wait(); err != nil { ... } // the FIRST error, and ctx is cancellederrgroup should be your default for bounded, fallible, cancellable fan-out. Writing the equivalent by hand needs a WaitGroup, a semaphore channel, an error-capture mutex, and a context.CancelFunc — roughly forty lines with three places to get it wrong (Go Sync Package and Mutexes).
or-done: Making Any Channel Cancellable
The problem: for v := range ch cannot be cancelled. This wrapper fixes it.
func orDone[T any](ctx context.Context, in <-chan T) <-chan T {
out := make(chan T)
go func() {
defer close(out)
for {
select {
case v, ok := <-in:
if !ok { return }
select {
case out <- v: // the SEND must also be cancellable
case <-ctx.Done():
return
}
case <-ctx.Done():
return
}
}
}()
return out
}
// now this loop respects cancellation:
for v := range orDone(ctx, someChannel) { ... }The nested select on the send is the detail people omit. Without it, the goroutine blocks forever trying to deliver a value to a consumer that has already given up — a leak (Go Goroutines).
tee and bridge
// tee: duplicate one stream to two consumers
func tee[T any](ctx context.Context, in <-chan T) (<-chan T, <-chan T) {
out1, out2 := make(chan T), make(chan T)
go func() {
defer close(out1)
defer close(out2)
for v := range orDone(ctx, in) {
o1, o2 := out1, out2 // local copies we can nil out
for i := 0; i < 2; i++ {
select {
case <-ctx.Done():
return
case o1 <- v:
o1 = nil // ← nil disables this branch
case o2 <- v:
o2 = nil
}
}
}
}()
return out1, out2
}The o1 = nil trick is the nil-channel-disables-a-select-branch idiom (Go Channels and Select). Setting a local channel variable to nil after sending on it means the second loop iteration can only take the other branch — so each value goes to both consumers exactly once, in whichever order they are ready.
Examples
Example 1: A full regression pipeline with cancellation
package main
import (
"context"
"fmt"
"os"
"os/signal"
"runtime"
"sync"
"time"
)
type Job struct {
Test string
Seed uint64
}
type Result struct {
Job Job
Passed bool
Dur time.Duration
Err error
}
// ── STAGE 1: generate ─────────────────────────────────────────────
func generate(ctx context.Context, tests []string, seeds int) <-chan Job {
out := make(chan Job)
go func() {
defer close(out)
for _, t := range tests {
for s := 1; s <= seeds; s++ {
select {
case out <- Job{Test: t, Seed: uint64(s)}:
case <-ctx.Done():
return // ← cancellable send
}
}
}
}()
return out
}
// ── STAGE 2: fan out to N workers, fan in to one channel ──────────
func run(ctx context.Context, in <-chan Job, workers int) <-chan Result {
out := make(chan Result)
var wg sync.WaitGroup
for w := 0; w < workers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := range in {
r := simulate(ctx, j)
select {
case out <- r:
case <-ctx.Done():
return
}
}
}()
}
go func() { wg.Wait(); close(out) }() // ← the closer
return out
}
func simulate(ctx context.Context, j Job) Result {
start := time.Now()
d := time.Duration(30+j.Seed*7%80) * time.Millisecond
select {
case <-time.After(d):
return Result{Job: j, Passed: j.Seed%9 != 0, Dur: time.Since(start)}
case <-ctx.Done():
return Result{Job: j, Err: ctx.Err(), Dur: time.Since(start)}
}
}
// ── STAGE 3: aggregate, in main ───────────────────────────────────
func main() {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
defer stop()
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
workers := runtime.NumCPU()
tests := []string{"axi_smoke", "pcie_link", "wifi_rx"}
start := time.Now()
var passed, failed, cancelled int
var busy time.Duration
for r := range run(ctx, generate(ctx, tests, 12), workers) {
busy += r.Dur
switch {
case r.Err != nil:
cancelled++
case r.Passed:
passed++
default:
failed++
fmt.Printf("FAIL %-12s seed=%d\n", r.Job.Test, r.Job.Seed)
}
}
wall := time.Since(start)
fmt.Printf("\n%d passed %d failed %d cancelled\n", passed, failed, cancelled)
fmt.Printf("%v of work in %v wall (%.1fx speedup on %d workers)\n",
busy.Round(time.Millisecond), wall.Round(time.Millisecond),
float64(busy)/float64(wall), workers)
fmt.Printf("goroutines remaining: %d\n", runtime.NumGoroutine())
}FAIL axi_smoke seed=9
FAIL pcie_link seed=9
FAIL wifi_rx seed=9
33 passed 3 failed 0 cancelled
2.484s of work in 341ms wall (7.3x speedup on 8 workers)
goroutines remaining: 1Every property you want is present and each comes from one specific line: bounded concurrency (a fixed worker count), backpressure (unbuffered channels — a slow aggregator throttles the whole pipeline), cancellation (Ctrl-C and a timeout share one context), and clean shutdown (goroutines remaining: 1).
Ctrl-C mid-run produces partial results and a clean exit, because the cancellation propagates through every stage's select (Go Context).
Example 2: Fold-and-merge instead of a shared lock
package main
import (
"runtime"
"sync"
"testing"
)
// ── SHARED: every increment contends on one mutex ─────────────────
func countShared(lines []string) map[string]int {
m := make(map[string]int)
var mu sync.Mutex
var wg sync.WaitGroup
chunk := (len(lines) + 7) / 8
for i := 0; i < len(lines); i += chunk {
hi := min(i+chunk, len(lines))
wg.Add(1)
go func(part []string) {
defer wg.Done()
for _, l := range part {
mu.Lock()
m[key(l)]++ // ← lock per LINE
mu.Unlock()
}
}(lines[i:hi])
}
wg.Wait()
return m
}
// ── SHARDED: no sharing at all until the end ──────────────────────
func countSharded(lines []string) map[string]int {
workers := runtime.NumCPU()
partials := make([]map[string]int, workers)
var wg sync.WaitGroup
chunk := (len(lines) + workers - 1) / workers
for w := 0; w < workers; w++ {
lo := w * chunk
hi := min(lo+chunk, len(lines))
partials[w] = make(map[string]int)
if lo >= hi { continue }
wg.Add(1)
go func(w int, part []string) {
defer wg.Done()
local := partials[w] // each goroutine owns ITS map
for _, l := range part { local[key(l)]++ }
}(w, lines[lo:hi])
}
wg.Wait()
merged := make(map[string]int)
for _, p := range partials {
for k, v := range p { merged[k] += v }
}
return merged
}$ go test -bench=Count -benchmem
BenchmarkSequential-8 42 28.1 ms/op
BenchmarkShared-8 14 82.4 ms/op ← SLOWER than sequential!
BenchmarkSharded-8 271 4.4 ms/op ← 6.4x over sequentialThe shared-mutex version is three times slower than not parallelizing at all. Eight goroutines contending on one lock for every increment means the lock, not the CPU, is the bottleneck — plus cache-line ping-pong between cores on the mutex itself (Go Sync Package and Mutexes).
The sharded version has zero synchronization in the hot loop. Each goroutine writes only its own map and its own index of partials (distinct addresses, so no race — Go Memory Model), and the merge happens once, single-threaded, at the end.
THE GENERAL PRINCIPLE
avoid sharing, do not synchronize sharing.
this is the same lesson as Rust's rayon fold/reduce
(Rust Performance and Zero Cost Abstractions) and it applies to
any aggregation: counting, summing, histogramming, coverage merging.Example 3: A rate-limited, retrying dispatcher
package main
import (
"context"
"fmt"
"math"
"math/rand"
"time"
"golang.org/x/sync/errgroup"
"golang.org/x/time/rate"
)
type Job struct{ ID int }
func submit(ctx context.Context, j Job) error {
if rand.Float64() < 0.3 {
return fmt.Errorf("job %d: license server busy", j.ID)
}
select {
case <-time.After(50 * time.Millisecond):
return nil
case <-ctx.Done():
return ctx.Err()
}
}
// retry with exponential backoff and jitter
func withRetry(ctx context.Context, attempts int, fn func() error) error {
var err error
for i := 0; i < attempts; i++ {
if err = fn(); err == nil { return nil }
if ctx.Err() != nil { return ctx.Err() }
backoff := time.Duration(math.Pow(2, float64(i))) * 100 * time.Millisecond
jitter := time.Duration(rand.Int63n(int64(backoff / 2)))
select {
case <-time.After(backoff + jitter):
case <-ctx.Done():
return ctx.Err()
}
}
return fmt.Errorf("after %d attempts: %w", attempts, err)
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
limiter := rate.NewLimiter(rate.Limit(20), 5) // 20/sec, burst of 5
g, gctx := errgroup.WithContext(ctx)
g.SetLimit(4) // at most 4 concurrent
for i := 1; i <= 20; i++ {
job := Job{ID: i}
g.Go(func() error {
if err := limiter.Wait(gctx); err != nil { return err } // rate limit
return withRetry(gctx, 3, func() error { return submit(gctx, job) })
})
}
if err := g.Wait(); err != nil {
fmt.Println("regression failed:", err)
return
}
fmt.Println("all jobs submitted")
}Four concerns composed cleanly, each from one line: bounded concurrency (g.SetLimit(4)), rate limiting (limiter.Wait), retry with backoff and jitter (withRetry), and cancellation threading through all of them via gctx.
The jitter matters and is often omitted: without it, twenty jobs that fail simultaneously all retry at exactly the same moment, producing a thundering herd against the resource that was already overloaded. Adding a random fraction of the backoff spreads them out.
This shape — limit, rate, retry, cancel — is what a real dispatcher against a licensed EDA tool or a shared compute farm needs, and it is about forty lines in Go.
What Trips People Up
- A receiver closing the channel. Only the sender closes. With multiple senders, use the
rangeover a channel nobody closes. Hangs forever at the end of the data.- Forgetting the cancellable send. A goroutine that checks
ctx.Done()before the work but - Acquiring the semaphore inside the goroutine. All N goroutines spawn immediately and
Arc-style shared accumulator under a lock. Often slower than sequential. Shard and- Unbuffered channels between mismatched stages. Correct, but every handoff is a rendezvous.
time.Afterinside a loop. Allocates a timer per iteration that lives until it fires.- Retrying without jitter. Thundering herd.
- Reaching for a channel where a mutex belongs. A channel of capacity 1 holding one value
- Not measuring. Concurrency has real overhead; a parallel version can easily be slower.
wg.Wait(); close(out) closer goroutine.
not on the result send still leaks.
queue. Acquire before go.
merge.
A small buffer decouples rates; a huge one is an unbounded queue.
Use time.NewTimer and Reset, or a Ticker with defer Stop().
that goroutines take and put back is a lock with extra allocations.
Related
- Go Channels and Select — the primitives and their close rules
- Go Goroutines — leaks, bounding, and lifecycle
- Go Context — cancellation threading through every pattern
- Go Sync Package and Mutexes —
errgroup,WaitGroup, and when locks win - Go Memory Model — why sharding is race-free
- Go Scheduler — matching worker counts to
GOMAXPROCS - Go Profiling — finding contention the CPU profile hides
- Rust Threads and Channels — the same patterns with statically checked sharing
- Go Index — full topic map
Practice lab
Build the smallest runnable Go program that demonstrates Go Concurrency Patterns. Add a table-driven test, run the relevant Go diagnostics, inject one realistic failure, and explain the evidence used to correct it. Add an operational constraint such as concurrency, recovery, security, latency, or cost, and defend the resulting design trade-off.
Review questions
- What problem does Go Concurrency Patterns solve, and what assumptions does it rely on?
- Which boundary or failure case is easiest to miss, and how would you expose it?
- What alternative design would you consider, and what trade-off would change the decision?
- What artifact, trace, test, or metric proves that your implementation is correct?
Completion evidence
- A working artifact, annotated trace, or reproducible experiment
- At least one normal case and one deliberately failing or boundary case
- A concise explanation of the design choice and its trade-offs
- Saved output showing how correctness was evaluated