Fan-Out and Fan-In

The previous lesson measured a setup where a fixed number of workers listened to the same job channel: a single source, many consumers, and a termination signal that reached all of them through a single close. But work does not always arrive at a rate a single worker could keep up with on its own. Sometimes the values being produced themselves must be distributed across more than one processor — this is fan-out — and the results of that distributed work must be merged back into a single stream — this is fan-in. The previous lesson’s “who closes the channel” question applies here too, but the answer sits in a different place: in the worker pool, the closing side was the one distributing the work, while here, more than one side writes to the channel that needs closing — and this, for the first time, turns “who closes it” into a real problem.

Fan-out and fan-in cover two separate directions. Fan-out is a production source (a queue, a channel) opening out to more than one processor — nearly the same mechanism as the previous lesson’s job channel, because a fixed number of goroutines reading from a shared channel is already a fan-out. Fan-in works in the opposite direction: more than one source merging into a single channel. When the two directions are set up together, a new question arises — in fan-out the closing side is clear (the single side managing the source), but in fan-in, however many sides write to the channel that needs closing, that many sides carry the risk of the closing responsibility being split among them.

A Source Fanned Out to Many, Results Fanned Into One Channel

The measurement below fans six values out to three producer goroutines; each producer squares the values falling to its share and writes them to a single fanin channel. Who will close the fan-in channel is already resolved here: a separate goroutine waits for a wait group (sync.WaitGroup) to report that all producers are finished, and only then closes fanin. Sole ownership is established here not in one producer, but outside all the producers.

// fanoutfanin.go — a source fans out to three producers, results fan in to one channel; the wait group closes the fan-in channel
package main

import (
	"fmt"
	"sort"
	"sync"
)

func producer(input <-chan int, fanin chan<- int, group *sync.WaitGroup) {
	defer group.Done()
	for value := range input {
		fanin <- value * value
	}
}

func main() {
	const jobCount = 6
	const producerCount = 3

	input := make(chan int, jobCount)
	fanin := make(chan int, jobCount)
	var group sync.WaitGroup

	for id := 1; id <= producerCount; id++ {
		group.Add(1)
		go producer(input, fanin, &group)
	}

	for i := 1; i <= jobCount; i++ {
		input <- i
	}
	close(input)

	// Fan-in: closes the fanin channel ITSELF once all producers finish;
	// no producer closes it on its own.
	go func() {
		group.Wait()
		close(fanin)
	}()

	fmt.Println("-- source fanned out to 3 producers, results fanned in to one channel --")
	results := make([]int, 0, jobCount)
	for s := range fanin {
		results = append(results, s)
	}
	sort.Ints(results)

	total := 0
	for _, s := range results {
		total += s
	}
	fmt.Printf("fanned-out job count: %d\n", jobCount)
	fmt.Printf("collected result count: %d, total: %d\n", len(results), total)
	fmt.Println("fan-in's reason for stopping: what it waited for arrived (the fan-in channel closed)")
}

-- source fanned out to 3 producers, results fanned in to one channel --
fanned-out job count: 6
collected result count: 6, total: 91
fan-in's reason for stopping: what it waited for arrived (the fan-in channel closed)

CP6. The input channel plays the same role as the previous lesson’s jobs — a queue shared among three producers. What is different is the fanin channel: all three producers write to it at the same time, not a single worker. The result count reaching six and the total reaching ninety-one (the squares 1+4+9+16+25+36) is independent of which producer processed which value; this is exactly why sorting (sort.Ints) is needed — the order in which the three producers write to the fan-in channel can vary from run to run, but the set itself does not change.

In this measurement, the producer function has almost the same body as the previous lesson’s worker: both read from an input channel, both write to an output channel. The difference is not in the function’s own code, it is in the context it is called in — worker was set up so a fixed number of consumers would share the same job, while producer runs the same function this time in the context of more than one writer writing to the same destination. The pattern’s name changes (from worker pool to fan-out), but the underlying channel mechanism stays the same; only where the closing responsibility sits changes.

CP7. The closing goroutine here is a third party: neither one of the producers, nor the fan-in side itself. group.Wait() does not return until all three of the three producers have reached group.Done(), and only then does close(fanin) run — this guarantees that the last write to the fan-in channel has already happened. Both the input and fanin channels are opened buffered with jobCount capacity, for the same reason as the previous lesson: so the producers can take their share and write without being blocked. This third-party approach raises the next section’s question: what if this separate goroutine did not exist at all, and each producer tried to close the channel itself once its own work was done?

Who Closes the Fan-In Channel

Closing a channel in Go is a one-time operation: closing the same channel twice panics. If all three producers share the same channel and each one says “I am done, let me close the channel” once it finishes its own work, the first one to close succeeds, but every closing attempt after that panics. The measurement below sets exactly this up: each producer first sends its own value, then (after a wait point makes sure all producers have finished sending) tries close(fanin) on its own.

// naive.go — if every producer tries to close on its own, all but the first panic
package main

import (
	"fmt"
	"sync"
)

func naiveProducer(id int, fanin chan int, sent *sync.WaitGroup, result chan<- bool, finished *sync.WaitGroup) {
	defer finished.Done()
	fanin <- id
	sent.Done()
	sent.Wait() // wait until every producer has sent its own value
	panicked := false
	func() {
		defer func() {
			if r := recover(); r != nil {
				panicked = true
			}
		}()
		close(fanin)
	}()
	result <- panicked
}

func main() {
	const producerCount = 3
	fanin := make(chan int, producerCount)
	result := make(chan bool, producerCount)
	var sent sync.WaitGroup
	var finished sync.WaitGroup
	sent.Add(producerCount)
	finished.Add(producerCount)

	for id := 1; id <= producerCount; id++ {
		go naiveProducer(id, fanin, &sent, result, &finished)
	}
	finished.Wait()
	close(result)

	succeeded, panicking := 0, 0
	for panicked := range result {
		if panicked {
			panicking++
		} else {
			succeeded++
		}
	}
	fmt.Println("-- each producer sends its own value first, then tries to close --")
	fmt.Printf("producers that closed successfully: %d / %d\n", succeeded, producerCount)
	fmt.Printf("producers that panicked: %d / %d\n", panicking, producerCount)
}

-- each producer sends its own value first, then tries to close --
producers that closed successfully: 1 / 3
producers that panicked: 2 / 3

CP8. The result is single-answered in a not-so-surprising way: on every run, exactly one producer closes successfully, and the remaining two panic. Which producer succeeds can change from run to run (all three start trying to close at the same time), but how many succeed never changes — Go’s runtime protects a channel’s closing with an internal lock, so even if more than one goroutine calls close at the same moment, only one of them really closes the channel, and the rest find it already closed and panic. This is why the lesson prints not which one succeeded, but how many did.

This internal lock means that close itself is an atomic operation: whether a channel is “closed or not” always has a single, sequential history, even when more than one goroutine tries to ask and change it at the same instant. Even if three goroutines really call close(fanin) at the same moment, the runtime processes them in some invisible order — the first one processed succeeds, every one after it finds an already-closed channel and panics. This behavior is very different from writing to a shared variable without protection: if two goroutines write to a shared counter at the same time, the result is an undefined race condition (not measured here, only established as a concept in M01/K05 Operating System Concepts), but closing a channel is an operation the language itself guarantees, not open to a race — whoever wins, exactly one winner comes out.

CP9. The second synchronization point (sent) is a deliberate design decision here: the measurement makes sure every producer sends its own value first, then tries to close. Without this separation, one producer could close the channel before another had sent its own value, and that producer’s send (fanin <- id) would end up not before the close line, but as a panic — which would confuse which line panicked for which reason. sent.Wait(), just like group.Wait() seen in the previous lesson, separates the two phases (sending, closing) without leaving the order to the scheduler.

This separation itself hides one observation: a panic about a closed channel can actually be triggered in two separate forms — closing a closed channel again (close of closed channel) and sending to a closed channel (send on closed channel). Without the sent barrier above, a producer might close the channel before another had sent its own value (fanin <- id), and since that send line sits outside its own recover block, the panic would go uncaught and the program would crash — the lesson would stop before ever reaching kod_dogrula.py. The sent.Wait() barrier prevents exactly this: no producer tries to close until it is certain all sends have finished, which is why the observed panic type is always the same (closing an already-closed channel), never the other one.

This measurement shows why the correct arrangement in the previous section is correct: the responsibility for closing the fan-in channel does not belong to any of the producers, because none of them knows when the others will finish. The responsibility can only be given to a third party that knows when all the producers have finished — a separate goroutine waiting on a wait group.

An Unclosed Fan-In Channel

The previous two setups showed two separate outcomes of splitting the closing responsibility between zero and more than one side: in the first setup, responsibility belonged to exactly one side, and everything ended silently; in the second setup, responsibility (mistakenly) belonged to three sides, and two panics resulted. A third possibility remains: what happens if responsibility is given to no one at all? If the third-party goroutine (the separate goroutine above, waiting on the wait group and closing the fan-in channel) is never set up at all, the fan-in channel never closes. The measurement below sets this up in its plainest form: a fan-in goroutine ranges over the fanin channel, but there is no one at all to close the channel.

// leak.go — if the fan-in channel is never closed, the fan-in side never stops
package main

import (
	"fmt"
	"runtime"
)

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

	fanin := make(chan int)
	go func() {
		// Since no one closes the fan-in channel, this range will never end.
		for range fanin {
		}
	}()

	fmt.Println("-- fan-in channel never closed --")
	fmt.Printf("did a non-stopping fan-in side remain: %v\n", runtime.NumGoroutine() > base)
}

-- fan-in channel never closed --
did a non-stopping fan-in side remain: true

CP10. This is the fan-in side’s mirror image of the previous lesson’s unclosed job channel measurement: there, the workers waited forever; here, a single fan-in side waits forever. The difference is where the leak sits: in the worker pool, the leaking side was many consumers; here, the leaking side is a single fan-in goroutine — but in both, the cause is the same, ranging over a channel that no one closes. Since the fan-in side is itself a goroutine too, its leak is measured not by a different tool than the worker leak, but with the exact same runtime.NumGoroutine() comparison.

What these three setups have in common is that none of them solves the problem at its source: even the correct setup assumes that all producers will eventually finish. If one of the producers never finishes its own work (say, it enters an infinite loop), group.Wait() never returns either, and the fan-in channel again never closes — this setup gives the closing responsibility to the right side, but that side itself still depends on a condition. Bounding the possibility of a producer never finishing its own work (with a fixed duration or a signal coming from outside) is not this lesson’s subject; the next lesson, measuring cancellation propagation with the context package, takes up that question.

The three measurements together complete this topic’s thesis: a correctly built fan-in (responsibility gathered in one side) stopped deterministically and silently; an incorrectly built fan-in (everyone trying to close on their own) stopped deterministically but noisily — with panics; and a never-built fan-in (no one closing) never stopped at all. All three are different answers to the same question: the responsibility for closing the fan-in channel must belong to exactly one side, neither fewer nor more. This principle is a generalization of the same principle behind the worker pool’s single closing side (the side distributing the work): no matter how many sides write to a channel, closing always stays a separate, singular responsibility.

Summary

• When a source is fanned out to a fixed number of producers (fan-out), the results merge back into a single fan-in channel (fan-in); the collected result count and its total are two fixed values independent of the producer count.
• The responsibility for closing the fan-in channel belongs to none of the producers — a third party closes it, one that a wait group tells once all producers have finished.
• If every producer tries to close on its own, exactly one succeeds and the rest panic; which producer succeeds can change, but how many succeed never does.
• If the fan-in channel is never closed, the fan-in side waits forever; the leak is again measured only as a boolean.
• The closing responsibility must belong to exactly one side: zero sides leads to a leak, more than one leads to a panic.
• Closing a channel is an atomic operation the language itself guarantees; even if more than one goroutine calls close at the same moment, exactly one succeeds, the rest panic.

Next Step

In this lesson, a source fanned out to many and the results fanned back into a single channel — but the flow was a single step: each producer took the input, squared it, wrote the result. Real processing usually passes through more than one step, and each step has its own speed, its own closing responsibility. The next lesson chains these steps — building a pipeline out of stages, with each stage closing its own output channel on its own. This time the question is not who holds a single closing responsibility: if a stage in the chain stops early, what happens to the stages ahead of it?