Pipelines in Go and Cancellation at Every Single Stage

A pipeline is the pattern to reach for when a task is a sequence of steps over a stream of items: read, parse, filter, transform, write. Written as one loop, the steps run one after another for each item. Written as a pipeline, each step is its own goroutine, items flow between them over channels, and while one stage is working on item 7, the next is already working on item 6. The structure also makes each step small enough to test and replace.

This guide builds a complete pipeline from stage functions, shows how cancellation passes through every stage, and covers backpressure, measurement and testing. It builds on channels and fan-out and fan-in, and it sets up pipeline errors. Everything ran on Go 1.25.6.

The shape of a stage

The convention from the Go blog's pipelines article is simple: a stage is a function that takes a receive-only channel, returns a receive-only channel, and owns the channel it returns. It starts a goroutine that reads the input, writes the output, and closes the output when the input is exhausted. Adding a context makes it stoppable, and the context has to guard both directions: the send to the next stage and the receive from the previous one.

package main

import "context"

// send delivers v on out, reporting false if ctx ends first.
func send[T any](ctx context.Context, out chan<- T, v T) bool {
	select {
	case out <- v:
		return true
	case <-ctx.Done():
		return false
	}
}

// recv takes a value from in, reporting false when in is closed or ctx ends.
func recv[T any](ctx context.Context, in <-chan T) (T, bool) {
	select {
	case v, ok := <-in:
		return v, ok
	case <-ctx.Done():
		var zero T
		return zero, false
	}
}

// Generate emits the values and closes the channel.
func Generate[T any](ctx context.Context, vals ...T) <-chan T {
	out := make(chan T)
	go func() {
		defer close(out)
		for _, v := range vals {
			if !send(ctx, out, v) {
				return
			}
		}
	}()
	return out
}

// Map applies f to every value.
func Map[T, U any](ctx context.Context, in <-chan T, f func(T) U) <-chan U {
	out := make(chan U)
	go func() {
		defer close(out)
		for {
			v, ok := recv(ctx, in)
			if !ok || !send(ctx, out, f(v)) {
				return
			}
		}
	}()
	return out
}

// Filter passes only the values for which keep returns true.
func Filter[T any](ctx context.Context, in <-chan T, keep func(T) bool) <-chan T {
	out := make(chan T)
	go func() {
		defer close(out)
		for {
			v, ok := recv(ctx, in)
			if !ok {
				return
			}
			if keep(v) && !send(ctx, out, v) {
				return
			}
		}
	}()
	return out
}

// Collect reads everything from in into a slice. It is the usual last stage.
func Collect[T any](ctx context.Context, in <-chan T) []T {
	var all []T
	for {
		v, ok := recv(ctx, in)
		if !ok {
			return all
		}
		all = append(all, v)
	}
}

Three conventions repeat in every stage. The stage creates its output channel, so it is the one that closes it: defer close(out) is the first line inside the goroutine. Every send is a select with ctx.Done() (the send helper), so a downstream stage that has stopped reading cannot block this one forever. And every receive is the same (the recv helper), so an upstream stage that has stalled without closing cannot hold this one hostage either. End-of-stream propagates down the chain because each stage returns, and closes its output, when its input closes.

I wrote the first version of these stages with for v := range in, which reads more directly, and a test that cancels a pipeline whose source never produces anything failed: the stages were parked in a plain receive and never saw the cancellation. range is safe only when the upstream is guaranteed to either send or close, which is true between two stages on the same context and false at the edge of the pipeline, where the source is a socket, a queue or any channel you did not write. Guarding both directions costs two small helpers and removes the question.

Composing a pipeline

Stages compose by feeding one's output to the next. The pipeline reads as a sentence:

package main

import (
	"context"
	"fmt"
	"strings"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	lines := Generate(ctx,
		"ERROR disk full",
		"INFO started",
		"ERROR timeout",
		"WARN slow query",
		"ERROR bad request",
		"INFO stopped",
	)
	errors := Filter(ctx, lines, func(l string) bool { return strings.HasPrefix(l, "ERROR") })
	messages := Map(ctx, errors, func(l string) string { return strings.TrimPrefix(l, "ERROR ") })
	shouted := Map(ctx, messages, strings.ToUpper)

	fmt.Println(Collect(ctx, shouted))
}

[DISK FULL TIMEOUT BAD REQUEST]

Six log lines go in; three error messages come out, upper-cased. Each stage is a goroutine, and the lines pass through all four at once, as a stream. The output is deterministic because every stage here has a single worker, which keeps the items in order. The defer cancel() at the top is the safety net: if anything returns early, cancelling the context releases every stage.

Cancellation passes through

The reason every send and receive is a select is that a pipeline is only as stoppable as its least stoppable stage. Suppose the consumer takes two results from a stream of a million and walks away:

package main

import (
	"context"
	"fmt"
	"runtime"
	"time"
)

func Example_cancellation() {
	before := runtime.NumGoroutine()

	ctx, cancel := context.WithCancel(context.Background())

	nums := make([]int, 1_000_000)
	for i := range nums {
		nums[i] = i
	}
	squares := Map(ctx, Generate(ctx, nums...), func(n int) int { return n * n })

	first, second := <-squares, <-squares
	fmt.Println("took", first, second, "then stopped reading")

	cancel() // every stage's select sees Done and returns
	time.Sleep(50 * time.Millisecond)
	fmt.Println("no goroutines left behind:", runtime.NumGoroutine() <= before)

	// Output:
	// took 0 1 then stopped reading
	// no goroutines left behind: true
}

The consumer reads two values and cancels. Both stages are blocked trying to send (Generate to Map, and Map to us), their selects pick the Done case, and both goroutines return, so the count is back to where it started (the example prints a boolean, because other tests in the package may still be winding down goroutines of their own). Remove the ctx.Done() cases and the program leaves two goroutines blocked forever on sends nobody will receive, which is the leak pattern from goroutine leaks. Cancellation through channels is manual in Go; the discipline is in writing every send and receive as an exit.

Backpressure for free

With unbuffered channels, a slow stage slows everything before it. The producer cannot generate faster than the slowest stage consumes, so memory stays flat however long the stream is. You can see it by making one stage slow and watching how far the source gets:

package main

import (
	"context"
	"fmt"
	"sync/atomic"
	"time"
)

func Example_backpressure() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var produced atomic.Int64
	counted := Map(ctx, Generate(ctx, make([]int, 1000)...), func(n int) int {
		produced.Add(1)
		return n
	})
	slow := Map(ctx, counted, func(n int) int {
		time.Sleep(10 * time.Millisecond) // a slow stage
		return n
	})

	deadline := time.After(105 * time.Millisecond)
	consumed := 0
loop:
	for {
		select {
		case _, ok := <-slow:
			if !ok {
				break loop
			}
			consumed++
		case <-deadline:
			break loop
		}
	}
	fmt.Println("consumed about ten:", consumed >= 8 && consumed <= 12)
	fmt.Println("fast stage stayed within a few items of the slow one:", produced.Load() < int64(consumed)+5)

	// Output:
	// consumed about ten: true
	// fast stage stayed within a few items of the slow one: true
}

After about 105 ms the slow stage has finished roughly ten items, and the fast stage upstream has produced only a few more than that, not all thousand, because it blocks as soon as the slow stage stops accepting. The example checks both facts and prints them as booleans, since the exact counts vary a little between runs. This is the property you lose with large buffers or unbounded queues between stages, where the fast stage runs ahead and the difference sits in memory. If you do add buffers, make them small and deliberate (buffered or unbuffered).

The slowest stage sets the pace

The throughput of a pipeline equals the throughput of its slowest stage; the others spend their time waiting. That makes measurement the first step in tuning: find the slow stage before adding anything. A cheap way is to wrap each stage with a counter and a timer, or to look at which stage's goroutines are blocked on sends (a stage blocked sending is waiting for the one after it; a stage blocked receiving is waiting for the one before it) in a goroutine dump. Then fix the bottleneck in the right order:

1. Make the slow stage's work cheaper.
2. Fan it out to several workers, keeping the pipeline otherwise as is (fan-out and fan-in). The stage function then becomes a ParallelMap, and the order caveats apply.
3. Batch items so per-item overhead is amortised.

Do not add buffers to the fast stages in the hope of speeding up the slow one; that only moves where the items wait.

Testing a stage in isolation

Because a stage is a function from channel to channel, testing it needs no pipeline: feed it from a slice, collect its output, and compare. Test the cancel behavior at the same time, with a stage given a source that never produces anything, inside a synctest bubble so that any goroutine left blocked fails the test by itself (detecting leaks):

package main

import (
	"context"
	"slices"
	"testing"
	"testing/synctest"
)

func TestFilterAndMap(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		ctx := context.Background()
		in := Generate(ctx, 1, 2, 3, 4, 5, 6)
		out := Map(ctx, Filter(ctx, in, func(n int) bool { return n%2 == 0 }), func(n int) int { return n * 10 })

		got := Collect(ctx, out)
		if want := []int{20, 40, 60}; !slices.Equal(got, want) {
			t.Errorf("got %v, want %v", got, want)
		}
	})
}

func TestStagesStopOnCancelEvenWithAStalledSource(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		ctx, cancel := context.WithCancel(context.Background())
		stalled := make(chan int) // a source that never sends and never closes
		out := Map(ctx, Filter(ctx, stalled, func(int) bool { return true }), func(n int) int { return n })

		cancel()
		_ = out
		// Both stages are waiting on a channel nobody feeds. After cancel they
		// must have returned, or synctest.Test reports a deadlock.
	})
}

The first test checks the composition. The second hands the stages a channel that never produces, cancels, and relies on the bubble to detect leftover goroutines. It is the test that failed against my first version of the stages, with a deadlock panic listing both goroutines as blocked in chan receive, and it passes now that receives are guarded. That is the value of testing cancellation as a behavior: the happy path passed the whole time.

A realistic pipeline: counting words

The stages above are building blocks. A pipeline closer to real work reads lines, splits them into words, normalises them, and counts. Splitting changes the item count (one line becomes many words), so it needs a stage of its own, which is also the pattern for any one-to-many step:

package main

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"unicode"
)

// FlatMap turns each input into zero or more outputs.
func FlatMap[T, U any](ctx context.Context, in <-chan T, f func(T) []U) <-chan U {
	out := make(chan U)
	go func() {
		defer close(out)
		for {
			v, ok := recv(ctx, in)
			if !ok {
				return
			}
			for _, u := range f(v) {
				if !send(ctx, out, u) {
					return
				}
			}
		}
	}()
	return out
}

func Example_wordCount() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	lines := Generate(ctx,
		"Go makes concurrency easy.",
		"Concurrency is not parallelism!",
		"Go channels, go routines.",
	)
	words := FlatMap(ctx, lines, func(l string) []string {
		return strings.FieldsFunc(l, func(r rune) bool { return !unicode.IsLetter(r) })
	})
	lower := Map(ctx, words, strings.ToLower)

	counts := map[string]int{}
	for w := range lower { // the sink: the last stage owns the map, so no lock is needed
		counts[w]++
	}

	var keys []string
	for k := range counts {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		if counts[k] > 1 {
			fmt.Println(k, counts[k])
		}
	}

	// Output:
	// concurrency 2
	// go 3
}

FlatMap is Map where the function returns a slice, and each element is sent separately. The sink at the end is a plain for range loop that owns the counts map, so the map needs no lock: only one goroutine, the one running the example, ever touches it. That is the ownership idea behind channels at work. The printed output lists the words that appear more than once: go three times and concurrency twice.

Measuring each stage

To find the bottleneck, count how many items pass through each stage and how long the stage spends waiting for input versus working. A small generic wrapper gives you that for any stage without editing it:

package main

import (
	"context"
	"fmt"
	"sync/atomic"
	"time"
)

// Tap passes values through unchanged and counts them.
func Tap[T any](ctx context.Context, in <-chan T, n *atomic.Int64) <-chan T {
	return Map(ctx, in, func(v T) T {
		n.Add(1)
		return v
	})
}

func Example_measuredStages() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var afterGenerate, afterFilter atomic.Int64

	src := Tap(ctx, Generate(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10), &afterGenerate)
	even := Filter(ctx, src, func(n int) bool { return n%2 == 0 })
	counted := Tap(ctx, even, &afterFilter)
	slow := Map(ctx, counted, func(n int) int {
		time.Sleep(time.Millisecond)
		return n
	})

	Collect(ctx, slow)
	fmt.Println("generated:", afterGenerate.Load(), "| survived the filter:", afterFilter.Load())

	// Output:
	// generated: 10 | survived the filter: 5
}

=== RUN   TestFilterAndMap
--- PASS: TestFilterAndMap (0.00s)
=== RUN   TestStagesStopOnCancelEvenWithAStalledSource
--- PASS: TestStagesStopOnCancelEvenWithAStalledSource (0.00s)
=== RUN   Example_backpressure
--- PASS: Example_backpressure (0.11s)
=== RUN   Example_cancellation
--- PASS: Example_cancellation (0.07s)
=== RUN   Example_measuredStages
--- PASS: Example_measuredStages (0.01s)
=== RUN   Example_wordCount
--- PASS: Example_wordCount (0.00s)
PASS
ok  	example.test/lab	1.416s

Counting at the boundaries between stages shows where items are dropped (ten generated, five survive the filter) and, sampled over time, where they pile up. In a long-running service you would export the counters as metrics and watch the rate at each point: a stage whose output rate is far below its input rate is the bottleneck, and one whose counter does not move is stuck. Combine that with goroutine dumps, where a stage parked on a send is waiting for the one after it (goroutine dumps).

Pipeline, errgroup or loop?

A pipeline has overhead: channels, goroutines, context checks. It earns that overhead in particular cases. Use one when the data is a stream or too large to hold, when the stages have different bottlenecks (one CPU-bound, one waiting on I/O) so overlapping them is a real gain, when you need backpressure, or when the steps are independently reusable. Use a plain loop for a small slice, or for work where each item goes through all the steps with no benefit from overlap: the loop is simpler, easier to debug and faster for cheap steps (fan-out costs measured the overhead). Use errgroup when you have a batch of independent jobs, want a bound on parallelism and need the first error (errgroup). Many programs mix them: a pipeline whose slow stage is a fan-out, run under an errgroup for lifecycle and errors.

Designing stages that last

A handful of rules keep stages reusable and the pipeline easy to read.

One job per stage. A stage that parses, validates and enriches is three stages with a single name. Small stages are easy to test, to reorder and to replace, and they make the bottleneck visible, because the stage with the slow job is a stage of its own.

Stages do not know their neighbours. A stage takes a channel of T and returns a channel of U, and nothing else about the pipeline. If a stage needs to know what comes before or after it, a boundary is in the wrong place.

Values are passed, then forgotten. Once a stage sends an item, it must not touch it again; the receiver owns it. If items are pointers or slices, this is a convention you enforce by discipline, and the race detector checks it. Passing small values or copying is the safe default.

The caller owns the context. Stages take a context and never create a root one. The code that assembles the pipeline decides when it ends, through a deadline, a cancel or a signal.

Cancel is for stopping early, closing is for finishing. When the stream ends normally, the source closes its output and the close ripples through. When something goes wrong, or the consumer has seen enough, cancel the context. Mixing them up is how pipelines hang: cancelling when you meant to wait loses items in flight, and waiting when you meant to cancel keeps a million items going to a consumer that has left.

Order is a property you choose. Single-worker stages preserve order. Fan-out stages do not, unless you carry an index, so say in the stage's comment whether it preserves order.

From a pipeline to a service

A command line tool runs a pipeline once and exits. A service runs one for as long as it lives, and that changes two things. The pipeline's context becomes the service's lifetime, so shutdown means cancelling it and waiting for the stages to drain, which needs a WaitGroup or an errgroup around the stage goroutines, something the compact stage functions above leave out for clarity. And the source is no longer a slice but an unbounded stream, so the memory behavior comes entirely from backpressure, and the metrics on each boundary become the way you notice that a downstream dependency has slowed. Plan for both from the start: give the assembled pipeline a Run(ctx) error method that returns when the stages have all exited, and export a counter per stage.

Where pipelines go wrong

A stage that forgets to close its output, so the next stage's range never ends. Closing the input from the wrong side. A send without a select, the leak above. Shared mutable state between stages; items should be passed and then forgotten by the sender. Unbounded fan-out inside a stage. Errors with nowhere to go: stages as written have no error path, and the usual ways to add one are an item type that carries (value, error) or an errgroup for the stage goroutines (pipeline errors). Using a pipeline for a small slice, where a loop would be simpler and faster. Pipelines pay for themselves on streams, on stages with different bottlenecks, and when stopping early matters.

The pattern is described in the Go blog's Pipelines and cancellation, the source of the stage conventions above, and the pieces are documented in context and testing/synctest.