const { Worker, MessageChannel } = require('node:worker_threads');
const path = require('node:path');
class WorkerPool {
constructor(size) {
this.workers = [];
this.idle = [];
this.queue = [];
for (let i = 0; i < size; i++) {
const worker = new Worker(path.join(__dirname, 'worker.js'));
this.workers.push(worker);
this.idle.push(worker);
}
}
run(payload) {
return new Promise((resolve, reject) => {
this.queue.push({ payload, resolve, reject });
this.dispatch();
});
}
dispatch() {
if (this.idle.length === 0 || this.queue.length === 0) return;
const worker = this.idle.pop();
const { payload, resolve, reject } = this.queue.shift();
const { port1, port2 } = new MessageChannel();
port1.once('message', (result) => {
port1.close();
this.idle.push(worker);
if (result.error) reject(new Error(result.error));
else resolve(result);
this.dispatch();
});
worker.postMessage({ payload, port: port2 }, [port2]);
}
async destroy() {
await Promise.all(this.workers.map((w) => w.terminate()));
}
}
module.exports = { WorkerPool };
const { parentPort } = require('node:worker_threads');
const { createHash } = require('node:crypto');
parentPort.on('message', ({ payload, port }) => {
try {
const { buffer, start, end } = payload;
const view = new Uint8Array(buffer, start, end - start);
const digest = createHash('sha256').update(view).digest('hex');
port.postMessage({ start, end, digest });
} catch (err) {
port.postMessage({ error: err.message });
} finally {
port.close();
}
});
const os = require('node:os');
const { createHash } = require('node:crypto');
const { WorkerPool } = require('./pool');
async function hashChunked(bytes, chunkCount) {
const shared = new SharedArrayBuffer(bytes.length);
new Uint8Array(shared).set(bytes);
const pool = new WorkerPool(os.cpus().length);
const chunkSize = Math.ceil(bytes.length / chunkCount);
const tasks = [];
for (let start = 0; start < bytes.length; start += chunkSize) {
const end = Math.min(start + chunkSize, bytes.length);
tasks.push(pool.run({ buffer: shared, start, end }));
}
try {
const parts = await Promise.all(tasks);
parts.sort((a, b) => a.start - b.start);
const combined = createHash('sha256');
for (const part of parts) combined.update(part.digest);
return combined.digest('hex');
} finally {
await pool.destroy();
}
}
async function main() {
const data = Buffer.alloc(64 * 1024 * 1024, 0x7a);
const rootHash = await hashChunked(data, 16);
console.log('merged digest:', rootHash);
}
main().catch((err) => {
console.error(err);
process.exit(1);
});
This snippet shows how to parallelize a CPU-bound task across a pool of worker_threads, then merge the partial results back on the main thread. The workload here is hashing large byte ranges of a buffer, but the shape applies to any embarrassingly-parallel job: split the input into chunks, hand each chunk to an idle worker, and reduce the responses.
In pool.js, WorkerPool spins up size workers eagerly and keeps two structures: an idle stack of ready workers and a queue of pending tasks. The key detail is that each run call creates a dedicated MessageChannel. The worker keeps its long-lived control Worker for lifecycle, but every individual task gets a fresh port2 transferred to it. This isolates task replies onto their own channel, so a slow or out-of-order response can never be confused with another task's result, and the port1 handler can be a one-shot once('message') that resolves a promise. Transferring the port (via the transferList) moves ownership rather than copying, which is what makes it cheap.
dispatch is the scheduler: it pops an idle worker and a queued task, and if either is missing it simply returns and waits. When a worker finishes it pushes itself back onto idle and calls dispatch again, which naturally provides backpressure — tasks only start when a worker is free, so the queue absorbs bursts without oversubscribing the CPU. destroy terminates every worker for clean shutdown.
worker.js is the other side. It listens on the main parentPort for a message carrying the transferred port, does the hashing on a SharedArrayBuffer-backed view so no bytes are copied per task, then posts the digest back on that dedicated port and closes it. Using a shared buffer means the fan-out never duplicates the payload across threads.
main.js ties it together: it splits the buffer into N ranges, maps each to pool.run, and uses Promise.all to await all partial hashes before folding them into one combined digest. The trade-offs are worth noting — worker startup and message serialization have real overhead, so this only wins for genuinely CPU-heavy chunks; tiny tasks are faster inline. Pinning size to os.cpus().length avoids thrashing the scheduler.
Related snips
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
package pools
import (
"bytes"
"sync"
)sync.Pool for bytes.Buffer to reduce allocations in hot paths
import type { IncomingMessage, ServerResponse } from "http";
const MIN_BYTES = 1024;
const INCOMPRESSIBLE = /^(image|video|audio)\/|application\/(zip|gzip|x-brotli|pdf|octet-stream)/i;Response compression (only when it helps)
Share this code
Here's the card — post it anywhere.
