merge с отменой и пайплайн
Миша отменил загрузку на середине большого пакета — как если бы клиент закрыл вкладку. Верхние стадии конвейера продолжили жить: их горутины висели в невидимой очереди.
наивный merge утекает, если читатель уходит раньше времени; done/ctx лишь сигналит передатчикам merge — они завершаются сами, а горутины, питающие входные каналы, обязаны слушать отмену отдельно.
Утечка наивного merge
func merge(chs ...<-chan *FileMeta) <-chan *FileMeta { /* наивная версия из L93 */ }
out := merge(a, b, c)
v := <-out
// читатель взял одно значение и ушёлПередатчики заблокированы на out <- v: никто больше не читает. Они висят вечно — это утечки горутин из. Наивный merge корректен, только если читатель дочитывает до конца.
Отмена в двух уровнях
func merge(done <-chan struct{}, chs ...<-chan *FileMeta) <-chan *FileMeta {
out := make(chan *FileMeta)
var wg sync.WaitGroup
for _, ch := range chs {
wg.Add(1)
go func(ch <-chan *FileMeta) {
defer wg.Done()
for {
select {
case v, ok := <-ch:
if !ok {
return
}
select {
case out <- v:
case <-done:
return
}
case <-done:
return
}
}
}(ch)
}
go func() {
wg.Wait()
close(out)
}()
return out
}Два уровня select: первый — «пришло значение или отмена», второй — «удалось отправить или отмена». Закрыл done — передатчики выходят, WaitGroup досчитывается, out закрывается. Горутины, питающие входные каналы, тоже должны уметь отменяться — сигнал через ctx или тот же done.
Пайплайн из стадий
Каждая стадия пайплайна — функция «канал в, канал наружу»: square читает числа и возвращает канал квадратов, sum — последняя стадия, читает и складывает. Отмена прокидывается через все стадии одним контекстом. Стадии независимы и проверяемы по отдельности.
Вы узнаете:
- как передаются ошибки по пайплайну
- кто закрывает выходной канал пайплайна
Наивный merge утекает без читателя; done/ctx в двух уровнях select сворачивают передатчики, но входные горутины должны слушать отмену сами; стадии пайплайна соединяются каналами и делят одну отмену.
10 000 ₽ · 60 минут · онлайн