Много сразу: all, forEach, validate, partition

Запускать файбер на каждого соседа руками и потом собирать их через join можно, ты это умеешь с урока 61. Но в девяти случаях из десяти задача одна и та же: сделать одну и ту же работу для каждого элемента списка, не дожидаясь каждого по очереди. Для этого у Effect есть готовое, и оно сразу решает три вопроса, которые руками легко забыть: сколько сразу, что с результатами и что с ошибками.

concurrency

Effect.forEach из EF4 делает работу для каждого элемента. По умолчанию по очереди. Второй аргумент настроек, concurrency, говорит, сколько сразу:

export const broadcast = (lines: readonly string[], limit: number): Effect.Effect<void, SendFailed, Wire> =>
  Effect.forEach(lines, (line) => Wire.use((wire) => wire.send(line)), {
    concurrency: limit,
    discard: true,
  });

Шесть строк, провод отвечает за двести миллисекунд:

limit 1: 1200 мс, одновременно 1
limit 3: 400 мс, одновременно 3
limit 6: 200 мс, одновременно 6

Число в concurrency это предел одновременности: не больше стольких файберов сразу, остальные ждут, пока какой-нибудь кончится. Есть ещё 'unbounded', все сразу, и 'inherit', «как скажет тот, кто меня позвал». discard: true говорит, что результаты не нужны: forEach не собирает массив и отдаёт void.

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

Effect.all то же самое для набора разных квитанций: массив, кортеж или объект. Effect.all({ book, journal }, { concurrency: 2 }) ждёт обе и отдаёт объект с теми же ключами.

Попробуй сам: три шага по очереди и все сразу.

Что с ошибками

forEach при первой же ошибке останавливается: прерывает тех, кто ещё работает, и падает этой ошибкой. Для отправки это правильно. Для разбора ночи нет: Рыдкулл хочет знать все битые строки, а не первую.

const night = ['мусор', 'T-3->T-1|1|Ping|', 'T-1->T-2|5|Smoke|'];

forEach: GarbledFrame
validate: [ 'GarbledFrame', 'UnknownTag' ]
partition: [ 1 ] [ 'GarbledFrame', 'UnknownTag' ]

Effect.validate проходит все элементы и копит ошибки. Если упала хоть одна, ошибка это непустой массив всех ошибок по порядку:

export const decodeStrict = (
  lines: readonly string[],
): Effect.Effect<Frame[], readonly [DecodeError, ...DecodeError[]]> =>
  Effect.validate(lines, (line) => decodeFrame(line));

Тип ошибки readonly [DecodeError, ...DecodeError[]] это массив, в котором хотя бы один элемент. Пустого массива ошибок не бывает: если ошибок нет, это успех. Та же мысль, что у валидации в kotel из урока 26, только с квитанциями.

Effect.partition не падает вообще. Она делит: что удалось, в одну кучу, что не удалось, в другую:

export const decodeSplit = (
  lines: readonly string[],
): Effect.Effect<[frames: Frame[], errors: DecodeError[]]> =>
  Effect.partition(lines, (line) => decodeFrame(line));

Порядок в паре: сначала удачи, потом ошибки. Это decodeAll из урока 42, только короче: там мы делали то же через Effect.result и два фильтра.

У обеих есть concurrency, как у forEach. Разбор строки быстрый, поэтому для него одновременность не нужна.

Под капотом

Внутри forEach с пределом заводит не файбер на каждый элемент, а столько работников, сколько разрешено. Каждый берёт следующий необработанный элемент, исполняет работу и кладёт результат в ячейку с номером элемента. Упал один, остальных прерывают, а из того, что успели положить, ничего не отдают. validate вместо прерывания запоминает ошибку и идёт дальше. Все работники это дети того, кто позвал forEach: структурная конкурентность из урока 62 работает и здесь.

Где тут дыра

await Effect.runPromise(
  Effect.forEach(neighbors, (to) => Wire.use((w) => w.send(to)), { concurrency: 'unbounded' }).pipe(
    Effect.provideService(Wire, wire),
  ),
);

unbounded на тысячу соседей, одновременно: 1000

На десяти соседях быстрее, это правда. На тысяче провод получил тысячу строк одновременно. Провод Магистрали не резиновый: он отвечает отказом, склад из EF6 выдаёт тысячу линий, а память башни держит тысячу ожидающих ответов. Быстрее не стало: стало «упало сразу всё».

Лечится пределом. Его подбирают по тому, сколько выдерживает самое узкое место: провод, склад, сосед. Если узкое место общее для разных частей программы, предел ставят не в forEach, а на само узкое место, семафором из урока 67.

Башня, этап 28

Башня учится делать много сразу. Рассылка не больше стольких-то строк одновременно, строгий разбор, который возвращает все ошибки, и мягкий, который делит кадры и ошибки.

Практика

Дальше

Рыдкулл раскладывает на столе схему Магистрали.

Гонки и время ждут в следующем уроке.