The missing standard library for multithreading in JavaScript

Multithreading.js – 安全で堅牢な WebWorker 並列処理を実現する TypeScript ライブラリ

概要

multithreading

• スレッドプールアーキテクチャ
• 厳格なメモリー安全セマンティクス
• 同期プリミティブ（Mutex、RwLock、Semaphore、Condvar）

WebWorker、シリアライズ、および

SharedArrayBuffer

インストール

npm install multithreading

コアコンセプト

速習例

import { spawn } from "multithreading";

// バックグラウンドスレッドでタスクを生成
const handle = spawn(() => {
  const result = Math.random();
  return result;
});

// 結果待ち
const result = await handle.join();

if (result.ok) {
  console.log("Result:", result.value);   // 例: 0.6378467071314606
} else {
  console.error("Worker error:", result.error);
}

データ転送：

move()

WebWorker は外部スコープの変数をキャプチャできないため、

move()

import { spawn, move } from "multithreading";

const largeData = new Uint8Array(10 * 1024 * 1024); // 10 MB
const metaData = { id: 1 };

const handle = spawn(move(largeData, metaData), (data, meta) => {
  console.log("Processing ID:", meta.id);
  return data.byteLength;
});

await handle.join();

• Transferable Objects – ゼロコピーで所有権を転送
• Non‑Transferable Objects –

  structuredClone

  によってクローン

Shared JSON バッファ

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

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

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

  console.log(`Current Score: ${state.score}`);
  state.score += 100;
  state.players.push("Worker1");
}).join();

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

同期プリミティブ

1. Mutex

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

const buffer = new SharedArrayBuffer(4);
const counterMutex = new Mutex(new Int32Array(buffer));

spawn(move(counterMutex), async (mutex) => {
  using guard = await mutex.acquire();   // スコープ終了時に自動解放
  guard.value[0]++;
});

• 自動 vs 手動管理
  

  using

  （推奨）を使用。

  using

  が利用できない環境では

  try…finally

  内で

  drop()

  を呼びます。

2. RwLock

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

const lock = new RwLock(new Int32Array(new SharedArrayBuffer(4)));

spawn(move(lock), async (l) => {
  using guard = await l.write();
  guard.value[0] = 42;
});

spawn(move(lock), async (l) => {
  using guard = await l.read();
  console.log(guard.value[0]);
});

3. Semaphore

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

const semaphore = new Semaphore(3);

for (let i = 0; i < 10; i++) {
  spawn(move(semaphore), async (sem) => {
    using _ = await sem.acquire();   // 許可が無ければ待機
    console.log("Acquired slot! Working...");
    await new Promise(r => setTimeout(r, 1000));
  });
}

4. Condvar

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

const mutex = new Mutex(new Int32Array(new SharedArrayBuffer(4)));
const cv = new Condvar();

spawn(move(mutex, cv), async (lock, cond) => {
  using guard = await lock.acquire();
  while (guard.value[0] === 0) {
    await cond.wait(guard);   // ロックを解放して待機、再取得
  }
  console.log("Received signal, value is:", guard.value[0]);
});

チャネル（MPMC）

SharedJsonBuffer

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

const [tx, rx] = channel<{ hello: string }>(); // デフォルト容量は1

// プロデューサー
spawn(move(tx), async (sender) => {
  await sender.send({ hello: "world" });
  await sender.send({ hello: "multithreading" });   // GC 時に自動クローズ
});

// コンシューマー
spawn(move(rx.clone()), async (receiver) => {
  for await (const value of receiver) {
    console.log(value);   // { hello: "world" }, 次に { hello: "multithreading" }
  }
});

主な特徴

• 任意の JSON データをサポート
• バウンド容量とバックプレッシャー
• クローン可能、参照カウンタ付きハンドル

ワーカー内でモジュールをインポート

ワーカーは

await import()

import { spawn } from "multithreading";

spawn(async () => {
  const utils = await import("./utils.ts");   // main.ts に対する相対パス
  const _ = await import("lodash");

  console.log("Magic number:", utils.magicNumber);
  console.log("Random number via lodash:", _.default.random(1, 100));
});

API リファレンス

spawn(fn)

spawn(move(...), fn)

initRuntime(config?)

shutdown()

move(...args)

drop(resource)

SharedJsonBuffer

channel<T>(capacity?)

[Sender<T>, Receiver<T>]

Sender<T>

send()

sendSync()

clone()

close()

Receiver<T>

recv()

recvSync()

clone()

close()

Mutex<T>

acquire()

tryLock()

acquireSync()

RwLock<T>

read()

write()

readSync()

writeSync()

Semaphore

acquire(amount?)

tryAcquire(amount?)

acquireSync(amount?)

Condvar

wait(guard)

notifyOne()

notifyAll()

waitSync(guard)

技術的実装詳細

• Serialization Protocol – “Envelope” 形式 (

  PayloadType.RAW

  /

  LIB

  ) を用い、ワーカー内で Mutex 等を再構築
• Atomics –

  Int32Array

  （

  SharedArrayBuffer

  ）上に構築し、

  Atomics.wait/notify

  を活用
• Import Patching –

  spawn

  は呼び出し元のスタックトレースを解析し、正しい相対パス解決のために

  import()

  呼び出しを正規表現で修正

これで JavaScript アプリケーションに安全で強力な並列処理を導入できます。

multithreading.js