Worker Threads

import { Worker, isMainThread, parentPort } from 'node:worker_threads';
import { createHash } from 'node:crypto';

const work = () => {
  let h = Buffer.from('seed');
  for (let i = 0; i < 1_000_000; i++) h = createHash('sha256').update(h).digest();
  return h.toString('hex').slice(0, 8);
};

if (!isMainThread) {
  parentPort.postMessage(work());
} else {
  async function measure(label, fn) {
    let maxGap = 0, last = performance.now();
    const ticker = setInterval(() => {
      const now = performance.now();
      maxGap = Math.max(maxGap, now - last);
      last = now;
    }, 10);
    const start = performance.now();
    const result = await fn();
    const ms = Math.round(performance.now() - start);
    await new Promise((r) => setTimeout(r, 30));          // let a late tick report the gap
    clearInterval(ticker);
    console.log(label, { result, ms, maxTickGapMs: Math.round(maxGap) });
  }
  await measure('on the main thread', async () => { await new Promise((r) => setImmediate(r)); return work(); });
  await measure('in a worker       ', () => new Promise((resolve, reject) => {
    const w = new Worker(new URL(import.meta.url));
    w.once('message', resolve);
    w.once('error', reject);
  }));
}

on the main thread { result: '6871b211', ms: 474, maxTickGapMs: 474 }
in a worker        { result: '6871b211', ms: 402, maxTickGapMs: 11 }

import { Worker, isMainThread, parentPort, workerData, threadId } from 'node:worker_threads';

if (isMainThread) {
  const worker = new Worker(new URL(import.meta.url), { workerData: { title: 'Dune', n: 35 } });
  worker.once('online', () => console.log('main: worker', worker.threadId, 'is online'));
  worker.on('message', (msg) => console.log('main: got', msg));
  worker.once('error', (err) => console.error('main: worker threw', err.message));
  worker.once('exit', (code) => console.log('main: worker exited with', code));
} else {
  const fib = (n) => (n < 2 ? n : fib(n - 1) + fib(n - 2));
  parentPort.postMessage({ from: threadId, title: workerData.title, fib: fib(workerData.n) });
}

main: worker 1 is online
main: got { from: 1, title: 'Dune', fib: 9227465 }
main: worker exited with 0

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

if (isMainThread) {
  const worker = new Worker(__filename, { workerData: [1, 2, 3] });
  worker.once('message', (sum) => console.log('sum from worker:', sum));
} else {
  parentPort.postMessage(workerData.reduce((a, b) => a + b, 0));
}

import { Worker, isMainThread, parentPort, workerData, threadId, threadName, setEnvironmentData, getEnvironmentData, SHARE_ENV } from 'node:worker_threads';

if (isMainThread) {
  process.env.BOOKSHELF_MODE = 'main-value';
  setEnvironmentData('locale', 'es-ES');                    // cloned into every Worker created afterwards
  const results = [];
  for (const options of [
    { name: 'copy-env' },                                  // default: process.env is copied
    { name: 'share-env', env: SHARE_ENV },                 // same object, read and write
    { name: 'custom-env', env: { ONLY_THIS: '1' } },
  ]) {
    await new Promise((resolve) => {
      const w = new Worker(new URL(import.meta.url), { ...options, argv: ['--verbose'], workerData: options.name });
      w.on('message', (m) => results.push(m));
      w.once('exit', resolve);
    });
    console.log(options.name, '-> main sees WORKER_WROTE =', process.env.WORKER_WROTE);
    delete process.env.WORKER_WROTE;
  }
  console.table(results);
} else {
  process.env.WORKER_WROTE = 'from-worker';
  parentPort.postMessage({
    name: workerData, threadId, threadName,
    modeVisible: process.env.BOOKSHELF_MODE ?? '(unset)',
    onlyThis: process.env.ONLY_THIS ?? '(unset)',
    locale: getEnvironmentData('locale'),
    argv: process.argv.slice(2).join(' '),
  });
}

copy-env -> main sees WORKER_WROTE = undefined
share-env -> main sees WORKER_WROTE = from-worker
custom-env -> main sees WORKER_WROTE = undefined
┌─────────┬──────────────┬──────────┬──────────────┬──────────────┬───────────┬─────────┬─────────────┐
│ (index) │ name         │ threadId │ threadName   │ modeVisible  │ onlyThis  │ locale  │ argv        │
├─────────┼──────────────┼──────────┼──────────────┼──────────────┼───────────┼─────────┼─────────────┤
│ 0       │ 'copy-env'   │ 1        │ 'copy-env'   │ 'main-value' │ '(unset)' │ 'es-ES' │ '--verbose' │
│ 1       │ 'share-env'  │ 2        │ 'share-env'  │ 'main-value' │ '(unset)' │ 'es-ES' │ '--verbose' │
│ 2       │ 'custom-env' │ 3        │ 'custom-env' │ '(unset)'    │ '1'       │ 'es-ES' │ '--verbose' │
└─────────┴──────────────┴──────────┴──────────────┴──────────────┴───────────┴─────────┴─────────────┘

import { Worker } from 'node:worker_threads';

const w = new Worker('console.log("from worker"); console.error("worker stderr")', { eval: true, stdout: true, stderr: true });
let captured = '';
w.stdout.on('data', (d) => { captured += `[out] ${d}`; });
w.stderr.on('data', (d) => { captured += `[err] ${d}`; });
w.once('exit', () => console.log(captured.trim()));

[out] from worker
[err] worker stderr

import { MessageChannel } from 'node:worker_threads';

const { port1, port2 } = new MessageChannel();
port1.on('message', (m) => { console.log('received', m); });
port1.on('messageerror', (e) => console.log('messageerror', e.message));

const circular = { name: 'loop' }; circular.self = circular;
port2.postMessage({ when: new Date(0), tags: new Set(['a']), map: new Map([[1, 'x']]), big: 10n, circular, re: /x/g });

try { port2.postMessage({ fn() {} }); } catch (e) { console.log(e.name, '-', e.message); }
class Book { constructor(t) { this.title = t; } get upper() { return this.title.toUpperCase(); } }
port2.postMessage(new Book('Dune'));               // arrives as a plain object

const bytes = new Uint8Array([1, 2, 3, 4]);
port2.postMessage(bytes);                           // copy
console.log('after copy    :', bytes.byteLength);
port2.postMessage(bytes, [bytes.buffer]);           // transfer
console.log('after transfer:', bytes.byteLength, bytes.buffer.detached);
setTimeout(() => port1.close(), 50);

DataCloneError - fn() {} could not be cloned.
after copy    : 4
after transfer: 0 true
received {
  when: 1970-01-01T00:00:00.000Z,
  tags: Set(1) { 'a' },
  map: Map(1) { 1 => 'x' },
  big: 10n,
  circular: <ref *1> { name: 'loop', self: [Circular *1] },
  re: /x/g
}
received { title: 'Dune' }
received Uint8Array(4) [ 1, 2, 3, 4 ]
received Uint8Array(4) [ 1, 2, 3, 4 ]

import { Worker, isMainThread, workerData } from 'node:worker_threads';

const N = 200_000, THREADS = 4;
if (isMainThread) {
  for (const mode of ['plain', 'atomic']) {
    const sab = new SharedArrayBuffer(4);
    const counter = new Int32Array(sab);
    await Promise.all(Array.from({ length: THREADS }, () => new Promise((res, rej) => {
      const w = new Worker(new URL(import.meta.url), { workerData: { sab, mode } });
      w.once('exit', res); w.once('error', rej);
    })));
    console.log(mode.padEnd(6), 'expected', N * THREADS, 'got', counter[0]);
  }
} else {
  const counter = new Int32Array(workerData.sab);
  for (let i = 0; i < N; i++) {
    if (workerData.mode === 'plain') counter[0]++;      // read-modify-write: a data race
    else Atomics.add(counter, 0, 1);                    // one indivisible step
  }
}

plain  expected 800000 got 338238
atomic expected 800000 got 800000

import { Worker, isMainThread, workerData } from 'node:worker_threads';

if (isMainThread) {
  const flag = new Int32Array(new SharedArrayBuffer(8));   // [0]=state, [1]=result
  const w = new Worker(new URL(import.meta.url), { workerData: { buf: flag.buffer } });
  // asynchronous wait: allowed on the main thread
  const { async, value } = Atomics.waitAsync(flag, 0, 0, 5000);
  console.log('main: waiting, async =', async);
  console.log('main: woke with', await value, 'result =', Atomics.load(flag, 1));
  await new Promise((r) => w.once('exit', r));
} else {
  const flag = new Int32Array(workerData.buf);
  Atomics.store(flag, 1, 42);
  Atomics.store(flag, 0, 1);
  Atomics.notify(flag, 0);                                   // wake one waiter
  // blocking wait is fine inside a worker (never on the main thread)
  console.log('worker: Atomics.wait ->', Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50));
}

main: waiting, async = true
main: woke with ok result = 42
worker: Atomics.wait -> timed-out

import { locks } from 'node:worker_threads';
await locks.request('bookshelf-index', { mode: 'exclusive' }, async (lock) => {
  console.log('holding', lock.name, lock.mode);
});
console.log(JSON.stringify(await locks.query()));

holding bookshelf-index exclusive
{"held":[],"pending":[]}

import { Worker } from 'node:worker_threads';

const w = new Worker(`
  const keep = [];
  for (;;) keep.push(new Array(1e5).fill('x'.repeat(10)));
`, { eval: true, resourceLimits: { maxOldGenerationSizeMb: 16, maxYoungGenerationSizeMb: 4 } });

console.log('limits seen by parent:', w.resourceLimits);
w.on('error', (e) => console.log('error event:', e.code, '-', e.message));
w.on('exit', (code) => console.log('exit code:', code));

limits seen by parent: {
  maxYoungGenerationSizeMb: 4,
  maxOldGenerationSizeMb: 16,
  codeRangeSizeMb: -1,
  stackSizeMb: 4
}
error event: ERR_WORKER_OUT_OF_MEMORY - Worker terminated due to reaching memory limit: JS heap out of memory
exit code: 1

import { Worker, isMainThread, parentPort } from 'node:worker_threads';

if (isMainThread) {
  const w = new Worker(new URL(import.meta.url));
  await new Promise((r) => w.once('online', r));
  const before = w.performance.eventLoopUtilization();
  await new Promise((r) => setTimeout(r, 300));
  const elu = w.performance.eventLoopUtilization(before);
  console.log('worker ELU while spinning:', elu.utilization.toFixed(2));
  const heap = await w.getHeapStatistics();
  console.log('worker heap used > 0:', heap.used_heap_size > 0);
  const cpu = await w.cpuUsage();
  console.log('worker cpu keys:', Object.keys(cpu).join(','));
  await w.terminate();
} else {
  for (;;) Math.sqrt(Math.random());
}

worker ELU while spinning: 1.00
worker heap used > 0: true
worker cpu keys: user,system

import { Worker } from 'node:worker_threads';

const times = [];
for (let i = 0; i < 20; i++) {
  const t = performance.now();
  await new Promise((resolve) => {
    const w = new Worker('require("node:worker_threads").parentPort.postMessage(1)', { eval: true });
    w.once('message', () => { times.push(performance.now() - t); w.terminate().then(resolve); });
  });
}
times.sort((a, b) => a - b);
console.log('round trip to a fresh worker (ms): median', times[10].toFixed(1), 'max', times.at(-1).toFixed(1));

const rss0 = process.memoryUsage().rss;
const ws = Array.from({ length: 8 }, () => new Worker('setInterval(() => {}, 1000)', { eval: true }));
await Promise.all(ws.map((w) => new Promise((r) => w.once('online', r))));
await new Promise((r) => setTimeout(r, 300));
console.log('RSS growth per idle worker (MiB):', ((process.memoryUsage().rss - rss0) / 8 / 2 ** 20).toFixed(1));
await Promise.all(ws.map((w) => w.terminate()));

import { parentPort } from 'node:worker_threads';
import { createHash } from 'node:crypto';

// One job = one cover image (an ArrayBuffer that was transferred to us).
parentPort.on('message', ({ id, bytes }) => {
  try {
    const data = new Uint8Array(bytes);
    let digest = createHash('sha256').update(data).digest();
    for (let i = 0; i < 150_000; i++) {                      // stand-in for real CPU work: perceptual hash, resize...
      digest = createHash('sha256').update(digest).update(data.subarray(0, 64)).digest();
    }
    parentPort.postMessage({ id, hash: digest.toString('hex') });
  } catch (error) {
    parentPort.postMessage({ id, error: { name: error.name, message: error.message } });
  }
});

import { Worker } from 'node:worker_threads';
import { availableParallelism } from 'node:os';
import { AsyncResource } from 'node:async_hooks';

export class WorkerPool {
  #file; #size; #idle = []; #all = new Set(); #queue = []; #jobs = new Map(); #nextId = 0; #closed = false;

  constructor(file, size = Math.max(1, availableParallelism() - 1)) {
    this.#file = file;
    this.#size = size;
  }

  /** Queue a job; resolves with the worker's answer. `transfer` lists buffers to move, not copy. */
  run(payload, transfer = []) {
    if (this.#closed) return Promise.reject(new Error('pool is closed'));
    return new Promise((resolve, reject) => {
      const id = this.#nextId++;
      const resource = new AsyncResource('WorkerPoolTask');  // keeps async stack traces / ALS context
      this.#queue.push({ id, payload, transfer, resolve: (v) => resource.runInAsyncScope(resolve, null, v), reject: (e) => resource.runInAsyncScope(reject, null, e) });
      this.#drain();
    });
  }

  #spawn() {
    const worker = new Worker(this.#file);
    worker.on('message', (msg) => this.#settle(worker, msg));
    worker.on('error', (err) => this.#crash(worker, err));         // uncaught error: the worker is already dead
    worker.on('exit', (code) => { if (this.#all.delete(worker)) this.#crash(worker, new Error(`worker exited with ${code}`), false); });
    this.#all.add(worker);
    return worker;
  }

  #drain() {
    while (this.#queue.length) {
      let worker = this.#idle.pop();
      if (!worker && this.#all.size < this.#size) worker = this.#spawn();
      if (!worker) return;                                          // everyone is busy; wait for a free worker
      const job = this.#queue.shift();
      this.#jobs.set(worker, job);
      worker.postMessage({ id: job.id, ...job.payload }, job.transfer);
    }
  }

  #settle(worker, msg) {
    const job = this.#jobs.get(worker);
    this.#jobs.delete(worker);
    this.#idle.push(worker);
    if (msg.error) job.reject(Object.assign(new Error(msg.error.message), { name: msg.error.name }));
    else job.resolve(msg);
    this.#drain();
  }

  #crash(worker, err, remove = true) {
    if (remove) this.#all.delete(worker);
    const job = this.#jobs.get(worker);
    this.#jobs.delete(worker);
    this.#idle = this.#idle.filter((w) => w !== worker);
    job?.reject(err);
    if (!this.#closed) this.#drain();                               // replacement is spawned on demand
  }

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

import { WorkerPool } from './pool.js';
import { randomFillSync } from 'node:crypto';
import { createHash } from 'node:crypto';

const covers = Array.from({ length: 16 }, (_, i) => ({ isbn: `978-0-00-${String(i).padStart(6, '0')}`, bytes: randomFillSync(new Uint8Array(256 * 1024)) }));

// baseline: the same work on the main thread, one cover after another
function hashHere(data) {
  let d = createHash('sha256').update(data).digest();
  for (let i = 0; i < 150_000; i++) d = createHash('sha256').update(d).update(data.subarray(0, 64)).digest();
  return d.toString('hex');
}
let t = performance.now();
const expected = covers.map((c) => hashHere(c.bytes));
console.log('sequential, main thread:', Math.round(performance.now() - t), 'ms');

const pool = new WorkerPool(new URL('./hash-worker.js', import.meta.url), 4);
t = performance.now();
const results = await Promise.all(covers.map(({ bytes }) => {
  const copy = bytes.slice();                                       // keep ours; transfer the copy
  return pool.run({ bytes: copy.buffer }, [copy.buffer]);
}));
console.log('pool of 4              :', Math.round(performance.now() - t), 'ms');
console.log('identical hashes       :', results.every((r, i) => r.hash === expected[i]));

// a job that fails inside the worker rejects its own promise and leaves the pool healthy
await pool.run({ bytes: -1 }).catch((e) => console.log('bad job rejected      :', e.name));
console.log('pool still works       :', (await pool.run({ bytes: new ArrayBuffer(8) })).hash.slice(0, 8));
await pool.close();

sequential, main thread: 997 ms
pool of 4              : 744 ms
identical hashes       : true
bad job rejected      : RangeError
pool still works       : 7efe150c

import http from 'node:http';
import { WorkerPool } from './pool.js';

const pool = new WorkerPool(new URL('./hash-worker.js', import.meta.url), 2);

const server = http.createServer(async (req, res) => {
  if (req.method === 'POST' && req.url === '/covers/hash') {
    const chunks = [];
    for await (const chunk of req) chunks.push(chunk);
    const body = Buffer.concat(chunks);
    const bytes = new Uint8Array(body).buffer;               // own copy, safe to transfer
    try {
      const { hash } = await pool.run({ bytes }, [bytes]);
      res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify({ bytes: body.length, hash }));
    } catch (err) {
      res.writeHead(500).end(err.message);
    }
  } else {
    res.writeHead(404).end();
  }
});

await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));
const { port } = server.address();

// exercise it: four uploads in parallel, the event loop stays responsive
const started = performance.now();
const replies = await Promise.all(Array.from({ length: 4 }, (_, i) =>
  fetch(`http://127.0.0.1:${port}/covers/hash`, { method: 'POST', body: Buffer.alloc(100_000, i) }).then((r) => r.json())));
console.log(replies.map((r) => `${r.bytes} B -> ${r.hash.slice(0, 12)}`));
console.log('total ms:', Math.round(performance.now() - started));

server.close();
await pool.close();

[
  '100000 B -> ed8e655a629c',
  '100000 B -> 0ed0ccc82325',
  '100000 B -> f0c9ce7e08de',
  '100000 B -> fd3bca9ecdff'
]
total ms: 362

import { Worker, isMainThread, workerData, parentPort } from 'node:worker_threads';
import { createHash } from 'node:crypto';
import { availableParallelism } from 'node:os';

// Shared state: [0] = stop flag, [1] = winning nonce (low 31 bits are enough here)
const STOP = 0, NONCE = 1;

if (!isMainThread) {
  const { shared, block, difficulty, start, step } = workerData;
  const state = new Int32Array(shared);
  const prefix = '0'.repeat(difficulty);
  let tried = 0;
  for (let nonce = start; ; nonce += step) {
    if ((tried++ & 0x3fff) === 0 && Atomics.load(state, STOP) === 1) break;   // cheap periodic check
    const hash = createHash('sha256').update(`${block}:${nonce}`).digest('hex');
    if (hash.startsWith(prefix)) {
      if (Atomics.compareExchange(state, STOP, 0, 1) === 0) {                  // first finder wins
        Atomics.store(state, NONCE, nonce);
        parentPort.postMessage({ nonce, hash });
      }
      break;
    }
  }
  parentPort.postMessage({ tried });
} else {
  const block = 'bookshelf-block-1|prev=0000|data=alice->bob:5';
  const difficulty = 5;
  const threads = Math.min(4, availableParallelism());
  const shared = new SharedArrayBuffer(8);
  const state = new Int32Array(shared);
  const started = performance.now();
  const workers = Array.from({ length: threads }, (_, i) =>
    new Worker(new URL(import.meta.url), { workerData: { shared, block, difficulty, start: i, step: threads } }));
  let answer, total = 0;
  await Promise.all(workers.map((w) => new Promise((resolve, reject) => {
    w.on('message', (m) => { if (m.hash) answer = m; if (m.tried) total += m.tried; });
    w.once('error', reject);
    w.once('exit', resolve);
  })));
  const check = createHash('sha256').update(`${block}:${answer.nonce}`).digest('hex');
  console.log({ threads, nonce: answer.nonce, hash: answer.hash, verified: check === answer.hash, attempts: total, ms: Math.round(performance.now() - started) });
}

{
  threads: 4,
  nonce: 2682751,
  hash: '000009bbcb0690171e8c8e21427f22a805ce737597ae87639657e512791a7e40',
  verified: true,
  attempts: 2685923,
  ms: 438
}

const worker = new Worker(new URL('./web-worker-script.js', import.meta.url), { name: 'greeter' });
worker.addEventListener('message', (event) => {
  console.log(event.data);
  worker.terminate();
});
worker.postMessage('Hello');

addEventListener('message', (event) => {
  postMessage(`${event.data} from ${name} (self === globalThis: ${self === globalThis}, has process: ${typeof process})`);
});

node --experimental-web-worker webworker.mjs

Hello from greeter (self === globalThis: true, has process: object)