Pipelines: One Stage Feeds the Next

A pipeline is a chain of stages connected by channels. The first stage produces values; each later stage receives on an inbound channel, does some work, and sends on an outbound channel; the last stage’s channel is what you consume. Nothing here is a new language feature. A pipeline is just goroutines and channels arranged in a particular shape, but it is such a useful shape that it deserves its own chapter, because once you see it you will reach for it constantly: streaming a file through a series of transforms, processing records as they arrive, any place where data flows in one direction through independent steps.

The shape has one strict convention that makes the whole thing compose. Every stage takes a receive-only channel and returns a receive-only channel, and it closes its output when its input runs dry. Get that rule right and stages snap together like pipe fittings.

A generator and a stage

The first stage has no inbound channel, only outbound. It is a generator: it turns something (a slice, a file, a network socket) into a stream of values on a channel.

// gen turns a fixed list of ints into a channel that emits them, then closes.
func gen(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, n := range nums {
			out <- n
		}
		close(out)
	}()
	return out
}

Read the type signature carefully, because it is the whole pattern in miniature. gen returns <-chan int, a receive-only channel. The caller can only read from it, never send or close, which is exactly the contract you want: the stage owns its channel and is the only thing that closes it. Inside, a goroutine walks the input, sends each value, and calls close(out) when done. The function itself returns immediately, handing back the channel while the goroutine keeps feeding it.

A middle stage looks almost identical, except it has an inbound channel too:

// square reads from in, emits each value squared, and closes out when in drains.
func square(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		for n := range in {
			out <- n * n
		}
		close(out)
	}()
	return out
}

The for n := range in loop is the load-bearing detail. Ranging over a channel receives values until the channel is closed and drained, then ends the loop cleanly. So square keeps working as long as its upstream keeps sending, and the moment gen closes its channel, square’s range loop ends, square closes its channel, and the signal propagates down the line. Close flows downstream, one stage at a time, with no coordination beyond each stage minding its own output.

Now compose them. gen feeds square, and you consume the result by ranging over it:

func main() {
	for v := range square(gen(2, 3, 4, 5)) {
		fmt.Println(v)
	}
}

4
9
16
25

Look at the output order: 4, 9, 16, 25, the squares of 2, 3, 4, 5 in the exact order they entered. This is worth pausing on, because the pipeline is running two goroutines concurrently and you might expect the ordering to be up for grabs. It is not. A single channel preserves order: values come out in the order they went in, first in first out. The concurrency buys you overlap (while square is squaring one number, gen can be producing the next), but along any one channel the sequence is fixed. Order only gets scrambled when you split work across multiple channels, which is the next chapter’s subject.

Stages compose because the types line up

Because every stage has the same shape (<-chan int in, <-chan int out), stages nest directly. Feeding square into another square is just another function call:

func main() {
	// Two square stages composed: n -> n^2 -> n^4.
	for v := range square(square(gen(2, 3, 4))) {
		fmt.Println(v)
	}
}

16
81
256

Those are the fourth powers of 2, 3, 4, still in order. Each square runs in its own goroutine, so the two stages execute concurrently, passing values along as they go. You can extend the chain as far as you like, and a real pipeline often has a generator, several transform stages, and a final consumer, each an independent goroutine linked only by its channels. The uniform signature is what lets you read square(square(gen(...))) as a data-flow diagram written left to right.

Stopping early without leaking

The pipelines so far run to completion: the generator exhausts its input, close propagates, every goroutine finishes. But what if the consumer wants to stop early, after the first value or the first match? If you simply stop ranging over the final channel and walk away, the stages upstream are still trying to send. A stage blocked on out <- n with nobody receiving will block forever. Its goroutine never returns, and neither does anything it is holding onto. That is a goroutine leak, and it gets its own chapter shortly.

The fix is to give every stage a second way out. You pass a shared done channel into each stage, and instead of a bare send, the stage uses a select (covered back in the select chapter) that races the send against a receive on done. When you close done, every blocked send is unblocked by the <-done case instead, and the goroutine returns.

// gen emits nums, but abandons the send the moment done is closed, so the
// goroutine can always return instead of blocking forever on a full channel.
func gen(done <-chan struct{}, nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			select {
			case out <- n:
			case <-done:
				return
			}
		}
	}()
	return out
}

Every stage takes the same done and wraps its send the same way. The consumer takes what it wants, then closes done to tear the whole line down:

func main() {
	done := make(chan struct{})
	results := square(done, gen(done, 2, 3, 4, 5, 6, 7, 8))

	// Consume only the first value, then stop the whole pipeline.
	fmt.Println("took:", <-results)
	close(done)

	time.Sleep(50 * time.Millisecond) // let the abandoned goroutines return
	fmt.Println("goroutines still running:", runtime.NumGoroutine())
}

took: 4
goroutines still running: 1

We read exactly one value, closed done, and every stage’s select fell through to its <-done case and returned. The 1 is main itself, the only goroutine left. The defer close(out) in each stage still runs on the way out, so downstream stages see their input close and unwind in turn. Closing a channel is a broadcast: every goroutine selecting on done observes it at once, which is why one close(done) can retire an entire pipeline.

A struct{} channel is the idiom for a pure signal. struct{} is a type that holds no data and occupies no memory, so chan struct{} says “this channel carries no value, only the fact that something happened.” Closing it is the event.

A note on buffering the stages

The channels above are unbuffered, so every send blocks until the next stage is ready to receive. That is lockstep: gen cannot produce value two until square has taken value one. Often that is fine, and it keeps memory flat, because no stage runs ahead of its consumer. When a stage’s work is bursty or its timing is uneven, though, giving its output channel a small buffer (make(chan int, 100)) lets it run a little ahead and smooths the flow, so a momentary stall in one stage does not immediately stall the one feeding it. Buffering is a throughput tuning knob, not a correctness fix. The close-propagates and order-preserved guarantees hold identically whether the channels are buffered or not, because a range still ends only when the channel is closed and fully drained, buffer included. Reach for a buffer when profiling says a stage is starving, not by default.

In real code you would usually thread a context.Context (from the context chapter) through the pipeline instead of a hand-rolled done channel, selecting on <-ctx.Done(). It is the same mechanism with cancellation, deadlines, and a value bag layered on top. The done-channel here is the bare bones of what context does for you, and seeing it in the raw makes the context version obvious.

Final thoughts

A pipeline is goroutines linked by channels, one stage per step, each stage receiving on one channel and sending on another. The discipline that makes it compose is uniform and simple: return a receive-only channel, and close it when your input drains, so that close propagates down the chain on its own. A single channel preserves order, so a straight pipeline is deterministic even though its stages run concurrently. And because a stage blocked on a send with no receiver leaks its goroutine, any pipeline that can stop early needs an escape hatch: a shared done channel (or a context) selected against every send, closed once to unwind the whole line at once. Next we split the stream across several goroutines, which is where bounded parallelism, and scrambled ordering, begin.

Next: worker pools: fan-out and fan-in — running a fixed set of workers over a shared stream, and merging their results back into one.