ОсновиTaskSet

TaskSet ​

У попередній главі з'ясувалося, що TaskGroup пам'ятає все. Це не випадковість, а властивість, на яку можна спертися: скільки б разів ти не запитував групу про результати, відповідь буде та сама. Така поведінка називається ідемпотентністю, і ми з нею вже стикалися: повторний await корутини щоразу повертає той самий результат.

Але пам'ять має ціну. Проженеш крізь групу сто тисяч задач, і всі сто тисяч результатів залишаться лежати всередині, навіть якщо кожен був потрібен рівно один раз, у момент готовності. Для конвеєра, що працює годинами, це не сховище, а витік.

Потрібен близнюк TaskGroup із протилежним характером: віддав результат і забув. Він називається TaskSet.

Споживання замість зберігання ​

Зовні все виглядає однаково: spawn, close, ліміт конкурентності. Різниця в тому, що відбувається з результатом після того, як його доставлено:

php
use Async\TaskSet;

$set = new TaskSet();

$set->spawn(fn() => 'alpha');
$set->spawn(fn() => 'beta');
$set->spawn(fn() => 'gamma');

echo $set->joinNext()->await(); // alpha
echo $set->joinNext()->await(); // beta, уже інший!
echo $set->joinNext()->await(); // gamma

echo $set->count(); // 0, набір порожній

Кожен виклик joinNext() повертає наступний готовий результат і видаляє його запис із набору. Порівняй це з race() у TaskGroup, що повертає того самого першого переможця скільки б разів ти не викликав. TaskSet поводиться не як сховище, а як черга: прочитав, і його вже немає. Так, це та сама семантика, що й у recv із глави про канали, тільки тепер у черзі стоять не значення, а задачі, що завершуються.

Методи очікування у близнюків римуються між собою:

  • joinNext() — як race(): перший, хто завершився, його запис видаляється.
  • joinAny() — як any(): перший успішний, його запис видаляється.
  • joinAll() — як all(): усі результати разом, набір спорожнюється.

Префікс join сам натякає на різницю: результат не просто прочитано, а вилучено.

Конвеєр без витоку ​

Перепишемо стотисячний імпорт із урахуванням того, що ми тепер знаємо:

php
$set = new TaskSet(concurrency: 10);

spawn(function () use ($set, $handle, $addressIndex) {
    while (($row = fgetcsv($handle)) !== false) {
        $set->spawn(fn() => checkAddress($row[$addressIndex]));
    }
    $set->close();
});

foreach ($set as $key => [$result, $error]) {
    if ($error !== null) {
        error_log("Адресу не перевірено: {$error->getMessage()}");
        continue;
    }
    saveAddress($pdo, $result);
}

Одна корутина читає файл і невпинно підкидає задачі, а головний потік обробляє результати в міру їхньої готовності. Кожна оброблена запис одразу видаляється з набору, тож пам'ять тримає лише задачі в польоті: десять працюючих плюс черга. Файл може бути будь-якого розміру, споживання пам'яті від нього не залежить.

Зверни увагу, як знайомі деталі склалися в нову картину: ліміт конкурентності замість саморобного пулу воркерів, close() як сигнал «задач більше не буде», пара [$result, $error] замість мовчки проковтнутих винятків, і PDO з дев'ятої глави, якому не страшні конкурентні виклики saveAddress.

Супервізор ​

Є і другий сценарій, де ця споживаюча семантика незамінна: код, що наглядає за довгоживучими задачами і реагує на їхнє завершення.

php
$set = new TaskSet();

$set->spawnWithKey('mailer',  runMailer(...));
$set->spawnWithKey('metrics', runMetrics(...));
$set->spawnWithKey('cleaner', runCleaner(...));

foreach ($set as $key => [$result, $error]) {
    error_log("Служба $key зупинилася" . ($error ? ": {$error->getMessage()}" : ''));

    // перезапускаємо службу, що впала
    $set->spawnWithKey($key, restartService($key));
}

Набір ніколи не закривається, тож foreach ніколи не завершується, він просто чекає на наступну подію. Кожна оброблена запис видаляється, перезапущена служба додається на її місце, і цикл живе вічно. Супервізору не потрібна історія всіх завершень від початку часів; йому потрібно саме це: хтось із підопічних зупинився, іди розберися чому і перезапусти. Такий цикл не написати з TaskGroup: його foreach щоразу починав би заново з першої служби, що зупинилася.

Ось, власне, і вся різниця між близнюками: пам'ять. Група зберігає результати і повторно відповідає на будь-яке запитання про них, тому вона добра там, де набір задач фіксований, а результати цінні як єдине ціле. Набір віддає кожен результат один раз і одразу звільняє пам'ять, тому йому до снаги нескінченний потік задач. Є просте правило вибору: якщо твоє запитання до задач звучить як «що ви там насчитали?», бери TaskGroup; якщо «що там далі?», бери TaskSet.

За останні чотири глави ми тричі побудували одне й те саме: обійти колекцію, виконуючи для кожного елемента конкурентну роботу з обмеженням. Раз через канал із воркерами, раз через TaskGroup, раз через TaskSet. Чи не час цьому патерну отримати власне ім'я і стиснутися до одного рядка?