Channels ​

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('hello');
});

echo $channel->recv(); // hello

• send — кладе значення в канал.
• recv — забирає значення з каналу.

Обидві операції блокують, і в цьому весь сенс. Якщо канал порожній, recv призупиняє корутину, доки не з'являться дані. Якщо канал заповнений, натомість призупиняється send. 5 у конструкторі задає розмір буфера: скільки значень канал готовий утримувати, доки їх ніхто не забере.

Rendezvous ​

Що станеться, якщо створити канал із буфером 0? Йому нема куди зберігати значення, тож send не завершиться, доки інша корутина не викличе recv:

$ch = new Channel(0);

spawn(function () use ($ch) {
    echo "before send\n";
    $ch->send('hello');
    echo "after send\n"; // виконається лише після recv
});

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

Такий канал називається rendezvous: відправник і отримувач змушені зустрітися в одній точці часу, звідси й назва.

Відмінність від буферизованого каналу тонша, ніж здається. Буфер означає "надіслано", але не "отримано": send лишає значення й рухається далі, і відправник гадки не має, чи буде, і коли буде його забрано. Rendezvous гарантує доставку: щойно send повертається, значення вже в руках отримувача. Це важливо, коли завдання не можна лишити чекати в черзі: наприклад, передати роботу воркеру лише якщо він зараз вільний.

Дані тут майже ні до чого: rendezvous це чиста синхронізація. Дві корутини гарантовано зустрічаються в один момент, навіть якщо все, що вони передавали, це 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 ніколи не більше десяти. Кількість воркерів це точно ліміт конкурентності, і він задається єдиною константою в циклі.

Backpressure ​

Чому буфер саме 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 на заповненому каналі присипляв читання файлу, доки воркери не розчищали чергу. Rendezvous зводив дві корутини разом в один момент часу. Кожна блокувальна операція виявлялася точкою, у якій корутини підлаштовуються одна під одну.

І тут варто відзначити тонкість однопотокового світу. Для передавання даних як таких канал не потрібен: корутини живуть у спільній пам'яті, і об'єкт можна просто передати їм через use. Гонитви даних не трапляються в межах одного потоку; дві корутини ніколи не виконуються точно в один і той самий момент. Канал використовується для іншого: обмежена черга, розподіл роботи між вільними воркерами, сигнал "роботи більше немає" через close.

Тож точніше думати про канал так: це спосіб синхронізувати корутини, який до того ж переносить дані, а не навпаки. Утім, у багатопотоковому коді, де корутини розподілені по різних потоках, канал стає незамінним в обох ролях: там немає спільної пам'яті, і він лишається єдиним безпечним містком.

Тепер у нас є два способи пов'язати корутини. Future обіцяє одне значення. Канал переносить цілий потік значень і принагідно задає роботі спільний ритм: виробник не знає, хто й коли забере дані, споживач не знає, звідки вони прийшли, але ніхто нікого не затоплює роботою.

Лишилася одна невідступна думка. Ми запустили десять воркерів і рушили далі. А що, як один із них помре з винятком посеред імпорту? Хто взагалі наглядає за всіма цими корутинами? З'ясуймо це в наступному розділі.