Implement mutex notifications using MessagePort

tshemsedinov · tshemsedinov · commit 31d6b314b0d5 · 2019-03-23T02:07:39.000+02:00
diff --git a/JavaScript/8-async.js b/JavaScript/8-async.js
@@ -0,0 +1,92 @@
+'use strict';
+
+const threads = require('worker_threads');
+const { Worker, isMainThread } = threads;
+
+const LOCKED = 0;
+const UNLOCKED = 1;
+
+class Mutex {
+  constructor(messagePort, shared, offset = 0, initial = false) {
+    this.port = messagePort;
+    this.lock = new Int32Array(shared, offset, 1);
+    if (initial) Atomics.store(this.lock, 0, UNLOCKED);
+    this.owner = false;
+    this.resolve = null;
+    if (messagePort) {
+      messagePort.on('message', kind => {
+        if (kind === 'leave') this.tryEnter();
+      });
+    }
+  }
+
+  enter() {
+    return new Promise(resolve => {
+      this.resolve = resolve;
+      this.tryEnter();
+    });
+  }
+
+  tryEnter() {
+    if (!this.resolve) return;
+    let prev = Atomics.exchange(this.lock, 0, LOCKED);
+    if (prev === UNLOCKED) {
+      this.owner = true;
+      this.resolve();
+      this.resolve = null;
+    }
+  };
+
+  leave() {
+    if (!this.owner) return false;
+    Atomics.store(this.lock, 0, UNLOCKED);
+    this.port.postMessage('leave');
+    this.owner = false;
+    return true;
+  }
+}
+
+class Thread {
+  constructor(data) {
+    const worker = new Worker(__filename, { workerData: data });
+    this.worker = worker;
+    Thread.workers.add(worker);
+    worker.on('message', kind => {
+      for (const next of Thread.workers) {
+        if (next !== worker) {
+          next.postMessage(kind);
+        }
+      }
+    })
+  }
+}
+
+Thread.workers = new Set();
+
+// Usage
+
+if (isMainThread) {
+  const buffer = new SharedArrayBuffer(4);
+  const mutex = new Mutex(null, buffer, 0, true);
+  console.dir({ mutex });
+  new Thread(buffer);
+  new Thread(buffer);
+} else {
+  const { threadId, workerData, parentPort } = threads;
+  const mutex = new Mutex(parentPort, workerData);
+
+  setInterval(() => {
+    console.log(`Interval ${threadId}`);
+  }, 500);
+
+  const f = async () => {
+    await mutex.enter();
+    console.log(`Enter ${threadId}`);
+    setTimeout(() => {
+      mutex.leave();
+      console.log(`Leave ${threadId}`);
+      setTimeout(f, 0);
+    }, 5000);
+  };
+  f();
+}
