Use async/await with semaphores

Author: tshemsedinov

JavaScript/8-binary-async.js

@@ -0,0 +1,54 @@
+'use strict';
+
+const threads = require('worker_threads');
+const { Worker, isMainThread } = threads;
+
+const LOCKED = 0;
+const UNLOCKED = 1;
+
+class BinarySemaphore {
+  constructor(shared, offset = 0, init = false) {
+    this.lock = new Int32Array(shared, offset, 1);
+    if (init) Atomics.store(this.lock, 0, UNLOCKED);
+  }
+
+  async enter() {
+    let prev = Atomics.exchange(this.lock, 0, LOCKED);
+    while (prev !== UNLOCKED) {
+      Atomics.wait(this.lock, 0, LOCKED);
+      prev = Atomics.exchange(this.lock, 0, LOCKED);
+    }
+    Atomics.store(this.lock, 0, LOCKED);
+  }
+
+  leave() {
+    if (Atomics.load(this.lock, 0) === UNLOCKED) {
+      throw new Error('Cannot leave unlocked BinarySemaphore');
+    }
+    Atomics.store(this.lock, 0, UNLOCKED);
+    Atomics.notify(this.lock, 0, 1);
+  }
+}
+
+// Usage
+
+if (isMainThread) {
+  const buffer = new SharedArrayBuffer(14);
+  const semaphore = new BinarySemaphore(buffer, 0, true);
+  console.dir({ semaphore });
+  new Worker(__filename, { workerData: buffer });
+  new Worker(__filename, { workerData: buffer });
+} else {
+  const { threadId, workerData } = threads;
+  const semaphore = new BinarySemaphore(workerData);
+  const array = new Int8Array(workerData, 4);
+  const value = threadId === 1 ? 1 : -1;
+  setInterval(async () => {
+    await semaphore.enter();
+    for (let i = 0; i < 10; i++) {
+      array[i] += value;
+    }
+    console.dir([ threadId, semaphore.lock[0], array ]);
+    semaphore.leave();
+  }, 100);
+}

JavaScript/9-counting-async.js

@@ -0,0 +1,52 @@
+'use strict';
+
+const fs = require('fs');
+const threads = require('worker_threads');
+const { Worker, isMainThread } = threads;
+
+class CountingSemaphore {
+  constructor(shared, offset = 0, initial) {
+    this.counter = new Int32Array(shared, offset, 1);
+    if (typeof initial === 'number') {
+      Atomics.store(this.counter, 0, initial);
+    }
+  }
+
+  async enter() {
+    Atomics.wait(this.counter, 0, 0);
+    const prev = Atomics.sub(this.counter, 0, 1);
+    if (prev > 0) return;
+    this.leave();
+    return await this.enter();
+  }
+
+  leave() {
+    Atomics.add(this.counter, 0, 1);
+    Atomics.notify(this.counter, 0, 1);
+  }
+}
+
+// Usage
+
+if (isMainThread) {
+  const buffer = new SharedArrayBuffer(4);
+  const semaphore = new CountingSemaphore(buffer, 0, 2);
+  console.dir({ semaphore: semaphore.counter[0] });
+  for (let i = 0; i < 20; i++) {
+    new Worker(__filename, { workerData: buffer });
+  }
+} else {
+  const { threadId, workerData } = threads;
+  const semaphore = new CountingSemaphore(workerData);
+  const REPEAT_COUNT = 1000000;
+  const file = `file-${threadId}.dat`;
+  console.dir({ threadId, semaphore: semaphore.counter[0] });
+
+  (async () => {
+    await semaphore.enter();
+    const data = `Data from ${threadId}`.repeat(REPEAT_COUNT);
+    await fs.promises.writeFile(file, data);
+    await fs.promises.unlink(file);
+    semaphore.leave();
+  })();
+}