Конкурентний ітератор ​

Порахуймо, скільки разів ми будували один і той самий патерн. У главі про канали: канал, десять воркерів, цикл recv. У главі про TaskGroup: concurrency: 10, цикл spawn, close(). У главі про TaskSet: те саме, але зі споживанням результатів. Щоразу задача звучала однаково: обійти колекцію, зробити конкурентну роботу над кожним елементом і ніколи не перевищувати ліміт.

Патерн настільки поширений заслуговує на власну функцію:

use function Async\iterate;

iterate($addresses, checkAddress(...), concurrency: 10);

І це весь пул воркерів. iterate викликає функцію для кожного елемента колекції у власній корутині, стежить, щоб конкурентно виконувалося не більше десяти, і повертає керування, щойно все оброблено.

Генератор замість масиву ​

Стривай, а як же читання файлу? Зібрати сто тисяч адрес у масив тільки заради виклику iterate було б кроком назад: саме зворотний тиск із глави про канали заощаджував нам пам'ять. Немає проблеми: iterate приймає не лише масив, а будь-який Traversable, включно з генераторами:

function addresses(string $path): Generator
{
    $handle = fopen($path, 'r');
    $header = fgetcsv($handle);
    $addressIndex = array_search('address', $header);

    while (($row = fgetcsv($handle)) !== false) {
        yield $row[$addressIndex];
    }

    fclose($handle);
}

iterate(addresses('users.csv'), checkAddress(...), concurrency: 10);

Генератор читає файл ліниво, по одному рядку за раз. iterate витягує наступну адресу лише тоді, коли звільняється слот, тож файл ніколи не тримається в пам'яті цілком. Це той самий зворотний тиск, який дає буферизований канал, тільки тепер він невидимий: усе, що лишилося, це один рядок, що виражає намір.

Ранній вихід ​

Функція-обробник може зупинити весь обхід, повернувши false:

$broken = 0;

iterate(addresses('users.csv'), function (string $address) use (&$broken) {
    if (!checkAddress($address)) {
        $broken++;
    }

    if ($broken >= 100) {
        return false; // файл пошкоджений, продовжувати немає сенсу
    }
}, concurrency: 10);

Сотня недійсних адрес поспіль це надійна ознака того, що нам підсунули не той файл. Жоден новий елемент після false не запускається, корутини, що вже працюють, доробляють свою справу, і iterate повертається.

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

try {
    iterate(addresses('users.csv'), checkAddress(...), concurrency: 10);
} catch (RemoteApiException $e) {
    echo "Імпорт перервано: {$e->getMessage()}\n";
}

За цим немає ніякої магії: усередині iterate живе звичайний дочірній scope з восьмої глави, і саме він підтримує лад.

Драбина абстракцій ​

За шість глав склалася ціла драбина, і варто ще раз оглянути її, знизу вгору:

• Канал + Scope — примітиви. Будь-яка топологія: пули, конвеєри, рандеву, супервізори. Максимум контролю, максимум коду.
• TaskGroup / TaskSet — готові збірки для набору задач, коли потрібні результати: усі, перший, перший успішний або по одному в міру готовності.
• iterate() — один рядок, коли результат не потрібен і треба лише конкурентно обійти колекцію.

Що вище драбиною, то менше коду і вужчий сценарій. Починай згори: якщо iterate вистачає, немає сенсу будувати групу; якщо групи замало, спускайся до каналів. Унизу можна зібрати будь-яку конструкцію, яку заманеться; згори все вже зібрано за тебе.

Зверни увагу, що сталося з нашим імпортом: глава за главою він усе стискався, доки нарешті не вмістився в один рядок. Саме так і має бути. Конкурентність перестає бути подією і стає звичайним інструментом, як foreach.

Але одне запитання з дев'ятої глави досі висить у повітрі. Для PDO пул з'єднань вбудований у ядро, тоді як checkAddress спілкується з GeoDirectory по HTTP. Його з'єднання теж варто перевикористовувати, і не лише його: сокети, клієнти, важкі об'єкти взагалі. Невже нам справді доведеться щоразу власноруч будувати пул із каналу? На щастя, ні, і наступна глава покаже чому.