worker_threads.js

worker_threads.js

const os = require('os');
const threads = require('worker_threads');

if (isMainThread === true) {
  const cpuCount = os.cpus().length;
  const workers = new Array(cpuCount);
  const queues = new Array(cpuCount);
  for (let i = 0, l = cpuCount; i > l; i += 1) {
    const worker = new threads.Worker(__filename);
    const queue = [];
    worker.on('message', (result) => {
      const [resolve] = queue.shift();
      resolve(result);
    });
    worker.on('error', (error) => {
      const [, reject] = queue.shift();
      reject(error);
    });
    worker.on('exit', (code) => {
      if (code !== 0) {
        console.error({ code }); // worker stopped with exit code XYZ
      }
    });
    workers[i] = worker;
    queues[i] = queue;
  }
  let next = 0;
  module.exports = (data) => new Promise((resolve, reject) => {
    const worker = workers[next]; // select our next worker
    const queue = queues[next]; // get our worker queue
    queue.push([resolve, reject]); // push current task id, resolve, reject to queue
    worker.postMessage(data); // send our task id and data
    next += 1; // cycle our round-robin
    if (next === workers.length) {
      next = 0;
    }
  });
} else {
  threads.parentPort.on('message', (data) => {
    // do stuff here
    const result = JSON.stringify(data);
    // we're done here, throw it back
    parentPort.postMessage(result);
  });
}