This content originally appeared on Bits and Pieces - Medium and was authored by Mukul Dutt
In the context of building libraries in Node.js, multi-threading can be particularly useful for improving performance and scalability. Node.js is a single-threaded runtime, meaning that it can only execute one task at a time. However, by using multi-threading, it is possible to offload certain tasks to separate threads.
Node-Threader
A simple and easy-to-use multi-threaded Node.js library that can execute multiple heavy synchronous tasks on separate threads.
Getting Started…
Here is the basic multi-thread architecture we are going to apply to our library. The Main thread is where execution will start, and the Task queue will communicate with all the threads for task distribution and sending responses back to the main thread after synchronous execution.
Main Thread
Thread forking initialization will start from here. The main thread’s responsibility is to receive an arrow function, which can be serialized and de-serialized while communicating with threads, and forward it to the task queue manager.
Task Queue Manager
While initializing the node-threader library, we need to provide the number of threads that will be utilized during the multi-thread process. We will also maintain an empty thread array to keep track of unused threads in a round-robin fashion. In addition, we will have invokers, which will keep track of the callback & promise response from the thread process and later forward it to the main thread.
Threads
We need to add a process listener that will listen to events from the main thread. The payload will include an executable function and an invoker ID. After function execution, we will send the response back along with the invoker ID. The task queue manager will then catch this response and invoke the callback of the invoker ID back to the main process, thus completing the cycle.
Final Code
if (!require("cluster").isMaster) {
process.on("message", async (data) => {
let isError = false;
const response = await (async () => {
try {
return await eval(data.data)();
} catch (e) {
isError = true;
return e;
}
})();
process.send({
response,
invokerId: data.invokerId,
isError,
});
});
return;
}
class NodeThreader {
/*
invokers will be key value of callback & promises of each thread
*/
threads = [];
invokers = {};
exitProcess = false;
/**
* @param {number|null} threads
* @param {boolean} exitProcess
*/
constructor(threads, exitProcess = false) {
this.initializeThreads(threads);
this.exitProcess = exitProcess;
}
/*
Thread initilization
*/
initializeThreads(threads = 1) {
const cluster = require("cluster");
this.threads = Array(threads || require("os").cpus().length)
.fill(true)
.map(cluster.fork);
this.initializeListener();
}
initializeListener(thread) {
if (thread) return this.addListener(thread);
this.threads.forEach(this.addListener);
}
/*
Adding listener for each thread for recieving response from different threads
*/
addListener = (thread) => {
thread.on("message", async ({ invokerId, response, isError }) => {
const currentInvokers = this.invokers[thread.id]?.[invokerId];
currentInvokers?.callback?.(
(isError && response) || null,
(!isError && response) || null
);
currentInvokers?.[(isError && "reject") || "resolve"]?.(response);
setTimeout(() => {
delete this.invokers[thread.id]?.[invokerId];
if (Object.keys(this.invokers[thread.id]).length === 0) {
delete this.invokers[thread.id];
}
this.exitProcessOnComplete();
});
});
thread.on("exit", () => this.onThreadExit(thread));
};
exitProcessOnComplete() {
if (!this.exitProcess) return;
if (Object.keys(this.invokers).length === 0) {
process.exit();
}
}
onThreadExit(thread) {
Object.values(this.invokers[thread.id] || {}).forEach((func) =>
func(new Error("Process died"))
);
const newThread = require("cluster").fork();
this.threads = [newThread, ...this.threads];
this.initializeListener(newThread);
}
/**
* @param {function} data
* @param {function} callback
* @returns Promise
*/
execute(data, callback) {
const thread = this.threads.slice(0, 1)[0];
const invokerId = this.initializeInvoker(thread, callback);
thread.send({
data: (typeof data === "string" && data) || `${data}`,
invokerId,
});
this.threads.push(this.threads.splice(0, 1)[0]);
const invoker = this.invokers[thread.id][invokerId];
invoker["callback"] = callback;
return new Promise((resolve, reject) => {
invoker["resolve"] = resolve;
invoker["reject"] = reject;
});
}
initializeInvoker(thread) {
const invokerId = Math.random() * 999999999;
this.invokers[thread.id] =
(!this.invokers[thread.id] && {}) || this.invokers[thread.id];
this.invokers[thread.id][invokerId] = {
...(this.invokers[thread.id][invokerId] || {}),
};
return invokerId;
}
}
if (typeof module !== "undefined" && module.exports) {
module.exports = NodeThreader;
}Usage
import NodeThreader from "node-threader";
//*REQUIRED* this condition check if mandatory because once thread is forked from master this whole block will be executed again to avoid this only allow NodeThreader to initialize only in master node.
if (!require("cluster").isMaster) return;
//thread -> number | null(consider all threads)
//exitProcess => boolean (exit process once all thread executions are completed)
const instance = new NodeThreader(2, true);
//executable MUST be an ARROW function as it can enclosed in string and retains its property after & before serializing/de-serializing while communicating with clusters.
//thread response supports both callback & promise
instance.execute(
() => {
return `Synchronus task : ${2e2 * 2e2}`;
},
(error, response) => {
console.log("callback response ", response);
}
);
instance
.execute(
() => {
throw "Let me throw error for no reason";
},
(error, response) => {
console.log("callback error ", error);
}
)
.catch((catchResponse) => {
console.log("Promise rejected ", catchResponse);
});
instance
.execute(() => {
return new Promise(async (r) => {
await new Promise((r) => setTimeout(r, 1000));
r("Completed after 1000ms");
});
})
.then((promiseResponse) => {
console.log("Promise resolved ", promiseResponse);
});
Repository Link : https://github.com/MUKUL47/node-threader
NPM Link : https://www.npmjs.com/package/node-threader
I hope you have enjoyed reading about the development and implementation of my multi-threaded Node.js library. Writing this library has been a rewarding and educational experience, and I am excited to share it with the community. Thank you for taking the time to read my first post on JavaScript
Bonus: Build Node.js apps with reusable components
Bit’s open-source tool help 250,000+ devs to build apps with components.
Turn any UI, feature, or page into a reusable component — and share it across your applications. It’s easier to collaborate and build faster.
Split apps into components to make app development easier, and enjoy the best experience for the workflows you want:
→ Micro-Frontends
→ Design System
→ Code-Sharing and reuse
→ Monorepo
Learn more
- How We Build Micro Frontends
- How we Build a Component Design System
- Bit - Component driven development
- 5 Ways to Build a React Monorepo
- How to Create a Composable React App with Bit
- Sharing JavaScript Utility Functions Across Projects
How to Create a Multi-Threaded Node.js Library was originally published in Bits and Pieces on Medium, where people are continuing the conversation by highlighting and responding to this story.
This content originally appeared on Bits and Pieces - Medium and was authored by Mukul Dutt
Mukul Dutt | Sciencx (2023-01-16T09:55:00+00:00) How to Create a Multi-Threaded Node.js Library. Retrieved from https://www.scien.cx/2023/01/16/how-to-create-a-multi-threaded-node-js-library/
Please log in to upload a file.
There are no updates yet.
Click the Upload button above to add an update.