Каналы ​

Future обещает ровно одно значение. А что, если значений много и они появляются постепенно?

Вспомним CSV-файл из первой главы. Теперь задача серьёзнее: при импорте каждый адрес пользователя нужно проверить через GeoDirectory. Корутины у нас уже есть, попробуем в лоб:

while (($row = fgetcsv($handle)) !== false) {
    spawn(checkAddress(...), $row[$addressIndex]);
}

В файле сто тысяч строк. Этот код создаст сто тысяч корутин, и все они конкурентно откроют соединение к GeoDirectory. Корутины дешёвые, а вот соединения нет. Сервис захлебнётся, а вместе с ним и наш импорт.

Хочется другого: пусть проверкой занимаются, скажем, десять корутин, а остальные адреса ждут своей очереди. Нужна очередь, из которой одни корутины забирают работу, а другие её добавляют.

Такая очередь называется каналом.

Channel ​

Канал соединяет корутины напрямую:

use Async\Channel;
use function Async\spawn;

$channel = new Channel(5);

spawn(function () use ($channel) {
    $channel->send('привет');
});

echo $channel->recv(); // привет

• send — кладёт значение в канал.
• recv — забирает значение из канала.

Обе операции блокирующие, и в этом вся суть. Если канал пуст, recv приостанавливает корутину до появления данных. Если канал полон, приостанавливается уже send. Число 5 в конструкторе задаёт размер буфера: сколько значений канал готов хранить, пока их никто не забрал.

Рандеву ​

А что будет, если создать канал с буфером 0? Хранить значения ему негде, поэтому send не завершится, пока другая корутина не вызовет recv:

$ch = new Channel(0);

spawn(function () use ($ch) {
    echo "до send\n";
    $ch->send('привет');
    echo "после send\n"; // выполнится только после recv
});

spawn(function () use ($ch) {
    echo "до recv\n";
    echo $ch->recv() . "\n";
});

Такой канал называется рандеву: отправитель и получатель обязаны встретиться в одной точке времени, отсюда и название.

Разница с буферизованным каналом тоньше, чем кажется. Буфер означает «отправлено», но не «получено»: send положил значение и пошёл дальше, а, когда его заберут и заберут ли вообще, отправитель не знает. Рандеву даёт гарантию вручения: раз send завершился, значение уже в руках получателя. Это важно, когда задаче нельзя лежать в очереди: например, работу отдают воркеру, только если он прямо сейчас свободен.

Данные здесь почти ни при чём: рандеву — синхронизация в чистом виде. Две корутины гарантированно сойдутся в одном моменте, даже если бы передавали null.

Пул воркеров ​

Соберём решение для импорта. Десять корутин-воркеров разбирают адреса из канала, а главный поток читает файл и наполняет канал:

use Async\Channel;
use Async\ChannelException;
use function Async\spawn;

$queue = new Channel(100);

for ($i = 0; $i < 10; $i++) {
    spawn(function () use ($queue) {
        foreach ($queue as $address) {
                checkAddress($address);
        }
    });
}

while (($row = fgetcsv($handle)) !== false) {
    $queue->send($row[$addressIndex]);
}

$queue->close();

Каждое значение из канала достаётся ровно одному воркеру, даже если recv ждут все десять. Значения не дублируются, поэтому воркеры не мешают друг другу: канал сам распределяет работу между свободными корутинами.

Сколько бы строк ни было в файле, соединений к GeoDirectory не больше десяти. Число воркеров и есть предел конкурентности, и задаётся он одной константой в цикле.

Обратное давление ​

А зачем буфер именно 100? Представьте, что воркеры проверяют адреса медленнее, чем главный поток читает файл. Без ограничения очередь разрасталась бы, пока весь стотысячный файл не оказался в памяти.

С буфером 100 происходит другое: как только в канале накапливается сто адресов, send приостанавливает главный поток. Чтение файла замирает и возобновляется, когда воркеры освободят место. Производитель автоматически подстраивается под скорость потребителей.

Этот механизм называется обратным давлением (backpressure). Заметьте, мы его не программировали: он следует из блокирующей природы send. Достаточно выбрать размер буфера, и дальше всё балансируется само.

Закрытие канала ​

Когда файл прочитан, воркеры продолжают ждать в recv. Знакомая ситуация: в главе про отмену так же вечно крутилась корутина прогресса. Но здесь cancel() по каждому воркеру не нужен, у канала есть более точный инструмент:

$queue->close();

close объявляет: новых значений не будет. Воркеры сначала дочитают то, что осталось в буфере, а затем recv бросит ChannelException, и цикл while (true) завершится. Обратите внимание на порядок: закрытие не обрывает работу, а даёт ей корректно закончиться.

И ещё одна знакомая деталь: recv и send принимают тот же токен отмены, что и await из главы про таймауты:

$address = $queue->recv(timeout(5000));

Если за пять секунд в канале ничего не появилось, воркер получит AsyncCancellation и сможет, например, записать предупреждение в лог.

Передача или синхронизация? ​

Присмотримся, что канал на самом деле делал в этой главе. recv на пустом канале усыплял воркера ровно до появления работы. send на полном канале усыплял чтение файла, пока воркеры не разгребут очередь. Рандеву сводил две корутины в одном моменте времени. Каждая блокирующая операция оказалась точкой, где корутины подстраиваются друг под друга.

И тут стоит заметить тонкость однопоточного мира. Для самой передачи данных канал не обязателен: корутины живут в общей памяти, и объект можно просто отдать им через use. Гонок данных в одном потоке не бывает, две корутины никогда не выполняются в один и тот же момент. Канал берут ради другого: очередь с ограничением, раздача работы свободным воркерам, сигнал «работы больше не будет» через close.

Поэтому точнее думать о канале так: это способ синхронизации корутин, который заодно передаёт данные, а не наоборот. А вот в многопоточном коде, когда корутины разнесены по разным потокам, канал становится незаменим уже в обеих ролях: там общей памяти нет, и он остаётся единственным безопасным мостом.

Теперь у нас два способа связать корутины между собой. Future обещает одно значение. Канал передаёт целый поток значений и заодно задаёт общий ритм работы: производитель не знает, кто и когда заберёт данные, потребитель не знает, откуда они пришли, но никто никого не заваливает работой.

Осталась одна тревожная мысль. Мы запустили десять воркеров и разошлись. А если один из них умрёт с исключением посреди импорта? Кто вообще присматривает за всеми этими корутинами? Разберёмся в следующей главе.