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

Посчитаем, сколько раз мы построили один и тот же паттерн. В главе про каналы: канал, десять воркеров, цикл 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: ошибка в любом обработчике останавливает обход, отменяет оставшиеся корутины и всплывает наружу. Fail-together по умолчанию:

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. Соединения к нему тоже стоило бы переиспользовать, и не только к нему: сокеты, клиенты, тяжёлые объекты. Неужели каждый раз собирать пул из канала руками? К счастью, нет, и следующая глава покажет почему.