Паттерны конкурентности в Go: пул воркеров, конвейер, fan-in

Что здесь на самом деле спрашивают

Лестница почти всегда такая:

1. «Напишите пул воркеров» — разминка; код пишут почти все, и вопросы начинаются после него.
2. «Сколько воркеров взять?» — здесь проверяют, отличает ли кандидат задачу для процессора от задачи, которая ждёт.
3. «Результаты придут в том же порядке, что задачи?»
4. «Кто и когда закрывает канал результатов?» — и что будет, если не закрыть или закрыть не там.
5. «Слейте N каналов в один» — fan-in.
6. «Как ограничить число одновременных запросов к базе?» — семафор.
7. «Потребитель ушёл раньше — что станет с горутинами конвейера?» — отмена и утечки; дальше — «как остановить всё по первой ошибке».

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

База: четыре схемы из горутин и каналов

Схемы собираются из четырёх деталей, и для урока хватит одной фразы о каждой. range ch получает значения из канала одно за другим. select ждёт, пока сможет выполниться хотя бы одна из перечисленных в нём операций с каналами, и выполняет одну из готовых. sync.WaitGroup — счётчик: Add(n) прибавляет, Done() вычитает единицу, Wait() ждёт нуля. ctx.Done() — канал, который закрывается, когда вызвали cancel() контекста. Подробно — в уроках о каналах, о sync и atomic и о context.

На всех четырёх схемах держится одно правило спецификации:

И то, как это видит получатель, — range по каналу:

Сам по себе range по каналу заканчивается только закрытием. Отсюда вся механика: горутина, которая читает range, выйдет, когда кто-то закроет её вход, — а значит, кто-то обязан это сделать. Вот пул воркеров, на котором в прогоне сняты разделы 1 и 2.

Фрагмент bench/gopattern/internals.go:

func pool(workers int, jobs []uint64, work func(uint64) uint64) []uint64 {
    in := make(chan int)
    out := make(chan result)
    var wg sync.WaitGroup
    for w := 0; w < workers; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for i := range in {
                out <- result{i, work(jobs[i])}
            }
        }()
    }
    go func() {
        for i := range jobs {
            in <- i
        }
        close(in)
    }()
    go func() {
        wg.Wait()
        close(out)
    }()
    res := make([]uint64, len(jobs))
    for r := range out {
        res[r.idx] = r.val
    }
    return res
}

Горутин здесь три вида, и у каждого канала ровно один закрывающий. Канал задач in закрывает раздатчик — он один знает, что задач больше нет. Канал результатов out пишут N воркеров, и ни один из них не знает, последний ли он, — поэтому его закрывает отдельная горутина, дождавшаяся wg.Wait(). Читает результаты вызывающий, и его range закончится, когда закроется out.

Остальные три схемы — из тех же деталей:

Fan-out — это раздача одного канала нескольким читателям; пул воркеров и есть fan-out, поэтому отдельной строки у него нет.

Этого достаточно для ответа уровня Junior: пул воркеров — N горутин, которые читают задачи из одного канала и пишут результаты в другой. Канал задач закрывает тот, кто их раздаёт; канал результатов — отдельная горутина после wg.Wait(), потому что ни один воркер не знает, последний ли он.

Механизм 1: сколько воркеров — зависит от того, чем занята задача

Второй вопрос лестницы, и на нём ошибаются чаще всего: «чем больше воркеров, тем быстрее». Документация пакета runtime ставит потолок:

Пул выше, 16 задач — цикл на процессоре без памяти и без ожиданий. Машина с двумя ядрами; каждая конфигурация прогнана 15 раз вперемешку с остальными.

Вывод bench/gopattern/internals.go, раздел 1:

1) ПУЛ ВОРКЕРОВ: ЗАДАЧА ДЛЯ ПРОЦЕССОРА
──────────────────────────────────────
  16 задач, каждая — 3 000 000 шагов xorshift; лучший из 15 кругов вперемешку
  GOMAXPROCS  воркеров  время     ускорение к 1 воркеру  ядер занято (CPU/время)
       2          1     102 мс    1.0                    1.0
       2          2     52 мс     2.0                    2.0
       2          4     53 мс     1.9                    1.9
       2          8     53 мс     1.9                    2.0
       1          1     104 мс    1.0                    1.0
       1          8     102 мс    1.0                    1.0

Два воркера — вдвое быстрее одного, а дальше прибавки нет: при восьми воркерах в каждый момент работают те же два, остальные стоят в очереди планировщика. Нижние две строки — та же проверка с другой стороны: при GOMAXPROCS=1 восемь воркеров не быстрее одного. Выше двух столбец «ядер занято» на двух ядрах подняться и не может. Что потолок ставит именно GOMAXPROCS, видно по нижним строкам: при GOMAXPROCS=1 восемь воркеров заняли одно ядро (1,0), хотя свободных было два.

Устроено это через P — логический процессор планировщика. Комментарий в начале runtime/proc.go (строка 28 на go1.24.7) определяет его так: P - processor, a resource that is required to execute Go code (P — процессор, ресурс, без которого нельзя выполнять код Go). Сколько P заводить, рантайм при старте берёт из GOMAXPROCS (строки 876–880, вызов procresize(procs)). Поток без P Go-код не выполняет, поэтому воркеров больше, чем P, просто ждут своей очереди. Подробно — в уроке о планировщике.

Теперь та же схема, но задача — time.Sleep(20 мс), как ожидание диска или сети.

Вывод bench/gopattern/internals.go, раздел 2:

2) ПУЛ ВОРКЕРОВ: ЗАДАЧА СО СНОМ
───────────────────────────────
  16 задач, каждая — time.Sleep(20 мс); лучший из 3 кругов
  воркеров   время     ускорение к 1 воркеру
      1      324 мс    1.0
      2      162 мс    2.0
      4      81 мс     4.0
      8      41 мс     8.0
     16      20 мс     16.0

Спящая горутина ядро не занимает, и в этом прогоне ускорение росло с числом воркеров, пока их не стало столько же, сколько задач. Отсюда ответ на «сколько воркеров»: для задачи на процессоре — порядка runtime.GOMAXPROCS(0): больше одновременно не выполняется, и в нашем замере больше не ускоряло; для задачи, которая ждёт, — столько, сколько выдержит то, чего она ждёт: база, чужой API, диск. Точное число для своей задачи определяют замером. Ядра здесь не ограничение, ограничение — внешний ресурс, и его ограничивают семафором (Механизм 4).

Результаты приходят в порядке готовности

Третий вопрос. Четыре задачи a b c d работают 80, 20, 60 и 40 мс, воркеров четыре — каждой задаче достался свой.

Вывод bench/gopattern/internals.go, раздел 3:

3) ПОРЯДОК РЕЗУЛЬТАТОВ
──────────────────────
  задачи a b c d, работают 80, 20, 60, 40 мс; 4 воркера
  в порядке прихода:        [b d c a]
  res[i] по индексу задачи: [A B C D]
  порядок прихода совпал с первым в 20 повторах из 20

Из канала результатов значения выходят в том порядке, в каком воркеры их отправили, — то есть в порядке готовности, а не задач. Здесь длительности разнесены на 20 мс, и в этом прогоне порядок совпал во всех 20 повторах. Но это не гарантия: порядок прихода определяется тем, какой воркер раньше успел отправить, и на загруженной машине задача на 40 мс может прийти раньше задачи на 20 мс. Порядок задач возвращает индекс: воркер отправляет не только ответ, но и номер задачи, а получатель кладёт ответ в res[i]. Так сделан pool выше — result{i, work(jobs[i])} и res[r.idx] = r.val. Записи в разные элементы среза из одной горутины-получателя гонки не создают: пишет их только она.

Механизм 2: конвейер держится на close

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

Фрагмент bench/gopattern/internals.go:

func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}
 
func sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

Вывод bench/gopattern/internals.go, раздел 4:

4) КОНВЕЙЕР: GEN → SQ → ПОТРЕБИТЕЛЬ
───────────────────────────────────
  потребитель получил: [1 4 9 16 25]
  range закончился: стадия gen закрыла свой выход, range в sq
  вышел, sq закрыла свой — сигнал «больше не будет» прошёл по цепочке
  горутин после конвейера: +0

Теперь sq забыла close(out) — одна строка. Программа cases/forgotclose: те же gen и sq, и main читает выход sq через range.

Вывод bench/gopattern/internals.go, раздел 5; cases/forgotclose падает намеренно, исправление — sq с close(out) выше:

5) СТАДИЯ ЗАБЫЛА CLOSE: ВСЕ СПЯТ
────────────────────────────────
  1 4 9 16 25 fatal error: all goroutines are asleep - deadlock!
  goroutine <N> [chan receive]:
  main.main()
      <cases>/forgotclose/main.go:37
  код выхода: 2

Все пять значений потребитель получил. Потом gen закрыла свой выход, range в sq вышел, sq вернулась — а её выход остался открытым. main ждёт шестого значения в range, которого не будет, и сказать ей «больше не будет» уже некому. Других горутин нет, и рантайм роняет программу. Стек показывает, где ждали: [chan receive] в строке range.

Ту же fatal error даёт ошибка, которую делают в пуле воркеров: wg.Wait() прямо в main, до чтения результатов (cases/waitbeforeread, три воркера, шесть задач, канал результатов без буфера).

Вывод bench/gopattern/internals.go, раздел 6; cases/waitbeforeread падает намеренно, исправление — pool в «Базе»:

6) WG.WAIT ДО ЧТЕНИЯ РЕЗУЛЬТАТОВ
────────────────────────────────
  fatal error: all goroutines are asleep - deadlock!
  goroutine <N> [sync.WaitGroup.Wait]:
  goroutine <N> [chan send]:
  goroutine <N> [chan send]:
  goroutine <N> [chan send]:
  goroutine <N> [chan send]:
  код выхода: 2

Отчёт рантайма здесь читается как схема взаимного ожидания: main стоит в [sync.WaitGroup.Wait] и ждёт воркеров, три воркера стоят в [chan send] и ждут, когда main начнёт читать, а раздатчик стоит в [chan send] с задачами, которые некому взять. Лечится это тем, что сделано в pool: wg.Wait() и close(out) — в отдельной горутине, а читает вызывающий сразу.

Буфер у канала результатов ошибку не лечит, а прячет. Та же программа с буфером на 6 (cases/waitbuffered).

Вывод bench/gopattern/internals.go, раздел 6, продолжение:

  cases/waitbuffered — та же ошибка, но буфер результатов на 6:
  задач 6: задач: 6 сумма квадратов: 91
  код выхода: 0
  задач 7: fatal error: all goroutines are asleep - deadlock!
  код выхода: 2

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

Буфер сам по себе не ошибка. Он — корректный дизайн, когда число результатов известно заранее и ёмкость выведена из того же числа: make(chan result, len(jobs)) при одном результате на задачу. Тогда каждая отправка находит место — спецификация разрешает отправку в буферизованный канал, если в буфере есть место, — и wg.Wait() до чтения не заблокирует. Ошибка — константа, не связанная с числом задач, как 6 в этом примере.

Почему забытый close в настоящем сервисе обычно не падает, а висит, — разобрано в разделе «Глубже».

Механизм 3: fan-in — закрыть выход может только тот, кто дождался всех

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

Фрагмент bench/gopattern/internals.go:

func merge(cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, c := range cs {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }()
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

Вывод bench/gopattern/internals.go, раздел 8:

8) FAN-IN: СЛИЯНИЕ КАНАЛОВ
──────────────────────────
  три источника по 4 значения: получено 12, сумма 78
  range закончился: close(out) стоит после wg.Wait()
  горутин после слияния: +0

Две детали этого кода спрашивают отдельно. wg.Add(1) стоит до go, а не внутри горутины — документация WaitGroup.Add говорит это прямо:

Иначе Wait может увидеть ноль раньше, чем горутина успеет сделать Add. И переменная цикла c у каждой горутины своя — с Go 1.22 каждая итерация for заводит новую переменную; почему это важно для функции, запущенной в цикле, разобрано в уроке о замыканиях.

Что будет, если close(out) поставить сразу, без wg.Wait() (cases/earlyclose; источники отвечают через 50 мс, как сеть).

Вывод bench/gopattern/internals.go, раздел 8, продолжение; cases/earlyclose падает намеренно, исправление — merge выше:

  cases/earlyclose — close(out) сразу, без wg.Wait():
  получено значений: 0
  panic: send on closed channel
  goroutine <N> [running]:
  main.merge.func1()
      <cases>/earlyclose/main.go:36
  created by main.merge in goroutine <N>
      <cases>/earlyclose/main.go:33
  код выхода: 2

Получатель увидел закрытый канал и вышел из range с нулём значений. А первая же пересылка (строка 36 — out <- v) упала: спецификация называет отправку в закрытый канал паникой во время выполнения. Это другая поломка, чем забытый close: забытый вешает получателя, ранний роняет отправителя и теряет данные.

С go1.25 у WaitGroup есть метод, который ставит Add на место сам:

В исходнике go1.25.1 у метода пять строк тела: wg.Add(1), затем горутина с defer wg.Done() и вызовом f() (sync/waitgroup.go, строки 235–241). На go1.24.7 этого метода нет — разница по версиям собрана в «Истории версий».

Механизм 4: семафор — где занимать место, решает, сколько будет горутин

Задача со сном из Механизма 1 ускорялась с числом воркеров, но в жизни «сон» — это запрос к базе, и база выдержит не всякое число одновременных запросов. Ограничивают это буферизованным каналом: занять место — отправить в него, освободить — прочитать. Отправка в полный буферизованный канал ждёт, поэтому одновременно внутри окажется не больше, чем его ёмкость.

Где именно занимать место — вопрос не стиля. Первая расстановка — горутина на задачу, место занимается внутри. Фрагмент bench/gopattern/semaphore.go (c — счётчики живых и работающих обработчиков):

for i := 0; i < tasks; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        c.enter()
        defer c.leave()
        sem <- struct{}{}
        defer func() { <-sem }()
        c.work()
    }()
}

Вывод bench/gopattern/internals.go, раздел 9 — та же расстановка, рядом вариант без семафора:

9) СЕМАФОР НА БУФЕРИЗОВАННОМ КАНАЛЕ
───────────────────────────────────
  12 горутин, каждая спит 20 мс; лучший по времени из 3 кругов
  семафор          одновременно работали   горутин было   время
  make(chan, 3)    3                       12             81 мс
  нет              12                      12             20 мс

Наблюдаемый максимум одновременно работающих — ровно ёмкость, 3; время в этом прогоне выросло вчетверо, потому что двенадцать задач прошли четырьмя волнами по три. Но горутин было 12: при такой расстановке все созданы сразу, и девять стоят на sem <- struct{}{}, каждая со своим стеком.

Вторая расстановка — место занимается до go, освобождается в defer запущенной горутины. Фрагмент того же файла:

for i := 0; i < tasks; i++ {
    sem <- struct{}{}
    wg.Add(1)
    go func() {
        defer wg.Done()
        defer func() { <-sem }()
        c.enter()
        defer c.leave()
        c.work()
    }()
}

Вывод bench/gopattern/semaphore.go, разделы 1 и 2; горутины считаются своим счётчиком живых обработчиков, а не runtime.NumGoroutine(), в котором есть и служебные горутины:

1) ГОРУТИНА НА ЗАДАЧУ, МЕСТО ЗАНИМАЕТСЯ ВНУТРИ
──────────────────────────────────────────────
  задач 12, ёмкость 3: обработчиков жило одновременно 12, работало 3

2) МЕСТО ЗАНИМАЕТСЯ ДО GO
─────────────────────────
  задач 12, ёмкость 3: обработчиков жило одновременно 3, работало 3

Теперь цикл раздачи сам встаёт на отправке в полный канал, и новая горутина не появляется, пока не освободится место: живых обработчиков не больше трёх. Если у раздачи есть context, ожидание места тоже должно слушать отмену — иначе цикл простоит на sem <- struct{}{} сколько угодно. Фрагмент функции acquireWithCancel из того же файла:

select {
case sem <- struct{}{}:
case <-ctx.Done():
    err = ctx.Err()
}
if err != nil {
    break
}

Вывод bench/gopattern/semaphore.go, раздел 3:

3) МЕСТО ДО GO, ОЖИДАНИЕ МЕСТА СЛУШАЕТ CTX.DONE()
─────────────────────────────────────────────────
  запущено обработчиков: 3 из 12
  раздача остановилась с ошибкой: context canceled (errors.Is context.Canceled: true)

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

Отсюда сравнение трёх схем по тому, что в них на самом деле разное:

Ограничить число горутин умеют и пул, и семафор с местом до go; различие в том, живёт ли горутина дольше одной задачи и где копится очередь. Схема «место внутри» на миллион задач создаст миллион горутин — это свойство расстановки, а не семафора.

Механизм 5: остановка — потребитель ушёл раньше, или случилась ошибка

Последний вопрос лестницы. Источник countFrom бесконечный — как чтение из сети или из очереди, — а потребитель takeThree берёт три значения через range, делает break и возвращается.

Вывод bench/gopattern/internals.go, раздел 10:

10) ПОТРЕБИТЕЛЬ УШЁЛ РАНЬШЕ: БЕЗ CONTEXT
────────────────────────────────────────
  takeThree(sq(countFrom())) -> [1 4 9]
  горутин после: +2
    countFrom: [chan send]
    sq: [chan send]

Программа не упала и ответ правильный — а две горутины остались навсегда: обе стоят на отправке, которую никто не примет. fatal error здесь нет, потому что main жива и работает. Сборщик мусора такие горутины не собирает; почему — и чем такие утечки ловят — разобрано в уроке о каналах, Механизм 5. Здесь важно, что счёт runtime.NumGoroutine() до и после показывает утечку. Для теста этого мало: в программе бывают служебные горутины, и снимок счётчика может случайно совпасть. Надёжнее ждать сигнала завершения — об этом ниже и в «Дальше спросят».

Первая попытка лечения — дать отправке второй выход, отмену. Фрагмент bench/gopattern/internals.go (в cancel.go та же функция называется sqRange):

func sqCtx(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case out <- n * n:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

Вывод bench/gopattern/internals.go, раздел 11 — в связке с источником countFromCtx, который тоже слушает отмену:

11) ПОТРЕБИТЕЛЬ УШЁЛ РАНЬШЕ: С CONTEXT
──────────────────────────────────────
  takeThree(sqCtx(ctx, countFromCtx(ctx))) -> [1 4 9]
  горутин после defer cancel(): +0

В этой связке всё сошлось: по defer cancel() источник вышел и закрыл свой выход, range в стадии кончился. Но у стадии две точки ожидания — получение из входа и отправка в выход, — а ctx.Done() слушает только вторая. range по входу отмену не слышит. bench/gopattern/cancel.go проверяет обе версии стадии на четырёх сценариях, и завершение определяет не счётом горутин, а сигналом: закрылся ли выход стадии за 500 мс.

Вывод bench/gopattern/cancel.go, раздел 1:

1) СТАДИЯ sqRange: RANGE ПО ВХОДУ, SELECT ТОЛЬКО НА ОТПРАВКЕ
────────────────────────────────────────────────────────────
  обычный поток 1, 2, 3:            получено [1 4 9], выход закрыт: да
  вход закрыт и пуст:               выход закрыт: да
  отмена, вход открыт и пуст:       выход закрыт за 500 мс: НЕТ
  отмена, вход nil:                 выход закрыт за 500 мс: НЕТ
  отмена, выход никто не читает:    выход закрыт за 500 мс: да

Если верхняя стадия ничего не отправляет и не закрывает канал, стадия с range на отмену не реагирует: cancel() — это сигнал, а не принудительная остановка, и горутина выходит, только если в момент ожидания слушает ctx.Done(). Исправление — select с отменой в обеих точках. Фрагмент bench/gopattern/cancel.go:

func sqCtx(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for {
            var n int
            var ok bool
            select {
            case <-ctx.Done():
                return
            case n, ok = <-in:
                if !ok {
                    return
                }
            }
            select {
            case <-ctx.Done():
                return
            case out <- n * n:
            }
        }
    }()
    return out
}

Вывод bench/gopattern/cancel.go, раздел 2:

2) СТАДИЯ sqCtx: SELECT И НА ПОЛУЧЕНИИ, И НА ОТПРАВКЕ
─────────────────────────────────────────────────────
  обычный поток 1, 2, 3:            получено [1 4 9], выход закрыт: да
  вход закрыт и пуст:               выход закрыт: да
  отмена, вход открыт и пуст:       выход закрыт за 500 мс: да
  отмена, вход nil:                 выход закрыт за 500 мс: да
  отмена, выход никто не читает:    выход закрыт за 500 мс: да

Все четыре сценария закончились закрытым выходом. Две оговорки. Приоритета у отмены нет: если одновременно готовы и отправка, и ctx.Done(), select выбирает одну из готовых веток, поэтому после cancel() стадия может успеть отправить ещё одно значение. И вход стадия не закрывает — он принадлежит тому, кто в него пишет. Как cancel закрывает Done и почему его зовут всегда, а не только при отмене, — в уроке о context.

Всё вместе помещается в короткую программу. Полный файл bench/gopattern/cases/pipeline/main.go — его можно скопировать в пустой каталог рядом с go.mod (module example, go 1.24) и запустить командой go run .:

package main
 
import (
	"context"
	"fmt"
)
 
func count(ctx context.Context) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := 1; ; n++ {
			select {
			case <-ctx.Done():
				return
			case out <- n:
			}
		}
	}()
	return out
}
 
func sq(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for {
			var n int
			var ok bool
			select {
			case <-ctx.Done():
				return
			case n, ok = <-in:
				if !ok {
					return
				}
			}
			select {
			case <-ctx.Done():
				return
			case out <- n * n:
			}
		}
	}()
	return out
}
 
func main() {
	ctx, cancel := context.WithCancel(context.Background())
	out := sq(ctx, count(ctx))
	for i := 0; i < 3; i++ {
		fmt.Println(<-out)
	}
	cancel()
	for range out { // drain: sq may still send one more value
	}
	fmt.Println("sq closed its output: pipeline stopped")
}

Вывод (runs/pipeline.txt, go1.24.7; на go1.25.1 тот же):

1
4
9
sq closed its output: pipeline stopped

Последняя строка печатается только после того, как range по выходу sq закончился, — то есть после того, как sq закрыла выход. Это и есть проверка остановки сигналом, а не счётом.

Остановка по первой ошибке — то же самое, только отменяет не потребитель, а стадия, у которой случилась ошибка. На стандартной библиотеке это около тридцати строк. Ниже — минимальный аналог errgroup.WithContext из golang.org/x/sync: сам модуль в стандартную библиотеку не входит, и здесь он не подключался, поэтому код — свой.

Фрагмент bench/gopattern/internals.go:

type group struct {
    wg     sync.WaitGroup
    once   sync.Once
    err    error
    cancel context.CancelCauseFunc
}
 
func withContext(parent context.Context) (*group, context.Context) {
    ctx, cancel := context.WithCancelCause(parent)
    return &group{cancel: cancel}, ctx
}
 
func (g *group) Go(f func() error) {
    g.wg.Add(1)
    go func() {
        defer g.wg.Done()
        if err := f(); err != nil {
            g.once.Do(func() {
                g.err = err
                g.cancel(err)
            })
        }
    }()
}
 
func (g *group) Wait() error {
    g.wg.Wait()
    g.cancel(g.err)
    return g.err
}

Пять горутин: источник, который на четвёртой записи возвращает ошибку, три обработчика и одна долгая работа на 10 секунд — все слушают ctx.Done().

Вывод bench/gopattern/internals.go, раздел 12:

12) ОСТАНОВКА ПО ПЕРВОЙ ОШИБКЕ
──────────────────────────────
  Wait вернул: запись 4: плохая запись
  errors.Is(err, errBad): true
  context.Cause(ctx): запись 4: плохая запись
  долгая работа на 10 с прервана: true (Wait занял меньше секунды: true)
  горутин после: +0

Первая ошибка отменила общий контекст, долгая работа вышла по ctx.Done(), не дожидаясь своих десяти секунд, и Wait вернул именно эту ошибку — обёрнутую, так что errors.Is её находит. context.Cause отдаёт ту же причину любому, у кого есть ctx. sync.Once здесь нужен затем, чтобы вторая ошибка не перезаписала первую.

Глубже: почему забытый close роняет маленькую программу и вешает настоящую

В Механизме 2 забытый close дал fatal error. Это удобный ответ на собеседовании, но опасная привычка: кажется, что такую ошибку видно сразу. Прогон проверяет ту же стадию sq без close в программе, где жив один таймер — time.After в соседнем select.

Вывод bench/gopattern/internals.go, раздел 7:

7) СТАДИЯ ЗАБЫЛА CLOSE, НО В ПРОГРАММЕ ЕСТЬ ТАЙМЕР
──────────────────────────────────────────────────
  через 500 мс потребитель всё ещё в range; fatal error нет
  горутин висит: +1
    потребитель, запущенный из main.main, [chan receive]

Никакого падения: потребитель стоит в [chan receive], и это просто одна лишняя горутина. Почему — видно в исходнике checkdead, функции, которая печатает эту ошибку. В runtime/proc.go на go1.24.7 она начинается в строке 6026, и до вызова fatal в строке 6127 у неё несколько ранних выходов; здесь важны два. Первый — в строках 6055–6058: если хоть один поток выполняет код, проверка заканчивается (if run > run0 { return }). Второй — в строках 6119–6124.

Фрагмент runtime/proc.go, строки 6119–6124 на go1.24.7:

// There are no goroutines running, so we can look at the P's.
for _, pp := range allp {
    if len(pp.timers.heap) > 0 {
        return
    }
}

Если на каком-либо P стоит таймер, проверка заканчивается, не дойдя до fatal. all goroutines are asleep означает буквально «спят все, и таймеров нет». В сервисе, где тикает таймер метрик, ждёт любой time.After или хоть одна горутина занята делом, это условие почти не наступает — забытый close там не роняет процесс, а копит висящие горутины, по одной на каждый такой конвейер. Ловят это счётом горутин, как в Механизме 5, а не ожиданием падения.

Как отвечать на собеседовании

Короткий ответ: пул воркеров — N горутин, которые читают задачи из одного канала и пишут результаты в другой; канал задач закрывает раздатчик, канал результатов — отдельная горутина после wg.Wait(). Для задачи на процессоре одновременно работают не больше GOMAXPROCS воркеров, поэтому прибавки сверх этого ждать неоткуда; для задачи, которая ждёт, число ограничивает внешний ресурс, и его ограничивают семафором. Конвейер — стадии через каналы, каждая закрывает свой выход; чтобы стадии не утекали, когда потребитель ушёл, каждое ожидание стадии — и получение, и отправка — стоит в select с ctx.Done().

Этого достаточно, чтобы ответить верно. Дальше — то, что добавляют, если собеседник копает.

Если интервьюер копает глубже

На «сколько воркеров» — две строки прогона. «На машине с двумя ядрами два воркера дали около 2,0, а восемь — не больше: одновременно Go-код выполняют не больше GOMAXPROCS потоков. На задаче со сном восемь воркеров — 8,0: спящая горутина ядро не занимает. На своей задаче это меряют».

На «в каком порядке результаты» — в порядке готовности. «Задачи a b c d пришли как [b d c a]. Порядок возвращают индексом: воркер отправляет номер задачи вместе с ответом, получатель кладёт в res[i]».

На «кто закрывает канал» — тот, кто знает, что отправок больше не будет. «Если отправителей несколько — отдельная горутина после wg.Wait(). Не закрыть — получатель висит; закрыть раньше времени — panic: send on closed channel».

На «что будет, если забыть close» — не всегда fatal error. «Рантайм роняет программу, только если спят все и нет таймеров — так написан checkdead. В сервисе это просто висящая горутина».

На «почему горутины остались после cancel» — точки ожидания. «cancel() — сигнал, а не остановка. Стадия с range по входу на открытом пустом входе отмену не услышала; с select и на получении, и на отправке — вышла во всех четырёх сценариях».

На «как остановить по ошибке» — общий контекст. «Первая ошибка вызывает cancel(err), все стадии слушают ctx.Done(), Wait возвращает первую ошибку. Это около тридцати строк на WaitGroup, Once и context.WithCancelCause».

Дальше спросят

  задач 6: задач: 6 сумма квадратов: 91
  код выхода: 0
  задач 7: fatal error: all goroutines are asleep - deadlock!
  код выхода: 2

func closedWithin(out <-chan int, d time.Duration) ([]int, bool) {
	var got []int
	timeout := time.After(d)
	for {
		select {
		case v, ok := <-out:
			if !ok {
				return got, true
			}
			got = append(got, v)
		case <-timeout:
			return got, false
		}
	}
}

  wggo/main.go:18:6: sync.Go requires go1.25 or later (module is go1.24)

Частые заблуждения

История версий

Урок прогонялся на go1.24.7 и go1.25.1 (linux/amd64, 2 ядра); документация и исходники сверены 14.10.2026. К этой дате вышли Go 1.26 и Go 1.27 — на них урок не прогонялся, и строк о них в таблице нет. Семантика close, range и select — спецификация языка; от версии в этом уроке зависит только WaitGroup.Go.

Практика

Две задачи. Сначала ответьте, потом сверьтесь с настоящим выводом: в обеих правильный ответ взят из прогона скрипта, а не назначен.

before := runtime.NumGoroutine()
in := sq(gen())
fmt.Println(<-in, <-in, <-in)
time.Sleep(50 * time.Millisecond)
fmt.Println("goroutines left:", runtime.NumGoroutine()-before)

sem := make(chan struct{}, 2)
var now, peak atomic.Int64
var wg sync.WaitGroup
for i := 0; i < 6; i++ {
wg.Add(1)
go func() {
	defer wg.Done()
	sem <- struct{}{}
	n := now.Add(1)
	for {
		m := peak.Load()
		if n <= m || peak.CompareAndSwap(m, n) {
			break
		}
	}
	time.Sleep(10 * time.Millisecond)
	now.Add(-1)
	<-sem
}()
}
wg.Wait()
fmt.Println("ran at once:", peak.Load())

1 4 9
goroutines left: 2
ran at once: 2

1 4 9
goroutines left: 0
ran at once: 2

1 4 9
goroutines left: 2
ran at once: 6

1 4 9
goroutines left: 0
ran at once: 6

Проверка знаний