Worker pool и лимиты параллелизма

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

🔥 Инцидент: сто тысяч горутин против десяти соединений

Импорт каталога: 100 000 строк из CSV, каждая — валидация, три запроса к базе и запись. Пул соединений — 10. Инженер запустил go importRow(row) на каждую строку: 100 000 горутин почти мгновенно встали в очередь за соединениями, память выросла на гигабайты, а часть импорта упала с context deadline exceeded. Хуже: после первой ошибки процесс не остановился — остаток строк продолжил молотить мусор, который всё равно не запишется, и зря держал соединения.

Нужен не запрет горутин, а явный лимит одновременной работы и правило, что делать, когда работа больше не нужна.

📖 Теория

7.1 Пул воркеров: jobs, results и закрытие

Пул — это N горутин, которые читают один канал задач. Схема:

продюсер пишет в jobs и закрывает его, когда задачи кончились;
каждый воркер делает for row := range jobs;
выходы сливаются в results, а закрывает его отдельная горутина после wg.Wait() — когда все воркеры закончили.

go
func pool(ctx context.Context, n int, jobs <-chan Row) <-chan Res {
    out := make(chan Res)
    var wg sync.WaitGroup
    wg.Add(n)
    for i := 0; i < n; i++ {
        go func() {
            defer wg.Done()
            for row := range jobs {
                select {
                case out <- process(ctx, row):
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

Три контракта, без которых пул ломается:

Закрывает jobs только продюсер, а results — только closer; воркер, закрывший чужой канал, роняет соседей паникой.
Воркер умеет выйти: select с ctx.Done() на отправке результата, иначе он навсегда зависнет на out <-.
Продюсер тоже умеет выйти: если воркеры ушли по отмене, а он пишет в jobs, нужен select с ctx.Done() и там.

7.2 errgroup: Go, SetLimit и отмена по первой ошибке

errgroup.WithContext возвращает группу и контекст. Первая ненулевая ошибка отменяет этот контекст, а Wait возвращает первую ошибку. SetLimit(n) ограничивает число одновременно выполняемых Go: при заполненном лимите сам вызов Go блокируется и ждёт слот.

go
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(10)
for _, row := range rows {
    g.Go(func() error {
        return importRow(ctx, row) // ctx отменится после первой ошибки
    })
}
if err := g.Wait(); err != nil {
    return fmt.Errorf("импорт: %w", err)
}

Разница с самодельным семафором важна:

SetLimit блокирует продюсера до старта горутины — горутин ровно n, но чтение входных данных замедляется (backpressure);
семафор внутри goroutine (как в сниппете) ограничивает только одновременное выполнение, но горутина под каждую задачу всё равно создаётся: 100 000 задач — 100 000 ждущих горутин;
TryGo не ждёт слот, а возвращает false — удобно для быстрого отказа при перегрузке.

В сниппете это видно буквально: 20 задач создают 20 горутин, хотя одновременно работают только 4 — остальные ждут слот семафора.

Не вызывайте Go из задачи той же группы при исчерпанном лимите: получится взаимная блокировка. И помните: errgroup отменяет только тех, кто смотрит в ctx, — об этом 7.4.

7.3 Семафор-канал и weighted semaphore

Канал с буфером — простейший семафор: отправка занимает слот, получение освобождает.

go
sem := make(chan struct{}, 10)

select {
case sem <- struct{}{}:
    defer func() { <-sem }()
case <-ctx.Done():
    return ctx.Err() // ждать слот бессмысленно
}

Если ждать не нужно, default превращает захват в «попробовать и отказать»: это уже load shedding, а не лимит.

semaphore.NewWeighted(n) нужен, когда операции стоят по-разному: тяжёлый отчёт берёт 3 слота, лёгкий запрос — 1. Acquire(ctx, 3) уважает отмену, Release(3) обязан вернуть ровно столько же; TryAcquire не блокируется. Очередь ожидающих соблюдается, но крупный захват ждёт, пока наберётся нужный вес, и может задержать мелкие запросы, стоящие за ним.

7.4 Отмена: кто проверяет ctx в задаче

Отмена в Go кооперативная: cancel() не убивает горутину, а только закрывает ctx.Done(). Проверять обязан тот, кто выполняет работу:

сетевые и DB-вызовы получают ctx и завершаются сами;
долгий цикл проверяет ctx.Err() каждые N итераций или перед каждой дорогой операцией;
задача, игнорирующая ctx, доработает до конца, и Wait будет ждать именно её.

Поэтому errgroup с отменой не отменяет уже начатые side effects: часть строк успеет записаться. Для импорта это значит, что нужен идемпотентный ключ или статус строки, иначе повторный запуск задвоит данные. Отмена экономит будущую работу, а не откатывает прошлую.

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

Пул vs семафор: пул держит ровно N горутин и тормозит продюсера; семафор проще встроить в существующий код, но не ограничивает число созданных горутин.
Ждать слот vs отказать: ожидание сглаживает всплеск, но скрывает перегрузку и копит задержку; быстрый отказ честнее, но требует retry или очереди выше по стеку.
Лимит по CPU vs по ресурсу: воркеров должно быть столько, сколько выдерживает узкий ресурс — пул БД, квота API, диск. Если задача делает три параллельных запроса, эффективный спрос втрое больше числа воркеров.
Fail-fast vs best-effort: первая ошибка отменяет всё и экономит время, но теряет частичный результат; сбор всех ошибок даёт полную картину ценой лишней работы.
Буфер jobs: сглаживает неровный вход, но прячет отставание; небуферизованный канал сразу показывает давление.

💻 Практика

Воспроизведите лавину. 100 000 «строк» с обработкой 1 мс и пулом соединений 10: вариант А — go importRow(row) без лимита, вариант Б — пул из 10 воркеров. Снимите пик горутин, памяти и число ошибок таймаута.
Соберите errgroup с лимитом. SetLimit(10), ошибка на 500-й строке. Проверьте, что ctx отменён, новые задачи не стартуют, а начатые видят отмену внутри importRow.
Взвешенный семафор. Отчёт занимает 3 слота, лёгкий запрос — 1, лимит 4. Покажите, что два отчёта не идут одновременно, и проверьте освобождение слота при ctx.Done().
Запустите сниппет модуля: мини-errgroup на 20 задачах с лимитом 4, затем вторая часть, где ошибка третьей задачи отменяет очередь. Вывод: пик параллельности 4 при лимите 4, ошибка группы: <nil>, затем [ok ok ERR skip skip skip] — хвост очереди не стартовал.

🧩 Quiz

1. Кто закрывает канал результатов пула воркеров?

2. Чем канал-семафор внутри горутины отличается от пула воркеров?

3. Кто проверяет ctx внутри длинной задачи?

4. Почему лимит воркеров выбирают по пулу БД, а не по числу CPU?

5. Что делает g.SetLimit(10) в errgroup?

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

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

Строить пул с каналами jobs/results и корректным закрытием
Использовать errgroup.SetLimit и объяснять его отличие от семафора
Ограничивать параллелизм каналом-семафором и weighted semaphore
Проводить отмену внутрь задачи, а не только до её старта
Выбирать лимит по узкому ресурсу и считать эффективный спрос
Различать fail-fast и best-effort при массовых ошибках
Ответить на все вопросы Quiz

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

Лимит параллелизма защищает ресурс от одновременного наплыва, но не задаёт скорость обращений во времени: 10 воркеров могут сделать 10 000 вызовов в секунду, если каждый быстрый. Дальше — rate limiting: token bucket, окна, лимиты по ключам и load shedding, когда пора отклонять запросы, а не ставить их в очередь.

← Модуль 6: Конвейеры: fan-out, fan-in и отмена Модуль 8: Rate limiting: token bucket и окна →