Node.js is awesome for building fast, scalable apps. But to really make it fly on modern multi-core systems, you’ve gotta harness the power of multi-threading and worker threads. Let’s dig into how to level up your Node.js skills and squeeze every ounce of performance out of those cores.
First things first - Node.js is traditionally single-threaded. That’s great for simplicity, but it means we’re leaving performance on the table with our fancy multi-core processors. Enter worker threads - a way to run JavaScript in parallel across multiple CPU cores.
To get started with worker threads, we need to import the ‘worker_threads’ module:
const { Worker, isMainThread, parentPort } = require('worker_threads');
Now we can spin up a new worker thread like this:
if (isMainThread) {
const worker = new Worker(__filename);
worker.on('message', (msg) => {
console.log('Received:', msg);
});
worker.postMessage('Hello from main thread!');
} else {
parentPort.on('message', (msg) => {
console.log('Worker received:', msg);
parentPort.postMessage('Hello from worker!');
});
}
This creates a worker that runs the same file. The main thread can send messages to the worker with postMessage(), and receive replies via the ‘message’ event. The worker does the reverse using parentPort.
But that’s just scratching the surface. To really leverage worker threads, we want to offload CPU-intensive tasks. Let’s say we’re building a image processing app. We could create a pool of workers to handle resizing images in parallel:
const { Worker } = require('worker_threads');
const os = require('os');
const workerPool = [];
const numWorkers = os.cpus().length;
for (let i = 0; i < numWorkers; i++) {
const worker = new Worker('./image-resizer.js');
workerPool.push(worker);
}
function resizeImage(imagePath, size) {
return new Promise((resolve, reject) => {
const worker = workerPool.pop();
worker.postMessage({ imagePath, size });
worker.once('message', (result) => {
workerPool.push(worker);
resolve(result);
});
worker.once('error', (error) => {
workerPool.push(worker);
reject(error);
});
});
}
In our image-resizer.js, we’d have something like:
const { parentPort } = require('worker_threads');
const sharp = require('sharp');
parentPort.on('message', async ({ imagePath, size }) => {
try {
const resizedBuffer = await sharp(imagePath)
.resize(size.width, size.height)
.toBuffer();
parentPort.postMessage(resizedBuffer);
} catch (error) {
parentPort.postMessage({ error: error.message });
}
});
Now we can resize multiple images in parallel, fully utilizing our CPU cores. Sweet!
But worker threads aren’t just for CPU-bound tasks. They’re great for I/O-bound operations too. Say we’re building a web scraper and want to fetch multiple pages simultaneously:
const { Worker } = require('worker_threads');
const urls = [/* list of URLs */];
const fetchWorker = new Worker(`
const { parentPort } = require('worker_threads');
const fetch = require('node-fetch');
parentPort.on('message', async (url) => {
try {
const response = await fetch(url);
const text = await response.text();
parentPort.postMessage({ url, text });
} catch (error) {
parentPort.postMessage({ url, error: error.message });
}
});
`);
urls.forEach(url => {
fetchWorker.postMessage(url);
});
fetchWorker.on('message', (result) => {
if (result.error) {
console.error(`Error fetching ${result.url}: ${result.error}`);
} else {
console.log(`Fetched ${result.url}: ${result.text.slice(0, 50)}...`);
}
});
This approach lets us fetch multiple pages concurrently, potentially speeding up our scraper significantly.
Now, worker threads are fantastic, but they’re not always the best solution. For simpler concurrency needs, we might want to use the built-in ‘cluster’ module instead. It’s great for scaling a web server across multiple cores:
const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;
if (cluster.isMaster) {
console.log(`Master ${process.pid} is running`);
// Fork workers.
for (let i = 0; i < numCPUs; i++) {
cluster.fork();
}
cluster.on('exit', (worker, code, signal) => {
console.log(`worker ${worker.process.pid} died`);
});
} else {
// Workers can share any TCP connection
// In this case it is an HTTP server
http.createServer((req, res) => {
res.writeHead(200);
res.end('hello world\n');
}).listen(8000);
console.log(`Worker ${process.pid} started`);
}
This creates a separate process for each CPU core, all handling incoming HTTP requests. It’s a simple way to make your web server utilize all available cores.
But what if we want even more control over our multi-threading? That’s where the ‘async_hooks’ module comes in. It lets us track asynchronous resources throughout their lifecycle. Here’s a simple example:
const async_hooks = require('async_hooks');
const eid = async_hooks.executionAsyncId();
const tid = async_hooks.triggerAsyncId();
console.log(`Current execution async id: ${eid}`);
console.log(`Trigger async id: ${tid}`);
const hook = async_hooks.createHook({
init(asyncId, type, triggerAsyncId, resource) {
console.log(`Init: ${asyncId}, Type: ${type}`);
},
before(asyncId) {
console.log(`Before: ${asyncId}`);
},
after(asyncId) {
console.log(`After: ${asyncId}`);
},
destroy(asyncId) {
console.log(`Destroy: ${asyncId}`);
}
});
hook.enable();
setTimeout(() => {
console.log('Timeout executed');
}, 100);
This lets us peek under the hood of Node.js’s event loop and understand how our asynchronous operations are being handled.
Now, all this multi-threading goodness is great, but it can lead to some gnarly concurrency issues if we’re not careful. That’s where atomics come in. The Atomics object provides atomic operations as static methods. They’re super useful for managing shared memory between workers:
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
if (isMainThread) {
const buffer = new SharedArrayBuffer(4);
const view = new Int32Array(buffer);
const worker = new Worker(__filename, { workerData: buffer });
Atomics.store(view, 0, 123);
worker.on('message', (msg) => {
console.log(msg);
console.log(Atomics.load(view, 0)); // This will print 456
});
} else {
const view = new Int32Array(workerData);
console.log(Atomics.load(view, 0)); // This will print 123
Atomics.store(view, 0, 456);
parentPort.postMessage('Changed shared value');
}
This demonstrates how we can safely share and modify data between threads using atomic operations.
But what if we want to coordinate between multiple threads? That’s where Atomics.wait() and Atomics.notify() come in handy:
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
if (isMainThread) {
const buffer = new SharedArrayBuffer(4);
const view = new Int32Array(buffer);
const worker = new Worker(__filename, { workerData: buffer });
setTimeout(() => {
Atomics.store(view, 0, 123);
Atomics.notify(view, 0);
console.log('Main thread notified worker');
}, 1000);
worker.on('message', (msg) => {
console.log(msg);
});
} else {
const view = new Int32Array(workerData);
console.log('Worker waiting...');
Atomics.wait(view, 0, 0);
console.log('Worker woke up!');
console.log(`Shared value is now ${Atomics.load(view, 0)}`);
parentPort.postMessage('Worker finished');
}
This script demonstrates how a worker can wait for the main thread to signal it, allowing for sophisticated synchronization between threads.
Now, all this low-level stuff is cool, but sometimes we want something a bit more high-level. That’s where libraries like Piscina come in. Piscina provides an easy-to-use thread pool for Node.js:
const Piscina = require('piscina');
const piscina = new Piscina({
filename: path.resolve(__dirname, 'worker.js')
});
(async () => {
const result = await piscina.run({ a: 4, b: 6 });
console.log(result); // Prints 10
})();
And in our worker.js:
module.exports = ({ a, b }) => a + b;
This makes it super easy to offload tasks to a pool of worker threads.
But what if we’re dealing with a lot of data and want to process it in chunks across multiple threads? We can combine worker threads with Node.js streams for some seriously powerful data processing:
const { Worker } = require('worker_threads');
const { Readable, Transform } = require('stream');
const dataSource = new Readable({
read() {
for (let i = 0; i < 100; i++) {
this.push(Buffer.from(`Data chunk ${i}`));
}
this.push(null);
}
});
class WorkerTransform extends Transform {
constructor(workerScript) {
super({ objectMode: true });
this.worker = new Worker(workerScript);
this.worker.on('message', (result) => {
this.push(result);
});
}
_transform(chunk, encoding, callback) {
this.worker.postMessage(chunk);
callback();
}
_flush(callback) {
this.worker.postMessage(null);
this.worker.on('exit', () => {
callback();
});
}
}
const workerTransform = new WorkerTransform(`
const { parentPort } = require('worker_threads');
parentPort.on('message', (chunk) => {
if (chunk === null) {
parentPort.close();
} else {
// Do some CPU-intensive processing here
const result = chunk.toString().toUpperCase();
parentPort.postMessage(result);
}
});
`);
dataSource
.pipe(workerTransform)
.on('data', (chunk) => {
console.log(chunk);
})
.on('end', () => {
console.log('Processing complete');
});
This setup allows us to process a stream of data across multiple CPU cores, potentially providing a significant performance boost for data-intensive applications.
As we wrap up, it’s worth mentioning that while multi-threading can significantly boost performance, it also introduces complexity. Always profile your application to ensure that the overhead of creating and managing threads doesn’t outweigh the performance benefits. Tools like node —prof and flame graphs can be invaluable here
