Конвейеры: fan-out, fan-in и отмена

Перед началом: модули 1–5 (планировщик, каналы, select, sync). Инструменты: Go 1.22+, go test -race, docker для лимита памяти. Трудоёмкость: ~4 ч (теория 1.5 ч · практика 2 ч · квиз 0.5 ч).

🔥 Инцидент: ночной ETL не уложился в окно, а «ускорение» положило сервис

Ночной ETL-сервис забирает 1,4 млн записей из очереди, валидирует, дополняет и пишет в базу. Окно — два часа; валидация одной записи занимает ~40 мс, и сервис стабильно не укладывался.

Инженерно очевидное решение: «каждая запись независима — запускай горутину на запись». В первую же ночь процесс был убит OOM-killer: 1,4 млн горутин, каждая с результатом и буферами, съели всю память, а база получила лавину одновременных вставок и упёрлась в пул соединений. Вторая ночь — с ограничением «только 10 000 горутин за раз»: стало лучше, но после первой невалидной записи пайплайн продолжал молотить остаток, и половина окна ушла на обработку мусора, который всё равно не запишется.

Правильная конструкция была не «горутина на запись», а конвейер с явными этапами, ограниченным fan-out и сквозной отменой — ровно то, из чего собран этот модуль.

📖 Теория

6.1 Каналы — это конвейер, если у этапов есть владелец

Этап конвейера — функция, которая берёт <-chan T, возвращает <-chan U и владеет своим выходным каналом: только он пишет и только он закрывает.

go
func validate(in <-chan Row) <-chan Row {
    out := make(chan Row)
    go func() {
        defer close(out) // владелец закрывает — это контракт
        for r := range in {
            if err := check(r); err != nil {
                continue
            }
            out <- r
        }
    }()
    return out
}

Два правила, без которых конвейер разваливается:

• Никто, кроме владельца, не закрывает канал. Потребитель, закрывший чужой канал, роняет продюсера паникой send on closed.
• Направленные типы (<-chan / chan<-) делают владение видимым в сигнатуре — компилятор запрещает писать в «только чтение».

Показать кодКопировать код

6.2 Fan-out и fan-in: считать быстрее, не теряя учёта

Fan-out — один канал читают несколько воркеров: каждый элемент достаётся ровно одному (каналы работают как очередь без повторов). Fan-in — выходы всех воркеров сливаются в один канал; закрывается он, когда закрыты все входные.

go
func merge(ins ...<-chan Result) <-chan Result {
    out := make(chan Result)
    var wg sync.WaitGroup
    wg.Add(len(ins))
    for _, in := range ins {
        go func(in <-chan Result) {
            defer wg.Done()
            for v := range in {
                out <- v
            }
        }(in)
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

Порядок элементов после fan-out больше не определён. Если он нужен — добавляйте к результату порядковый номер и сортируйте на выходе (или собирайте в срез по индексу); если не нужен — не платите за него.

6.3 Отмена: stop должен дойти до каждого этапа

Пайплайн без отмены — это генератор, который продолжает производить работу, когда потребитель уже упал. Каждый send и каждый receive обязаны уметь выйти:

go
select {
case out <- v:
case <-ctx.Done(): // или <-stop
    return
}

Сквозной ctx прокидывается во все этапы и в select каждого шага; первая ошибка вызывает cancel() — и весь конвейер сворачивается сверху вниз. Важно: отмена не отменяет уже начатую запись — это решает идемпотентность (модуль 13).

6.4 Backpressure: буфер — это долг, а не скорость

Небуферизованный канал — рукопожатие: продюсер ждёт, пока элемент не примут. Буфер переводит это ожидание в «долг»: продюсер бежит на N элементов вперёд. Отсюда правило:

• Медленный этап должен тормозить быстрый — иначе очередь растёт (сначала в канале, потом в памяти, потом в OOM).
• Буфер сглаживает неравномерность, но не увеличивает пропускную способность: если writer вставляет 500 строк/с, а generator даёт 5000 — буфер любого размера будет переполняться.

Мерило здоровья конвейера — не «нет блокировок», а стабильная длина очередей: металлики на длину канала (или счётчик len(ch) в экспозиции) показывают, где конвейер не справляется.

6.5 Компромиссы

• Fan-out шире vs уже: больше воркеров — выше параллелизм, но больше переключений и давления на общий ресурс (БД, внешний API); лишние воркеры просто стоят на канале.
• Буфер больше vs меньше: сглаживает пики и уменьшает блокировки, но увеличивает задержку «до обработки» и память; unbuffered честнее показывает узкое место.
• Порядок vs скорость: сохранение порядка требует индексов и сборки — иногда это дороже самой обработки.
• Один errgroup vs свой merge: errgroup проще и приносит отмену бесплатно, но смешивает «все задачи» в один пул; свой merge даёт контроль над этапами и их задержками.

💻 Практика

1. Воспроизведите OOM. ETL-заглушка: генератор 500k «записей», обработчик 1 мс; вариант А — go process(row) без ограничений, вариант Б — пул из 8 воркеров. Запустите оба в контейнере с --memory=256m, снимите пиковую память и число горутин.
2. Соберите конвейер с отменой. Генератор → валидатор → fan-out (4 воркера) → merge. «Грязная» запись на позиции 10 000 вызывает cancel(). Проверьте: все этапы завершаются, ни одна горутина не висит (leak-тест из модуля 17), счётчики показывают, сколько обработано до отмены.
3. Замерьте backpressure. Сделайте writer с задержкой 2 мс и буферы 0, 256, 4096. Покажите, как меняются длина очереди и пиковая память при неизменной пропускной способности; объясните, почему буфер 4096 не ускорил запись.
4. Запустите сниппет модуля: конвейер «генератор → квадрат → fan-out 3 → fan-in» с суммой и проверкой, что ни один элемент не потерян и не посчитан дважды.

🧩 Quiz

✅ Итоги модуля

Студент должен:

• Строить этапы конвейера с владением каналами
• Писать fan-out и fan-in без потерь и дублей
• Проводить отмену через все этапы через ctx и select
• Объяснить backpressure и почему буфер не равен скорости
• Диагностировать конвейер по длинам очередей, а не по CPU
• Не допускать горутину-на-элемент без ограничения
• Ответить на все вопросы Quiz

🔗 Мостик к Модулю 7

Конвейер задаёт структуру обработки, но сколько воркеров ставить в fan-out и как ограничить параллелизм внешнего ресурса — отдельный вопрос. Дальше — worker pool, семафоры, errgroup с SetLimit и цена каждого лишнего воркера.