Node.Js Worker Threads: Scale CPU-Bound Work Across All Cores

const cluster = require('cluster');
const http = require('http');
const os = require('os');
const numCPUs = os.cpus().length;

if (cluster.isPrimary) {
  console.log(`Primary ${process.pid} starting ${numCPUs} workers`);

  // Fork one worker per CPU core
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  // Restart dead workers
  cluster.on('exit', (worker, code, signal) => {
    console.log(`Worker ${worker.process.pid} died (${signal || code}). Restarting...`);
    cluster.fork();
  });
} else {
  // Each worker runs a full Express server
  const express = require('express');
  const app = express();

  app.get('/api/data', async (req, res) => {
    res.json(await fetchData());
  });

  app.listen(3000, () => {
    console.log(`Worker ${process.pid} listening on port 3000`);
  });
}
// Result: 8 workers, 8x throughput, OS load-balances connections

# ecosystem.config.js
module.exports = {
  apps: [{
    name: 'api',
    script: './src/server.js',
    instances: 'max',      // One instance per CPU core
    exec_mode: 'cluster',  // Cluster mode — shared port
    max_memory_restart: '500M',
    env_production: {
      NODE_ENV: 'production',
      PORT: 3000
    }
  }]
};

# Commands:
pm2 start ecosystem.config.js --env production
pm2 list          # View all instances
pm2 logs api      # View logs
pm2 reload api    # Zero-downtime reload (restart workers one by one)
pm2 monit         # Real-time monitoring

// main.js — spawn workers for CPU-intensive tasks
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

if (isMainThread) {
  // Main thread: receive task, dispatch to worker
  function runWorker(data) {
    return new Promise((resolve, reject) => {
      const worker = new Worker(__filename, { workerData: data });
      worker.on('message', resolve);
      worker.on('error', reject);
      worker.on('exit', code => {
        if (code !== 0) reject(new Error(`Worker stopped with exit code ${code}`));
      });
    });
  }

  // Worker pool — reuse workers instead of spawning per task
  const { WorkerPool } = require('./worker-pool');
  const pool = new WorkerPool('./hasher.js', os.cpus().length);
  const hash = await pool.run({ input: largeData });

} else {
  // Worker thread: do CPU-intensive work
  const result = heavyComputation(workerData.input);
  parentPort.postMessage(result);
}

// SharedArrayBuffer: share memory between worker threads
const { SharedArrayBuffer, Atomics } = globalThis;

// Main thread:
const sharedBuffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * 1000);
const sharedArray = new Int32Array(sharedBuffer);

const worker = new Worker('./worker.js', {
  workerData: { sharedBuffer }
});

// Worker thread:
const sharedArray = new Int32Array(workerData.sharedBuffer);
// Use Atomics for thread-safe operations:
Atomics.add(sharedArray, 0, 1); // Atomic increment — no race conditions