← Back to Articles
#golang#concurrency#architecture#pipelines

Go Concurrency Pipelines: Mastering Fan-In, Fan-Out, and errgroup

"Deep-dive into concurrent pipeline architectures in Go using Fan-In, Fan-Out, and errgroup.SetLimit for scalable batch and stream data processing."

•By huud•
0 views
~4 min read
Go Concurrency Pipelines: Mastering Fan-In, Fan-Out, and errgroup

Go Concurrency Pipelines: Mastering Fan-In, Fan-Out, and errgroup

Streaming data architectures and batch-processing jobs often require multi-stage pipelines. Go's native channels and concurrency primitives excel at composing streaming stages, but managing channel lifecycles, error propagation, and goroutine synchronization requires precise patterns.

This guide explores idiomatic Pipeline, Fan-Out, and Fan-In patterns in Go, alongside modern error-group orchestration using golang.org/x/sync/errgroup.

Pipeline Fundamentals: Channel Invariants

A pipeline stage is a function that:

  1. Receives values from an inbound channel.
  2. Performs transformation or I/O work.
  3. Sends values to an outbound channel.
  4. Closes the outbound channel when inbound channel closes or context cancels.

Channel Rule of Thumb

The sender owns the channel. Never let a consumer close a channel, or you will trigger a runtime panic: panic: send on closed channel.

Stage 1: Generator Stage

A generator converts a slice or slice-like source into a read-only streaming channel:

go 
package pipeline

import "context"

func Generator[T any](ctx context.Context, items ...T) <-chan T {
	out := make(chan T)
	go func() {
		defer close(out)
		for _, item := range items {
			select {
			case <-ctx.Done():
				return
			case out <- item:
			}
		}
	}()
	return out
}

Stage 2: Fan-Out Pattern

Fan-Out distributes work from a single channel across multiple concurrent goroutines. This is ideal when a processing step is computationally expensive or I/O heavy (e.g., image thumbnailing, external API calls, or cryptographic hashing).

go 
func Worker[T any, R any](
	ctx context.Context,
	in <-chan T,
	transform func(context.Context, T) (R, error),
) <-chan Result[R] {
	out := make(chan Result[R])
	go func() {
		defer close(out)
		for item := range in {
			res, err := transform(ctx, item)
			select {
			case <-ctx.Done():
				return
			case out <- Result[R]{Value: res, Err: err}:
			}
		}
	}()
	return out
}

type Result[T any] struct {
	Value T
	Err   error
}

Stage 3: Fan-In Pattern

Fan-In multiplexes multiple input channels into a single output stream.

go 
import "sync"

func FanIn[T any](ctx context.Context, channels ...<-chan T) <-chan T {
	out := make(chan T)
	var wg sync.WaitGroup

	multiplex := func(c <-chan T) {
		defer wg.Done()
		for val := range c {
			select {
			case <-ctx.Done():
				return
			case out <- val:
			}
		}
	}

	wg.Add(len(channels))
	for _, c := range channels {
		go multiplex(c)
	}

	// Closer goroutine
	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

Orchestrating Complex Workflows with errgroup

While raw channels are great for continuous streams, batch pipelines that require "all-or-nothing" execution and fast-fail error propagation are best managed with errgroup.Group.

Practical Example: Parallel Batch Processing with Rate Limiting

go 
package main

import (
	"context"
	"fmt"
	"time"

	"golang.org/x/sync/errgroup"
)

type Document struct {
	ID   string
	Data string
}

func ProcessDocuments(ctx context.Context, docs []Document) error {
	// Create an errgroup derived from caller's context
	g, ctx := errgroup.WithContext(ctx)
	
	// Limit concurrency to 8 parallel workers (Go 1.20+)
	g.SetLimit(8)

	for _, doc := range docs {
		d := doc // Capture loop variable for closure
		g.Go(func() error {
			// Early exit if another worker already failed
			select {
			case <-ctx.Done():
				return ctx.Err()
			default:
			}

			return transformAndUpload(ctx, d)
		})
	}

	// Wait blocks until all tasks complete or the first returns non-nil error
	if err := g.Wait(); err != nil {
		return fmt.Errorf("pipeline execution failed: %w", err)
	}

	return nil
}

func transformAndUpload(ctx context.Context, doc Document) error {
	// Simulate I/O work
	select {
	case <-time.After(50 * time.Millisecond):
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

Channel Pipeline vs. errgroup.SetLimit: When to Use What?

Criteria Channel Fan-Out/Fan-In errgroup.Group
Data Stream Continuous infinite streaming Finite slice or batch
Error Handling Custom result wrapper needed Built-in first-error propagation
Backpressure Managed via channel buffers Managed via SetLimit(N)
Complexity Higher (manual sync/closing) Minimal (declarative)

Conclusion

When building high-performance Go pipelines:

  1. Always couple channel reads and writes with ctx.Done().
  2. Encapsulate channel mutations within the creator goroutine.
  3. Use errgroup for batch parallel tasks and streaming channels for infinite consumer pipelines.

About the Author

huud

huud

@huud

About →

Systems architect and software engineer building high-performance distributed platforms.