Заблокировать цикл и вынести счёт в отдельный поток

Что долгий колбэк задерживает всех, мы уже видели: цикл не прерывает работающий код и заглядывает в таймеры, только когда колбэк вернулся. Чего мы пока не умеем — сказать, насколько это плохо в конкретном месте. Вопрос «подвисает ли» бесполезен; полезен вопрос «на сколько опаздывает то, что должно было случиться вовремя». На него отвечает прибор в пять строк: таймер с известным заранее сроком, который при каждом срабатывании записывает разницу между сроком и тем, что показывают часы.

Назовём его сердцем. Сердце бьётся каждые два тика и заводит себя заново из собственного колбэка, а не из расписания. Это не упрощение ради модели: setInterval устроен так же — следующий срок он отмеряет от того момента, когда колбэк сработал.

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

// «сердце» цикла: таймер на каждые два тика, который сам себя перезаводит
// и записывает, на сколько опоздал удар
const опоздания = [];
function сердце() {
  const срок = loop.now + 2;
  loop.timeout(() => {
    опоздания.push(loop.now - срок);
    if (опоздания.length < 12) сердце();
  }, 2);
}
сердце();

// долгий счёт: колбэк не возвращается двадцать тиков, и всё это время цикл стоит
loop.timeout(() => {
  console.log(`счёт начался на тике ${loop.now}`);
  loop.now += 20; // поток занят: ни одна фаза не выполнится, пока колбэк не вернулся
  console.log(`счёт кончился на тике ${loop.now}`);
}, 6);

loop.run();
console.log('опоздания ударов:', опоздания.join(' '));

Опоздал ровно один удар и ровно на длину счёта. Не десять опоздавших ударов подряд и не «сердце сбилось до конца прогона» — одна яма глубиной в двадцать тиков, и сразу за ней ровный ряд нулей. Это и есть мера: наибольшее опоздание равно длине самого долгого синхронного куска. Всё остальное — производные: сколько таких кусков в секунду и какой запас у соседей.

Из той же строки следует и то, чего не случилось. За двадцать тиков сердце должно было ударить десять раз, а недосчиталось девяти ударов: пропущенные не копятся и не обрушиваются пачкой, когда поток освободится. С кликами в браузере ровно наоборот — те ждут в очереди все. Разница в том, кто заводит следующий срок: клик приходит от человека, а следующий удар заводит предыдущий.

Прибор не обязан быть таймером на два тика. Важно одно: срок известен заранее, а опоздание считается по факту. Единицы — подробность.

Прибор переносится на Node почти без изменений: setInterval на десять миллисекунд, который только увеличивает счётчик. Работа — триста миллисекунд арифметики, сначала в главном потоке, потом в отдельном. Печатаем не миллисекунды (у каждой машины свои), а то, что от них не зависит: сколько ударов уместилось.

Прогноз перед прогоном — три строки. Сколько ударов насчитает сердце за триста миллисекунд синхронного счёта? Наверстает ли оно пропущенное, когда поток освободится? И изменится ли ответ, если ровно тот же счёт уедет в отдельный поток?

// «Сердце» цикла: таймер на 10 мс, который только считает свои удары. Рядом —
// триста миллисекунд арифметики: сначала в главном потоке, потом в отдельном.
import { Worker } from 'node:worker_threads';

const СЧЁТ_МС = 300;

let ударов = 0;
const сердце = setInterval(() => {
  ударов += 1;
}, 10);

// ровно та же работа, что делает count.mjs в отдельном потоке
function считать(мс) {
  const конец = Date.now() + мс;
  let сумма = 0;
  for (let i = 0; Date.now() < конец; i += 1) сумма += Math.sqrt(i);
  return сумма;
}

function вГлавномПотоке() {
  ударов = 0;
  считать(СЧЁТ_МС);
  // печать стоит внутри того же колбэка, до возврата из него
  console.log(`счёт в главном потоке: ударов сердца за время счёта — ${ударов}`);
  setTimeout(() => {
    console.log(`пропущенные удары наверстались пачкой: ${ударов > 5 ? 'да' : 'нет'}`);
    вОтдельномПотоке();
  }, 0);
}

function вОтдельномПотоке() {
  ударов = 0;
  const поток = new Worker(new URL('./count.mjs', import.meta.url));
  поток.on('message', () => {
    console.log(`счёт в отдельном потоке: ударов больше десяти — ${ударов > 10 ? 'да' : 'нет'}`);
    clearInterval(сердце);
    поток.terminate();
  });
  поток.postMessage(СЧЁТ_МС);
}

вГлавномПотоке();

счёт в главном потоке: ударов сердца за время счёта — 0
пропущенные удары наверстались пачкой: нет
счёт в отдельном потоке: ударов больше десяти — да

Ноль — это не «мало», это точно ноль, и он не зависит от машины: печать стоит внутри того же колбэка, до возврата из него, а значит, цикл ещё ни разу не заходил в фазу сроков. Сколько бы миллисекунд ни шёл счёт, ударов за это время не будет ни одного.

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

Третья строка — тот же счёт, та же арифметика, но выполняет её отдельный поток из worker_threads. Ударов за это время набирается три десятка (на машине автора 28—31), поэтому в выводе стоит порог, а не число: важно не «тридцать», а то, что сердце било как ни в чём не бывало. Работы меньше не стало — она просто перестала быть работой главного потока.

Лекарств, как и было обещано, два. Первое — резать счёт на порции и отдавать ход между ними: оно ничего не уносит из процесса и годится там, где работу можно разрезать. Второе — отдать работу целиком другому потоку; ему и посвящён остаток статьи. Как выбирать между ними — в последнем разделе.

Worker читатель уже заводил в главе про service worker, и урок там был про то, чего у потока нет: document. Здесь урок обратный — про то, что у него есть: собственный поток операционной системы, своя копия движка, своя куча и свой цикл событий. Отдельный поток не ускоряет счёт. Он меняет ответ на вопрос «чей поток сейчас занят».

В Node поток заводят файлом — new Worker(new URL('./count.mjs', import.meta.url)) из node:worker_threads, это видно в прогоне выше. Браузеру тоже нужен файл по адресу, а у страницы учебника второго файла нет; поэтому текст потока заворачивают в Blob и получают на него адрес через URL.createObjectURL. Приём не учебный: так же собирают воркеров сборщики и библиотеки, которым не хочется тащить рядом второй файл.

Считать будем одно и то же двумя способами, а мерить — не работу, а занятость главного потока. Прогноз: какая строка напечатается первой и совпадут ли два ответа?

// Одна и та же работа двумя способами. Меряем не саму работу, а то, насколько
// долго при этом занят главный поток.
const РАБОТА = 50000000;
const счёт = (до) => {
  let сумма = 0;
  for (let i = 0; i < до; i += 1) сумма += Math.sqrt(i);
  return Math.round(сумма);
};

// способ первый: считаем прямо здесь
const началиЗдесь = Date.now();
const ответЗдесь = счёт(РАБОТА);
const занятЗдесь = Date.now() - началиЗдесь;

// способ второй: текст той же функции уезжает в отдельный поток
const источник = `
  const счёт = ${счёт};
  self.onmessage = (e) => self.postMessage(счёт(e.data));
`;
const адрес = URL.createObjectURL(new Blob([источник], { type: 'text/javascript' }));
const поток = new Worker(адрес);

const началиТам = Date.now();
поток.postMessage(РАБОТА);
const занятТам = Date.now() - началиТам;
console.log('заказ отправлен, главный поток свободен');

поток.onmessage = (событие) => {
  console.log('ответ потока совпал с ответом главного потока:', событие.data === ответЗдесь);
  console.log('главный поток был занят хотя бы в десять раз меньше:', занятТам * 10 < занятЗдесь);
  поток.terminate();
  URL.revokeObjectURL(адрес);
};

Ответы совпали — это важнее, чем кажется: в поток уехал текст той же самой функции, подставленный в шаблонную строку, и посчитал он ровно то же. А главный поток потратил на всё про всё столько, сколько нужно на одно postMessage с числом внутри. Порог «хотя бы в десять раз меньше» взят с большим запасом: на машине автора счёт держал поток около сотни миллисекунд, а отправка — меньше одной.

Первая строка печатается раньше ответа — и не потому, что «поток быстрый», а потому, что ответ приходит сообщением, то есть обычным событием цикла, и ждёт своей фазы наравне со всеми. Отправка не ждёт ничего.

Текст функции уехал в поток подстановкой в шаблонную строку. Это сработало лишь потому, что внутри неё нет ни одного внешнего имени: в поток едет текст, а не замыкание.

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

Первое следствие — чего в поток не отдать. Ни объект запроса и ответа, ни соединение с базой, ни открытый файл: это не значения, а ресурсы процесса, за ними стоят дескрипторы операционной системы, и копировать там нечего. В поток отдают задание — данные, которых довольно, чтобы посчитать, — и получают обратно результат. Обработчик со своим сокетом остаётся в главном потоке и отвечает сам, когда придёт сообщение.

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

// ArrayBuffer — просто кусок памяти. Отправим два одинаковых куска по-разному.
const источник = `
  self.onmessage = (e) => self.postMessage(e.data.byteLength);
`;
const адрес = URL.createObjectURL(new Blob([источник], { type: 'text/javascript' }));
const поток = new Worker(адрес);
const доехало = [];

поток.onmessage = (событие) => {
  доехало.push(событие.data);
  if (доехало.length < 2) return;
  console.log('в потоке байт:', доехало.join(' и '));
  поток.terminate();
  URL.revokeObjectURL(адрес);
};

const копией = new ArrayBuffer(8000000);
поток.postMessage(копией);
console.log('после копии у отправителя байт:', копией.byteLength);

const передачей = new ArrayBuffer(8000000);
поток.postMessage(передачей, [передачей]);
console.log('после передачи у отправителя байт:', передачей.byteLength);

Второй аргумент postMessage — список того, что нужно передать, а не скопировать. После передачи буфер у отправителя отсоединён: длина ноль, читать нечего и писать некуда. Это не поломка, а условие сделки — память осталась одна, и владелец у неё теперь один. Копия проще, передача дешевле; выбирают по размеру, а не по вкусу.

У worker_threads всё то же самое и немного сверх: данные при запуске (workerData), отдельные каналы сообщений (MessagePort), пределы по памяти на поток и настоящая общая память — SharedArrayBuffer вместе с Atomics. Последнего в песочнице учебника не показать: общая память требует от страницы особых заголовков, которых у песочницы нет. Так что пример выше — предел того, что здесь можно потрогать руками.

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

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

Порции и поток не соперники. Порции оставляют работу в процессе и годятся там, где её можно разрезать; поток уносит её целиком и годится там, где резать нечего. А половина случаев не требует ни того, ни другого.

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