Основы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. Не пора ли этому паттерну получить собственное имя и уместиться в одну строчку?