Задача:
Реализуй обобщённую функцию MapOrdered — ограниченный по параллелизму пул воркеров, который применяет функцию
fn к входным элементам и отдаёт результаты строго в порядке входа, при этом:- не теряет отмену
context.Context,- не допускает утечек горутин и зависаний,
- корректно обрабатывает
panic внутри fn,- поддерживает «раннее завершение» (как только достаточно результатов),
- обеспечивает backpressure (не раздувает буферы).
Сигнатуры:
type Result[R any] struct {
Value R
Err error
}
type Options struct {
Workers int // >0, число воркеров
MaxInFlight int // ≥ Workers, ограничение внутренних буферов
EarlyStopN int // если >0, остановиться после N успешных результатов
PanicAsError bool // если true, паники маппятся в error
TaskTimeout time.Duration // если >0, дедлайн на одну задачу
}
func MapOrdered[T any, R any](
ctx context.Context,
input <-chan T,
fn func(context.Context, T) (R, error),
opt Options,
) <-chan Result[R]
Требования и тонкости
- Строгий порядок.
Выходные элементы должны соответствовать порядку поступления во input. Параллелизм допустим, но публикация результата — строго по индексу.
- Отмена и завершение.
- При отмене ctx функция должна без утечек завершить все горутины и закрыть выходной канал.
- Если EarlyStopN > 0, как только выдали N успешных результатов — корректно останавливаем обработку оставшихся задач (не зависаем, не «подвешиваем» воркеров).
🟠Backpressure.
Никаких неограниченных буферов. Учитывай MaxInFlight, чтобы не переполнять память при медленном fn.
🟠Panic-handling.
Если PanicAsError=true, паники из fn перехватываются и превращаются в error. Если false — паника должна «пробить» наружу (но без гонок и утечек).
🟠Timeout на задачу.
При TaskTimeout>0 каждая задача исполняется с отдельным контекстом-дедлайном; таймаут — это ошибка задачи, а не общий стоп пула.
🟠Гарантия отсутствия утечек.
После закрытия input и/или отмены контекста пул завершает все свои горутины. Проверь runtime.NumGoroutine() до/после и убедись в стабильном числе.
🟠Без гонок.
Решение обязано проходить -race.
🟠Zero-copy по возможности.
Не копируй большие данные лишний раз; не «складывай» всё в память — обрабатывай потоково. Допустимы минимальные накладные структуры (индексы, слоты).
Подсказки (но не решение)
- Для сохранения порядка пригодится «кольцо результатов» или слайс «слотов» с публикацией по индексу и «ползунком» выдачи.
- Отдельно продумай: кто закрывает выходной канал и когда (все задачи обработаны, или ранний стоп).
- Аккуратно обращайся с контекстами: каждому fn — свой дочерний ctx (для таймаутов), общий ctx — для остановки пула.
- Не забудь про range-variable capture в горутинах.
- Паники оборачивай через recover, если включён PanicAsError.