Обмен памятью между рабочими потоками Node.js с использованием SharedArrayBuffer и Atomics

Рабочие потоки Node.js — это настоящие потоки операционной системы, каждый из которых имеет собственный экземпляр движка JavaScript. Это побуждает предполагать, что они делят память так же, как потоки в C или Java, но это предположение оказывается ошибочным при первой попытке координации действий двух рабочих потоков над одними и теми же данными. В этой статье объясняется, что на самом деле делает функция postMessage с вашими данными, когда действительно оправдана настоящая общая память, и как класс Atomics обеспечивает корректность одновременных обновлений.

Если вы только начинаете знакомство с рабочими потоками в целом, начните с статьи переноса тяжелых вычислений в Node.js с использованием рабочих потоков и пулов; здесь же рассматривается более узкая тема: что происходит с памятью после создания рабочего потока.

Функция postMessage отправляет копию, а не ссылку

Эксперимент начинается с того, что основной поток создаёт рабочий процесс из отдельного файла.

// main.js
const { Worker } = require("worker_threads");
const worker = new Worker("./worker.js");

Затем он отправляет обычный массив рабочему процессу, а через мгновение записывает в лог свою собственную копию этого массива.

const sharedLookingArray = [1, 2, 3];
worker.postMessage(sharedLookingArray);setTimeout(() => {
  console.log("Main thread's array:", sharedLookingArray); // still [1, 2, 3]
}, 100);

С другой стороны, рабочий процесс получает массив, добавляет к нему элемент и записывает результат.

// worker.js
const { parentPort } = require("worker_threads");
parentPort.on("message", (arr) => {
  arr.push(4);
  console.log("Worker's array:", arr); // [1, 2, 3, 4]
});

Рабочий процесс выводит [1, 2, 3, 4], в то время как основной поток по-прежнему выводит [1, 2, 3], независимо от времени ожидания. Причина в том, что функция postMessage обрабатывает значение с использованием алгоритма структурированного клона, того же механизма, который глобально доступен под именем structuredClone. Данные сериализуются на стороне отправки и воссоздаются в виде независимого объекта на стороне приёма, поэтому после вызова существуют два не связанных между собой массива — по одному на поток.

Это преднамеренный и безопасный стандартный вариант. Два потока не могут одновременно работать с памятью, которая никогда не предоставлялась им обоим. Затраты связаны с самой копией, размер которой растет вместе с объемом данных и возникают при каждой передаче сообщения. (Для крупных двоичных данных Node также может переместить ArrayBuffer в другой поток в виде объекта, передаваемого между процессами, что позволяет избежать копирования, но делает данные недоступными для отправителя; это передача, а не совместное использование.)

Когда общая память — правильный инструмент

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

Основной поток снова загружает модуль рабочего процесса:

// main.js
const { Worker } = require("worker_threads");

Затем выделяется четырехбайтный общий буфер, достаточный для одного 32-битного целого числа; он оборачивается в представление типа Int32Array, после чего буфер передается рабочему процессу через параметр workerData.

const sharedBuffer = new SharedArrayBuffer(4); // 4 bytes, one 32-bit integer
const sharedCounter = new Int32Array(sharedBuffer);const worker = new Worker("./worker.js", { workerData: { sharedBuffer } });

Рабочий процесс создает собственное представление типа Int32Array над тем же самым буфером:

// worker.js
const { workerData } = require("worker_threads");
const sharedCounter = new Int32Array(workerData.sharedBuffer);

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

sharedCounter[0] += 1; // this thread and the main thread are now touching the same memory

Это существенно отличается от обычного поведения конкурентности в JavaScript. Код, находящийся в одной области выполнения, работает в рамках одного потока, поэтому знакомые ситуации с асинхронными конфликтами, когда две функции с модификатором async выполняются в неожиданном порядке, все равно связаны с одновременной работой потока в определенных моментах. С использованием SharedArrayBuffer два потока могут читать и писать в одну и ту же память буквально в один и тот же момент.

Потерянное обновление, которое теперь стало возможным

Рассмотрим очевидный способ увеличения общего счетчика:

// naive, and genuinely unsafe with real shared memory
sharedCounter[0] = sharedCounter[0] + 1;

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

Атомарные операции: неделимые операции над общей памятью

Объект Atomics предоставляет операции над общими целочисленными массивами, которые выполняются как единая единица. Никакой другой поток не может наблюдать или вмешиваться в атомарную операцию на ее промежуточном этапе.

Atomics.add(sharedCounter, 0, 1); // read, increment, and write, as one indivisible step

Если каждая поток увеличивает значение с помощью Atomics.add вместо операции читать-изменить-записать, никакие обновления не могут быть потеряны, поскольку движок гарантирует, что другие атомарные операции над этим элементом не могут выполняться одновременно с ней. Эта гарантия действует только в том случае, если все потоки используют атомарные операции; один поток, продолжающий использовать обычное увеличение значения, снова создает риск конкуренции. Все более сложные структуры, создаваемые на основе общей памяти — будь то счётчики, блокировки или очереди, — в конечном итоге состоят из операций подобного типа, которые являются атомарными на уровне аппаратного обеспечения, а не просто «обычно работающими».

Atomics.wait и Atomics.notify: блокировка между потоками

Atomics также предоставляет то, чего нет в JavaScript: возможность полностью остановить выполнение одной потоковой задачи до тех пор, пока другая потоковая задача не разбудит её. В рабочем потоке функция Atomics.wait блокируется до тех пор, пока счётчик не примет значение 0:

// worker.js — waits for main thread to signal it's time to proceed
Atomics.wait(sharedCounter, 0, 0); // blocks this thread while sharedCounter[0] === 0
console.log("Signaled to continue, value is now:", sharedCounter[0]);

Основной поток сохраняет новое значение, а затем разбудит один из ожидающих потоков:

// main.js — does the signaling
Atomics.store(sharedCounter, 0, 1);
Atomics.notify(sharedCounter, 0, 1); // wakes up one waiting thread

Важно сначала сохранить значение, а уже затем отправить уведомление. Функция Atomics.wait сначала проверяет значение и немедленно возвращает, если оно больше не совпадает с ожидаемым, поэтому сигнал, отправленный до того, как рабочий поток начнёт ожидать, не теряется.

Это настоящий блокирующий вызов, что является необычным для JavaScript, где асинхронная работа обычно выполняется через обратные вызовы в цикле событий. Блокировка потока, запускающего цикл событий, приведет к заморозке всего остального, что обрабатывается этим потоком, именно поэтому браузеры не разрешают использование Atomics.wait в основном потоке, и поэтому он, как правило, предназначен для рабочих потоков. (Браузеры также требуют, чтобы страница была изолирована по принципу cross-origin, прежде чем SharedArrayBuffer вообще станет доступен; у Node нет таких ограничений.)

Реалистичное применение: отображение статуса выполнения в рабочей пуле

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

// main.js
const progress = new Int32Array(new SharedArrayBuffer(4));
const workers = chunks.map((chunk) =>
  new Worker("./process-chunk.js", { workerData: { chunk, progressBuffer: progress.buffer } })
);

Затем он считывает значение счётчика каждые полсекунды с помощью Atomics.load, что гарантирует получение полного и актуального значения:

setInterval(() => {
  console.log(`Processed ${Atomics.load(progress, 0)} items so far`);
}, 500);

Каждый процесс создаёт собственный вариант общего буфера:

// process-chunk.js
const { workerData } = require("worker_threads");
const progress = new Int32Array(workerData.progressBuffer);

При обработке элементов он атомарно увеличивает общий счётчик:

for (const item of workerData.chunk) {
  processItem(item);
  Atomics.add(progress, 0, 1); // every worker safely increments the same shared counter
}

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

Большинству кодов не следует прибегать к этому

Типичный код backend на Node редко нуждается в подобных решениях. Большая часть задач, связанных с конкурентностью в API-сервере, связана с операциями ввода-вывода, которые уже хорошо обрабатываются однопоточным циклом событий. Для задач, требующих высокой производительности процессора, таких как обработка изображений, преобразование данных или криптография, обычно достаточно рабочих потоков с использованием простого механизма копирования через postMessage.

SharedArrayBuffer и Atomics предназначены для более узкой ситуации: высокочастотной координации между параллельными потоками, когда копирование данных при каждом обмене само по себе становится узким местом. Использование общей памяти до того, как будет подтверждена наличие этой проблемы, не делает код быстрее; оно лишь вводит проблемы конкуренции на уровне аппаратного обеспечения, которые модель копирования была создана для предотвращения.

Основные выводы

• postMessage создает структурированные копии данных, так что каждый поток работает со своей копией, и изменения никогда не перекрываются.
• SharedArrayBuffer позволяет нескольким потокам получать прямой доступ к одним и тем же байтам через представления типизированных массивов.
• Простое присваивание x = x + 1 в общей памяти может привести к потере обновлений без видимых последствий; для каждой записи следует использовать Atomics.add и подобные методы.

Atomics.wait действительно блокирует поток и предназначен для использования в рабочих потоках; его следует сочетать с Atomics.store и Atomics.notify.

Обращайтесь к общей памяти только после того, как будет установлено, что копирование данных является узким местом.