Pipelines, Fan-out & Fan-in

"We should have some ways of coupling programs like garden hose — screw in another segment when it becomes necessary to massage data in another way." — Doug McIlroy, internal Bell Labs memo (1964)

Part IV was the vocabulary: goroutines, channels, select, the sync package, context. Each was introduced on its own terms, in examples sized to demonstrate one idea. Part V is what happens when they meet a real system, and the shapes that recur often enough to have names.

The first of those shapes is the pipeline, and it is the oldest idea in this book. McIlroy's garden-hose memo led to Unix pipes six years later, and the insight has not changed since: a program that reads a stream and writes a stream can be joined to any other program that does the same, and the composition is more useful than the parts. Go's version replaces processes with goroutines and pipes with channels, which makes the segments cheap enough to use casually and puts them inside one program where they share a type system.

What Go adds — and what most of this chapter is about — is that the segments can fan out. A Unix pipe is strictly one-to-one. When a stage is slow, Go lets you run ten copies of it reading the same input, and merge their outputs back into one stream. That is the pattern that makes concurrency pay for itself, and it is also where the leaks live.

A stage is a function that returns a channel

The convention is uniform across Go, and once you see it the rest follows.

Four rules are encoded in those signatures, and every one of them prevents a specific bug.

A stage takes <-chan T and returns <-chan U. The receive-only return type means the caller physically cannot close the channel — the compiler rejects it — which enforces Chapter 2's rule that only the sender closes.

The goroutine is started inside the stage, not by the caller. The function returns immediately with a channel that will fill up later. This is what makes composition work: double(square(gen(...))) builds the whole pipeline in one expression, and nothing has run to completion yet.

defer close(out) is the first line of the goroutine. When the input channel closes, the range ends, the function returns, and the deferred close propagates the end-of-stream signal downstream. A stage that forgets it leaves its consumer ranging forever — a hang with no error, exactly the leak of Chapter 2 dressed as an ordinary loop.

The stage owns exactly one channel: the one it created. It reads someone else's and writes its own, and that division is what makes ownership unambiguous all the way along the chain.

Notice also that the intermediate channels are unbuffered. That is the right default here: it means each stage runs only as fast as the next one consumes, so a slow final stage exerts backpressure all the way to the source rather than letting work pile up in memory.

Fan-out: many workers, one input

When one stage is slower than the rest, run several copies of it against the same input channel.

The mechanism is worth being precise about, because it looks like magic and is not. Several goroutines ranging over the same channel is well-defined: each value goes to exactly one of them, chosen by the runtime, and when the channel closes every one of their range loops ends. There is no coordination, no work-stealing queue, and no need for one — the channel is the queue, and receive is atomic.

That gives you a distribution property for free: work is pulled rather than pushed, so a worker that finishes early takes the next item, and a worker stuck on something slow simply does not. Compare that with splitting the input into three slices up front, where one unlucky partition full of slow items leaves two workers idle while the third grinds. Pull-based distribution is self-balancing, and it is the single strongest argument for this shape.

Two limits are worth knowing now. Fan-out only helps when the stage is the bottleneck — parallelising a fast stage in front of a slow one just moves the queue. And the number of workers is a decision, not a detail: the next chapter is entirely about why it must be bounded, and what happens when it is not.

Fan-in: many outputs, one channel

Fan-out produces several channels. Merging them back into one is the piece that needs care, because it is where "who closes this?" gets hard.

The wg.Wait(); close(out) goroutine is the whole trick, and it deserves to be understood rather than copied.

out has several senders, so by Chapter 2's rule none of them may close it — a close while another sender is live panics that sender and kills the process. But somebody must, or the consumer's range never ends. The resolution is a separate goroutine whose only job is to wait for every sender and then close, which orders the close after the last send by construction rather than by hope.

It has to be a goroutine, not a plain wg.Wait() before the return. merge returns immediately with a channel that fills later, exactly like every other stage; blocking on Wait in the function body would mean nothing could receive from out until every sender had finished, and since the senders block on out, that deadlocks instantly. This is the most common way people get merge wrong, and the symptom is a total hang on the first item.

One playground note: this book's interpreter cannot iterate a slice of channels with the two-variable form (for _, c := range cs), so the loop above is indexed. for i := range cs works too. Both are ordinary Go; only the value form trips this runtime.

Abandoning a pipeline leaks it

Everything so far assumed the consumer drains to the end. Here is what happens when it does not, and it is worth watching closely.

The three cases are the lesson. The first producer ran to completion and set its flag. The second is still sitting on out <- 3 and will be for as long as the process lives, holding its stack, its loop variables, and — in a real pipeline — a database cursor, an open file, or a buffer. The third took the same early exit and returned cleanly, because cancel() reached it.

Now scale the middle case. That was one goroutine. A pipeline with four stages and a fan-out of eight leaks every stage above the abandonment point, so one break in a handler can strand a dozen goroutines. Do that per request and the graph is the straight line from Part IV: memory climbing with no sawtooth, goroutine count climbing in lockstep, and a restart as the only remedy.

Every send in a cancellable stage becomes a select. That is the whole change, and it must be applied at every stage — one stage without it is enough to strand everything above it, because that stage will block on its own send and never reach its ctx.Done() check.

The older idiom, still common in code written before Go 1.7, uses a plain done chan struct{} closed by the consumer, which works identically because closing broadcasts. context supersedes it by carrying deadlines and threading through APIs you do not own, but the mechanism underneath is the same closed channel.

Generic stages, and when not to build one

Type parameters make pipeline stages reusable, and this is one of the cases where generics earn their place.

Map changes the element type — <-chan int in, <-chan string out — with the compiler checking both ends. Before generics this needed either chan interface{} and a type assertion per element, or a hand-written stage per type pair. It is a genuine improvement, and Map, Filter and Take cover most of what a pipeline library needs.

Now the part that matters more than any of the above. Most pipelines should not be pipelines.

That last example processes eight integers using four goroutines and three channels, and a plain loop would do it in three lines with no concurrency, no leak risk, and better performance. Channel operations are not free — each send and receive is a scheduler interaction costing on the order of a hundred nanoseconds — so a pipeline over cheap per-item work is slower than the loop it replaced, sometimes by an order of magnitude.

A pipeline earns its complexity under three conditions, and it wants at least two of them: the stages block on something external (network, disk, a lock), so overlapping them recovers real time; the stream is unbounded or large enough that you cannot hold it in memory; or the stages have genuinely different rates and you want backpressure to manage them rather than doing it yourself. Absent those, you are paying goroutine coordination costs and inheriting an entire class of lifecycle bugs to express a for loop with extra steps.

Common Mistakes

1. Forgetting defer close(out) in a stage. The downstream range never ends and its goroutine never returns — a hang with no error, and the most common pipeline bug.

2. Closing a merged channel from a sender. With several senders, any one closing panics the others. Use a WaitGroup and a dedicated goroutine that waits, then closes.

3. Calling wg.Wait() in the body of merge instead of a goroutine. The senders block on the output channel that nobody can drain until Wait returns, so it deadlocks on the first item.

4. Returning chan T instead of <-chan T. The receive-only type is what stops a caller closing a channel it does not own. Give it away for free and the compiler stops helping.

5. Breaking out of a consumer loop without cancelling. Every stage upstream parks on its next send, permanently. One break in a handler can strand a dozen goroutines and a database cursor.

6. Adding cancellation to some stages but not all. A stage without a select on its send blocks there and never reaches its own ctx.Done() check, stranding everything above it. It has to be every stage.

7. Buffering intermediate channels to "speed things up". Buffers remove the backpressure that was keeping memory bounded, so a slow final stage now accumulates a queue in your heap instead of slowing the source.

8. Fanning out a stage that is not the bottleneck. It adds goroutines, a merge, and new failure modes, and moves the queue rather than shortening it. Measure first.

9. Unbounded fan-out. One worker per input item makes concurrency a function of load, which is the subject of the next chapter and the usual way a pipeline exhausts a connection pool.

10. Building a pipeline for cheap per-item work. Channel operations cost far more than the arithmetic they carry. If nothing blocks and the data fits in memory, write the loop.

Interview Corner

Hands-on

Summary

• A stage takes <-chan T and returns <-chan U, starts its own goroutine, and returns immediately — which is what lets stages compose as expressions.
• defer close(out) is the first line of every stage's goroutine. Forgetting it leaves the consumer ranging forever: a hang with no error.
• Return receive-only channels. The type is what stops a caller closing a channel it does not own, and the compiler enforces it.
• Unbuffered intermediate channels give you backpressure. A slow final stage throttles the source instead of accumulating a queue in your heap.
• Fan-out is several goroutines ranging one channel. Each value goes to exactly one, distribution is pull-based and self-balancing, and no coordination is needed.
• Fan-in needs a WaitGroup and a dedicated closing goroutine. No sender may close a shared channel, and wg.Wait() in the function body deadlocks instead.
• An abandoned pipeline leaks every stage above the abandonment point — parked on a send, holding cursors and buffers, with no error and no way to notice.
• Cancellation means every send becomes a select, at every stage. One stage without it blocks on its own send and strands everything upstream.
• Generics make stages reusable across types — Map can change <-chan int into <-chan string with both ends checked.
• Most pipelines should be loops. Channel operations cost far more than the work they usually carry; a pipeline earns its complexity when stages block, the stream is large, or rates genuinely differ.

This chapter fanned out to three workers because three looked reasonable. That number was never justified, and in a real system it is the most consequential line in the file — one goroutine per item is a design that works beautifully in testing and takes down a database in production, because concurrency becomes a function of load rather than of what your dependencies can absorb. The next chapter is worker pools and bounded concurrency: how to size a pool against the scarcest resource downstream rather than the arrival rate, why a full channel is a feature, and why "add more workers" is usually the wrong response to a queue.