Refactor sync primitives

Author: W4G1

src/lib/json_buffer.ts

@@ -53,7 +53,10 @@ export function serialize(arg: any): {
   }
 
   // Library Object (Instance of Serializable)
-  if (typeof arg === "object" && typeof arg[toSerialized] === "function") {
+  if (
+    typeof arg === "object" && arg !== null &&
+    typeof arg[toSerialized] === "function"
+  ) {
     const { value, transfer, typeId } = arg[toSerialized]();
     const Ctor = arg.constructor as SerializableConstructor;
 

src/lib/shared.ts

@@ -7,6 +7,11 @@ import {
 } from "../shared.ts";
 import type { SharedMemoryView } from "../types.ts";
 
+const IDX_NOTIFY_SEQ = 0;
+const SEQ_INCREMENT = 1;
+const NOTIFY_ONE = 1;
+const NOTIFY_ALL = Infinity;
+
 export class Condvar extends Serializable {
   static {
     register(1, this);
@@ -16,37 +21,35 @@ export class Condvar extends Serializable {
 
   constructor(_buffer?: SharedArrayBuffer) {
     super();
-    if (_buffer) {
-      this.#atomic = new Int32Array(_buffer);
-    } else {
-      this.#atomic = new Int32Array(new SharedArrayBuffer(4));
-    }
+    this.#atomic = new Int32Array(
+      _buffer ?? new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT),
+    );
   }
 
   blockingWait<T extends SharedMemoryView | void>(guard: MutexGuard<T>): void {
     const controller = guard[INTERNAL_MUTEX_CONTROLLER];
-    const seq = Atomics.load(this.#atomic, 0);
+    const seq = Atomics.load(this.#atomic, IDX_NOTIFY_SEQ);
 
     controller.unlock();
 
-    Atomics.wait(this.#atomic, 0, seq);
+    Atomics.wait(this.#atomic, IDX_NOTIFY_SEQ, seq);
 
     controller.blockingLock();
   }
 
   /**
    * Asynchronously waits for a notification. Safe to use on the Main Thread.
-   * * @param guard The MutexGuard protecting the shared state.
+   * @param guard The MutexGuard protecting the shared state.
    */
   async wait<T extends SharedMemoryView | void>(
     guard: MutexGuard<T>,
   ): Promise<void> {
     const controller = guard[INTERNAL_MUTEX_CONTROLLER];
-    const seq = Atomics.load(this.#atomic, 0);
+    const seq = Atomics.load(this.#atomic, IDX_NOTIFY_SEQ);
 
     controller.unlock();
 
-    const result = Atomics.waitAsync(this.#atomic, 0, seq);
+    const result = Atomics.waitAsync(this.#atomic, IDX_NOTIFY_SEQ, seq);
     if (result.async) {
       await result.value;
     }
@@ -58,16 +61,16 @@ export class Condvar extends Serializable {
    * Wakes up one blocked thread waiting on this Condvar.
    */
   notifyOne() {
-    Atomics.add(this.#atomic, 0, 1);
-    Atomics.notify(this.#atomic, 0, 1);
+    Atomics.add(this.#atomic, IDX_NOTIFY_SEQ, SEQ_INCREMENT);
+    Atomics.notify(this.#atomic, IDX_NOTIFY_SEQ, NOTIFY_ONE);
   }
 
   /**
    * Wakes up all blocked threads waiting on this Condvar.
    */
   notifyAll() {
-    Atomics.add(this.#atomic, 0, 1);
-    Atomics.notify(this.#atomic, 0, Infinity);
+    Atomics.add(this.#atomic, IDX_NOTIFY_SEQ, SEQ_INCREMENT);
+    Atomics.notify(this.#atomic, IDX_NOTIFY_SEQ, NOTIFY_ALL);
   }
 
   [toSerialized]() {

src/lib/sync/condvar.ts

@@ -10,14 +10,17 @@ import type { Result } from "../types.ts";
 import { INTERNAL_SEMAPHORE_CONTROLLER, Semaphore } from "./semaphore.ts";
 import { SharedJsonBuffer } from "../json_buffer.ts";
 
-const HEAD_IDX = 0;
-const TAIL_IDX = 1;
-const CLOSED_IDX = 2;
-const CAP_IDX = 3;
-const TX_COUNT_IDX = 4;
-const RX_COUNT_IDX = 5;
+const IDX_HEAD = 0;
+const IDX_TAIL = 1;
+const IDX_CLOSED = 2;
+const IDX_CAP = 3;
+const IDX_TX_COUNT = 4;
+const IDX_RX_COUNT = 5;
 const META_SIZE = 6;
 
+const OPEN = 0;
+const CLOSED = 1;
+
 const ERR_DISPOSED_SENDER = new Error("Sender is disposed");
 const ERR_DISPOSED_RECEIVER = new Error("Receiver disposed");
 const ERR_CLOSED = new Error("Channel closed");
@@ -41,29 +44,29 @@ class ChannelInternals<T> extends Serializable {
   }
 
   write(value: T): void {
-    const tail = this.state[TAIL_IDX]!;
+    const tail = this.state[IDX_TAIL]!;
     this.items[tail] = value;
-    this.state[TAIL_IDX] = (tail + 1) % this.state[CAP_IDX]!;
+    this.state[IDX_TAIL] = (tail + 1) % this.state[IDX_CAP]!;
   }
 
   read(): T | null {
-    const head = this.state[HEAD_IDX]!;
+    const head = this.state[IDX_HEAD]!;
     const val = this.items[head] as T;
 
     // Optimistic read check
     if (val === null) return null;
 
     this.items[head] = null;
-    this.state[HEAD_IDX] = (head + 1) % this.state[CAP_IDX]!;
+    this.state[IDX_HEAD] = (head + 1) % this.state[IDX_CAP]!;
     return val;
   }
 
   isClosed(): boolean {
-    return Atomics.load(this.state, CLOSED_IDX) === 1;
+    return Atomics.load(this.state, IDX_CLOSED) === CLOSED;
   }
 
   hasReceivers(): boolean {
-    return Atomics.load(this.state, RX_COUNT_IDX) > 0;
+    return Atomics.load(this.state, IDX_RX_COUNT) > 0;
   }
 
   [toSerialized]() {
@@ -140,7 +143,7 @@ export class Sender<T> extends ChannelHandle<T> {
 
   clone(): Sender<T> {
     if (this.disposed) throw new Error("Cannot clone disposed Sender");
-    Atomics.add(this.internals.state, TX_COUNT_IDX, 1);
+    Atomics.add(this.internals.state, IDX_TX_COUNT, 1);
     return new Sender(this.internals);
   }
 
@@ -224,7 +227,7 @@ export class Sender<T> extends ChannelHandle<T> {
 
     try {
       if (this.internals.isClosed()) return;
-      Atomics.store(state, CLOSED_IDX, 1);
+      Atomics.store(state, IDX_CLOSED, CLOSED);
       // Wake up everyone
       slotsAvailable[INTERNAL_SEMAPHORE_CONTROLLER].release(1_073_741_823);
       itemsAvailable[INTERNAL_SEMAPHORE_CONTROLLER].release(1_073_741_823);
@@ -236,7 +239,7 @@ export class Sender<T> extends ChannelHandle<T> {
 
   [Symbol.dispose]() {
     if (this.disposed) return;
-    const prevCount = Atomics.sub(this.internals.state, TX_COUNT_IDX, 1);
+    const prevCount = Atomics.sub(this.internals.state, IDX_TX_COUNT, 1);
     if (prevCount === 1) this.close();
     this.disposed = true;
   }
@@ -259,7 +262,7 @@ export class Receiver<T> extends ChannelHandle<T> {
 
   clone(): Receiver<T> {
     if (this.disposed) throw new Error("Cannot clone disposed Receiver");
-    Atomics.add(this.internals.state, RX_COUNT_IDX, 1);
+    Atomics.add(this.internals.state, IDX_RX_COUNT, 1);
     return new Receiver(this.internals);
   }
 
@@ -348,7 +351,7 @@ export class Receiver<T> extends ChannelHandle<T> {
   [Symbol.dispose]() {
     if (this.disposed) return;
     this.disposed = true;
-    const prevCount = Atomics.sub(this.internals.state, RX_COUNT_IDX, 1);
+    const prevCount = Atomics.sub(this.internals.state, IDX_RX_COUNT, 1);
     if (prevCount === 1) this.close();
   }
 
@@ -360,15 +363,16 @@ export class Receiver<T> extends ChannelHandle<T> {
 }
 
 export function channel<T>(capacity: number = 32): [Sender<T>, Receiver<T>] {
-  const stateSab = new SharedArrayBuffer(META_SIZE * 4);
-  const state = new Int32Array(stateSab);
-
-  state[CAP_IDX] = capacity;
-  state[HEAD_IDX] = 0;
-  state[TAIL_IDX] = 0;
-  state[CLOSED_IDX] = 0;
-  state[TX_COUNT_IDX] = 1;
-  state[RX_COUNT_IDX] = 1;
+  const state = new Int32Array(
+    new SharedArrayBuffer(META_SIZE * Int32Array.BYTES_PER_ELEMENT),
+  );
+
+  state[IDX_CAP] = capacity;
+  state[IDX_HEAD] = 0;
+  state[IDX_TAIL] = 0;
+  state[IDX_CLOSED] = OPEN;
+  state[IDX_TX_COUNT] = 1;
+  state[IDX_RX_COUNT] = 1;
 
   const initialData = new Array<T | null>(capacity).fill(null);
   const items = new SharedJsonBuffer(initialData);

src/lib/sync/mpmc.ts

@@ -12,7 +12,7 @@ export const INTERNAL_MUTEX_CONTROLLER = Symbol(
   "Thread.InternalMutexController",
 );
 
-const INDEX = 0;
+const IDX_LOCK_STATE = 0;
 const LOCKED = 1;
 const UNLOCKED = 0;
 
@@ -25,7 +25,7 @@ export interface MutexController {
 export class MutexGuard<T extends SharedMemoryView | void>
   implements Disposable {
   #data: T;
-  #mutex: MutexController;
+  readonly #mutex: MutexController;
   #released = false;
 
   constructor(data: T, mutex: MutexController) {
@@ -65,14 +65,14 @@ export class Mutex<T extends SharedMemoryView | void = void>
 
   #data: T;
   #lockState: Int32Array<SharedArrayBuffer>;
-  #controller: MutexController;
+  readonly #controller: MutexController;
 
-  constructor(data?: T, _existingLockBuffer?: SharedArrayBuffer) {
+  constructor(data?: T, _lockBuffer?: SharedArrayBuffer) {
     super();
     this.#data = data as T;
-    this.#lockState = _existingLockBuffer
-      ? new Int32Array(_existingLockBuffer)
-      : new Int32Array(new SharedArrayBuffer(4));
+    this.#lockState = _lockBuffer
+      ? new Int32Array(_lockBuffer)
+      : new Int32Array(new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT));
     this.#controller = {
       unlock: () => this.#unlock(),
       blockingLock: () => this.#performBlockingLock(),
@@ -82,19 +82,29 @@ export class Mutex<T extends SharedMemoryView | void = void>
 
   #tryLock(): boolean {
     return (
-      Atomics.compareExchange(this.#lockState, INDEX, UNLOCKED, LOCKED) ===
+      Atomics.compareExchange(
+        this.#lockState,
+        IDX_LOCK_STATE,
+        UNLOCKED,
+        LOCKED,
+      ) ===
         UNLOCKED
     );
   }
 
   #unlock(): void {
     if (
-      Atomics.compareExchange(this.#lockState, INDEX, LOCKED, UNLOCKED) !==
+      Atomics.compareExchange(
+        this.#lockState,
+        IDX_LOCK_STATE,
+        LOCKED,
+        UNLOCKED,
+      ) !==
         LOCKED
     ) {
       throw new Error("Mutex was not locked or locked by another thread");
     }
-    Atomics.notify(this.#lockState, INDEX, 1);
+    Atomics.notify(this.#lockState, IDX_LOCK_STATE, 1);
   }
 
   /**
@@ -104,7 +114,7 @@ export class Mutex<T extends SharedMemoryView | void = void>
   #performBlockingLock(): void {
     while (true) {
       if (this.#tryLock()) return;
-      Atomics.wait(this.#lockState, INDEX, LOCKED);
+      Atomics.wait(this.#lockState, IDX_LOCK_STATE, LOCKED);
     }
   }
 
@@ -115,7 +125,7 @@ export class Mutex<T extends SharedMemoryView | void = void>
   async #performAsyncLock(): Promise<void> {
     while (true) {
       if (this.#tryLock()) return;
-      const result = Atomics.waitAsync(this.#lockState, INDEX, LOCKED);
+      const result = Atomics.waitAsync(this.#lockState, IDX_LOCK_STATE, LOCKED);
       if (result.async) {
         await result.value;
       }