TaskGroup ​

Допустим, что страница профиля пользователя собирается из трёх источников: данные пользователя из базы, заказы из базы, отзывы из внешнего API. Источники не зависят друг от друга, значит, запрашивать их нужно конкурентно. Мы так уже умеем:

$user    = spawn(fetchUser(...), $userId);
$orders  = spawn(fetchOrders(...), $userId);
$reviews = spawn(fetchReviews(...), $userId);

$profile = new UserProfile(await($user), await($orders), await($reviews));

Для трёх задач терпимо. Но приглядитесь: это снова ручная бухгалтерия из главы про Scope, только теперь нужны ещё и результаты. Каждую корутину надо не забыть подождать, а если fetchUser бросит исключение, fetchOrders и fetchReviews продолжат работать вхолостую: отменять их придётся "руками" в catch.

А есть задачи, где ручной вариант становится по-настоящему болезненным. Например, взять результат первой успевшей корутины и отменить остальные. Попробуйте написать такое через await в цикле: получится клубок из проверок и отмен!

Scope здесь тоже не помощник: он управляет жизнью корутин, но ничего не знает про их результаты. Нужен примитив уровнем выше.

Группа задач ​

TaskGroup объединяет задачи в единое целое: запускает их в собственном Scope, хранит результаты и позволяет ждать группу целиком:

use Async\TaskGroup;

$group = new TaskGroup();

$group->spawnWithKey('user',    fn() => fetchUser($userId));
$group->spawnWithKey('orders',  fn() => fetchOrders($userId));
$group->spawnWithKey('reviews', fn() => fetchReviews($userId));

$data = $group->all()->await();

$profile = new UserProfile($data['user'], $data['orders'], $data['reviews']);

Метод all() возвращает знакомый Future, который разрешится массивом результатов, когда завершатся все задачи. Ключи мы задали сами через spawnWithKey, поэтому в массиве не безликие индексы, а имена.

И раз это Future, ограничение по времени достаётся бесплатно, тем же токеном, что и всегда:

$data = $group->all()->await(timeout(5000));

Если хотя бы одна задача бросит исключение, группа поведёт себя как Scope из восьмой главы: остальные задачи отменятся, а await выбросит CompositeException со всеми ошибками внутри. Группа либо собрала всё, либо не собрала ничего, промежуточных состояний нет.

Первый успевший: race ​

all() — только одна из стратегий ожидания. Вспомним GeoDirectory: у него есть три реплики, и иногда какая-нибудь тормозит. Классический приём: отправить запрос на все реплики и взять первый ответ:

$group = new TaskGroup();

foreach (['geo-1', 'geo-2', 'geo-3'] as $host) {
    $group->spawn(fn() => checkAddressAt($host, $address));
}

$verdict = $group->race()->await();

race() разрешается результатом первой завершившейся задачи, неважно, успехом или ошибкой. Тот самый сценарий «взять первый и не ждать остальных», который так мучительно писать вручную.

Первый успешный: any ​

У race() есть жёсткий характер: если первой финиширует упавшая задача, вы получите её исключение. Иногда нужно мягче: перебрать несколько провайдеров и взять первый успешный ответ, а на ошибки закрыть глаза:

$group = new TaskGroup();

$group->spawn(fn() => geocodeViaGoogle($address));
$group->spawn(fn() => geocodeViaOsm($address));
$group->spawn(fn() => geocodeViaYandex($address));

$coords = $group->any()->await();
$group->suppressErrors();

any() игнорирует упавших и возвращает первого победителя. Исключение вы получите, только если упали все, и это будет CompositeException с полным списком причин. Обратите внимание на suppressErrors(): ошибки проигравших провайдеров никто не обработал, и группа хочет явного подтверждения, что так и задумано. Знакомый принцип из главы про исключения: ошибка не может исчезнуть молча.

Лимит конкурентности ​

А теперь неожиданное. Помните пул воркеров из главы про каналы: канал, десять корутин, цикл recv? TaskGroup умеет то же самое в несколько строк:

$group = new TaskGroup(concurrency: 10);

while (($row = fgetcsv($handle)) !== false) {
    $group->spawn(fn() => checkAddress($row[$addressIndex]));
}

$group->close();

foreach ($group as $key => [$result, $error]) {
    // результаты приходят по мере готовности
}

Параметр concurrency: 10 ограничивает число одновременно работающих задач: остальные ждут в очереди и даже не создают корутину, пока не освободится слот. close() играет ту же роль, что и у канала: объявляет, что новых задач не будет. А foreach отдаёт результаты по мере их готовности, не дожидаясь конца всей группы.

Означает ли это, что канал был не нужен? Нет. Канал — примитив синхронизации, из которого можно собрать что угодно. TaskGroup — готовая сборка для самого частого случая «выполнить набор задач и получить результаты». Когда задача укладывается в этот шаблон, берите TaskGroup; когда нужна нестандартная топология, в руках всегда остаются каналы и Scope.

В сухом остатке TaskGroup — это Scope плюс результаты. Группа задач превращается в единое значение, у которого можно попросить всё сразу, первого успевшего или первого успешного.

И одна деталь напоследок: TaskGroup бережно хранит все результаты. Вызовите race() дважды, и оба раза получите один и тот же ответ. Обойдите группу через foreach повторно, и она снова отдаст всё с начала. Для страницы профиля это удобно. А теперь представьте конвейер, который прогоняет через группу сто тысяч задач. Всё, что группа помнит, живёт в памяти. Чувствуете подвох? О нём и поговорим в следующей главе.