---
title: Pipelines
source: 'https://academia.sh/en/courses/go-concurrency/pipelines'
course: Concurrency
language: en
updated: '2026-09-14T10:59:31+00:00'
license: 'CC BY-SA 4.0'
objectives:
    - 'Builds a pipeline arrangement where each stage opens and closes its own output channel, measuring the count of values flowing through'
    - "Shows that the chain's order is preserved with no channel or synchronization tool at all, guaranteed by the single-producer/single-consumer relationship"
    - 'Proves that the stages ahead never stop when the consumer exits the chain early, by measuring the leak as a boolean'
---

# Pipelines

In a pipeline arrangement, each stage opens its own output channel and closes it itself; five values passing through a generation stage, a squaring stage, and an add-one stage flow through a chain where no stage disturbs the order. Limiting measurement: when the consumer gives up early from the middle of the chain, the stages ahead of it never stop at all — even though every stage's closing responsibility is correctly set up, none of them knows when the consumer at the far end of the chain has given up. The fix is the subject of the next lesson.

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.

```go
// 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`.

```go
// 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.