Fan-Out/Fan-In in Go: pattern for parallel data processing

Fan-Out/Fan-In Pattern in Go: Flexible Parallel Data Processing

The Fan-Out/Fan-In pattern gives you a powerful way in Go to distribute tasks across goroutines and then aggregate the results. Unlike the Worker Pool pattern, it offers greater flexibility through individual processing pipelines for each worker, which is especially handy for complex data transformation chains.

Core Principles of the Pattern

The pattern breaks down into two key phases. The Fan-Out phase distributes input data across multiple workers (goroutines). Each worker gets its own task, processes it independently, and sends the result to its dedicated channel. The Fan-In phase then collects results from all these individual worker channels into a single output channel for downstream processing.

The main difference from Worker Pool lies in the channel architecture:

• Worker Pool: All workers typically read tasks from one shared channel and write results to another shared channel.
• Fan-Out/Fan-In: Each worker has its own output channel, requiring an extra merge step.

This approach adds some overhead due to the extra goroutines and channels for merging, but it shines in pipeline architectures where later stages need separate data streams.

Practical Implementation

Let's walk through a hands-on implementation that's deliberately different from a classic Worker Pool to highlight the key features.

Worker Function

In this setup, the worker is a function that spins up a goroutine and returns a results channel. This differs from passing a channel as a parameter.

func worker(x int, workerID int) chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        fmt.Printf("worker %d, job %d\n", workerID, x)
        out <- x * x
    }()
    return out
}

Key implementation details:

• The goroutine closes its out channel with defer after sending the result.
• For demo purposes, the work is trivial—squaring a number.
• In production code, you'd need to refine this: manage lifecycle with context for graceful shutdowns and timeouts.

Fan-Out Phase (Distribution)

The distributor (producer) launches one worker per input data item. The channels returned by workers are stored in a slice for later merging.

for i, x := range data {
    chnls = append(chnls, worker(x, i+1))
}

Key feature: The number of workers scales dynamically with the task count, which isn't mandatory but shows the pattern's flexibility. In real apps, you'd usually cap the number of concurrent workers.

Fan-In Phase and Merge Function

The merge function handles the crucial job of combining data from multiple channels into one. It:

• Creates a single unbuffered output channel out.
• For each input channel, launches a goroutine that reads data and forwards it to out.
• In a separate goroutine, waits for all reader goroutines to finish using sync.WaitGroup and then closes out.

func merge(chnls ...chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, ch := range chnls {
        wg.Add(1)
        go func(c chan int) {
            defer wg.Done()
            for x := range c {
                out <- x
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

Technical note: The wait (wg.Wait()) runs in a separate goroutine. Doing it in the main flow could cause a deadlock if the consumer hasn't started reading from out yet, blocking writes.

Consumer

The consumer reads results from the single merged channel. The for range loop exits automatically when merge closes out.

for x := range out {
    results = append(results, x)
}

Key Takeaways

• Architectural flexibility: Individual pipelines per worker make it easy to insert extra data processing stages.
• Dynamic scaling: It theoretically supports one worker per task, though you'd control concurrency in practice.
• Overhead: The pattern uses extra resources (channels, goroutines) for merging compared to simpler approaches.
• Lifecycle management: This demo skips contexts, which is fine for examples but essential in production for cancellations and timeouts.
• Code readability: Unbuffered channels and for range loops keep things concise and clear.

Production Recommendations

This demo code is for learning. For real-world use, make these upgrades:

• Add context (context.Context): Pass it to workers and merge for handling cancellations, deadlines, and timeouts.
• Limit concurrency: Unlimited goroutines (one per task) can exhaust resources. Use semaphores or a fixed-size worker pool in Fan-Out.
• Error handling: Build in a way to propagate errors from workers to the consumer, like a dedicated error channel or result struct.
• Buffered channels: They can boost performance by reducing blocking in some cases, but test carefully.

The Fan-Out/Fan-In pattern is a powerhouse in every Go developer's toolkit for building efficient, scalable data processing pipelines—especially when you need flexibility in managing data flows between stages.

— Editorial Team