The basic channel patterns — fan-out, fan-in, pipelines — appear in Go worker pools and Go channels fundamentals. This guide goes deeper: patterns that solve specific problems in long-running concurrent programs, including how to combine multiple done signals, how to consume from channels that may never close, and how to flatten a stream of streams.
These patterns are drawn from real concurrent system design. Each one addresses a concrete problem you’ll encounter when the basic patterns aren’t enough.
The Or-Done Channel: Exiting Blocked Receives
When you have a goroutine ranging over a channel, it blocks until the channel closes. If the channel never closes — because the producer leaked, or the context was cancelled before the producer finished — your goroutine leaks too.
The or-done pattern wraps any channel receive with a context check, making it safe to range over channels that might not close:
// orDone yields values from ch, stopping when ctx is done
func orDone(ctx context.Context, ch <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for {
select {
case v, ok := <-ch:
if !ok {
return // source closed
}
select {
case out <- v:
case <-ctx.Done():
return
}
case <-ctx.Done():
return
}
}
}()
return out
}
The two-level select is deliberate. The outer select checks for both a new value and cancellation simultaneously. The inner select checks cancellation again when forwarding — important when out is a slow consumer and ctx is cancelled while waiting to send.
Usage is clean — range over the wrapped channel without worrying about the underlying channel’s close behavior:
for v := range orDone(ctx, slowProducer()) {
process(v)
}
// exits cleanly whether slowProducer closes or ctx is cancelled
The Tee Pattern: Duplicating a Stream
Sometimes you need to send every value from one source to two independent consumers. A naive approach of having both consumers read from the same channel won’t work — each value is consumed once. The tee pattern duplicates the stream:
func tee(ctx context.Context, in <-chan int) (<-chan int, <-chan int) {
out1 := make(chan int)
out2 := make(chan int)
go func() {
defer close(out1)
defer close(out2)
for v := range orDone(ctx, in) {
// Shadow with local vars so the goroutine captures the right value
v1, v2 := out1, out2
// Both outputs must receive before the loop continues
for i := 0; i < 2; i++ {
select {
case v1 <- v:
v1 = nil // disable this case
case v2 <- v:
v2 = nil
}
}
}
}()
return out1, out2
}
The trick of setting v1 = nil after a send disables that select case (nil channels never select), ensuring both consumers receive before the loop advances. This makes the tee synchronous — neither consumer gets ahead of the other. If you want independent buffering, add buffer sizes to out1 and out2.
Use tee when you need two independent processing pipelines from one source — for example, logging every event while simultaneously aggregating metrics.
The Bridge Channel: Flattening a Stream of Streams
If you have a <-chan <-chan int (a channel of channels), processing it requires receiving from the outer channel to get inner channels, then receiving from each inner channel in turn. The bridge pattern flattens this into a single <-chan int:
func bridge(ctx context.Context, chanStream <-chan <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for {
var stream <-chan int
select {
case maybeStream, ok := <-chanStream:
if !ok {
return
}
stream = maybeStream
case <-ctx.Done():
return
}
// Drain the inner channel
for v := range orDone(ctx, stream) {
select {
case out <- v:
case <-ctx.Done():
return
}
}
}
}()
return out
}
Bridge is useful when work is chunked into batches (each batch produces its own channel of results) and the consumer wants a flat stream of all results in order.
Queuing: Controlled Backpressure
An unbuffered pipeline stalls completely when any stage is slow — every upstream stage blocks waiting for the slow stage. Adding a buffer between stages lets upstream goroutines continue while the slow stage catches up:
// withBuffer adds a buffer between in and out, absorbing bursts
func withBuffer(ctx context.Context, in <-chan int, size int) <-chan int {
out := make(chan int, size)
go func() {
defer close(out)
for v := range orDone(ctx, in) {
select {
case out <- v:
case <-ctx.Done():
return
}
}
}()
return out
}
Choosing the buffer size involves a tradeoff: a larger buffer absorbs more bursts but uses more memory and increases end-to-end latency. A buffer of zero gives maximum backpressure (upstream is tightly coupled to downstream speed). For most pipelines, a buffer of 10–100 between stages is a reasonable starting point.
Combining Done Signals: The Or-Channel
context.Context is the idiomatic way to cancel a tree of goroutines. But sometimes you have multiple independent cancellation signals and need any one of them to stop a goroutine. The or-channel combines N done channels into one:
func or(channels ...<-chan struct{}) <-chan struct{} {
switch len(channels) {
case 0:
return nil
case 1:
return channels[0]
}
out := make(chan struct{})
go func() {
defer close(out)
switch len(channels) {
case 2:
select {
case <-channels[0]:
case <-channels[1]:
}
default:
select {
case <-channels[0]:
case <-channels[1]:
case <-channels[2]:
case <-or(append(channels[3:], out)...):
}
}
}()
return out
}
This recursive construction handles any number of input channels with O(log n) goroutines. When any input channel closes, the output closes, which propagates through any recursive instances.
In practice you should prefer context.WithCancel and derived contexts — they’re more readable and well-integrated with the standard library. Reach for or-channel when you’re integrating with code that uses raw done channels rather than context.Context.
Retry with Exponential Backoff
Network calls and I/O operations fail transiently. A retry channel pattern yields a channel that emits retry signals with exponential backoff:
func retryBackoff(ctx context.Context, initial, max time.Duration) <-chan struct{} {
out := make(chan struct{})
go func() {
defer close(out)
delay := initial
for {
select {
case out <- struct{}{}: // signal: try now
case <-ctx.Done():
return
}
select {
case <-time.After(delay):
delay = min(delay*2, max)
case <-ctx.Done():
return
}
}
}()
return out
}
// Usage
for range retryBackoff(ctx, 100*time.Millisecond, 30*time.Second) {
err := callExternalAPI()
if err == nil {
break
}
log.Printf("API call failed: %v, retrying...", err)
}
The first receive signals “try now” without any wait. Subsequent receives include the backoff delay. The context cancels the whole loop cleanly.
Pattern Composition: Building a Real Pipeline
These patterns compose — chain them to build pipelines with exactly the behavior you need:
func processEvents(ctx context.Context, events <-chan Event) <-chan Result {
// Step 1: don't leak if events never closes
safe := orDone(ctx, events)
// Step 2: add a buffer to absorb producer bursts
buffered := withBuffer(ctx, safe, 50)
// Step 3: fan out to 5 workers
type worker func(context.Context, <-chan Event) <-chan Result
var workerResults []<-chan Result
for i := 0; i < 5; i++ {
workerResults = append(workerResults, processWorker(ctx, buffered))
}
// Step 4: fan in all worker results
return fanIn(ctx, workerResults...)
}
Each step addresses one concern: leak safety, backpressure, parallelism, result collection. The composition is readable because each function has a clear input and output channel.
Common Mistakes
Not checking context in both send and receive selects. A goroutine that checks context only on receive can get stuck trying to send to a full downstream channel. Put case <-ctx.Done(): return in every select that sends or receives.
Tee without synchronization. A naive tee that spawns two goroutines — one sending to each output — doesn’t synchronize between the two sends. If one consumer is slower, the other gets ahead and the values appear at different rates. Use the nil-channel trick shown above to ensure both consumers receive each value before moving to the next.
Bridge without orDone on inner channels. If inner channels might not close (because the goroutine producing them leaked), bridge will stall waiting for them. Always wrap inner channel reads with orDone.
Large recursive or-channels. The recursive or-channel creates O(log n) goroutines per invocation. For very large N, this accumulates. Prefer context.WithCancel and structured context trees for most cancellation needs.
Summary
- or-done: safe range over channels that may not close — wraps every receive with a context check
- tee: duplicate a stream for two independent consumers, synchronized so neither gets ahead
- bridge: flatten
<-chan <-chan Tinto<-chan T— useful when work is batched into chunks each producing their own channel - buffer stage: absorb producer bursts between pipeline stages by inserting a buffered channel
- or-channel: combine N done signals so any one cancels the group — prefer
context.Contextwhere possible - Compose these patterns: each takes and returns channels, making them chainable
Resources
- Go Blog: Pipelines and Cancellation
- Concurrency in Go — Katherine Cox-Buday (the source of many of these patterns)
- Go by Example: Select
- golang.org/x/sync
Comments