Skip to content

A pipeline of stages

Level: 201 · you can start a goroutine and range over a channel, and want to chain several

One line: A pipeline is a chain of goroutines, each the only sender on the channel it returns and the one that closes it. When the source runs out, the closes cascade down to the reader. When the reader stops early, a canceled context lets the stages upstream stop too; without one, they stay blocked on their sends for as long as the program runs.

Sameer Ajmani's Go blog post Go Concurrency Patterns: Pipelines and cancellation ↗ gives the informal definition this page follows: "a series of stages connected by channels", where each stage receives from upstream, does its work, and sends downstream. The first stage is the source and the last is the sink. Here the source is count, the middle stage is square, and the sink is main.

The output

Verified output of a_pipeline_of_stages_go.go — regenerated by tools/run_examples.py, never hand-typed.

1. count(1..5) -> square -> main reads to the end
   main received:  [1 4 9 16 25]
   stages.Wait()   returned: count and square have ended
2. count(1..) -> square -> main reads 3, then cancels
   main received:  [1 4 9]
   stages.Wait()   returned: count and square have ended
   squares open:   false
a_pipeline_of_stages_go.go

a_pipeline_of_stages_go.go in full — pasted here by tools/run_examples.py from the file CI runs.

// A pipeline of stages: count -> square -> main. Each stage is a goroutine that
// owns its output channel: it is the only sender on it, and it closes it on its
// way out. A context tells the stages to stop early, so that none is left
// blocked on a send that nobody will ever receive.
//
//  go build a_pipeline_of_stages_go.go && ./a_pipeline_of_stages_go
package main

import (
    "context"
    "fmt"
    "math"
    "sync"
)

// count is the first stage, the source: it sends 1, 2, ... last, then closes
// its channel. It stops early if ctx is canceled.
func count(ctx context.Context, stages *sync.WaitGroup, last int) <-chan int {
    out := make(chan int)
    stages.Go(func() {
        defer close(out) // the owner closes, on every way out
        for n := 1; n <= last; n++ {
            select {
            case out <- n:
            case <-ctx.Done():
                return
            }
        }
    })
    return out
}

// square is a middle stage: it receives until its input is closed, sends the
// square of each value, and closes its own channel when it returns.
func square(ctx context.Context, stages *sync.WaitGroup, in <-chan int) <-chan int {
    out := make(chan int)
    stages.Go(func() {
        defer close(out)
        for n := range in {
            select {
            case out <- n * n:
            case <-ctx.Done():
                return
            }
        }
    })
    return out
}

func main() {
    var stages sync.WaitGroup // counts the stage goroutines, to prove they returned

    fmt.Println("1. count(1..5) -> square -> main reads to the end")
    ctx := context.Background()
    squares := square(ctx, &stages, count(ctx, &stages, 5))
    var got []int
    for s := range squares { // ends when square closes its channel
        got = append(got, s)
    }
    fmt.Println("   main received: ", got)
    stages.Wait()
    fmt.Println("   stages.Wait()   returned: count and square have ended")

    fmt.Println("2. count(1..) -> square -> main reads 3, then cancels")
    ctx, cancel := context.WithCancel(context.Background())
    squares = square(ctx, &stages, count(ctx, &stages, math.MaxInt))
    got = nil
    for range 3 {
        got = append(got, <-squares)
    }
    fmt.Println("   main received: ", got)
    cancel()
    stages.Wait()
    fmt.Println("   stages.Wait()   returned: count and square have ended")
    _, open := <-squares
    fmt.Println("   squares open:  ", open)
}

Reading the output

1. The closes cascade. count sends 1 to 5 and returns, and its deferred close(out) ends the for n := range in loop in square. square returns in turn, and its own deferred close ends main's loop. No stage was told to stop: each one finished because its input did (Closing a channel ends a range). The stages.Wait() line shows that both goroutines ended, because each one was started with stages.Go, which counts a goroutine in when it starts and out when it returns (A WaitGroup counts goroutines).

The stage that makes a channel is the one that closes it because it is the only goroutine that knows no more values are coming, and a send on a closed channel panics (spec: Close ↗). Returning <-chan int writes that ownership into the type: the same section of the spec makes it an error to close a receive-only channel, so a stage cannot close a channel it was handed. The blog's first guideline for pipelines says the same thing: a stage closes its outbound channels once all its sends are done.

2. Cancel reaches upstream. This time count would count to math.MaxInt. main reads three squares and calls cancel(). From then on nothing receives from squares, so a stage that is sending has to have another way out. Each send sits in a select with a second case, <-ctx.Done(). Done returns a channel that is closed when the context is canceled, and a receive from a closed channel proceeds at once (spec: Receive operator ↗), so every stage waiting in that select returns. stages.Wait() returned again, and squares open: false shows that square ran its deferred close on the way out.

ctx.Done() is the blog's done channel under another name. In the blog, main makes a chan struct{} and defers close(done), so that every return path from main signals every stage at once. Closing a channel is a broadcast: it releases every goroutine receiving from it, however many there are. A context carries that same channel and adds deadlines and causes to it (Cancel reaches every goroutine).

Without the cancel

This driver builds the same pipeline with the selects and the context taken out. main reads three squares, then waits for the stages:

Verified output of a_pipeline_of_stages_no_cancel_sh.sh — regenerated by tools/run_examples.py, never hand-typed.

$ go build no_cancel.go
$ ./no_cancel
main received: 1
main received: 4
main received: 9
main: done reading, waiting for the stages to return
fatal error: all goroutines are asleep - deadlock!
exit status 2

They never return. Each stage is blocked on a send that nothing will receive. main is blocked too, waiting for them, so no goroutine can run, and the runtime ends the program with exit status 2 (All goroutines are asleep). That is the lucky case. If main were busy with something else, such as serving requests, nothing would notice, and the two goroutines would stay blocked for the life of the process. The blog calls this a resource leak, and points out that the garbage collector does not reclaim a goroutine: it has to return on its own (A leaked goroutine never ends).

The blog also considers a different fix, a buffer with room for the values nobody reads, and rejects it as fragile. It only works as long as someone knows how many values will be sent and how many the reader will skip.

a_pipeline_of_stages_no_cancel_sh.sh

a_pipeline_of_stages_no_cancel_sh.sh in full — pasted here by tools/run_examples.py from the file CI runs.

#!/usr/bin/env bash
# The pipeline from a_pipeline_of_stages_go.go with the cancellation taken out:
# the stages have no context to watch. main reads three squares and then waits
# for the stages to return, which they never do: each is blocked on a send that
# nothing will receive. Once every goroutine is blocked, Go's runtime ends the
# program. The goroutine dump after its first line names goroutine numbers, so
# this script keeps the first line and the exit status.
set -u

dir=$(mktemp -d)
trap 'rm -rf "$dir"' EXIT
export GOTOOLCHAIN=local

cat >"$dir/no_cancel.go" <<'GO'
package main

import (
    "fmt"
    "math"
    "sync"
)

func count(stages *sync.WaitGroup, last int) <-chan int {
    out := make(chan int)
    stages.Go(func() {
        defer close(out)
        for n := 1; n <= last; n++ {
            out <- n // no select, no way to stop
        }
    })
    return out
}

func square(stages *sync.WaitGroup, in <-chan int) <-chan int {
    out := make(chan int)
    stages.Go(func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    })
    return out
}

func main() {
    var stages sync.WaitGroup
    squares := square(&stages, count(&stages, math.MaxInt))
    for range 3 {
        fmt.Println("main received:", <-squares)
    }
    fmt.Println("main: done reading, waiting for the stages to return")
    stages.Wait()
    fmt.Println("main: the stages returned") // never printed
}
GO

say() { printf '$ %s\n' "$*"; }

say 'go build no_cancel.go'
(cd "$dir" && go build -trimpath no_cancel.go) || exit 1

say './no_cancel'
(cd "$dir" && ./no_cancel) 2>"$dir/stderr"
status=$?
grep '^fatal error' "$dir/stderr"
echo "exit status $status"

What to do

  • Give every stage the same shape: a function that takes a context and an input channel, makes its output channel, starts one goroutine that is the only sender on it, and returns it as <-chan T.
  • Make defer close(out) the first line of that goroutine, so that every way out closes the channel.
  • Put every send in a select with case <-ctx.Done(): return. Receiving with range is safe, because upstream will close. Sending is where a stage gets stuck.
  • Whoever stops reading early cancels, with defer cancel() right after context.WithCancel, so that every return path reaches the stages.
  • Do not size a buffer to hide a stuck stage.

In other languages

  • The Rust library's Channels ↗: a Rust channel ends when every Sender has been dropped, not when someone remembers to call close. When the Receiver is dropped, send fails and hands the value back, so an upstream Rust stage learns that the reader has gone on its next send. A Go stage in the same position blocks forever unless it also watches a cancel channel.
  • The Linux library's head closes the pipe early ↗: a shell pipeline is the same shape built from processes. When head exits, the writer's next write kills it with SIGPIPE. That is cancellation delivered by the operating system, and nothing like it reaches a goroutine.
  • The Concurrency library's chapter on passing messages in six languages, 05 Message passing, is planned.

Sources