Fan Out and Fan In Patterns in Go With Cancellation

Two names, one idea. Fan-out is taking a stream of work and spreading it across several goroutines. Fan-in is gathering what they produce back into a single stream. Together they are how Go programs use all their cores on a pipeline stage, or overlap many slow calls, and the pattern is small enough to fit in a screen of code. The details are where the bugs are: closing the output at the right time, stopping without leaking, and getting results in the order you expect.

This guide builds both operations as reusable generic functions, shows how cancellation threads through them, and ends with the tests that prove they work. It builds on pipelines, channels and worker pools. Everything ran on Go 1.25.6.

Fan-out: many readers on one channel

The simplest fan-out is a channel and several goroutines ranging over it. Go's channel semantics do the load balancing: each value is delivered to exactly one receiver, whichever is free next.

package main

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

func main() {
	jobs := make(chan int)
	var mu sync.Mutex
	handled := map[int][]int{} // worker id -> jobs it handled
	var wg sync.WaitGroup

	for w := 1; w <= 3; w++ {
		wg.Go(func() {
			for j := range jobs {
				mu.Lock()
				handled[w] = append(handled[w], j)
				mu.Unlock()
			}
		})
	}

	for j := 1; j <= 9; j++ {
		jobs <- j
	}
	close(jobs)
	wg.Wait()

	total := 0
	for _, js := range handled {
		total += len(js)
	}
	var ids []int
	for id := range handled {
		ids = append(ids, id)
	}
	sort.Ints(ids)
	fmt.Println("jobs handled in total:", total, "| workers that took part:", len(ids) > 0)
}

jobs handled in total: 9 | workers that took part: true

Nine jobs go to whichever of three workers is ready; each is handled exactly once, and which worker gets which job varies from run to run. There is no scheduler of your own: the channel is the distribution mechanism. Slow jobs naturally go to fewer workers than fast ones, because a busy worker is not receiving. That self-balancing is the main reason fan-out is built on a shared channel and not on pre-splitting the input into slices.

Fan-in: merging outputs

Fan-in is the reverse: several channels in, one out. A goroutine per input copies values to the output, and a coordinator closes the output when all of them have finished:

package main

import (
	"context"
	"sync"
)

// Merge combines the inputs into one channel. The output is closed when every
// input is closed, or when ctx ends.
func Merge[T any](ctx context.Context, inputs ...<-chan T) <-chan T {
	out := make(chan T)
	var wg sync.WaitGroup

	for _, in := range inputs {
		wg.Go(func() {
			for v := range in {
				select {
				case out <- v:
				case <-ctx.Done():
					return
				}
			}
		})
	}

	go func() {
		wg.Wait()
		close(out) // the only closer, after every sender has returned
	}()
	return out
}

The structure is the standard fan-in: one goroutine per input, a send that can be abandoned on cancellation, and a closer goroutine that waits for them all. No sender closes out, because each knows only about its own input (who closes a channel). If the context ends, senders stop, the WaitGroup completes, and the output closes, so a consumer ranging over it terminates.

A generic Map stage

With Merge in hand, the fan-out half is a function that starts N workers over one input and merges their outputs:

package main

import "context"

// ParallelMap applies f to every value from in using n workers and returns the
// merged results. Results arrive in completion order, not input order.
func ParallelMap[T, U any](ctx context.Context, in <-chan T, n int, f func(T) U) <-chan U {
	outs := make([]<-chan U, n)
	for i := range n {
		out := make(chan U)
		outs[i] = out
		go func() {
			defer close(out)
			for v := range in {
				select {
				case out <- f(v):
				case <-ctx.Done():
					return
				}
			}
		}()
	}
	return Merge(ctx, outs...)
}

// Source sends the values on a channel and closes it, stopping early if ctx ends.
func Source[T any](ctx context.Context, vals ...T) <-chan T {
	out := make(chan T)
	go func() {
		defer close(out)
		for _, v := range vals {
			select {
			case out <- v:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

ParallelMap is fan-out followed by fan-in: n workers each reading the shared in and writing to their own output, then Merge collapsing them. The worker's function runs before the send (out <- f(v)), and that is where the parallel speedup comes from. Using it:

package main

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

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

	words := Source(ctx, "go", "channels", "select", "context", "pipeline", "worker")
	upper := ParallelMap(ctx, words, 3, strings.ToUpper)

	var got []string
	for w := range upper {
		got = append(got, w)
	}
	sort.Strings(got) // completion order varies, so sort for a stable print
	fmt.Println(got)
}

[CHANNELS CONTEXT GO PIPELINE SELECT WORKER]

The six words go through three parallel workers, and the output order is not guaranteed, so the example sorts before printing. The channel range ends when everything has been processed because the closes cascade: Source closes words, each worker's range ends and it closes its output, Merge sees all inputs closed and closes upper.

Keeping the input order

Because results arrive as workers finish, a slow item can be overtaken by a fast one. If the caller needs the output in input order, carry the position with each item and reassemble. The simplest reassembly collects into a slice by index:

package main

import (
	"fmt"
	"sync"
	"time"
)

type indexed[T any] struct {
	i int
	v T
}

func main() {
	inputs := []int{5, 1, 4, 2, 3}

	jobs := make(chan indexed[int])
	results := make(chan indexed[string])

	var wg sync.WaitGroup
	for range 3 {
		wg.Go(func() {
			for j := range jobs {
				time.Sleep(time.Duration(j.v) * 4 * time.Millisecond) // uneven work
				results <- indexed[string]{j.i, fmt.Sprint("item-", j.v)}
			}
		})
	}

	go func() {
		for i, v := range inputs {
			jobs <- indexed[int]{i, v}
		}
		close(jobs)
	}()
	go func() { wg.Wait(); close(results) }()

	ordered := make([]string, len(inputs))
	for r := range results {
		ordered[r.i] = r.v // place by position
	}
	fmt.Println(ordered)
}

[item-5 item-1 item-4 item-2 item-3]

The work has uneven durations, so items complete in a different order from the input, and the position index puts each result where it belongs: the output is [item-5 item-1 item-4 item-2 item-3], the input order. Reassembly by index needs the whole result set in memory. For streams too large for that, either accept completion order, or use a resequencer that releases items in order as the next expected index arrives and buffers the early ones, and note that a slow head-of-line item then delays everything behind it.

Cancellation and leaks

Every goroutine in these functions has an exit on cancellation: the senders select on ctx.Done(), and the closer exits when its WaitGroup completes. The weak point in a design like this is a consumer that stops reading without cancelling; the senders then block on their sends forever. The rule is that whoever stops consuming must cancel the context. A test proves it, and it can run in a synctest bubble so that a leak fails by itself (detecting leaks):

package main

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

func TestParallelMapProcessesEverything(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()

		out := ParallelMap(ctx, Source(ctx, "a", "b", "c", "d", "e"), 3, strings.ToUpper)

		var got []string
		for v := range out {
			got = append(got, v)
		}
		slices.Sort(got)
		if !slices.Equal(got, []string{"A", "B", "C", "D", "E"}) {
			t.Errorf("got %v", got)
		}
	})
}

func TestStoppingEarlyDoesNotLeak(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		ctx, cancel := context.WithCancel(context.Background())

		vals := make([]int, 1000)
		out := ParallelMap(ctx, Source(ctx, vals...), 4, func(n int) int { return n + 1 })

		<-out // take one result, then walk away
		cancel()
		for range out { // drain so senders can finish
		}
		// If any goroutine were still blocked, synctest.Test would fail here.
	})
}

=== RUN   TestParallelMapProcessesEverything
--- PASS: TestParallelMapProcessesEverything (0.00s)
=== RUN   TestStoppingEarlyDoesNotLeak
--- PASS: TestStoppingEarlyDoesNotLeak (0.00s)
PASS
ok  	example.test/lab	1.264s

The first test checks completeness and that closes cascade (otherwise the range would hang and the bubble would report a deadlock). The second takes one value from a thousand, cancels, drains the rest and returns; if any worker or the merge goroutine had been left blocked, synctest.Test would fail with its deadlock panic. Draining after cancel matters: a sender that is already inside its select may still complete a send before it sees the cancellation, and the drain lets such a send finish. Without a drain the senders see ctx.Done() first and exit by themselves, so the drain is belt and braces here, and necessary when senders do not select.

Does it pay? Measure it

Fan-out is not free: every item crosses at least two channels, and the workers coordinate through the scheduler. Whether the parallelism wins depends on how heavy f is. This benchmark runs the same thousand items through ParallelMap with a heavy function (about 63 microseconds of arithmetic per item) and a trivial one, and compares with a plain loop:

package main

import (
	"context"
	"fmt"
	"testing"
)

var sink int

func heavy(n int) int {
	x := n
	for i := 0; i < 200_000; i++ {
		x += i * i
	}
	return x
}

func light(n int) int { return n + 1 }

func items() []int {
	v := make([]int, 1000)
	for i := range v {
		v[i] = i
	}
	return v
}

func BenchmarkHeavy(b *testing.B) {
	vals := items()
	b.Run("sequential", func(b *testing.B) {
		for i := 0; i < b.N; i++ {
			s := 0
			for _, v := range vals {
				s += heavy(v)
			}
			sink = s
		}
	})
	for _, n := range []int{2, 4, 10} {
		b.Run(fmt.Sprintf("workers=%d", n), func(b *testing.B) {
			for i := 0; i < b.N; i++ {
				ctx, cancel := context.WithCancel(context.Background())
				s := 0
				for r := range ParallelMap(ctx, Source(ctx, vals...), n, heavy) {
					s += r
				}
				cancel()
				sink = s
			}
		})
	}
}

func BenchmarkLight(b *testing.B) {
	vals := items()
	b.Run("sequential", func(b *testing.B) {
		for i := 0; i < b.N; i++ {
			s := 0
			for _, v := range vals {
				s += light(v)
			}
			sink = s
		}
	})
	b.Run("workers=4", func(b *testing.B) {
		for i := 0; i < b.N; i++ {
			ctx, cancel := context.WithCancel(context.Background())
			s := 0
			for r := range ParallelMap(ctx, Source(ctx, vals...), 4, light) {
				s += r
			}
			cancel()
			sink = s
		}
	})
}

goos: darwin
goarch: arm64
pkg: example.test/lab
cpu: Apple M1 Max
BenchmarkHeavy/sequential-10         	       4	  63931292 ns/op
BenchmarkHeavy/sequential-10         	       4	  62372490 ns/op
BenchmarkHeavy/sequential-10         	       4	  62660510 ns/op
BenchmarkHeavy/workers=2-10          	       6	  34949688 ns/op
BenchmarkHeavy/workers=2-10          	       6	  34951229 ns/op
BenchmarkHeavy/workers=2-10          	       6	  34782326 ns/op
BenchmarkHeavy/workers=4-10          	      12	  18335806 ns/op
BenchmarkHeavy/workers=4-10          	      12	  18604878 ns/op
BenchmarkHeavy/workers=4-10          	      12	  18425670 ns/op
BenchmarkHeavy/workers=10-10         	      22	  10212561 ns/op
BenchmarkHeavy/workers=10-10         	      20	  10801069 ns/op
BenchmarkHeavy/workers=10-10         	      20	  10884192 ns/op
BenchmarkLight/sequential-10         	  387218	       619.3 ns/op
BenchmarkLight/sequential-10         	  382395	       611.2 ns/op
BenchmarkLight/sequential-10         	  395685	       617.0 ns/op
BenchmarkLight/workers=4-10          	     164	   1413612 ns/op
BenchmarkLight/workers=4-10          	     164	   1544886 ns/op
BenchmarkLight/workers=4-10          	     159	   1414999 ns/op
PASS
ok  	example.test/lab	5.928s

On an Apple M1 Max, the plain loop over a thousand heavy items took 62.7 ms. Two workers took 34.9 ms (1.8 times faster), four took 18.4 ms (3.4 times) and ten took 10.8 ms (5.8 times), so the speedup tracks the worker count, though not linearly, because of the channel handoffs and the machine's mix of performance and efficiency cores. For the trivial function the picture reverses: the plain loop took 617 ns for the thousand items, and the four-worker pipeline took 1.4 ms, about 2,300 times slower. The thousand additions cost a microsecond sequentially, and pushing a thousand items through three channel hops and a goroutine handoff each costs far more than that. Fan-out is for work that is expensive per item, or slow per item because it waits. If your per-item work is a few nanoseconds, batch the items (send slices of a hundred) so the overhead is amortised, or do not parallelise at all.

Backpressure and buffering

All the channels in these examples are unbuffered, and that is a choice with a consequence. A fast Source cannot run ahead of the workers, and workers cannot run ahead of the consumer, so memory use is flat however long the stream is. A slow consumer slows the whole pipeline, which is the right behavior: the alternative is a queue that grows until the process runs out of memory. Add a buffer when you can name the reason, for example a buffer the size of the worker count between stages to let each stage ride out small variations in the others (buffered or unbuffered), and resist adding large buffers to make a benchmark look better. A large buffer also changes cancellation: items sitting in buffers when you cancel are work done and discarded.

Fan-out over a stream you do not control

The Source above reads a slice. Real fan-out often reads a stream whose size is unknown: lines from a file, rows from a query, messages from a queue. The shape is the same, a producer goroutine that closes its output at the end of the stream and stops early on cancellation. Two additional points matter. First, when the producer blocks on a slow source (a network read), cancellation does not reach it unless the read itself honors the context, so close the connection or set a deadline in a context.AfterFunc (stopping a goroutine). Second, a worker blocked on a slow dependency takes its item with it; if the dependency hangs, workers accumulate behind it and the pipeline stalls, so give each item's work its own timeout.

Variant: keep per-key order

Sometimes you want the parallelism of fan-out and the ordering of a single worker, for items that belong together: all events for one user, all writes to one file, all updates to one account. The way to get both is to route by key. Instead of one shared channel, each worker has its own input, and a dispatcher sends an item to worker hash(key) % n. Items with the same key always land on the same worker and are processed in arrival order; different keys run in parallel.

package main

import (
	"fmt"
	"hash/fnv"
	"sync"
)

type event struct {
	user string
	seq  int
}

func shard(key string, n int) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32()) % n
}

func main() {
	const workers = 3
	inputs := make([]chan event, workers)
	var wg sync.WaitGroup
	var mu sync.Mutex
	order := map[string][]int{} // the order in which each user's events were handled

	for i := range workers {
		inputs[i] = make(chan event, 8)
		wg.Go(func() {
			for e := range inputs[i] {
				mu.Lock()
				order[e.user] = append(order[e.user], e.seq)
				mu.Unlock()
			}
		})
	}

	for seq := 1; seq <= 5; seq++ {
		for _, u := range []string{"ada", "linus", "grace", "ken"} {
			inputs[shard(u, workers)] <- event{u, seq}
		}
	}
	for _, in := range inputs {
		close(in)
	}
	wg.Wait()

	inOrder := true
	for _, seqs := range order {
		for i := 1; i < len(seqs); i++ {
			if seqs[i] < seqs[i-1] {
				inOrder = false
			}
		}
	}
	fmt.Println("users:", len(order), "| every user's events handled in order:", inOrder)
}

users: 4 | every user's events handled in order: true

Four users and three workers: each user's five events go to one worker, in sequence, so the per-user order is preserved, while different users may be handled on different workers at once. The trade-off is that load is no longer self-balancing. A hot key sends all its traffic to one worker while the others idle, and a slow event blocks the events queued behind it on that shard. Hash routing is the right tool when ordering per key matters more than perfect balance; when it does not, the shared channel is better.

Fan-out in the standard library and elsewhere

You have met this pattern without building it. net/http's server is fan-out: the accept loop is the producer, and each connection gets a goroutine. database/sql's connection pool bounds a fan-out of queries. golang.org/x/sync/errgroup with SetLimit is a bounded fan-out for a known batch, with error handling built in, and it is usually the better choice when you have a slice of work and want the first error (errgroup). The channel-based form in this article is for streams, where items arrive over time and stages are connected, and for the cases where you need to control ordering and buffering yourself. A useful rule: for a finite batch with failures that should stop the rest, use errgroup; for a stream passing through stages, use channels with the shapes above.

A review checklist

When you read a fan-out or fan-in in someone else's code, check five things. Who closes the output, and is it exactly one goroutine, after all senders are done? Does every send and receive have a way out on cancellation? Is the number of workers bounded? Does anything depend on result order, and if so, how is it preserved? And is the per-item work expensive enough to pay for the handoffs? If each answer is clear, the pattern is probably sound.

Common mistakes

Closing the merged output from a sender, which races the other senders. Use the coordinator. Forgetting to cancel when the consumer stops early, leaving senders blocked. Assuming order, then discovering it under load. Starting a goroutine per item with no limit; fan-out has a fixed number of workers for a reason. Sharing mutable state between the workers without synchronisation: the whole design is that items are passed, not shared. Ignoring errors from f by giving the function no way to return one. Parallelising work that is too cheap. The benchmark is the antidote.

When to use fan-out, and when not

Fan-out helps when the per-item work is independent and either CPU-bound (more workers use more cores, up to the core count) or latency-bound (more workers overlap more waiting). It does not help when the work is tiny, because channel handoffs cost more than the work (channel costs); when the stages share one bottleneck, such as a single database connection or a lock; or when ordering constraints force serial processing. How many workers to start follows the rules in pool sizing: cores for CPU work, rate times latency, capped by the dependency, for I/O work.

For failures, the stage functions above do not return errors. If f can fail, make the output type carry an error alongside the value, so one failure does not stop the others, or use errgroup where the first failure should cancel the rest (pipeline errors and errgroup). The names come from the Go blog's Pipelines and cancellation, which defines fan-out and fan-in in exactly this sense, and the building blocks are in sync and context.