go-concurrency-patterns

Go Concurrency Patterns

Production-grade concurrency patterns for Go. Each pattern includes when to use it, the implementation, and common pitfalls.

How It Works

1. Agent identifies the concurrency need from the task description
2. Agent selects the appropriate pattern(s) from this catalog
3. Agent implements the pattern with proper error handling and cancellation
4. Agent verifies goroutine lifecycle management (no leaks)

Patterns

1. Worker Pool (Critical)

Bounded concurrency for processing a stream of work items. Controls resource usage.

When to use: Processing N items with at most M concurrent workers.

func WorkerPool[T any, R any](
    ctx context.Context,
    items []T,
    workers int,
    process func(context.Context, T) (R, error),
) ([]R, error) {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(workers)

    var mu sync.Mutex
    results := make([]R, 0, len(items))

    for _, item := range items {
        g.Go(func() error {
            result, err := process(ctx, item)
            if err != nil {
                return err
            }
            mu.Lock()
            results = append(results, result)
            mu.Unlock()
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

Pitfalls:

• Without SetLimit, all items start concurrently (not a pool)
• Results are unordered — use index-based collection if order matters
• First error cancels remaining work via errgroup context

2. Fan-Out / Fan-In (Critical)

Distribute work across multiple goroutines, then collect results.

func FanOutFanIn[T any](
    ctx context.Context,
    input <-chan T,
    workers int,
    process func(context.Context, T) T,
) <-chan T {
    // Fan-out: create N worker channels
    channels := make([]<-chan T, workers)
    for i := 0; i < workers; i++ {
        channels[i] = worker(ctx, input, process)
    }

    // Fan-in: merge all worker channels into one
    return merge(ctx, channels...)
}

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

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

    output := func(ch <-chan T) {
        defer wg.Done()
        for item := range ch {
            select {
            case out <- item:
            case <-ctx.Done():
                return
            }
        }
    }

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

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

    return out
}