Node.js Worker Threads for CPU Tasks

Worker Threads enable parallel JavaScript execution. Here's how to use them for CPU-intensive work.

Basic Worker#

// main.js
const { Worker } = require('worker_threads');

const worker = new Worker('./worker.js', {
  workerData: { numbers: [1, 2, 3, 4, 5] },
});

worker.on('message', (result) => {
  console.log('Result:', result);
});

worker.on('error', (error) => {
  console.error('Worker error:', error);
});

worker.on('exit', (code) => {
  if (code !== 0) {
    console.error(`Worker stopped with exit code ${code}`);
  }
});

// worker.js
const { parentPort, workerData } = require('worker_threads');

function heavyComputation(numbers) {
  return numbers.reduce((sum, n) => sum + n * n, 0);
}

const result = heavyComputation(workerData.numbers);
parentPort.postMessage(result);

Promise Wrapper#

// worker-wrapper.ts
import { Worker } from 'worker_threads';

interface WorkerOptions<T> {
  workerPath: string;
  workerData: T;
  timeout?: number;
}

export function runWorker<T, R>({
  workerPath,
  workerData,
  timeout = 30000,
}: WorkerOptions<T>): Promise<R> {
  return new Promise((resolve, reject) => {
    const worker = new Worker(workerPath, { workerData });

    const timer = setTimeout(() => {
      worker.terminate();
      reject(new Error('Worker timeout'));
    }, timeout);

    worker.on('message', (result: R) => {
      clearTimeout(timer);
      resolve(result);
    });

    worker.on('error', (error) => {
      clearTimeout(timer);
      reject(error);
    });

    worker.on('exit', (code) => {
      clearTimeout(timer);
      if (code !== 0) {
        reject(new Error(`Worker exited with code ${code}`));
      }
    });
  });
}

// Usage
const result = await runWorker<{ data: number[] }, number>({
  workerPath: './compute-worker.js',
  workerData: { data: [1, 2, 3, 4, 5] },
  timeout: 10000,
});

Worker Pool#

// worker-pool.ts
import { Worker } from 'worker_threads';
import os from 'os';

interface Task<T, R> {
  data: T;
  resolve: (result: R) => void;
  reject: (error: Error) => void;
}

class WorkerPool<T, R> {
  private workers: Worker[] = [];
  private freeWorkers: Worker[] = [];
  private taskQueue: Task<T, R>[] = [];
  private workerPath: string;

  constructor(workerPath: string, poolSize = os.cpus().length) {
    this.workerPath = workerPath;

    for (let i = 0; i < poolSize; i++) {
      this.addNewWorker();
    }
  }

  private addNewWorker(): void {
    const worker = new Worker(this.workerPath);

    worker.on('message', (result: R) => {
      const task = (worker as any).currentTask as Task<T, R>;
      task.resolve(result);
      this.freeWorkers.push(worker);
      this.processQueue();
    });

    worker.on('error', (error) => {
      const task = (worker as any).currentTask as Task<T, R>;
      if (task) {
        task.reject(error);
      }
      // Replace failed worker
      this.workers = this.workers.filter(w => w !== worker);
      this.addNewWorker();
    });

    this.workers.push(worker);
    this.freeWorkers.push(worker);
  }

  private processQueue(): void {
    if (this.taskQueue.length === 0 || this.freeWorkers.length === 0) {
      return;
    }

    const worker = this.freeWorkers.pop()!;
    const task = this.taskQueue.shift()!;

    (worker as any).currentTask = task;
    worker.postMessage(task.data);
  }

  run(data: T): Promise<R> {
    return new Promise((resolve, reject) => {
      this.taskQueue.push({ data, resolve, reject });
      this.processQueue();
    });
  }

  async close(): Promise<void> {
    await Promise.all(this.workers.map(w => w.terminate()));
  }
}

// Usage
const pool = new WorkerPool<number[], number>('./sum-worker.js', 4);

const results = await Promise.all([
  pool.run([1, 2, 3]),
  pool.run([4, 5, 6]),
  pool.run([7, 8, 9]),
  pool.run([10, 11, 12]),
]);

await pool.close();

Inline Worker#

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

function runInlineWorker<T, R>(fn: (data: T) => R, data: T): Promise<R> {
  return new Promise((resolve, reject) => {
    const workerCode = `
      const { parentPort, workerData } = require('worker_threads');
      const fn = ${fn.toString()};
      const result = fn(workerData);
      parentPort.postMessage(result);
    `;

    const worker = new Worker(workerCode, {
      eval: true,
      workerData: data,
    });

    worker.on('message', resolve);
    worker.on('error', reject);
  });
}

// Usage
const result = await runInlineWorker(
  (nums: number[]) => nums.reduce((a, b) => a + b, 0),
  [1, 2, 3, 4, 5]
);

Shared Memory#

// main.ts
import { Worker } from 'worker_threads';

// Create shared buffer
const sharedBuffer = new SharedArrayBuffer(4 * 1024); // 4KB
const sharedArray = new Int32Array(sharedBuffer);

// Initialize data
for (let i = 0; i < 1000; i++) {
  sharedArray[i] = i;
}

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

worker.on('message', () => {
  console.log('First 10 values:', sharedArray.slice(0, 10));
});

// shared-worker.js
const { parentPort, workerData } = require('worker_threads');

const sharedArray = new Int32Array(workerData.sharedBuffer);

// Modify shared memory
for (let i = 0; i < sharedArray.length; i++) {
  sharedArray[i] = sharedArray[i] * 2;
}

parentPort.postMessage('done');

Atomics for Synchronization#

// Using Atomics for thread-safe operations
import { Worker, isMainThread, parentPort, workerData } from 'worker_threads';

if (isMainThread) {
  const sharedBuffer = new SharedArrayBuffer(4);
  const sharedArray = new Int32Array(sharedBuffer);

  // Start multiple workers
  const workers = Array.from({ length: 4 }, () =>
    new Worker(__filename, { workerData: { sharedBuffer } })
  );

  Promise.all(
    workers.map(w => new Promise(resolve => w.on('exit', resolve)))
  ).then(() => {
    console.log('Final count:', sharedArray[0]);
    // Should be 4000 (4 workers * 1000 increments)
  });
} else {
  const sharedArray = new Int32Array(workerData.sharedBuffer);

  // Atomic increment
  for (let i = 0; i < 1000; i++) {
    Atomics.add(sharedArray, 0, 1);
  }

  process.exit(0);
}

// Using Atomics.wait and Atomics.notify
const sharedBuffer = new SharedArrayBuffer(4);
const sharedArray = new Int32Array(sharedBuffer);

// Worker waits
Atomics.wait(sharedArray, 0, 0); // Wait while value is 0

// Main thread notifies
Atomics.store(sharedArray, 0, 1);
Atomics.notify(sharedArray, 0, 1); // Wake one waiting worker

Message Channels#

import { Worker, MessageChannel } from 'worker_threads';

// Create channel for direct worker-to-worker communication
const channel = new MessageChannel();

const worker1 = new Worker('./worker1.js', {
  workerData: { port: channel.port1 },
  transferList: [channel.port1],
});

const worker2 = new Worker('./worker2.js', {
  workerData: { port: channel.port2 },
  transferList: [channel.port2],
});

// worker1.js
const { workerData } = require('worker_threads');
const { port } = workerData;

port.on('message', (msg) => {
  console.log('Worker 1 received:', msg);
});

port.postMessage('Hello from Worker 1');

// worker2.js
const { workerData } = require('worker_threads');
const { port } = workerData;

port.on('message', (msg) => {
  console.log('Worker 2 received:', msg);
  port.postMessage('Hello back from Worker 2');
});

Practical Example: Image Processing#

// image-processor.ts
import { Worker } from 'worker_threads';
import path from 'path';

interface ProcessOptions {
  width: number;
  height: number;
  quality: number;
}

async function processImage(
  imagePath: string,
  options: ProcessOptions
): Promise<Buffer> {
  return new Promise((resolve, reject) => {
    const worker = new Worker(
      path.join(__dirname, 'image-worker.js'),
      {
        workerData: { imagePath, options },
      }
    );

    worker.on('message', resolve);
    worker.on('error', reject);
  });
}

// image-worker.js
const { parentPort, workerData } = require('worker_threads');
const sharp = require('sharp');

async function process() {
  const { imagePath, options } = workerData;

  const result = await sharp(imagePath)
    .resize(options.width, options.height)
    .jpeg({ quality: options.quality })
    .toBuffer();

  parentPort.postMessage(result);
}

process();

Express Integration#

import express from 'express';
import { Worker } from 'worker_threads';
import path from 'path';

const app = express();

// Worker pool for CPU tasks
const pool = new WorkerPool(
  path.join(__dirname, 'compute-worker.js'),
  4
);

app.post('/compute', async (req, res) => {
  try {
    const result = await pool.run(req.body.data);
    res.json({ result });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

// Don't block event loop
app.get('/heavy', async (req, res) => {
  const result = await runWorker({
    workerPath: './heavy-worker.js',
    workerData: req.query,
  });

  res.json(result);
});

Best Practices#

When to Use:
✓ CPU-intensive computations
✓ Image/video processing
✓ Cryptographic operations
✓ Complex parsing/serialization

When NOT to Use:
✗ I/O bound operations
✗ Simple tasks (overhead > benefit)
✗ Operations needing main thread access
✗ Tasks requiring DOM access

Patterns:
✓ Use worker pools for reuse
✓ Transfer large data with transferList
✓ Use SharedArrayBuffer for shared state
✓ Implement proper error handling

Worker Threads enable true parallelism for CPU-intensive tasks. Use worker pools for efficiency, SharedArrayBuffer for shared state, and Atomics for synchronization. Reserve workers for computationally heavy operations where the parallelism benefit outweighs the overhead.