Паттерны concurrency: pipelines, fan-in/out, errgroup, singleflight

Урок-мост между теорией каналов из 04-concurrency-deep и продакшн-кодом. Шесть паттернов, которые встречаются в реальных сервисах каждый день, плюс типовые ошибки. Считаем, что про goroutine, channel, select, context уже понятно, иначе сначала пройди 04.

В качестве референса полезно держать рядом блог-пост Go-team Pipelines and cancellation, доклад Брайана Миллса Rethinking Classical Concurrency Patterns и книгу Кэтрин Кокс-Бьюди “Concurrency in Go”.

1. Главный закон: канал, всегда с владельцем

Перед любым паттерном, одно правило, которое спасает от 80% leak-ов и паник:

У каждого канала ровно один писатель и явный владелец, который его закрывает.

Закрывать канал из читателя, ошибка. Закрывать дважды, паника. Писать в закрытый канал, паника. Если правило соблюдено, дальше всё проще: читатель видит закрытый канал через for v := range ch или через v, ok := <-ch, сам останавливается, без флагов и счётчиков.

Закрыть канал, идиоматично через defer сразу после старта горутины:

func source(ctx context.Context, out chan<- int) {
    defer close(out)
    for i := 0; i < 1000; i++ {
        select {
        case <-ctx.Done():
            return
        case out <- i:
        }
    }
}

out chan<- int, write-only direction. Компилятор не даст из неё читать. Это контракт: получатель функции теперь точно знает, что эту сторону никто извне не закроет.

2. Pipeline через каналы

Pipeline, последовательность стадий. Каждая стадия, отдельная горутина, читает из входа, пишет в выход. Связи через каналы.

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

func transform(ctx context.Context, in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for item := range in {
            res := strings.ToUpper(item)
            select {
            case <-ctx.Done():
                return
            case out <- res:
            }
        }
    }()
    return out
}

func sink(in <-chan string) []string {
    var acc []string
    for item := range in {
        acc = append(acc, item)
    }
    return acc
}

// Сборка
ctx, cancel := context.WithCancel(ctx)
defer cancel()
result := sink(transform(ctx, source(ctx, items)))

Что здесь полезно подметить:

• Direction-аннотации, <-chan string для чтения, chan<- string для записи. Стадии не могут случайно закрыть чужой канал.
• select с <-ctx.Done() в каждой записи. Без этого при cancel стадия залипнет на полном канале (back-pressure из соседней стадии).
• defer close(out) в каждой стадии. Закрытие распространяется каскадом до sink-а, тот выходит из range.

Поиграй с buffer size, workers и source rate ниже. Когда буфер маленький плюс воркеров мало, source блокируется, видно красную рамку и счётчик blocked. Это и есть back-pressure, которая в pprof-е ловится через block-profile.

source

idle

chan

0/4

sink

0

0 emitted0 blocked0 done0 tps

Подними buffer до 0, и source залипает на каждом отправке (back-pressure через `chan` без буфера). Прибавь воркеров, и канал перестанет наполняться. Подними source rate выше суммарной пропускной воркеров, и канал вечно полный, source блокируется. Это ровно то, что увидишь в pprof block-profile.

3. Fan-out: распараллелить тяжёлую стадию

Когда стадия медленная (HTTP, GPU, дорогая регулярка), хочется тянуть её в N параллельных воркеров. Это fan-out. Технически, N горутин читают из одного канала, пишут в другой. Идиома:

func fanOut[T, R any](ctx context.Context, in <-chan T, n int, fn func(context.Context, T) R) <-chan R {
    out := make(chan R)
    var wg sync.WaitGroup
    wg.Add(n)
    for i := 0; i < n; i++ {
        go func() {
            defer wg.Done()
            for item := range in {
                res := fn(ctx, item)
                select {
                case <-ctx.Done():
                    return
                case out <- res:
                }
            }
        }()
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

Главные моменты:

• N воркеров читают range in. Когда in закрывается, каждый воркер выходит из цикла сам.
• sync.WaitGroup плюс отдельная горутина закрывает out. Без этой развязки невозможно безопасно закрыть выходной канал: ни один воркер не знает, “последний он” или нет.
• Порядок не сохраняется. На выходе fan-out элементы перемешаны. Если порядок важен, либо нумеруй на входе и сортируй на выходе, либо не делай fan-out.

Размер N это эмпирика: для CPU-bound, runtime.NumCPU(); для IO-bound, начинают с 4 * NumCPU и подкручивают.

4. Fan-in: собрать N каналов в один

Зеркало fan-out. Несколько источников пишут разные подмножества данных, их нужно мерджить в один поток. Идиома по sync.WaitGroup:

func fanIn[T any](ctx context.Context, ins ...<-chan T) <-chan T {
    out := make(chan T)
    var wg sync.WaitGroup
    wg.Add(len(ins))
    for _, ch := range ins {
        go func(c <-chan T) {
            defer wg.Done()
            for v := range c {
                select {
                case <-ctx.Done():
                    return
                case out <- v:
                }
            }
        }(ch)
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

Альтернатива через рефлексию (reflect.Select) бывает нужна, когда число каналов известно только в рантайме. Скучнее и медленнее, поэтому почти всегда хватает варианта выше.

5. errgroup: первый сбойный отменяет остальных

В пакете golang.org/x/sync/errgroup лежит самый используемый concurrency-инструмент после goroutine с каналами. Запускаем N задач, ждём всех, при первой ошибке отменяем context и собираем первую (или все) ошибки.

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

func fetchAll(ctx context.Context, urls []string) ([]string, error) {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(8)

    bodies := make([]string, len(urls))
    for i, u := range urls {
        i, u := i, u
        g.Go(func() error {
            body, err := fetch(ctx, u)
            if err != nil {
                return fmt.Errorf("fetch %s: %w", u, err)
            }
            bodies[i] = body
            return nil
        })
    }
    if err := g.Wait(); err != nil {
        return nil, err
    }
    return bodies, nil
}

Что важно:

• errgroup.WithContext(ctx) возвращает новый context. При первом return err он отменяется, и все живые g.Go(...) видят ctx.Done(), должны выйти.
• g.SetLimit(N) ограничивает максимум одновременно работающих goroutine. Без этого 1000 URL запускают 1000 горутин и могут уложить и тебя, и провайдера.
• Запись в общий слайс, безопасна когда каждая горутина пишет в свой индекс. Если на ту же позицию пишут двое, нужен mutex (или лучше, отдельный канал результатов).
• return nil обязательно. Если функция возвращает error, его обязательно вернуть, даже когда успех.

С Go 1.22 i, u := i, u уже не нужен (захват переменной цикла починили), но в legacy-коде ты эту строку увидишь часто.