Implement Barrier

Author: W4G1

README.md

@@ -282,6 +282,39 @@ spawn(move(mutex, cv), async (mutex, cv) => {
 });
 ```
 
+### 5\. Barrier
+
+A `Barrier` acts as a synchronization checkpoint. It blocks all participating threads until a specific number (`N`) of threads have reached the barrier. Once the last thread arrives, all threads are released simultaneously.
+
+Barriers are **cyclic**, meaning they automatically reset and can be reused immediately for subsequent phases of work (e.g., in iterative algorithms like simulations).
+
+**Leader Election:** The `wait()` method returns an object `{ isLeader: boolean }`. Exactly **one** thread is guaranteed to be the leader per cycle. This is useful for performing single-threaded setup or cleanup tasks between parallel phases.
+
+```typescript
+import { spawn, move, Barrier } from "multithreading";
+
+const N = 4;
+const barrier = new Barrier(N);
+
+for (let i = 0; i < N; i++) {
+  spawn(move(barrier), async (b) => {
+    // Phase 1
+    console.log("Working on Phase 1...");
+    
+    // Wait for all 4 threads to reach this point
+    const res = await b.wait();
+    
+    // Only one thread runs this code
+    if (res.isLeader) {
+      console.log("All threads finished Phase 1. preparing Phase 2...");
+    }
+    
+    // Phase 2 (All threads start this simultaneously)
+    console.log("Working on Phase 2...");
+  });
+}
+```
+
 -----
 
 ## Channels (MPMC)
@@ -440,6 +473,9 @@ Content-Security-Policy: default-src 'self'; worker-src 'self' blob:; script-src
       * `notifyOne()`: Wake one waiting thread.
       * `notifyAll()`: Wake all waiting threads.
       * `blockingWait(guard)`: Blocking wait (Halts Worker).
+  * **`Barrier`**:
+      * `wait()`: Async wait (Recommended). Returns `Promise<{ isLeader: boolean }>`.
+      * `blockingWait()`: Blocking wait (Halts Worker). Returns `{ isLeader: boolean }`.
 
 -----
 

src/lib/check_move_args.ts

@@ -1,4 +1,12 @@
-import { Condvar, Mutex, Receiver, RwLock, Semaphore, Sender } from "./lib.ts";
+import {
+  Barrier,
+  Condvar,
+  Mutex,
+  Receiver,
+  RwLock,
+  Semaphore,
+  Sender,
+} from "./lib.ts";
 import { toSerialized } from "./shared.ts";
 
 export function checkMoveArgs(args: any[]) {
@@ -9,7 +17,8 @@ export function checkMoveArgs(args: any[]) {
       arg.buffer instanceof SharedArrayBuffer;
     const isThreadSafe = arg instanceof Mutex || arg instanceof Condvar ||
       arg instanceof RwLock || arg instanceof Sender ||
-      arg instanceof Receiver || arg instanceof Semaphore;
+      arg instanceof Receiver || arg instanceof Semaphore ||
+      arg instanceof Barrier;
     const isLibrarySAB = !isThreadSafe &&
       typeof arg[toSerialized] !== "undefined";
 

src/lib/sync/barrier.ts

@@ -0,0 +1,171 @@
+import {
+  register,
+  Serializable,
+  toDeserialized,
+  toSerialized,
+} from "../shared.ts";
+
+const IDX_LOCK = 0;
+const IDX_CAP = 1; // Capacity (N)
+const IDX_COUNT = 2; // Current count (Starts at 0, goes to N, or N down to 0)
+const IDX_GEN = 3; // Generation ID
+
+const META_SIZE = 4;
+
+const LOCK_UNLOCKED = 0;
+const LOCK_LOCKED = 1;
+
+export interface BarrierWaitResult {
+  isLeader: boolean;
+}
+
+export class Barrier extends Serializable {
+  static {
+    register(8, this);
+  }
+
+  readonly #state: Int32Array<SharedArrayBuffer>;
+
+  constructor(n?: number, _buffer?: SharedArrayBuffer) {
+    super();
+    if (_buffer) {
+      this.#state = new Int32Array(_buffer);
+    } else {
+      if (n === undefined) {
+        throw new Error("Barrier capacity must be provided");
+      }
+      this.#state = new Int32Array(
+        new SharedArrayBuffer(META_SIZE * Int32Array.BYTES_PER_ELEMENT),
+      );
+      this.#state[IDX_LOCK] = LOCK_UNLOCKED;
+      this.#state[IDX_CAP] = n;
+      this.#state[IDX_COUNT] = n; // We count down from N to 0
+      this.#state[IDX_GEN] = 0;
+    }
+  }
+
+  /**
+   * Internal Spin/Wait Lock for protecting state updates.
+   * Updates are very fast, so we use a simple atomic lock mechanism.
+   */
+  #lock() {
+    while (
+      Atomics.compareExchange(
+        this.#state,
+        IDX_LOCK,
+        LOCK_UNLOCKED,
+        LOCK_LOCKED,
+      ) !== LOCK_UNLOCKED
+    ) {
+      Atomics.wait(this.#state, IDX_LOCK, LOCK_LOCKED);
+    }
+  }
+
+  #unlock() {
+    if (
+      Atomics.compareExchange(
+        this.#state,
+        IDX_LOCK,
+        LOCK_LOCKED,
+        LOCK_UNLOCKED,
+      ) !== LOCK_LOCKED
+    ) {
+      throw new Error("Barrier lock state corrupted");
+    }
+    Atomics.notify(this.#state, IDX_LOCK, 1);
+  }
+
+  /**
+   * Async version of the internal lock for Main Thread compatibility.
+   */
+  async #lockAsync() {
+    while (
+      Atomics.compareExchange(
+        this.#state,
+        IDX_LOCK,
+        LOCK_UNLOCKED,
+        LOCK_LOCKED,
+      ) !== LOCK_UNLOCKED
+    ) {
+      const res = Atomics.waitAsync(this.#state, IDX_LOCK, LOCK_LOCKED);
+      if (res.async) {
+        await res.value;
+      }
+    }
+  }
+
+  /**
+   * Blocks the current thread until all participating threads have reached the barrier.
+   */
+  blockingWait(): BarrierWaitResult {
+    this.#lock();
+    const localGen = Atomics.load(this.#state, IDX_GEN);
+    const count = Atomics.load(this.#state, IDX_COUNT) - 1;
+
+    if (count === 0) {
+      // We are the leader (the last one to arrive)
+      Atomics.store(this.#state, IDX_COUNT, Atomics.load(this.#state, IDX_CAP));
+      Atomics.add(this.#state, IDX_GEN, 1);
+      this.#unlock();
+
+      // Wake everyone up. They are waiting on IDX_GEN changing.
+      Atomics.notify(this.#state, IDX_GEN, Infinity);
+      return { isLeader: true };
+    } else {
+      // We are a follower
+      Atomics.store(this.#state, IDX_COUNT, count);
+      this.#unlock();
+
+      // Wait until the generation changes
+      while (Atomics.load(this.#state, IDX_GEN) === localGen) {
+        Atomics.wait(this.#state, IDX_GEN, localGen);
+      }
+      return { isLeader: false };
+    }
+  }
+
+  /**
+   * Asynchronously waits until all participating threads have reached the barrier.
+   */
+  async wait(): Promise<BarrierWaitResult> {
+    await this.#lockAsync();
+    const localGen = Atomics.load(this.#state, IDX_GEN);
+    const count = Atomics.load(this.#state, IDX_COUNT) - 1;
+
+    if (count === 0) {
+      // We are the leader
+      Atomics.store(this.#state, IDX_COUNT, Atomics.load(this.#state, IDX_CAP));
+      Atomics.add(this.#state, IDX_GEN, 1);
+      this.#unlock();
+
+      Atomics.notify(this.#state, IDX_GEN, Infinity);
+      return { isLeader: true };
+    } else {
+      // We are a follower
+      Atomics.store(this.#state, IDX_COUNT, count);
+      this.#unlock();
+
+      // Wait until the generation changes
+      while (Atomics.load(this.#state, IDX_GEN) === localGen) {
+        const res = Atomics.waitAsync(this.#state, IDX_GEN, localGen);
+        if (res.async) {
+          await res.value;
+        }
+      }
+      return { isLeader: false };
+    }
+  }
+
+  [toSerialized]() {
+    return {
+      value: this.#state.buffer,
+      transfer: [],
+    };
+  }
+
+  static override [toDeserialized](
+    buffer: ReturnType<Barrier[typeof toSerialized]>["value"],
+  ) {
+    return new Barrier(undefined, buffer);
+  }
+}

src/lib/sync/condvar.ts

@@ -8,6 +8,7 @@ import {
 import type { SharedMemoryView } from "../types.ts";
 
 const IDX_NOTIFY_SEQ = 0;
+
 const SEQ_INCREMENT = 1;
 const NOTIFY_ONE = 1;
 const NOTIFY_ALL = Infinity;

src/lib/sync/mod.ts

@@ -3,3 +3,4 @@ export * from "./mutex.ts";
 export * from "./rwlock.ts";
 export * from "./semaphore.ts";
 export * from "./mpmc.ts";
+export * from "./barrier.ts";

src/lib/sync/mpmc.ts

@@ -16,6 +16,7 @@ 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;

src/lib/sync/mutex.ts

@@ -13,6 +13,7 @@ export const INTERNAL_MUTEX_CONTROLLER = Symbol(
 );
 
 const IDX_LOCK_STATE = 0;
+
 const LOCKED = 1;
 const UNLOCKED = 0;
 

src/lib/sync/rwlock.ts

@@ -17,6 +17,7 @@ export interface RwLockController {
 }
 
 const IDX_LOCK_STATE = 0;
+
 const UNLOCKED = 0;
 const WRITE_LOCKED = -1;
 const READ_ONE = 1;

src/lib/sync/semaphore.ts

@@ -16,6 +16,8 @@ export interface SemaphoreController {
 const IDX_PERMITS = 0;
 const IDX_WAITERS = 1;
 
+const META_SIZE = 2;
+
 export class SemaphoreGuard implements Disposable {
   readonly #amount: number;
   readonly #controller: SemaphoreController;
@@ -60,7 +62,7 @@ export class Semaphore extends Serializable {
       this.#state = new Int32Array(_buffer);
     } else {
       this.#state = new Int32Array(
-        new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * 2),
+        new SharedArrayBuffer(META_SIZE * Int32Array.BYTES_PER_ELEMENT),
       );
       this.#state[IDX_PERMITS] = initialCount;
       this.#state[IDX_WAITERS] = 0;