スループット向け再利用可能なワーカープールの構築

Why a Worker Pool?

The Blocking Problem

function fib(n) {
  return n < 2 ? n : fib(n - 1) + fib(n - 2);
}

let ticks = 0;
const timer = setInterval(() => {
  console.log('heartbeat', ++ticks);
  if (ticks >= 3) clearInterval(timer);
}, 50);

console.log('start blocking work');
console.log('fib(38) =', fib(38)); // blocks the event loop
console.log('done blocking work');

A Single Worker Thread

const { Worker, isMainThread, parentPort } = require('node:worker_threads');

if (isMainThread) {
  const worker = new Worker(__filename);
  worker.on('message', (result) => {
    console.log('result =', result);
    worker.terminate();
  });
  worker.postMessage(40);
} else {
  parentPort.on('message', (n) => {
    const fib = (x) => (x < 2 ? x : fib(x - 1) + fib(x - 2));
    parentPort.postMessage(fib(n));
  });
}

Designing the Pool's Pieces

The Worker Script (worker.js)

// worker.js
const { parentPort } = require('node:worker_threads');

function heavyTask(n) {
  const fib = (x) => (x < 2 ? x : fib(x - 1) + fib(x - 2));
  return fib(n);
}

parentPort.on('message', ({ id, payload }) => {
  try {
    const result = heavyTask(payload);
    parentPort.postMessage({ id, result });
  } catch (err) {
    parentPort.postMessage({ id, error: err.message });
  }
});

Pool Skeleton: Spawning Workers

const { Worker } = require('node:worker_threads');
const os = require('node:os');

class WorkerPool {
  constructor(workerPath, size = os.cpus().length) {
    this.workerPath = workerPath;
    this.idle = [];
    this.queue = [];
    this.pending = new Map(); // id -> { resolve, reject }
    this.nextId = 0;
    for (let i = 0; i < size; i++) this._spawn();
  }

  _spawn() {
    const worker = new Worker(this.workerPath);
    worker.on('message', (msg) => this._onResult(worker, msg));
    worker.on('error', (err) => this._onError(worker, err));
    this.idle.push(worker);
  }
}

Submitting Tasks and the Queue

  run(payload) {
    return new Promise((resolve, reject) => {
      const id = this.nextId++;
      this.pending.set(id, { resolve, reject });
      this.queue.push({ id, payload });
      this._dispatch();
    });
  }

  _dispatch() {
    if (this.queue.length === 0 || this.idle.length === 0) return;
    const worker = this.idle.pop();
    const task = this.queue.shift();
    worker._currentId = task.id;
    worker.postMessage(task);
  }

Recycling: Handling Results

  _onResult(worker, msg) {
    const { id, result, error } = msg;
    const job = this.pending.get(id);
    this.pending.delete(id);
    worker._currentId = null;
    this.idle.push(worker);   // recycle the worker
    this._dispatch();          // pull the next queued task
    if (!job) return;
    if (error) job.reject(new Error(error));
    else job.resolve(result);
  }

Recycling on Failure

  _onError(worker, err) {
    const id = worker._currentId;
    if (id != null && this.pending.has(id)) {
      this.pending.get(id).reject(err);
      this.pending.delete(id);
    }
    // remove the dead worker, keep pool size constant
    this.idle = this.idle.filter((w) => w !== worker);
    worker.terminate();
    this._spawn();
    this._dispatch();
  }

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

A Complete, Runnable Pool

const { Worker, isMainThread, parentPort } = require('node:worker_threads');
const os = require('node:os');

if (!isMainThread) {
  const fib = (x) => (x < 2 ? x : fib(x - 1) + fib(x - 2));
  parentPort.on('message', ({ id, payload }) => {
    parentPort.postMessage({ id, result: fib(payload) });
  });
} else {
  class Pool {
    constructor(size) {
      this.idle = []; this.queue = []; this.pending = new Map(); this.id = 0;
      for (let i = 0; i < size; i++) this._spawn();
    }
    _spawn() {
      const w = new Worker(__filename);
      w.on('message', ({ id, result }) => {
        this.pending.get(id).resolve(result);
        this.pending.delete(id);
        this.idle.push(w); this._dispatch();
      });
      this.idle.push(w);
    }
    _dispatch() {
      if (!this.queue.length || !this.idle.length) return;
      const w = this.idle.pop(); const t = this.queue.shift();
      w.postMessage(t);
    }
    run(payload) {
      return new Promise((resolve) => {
        const id = this.id++;
        this.pending.set(id, { resolve });
        this.queue.push({ id, payload }); this._dispatch();
      });
    }
    destroy() { this.idle.forEach((w) => w.terminate()); }
  }

  const pool = new Pool(Math.min(4, os.cpus().length));
  const jobs = [30, 31, 32, 33, 30, 31, 32, 33];
  Promise.all(jobs.map((n) => pool.run(n))).then((results) => {
    console.log('results:', results);
    pool.destroy();
  });
}

Sizing and Tuning for Throughput

const buf = new Uint8Array(1024 * 1024).fill(7);
// Transfer the buffer instead of copying it (zero-copy handoff)
worker.postMessage({ id, payload: buf }, [buf.buffer]);
// After transfer, buf is detached in the sender: buf.byteLength === 0

Quick Check

Recap