feat: support for assigning new tasks in batches , to avoid network overhead

feat: support for assigning new tasks in batches , to avoid network overhead

Author: Kos-M

env.sample

@@ -3,7 +3,10 @@
 token=
 # transferEncryptToken should be 32 characters long
 transferEncryptToken=
-
+# enable/disable assigning new tasks in batches , values true/false
+getBatch=true
+# set batch size   if enabled , default is 5
+batchSize=5
 # LOG_LEVEL=off
 LOG_LEVEL=info
 # LOG_LEVEL=warn

example/index.js

@@ -5,6 +5,9 @@ const Worker = require('../index');
 process.env.transferEncryptToken = '00000000000000000000000000000000';
 process.env.token = 'demoCHannel0';
 process.env.LOG_LEVEL = 'debug';
+process.env.getBatch=true;
+process.env.batchSize=10;
+
 
 async function run() {
   // eslint-disable-next-line no-unused-vars

src/index.js

@@ -37,6 +37,8 @@ class Worker {
     this.taskFile = 'null';
     this.jobsDone = 0;
     this.jobsToDo = []; // push here jobs assigned due working..
+    this.getBatch = Object.prototype.hasOwnProperty.call(process.env, 'getBatch') ?  process.env.getBatch  : false
+    this.batchSize =  Object.prototype.hasOwnProperty.call(process.env, 'batchSize') ?  process.env.batchSize  :  5
     this.init = this.init.bind(this);
     this.onSeen = this.onSeen.bind(this);
     this.requestWork = this.requestWork.bind(this);
@@ -112,8 +114,8 @@ class Worker {
     if (Helper.getTimestamp() - this.lastRequestWork > minRequestWorkWindow) {
       // prevent unnecessary requests , flooding Master
       this.lastRequestWork = Helper.getTimestamp(); // track last request for work..
-      this.peer.rpc(this.MasterAdress, 'requestWork', {}, async (masterAns) => {
-        if (masterAns.task) {
+      this.peer.rpc(this.MasterAdress, 'requestWork', { getBatch: this.getBatch , batchSize: this.batchSize }, async (masterAns) => {
+        if (masterAns.task) { // single task
           // null if no jobs available
           const job = this.crypt.decrypt(JSON.parse(masterAns.task)); // decrypt once incoming job data
           const startedOn = Helper.getTimestamp();
@@ -126,6 +128,14 @@ class Worker {
               this.log.warn(e.message);
             });
         }
+        if (masterAns.batchTasks ){ // received batch tasks
+          masterAns.batchTasks.forEach((encryptedTask)=>{
+            const job = this.crypt.decrypt(JSON.parse(encryptedTask));
+            this.jobsToDo.push(JSON.parse(job)) // push decrypted job to internal queue
+          })
+          // this.log.fatal(this.jobsToDo)
+          this.event.emit('requestWork');
+        }
       });
     }
   }