Паттерны конкурентности

О чём глава

Горутины, каналы, select и context по отдельности решают маленькие задачи. В реальном коде из них снова и снова собирают одни и те же конструкции. Конвейер, который пропускает данные через несколько стадий обработки. Пул воркеров, который не даёт открыть миллион соединений с базой. Семафор, ограничивающий число одновременных запросов. Батчер, который копит мелкие записи и сохраняет их пачкой.

Эта глава — каталог таких конструкций. В каждой корректность держится на двух вопросах, и на ревью я начинаю именно с них:

1. Как до каждой горутины доходит сигнал «данные кончились» или «останавливаемся»?
2. Кто закрывает каждый канал?

Ответы задают протокол завершения. Но и этого мало: каждая операция, на которой горутина может ждать, должна либо гарантированно завершиться, либо реагировать на отмену. Иначе горутина зависнет даже при правильно закрытых каналах.

Перед чтением

Нужны каналы, select, WaitGroup, срезы и context. Сначала разберите маленький конвейер и фиксированный пул целиком, от первого значения до выхода последней горутины. Потом посмотрите, что меняется, если потребитель уходит раньше конца.

Один термин встретится много раз. Обратное давление (backpressure) — ситуация, когда медленный получатель притормаживает быстрого отправителя: отправка ждёт, пока получатель не освободится. Это хорошо: источник не может завалить систему данными быстрее, чем она их обрабатывает.

На первом проходе изучите конвейер, fan-out/fan-in и пул воркеров. Семафор, батчинг и errgroup — следующие шаги. Разделы в конце главы готовят к задачам блоков 5 и 7.

Конвейер (pipeline)

Конвейер — цепочка стадий, где выход одной стадии становится входом следующей. Каждая стадия — горутина: читает входной канал, обрабатывает значение, пишет в выходной. Стадии работают одновременно: пока последняя печатает первый результат, первая уже готовит следующий.

Дисциплина одна: стадия закрывает свой выходной канал, когда её вход закончился. Обычно это defer close(out) в начале горутины. Так сигнал «данные кончились» идёт по конвейеру тем же путём, что и данные: следующая стадия увидит закрытие через range и закроет свой выход.

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)
	}
}

1
4
9
16

Здесь одна стадия на каждом шаге, и порядок сохраняется. Ни одна стадия не знает, сколько чисел придёт: она читает, пока вход открыт, и закрывает выход, когда range закончился. Можно добавить третью и четвёртую стадию, и код каждой не изменится.

Когда потребитель уходит раньше

Пока потребитель дочитывает всё до конца, проблем нет. Теперь пусть он взял первое значение и вышел из цикла через break. sq пытается отправить второй квадрат и ждёт на out <- n * n вечно: читать некому. gen тоже ждёт на отправке следующего числа в sq, которая его больше не принимает. Две горутины зависли.

Чтобы стадия могла узнать «всё, заканчиваем», ей передают context и слушают ctx.Done() рядом с каждой операцией, на которой можно ждать:

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() и закрывает выход.

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)
}

[1 4 9 16 25 36]

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

В рабочем merge к каждому приёму и отправке добавляют ветку <-ctx.Done() по той же причине, что в конвейере: если потребитель ушёл, горутины-пересыльщики не должны зависнуть.

Пул воркеров (worker pool)

Fan-out с квадратами — это уже почти пул. Пул делает идею явной: фиксированное число воркеров разбирает задачи из общего канала и пишет результаты в общий канал.

Зачем фиксированное число? Если на каждую задачу запускать горутину, миллион задач превратится в миллион одновременных запросов к базе. Пул из восьми воркеров обработает тот же миллион, но одновременно к базе будет не больше восьми запросов. Размер пула — главный рычаг, которым вы ограничиваете нагрузку на зависимые системы.

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)
}

[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 заняты, отправка ждёт. После работы горутина забирает значение из канала (возвращает пропуск).

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)
}

лимит соблюдён: true

Горутин шесть, но в участке между «занять» и «вернуть» одновременно бывает не больше двух. Остальные стоят на sem <- struct{}{}: полный буфер и есть обратное давление. maxRunning на практике будет равен 2, но гарантирована только граница сверху, поэтому программа печатает проверку, а не число. Почему это корректно с точки зрения модели памяти, сказано в главе о ней: k-й приём из канала с буфером C happens-before завершения (k+C)-й отправки.

В рабочем коде вместо самодельного семафора часто берут golang.org/x/sync/semaphore. В нём ожидание пропуска принимает context и прерывается по отмене, а одна задача может занимать несколько пропусков сразу.

Батчинг

Писать в базу по одной записи дорого: каждая запись — отдельный запрос по сети. Батчер копит элементы и сохраняет их пачкой. Если ждать только полной пачки, можно застрять: пришло три элемента из ста, поток затих, и эти три не сохранятся никогда. Поэтому триггеров два: набралось N или прошло время, что наступит раньше. Оба живут в одном select.

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
}

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). В рабочем батчере есть ещё ветка <-ctx.Done(): при остановке сервиса он делает последний сброс и выходит, не теряя накопленное.

errgroup: параллельные задачи с ошибками

Частая задача: запустить N операций, дождаться всех, и если хотя бы одна упала, вернуть её ошибку, а остальные отменить. Вручную это WaitGroup, контекст и аккуратный сбор первой ошибки под мьютексом. Пакет golang.org/x/sync/errgroup делает это за вас:

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 не хранит результат дольше вызова; это делает кэш, и обычно их используют вместе.

Шардирование

Одна карта под одним мьютексом: горутины, работающие с разными ключами, всё равно ждут друг друга. Шардирование разбивает карту на несколько независимых частей (шардов), каждая со своим мьютексом. Номер шарда вычисляют из ключа: хеш-функция превращает строку в число, а остаток от деления на число шардов выбирает шард.

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))
	}
}

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. Оба ожидания должны слушать общий сигнал отмены. Каркас стадии:

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; если пока ждёт получателя — второй.

Что дальше

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

Как проверить, что после отмены действительно не осталось ни одной горутины, и как поймать гонку данных в таком графе, — в следующей главе, «Утечки горутин и гонки».

В задачах курса паттерны — это блок 3 (задачи 09–12: fan-in, пул с ошибками, многостадийный конвейер, батчер), блок 5 (лимитеры, singleflight, шардирование, обход графа) и весь блок 7 (десять вариантов пула). Тесты задач проверяют в первую очередь завершение и отсутствие утечек: ровно то, ради чего в примерах выше стоят ветки <-ctx.Done() и единственный владелец закрытия.