Node.js Worker Pool with worker

A WorkerPool implementation using Node.js worker_threads: fixed-size pool, job queue, in-flight tracking, and graceful error handling when a worker crashes mid-job.

This creates a WorkerPool with a predefined number of workers, each consuming a thread. Threads do not share memory by default, but data can be copied via postMessage.

JavaScript

import { Worker } from "node:worker_threads";

export class WorkerPool {
  constructor({ size, workerFile }) {
    this.size = size;
    this.workerFile = workerFile;
    this.workers = [];
    this.idle = [];
    this.queue = [];
    this.inflight = new Map(); // jobId -> { resolve, reject }

this.nextJobId = 1;

for (let i = 0; i < size; i++) {
      const w = new Worker(workerFile, { stdout: false, stderr: false });
      w.on("message", (msg) => this.#onMessage(w, msg));
      w.on("error", (err) => this.#onError(w, err));
      w.on("exit", (code) => this.#onExit(w, code));
      this.workers.push(w);
      this.idle.push(w);
    }
  }

run(job) {
    return new Promise((resolve, reject) => {
      const jobId = this.nextJobId++;
      this.queue.push({ jobId, job, resolve, reject });
      this.#drain();
    });
  }

async runAll(jobs) {
    // Wait for all jobs; pool ensures bounded parallelism
    await Promise.all(jobs.map((j) => this.run(j)));
  }

async close() {
    await Promise.all(this.workers.map(w => w.terminate()));
  }

#drain() {
    while (this.idle.length > 0 && this.queue.length > 0) {
      const worker = this.idle.pop();
      const { jobId, job, resolve, reject } = this.queue.shift();

this.inflight.set(jobId, { resolve, reject, worker });
      worker.postMessage({ type: "job", jobId, job });
    }
  }

#onMessage(worker, msg) {
    if (msg?.type === "done") {
      const rec = this.inflight.get(msg.jobId);
      if (!rec) return;

this.inflight.delete(msg.jobId);
      this.idle.push(worker);
      rec.resolve(msg.result);
      this.#drain();
      return;
    }

if (msg?.type === "error") {
      const rec = this.inflight.get(msg.jobId);
      if (!rec) return;

this.inflight.delete(msg.jobId);
      this.idle.push(worker);
      rec.reject(Object.assign(new Error(msg.error?.message || "worker error"), { cause: msg.error }));
      this.#drain();
    }
  }

#onError(worker, err) {
    // Fail the job that was running on this worker (if any)
    for (const [jobId, rec] of this.inflight.entries()) {
      if (rec.worker === worker) {
        this.inflight.delete(jobId);
        rec.reject(err);
      }
    }
  }

#onExit(worker, code) {
    if (code !== 0) {
      // Same idea: fail any inflight job on this worker
      for (const [jobId, rec] of this.inflight.entries()) {
        if (rec.worker === worker) {
          this.inflight.delete(jobId);
          rec.reject(new Error(`worker exited with code ${code}`));
        }
      }
    }
  }
}

WorkerPool entrypoint

someJob is some async task you've defined in someJob.js

JavaScript

import { parentPort } from "node:worker_threads";
import { someJob } from "./someJob.js";

parentPort.on("message", async (msg) => {
  if (msg?.type !== "job") return;

const { jobId, job } = msg;

try {
    await someJob(job); // same function as in promise version
    parentPort.postMessage({ type: "done", jobId, result: null });
  } catch (err) {
    parentPort.postMessage({
      type: "error",
      jobId,
      error: { message: err?.message || String(err), stack: err?.stack },
    });
  }
});