Пайплайны под отменой: защищённые send, дисциплина закрытия и каноничная утечка горутин

Стадии и каналы — форма

Пайплайн — это горутины, связанные каналами: каждая стадия принимает значения сверху, преобразует и отправляет вниз. Версия из блога 2014 года использовала канал done на пайплайн; современная форма — context:

func parse(ctx context.Context, in <-chan []byte) <-chan Record {
	out := make(chan Record)
	go func() {
		defer close(out) // стадия владеет out — закрывает только отправитель
		for raw := range in {
			rec, err := decode(raw)
			if err != nil {
				continue // или: сообщить в канал ошибок, см. ниже
			}
			select {
			case out <- rec:
			case <-ctx.Done():
				return // бросаем отправку; defer закроет out
			}
		}
	}()
	return out
}

Композируемость дают два структурных факта. Стадия владеет своим выходным каналом: она его создаёт, только она отправляет, и defer close(out) сигнализирует вниз «значений больше не будет» — именно это позволяет следующей стадии использовать простой for range. И функция сразу возвращает канал только-для-чтения; пайплайн собран до того, как потекли данные.

Каноничная утечка: неприкрытый send

Уберите select — стадия станет читаться чище, out <- rec, — и станет неправильной. Отправка в канал блокируется, пока получатель не готов. В момент, когда нижняя стадия перестала читать — упала, вернулась раньше на плохом входе, как в Hook, или просто решила, что ей хватит, — эта отправка блокируется навсегда. Горутина припаркована на chan send, GC её не тронет (припаркованная горутина — корень, пришпиливающий её стек и каждую запись, на которую он ссылается), и нигде не всплывает никакая ошибка. Это самая частая реальная утечка горутин, и правило против неё механическое: каждая отправка в канал пайплайна сидит внутри select с веткой ctx.Done(). Приём через for range безопасен лишь потому, что defer close отправителя гарантирован — если стадия может выйти, не закрыв выход, нижние range текут точно так же.

Дисциплина закрытия и fan-in

Когда вы впервые собираете fan-in, вопрос «кто закрывает объединённый канал?» легко решить неправильно: каждый отправитель кажется кандидатом, но любой из них, закрыв канал раньше времени, вызовет панику у остальных. У правила закрытия нет исключений: канал закрывает только отправитель, потому что отправка в закрытый канал паникует и повторное закрытие паникует — получатель в принципе не может знать, что закрывать безопасно. Стадии с одним отправителем получают это бесплатно через defer close(out). Fan-in — N воркеров, шлющих в один общий канал, — место, где дисциплина обретает форму: закрывать нельзя, пока не закончили все отправители, поэтому их считают:

func merge(ctx context.Context, ins ...<-chan Record) <-chan Record {
	out := make(chan Record)
	var wg sync.WaitGroup
	for _, in := range ins {
		wg.Add(1)
		go func() { // по форвардеру на вход
			defer wg.Done()
			for rec := range in {
				select {
				case out <- rec:
				case <-ctx.Done():
					return
				}
			}
		}()
	}
	go func() { wg.Wait(); close(out) }() // WaitGroup, затем close — в отдельной горутине
	return out
}

Пара wg.Wait(); close(out) живёт в отдельной горутине, чтобы merge вернулся сразу — закрытие случается ровно один раз, ровно после выхода последнего форвардера, чем бы этот выход ни был: «вход дочитан» или «ctx отменён посреди send».

Осушать или бросать — кто кого разблокирует при отмене

Отмена разбирает пайплайн с обоих концов, и стоит знать, на какой механизм вы полагаетесь. Бросание — дефолт выше: каждый отправитель защищает свой send через ctx.Done, и при отмене каждая стадия разблокирует себя сама и возвращается; значения в полёте теряются. Осушение — альтернатива: потребитель продолжает for range до закрытия канала, верхние стадии завершаются естественно и ничего не теряется — ценой обработки (или хотя бы приёма) всего уже произведённого. Осушение без защищённых отправок — способ убедить себя, что защита необязательна; оно держится, лишь пока дисциплинирован каждый потребитель, что и сломал один ранний return из Hook. Надёжная стойка: защищайте каждый send в любом случае, а осушение выбирайте, только когда данные нельзя терять (и тогда сам цикл осушения нуждается в границе — счётчик или дедлайн, — иначе медленный производитель превращает остановку в зависание).

Буферы: что они покупают, честно

Сделать каналы стадий буферизованными — make(chan Record, 1024) — выглядит как фикс пропускной способности. Будьте точны в том, что это делает. Буфер поглощает джиттер: короткие всплески, паузы GC, на миг замедлившегося потребителя; производители движутся дальше вместо передачи в ногу, и латентность небуферизованного рандеву (~сотни наносекунд на пару send-receive) перестаёт доминировать на мелких элементах. Чего буфер не чинит — дисбаланс: если стадия производит 12 тыс./с, а следующая потребляет 9 тыс./с, буфер на 1024 слота полон через ~340 мс, и пайплайн идёт ровно на 9 тыс./с — как небуферизованный, плюс 1024 записи лишней памяти и устаревания. Установившаяся пропускная способность — это самая медленная стадия, точка. Буферы ещё и оттягивают симптомы утечки: неприкрытый send в большой буфер отказывает, только когда буфер заполнится, нередко через минуты после смерти потребителя — баг из 02:13 всплывает в 02:47 со стеком, который уже не указывает на причину.

Ошибки и наблюдаемость

Стадии, которая может отказать, нужен путь для ошибок. Два рабочих дизайна. Канал ошибок на стадию: стадия шлёт ошибки в errc, оркестратор делает select по нему — явно, но каждый потребитель обязан осушать errc, иначе вы построили вторую утечку. Стадии под errgroup — обычно лучший дефолт: каждая стадия бежит как g.Go, возвращает свою ошибку, а групповой ctx отменяет защищённые отправки всех остальных; g.Wait отдаёт первый отказ после того, как весь пайплайн размотался. Для видимости инструментируйте сами каналы: периодический gauge len(ch) против cap(ch) по стадиям показывает, где сидит очередь — буфер, прибитый к ёмкости, метит стадию-бутылочное-горлышко, буфер, прибитый к нулю, метит голодание выше по течению, а растущий runtime.NumGoroutine() при ровном трафике — сигнал утечки из прошлого урока. Дешёвые числа, превращающие «батч тормозит» в «стена — стадия 3».