--- title: "Паттерны конкурентности" description: "Горутины, каналы, select и context по отдельности решают маленькие задачи. В реальном коде из них снова и снова собирают одни и те же конструкции. Конвейер, который пропускает…" canonical: https://graphlms.ru/go/book/patterns course: "Конкурентность Go" updated: 2026-10-12 lang: ru publisher: GraphLMS --- # Паттерны конкурентности > Источник: https://graphlms.ru/go/book/patterns — курс «Конкурентность Go», GraphLMS. ## О чём глава [Горутины](/go/book/goroutines), [каналы](/go/book/channels), [select](/go/book/select) и [context](/go/book/context) по отдельности решают маленькие задачи. В реальном коде из них снова и снова собирают одни и те же конструкции. Конвейер, который пропускает данные через несколько стадий обработки. Пул воркеров, который не даёт открыть миллион соединений с базой. Семафор, ограничивающий число одновременных запросов. Батчер, который копит мелкие записи и сохраняет их пачкой. Эта глава — каталог таких конструкций. В каждой корректность держится на двух вопросах, и на ревью я начинаю именно с них: 1. Как до каждой горутины доходит сигнал «данные кончились» или «останавливаемся»? 2. Кто закрывает каждый канал? Ответы задают протокол завершения. Но и этого мало: каждая операция, на которой горутина может ждать, должна либо гарантированно завершиться, либо реагировать на отмену. Иначе горутина зависнет даже при правильно закрытых каналах. ## Перед чтением Нужны каналы, `select`, `WaitGroup`, срезы и [context](/go/book/context). Сначала разберите маленький конвейер и фиксированный пул целиком, от первого значения до выхода последней горутины. Потом посмотрите, что меняется, если потребитель уходит раньше конца. Один термин встретится много раз. **Обратное давление** (backpressure) — ситуация, когда медленный получатель притормаживает быстрого отправителя: отправка ждёт, пока получатель не освободится. Это хорошо: источник не может завалить систему данными быстрее, чем она их обрабатывает. На первом проходе изучите конвейер, fan-out/fan-in и пул воркеров. Семафор, батчинг и errgroup — следующие шаги. Разделы в конце главы готовят к задачам блоков 5 и 7. ## Конвейер (pipeline) Конвейер — цепочка стадий, где выход одной стадии становится входом следующей. Каждая стадия — горутина: читает входной канал, обрабатывает значение, пишет в выходной. Стадии работают одновременно: пока последняя печатает первый результат, первая уже готовит следующий. Дисциплина одна: **стадия закрывает свой выходной канал, когда её вход закончился**. Обычно это `defer close(out)` в начале горутины. Так сигнал «данные кончились» идёт по конвейеру тем же путём, что и данные: следующая стадия увидит закрытие через `range` и закроет свой выход. ```go play package main import "fmt" // gen — источник: отдаёт числа и закрывает выход func gen(nums ...int) <-chan int { out := make(chan int) go func() { defer close(out) for _, n := range nums { out <- n } }() return out } // sq — стадия: возводит в квадрат всё, что пришло func sq(in <-chan int) <-chan int { out := make(chan int) go func() { defer close(out) for n := range in { out <- n * n } }() return out } func main() { for v := range sq(gen(1, 2, 3, 4)) { // gen -> sq -> потребитель fmt.Println(v) } } ``` ```text 1 4 9 16 ``` Здесь одна стадия на каждом шаге, и порядок сохраняется. Ни одна стадия не знает, сколько чисел придёт: она читает, пока вход открыт, и закрывает выход, когда `range` закончился. Можно добавить третью и четвёртую стадию, и код каждой не изменится. ### Когда потребитель уходит раньше Пока потребитель дочитывает всё до конца, проблем нет. Теперь пусть он взял первое значение и вышел из цикла через `break`. `sq` пытается отправить второй квадрат и ждёт на `out <- n * n` вечно: читать некому. `gen` тоже ждёт на отправке следующего числа в `sq`, которая его больше не принимает. Две горутины зависли. Чтобы стадия могла узнать «всё, заканчиваем», ей передают [context](/go/book/context) и слушают `ctx.Done()` рядом с каждой операцией, на которой можно ждать: ```go func sq(ctx context.Context, in <-chan int) <-chan int { out := make(chan int) go func() { defer close(out) for { select { case <-ctx.Done(): return // отмена, пока ждём вход case n, ok := <-in: if !ok { return // вход закончился } select { case out <- n * n: case <-ctx.Done(): return // отмена, пока ждём получателя } } } }() return out } ``` У стадии два места ожидания: приём из `in` и отправка в `out`. Поэтому и `select` с `ctx.Done()` два. `defer close(out)` по-прежнему сообщает следующей стадии, что данных больше не будет. Обратите внимание: отмена не передаётся через `close`. Закрытие идёт вниз по конвейеру, к потребителю, а застрявший источник находится вверху. Поэтому общий `ctx` передают всем стадиям, включая `gen`. Если `gen` контекст не принимает, отменяемая `sq` выйдет, а `gen` останется висеть на отправке. ## Fan-out и fan-in Иногда одна стадия — узкое место: операция над каждым элементом долгая, а элементов много. Тогда стадию распараллеливают. **Fan-out** — несколько горутин читают **один** входной канал. Каждая берёт следующий элемент, как только освободилась, поэтому нагрузка распределяется сама: быстрые воркеры берут больше. **Fan-in** — обратная операция: слить несколько каналов в один, чтобы потребитель видел единый поток. В fan-in в один канал пишут несколько горутин, и снова встаёт вопрос, кто его закроет. Ни одна из них этого сделать не может: она не знает, закончили ли остальные. Поэтому закрывает отдельная горутина-координатор: каждый источник по окончании вызывает `wg.Done()`, координатор ждёт `wg.Wait()` и закрывает выход. ```go play package main import ( "fmt" "sort" "sync" ) func gen(nums ...int) <-chan int { out := make(chan int) go func() { defer close(out) for _, n := range nums { out <- n } }() return out } // worker: одна из параллельных копий стадии func worker(in <-chan int) <-chan int { out := make(chan int) go func() { defer close(out) for n := range in { out <- n * n } }() return out } // merge: fan-in, выход закрывается после того, как замолчали все источники func merge(chans ...<-chan int) <-chan int { out := make(chan int) var wg sync.WaitGroup wg.Add(len(chans)) for _, c := range chans { go func(c <-chan int) { defer wg.Done() for v := range c { out <- v } }(c) } go func() { wg.Wait() // все источники закончили close(out) // закрывает один координатор }() return out } func main() { in := gen(1, 2, 3, 4, 5, 6) // fan-out: три воркера читают один вход w1, w2, w3 := worker(in), worker(in), worker(in) var got []int for v := range merge(w1, w2, w3) { // fan-in got = append(got, v) } sort.Ints(got) // порядок не определён — сортируем для стабильного вывода fmt.Println(got) } ``` ```text [1 4 9 16 25 36] ``` Без сортировки порядок менялся бы от запуска к запуску: три воркера работают независимо. **Fan-out почти всегда перемешивает порядок.** Если порядок важен, его несут вместе с элементом (порядковый номер) и восстанавливают на выходе; см. раздел о сохранении порядка в конце главы. В рабочем `merge` к каждому приёму и отправке добавляют ветку `<-ctx.Done()` по той же причине, что в конвейере: если потребитель ушёл, горутины-пересыльщики не должны зависнуть. ## Пул воркеров (worker pool) Fan-out с квадратами — это уже почти пул. Пул делает идею явной: фиксированное число воркеров разбирает задачи из общего канала и пишет результаты в общий канал. Зачем фиксированное число? Если на каждую задачу запускать горутину, миллион задач превратится в миллион одновременных запросов к базе. Пул из восьми воркеров обработает тот же миллион, но одновременно к базе будет не больше восьми запросов. Размер пула — главный рычаг, которым вы ограничиваете нагрузку на зависимые системы. ```go play package main import ( "context" "fmt" "sort" "sync" ) func pool(ctx context.Context, jobs <-chan int, n int) <-chan int { results := make(chan int) var wg sync.WaitGroup for i := 0; i < n; i++ { wg.Add(1) go func() { defer wg.Done() for { select { case <-ctx.Done(): return case j, ok := <-jobs: if !ok { return // задачи кончились } select { case results <- j * j: case <-ctx.Done(): return } } } }() } go func() { wg.Wait() close(results) // закрывает координатор, а не воркеры }() return results } func main() { ctx := context.Background() jobs := make(chan int) go func() { defer close(jobs) // закрытие — сигнал воркерам, что задач больше не будет for i := 1; i <= 6; i++ { jobs <- i } }() var got []int for r := range pool(ctx, jobs, 3) { // 3 воркера на 6 задач got = append(got, r) } sort.Ints(got) fmt.Println(got) } ``` ```text [1 4 9 16 25 36] ``` Пройдём завершение по шагам, потому что в нём вся суть: 1. Поставщик отправил шесть задач и закрыл `jobs`. 2. Каждый воркер, получив `ok == false`, выходит и вызывает `wg.Done()`. 3. Когда вышли все три, `wg.Wait()` возвращается, и координатор закрывает `results`. 4. `range` в `main` видит закрытие и заканчивается. Ни одна горутина не осталась висеть: у каждого воркера есть обычный выход (закрыт `jobs`) и аварийный (`ctx.Done()`). Структура та же, что в fan-in: воркеров много, а закрывает `results` один координатор после `wg.Wait()`. ## Семафор: не больше N одновременно Пул хорош, когда есть поток однотипных задач и канал результатов. Иногда нужно проще: запустить горутину на каждую задачу, но чтобы одновременно работало не больше N. Для этого хватает буферизованного канала, используемого как счётный семафор. Канал с буфером N — это N пропусков. Перед работой горутина кладёт в канал значение (занимает пропуск). Если все N заняты, отправка ждёт. После работы горутина забирает значение из канала (возвращает пропуск). ```go play package main import ( "fmt" "sync" "time" ) func main() { const limit = 2 sem := make(chan struct{}, limit) var mu sync.Mutex running, maxRunning := 0, 0 var wg sync.WaitGroup for i := 1; i <= 6; i++ { wg.Add(1) go func(id int) { defer wg.Done() sem <- struct{}{} // занять пропуск; ждём, если все заняты defer func() { <-sem }() // вернуть пропуск mu.Lock() running++ if running > maxRunning { maxRunning = running } mu.Unlock() time.Sleep(10 * time.Millisecond) // «работа» mu.Lock() running-- mu.Unlock() }(i) } wg.Wait() fmt.Println("лимит соблюдён:", maxRunning <= limit) } ``` ```text лимит соблюдён: true ``` Горутин шесть, но в участке между «занять» и «вернуть» одновременно бывает не больше двух. Остальные стоят на `sem <- struct{}{}`: полный буфер и есть обратное давление. `maxRunning` на практике будет равен 2, но гарантирована только граница сверху, поэтому программа печатает проверку, а не число. Почему это корректно с точки зрения модели памяти, сказано в [главе о ней](/go/book/memory-model): k-й приём из канала с буфером C happens-before завершения (k+C)-й отправки. В рабочем коде вместо самодельного семафора часто берут `golang.org/x/sync/semaphore`. В нём ожидание пропуска принимает `context` и прерывается по отмене, а одна задача может занимать несколько пропусков сразу. ## Батчинг Писать в базу по одной записи дорого: каждая запись — отдельный запрос по сети. Батчер копит элементы и сохраняет их пачкой. Если ждать только полной пачки, можно застрять: пришло три элемента из ста, поток затих, и эти три не сохранятся никогда. Поэтому триггеров два: набралось N или прошло время, что наступит раньше. Оба живут в одном `select`. ```go play package main import ( "fmt" "time" ) func main() { in := make(chan int) done := make(chan struct{}) go func() { defer close(done) const size = 3 const window = 50 * time.Millisecond buf := make([]int, 0, size) timer := time.NewTimer(window) defer timer.Stop() flush := func(reason string) { if len(buf) > 0 { fmt.Printf("flush(%s): %v\n", reason, buf) buf = buf[:0] } timer.Reset(window) // новое окно отсчитывается от сброса } for { select { case it, ok := <-in: if !ok { flush("close") // вход закрыт: сохраняем остаток return } buf = append(buf, it) if len(buf) >= size { flush("size") } case <-timer.C: flush("time") } } }() for i := 1; i <= 4; i++ { // 4 элемента без пауз in <- i } close(in) <-done } ``` ```text flush(size): [1 2 3] flush(close): [4] ``` Первая пачка уходит по размеру, остаток — по закрытию входа. Триггер по времени здесь не срабатывает: все четыре элемента приходят быстрее 50 мс. Если между `in <- 3` и `in <- 4` вставить `time.Sleep(100 * time.Millisecond)`, четвёрка уйдёт строкой `flush(time): [4]`. `Reset` после сброса пачки начинает новое окно. В модуле с `go 1.23` и новее это безопасно, даже если таймер уже сработал и значение из `timer.C` никто не забрал; до Go 1.23 в такой ситуации в канале оставалось старое значение, и следующий `select` сразу делал лишний сброс по времени (подробно — в [главе select](/go/book/select)). В рабочем батчере есть ещё ветка `<-ctx.Done()`: при остановке сервиса он делает последний сброс и выходит, не теряя накопленное. ## errgroup: параллельные задачи с ошибками Частая задача: запустить N операций, дождаться всех, и если хотя бы одна упала, вернуть её ошибку, а остальные отменить. Вручную это `WaitGroup`, контекст и аккуратный сбор первой ошибки под мьютексом. Пакет `golang.org/x/sync/errgroup` делает это за вас: ```go g, ctx := errgroup.WithContext(ctx) g.SetLimit(8) // не больше 8 задач одновременно; без вызова — без ограничения for _, u := range urls { g.Go(func() error { return fetch(ctx, u) }) } if err := g.Wait(); err != nil { return err // первая ошибка; ctx к этому моменту уже отменён } ``` `g.Wait()` ждёт все задачи и возвращает первую ненулевую ошибку. Производный `ctx` отменяется, как только какая-то задача вернула ошибку, поэтому остальные `fetch`, если они следят за контекстом, сворачиваются сами. `SetLimit` превращает группу в ограниченный пул: `g.Go` будет ждать, пока не освободится место. ## Перед задачами блоков 5 и 7 Эти разделы не решают задачи, а готовят понятия, на которых они построены. ### Частота и одновременность Это разные ограничения. «Не больше трёх воркеров» ограничивает, сколько вызовов идёт одновременно, но не сколько их будет в секунду: если вызов длится 1 мс, три воркера сделают около 3000 вызовов в секунду. RPS (requests per second) — число начатых запросов в секунду, его ограничивают отдельно. Чтобы ограничить частоту для всего пула, воркеры перед каждым вызовом берут разрешение у одного общего лимитера, например ждут тик общего `time.Ticker`. Если дать каждому воркеру свой тикер на 50 RPS, три воркера в сумме выдадут 150. В задаче 27 лимит задан именно на весь пул. ### Token bucket и leaky bucket **Token bucket** (ведро с жетонами) хранит запас разрешений, не больше ёмкости. Каждая операция тратит одно разрешение; время пополняет запас с заданной скоростью. Пример: ёмкость 3, скорость 3 жетона в секунду, в момент 0 ведро полное. - 0 мс: три вызова подряд разрешены, ведро пусто; четвёртый получает отказ. - 100 мс: накопилось 0,3 жетона — меньше целого, вызов получает отказ. - около 333 мс: накопился ровно один жетон, следующий вызов разрешён. Дробную часть запаса нельзя выбрасывать при каждом пересчёте, иначе реальная частота окажется ниже заявленной. Token bucket допускает короткий всплеск (ёмкость), а в среднем держит заданную скорость. Это не то же самое, что «не больше N вызовов в каждом окне в одну секунду». **Leaky bucket** (дырявое ведро) выдаёт разрешения равномерно, с фиксированным интервалом, и вызывающий ждёт своей очереди. Всплесков нет. В задаче 16 `Allow()` не ждёт (token bucket), в задаче 17 `Acquire()` ждёт (leaky bucket). ### Singleflight и cache stampede Популярный ключ пропал из кэша, и в ту же миллисекунду сто запросов идут за ним в базу. Это называют **cache stampede** («давка» при промахе кэша). Мьютекс вокруг карты не спасает: он защищает карту, но не объединяет одинаковые запросы. **Singleflight** объединяет одновременные вызовы с одним ключом: первый вызывающий выполняет функцию, остальные ждут и получают тот же результат. Когда вызов закончился, следующая волна снова выполнит функцию. Singleflight не хранит результат дольше вызова; это делает кэш, и обычно их используют вместе. ### Шардирование Одна карта под одним мьютексом: горутины, работающие с разными ключами, всё равно ждут друг друга. **Шардирование** разбивает карту на несколько независимых частей (шардов), каждая со своим мьютексом. Номер шарда вычисляют из ключа: хеш-функция превращает строку в число, а остаток от деления на число шардов выбирает шард. ```go play package main import ( "fmt" "hash/fnv" ) func shardOf(key string, shards uint32) uint32 { h := fnv.New32a() h.Write([]byte(key)) return h.Sum32() % shards } func main() { for _, k := range []string{"alice", "bob", "alice"} { fmt.Println(k, "->", shardOf(k, 32)) } } ``` ```text alice -> 7 bob -> 20 alice -> 7 ``` Главный инвариант: один и тот же ключ всегда попадает в один шард, иначе `Get` будет искать в другом шарде, чем писал `Set`. Сначала проверьте его на нескольких строках, потом добавляйте конкурентность. Насколько равномерно ключи распределяются и стало ли быстрее — отдельные вопросы, их проверяют бенчмарком. ### Обход графа в ширину (задача 20) В социальном графе пользователи — вершины, дружба — рёбра. Обход в ширину (BFS) сначала посещает всех соседей стартовой вершины, потом соседей соседей, и так по уровням. Возьмём граф `1 → [2, 3]`, `2 → [4]`, `3 → [4, 5]`, `4 → [1]` и начнём с 1: - уровень 1 (один шаг): 2 и 3; - уровень 2: соседи 2 и 3 — это 4, 4 и 5; новые из них 4 и 5; - уровень 3: сосед 4 — это 1, но 1 уже посещена; новых вершин нет. Множество `visited` (уже найденные вершины) не даёт посетить 4 дважды и уйти по кругу обратно в 1. Сначала напишите последовательный обход: текущий уровень — срез вершин, новые соседи собираются в срез следующего уровня. Конкурентная версия обрабатывает вершины одного уровня параллельно и дожидается всего уровня, прежде чем перейти к следующему. Проверка «есть ли вершина в `visited`» и добавление в него должны быть одной критической секцией: иначе два воркера одновременно увидят «нет» и оба возьмут вершину. Семафор ограничивает число одновременных запросов к источнику данных, а не число найденных вершин. ### Приоритеты (задача 25) Для трёх уровней приоритета достаточно трёх срезов-очередей под одним мьютексом. Воркер берёт задачу из очереди 3, если она не пуста, иначе из 2, иначе из 1; если пусты все, ждёт на `sync.Cond` в цикле проверки. Если в пустую очередь положили `(1, A)`, `(3, B)`, `(2, C)`, воркер возьмёт их в порядке B, C, A. Приоритет влияет на выбор следующей задачи, но не прерывает уже выполняющуюся. Для произвольного числа приоритетов есть `container/heap` — структура, из которой быстро достаётся элемент с наибольшим приоритетом. Для трёх фиксированных уровней очереди проще. Строгий приоритет может оставить низкий уровень голодать при непрерывном потоке высокого; если нужна справедливость, нужна отдельная политика, например квота. ### Сохранение порядка (задача 26) Результат второй задачи может быть готов раньше первого. Чтобы отдавать результаты в порядке входа, отдельно учитывают, что уже готово и что пора отдавать. Первый способ: нумеровать входы, складывать готовые результаты по номеру и отдавать только следующий ожидаемый номер. Второй способ — канал каналов: `chan (<-chan int)` передаёт не числа, а каналы, из которых можно прочитать число. На каждую задачу создают свой канал результата с буфером 1 и кладут его в общую очередь в порядке входа. Воркер может заполнить канал второй задачи раньше первой, но сборщик читает каналы из очереди по порядку: сначала ждёт первый, потом второй. Если первый результат задерживается, готовые следом тоже ждут — это цена требования порядка. Буфер 1 позволяет воркеру отдать результат и взять следующую задачу, не дожидаясь сборщика. ### Масштабирование и остановка пула (задачи 23, 30, 32) Начните с фиксированного пула. Затем различайте два числа: сколько воркеров живо и сколько из них занято задачей. Спящий в ожидании воркер жив. Новый воркер запускают, только если есть очередь и не достигнут максимум; решение и изменение счётчика должны происходить под одной блокировкой, иначе два менеджера одновременно решат «можно ещё одного». Лишний воркер уходит по простою, но не ниже минимума. После `Stop` минимум уже не действует: должны завершиться все. Остановка пула состоит из четырёх разных действий, и условие задачи определяет, какие из них нужны: **перестать принимать** новые задачи; **дренировать** очередь — выполнить уже принятые; **отменить** выполняющиеся — попросить их остановиться через контекст; **дождаться** — убедиться, что воркеры вышли. Произвольную `func()` нельзя прервать снаружи: функция с контекстом должна проверять его сама. Если `Stop` вернулся по таймауту, это значит «ждать больше не стали», а не «медленная задача остановлена». ### Паника в задаче, периодические задачи, pub/sub (задачи 24, 29, 31) `recover` ставят вокруг вызова одной задачи, а не вокруг всей функции воркера. Если `recover` стоит на уровне воркера, после паники функция воркера вернётся, и следующие задачи этот воркер уже не возьмёт. В периодическом планировщике `Ticker` отвечает на вопрос «когда отдать задачу», а пул — «кто её выполнит». `ticker.Stop()` не закрывает `ticker.C`, поэтому ожидание тиков совмещают с `done` или `ctx.Done()`. Pub/sub отличается от fan-out: сообщение получает каждый подходящий подписчик, а не один из конкурирующих воркеров. При маршрутизации по тегам сначала под мьютексом скопируйте список подходящих обработчиков, а вызывайте их уже без мьютекса. Иначе один медленный обработчик задержит подписку, публикацию и остановку всего маршрутизатора. ## Частые ошибки - **Канал закрывает не тот или несколько.** Если отправителей несколько, закрывает координатор после `wg.Wait()`, а не каждый воркер. Закрытие из получателя или из нескольких мест — паника. - **Отправка без ветки отмены.** Любая `out <- v` в долгоживущей горутине должна стоять в `select` рядом с `<-ctx.Done()`. Иначе ушедший потребитель оставит её висеть навсегда. - **Отмена передана не всем стадиям.** Отменяемая последняя стадия не освобождает источник, который ждёт на отправке выше по конвейеру. - **Большой буфер вместо протокола завершения.** Буфер откладывает зависание, но не убирает его: когда буфер заполнится, горутина повиснет точно так же. - **`go func()` без ответа на вопрос «кто её остановит».** Нет ответа — будущая утечка. ## Попробуйте сами В конвейере `gen → sq` потребитель принимает только первое число и уходит. Назовите операции, на которых оставшиеся горутины будут ждать. Затем сделайте обе стадии отменяемыми: и на ожидании входа, и на отправке выхода. Передайте им общий контекст, после первого результата отмените его и через `WaitGroup` дождитесь завершения обеих стадий. Большим буфером проблему не решайте. ### Проверка После ухода потребителя `sq` ждёт на отправке следующего квадрата, а `gen` — на отправке следующего числа в `sq`. Оба ожидания должны слушать общий сигнал отмены. Каркас стадии: ```go func squareStage(ctx context.Context, in <-chan int, wg *sync.WaitGroup) <-chan int { out := make(chan int) wg.Add(1) go func() { defer wg.Done() defer close(out) for { select { case <-ctx.Done(): return case n, ok := <-in: if !ok { return } select { case out <- n * n: case <-ctx.Done(): return } } } }() return out } ``` Источник устроен так же: каждая отправка в `select` с `ctx.Done()`, `wg.Done()` при выходе. `wg.Add` делается внутри функции-конструктора, до запуска горутины и до `wg.Wait()` в `main`. Если отмена пришла, пока стадия ждёт вход, сработает первый `select`; если пока ждёт получателя — второй. ## Что дальше Конкурентные паттерны — это графы: горутины в узлах, каналы на рёбрах. Корректность любого из них проверяется теми же двумя вопросами, с которых началась глава: как сигнал остановки доходит до каждого узла и кто закрывает каждое ребро. Как проверить, что после отмены действительно не осталось ни одной горутины, и как поймать гонку данных в таком графе, — в следующей главе, [«Утечки горутин и гонки»](/go/book/leaks-and-races). В задачах курса паттерны — это [блок 3](/go/topics/3) (задачи 09–12: fan-in, пул с ошибками, многостадийный конвейер, батчер), [блок 5](/go/topics/5) (лимитеры, singleflight, шардирование, обход графа) и весь [блок 7](/go/topics/7) (десять вариантов пула). Тесты задач проверяют в первую очередь завершение и отсутствие утечек: ровно то, ради чего в примерах выше стоят ветки `<-ctx.Done()` и единственный владелец закрытия.