What's the actual use of the Atomics object in ECMAScript?

Question

The ECMAScript specification defines the Atomics object in the section 24.4.

Among all the global objects this is the more obscure for me since I didn't know about its existence until I didn't read its specification, and also Google hasn't many references to it (or maybe the name is too much generic and everything gets submerged?).

According its official definition

The Atomics object provides functions that operate indivisibly (atomically) on shared memory array cells as well as functions that let agents wait for and dispatch primitive events

So it has the shape of an object with a number of methods to handle low-level memory and regulate the access to it. And also its public interface makes me suppose it. But what's the actual use of such object for the end-user? Why is it public? Are there some examples where it can be useful?

Thank you

Answer 1

Atomic operation is "everything or nothing" group of smaller operations.

Let's have a look at

let i=0;

i++

i++ is actually evaluated with 3 steps

1. read current i value
2. increment i by 1
3. return the old value

What happens if you have 2 threads doing the same operation? they can both read the same value 1 and increment it at the exact same time.

But this and Javascript, isn't it's single threaded?

Yes! JavaScript indeed single threads but browsers / node allows today usage of several JavaScript runtimes in parallel (Worker Threads, Web Workers).

Chrome and Node (v8 based) creates Isolate for each thread, which they all run in their own context.

And the only way the can share memory is via ArrayBuffer / SharedArrayBuffer

What will be the output of the next program?

Run with node > =10 (you might need --experimental_worker flag)

node example.js

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

if (isMainThread) {
  // main thread, create shared memory to share between threads
  const shm = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT);

  process.on('exit', () => {
    // print final counter
    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 {
  // worker thread, iteratres 500k and doing i++
  const arr = new Int32Array(workerData);
  for (let i = 0; i < 500000; i++) {
    arr[i]++;
  }
}

The output might be 2,500,000 but we don't know that and in most of the cases it won't be 2.5M, actually, the chance that you'll get the same output twice is pretty low, and as programmers we surely don't like code that we have no idea how it's going to end.

This is an example for race condition, where n threads race each other and not synced in any way.

Here comes the Atomic operation, that allows us to make arithmetic operations from start to end.

Let's change the program a bit and now run:

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);
    }
}

Now the output will always be 2,500,000

Bonus, Mutex using Atomics

Sometimes, we wish to an operation that only 1 thread can access at the same time, let's have a look at the next class

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;
    }
}

Now, you need to allocate SharedArrayBuffer and share them between all the threads and see that each time only 1 threads go inside the critical section

Run with node > 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();
    })();
}