Articles / Go

767 words

  • go
  • concurrency
  • goroutine-leaks
  • channels
  • context
  • debugging

Why Your Go Fan-In Goroutine Leaks When the Consumer Walks Away

The textbook fan-in function looks airtight and leaks on the first early exit. If the caller reads one value from the merged channel and then stops, every forwarder goroutine inside the merge sits blocked on a send forever. The goroutine that was meant to close the output is stuck as well, so you leak N+1 goroutines per call.

The merge function everyone writes

Here it is, close to how most tutorials present it:

func merge(ins ...<-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup
	wg.Add(len(ins))
	for _, in := range ins {
		go func() {
			defer wg.Done()
			for v := range in {
				out <- v
			}
		}()
	}
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}

One goroutine per input copies values to out, and one more waits for all of them and closes out. The happy path is correct: if the caller ranges over out until it closes, everything finishes. The trouble is that the design quietly assumes the consumer always reads to the end.

What happens when the consumer stops early

Suppose the caller only wants the first result:

func first(ins ...<-chan int) int {
	out := merge(ins...)
	return <-out
}

After first returns, nobody will ever receive from out again. out is unbuffered, so each forwarder that holds a value is parked on out <- v for good. Nothing closes the channel, nothing cancels anything, and the garbage collector will not rescue you because a goroutine blocked on a channel is still a live goroutine.

The closer is the part people miss. It is blocked in wg.Wait(), because the forwarders never call Done. So with three inputs you strand four goroutines per call, each holding its stack and whatever it captured.

Seeing it happen

You do not need a tool to prove this. Count goroutines before and after:

func gen(start int) <-chan int {
	ch := make(chan int)
	go func() {
		defer close(ch)
		for i := start; ; i++ {
			ch <- i
		}
	}()
	return ch
}

func main() {
	fmt.Println("before:", runtime.NumGoroutine())
	for range 100 {
		first(gen(0), gen(1000), gen(2000))
	}
	time.Sleep(100 * time.Millisecond)
	fmt.Println("after:", runtime.NumGoroutine())
}

Each call also strands three generators, which is deliberate: in a real system the sources are usually the thing that costs you. Expect about 700 extra goroutines from 100 calls, three generators plus four from the merge each time. If you want the exact culprits, a goroutine dump (pprof.Lookup("goroutine").WriteTo(os.Stdout, 2), or /debug/pprof/goroutine?debug=2 on a server) lists them in a chan send state, pointing at the line out <- v.

Quick detour: why the generators leak too

Hang on, the fix to merge will not rescue the generators. A forwarder that stops reading from in leaves the generator blocked on its own send. Fan-in only controls its own goroutines; whoever created the inputs has to give them a way out.

That is why the cancellation signal has to be shared: the same context goes to the generators and to the merge. Passing it to one but not the other just moves the leak upstream.

The fix: every blocking operation also watches the context

A forwarder blocks in two places: receiving from in and sending to out. Both need an escape hatch.

func merge(ctx context.Context, ins ...<-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup
	wg.Add(len(ins))
	for _, in := range ins {
		go func() {
			defer wg.Done()
			for {
				select {
				case v, ok := <-in:
					if !ok {
						return
					}
					select {
					case out <- v:
					case <-ctx.Done():
						return
					}
				case <-ctx.Done():
					return
				}
			}
		}()
	}
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}

The inner select is the important one: it is what frees a forwarder stuck holding a value. The outer select covers the case where the input is idle and would otherwise block the forwarder until the source next speaks, which might be never.

The closer now finishes naturally. Once every forwarder returns, wg.Wait() unblocks and out is closed, so a late reader sees a closed channel rather than a hang.

The caller owns the cancel

The merge cannot know the consumer has left, so the consumer has to say so:

func first(ctx context.Context, start ...int) int {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	ins := make([]<-chan int, len(start))
	for i, s := range start {
		ins[i] = gen(ctx, s)
	}
	return <-merge(ctx, ins...)
}

The defer cancel() is the whole contract. Generators need the same two-way select on their send; the shape is identical to the inner one above. Rerun the goroutine counter with this version and the number settles back to where it started, usually within a few milliseconds of the last return.

Things that look like fixes but are not

  • Buffering out only delays the leak until the buffer fills. It also hides the bug in tests that send fewer values than the buffer holds.
  • Draining out in a goroutine after the early return works, but it costs a goroutine and forces the sources to run to completion, which is no good for infinite ones like gen.
  • Closing out from the consumer turns the leak into a panic, because a forwarder mid-send on a closed channel panics. Only the sending side closes.

Catching it before production does

This is a bug that passes every functional test, because the answer is right; only the goroutine count is wrong. A cheap guard is a test that records runtime.NumGoroutine() before the call, calls it in a loop, then polls briefly for the count to return to baseline. Polling matters, since goroutines take a moment to notice cancellation and exit.

A leaked forwarder is small, so the symptom is rarely a crash. It is memory creeping up on a long-running service until something unrelated gets blamed.

← All articles