go-concurrency

Go Concurrency Patterns

Context Propagation

// Every I/O function: context as first param
func (s *Service) Process(ctx context.Context, id string) error {
    data, err := s.repo.Get(ctx, id)
    if err != nil { return fmt.Errorf("getting data: %w", err) }

    if err := ctx.Err(); err != nil { return err } // check before expensive ops
    return s.handle(ctx, data)
}

// Timeout
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()

// Values (sparingly, use unexported key types)
type contextKey string
const reqIDKey contextKey = "requestID"
func WithReqID(ctx context.Context, id string) context.Context {
    return context.WithValue(ctx, reqIDKey, id)
}

errgroup

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

// Basic — concurrent fetch
g, ctx := errgroup.WithContext(ctx)
results := make([]Data, len(ids))
for i, id := range ids {
    g.Go(func() error {
        data, err := fetch(ctx, id)
        if err != nil { return fmt.Errorf("fetching %s: %w", id, err) }
        results[i] = data
        return nil
    })
}
if err := g.Wait(); err != nil { return nil, err }

// With concurrency limit
g.SetLimit(10)
for _, item := range items {
    g.Go(func() error { return process(ctx, item) })
}

Goroutine Lifecycle

func (w *Worker) Start(ctx context.Context) error {
    g, ctx := errgroup.WithContext(ctx)
    for range w.numWorkers {
        g.Go(func() error {
            for {
                select {
                case <-ctx.Done():
                    return ctx.Err()
                case job, ok := <-w.jobs:
                    if !ok { return nil }
                    if err := w.process(ctx, job); err != nil { return err }
                }
            }
        })
    }
    return g.Wait()
}

Channel Patterns

// Pipeline
func Pipeline(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for { select {
            case <-ctx.Done(): return
            case v, ok := <-in:
                if !ok { return }
                select { case out <- transform(v): case <-ctx.Done(): return }
        }}
    }()
    return out
}