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

Реалістичне застосування: показ прогресу в режимі реального часу через пул робочих потоків

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

// 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 є I/O-залежними, і їх добре обробляє однопотоковий цикл подій. Для завдань, що залежать від процесора, таких як обробка зображень, трансформація даних чи криптографія, зазвичай достатньо працівничих потоків із звичайним копіюванням за допомогою postMessage.

SharedArrayBuffer та Atomics призначені для більш вузької ситуації: високочастотної координації між паралельними потоками, де копіювання під час кожної зміни саме по собі стає вузьким місцем. Використання спільної пам’яті до того, як ви оціните цю проблему, не робить код швидшим; воно лише створює конкурентні ситуації на рівні апаратного забезпечення, які модель копіювання була розроблена для усунення.

Ключові висновки

• postMessage створює структуровані копії даних, тож кожен поток працює зі своєю копією, і зміни ніколи не поширюються між ними.
• SharedArrayBuffer дає кільком потокам прямий доступ до одних і тих самих байтів через перегляди типових масивів.
• Звичайна операція x = x + 1 у спільній пам’яті може мовчки втрачати оновлення; для кожного запису слід використовувати Atomics.add та подібні функції.

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

Звертайтеся до спільної пам’яті лише після перевірки того, що копіювання даних є вузьким місцем.