Потоки ​

Все наши инструменты до сих пор жили в одном потоке операционной системы. Корутины создавали ощущение одновременности, но в каждый момент времени работала ровно одна из них. Для импорта и запросов к GeoDirectory этого хватало с избытком: задачи в основном ждали, а ожидание отлично делится.

Теперь другая задача. Раз в сутки ProfileService строит годовой отчёт: минута чистых вычислений, без единого запроса к сети или базе. Запустим его в корутине, рядом с привычным тикером:

spawn(function () {
    while (true) {
        echo "тик\n";
        delay(1000);
    }
});

spawn(function () {
    $report = buildYearlyReport($rows); // минута чистых вычислений
    echo "отчёт готов\n";
});

Тикер замолкает на минуту. Вспомните первую главу: переключение корутин происходит в точках ожидания, на sleep, delay, вводе-выводе. А в buildYearlyReport нет ни одной точки ожидания, только циклы и арифметика. Планировщик не получает управление, и весь процесс, включая тикер, воркеров и обработку запросов, стоит и смотрит, как одна корутина считает.

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

spawn_thread ​

use function Async\spawn_thread;
use function Async\await;

$thread = spawn_thread(fn() => buildYearlyReport($rows));

$report = await($thread);

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

Снаружи поток выглядит почти как корутина: его можно передать в знакомый await, который приостановит только ожидающую корутину. Исключение из потока тоже доедет до await, обёрнутое в Async\RemoteException.

Ничего общего ​

Почти как корутина, но с одним фундаментальным отличием. В главе про каналы мы говорили: корутины живут в общей памяти, объект можно просто отдать через use, гонок данных в одном потоке не бывает. Так вот, с потоками этот фокус не работает. У каждого потока собственное окружение PHP: свои переменные, свои классы, свои статические свойства. Общей памяти нет вовсе.

Поэтому всё, что уходит в поток, копируется целиком:

$rows = loadRows();

$thread = spawn_thread(function () use ($rows) {
    // это КОПИЯ $rows: изменения здесь снаружи не видны
    return buildYearlyReport($rows);
});

Из этого правила следуют и запреты: в поток нельзя передать PHP-ссылку (&$var), ресурс вроде открытого файла или объект с динамическими свойствами. Попытка обернётся ThreadTransferException сразу, на старте, а не загадочным поведением потом.

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

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

$state  = new FutureState();
$future = new Future($state);

spawn_thread(function () use ($state) {
    $state->complete(buildYearlyReport(loadRows()));
});

$report = await($future);

Производитель в одном потоке, потребитель в другом, а контракт тот же. Вот зачем Future был разделён на два объекта: граница между ними оказалась настолько прочной, что через неё можно провести границу потоков.

ThreadPool ​

Поток — дорогая сущность: собственное окружение PHP надо создать, инициализировать и прогреть. Одна задача в минуту — пожалуйста, а вот запускать поток на каждую мелкую задачу так же расточительно, как открывать соединение к базе на каждый запрос.

Вы уже знаете, что делают с дорогими ресурсами. Правильно, пул:

use Async\ThreadPool;

$pool = new ThreadPool(workers: 8);

$futures = [];
foreach ($uploads as $path) {
    $futures[] = $pool->submit(makeThumbnail(...), $path);
}

foreach ($futures as $future) {
    echo await($future), "\n";
}

$pool->close();

Восемь воркер-потоков создаются один раз и живут, пока живёт пул. submit кладёт задачу в очередь, свободный воркер забирает её и возвращает результат. И посмотрите, что возвращает submit: Future, обещание результата, который произведёт кто-то другой. В шестой главе этим «кем-то» была корутина, теперь целый отдельный поток, а контракт остался прежним.

Очередь пула ограничена, и если она заполнена, submit приостановит вызывающую корутину. Узнаёте? Обратное давление из главы про каналы: производитель задач автоматически подстраивается под скорость воркеров.

Число воркеров стоит держать близким к числу ядер процессора: вычислениям, в отличие от ожиданий, нужны физические ядра, и двадцать потоков на восьми ядрах будут только толкаться локтями. По умолчанию, без параметра workers, пул сам определит доступный параллелизм, учитывая даже квоты контейнера.

А для самого частого случая «обработать список параллельно» есть знакомая по iterate форма одной строчки:

$thumbs = $pool->map($uploads, makeThumbnail(...));

map раздаёт элементы воркерам, ждёт всех и возвращает результаты в исходном порядке.

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

Напоследок обратите внимание, во что превратился наш процесс: сотни корутин вперемешку обслуживают разных пользователей, задачи скачут между воркерами и потоками. И в этой толчее простой вопрос звучит неожиданно сложно: а где хранить «текущее»? Текущего пользователя, язык, идентификатор запроса? Глобальная переменная одна на всех, а корутин тысячи. Об этом в последней главе.