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
- read current
i value
- increment
i by 1
- 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();
})();
}
