Worker Pool Using worker_threads

import { Worker } from "node:worker_threads";

interface QueueItem {
  payload: unknown;
  resolve: (value: unknown) => void;
  reject: (reason?: unknown) => void;
}

interface PendingEntry {
  worker: Worker;
  resolve: (value: unknown) => void;
  reject: (reason?: unknown) => void;
  timer: ReturnType<typeof setTimeout>;
}

export interface WorkerPoolOptions {
  size?: number;
  workerSource?: string;
  workerUrl?: string | URL;
  taskTimeoutMs?: number;
}

export class WorkerPool {
  private readonly workers: Worker[] = [];
  private readonly idle: Worker[] = [];
  private readonly queue: QueueItem[] = [];
  private readonly pending = new Map<number, PendingEntry>();
  private readonly size: number;
  private readonly workerSource?: string;
  private readonly workerUrl?: string | URL;
  private readonly taskTimeoutMs: number;
  private nextId = 1;
  private closed = false;

  public constructor(options: WorkerPoolOptions) {
    const {
      size = 2,
      workerSource,
      workerUrl,
      taskTimeoutMs = 10_000,
    } = options;
    if (!Number.isInteger(size) || size < 1 || size > 64) {
      throw new RangeError("size must be an integer between 1 and 64");
    }
    if ((workerSource === undefined) === (workerUrl === undefined)) {
      throw new TypeError(
        "provide exactly one of workerSource or workerUrl",
      );
    }
    if (!Number.isFinite(taskTimeoutMs) || taskTimeoutMs <= 0) {
      throw new RangeError("taskTimeoutMs must be a positive finite number");
    }
    this.size = size;
    this.workerSource = workerSource;
    this.workerUrl = workerUrl;
    this.taskTimeoutMs = taskTimeoutMs;

    try {
      for (let index = 0; index < this.size; index += 1) {
        const worker = this.createWorker();
        this.workers.push(worker);
        this.idle.push(worker);
      }
    } catch (error) {
      for (const worker of this.workers) {
        void worker.terminate().catch(() => undefined);
      }
      throw error;
    }
  }

  public run<TResult = unknown>(payload: unknown): Promise<TResult> {
    if (this.closed) {
      return Promise.reject(new Error("worker pool is closed"));
    }
    return new Promise<TResult>((resolve, reject) => {
      this.queue.push({
        payload,
        resolve: resolve as (value: unknown) => void,
        reject,
      });
      this.dispatch();
    });
  }

  public async close(): Promise<void> {
    if (this.closed) return;
    this.closed = true;
    const error = new Error("worker pool is closed");
    for (const item of this.queue.splice(0)) item.reject(error);
    for (const entry of this.pending.values()) {
      clearTimeout(entry.timer);
      entry.reject(error);
    }
    this.pending.clear();
    const workers = this.workers.splice(0);
    this.idle.length = 0;
    await Promise.all(workers.map((worker) => worker.terminate()));
  }

  private createWorker(): Worker {
    const worker =
      this.workerSource !== undefined
        ? new Worker(this.workerSource, { eval: true })
        : new Worker(this.workerUrl as string | URL);
    worker.on("message", (message: unknown) => {
      this.handleMessage(worker, message);
    });
    worker.on("error", (error: Error) => {
      this.handleFailure(worker, error);
    });
    worker.on("exit", (code) => {
      if (!this.closed && this.workers.includes(worker)) {
        this.handleFailure(
          worker,
          new Error(`worker exited with code ${code}`),
        );
      }
    });
    return worker;
  }

  private dispatch(): void {
    while (
      !this.closed &&
      this.idle.length > 0 &&
      this.queue.length > 0
    ) {
      const worker = this.idle.pop();
      const task = this.queue.shift();
      if (worker === undefined || task === undefined) return;

      const id = this.nextId;
      this.nextId += 1;
      const timer = setTimeout(() => {
        if (this.closed) return;
        const entry = this.pending.get(id);
        if (entry === undefined) return;
        this.pending.delete(id);
        entry.reject(
          new Error(`task ${id} timed out after ${this.taskTimeoutMs}ms`),
        );
        this.replaceWorker(worker);
      }, this.taskTimeoutMs);

      this.pending.set(id, {
        worker,
        resolve: task.resolve,
        reject: task.reject,
        timer,
      });

      try {
        worker.postMessage({ id, payload: task.payload });
      } catch (error) {
        clearTimeout(timer);
        this.pending.delete(id);
        task.reject(error);
        this.idle.push(worker);
      }
    }
  }

  private handleMessage(worker: Worker, message: unknown): void {
    if (typeof message !== "object" || message === null) return;
    const data = message as {
      id?: unknown;
      ok?: unknown;
      result?: unknown;
      error?: unknown;
    };
    if (typeof data.id !== "number" || typeof data.ok !== "boolean") return;

    const entry = this.pending.get(data.id);
    if (entry === undefined || entry.worker !== worker) return;
    this.pending.delete(data.id);
    clearTimeout(entry.timer);

    if (data.ok) {
      entry.resolve(data.result);
    } else {
      entry.reject(
        new Error(
          typeof data.error === "string"
            ? data.error
            : "worker task failed",
        ),
      );
    }
    this.markIdle(worker);
  }

  private markIdle(worker: Worker): void {
    if (
      this.closed ||
      !this.workers.includes(worker) ||
      this.idle.includes(worker)
    ) {
      return;
    }
    this.idle.push(worker);
    this.dispatch();
  }

  private handleFailure(worker: Worker, error: unknown): void {
    const index = this.workers.indexOf(worker);
    if (index < 0) return;
    const failure =
      error instanceof Error ? error : new Error(String(error));
    this.workers.splice(index, 1);
    const idleIndex = this.idle.indexOf(worker);
    if (idleIndex >= 0) this.idle.splice(idleIndex, 1);

    for (const [id, entry] of this.pending) {
      if (entry.worker !== worker) continue;
      this.pending.delete(id);
      clearTimeout(entry.timer);
      entry.reject(failure);
    }

    void worker.terminate().catch(() => undefined);

    if (!this.closed) {
      try {
        const replacement = this.createWorker();
        this.workers.push(replacement);
        this.idle.push(replacement);
        this.dispatch();
      } catch (replacementError) {
        for (const item of this.queue.splice(0)) {
          item.reject(replacementError);
        }
      }
    }
  }

  private replaceWorker(worker: Worker): void {
    const index = this.workers.indexOf(worker);
    if (index < 0) return;
    this.workers.splice(index, 1);
    const idleIndex = this.idle.indexOf(worker);
    if (idleIndex >= 0) this.idle.splice(idleIndex, 1);
    void worker.terminate().catch(() => undefined);

    if (this.closed) return;
    try {
      const replacement = this.createWorker();
      this.workers.push(replacement);
      this.idle.push(replacement);
      this.dispatch();
    } catch (error) {
      for (const item of this.queue.splice(0)) item.reject(error);
    }
  }
}

const WORKER_SOURCE = `
const { parentPort } = require("node:worker_threads");
parentPort.on("message", ({ id, payload }) => {
  try {
    if (!payload || !Array.isArray(payload.values)) {
      throw new TypeError("payload.values must be an array");
    }
    const total = payload.values.reduce((sum, value) => {
      if (typeof value !== "number" || !Number.isFinite(value)) {
        throw new TypeError("values must be finite numbers");
      }
      return sum + value;
    }, 0);
    parentPort.postMessage({ id, ok: true, result: total });
  } catch (error) {
    parentPort.postMessage({
      id,
      ok: false,
      error: error instanceof Error ? error.message : String(error),
    });
  }
});
`;

async function main(): Promise<void> {
  const pool = new WorkerPool({
    size: 2,
    workerSource: WORKER_SOURCE,
    taskTimeoutMs: 2000,
  });

  const totals = await Promise.all([
    pool.run<number>({ values: [1, 2, 3] }),
    pool.run<number>({ values: [4, 5, 6] }),
    pool.run<number>({ values: [7, 8, 9] }),
  ]);
  console.log(totals);

  const failure = await pool
    .run({ values: [1, "x"] })
    .catch((error: unknown) =>
      error instanceof Error ? error.message : String(error),
    );
  console.log(failure);

  await pool.close();
  console.log("closed");
}

main().catch((error) => {
  console.error(error);
  process.exitCode = 1;
});