Scope
В прошлой главе мы запустили десять воркеров и закрыли канал. Файл прочитан, close() вызван, главный поток пошёл дальше. Но подождите: воркеры ещё разбирают буфер. Функция импорта уже вернула управление, а работа не закончена. Если PHP-скрипт завершится прямо сейчас, часть адресов останется непроверенной.
И вторая тревога из той же главы: что, если checkAddress внутри воркера бросит исключение? Из главы про исключения мы знаем, что оно сохранится в дескрипторе корутины и будет ждать await. Но никто не собирался делать await воркерам. Ошибка всплывёт в самом конце, когда что-то исправлять уже поздно.
Можно решить обе проблемы вручную: складывать корутины в массив и ждать каждую.
$workers = [];
for ($i = 0; $i < 10; $i++) {
$workers[] = spawn(worker(...), $queue);
}
// ... чтение файла ...
foreach ($workers as $worker) {
await($worker);
}Работает. Но это бухгалтерия: массив нужно не забыть завести, пронести через весь код и обойти в конце. А если корутина запускается где-то глубоко в недрах вызванной функции, она в этот массив уже не попадёт. Хочется, чтобы корутины сами знали, кому они принадлежат.
Для этого существует Scope.
Песочница для корутин
Scope — это пространство, внутри которого живут корутины. Он знает о каждой корутине, запущенной в нём, и умеет работать с ними как с группой:
use Async\Scope;
use Async\Channel;
use Async\ChannelException;
$queue = new Channel(100);
$workers = new Scope();
for ($i = 0; $i < 10; $i++) {
$workers->spawn(function () use ($queue) {
foreach ($queue as $address) {
checkAddress($address);
}
});
}
while (($row = fgetcsv($handle)) !== false) {
$queue->send($row[$addressIndex]);
}
$queue->close();
$workers->awaitCompletion(timeout(60000));
echo "Импорт завершён, все адреса проверены\n";Разница с прошлой главой в двух строчках: $workers->spawn() вместо spawn() и awaitCompletion() в конце. Метод awaitCompletion ждёт, пока завершатся все корутины scope, сколько бы их ни было. Никакого массива, никакой бухгалтерии: scope сам ведёт учёт.
Токен отмены здесь не опция, а обязательный аргумент: scope принципиально не позволяет ждать группу без предела. Знакомый timeout подойдёт, и если импорт не уложился в минуту, ожидание прервётся с OperationCanceledException, а дальше решать вам: подождать ещё или отменить группу.
Ошибки больше не теряются
Вернёмся к упавшему воркеру. Корутина в scope не может умереть незаметно: необработанное исключение всплывает в родительский scope. По умолчанию scope реагирует строго: ошибка одной корутины отменяет все остальные, а исключение доставляется тому, кто ждёт в awaitCompletion.
try {
$workers->awaitCompletion(timeout(60000));
} catch (RemoteApiException $e) {
echo "Импорт прерван: {$e->getMessage()}\n";
}Такая стратегия называется fail-together: группа либо доработала целиком, либо целиком остановилась. Для импорта это разумно: если GeoDirectory лёг, продолжать бомбардировать его оставшимися девятью воркерами незачем.
Но строгость нужна не всегда. Один битый адрес в файле не повод бросать остальные девяносто девять тысяч. Тогда scope назначают обработчик ошибок, и корутины становятся независимыми: упавшая логируется, остальные работают:
$workers->setExceptionHandler(function ($scope, $coroutine, Throwable $e) {
error_log("Адрес не проверен: {$e->getMessage()}");
});Кроме самого исключения обработчик получает scope и упавшую корутину: пригодится, чтобы понять, кто именно умер, или перезапустить работу.
Выбор стратегии за вами, и это главное отличие от простого spawn: там стратегия одна, «запустил и забыл».
Отмена всей группы
В главе про отмену мы останавливали одну корутину через cancel(). Scope делает то же самое для всей группы разом:
$workers->cancel();Каждая корутина внутри получит знакомую AsyncCancellation в своей точке ожидания: кто-то в recv, кто-то в delay. Механизм тот же, кооперативный, просто сигнал рассылается всем сразу.
Scope может содержать дочерние scope, и отмена спускается по иерархии рекурсивно: отменили родителя, отменилась вся ветка. Корутины перестают быть россыпью независимых задач и выстраиваются в дерево, где у каждой есть место и хозяин. Этот подход называется структурной конкурентностью, и он уже проверен в Kotlin, Swift и Java. TrueAsync приносит его в PHP.
Scope принадлежит объекту
Самое элегантное применение scope: отдать его во владение объекту.
use Async\Scope;
final class ImportService
{
private Scope $scope;
public function __construct()
{
$this->scope = new Scope();
}
public function import(string $path): void
{
$this->scope->spawn(/* воркеры и чтение файла */);
$this->scope->spawn(/* корутина прогресса из первой главы */);
}
public function __destruct()
{
$this->scope->dispose();
}
}Время жизни корутин теперь совпадает с временем жизни сервиса. Пока объект существует, его корутины работают. Удалили объект, и dispose() отменил всё, что тот успел запустить.
Помните $progress->cancel() из первой главы? Мы вручную ловили момент, когда корутина прогресса становится ненужной. Со scope этот вопрос отпадает сам собой: прогресс нужен, пока идёт импорт, а идёт он ровно столько, сколько живёт ImportService. Владение выражено прямо в коде, и забыть корутину стало попросту негде.
Весь импорт целиком
Соберём всё, что накопилось за восемь глав, в один работающий класс: корутины и прогресс из первой главы, канал с обратным давлением из седьмой, scope из этой.
use Async\Scope;
use Async\Channel;
use Async\ChannelException;
use function Async\delay;
use function Async\timeout;
final class ImportService
{
private Scope $scope;
public function __construct(private readonly int $workers = 10)
{
$this->scope = new Scope();
}
public function import(string $path, int $total): void
{
$queue = new Channel(100);
$counter = 0;
// Воркеры: разбирают адреса из канала, не более $this->workers конкурентно
for ($i = 0; $i < $this->workers; $i++) {
$this->scope->spawn(function () use ($queue, &$counter) {
foreach ($queue as $address) {
checkAddress($address);
$counter++;
}
});
}
// Прогресс: рисует состояние раз в секунду, пока импорт не закончится
$this->scope->spawn(function () use (&$counter, $total) {
while ($counter < $total) {
printProgress($counter, $total);
delay(1000);
}
printProgress($total, $total);
});
// Производитель: читает файл, обратное давление канала бережёт память
$handle = fopen($path, 'r');
$header = fgetcsv($handle);
$addressIndex = array_search('address', $header);
while (($row = fgetcsv($handle)) !== false) {
$queue->send($row[$addressIndex]);
}
fclose($handle);
$queue->close();
// Дожидаемся всех: и воркеров, и прогресса
$this->scope->awaitCompletion(timeout(600000));
}
public function __destruct()
{
$this->scope->dispose();
}
}
$importer = new ImportService();
$importer->import('users.csv', 100_000);Пара деталей, на которые стоит посмотреть внимательно. Прогресс больше не бесконечный цикл: условие while ($counter < $total) завершает его кооперативно, когда обработан последний адрес, поэтому awaitCompletion дожидается всех без единого cancel. А dispose() в деструкторе не участвует в нормальной работе вовсе: это страховка на случай, когда импорт бросил исключение или объект выкинули на полпути.
На этом главное про scope сказано. spawn отвечает только на вопрос «как запустить конкурентную работу». Scope закрывает вопросы, которые приходят следом: кто эту работу ждёт, кто узнает об ошибке и кто её остановит. Без scope корутина предоставлена сама себе, а внутри scope у неё есть владелец и своё место в структуре программы.
До сих пор воркеры только читали из внешнего мира. Но проверенные адреса надо ещё сохранить в базу данных. Можно ли просто отдать десяти корутинам один объект PDO? Хороший вопрос, чтобы открыть следующую главу.