Fan-out, fan-in¶
Level: 201 · you have a pipeline with one slow stage, and want several goroutines on it
One line: Several goroutines receiving from one channel split its values between them, so each value goes to exactly one of them. A merge that closes its output only after a WaitGroup has seen every copying goroutine finish joins their results back into one channel. What fan-in cannot give back is the order, so the program sorts before it prints.
The Go blog's Pipelines and cancellation ↗ names both halves. Fan-out is several functions reading from the same channel until it is closed. Fan-in is copying several input channels onto one channel, which is closed once all the inputs are. Here words sends eight words, three digesters share that channel and compute the SHA-256 of each word, standing in for work slow enough to be worth splitting, and merge fans their three channels back into one.
The output¶
Verified output of fan_out_fan_in_go.go — regenerated by tools/run_examples.py, never hand-typed.
words sent: 8
digests received: 8
sorted by word:
channel 69e36568
context ea7792a2
goroutine a0af8076
mutex 96cf2011
pipeline 23bf0d24
select b1a36d25
semaphore 2bbd2a7a
worker 87eba76e
fan_out_fan_in_go.go
fan_out_fan_in_go.go in full — pasted here by tools/run_examples.py from the file CI runs.
// Fan-out, fan-in. Three digesters read words from one channel (fan-out): each
// word goes to exactly one of them. Each digester sends on a channel of its own,
// and merge copies all three onto a single channel, which it closes once every
// input is closed and drained (fan-in). Digests arrive in whatever order the
// digesters finished, so main sorts them before it prints.
//
// go build fan_out_fan_in_go.go && ./fan_out_fan_in_go
package main
import (
"crypto/sha256"
"encoding/hex"
"fmt"
"slices"
"strings"
"sync"
)
type digest struct {
word string
sum string // the first 4 bytes of the word's SHA-256, in hex
}
// words is the source: it sends each word and closes its channel.
func words(list ...string) <-chan string {
out := make(chan string)
go func() {
defer close(out)
for _, w := range list {
out <- w
}
}()
return out
}
// digester is one worker. It owns its output channel and closes it when its
// input is closed.
func digester(in <-chan string) <-chan digest {
out := make(chan digest)
go func() {
defer close(out)
for w := range in {
sum := sha256.Sum256([]byte(w))
out <- digest{word: w, sum: hex.EncodeToString(sum[:4])}
}
}()
return out
}
// merge starts one copying goroutine per input, and one more that closes the
// output after every copy has finished, so that no send can follow the close.
func merge(inputs ...<-chan digest) <-chan digest {
out := make(chan digest)
var copies sync.WaitGroup
for _, in := range inputs {
copies.Go(func() {
for d := range in {
out <- d
}
})
}
go func() {
copies.Wait()
close(out)
}()
return out
}
func main() {
list := []string{"goroutine", "channel", "select", "mutex", "context", "pipeline", "worker", "semaphore"}
in := words(list...)
// Fan-out: three digesters share one input channel.
a, b, c := digester(in), digester(in), digester(in)
// Fan-in: one channel, closed when all three are done.
var got []digest
for d := range merge(a, b, c) {
got = append(got, d)
}
fmt.Printf("words sent: %d\n", len(list))
fmt.Printf("digests received: %d\n", len(got))
// Arrival order is the scheduler's; the report's order is main's.
slices.SortFunc(got, func(x, y digest) int { return strings.Compare(x.word, y.word) })
fmt.Println("sorted by word:")
for _, d := range got {
fmt.Printf(" %-10s %s\n", d.word, d.sum)
}
}
Reading the output¶
Eight words sent, eight digests received, each word once. Three goroutines range over the same channel, and each value sent is received by only one of them: the sorted list has every word exactly once, with none missing and none repeated. Which digester took which word changes between runs, and the program never prints it.
How merge knows when to close. Each digester closes its own channel when in is closed, as every stage in A pipeline of stages does. But the out channel in merge has three senders, the three copying goroutines, and none of them can close it: a close while another copy is still sending would make that send panic (spec: Close ↗). So merge counts the copies in a WaitGroup and starts one more goroutine, which waits for all of them and then closes out (A WaitGroup counts goroutines). The wait needs a goroutine of its own because merge has to return out before anyone can receive from it. If merge called copies.Wait() itself, it would never return: the copies would be blocked sending on a channel that nobody can receive from until merge returns it. The blog's merge calls wg.Add before it starts the copies and wg.Done at the end of each; WaitGroup.Go, added in Go 1.25, does both.
Why sort. Digests come out of merge in the order the copies delivered them, which depends on which digester got which word and when each one ran. Here is the same program without the sort, run 300 times:
input order: goroutine channel select mutex context pipeline worker semaphore
runs: 300
distinct arrival orders: 183
runs in input order: 11
most common:
11 goroutine select mutex context pipeline worker semaphore channel
11 goroutine channel select mutex context pipeline worker semaphore
9 goroutine mutex context select pipeline worker semaphore channel
8 select mutex context pipeline worker semaphore goroutine channel
8 select mutex context pipeline worker semaphore channel goroutine
That is 183 different orders from 300 runs of one program, and no single order came up more than 11 times. The input order came up 11 times as well, so a check that expected the words in the order they were sent would pass about one run in 27. The answer key holds the sorted list, which is the same on every run. To count on your own machine, run bash demo/tally.sh from this folder. It builds demo/arrival_order.go, this program without its sort. Your numbers will not match these, and that is the point.
A sort needs a key inside each result, here the word. A worker pool restores the order with an index instead, which works for any job.
What to do¶
- Fan out by starting several goroutines on the same input channel. The channel does the distributing.
- Fan in with a merge, or let the workers share one output channel and close it after a
WaitGroup, as A worker pool does. - Close a channel that has several senders from one goroutine that waits for all of them.
- Sort or index before you print, compare or test. After a fan-in, the arrival order belongs to the scheduler.
- Add cancellation as in a pipeline once a reader can stop early: every send, in the digesters and in the copies inside
merge, then needs aselectwithctx.Done(). The blog's finalmergedoes this with itsdonechannel.
In other languages¶
- The Rust library's Channels ↗: fan-in comes with the channel. Clone the
Senderfor each worker, drop the original, and the oneReceiverends when the last clone is dropped, with no merge goroutines and noWaitGroup. Fan-out is the harder half, because theReceiveris notClone: several threads cannot each hold it the way several goroutines range overin. - The Rust library's Spawning a thread ↗: for a fan-out over data already in hand,
thread::scopestarts threads that borrow it, and joining their handles in a fixed order gives a deterministic report with no sort at all. - The Concurrency library's chapters that will compare this across languages, 05 Message passing and 07 Parallelism, are planned.
Sources¶
- Sameer Ajmani, Go Concurrency Patterns: Pipelines and cancellation ↗, the Go blog, 13 March 2014: section "Fan-out, fan-in".
- The Go specification: Close ↗.
sync.WaitGroup.Go↗, added in Go 1.25, andslices.SortFunc↗.- Katherine Cox-Buday, Concurrency in Go (O'Reilly, 2017), chapter 4, "Concurrency Patterns in Go": "Fan-Out, Fan-In".
- Burak Serdar, Effective Concurrency in Go (Packt, 2023), chapter 5, "Worker Pools and Pipelines": "Fan-out/fan-in" and "Fan-in with ordering".
- James Cutajar, Learn Concurrent Programming with Go (Manning, 2024), chapter 9, "Programming with channels": "Fanning in and out".