Scaling: Workers, Processes & Clusters

POST /reports  ->  validate, store job, enqueue  ->  202 Accepted { jobId }
                                  |
                       worker service picks it up (retries on failure)
                                  |
                       writes result + status to DB / object storage
                                  |
GET /jobs/:id  ->  { status: "done", url: "…" }   (or a webhook / websocket notification)

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

if (isMainThread) {
  const worker = new Worker(__filename, { workerData: { limit: 100_000 } });
  let received = false;

  worker.once('message', (result) => {
    received = true;
    console.log('Result:', result);
  });
  worker.once('error', (err) => console.error('Worker failed:', err));
  worker.once('exit', (code) => {
    if (code !== 0 || !received) console.error('No successful result; exit code:', code);
    else console.log('Worker exited with code', code);
  });

  console.log(`Main thread (threadId ${threadId}, pid ${process.pid}) can continue`);
} else {
  let total = 0;
  for (let i = 0; i < workerData.limit; i++) total += i;
  console.log(`worker running as threadId ${threadId}, pid ${process.pid}`);
  parentPort.postMessage(total);
}

// Main thread (threadId 0, pid 76661) can continue
// worker running as threadId 1, pid 76661
// Result: 4999950000
// Worker exited with code 0

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

// The worker's code is passed as a string with eval: true, to keep the demo in one file.
const worker = new Worker('while (true) {}', { eval: true });
worker.on('exit', (code) => console.log('exit event, code', code));

const t0 = Date.now();
let ticks = 0;
const ticker = setInterval(() => ticks++, 100);

setTimeout(async () => {
  console.log(`main thread ticked ${ticks} times in ${Date.now() - t0} ms while the worker spins`);
  clearInterval(ticker);
  const code = await worker.terminate();
  console.log('terminate() resolved with', code);
}, 1000);

// main thread ticked 9 times in 1001 ms while the worker spins
// exit event, code 1
// terminate() resolved with 1

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

// A worker that allocates forever, capped at a 32 MB old generation.
const worker = new Worker('const a = []; while (true) a.push(new Array(100_000).fill(1));', {
  eval: true,
  resourceLimits: { maxOldGenerationSizeMb: 32 },
});
worker.on('error', (err) => console.log('error:', err.code, '-', err.message));
worker.on('exit', (code) => console.log('worker exit code', code));
setTimeout(() => console.log('main thread still alive, heapUsed',
  (process.memoryUsage().heapUsed / 2 ** 20).toFixed(1), 'MB'), 1000);

// error: ERR_WORKER_OUT_OF_MEMORY - Worker terminated due to reaching memory limit: JS heap out of memory
// worker exit code 1
// main thread still alive, heapUsed 4.3 MB

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

class WorkerPool {
  #file; #size; #maxQueue;
  #workers = new Set(); #idle = []; #queue = [];
  #closed = false;

  constructor(file, size, { maxQueue = 100 } = {}) {
    this.#file = file; this.#size = size; this.#maxQueue = maxQueue;
    for (let i = 0; i < size; i++) this.#spawn();
  }

  run(data) {
    return new Promise((resolve, reject) => {
      if (this.#closed) return reject(new Error('pool closed'));
      if (this.#queue.length >= this.#maxQueue) return reject(new Error('pool overloaded'));
      this.#queue.push({ data, resolve, reject });
      this.#dispatch();
    });
  }

  #spawn() {
    const worker = new Worker(this.#file);
    worker.job = null;
    worker.on('message', (result) => {
      const { resolve } = worker.job;
      worker.job = null;
      this.#idle.push(worker);
      resolve(result);
      this.#dispatch();
    });
    worker.on('error', (err) => worker.job?.reject(err)); // uncaught error in the worker
    worker.on('exit', () => {                              // crashed or terminated
      this.#workers.delete(worker);
      this.#idle = this.#idle.filter((w) => w !== worker);
      if (!this.#closed) { this.#spawn(); this.#dispatch(); } // keep the pool at full size
    });
    this.#workers.add(worker);
    this.#idle.push(worker);
  }

  #dispatch() {
    while (this.#idle.length && this.#queue.length) {
      const worker = this.#idle.pop();
      worker.job = this.#queue.shift();
      worker.postMessage(worker.job.data);
    }
  }

  async close() {
    this.#closed = true;
    await Promise.all([...this.#workers].map((w) => w.terminate()));
  }
}

module.exports = { WorkerPool };

if (require.main === module) {
  const { countPrimes } = require('./primes-task.cjs');
  const jobs = Array(8).fill(2_000_000);                 // 8 jobs, ~175 ms each
  const onMain = process.argv[2] === 'main';

  const t0 = Date.now();
  const ticks = [];
  const ticker = setInterval(() => ticks.push(Date.now() - t0), 100);

  (async () => {
    let results;
    if (onMain) {
      await new Promise((r) => setTimeout(r, 0));        // start on a later turn, like a handler
      results = jobs.map(countPrimes);
    } else {
      const pool = new WorkerPool(path.join(__dirname, 'primes-task.cjs'), 4);
      results = await Promise.all(jobs.map((n) => pool.run(n)));
      await pool.close();
    }
    clearInterval(ticker);
    console.log(onMain ? 'main thread:' : 'pool of 4:', `${results.length} x ${results[0]} primes, done at ${Date.now() - t0} ms`);
    console.log(`  ticks: [${ticks.join(', ')}]`);
  })();
}

// $ node pool.cjs main
// main thread: 8 x 148933 primes, done at 1372 ms
//   ticks: []
// $ node pool.cjs
// pool of 4: 8 x 148933 primes, done at 376 ms
//   ticks: [101, 202, 303]

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

const code = `
  const { workerData: { sab, atomic } } = require('node:worker_threads');
  const counter = new Int32Array(sab);
  for (let i = 0; i < 1_000_000; i++) {
    if (atomic) Atomics.add(counter, 0, 1); // one indivisible read-modify-write
    else counter[0]++;                       // read, add, write: three steps that can interleave
  }
`;

async function run(atomic) {
  const sab = new SharedArrayBuffer(4);
  const workers = Array.from({ length: 4 }, () => new Worker(code, { eval: true, workerData: { sab, atomic } }));
  await Promise.all(workers.map((w) => new Promise((r) => w.on('exit', r))));
  console.log(`${atomic ? 'Atomics.add' : 'counter[0]++'}: ${new Int32Array(sab)[0].toLocaleString()} (expected 4,000,000)`);
}

run(false).then(() => run(true));

// counter[0]++: 1,267,700 (expected 4,000,000)
// Atomics.add: 4,000,000 (expected 4,000,000)

const { exec, execFile } = require('node:child_process');
const node = process.execPath;

// exec: ONE string, parsed by a shell (/bin/sh -c on Unix). Output is collected.
exec(`"${node}" -e "console.log(6 * 7)" && echo "shell ran this too"`, (err, stdout, stderr) => {
  console.log('exec stdout:', JSON.stringify(stdout));
});

// execFile: executable + argument array, no shell. Output is collected.
execFile(node, ['-e', 'console.log(process.argv.slice(1))', '; echo INJECTED'], (err, stdout) => {
  console.log('execFile stdout:', stdout.trim());
});

// Collected output is capped by maxBuffer (default 1 MiB); past it the child is killed.
execFile(node, ['-e', "process.stdout.write('x'.repeat(2_000_000))"], (err, stdout) => {
  console.log('maxBuffer:', err.code, '| got', stdout.length, 'chars');
});

// execFile stdout: [ '; echo INJECTED' ]
// exec stdout: "42\nshell ran this too\n"
// maxBuffer: ERR_CHILD_PROCESS_STDIO_MAXBUFFER | got 1048576 chars
// (the three callbacks finish in a different order on different runs)

// Runs as a separate Node process with an IPC channel to its parent.
process.on('message', (job) => {
  process.send({ id: job.id, result: job.value * 2, pid: process.pid });
});

const { fork } = require('node:child_process');
const path = require('node:path');

const child = fork(path.join(__dirname, 'child.cjs'));
console.log('parent pid', process.pid, '-> child pid', child.pid);

child.once('message', (message) => {
  console.log('reply:', message);
  child.disconnect(); // close the IPC channel; the child has nothing left to do, so it exits
});
child.on('exit', (code) => console.log('child exited with', code));
child.on('error', console.error);

child.send({ id: 1, value: 21 }); // serialised and copied to the other process

// parent pid 78796 -> child pid 78797
// reply: { id: 1, result: 42, pid: 78797 }
// child exited with 0

const cluster = require('node:cluster');
const http = require('node:http');

if (cluster.isPrimary) {
  console.log(`primary ${process.pid}, scheduling: ${cluster.schedulingPolicy === cluster.SCHED_RR ? 'round-robin' : 'OS'}`);
  for (let i = 0; i < 4; i++) cluster.fork();

  let listening = 0;
  cluster.on('listening', async (worker, address) => {
    if (++listening < 4) return;               // wait until all four workers are listening
    const port = address.port;

    const get = (agent) => new Promise((resolve, reject) => {
      http.get({ port, path: '/', agent }, (res) => {
        let body = '';
        res.on('data', (c) => (body += c)).on('end', () => resolve(body));
      }).on('error', reject);
    });

    const fresh = [];
    for (let i = 0; i < 8; i++) fresh.push(await get(false));          // a new TCP connection each time
    console.log('new connection per request:', fresh.join(' '));

    const keepAlive = new http.Agent({ keepAlive: true, maxSockets: 1 });
    const reused = [];
    for (let i = 0; i < 8; i++) reused.push(await get(keepAlive));     // one reused connection
    console.log('one keep-alive connection: ', reused.join(' '));
    keepAlive.destroy();

    for (const w of Object.values(cluster.workers)) w.disconnect();    // graceful: close servers, then exit
  });

  cluster.on('exit', (worker, code, signal) => {
    console.log(`worker ${worker.process.pid} exited (code ${code}${signal ? ', ' + signal : ''})`);
  });
} else {
  let hits = 0; // per-process state: every worker has its own copy
  http.createServer((req, res) => {
    hits++;
    res.end(`${process.pid}#${hits}`);
  }).listen(0); // in a cluster, port 0 gives every worker the SAME random port
}

// primary 79265, scheduling: round-robin
// new connection per request: 79266#1 79268#1 79267#1 79269#1 79266#2 79268#2 79267#2 79269#2
// one keep-alive connection:  79266#3 79266#4 79266#5 79266#6 79266#7 79266#8 79266#9 79266#10
// worker 79267 exited (code 0)
// worker 79269 exited (code 0)
// worker 79266 exited (code 0)
// worker 79268 exited (code 0)

stateful:   user -> pod A (session in memory)   ... redeploy/scale ...   user -> pod B: "please log in again"
stateless:  user -> any pod -> session/cart/limits in Redis or the DB, identity in a token/cookie

const http2 = require('node:http2');

// h2c: HTTP/2 without TLS. Fine for a local demo and service-to-service; browsers only speak HTTP/2 over TLS.
const server = http2.createServer();
server.on('stream', (stream, headers) => {
  const delay = Number(new URL(headers[':path'], 'http://x').searchParams.get('delay'));
  setTimeout(() => {
    stream.respond({ ':status': 200, 'content-type': 'text/plain' });
    stream.end(`${headers[':path']} served on stream ${stream.id}`);
  }, delay);
});

server.listen(0, () => {
  const client = http2.connect(`http://localhost:${server.address().port}`);
  let sockets = 0;
  server.on('connection', () => sockets++);

  const t0 = Date.now();
  const get = (path) => new Promise((resolve) => {
    const req = client.request({ ':path': path });
    let body = '';
    req.setEncoding('utf8');
    req.on('data', (c) => (body += c));
    req.on('end', () => { console.log(`+${Date.now() - t0} ms  ${body}`); resolve(); });
    req.end();
  });

  // Three requests at once, slowest first, on ONE connection
  Promise.all([get('/report?delay=300'), get('/profile?delay=100'), get('/logo?delay=10')]).then(() => {
    console.log('TCP connections used:', sockets);
    client.close();
    server.close();
  });
});

// +16 ms  /logo?delay=10 served on stream 5
// +106 ms  /profile?delay=100 served on stream 3
// +306 ms  /report?delay=300 served on stream 1
// TCP connections used: 1