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 ще раз, і вона знову роздасть усе від початку. Для сторінки профілю це зручно. А тепер уявіть конвеєр, що пропускає крізь групу сто тисяч задач. Усе, що група пам'ятає, живе в пам'яті. Бачите пастку? Ось про це ми й поговоримо в наступному розділі.