Merge pull request #42 from koxu1996/master

RemoteJobConnector - avoidance of duplicated redis clients

Author: balis

wflib/connector.js

@@ -0,0 +1,118 @@
+
+/**
+ * Class for getting notifications about tasks' results.
+ */
+class RemoteJobConnector {
+    /**
+     * Constructor.
+     * @param {RedisClient} redisClient redis client
+     * @param {string} wfId workflow ID
+     * @param {number} checkInterval loop interval in ms.
+     */
+    constructor(redisClient, wfId, checkInterval) {
+        this.jobPromiseResolves = {};
+        this.rcl = redisClient;
+        this.running = false;
+        this.completedNotificationQueueKey = "wf:" + wfId + ":tasksPendingCompletionHandling";
+        this.checkInterval = checkInterval;
+    }
+
+    /**
+     * Gives promise, that will be resolved on remote
+     * job completion.
+     * @param {*} taskId task ID
+     */
+    waitForTask(taskId) {
+        if (this.jobPromiseResolves[taskId] !== undefined) {
+            console.error("[RemoteJobConnector] Task", taskId, "is already observed");
+            return;
+        }
+        console.log("[RemoteJobConnector] Waiting for task", taskId);
+        let promise = new Promise((resolve, reject) => {
+            this.jobPromiseResolves[taskId] = resolve;
+        });
+
+        return promise;
+    }
+
+    /**
+     * Runs connector, that fetches notifications about
+     * task completions, then makes relevant waiting promises
+     * resolved.
+     */
+    async run() {
+        this.running = true;
+        while (true) {
+            if (this.running == false) {
+                console.log("[RemoteJobConnector] Stopping");
+                break;
+            }
+
+            let taskId = null;
+            try {
+                taskId = await new Promise((resolve, reject) => {
+                    this.rcl.srandmember(this.completedNotificationQueueKey, function(err, reply) {
+                        err ? reject(err): resolve(reply);
+                    });
+                });
+            } catch (error) {
+                console.error("[RemoteJobConnector] Unable to fetch new complated jobs", error);
+            }
+
+            if (taskId == null) {
+                await new Promise((resolve) => setTimeout(resolve, this.checkInterval));
+                continue;
+            }
+
+            console.log("[RemoteJobConnector] Got completed job:", taskId);
+
+            let taskResult = null;
+            try {
+                taskResult = await new Promise((resolve, reject) => {
+                    this.rcl.spop(taskId, function(err, reply) {
+                        /* Wrap results into array to preserve
+                         * compatibility with blpop format. */
+                        let replyArr = [null, reply];
+                        err ? reject(err): resolve(replyArr);
+                    });
+                });
+            } catch (error) {
+                console.error("[RemoteJobConnector] Unable to get result of job", taskId);
+                continue;
+            }
+
+            if (this.jobPromiseResolves[taskId] === undefined) {
+                console.error("[RemoteJobConnector] Observer for task", taskId, "not found");
+                continue;
+            }
+            let promiseResolve = this.jobPromiseResolves[taskId];
+            delete this.jobPromiseResolves[taskId];
+
+            try {
+                await new Promise((resolve, reject) => {
+                    this.rcl.srem(this.completedNotificationQueueKey, taskId, function(err, reply) {
+                        err ? reject(err): resolve(reply);
+                    });
+                });
+            } catch (error) {
+                console.error("[RemoteJobConnector] Unable to delete job from completed queue", error);
+            }
+
+            console.log("[RemoteJobConnector] Resolving promise for task", taskId, "| result =", taskResult);
+            promiseResolve(taskResult);
+        }
+
+        return;
+    }
+
+    /**
+     * Stops connector.
+     */
+    async stop() {
+        console.log("[RemoteJobConnector] Requesting stop");
+        this.running = false;
+        return;
+    }
+}
+
+module.exports = RemoteJobConnector

wflib/index.js

@@ -13,6 +13,7 @@ var fs = require('fs'),
     //toobusy = require('toobusy'),
     shortid = require('shortid'),
     Mustache = require('mustache'),
+    RemoteJobConnector = require('./connector'),
     rcl;
 
 
@@ -24,6 +25,8 @@ var sendSignalTime = 0;
 var global_hfid = 0; // global UUID of this HF engine instance (used for logging)
 var globalInfo = {}; // object holding global information for a given HF engine instance
 
+let jobConnectors = {}; // object holding remote jobs' connectors
+
 function p0() {
     return (new Date()).getTime();
 }
@@ -209,6 +212,9 @@ exports.init = function(redisClient) {
                     "status", "waiting",
                     function(err, ret) { });
 
+            jobConnectors[wfId] = new RemoteJobConnector(rcl, wfId, 3000);
+            jobConnectors[wfId].run();
+
            var multi = rcl.multi(); // FIXME: change this to async.parallel
 
             var addSigInfo = function(sigId) {
@@ -1417,22 +1423,15 @@ function public_invokeProcFunction(wfId, procId, firingId, insIds_, insValues, o
                 conf.wfname = procInfo.wfname;
 
                 // This function is passed to the Process' Function (through 'context')
-                // and can be used to wait for task completion. It reads a key from redis 
-                // that should be set by the task's executor. 
-                // 'taskId' to be waited for is read from the process context, but 
-                // optionally it can be set by the caller via parameter 'taskIdentifier' 
+                // and can be used to wait for task completion. It reads a key from redis
+                // that should be set by the task's executor.
+                // 'taskId' to be waited for is read from the process context, but
+                // optionally it can be set by the caller via parameter 'taskIdentifier'
                 var getJobResult = async function(timeout, taskIdentifier) {
-                    return new Promise(function(resolve, reject) {
-                        // Create duplicate Redis client for blocking blpop (one per task)
-                        var redisCliBlocking = rcl.duplicate();
-                        const taskId = taskIdentifier || conf.taskId;
-
-                        redisCliBlocking.blpop(taskId, timeout, function(err, reply) {
-                            err ? reject(err): resolve(reply);
-                            // Close connection of the duplicate Redis client
-                            redisCliBlocking.quit();
-                        });
-                    });
+                    const taskId = taskIdentifier || conf.taskId;
+                    let wfId = taskId.split(':')[1];
+                    let connector = jobConnectors[wfId];
+                    return connector.waitForTask(taskId);
                 }
 
                 conf.jobResult = getJobResult;