thekdar - npm Package Compare versions

Comparing version 1.0.0 to 1.0.1

core/Thekdar.js

@@ -1,8 +0,8 @@
-const uuid = require("uuid");
-const EventEmitter = require("events");
-const events = require("./events");
-const util = require("util");
-const Task = require("./Task");
-const Worker = require("./Worker");
-const debug = require("debug")("Thekdar:main");
+const uuid = require('uuid');
+const EventEmitter = require('events');
+const events = require('./events');
+const util = require('util');
+const Task = require('./Task');
+const Worker = require('./Worker');
+const debug = require('debug')('Thekdar:main');
 
@@ -19,2 +19,4 @@ class Thekdar extends EventEmitter {
     this._workersAddress = new Map();
+    this._maxWorkers = 20;
+    this._maxTaskPerWorker = 10;
   }
@@ -31,6 +33,6 @@
     if (!task) {
-      throw new Error("Please provide task object instance of Task Class");
+      throw new Error('Please provide task object instance of Task Class');
     }
     if (!task.getType()) {
-      throw new Error("Please provide task type, can be fork, spwan, exec");
+      throw new Error('Please provide task type, can be fork, spwan, exec');
     }
@@ -42,3 +44,3 @@ const taskId = uuid();
       if (!worker) {
-        debug("No Worker found.");
+        debug('No Worker found.');
         return null;
@@ -48,7 +50,4 @@ }
       worker.addTask(this._tasks.get(taskId));
-      if (!this._workerTaskLookup.get(worker.getId())) {
-        this._workerTaskLookup.set(worker.getId(), []);
-      }
       this._workerTaskLookup.get(worker.getId()).push(task.getId());
-      this.emit("add", { worker });
+      this.emit('add', { worker });
       debug(`New task added, previous task count ${this._tasks.size}`);
@@ -62,33 +61,41 @@ return worker;
 
-  _getFreeWorker(task, workerAddressIndex) {
-    if (this._workerTaskLookup.size > Thekdar.MAX_WORKERS) {
-      throw new Error(
-        `Maximum workers are working ${
-          this._workerTaskLookup.size
-        }, no further workers can work.`
-      );
+  /**
+   * Deploy all workers at a time so later we don't
+   * have to create worker just before assigining task.
+   *
+   * @param {*} taskType
+   * @param {*} workerAddressIndex
+   */
+  deployWorkers(taskType = Task.TYPE_FORK, workerAddressIndex) {
+    if (this._isWorkersDeployed) {
+      return false;
     }
-    const taskType = task.getType();
+    this._isWorkersDeployed = true;
     let workers = this._workers.get(taskType);
-    let newWorker;
     if (!workers) {
       this._workers.set(taskType, new Map());
-      newWorker = this._createWorker(taskType, workerAddressIndex);
-      return newWorker;
     }
-    for (let [index, worker] of this._workers.get(taskType).entries()) {
-      let lWorker = this._workerTaskLookup.get(worker.getId());
-      if (lWorker.length === Thekdar.MAX_TASK_PER_WORKER - 1) {
-        const nextWorker = this._createWorker(taskType, workerAddressIndex);
-        this._workerTaskLookup.set(nextWorker.getId(), []);
-      }
-      if (lWorker.length >= Thekdar.MAX_TASK_PER_WORKER) {
-        debug("This worker has maximum task.");
-        newWorker = null;
-        continue;
-      } else {
-        newWorker = worker;
+    for (let i = 0; i < this._maxWorkers; i++) {
+      const worker = this._createWorker(Task.TYPE_FORK, workerAddressIndex);
+      this._workerTaskLookup.set(worker.getId(), []);
+    }
+  }
+  _getFreeWorker(task, workerTaskLength = 0) {
+    if (workerTaskLength === this._maxTaskPerWorker) {
+      debug(`All worker are working on with maximum work load.`);
+      return false;
+    }
+    const taskType = task.getType();
+    let workers = this._workers.get(taskType);
+    let newWorker;
+
+    for (let [workerId, works] of this._workerTaskLookup) {
+      if (works.length === workerTaskLength) {
+        newWorker = workers.get(workerId);
         break;
       }
     }
+    if (!newWorker) {
+      newWorker = this._getFreeWorker(task, workerTaskLength + 1);
+    }
     return newWorker;
@@ -108,3 +115,3 @@ }
     if (!address) {
-      throw new Error("Please specify address of worker");
+      throw new Error('Please specify address of worker');
     }
@@ -129,3 +136,3 @@ worker.setAddress(address);
       }
-      this.emit("message", data);
+      this.emit('message', data);
     };
@@ -158,3 +165,3 @@ }
       this._tasks.delete(taskId);
-      debug("A task deleted with id of %s", taskId);
+      debug('A task deleted with id of %s', taskId);
       return true;
@@ -178,3 +185,3 @@ } catch (er) {
         worker = workerGroup.get(workerId);
-        debug("Worker with %s id found, %o", workerId, worker);
+        debug('Worker with %s id found, %o', workerId, worker);
         break;
@@ -194,3 +201,3 @@ }
       this._workerTaskLookup.delete(workerId);
-      debug("Worker with id %s has been deleted", worker.getId());
+      debug('Worker with id %s has been deleted', worker.getId());
       return worker.kill();
@@ -210,2 +217,10 @@ } catch (e) {
   }
+
+  setMaxWorker(maxWorkers) {
+    this._maxWorkers = maxWorkers;
+  }
+
+  setMaxTaskPerWorker(maxTaskPerWorker) {
+    this._maxTaskPerWorker = maxTaskPerWorker;
+  }
 }
@@ -212,0 +227,0 @@ Thekdar.MAX_TASK_PER_WORKER = 10;

core/Worker.js

@@ -1,6 +0,6 @@
-const Task = require("./Task");
-const path = require("path");
-const { fork } = require("child_process");
-const debug = require("debug")("Thekdar:worker");
-const events = require("./events");
+const Task = require('./Task');
+const path = require('path');
+const { fork } = require('child_process');
+const debug = require('debug')('Thekdar:worker');
+const events = require('./events');
 
@@ -16,6 +16,6 @@ class Worker {
   on(handler) {
-    this._worker.on("message", data => {
+    this._worker.on('message', data => {
       handler({
         workerId: this.getId(),
-        ...data
+        ...data,
       });
@@ -41,3 +41,3 @@ });
   addTask(task) {
-    debug("New task added to worker %s", this._id);
+    debug('New task added to worker %s', this._id);
     this.send(events.TASK_ADD, task);
@@ -51,3 +51,3 @@ this._tasks.set(task.getId(), task);
       workerId: this.getId(),
-      task: data
+      task: data,
     };
@@ -54,0 +54,0 @@ if (this._worker) {

examples/types/fork.js

@@ -1,14 +0,12 @@
-const getPort = require("get-port");
-const http = require("http");
+process.on('message', data => {
+  // console.log(data);
 
-// const server = http.createServer((req, res) => {
-process.send({
-  type: "random",
-  time: Date.now()
+  setTimeout(() => {
+    process.send({
+      type: 'task:complete',
+      taskId: data.task.id,
+      workerId: data.workerId,
+      data: {},
+    });
+  }, Math.round(Math.random() * 5000) + 3000);
 });
-// res.end("hello world");
-// });
-// getPort().then(port => {
-// console.log(`Server listening on port ${port}`);
-// server.listen(port);
-// });

package.json

 {
     "name": "thekdar",
-    "version": "1.0.0",
+    "version": "1.0.1",
     "description": "",
@@ -5,0 +5,0 @@ "main": "index.js",

New alerts

Shell access

Supply chain risk

This module accesses the system shell. Accessing the system shell increases the risk of executing arbitrary code.

Found 1 instance in 1 package

Fixed alerts

Network access

Supply chain risk

This module accesses the network.

Found 1 instance in 1 package

Shell access

Supply chain risk

This module accesses the system shell. Accessing the system shell increases the risk of executing arbitrary code.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

18421

increased by4.19%

Lines of code

539

increased by5.89%

Number of medium supply chain risk alerts

1

decreased by-50%

No dependency changes