multithreading.js

The library provides a Multi-Producer, Multi-Consumer (MPMC) bounded channel that acts as a work-stealing queue. It handles backpressure (blocking send() when full) and load-balances messages across receivers (blocking recv() when empty).

Key characteristics:

• Exactly-once delivery: Each message is delivered to exactly one consumer.
• JSON Support: Uses SharedJsonBuffer to allow sending any JSON-serializable value (objects, arrays, etc.).
• Bounded: You define a capacity.
• Clonable: Sender and Receiver can be cloned and moved to different workers.
• Reference Counted: The channel closes automatically when all Sender handles or all Receiver handles are dropped.

import { spawn, move, channel } from "multithreading";

const [tx, rx] = channel();

// Producer
spawn(move(tx), async (sender) => {
  await sender.send({ hello: "world" });
  await sender.send({ hello: "multithreading" });
});

// Consumer
await spawn(move(rx.clone()), async (receiver) => {
  const result = await receiver.recv();
  console.log("Worker got:", result.value); // { hello: "world" }
}).join();

// Because we cloned rx, the main thread also still has a rx handle
for await (const value of rx) {
  console.log("Main thread got:", value); // { hello: "multithreading" }
}

The library supports standard dynamic await import() calls inside workers. It automatically handles path resolution so that imports are relative to the file that called spawn, rather than the worker's internal location.

You can import:

1. External Libraries: npm packages or CDNs.
2. Relative Files: Local modules relative to the caller.

Note: The function passed to spawn must be self-contained or explicitly import its dependencies. It cannot access variables from the outer scope unless they are passed via move().

// main.ts
import { spawn } from "multithreading";

spawn(async () => {
  // Importing relative files
  const utils = await import("./utils.ts");
  // Importing external libraries
  const { v4: uuiv4 } = await import("uuid");

  console.log("Magic number from relative file:", utils.magicNumber);
  console.log("Random UUID:", uuiv4());
  
  return utils.magicNumber;
});

A Mutex ensures only one thread can access specific data at a time.

Automatic Management (Recommended)

Use the using keyword (Explicit Resource Management) to automatically release the lock when the guard goes out of scope.

Manual Management (Bun / Standard JS)

If using Bun (where using may fail in worker contexts) or standard JS, you must manually call .dispose() on the guard. Always use a try...finally block to ensure the lock is released.

Best Practice: Use asynchronous methods (lock()) to avoid halting the entire Worker thread.

import { spawn, move, Mutex } from "multithreading";

// Option A: Automatic Management
const buffer = new SharedArrayBuffer(4);
const counterMutex = new Mutex(new Int32Array(buffer));

spawn(move(counterMutex), async (mutex) => {
  using guard = await mutex.lock();
  guard.value[0]++;
});

// Option B: Manual Management (Required for Bun)
const counterMutexManual = new Mutex(new Int32Array(new SharedArrayBuffer(4)));

spawn(move(counterMutexManual), async (mutex) => {
  const guard = await mutex.lock();
  try {
    guard.value[0]++;
  } finally {
    guard.dispose();
  }
});

SharedJsonBuffer allows for Mutex-protected shared memory for JSON objects. It is optimized for high-performance state synchronization of large, persistent objects by using Proxies to reserialize only changed bytes rather than the entire tree.

Note: SharedJsonBuffer has an initialization cost. It is best used for frequent incremental updates to large objects, rather than single-use transfers where standard cloning is faster.

import { spawn, move, Mutex, SharedJsonBuffer } from "multithreading";

const sharedState = new Mutex(new SharedJsonBuffer({
  score: 0,
  players: ["Main Thread"],
  level: {
    id: 1,
    title: "Start",
  },
}));

await spawn(move(sharedState), async (sharedState) => {
  using guard = await sharedState.lock();

  const state = guard.value;

  console.log(`Current Score: ${state.score}`);

  // Modify the data
  state.score += 100;
  state.players.push("Worker1");

  // End of scope: Lock is automatically released here
}).join();

// Verify on main thread
using guard = await sharedState.lock();

console.log(guard.value); // { score: 100, players: ["Main Thread", "Worker1"], ... }