Ответ 1

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

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

Пример:

var arr = new SharedArrayBuffer(1024);

// send a reference to the memory to any number of webworkers
workers.forEach(worker => worker.postMessage(arr));

// Normally, simultaneous access to the memory from multiple threads 
// (where at least one access is a write)
// is not safe, but the Atomics methods are thread-safe.
// This adds 2 to element 0 of arr.
Atomics.add(arr, 0, 2)

SharedArrayBuffer был ранее включен в основных браузерах, но после инцидента с Spectre он был отключен, поскольку совместно используемая память позволяет реализовать таймеры наносекундной точности, которые позволяют использовать Spectre.

Чтобы сделать это безопасным, браузеры должны запускать страницы отдельным процессом для каждого домена. Chrome начал делать это в версии 67, а общая память была снова включена в версии 68.

Ответ 2

Если у вас есть какие-то сложные вычисления, вам могут понадобиться WebWorkers, чтобы ваш основной скрипт мог продолжать свою работу, пока тяжелые вещи выполняются параллельно.

Проблема, которую решает Atomics, заключается в том, как WebVorkers могут общаться друг с другом (легко, быстро и надежно). Вы можете прочитать о ArrayBuffer, SharedArrayBuffer, Atomics и о том, как вы можете использовать их для своих преимуществ, здесь.

Вы не должны беспокоиться об этом, если:

• Вы создаете что-то простое (например, магазин, форум и т.д.)

Вам может понадобиться, если:

• Вы хотите создать что-то сложное, занимающее много памяти (например, figma или google drive)
• Это необходимо, если вы хотите работать с WebAssembly или webgl и хотите оптимизировать производительность
• Также вам может понадобиться, если вы хотите создать какой-то сложный модуль Node.js
• Или, если вы создаете сложное приложение через Electron, такое как Skype или Discord

Спасибо Павло Мур и Саймон Пэрис за ваши ответы!

Ответ 3

В дополнение к тому, что заявили Арсений-II и Саймон Пэрис, Atomics также удобна, когда вы встраиваете движок JavaScript в какое-то хост-приложение (чтобы включить в нем скрипты). Затем можно напрямую обращаться к разделяемой памяти из разных параллельных потоков одновременно, как из JS, так и из C/C++ или любого языка, на котором написано ваше хост-приложение, без использования JavaScript API для доступа на стороне C/C++/OtherLanguage.

Ответ 4

Атомарная операция - это группа "все или ничего" из небольших операций.

Давайте посмотрим на

let i=0;

i++

i++ фактически оценивается с 3 шагами

1. читать текущее значение i
2. приращение i на 1
3. вернуть старое значение

Что произойдет, если у вас 2 потока выполняют одну и ту же операцию? они оба могут читать одно и то же значение 1 и увеличивать его в одно и то же время.

Но это и Javascript, разве это не однопоточный?

Да! JavaScript действительно один поток, но браузеры/узел позволяет сегодня использовать несколько сред выполнения JavaScript параллельно (Worker Threads, Web Workers).

Chrome и Node (на основе v8) создают Isolate для каждого потока, который все они запускают в своем собственном context.

И единственный способ, которым можно share memory - через ArrayBuffer/SharedArrayBuffer

Каким будет выход следующей программы?

Запустить с узлом> = 10

node --experimental_worker example.js

const { isMainThread, Worker, workerData } = require('worker_threads');

if (isMainThread) {
  const shm = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT);
  process.on('exit', () => {
    const res = new Int32Array(shm);
    console.log(res[0]); // expected 5 * 500,000 = 2,500,000
  });
  Array(5).fill(null).map(() => new Worker(__filename, { workerData: shm }));
} else {
  const arr = new Int32Array(workerData);
  for (let i = 0; i < 500000; i++) {
    arr[i]++;
  }
}

Вывод может быть 2,500,000 но мы этого не знаем, и в большинстве случаев это не будет 2,5 МБ, на самом деле, вероятность того, что вы получите один и тот же результат дважды, довольно низка, и, как программисты, мы, конечно же, этого не делаем. Мне не нравится код, который мы понятия не имеем, чем он закончится.

Это пример состояния состязания, когда n потоков состязаются друг с другом и никак не синхронизируются.

Здесь идет Atomic операция, которая позволяет нам выполнять арифметические операции от начала до конца.

Давайте немного изменим программу и теперь запустим:

const { isMainThread, Worker, workerData } = require('worker_threads');


if (isMainThread) {
    const shm = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT);
    process.on('exit', () => {
        const res = new Int32Array(shm);
        console.log(res[0]); // expected 5 * 500,000 = 2,500,000
    });
    Array(5).fill(null).map(() => new Worker(__filename, { workerData: shm }));
} else {
    const arr = new Int32Array(workerData);
    for (let i = 0; i < 500000; i++) {
        Atomics.add(arr, 0, 1);
    }
}

Теперь на выходе всегда будет 2,500,000

Бонус, Mutex с использованием Atomics

Иногда нам нужна операция, к которой одновременно может обращаться только 1 поток, давайте посмотрим на следующий класс

class Mutex {

    /**
     * 
     * @param {Mutex} mutex 
     * @param {Int32Array} resource 
     * @param {number} onceFlagCell 
     * @param {(done)=>void} cb
     */
    static once(mutex, resource, onceFlagCell, cb) {
        if (Atomics.load(resource, onceFlagCell) === 1) {
            return;
        }
        mutex.lock();
        // maybe someone already flagged it
        if (Atomics.load(resource, onceFlagCell) === 1) {
            mutex.unlock();
            return;
        }
        cb(() => {
            Atomics.store(resource, onceFlagCell, 1);
            mutex.unlock();
        });
    }
    /**
     * 
     * @param {Int32Array} resource 
     * @param {number} cell 
     */
    constructor(resource, cell) {
        this.resource = resource;
        this.cell = cell;
        this.lockAcquired = false;
    }

    /**
     * locks the mutex
     */
    lock() {
        if (this.lockAcquired) {
            console.warn('you already acquired the lock you stupid');
            return;
        }
        const { resource, cell } = this;
        while (true) {
            // lock is already acquired, wait
            if (Atomics.load(resource, cell) > 0) {
                while ('ok' !== Atomics.wait(resource, cell, 0));
            }
            const countOfAcquiresBeforeMe = Atomics.add(resource, cell, 1);
            // someone was faster than me, try again later
            if (countOfAcquiresBeforeMe >= 1) {
                Atomics.sub(resource, cell, 1);
                continue;
            }
            this.lockAcquired = true;
            return;
        }
    }

    /**
     * unlocks the mutex
     */
    unlock() {
        if (!this.lockAcquired) {
            console.warn('you didn\'t acquire the lock you stupid');
            return;
        }
        Atomics.sub(this.resource, this.cell, 1);
        Atomics.notify(this.resource, this.cell, 1);
        this.lockAcquired = false;
    }
}

Теперь вам нужно выделить SharedArrayBuffer и разделить их между всеми потоками и увидеть, что каждый раз, когда только 1 поток проходит в critical section

Запустить с узлом> 10

node --experimental_worker example.js

const { isMainThread, Worker, workerData, threadId } = require('worker_threads');


const { promisify } = require('util');
const doSomethingFakeThatTakesTimeAndShouldBeAtomic = promisify(setTimeout);

if (isMainThread) {
    const shm = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT);
    Array(5).fill(null).map(() => new Worker(__filename, { workerData: shm }));
} else {
    (async () => {
        const arr = new Int32Array(workerData);
        const mutex = new Mutex(arr, 0);
        mutex.lock();
        console.log('[${threadId}] ${new Date().toISOString()}');
        await doSomethingFakeThatTakesTimeAndShouldBeAtomic(1000);
        mutex.unlock();
    })();
}