Pipelines

The previous lesson measured how many sides the closing responsibility should belong to: zero sides led to a leak, more than one led to a panic, and the correct answer was exactly one side. But that production was a single step — input taken, squared, result written. Real processing usually passes through more than one step, and each step has its own speed, its own output. This lesson chains these steps: a pipeline — a chain where each stage takes the previous stage’s output as its own input, produces its own output, and closes its own output channel itself. The closing responsibility is no longer in a single place; it sits separately in every link of the chain.

Compared to the previous two lessons’ arrangements, a pipeline’s shape is different. In the worker pool, many goroutines read from the same channel (a shared queue); in fan-out, many goroutines wrote to the same channel (a shared destination). In a pipeline, each channel sits between exactly one writer and one reader — no sharing, only chaining. This means the hardest problem from the previous two lessons (how many sides should take on the closing responsibility) never comes up here at all: every channel has exactly one owner, the stage that opens it. The chain’s own internal structure gets determinism for free; the real question in this lesson comes up in its second half — from outside the chain, from a consumer giving up.

Pipeline: Every Stage Carries Its Own Closing Responsibility

The measurement below builds a three-stage line: generate produces a range of numbers, square squares every value, addOne adds one to every value. All three are written the same way: each opens a channel, starts a goroutine doing its own work, and once that goroutine finishes its work, closes its own channel itself and returns the channel itself to its caller.

// stages.go — a three-stage pipeline: each stage opens its own output and closes it itself
package main

import "fmt"

// generate writes the values in 1..n to its own output and closes it once done.
func generate(n int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for i := 1; i <= n; i++ {
			out <- i
		}
	}()
	return out
}

// square replaces each value with its square; closes its own output once the input closes.
func square(input <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for value := range input {
			out <- value * value
		}
	}()
	return out
}

// addOne adds one to each value; closes its own output once the input closes.
func addOne(input <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for value := range input {
			out <- value + 1
		}
	}()
	return out
}

func main() {
	stage1 := generate(5)
	stage2 := square(stage1)
	stage3 := addOne(stage2)

	fmt.Println("-- three-stage pipeline, full consumption --")
	results := make([]int, 0, 5)
	for s := range stage3 {
		results = append(results, s)
	}
	fmt.Printf("value count flowing through: %d\n", len(results))
	fmt.Println("values:", results)
	fmt.Println("each stage's reason for stopping: what it waited for arrived (the previous stage's channel closed)")
}

-- three-stage pipeline, full consumption --
value count flowing through: 5
values: [2 5 10 17 26]
each stage's reason for stopping: what it waited for arrived (the previous stage's channel closed)

CP11. Five values (1..5) are squared in order (1,4,9,16,25) and each gets one added (2,5,10,17,26) — the result list is printed in this exact order, and this order arises with no sorting anywhere at all. The reason is different from the previous two lessons’ fan-out/fan-in arrangements: there, more than one producer wrote to the same channel at the same time, which is why order could be lost and the collected results needed sorting. Here every channel has exactly one writer and exactly one reader (generate’s single goroutine, square’s single goroutine, the main goroutine itself); in a single-writer/single-reader relationship, a channel always preserves send order, because there is no second side to race against.

The main goroutine itself is also part of the chain: it is the sole reader of stage3, and the consuming loop runs until addOne’s goroutine closes its own output. Four goroutines run from the start of the chain to the end — the three of generate, square, addOne, plus the main goroutine itself — but none of them needs a special synchronization tool (a wait group, a select) to wait for another; the channels themselves both carry the data and enforce the order.

CP12. Every stage’s reason for stopping is the same too: square and addOne close their own output channel once their own input channel closes (once for value := range input ends) — defer close(out) guarantees this even if a panic happens while consuming the input. generate‘s own goroutine closes the same way with the same defer, once its loop ends after sending five values. Every link in the chain applies the previous lesson’s “the closing responsibility belongs to exactly one side” principle separately, for its own channel — all three channels in the chain are closed by their own producer, none of them is shared.

This arrangement’s scaling rests on the same principle: if a fourth or fifth stage were added to the chain, each new stage would only take the previous stage’s output as its input and open its own channel — a call like stage4 := anotherTransform(stage3) can be added to the chain without touching any of the previous three stages at all. Determinism is independent of the chain’s length, because every added link follows the same single-writer/single-reader constraint; what is true for a three-stage chain (order preserved, closing propagating on its own) stays true for a ten-stage chain too.

The term pipeline is used here with a qualifier: this is a stream-processing pipeline — a separate concept from a shell programming command pipeline or the steps in a continuous delivery pipeline, sharing only the name. What is common between them is that in both, one output becomes the next step’s input; what differs is that a “stage” here is not a shell process or a deployment step, it is a goroutine built around a pair of channels.

Every stage function’s return type reflects this too: generate, square, and addOne all return <-chan int — a channel that can only be received from, not sent to. This prevents the caller from mistakenly sending its own value into the returned channel, at compile time; the channel’s sole writer is the goroutine of the function that created it, and the return type encodes this exclusivity into the type system. The same restriction holds on the input parameters too: square and addOne take girdi <-chan int — they can only read from their own channel, they cannot write to it. This small type detail establishes the “every channel has exactly one owner” principle not just as a design intention, but as a rule the compiler checks.

square and addOne’s bodies follow the exact same pattern, word for word: open an output channel, start a goroutine, range over the input channel, transform each value and write it to the output, close your own output once the input ends. This repetition is no accident — any transforming stage follows the same three steps, the only thing that changes is the transformation in between (value * value or value + 1). generate is one end of this pattern (no input channel, only production) and the consuming loop in the main goroutine is the other end (no output channel, only consumption); every stage in between repeats the same template, having both an input and an output. No matter how many stages a chain has, the closing responsibility always sits inside this same template, in the defer close(out) line.

Giving Up Early: The Stages Ahead Never Stop

The measurement above consumed all the way to the end of the chain — the loop ran until every value in stage3 was taken. But a consumer does not always go all the way to the end of the chain: it might stop the moment it finds the first result it was looking for, it might get an error, or it might give up for some other reason — a search operation, say, might not want the remaining values at all once it finds its first match. The measurement below sets this situation up in its plainest form: a two-stage line opens and the consumer gives up after the second value, exiting the for range loop with a break.

// early.go — if the consumer gives up early, the stages ahead of it never stop (fixed in the next lesson)
package main

import (
	"fmt"
	"runtime"
)

func generate(n int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for i := 1; i <= n; i++ {
			out <- i
		}
	}()
	return out
}

func square(input <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for value := range input {
			out <- value * value
		}
	}()
	return out
}

func main() {
	base := runtime.NumGoroutine()

	stage1 := generate(5)
	stage2 := square(stage1)

	taken := 0
	for range stage2 {
		taken++
		if taken == 2 {
			break // the consumer gives up after two values
		}
	}

	fmt.Println("-- consumer gives up after 2 values --")
	fmt.Printf("values taken: %d\n", taken)
	fmt.Printf("did a non-stopping stage remain: %v\n", runtime.NumGoroutine() > base)
}

-- consumer gives up after 2 values --
values taken: 2
did a non-stopping stage remain: true

CP13. After two values are taken, the loop exits with break — and the rest of the chain is never told about this at all. Exactly where the chain gets stuck depends on timing, but it is instructive to count the possible ends: either square’s goroutine gets stuck trying to send its own squaring result to stage2 — because the main goroutine no longer takes it — or it is waiting to take the next value from stage1. If square is stuck sending, generate’s goroutine also gets stuck when it tries to send its next value to stage1, because square is now too busy to read it. Since the chain is short, both possible situations arrive at the same result: two goroutines (generate and square) get permanently blocked. Since the consumer took exactly two values, square gets stuck on its third send, generate on its fourth; which order they get stuck in depends on timing, but the count of those stuck does not change — which is why the measurement only prints the runtime.NumGoroutine() comparison, not how many goroutines remain.

CP14. This arises for a different reason than the unclosed-channel leaks seen in the previous two lessons. There, the problem was that closing was never done at all; here, every stage correctly carries its own closing responsibility — both generate and square will close their own output once their work is done. The problem is that the work never finishes: generate’s “finishing” depends on it sending its fifth value, but if no one is taking after the third value, the send never completes, and the defer close line is therefore never reached either. A correctly built closing responsibility rests on the assumption that every link in the chain waits for the other; this assumption breaks when the consumer exits early, and no stage has any mechanism at all to notice this.

This is the chained version of the previous lesson’s “even the correct setup assumes producers will eventually finish” observation. There, there was a single assumption (the producers will finish); here, every stage in the chain assumes the one before it and the one after it are running normally, and the chain effect of these assumptions grows as the chain gets longer. In a three-stage chain, a consumer exiting early can affect two goroutines; in a ten-stage chain, the same early exit can block nine goroutines — the size of the leak grows together with the chain’s length, independent of how early the consumer gave up.

Placing the three lessons’ leak measurements side by side makes the difference clear. In the worker pool, the cause of the leak was closing being forgotten — the close(jobs) line was never written at all. In fan-out, the cause was the responsibility being given to no one at all — the third-party goroutine was never set up. Here, the responsibility is both written and given to the right side, but the event that should trigger it (the input running out) never happens at all, because the side that would have consumed the input has already stepped away. All three are caught by the same boolean measure (runtime.NumGoroutine() > base), but the root cause of all three sits in three separate places: missing closing, ownerless closing, untriggered closing.

This limit is not solved within the lesson — it is only measured. Equipping a stage with the information “the consumer is no longer listening” requires that stage to check a signal coming from outside its own loop; every stage’s for value := range input loop would need an additional branch, and that branch would need to come from a single source reaching all stages at the same time. This is exactly the next lesson’s subject.

Summary

• In a pipeline (stream-processing pipeline), each stage opens its own output channel and closes it itself; the closing responsibility does not sit in a single place, it sits separately in every link of the chain.
• Because every channel in the chain has exactly one writer and one reader, order is preserved with no sorting tool needed at all — unlike the shared channels in fan-out/fan-in.
• The term pipeline is used here in the stream-processing sense, a separate concept from a shell or delivery pipeline.
• When the consumer exits the chain early, the stages ahead of it never find out and stay permanently blocked; the leak is again measured only as a boolean.
• This leak’s source is different from the previous lessons’ unclosed channels: the closing responsibility is correctly built, but the work itself never finishes.
• The size of the leak grows together with the chain’s length: a consumer exiting early can block one goroutine in a short pipeline, the rest of the chain in a long one.

Next Step

The fix for the leak seen in this lesson requires a stage to listen not only to its own input channel, but also to a give-up signal coming from outside — and this signal needs to reach every stage from the beginning to the end of the chain at the same time, without being passed to each stage by hand one by one. The next lesson measures the tool that carries this signal: how a cancellation reaches every waiter in the chain at the same time, and what happens when a goroutine never checks this signal at all.