- Book: The Complete Guide to Go Programming
- Also by me: Hexagonal Architecture in Go — the companion book in the Thinking in Go series
- My project: Hermes IDE | GitHub — an IDE for developers who ship with Claude Code and other AI coding tools
- Me: xgabriel.com | GitHub
You have a stream of work to move through three steps. Read a
record, transform it, write the result. The naive version is one
big loop that does all three inline. It works until the transform
gets slow, or the write blocks on a remote service, and now the
read is stalled behind the write for no reason.
Go's answer is the pipeline. Each step becomes a stage. Each stage
is a goroutine that reads from an input channel and writes to an
output channel. The channels between them are the only shared
state. Wire the stages together and the runtime handles the
scheduling. This is not a framework or a library. It is a shape you
build from go, chan, and range, and the Go blog laid it out
back in 2014. What has changed
since is context, which fixes the part the original article had
to do by hand.
A stage is a function that returns a channel
Start with the smallest possible unit. A stage takes an input
channel and returns an output channel. It owns the goroutine that
feeds the output, and it owns closing that output.
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
out <- n * n
}
}()
return out
}
Three details carry the whole pattern. The output channel is
created by the stage, so the stage decides when it closes. The
defer close(out) runs when the goroutine returns, which happens
when the range in loop ends. And range in ends exactly when the
upstream channel closes. Closing cascades: close the source, and
every stage downstream drains and closes in turn.
The source is a stage with no input:
func gen(nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
out <- n
}
}()
return out
}
Composing them reads top to bottom:
func main() {
nums := gen(2, 3, 4)
squared := square(nums)
for n := range squared {
fmt.Println(n)
}
}
Output:
4
9
16
The main goroutine is the final consumer. It ranges over the last
channel until that closes, which only happens after gen finishes
and the close cascades through square. No WaitGroup, no manual
signalling. The channel closes are the coordination.
Who closes the channel
The rule that keeps pipelines correct: the goroutine that sends on
a channel is the goroutine that closes it. Never the receiver. A
receiver that closes a channel someone else still sends to causes a
panic: send on closed channel, and it will not be the receiver's
goroutine that panics.
In the stages above, each stage closes only its own out. It never
touches its in. That channel belongs to the upstream stage. Keep
that boundary and closing never races. Break it and you get
intermittent panics that only show up under load.
The one place beginners get this wrong is fan-out, where several
goroutines share one output channel. There, no single sender can
close it, because the others are still going. That case needs a
sync.WaitGroup and one closer goroutine that waits for all senders
to finish before it closes.
func merge(cs ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
for _, c := range cs {
wg.Add(1)
go func(c <-chan int) {
defer wg.Done()
for n := range c {
out <- n
}
}(c)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
The closer goroutine does nothing but wait and close. It is the
only writer allowed to close out, and it does so only after every
sender has returned. That is the single-closer rule holding even
when the senders are plural.
Unbuffered channels give you backpressure for free
Look again at make(chan int). No buffer. That is deliberate.
An unbuffered channel send blocks until a receiver is ready. So if
your write stage is slow, its receive falls behind, which blocks the
transform stage's send, which blocks the read stage's send. The
whole pipeline runs at the speed of its slowest stage, and nothing
piles up in memory in between. That is backpressure, and you get it
by doing nothing.
Add a buffer and you trade memory for slack:
out := make(chan int, 100)
Now the fast stage can run up to 100 items ahead of the slow one
before it blocks. That smooths out bursts. It also means up to 100
items sit in the channel, and if the slow stage crashes or the
context cancels, those items are in flight and unaccounted for. Use
a buffer when you have a measured burst to absorb. Reach for
unbuffered by default. The absence of a buffer is a feature: it
tells you, structurally, that no stage can outrun the one behind it.
Cancellation: the part the 2014 blog did by hand
The original pipeline article used an explicit done channel that
every stage watched. Every send became a select on the real send
and <-done. It worked, but you threaded that channel through every
signature by hand. context is that pattern standardized. A
cancelled context closes its Done() channel, and every stage
selects on it.
Here is square with cancellation:
func square(
ctx context.Context, in <-chan int,
) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
select {
case out <- n * n:
case <-ctx.Done():
return
}
}
}()
return out
}
The select is the whole point. Without it, out <- n*n blocks
forever if the consumer stops reading, and the goroutine leaks. With
it, a cancelled context wins the select, the goroutine returns,
defer close(out) fires, and the close cascades downstream. The
source stage needs the same treatment on its send so it stops
producing when the context dies.
func gen(ctx context.Context, nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
select {
case out <- n:
case <-ctx.Done():
return
}
}
}()
return out
}
Now cancellation is honest end to end. The consumer stops reading,
cancels the context, and every stage from source to sink notices on
its next send attempt and shuts down. No goroutine is left blocked
on a send that will never complete.
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
out := square(ctx, gen(ctx, 2, 3, 4, 5))
for n := range out {
fmt.Println(n)
if n == 9 {
cancel() // stop early
break
}
}
}
When main calls cancel() and breaks, it stops reading out.
The square goroutine, blocked trying to send the next value, wakes
on ctx.Done() and returns. The gen goroutine does the same. Both
close their outputs on the way out. Nothing leaks, even though the
consumer walked away mid-stream.
The one thing that still leaks
Cancellation covers the send side. It does not cover a stage that is
blocked on something other than a channel send or receive. A stage
in the middle of a slow http.Get, a sql query, or a disk read is
not sitting in your select. It is inside a library call, and it
returns to the select only when that call finishes.
The fix is the same as everywhere else in Go: pass the context down
into the blocking call so it can cancel it too. An
http.NewRequestWithContext, a QueryContext, a net.Conn with a
deadline derived from ctx.Deadline(). The pipeline's select
handles the channel plumbing. The context inside the I/O call
handles the syscall. You need both. One without the other still
leaves a goroutine parked on a slow read after the consumer is long
gone.
What you actually get
A Go pipeline is four rules doing the heavy lifting:
- A stage owns its output channel and closes it with
defer close. - Only the sender closes; closing cascades downstream through
range. - Unbuffered channels between stages give you backpressure without a single line of throttling code.
- Every send selects on
ctx.Done()so cancellation reaches every stage.
Keep those four and the pipeline stays composable. You can insert a
stage, fan out to parallel workers and merge them back, or swap
the source, without rewriting the coordination. The channels are the
contract, and the contract holds.
Channel-composed pipelines are Go leaning into what its runtime does
well, and the details (buffered versus unbuffered, when close
cascades, how the scheduler parks a blocked send) are the kind of
thing The Complete Guide to Go Programming works through from the
runtime up. Hexagonal Architecture in Go is the companion for the
next question: where a pipeline like this belongs behind a port so
the stages stay testable and the transport stays swappable.
Top comments (0)