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 "Import finished, all addresses checked\n";Відмінність від попереднього розділу це два рядки: $workers->spawn() замість spawn(), і awaitCompletion() наприкінці. Метод awaitCompletion чекає, доки завершиться кожна корутина в скоупі, хоч би скільки їх було. Ані масиву, ані рахівництва: скоуп веде облік сам.
Токен скасування тут не опційний, це обов'язковий аргумент: скоуп навмисне не дає вам чекати на групу без обмеження. Знайомий timeout пасує чудово, і якщо імпорт не встигає протягом хвилини, очікування переривається винятком OperationCanceledException, і вам вирішувати: почекати ще трохи чи скасувати групу.
Помилки більше не губляться
Повернімося до воркера, який упав. Корутина всередині скоупу не може померти тихо: необроблений виняток спливає до батьківського скоупу. За замовчуванням скоуп реагує суворо: помилка в одній корутині скасовує всі інші, а виняток доставляється тому, хто чекає в awaitCompletion.
try {
$workers->awaitCompletion(timeout(60000));
} catch (RemoteApiException $e) {
echo "Import aborted: {$e->getMessage()}\n";
}Ця стратегія називається fail-together: група або завершується цілком, або зупиняється цілком. Для імпорту це розумно: якщо GeoDirectory вийде з ладу, немає сенсу бомбардувати його рештою дев'яти воркерів.
Але суворість не завжди те, чого хочеться. Одна погана адреса у файлі це не привід кидати решту дев'яноста дев'яти тисяч. У такому разі ви призначаєте скоупу обробник помилок, і корутини стають незалежними: невдалу залоговано, решта далі працює:
$workers->setExceptionHandler(function ($scope, $coroutine, Throwable $e) {
error_log("Address not checked: {$e->getMessage()}");
});Окрім самого винятку, обробник отримує скоуп і корутину, що впала: зручно, щоб з'ясувати, хто саме помер, або щоб перезапустити роботу.
Вибір стратегії за вами, і це головна відмінність від звичайного spawn: там єдина стратегія це "запустити й забути".
Скасування всієї групи
У розділі про скасування ми зупиняли одну корутину за допомогою cancel(). Скоуп робить те саме для цілої групи за раз:
$workers->cancel();Кожна корутина всередині отримує знайомий AsyncCancellation у своїй точці очікування: одні в recv, інші в delay. Механізм той самий, кооперативний, просто сигнал іде до всіх за раз.
Скоуп може містити дочірні скоупи, і скасування тече вниз ієрархією рекурсивно: скасуйте батьківський, і скасовується вся гілка. Корутини перестають бути розкиданою купою незалежних завдань і утворюють дерево, де кожна має своє місце й власника. Цей підхід називається структурованою конкурентністю, і він уже довів себе в Kotlin, Swift і Java. TrueAsync приносить його в PHP.
Скоуп належить об'єкту
Найелегантніше застосування скоупу: передати володіння ним об'єкту.
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() із першого розділу? Ми вручну ловили момент, коли корутина прогресу ставала непотрібною. Зі скоупом це питання просто зникає: прогрес потрібен, доки триває імпорт, а він триває рівно стільки, скільки живе ImportService. Володіння виражене прямо в коді, і забути корутину просто ніде.
Увесь імпорт від початку до кінця
Зберімо все, що ми накопичили за вісім розділів, в один робочий клас: корутини й прогрес із першого розділу, канал зі зворотним тиском із сьомого, скоуп із цього.
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() у деструкторі під час звичайної роботи не грає жодної ролі: це запобіжна сітка на випадок, коли імпорт кине виняток або об'єкт буде відкинуто на півдорозі.
Це основне про скоуп. spawn лише відповідає на питання "як мені почати конкурентну роботу". Скоуп бере на себе питання, що постають одразу після: хто чекає на цю роботу, хто дізнається про помилку і хто її зупиняє. Без скоупу корутина полишена сама на себе; усередині скоупу вона має власника й місце в структурі програми.
Досі воркери лише читали із зовнішнього світу. Але перевірені адреси все ще потрібно зберегти в базу даних. Чи можна просто передати десяти корутинам один об'єкт PDO? Гарне питання, щоб відкрити наступний розділ.