Extend job submission and notification via Redis

Author: balis

examples/TaskStatusRedis/functions.js

@@ -1,13 +1,26 @@
 var spawn = require('child_process').spawn;
 
 // Spawns a job "node handler.js" and waits for the notification of its
-// completion using the Redis task status notification mechanism
-async function task_status_redis_test(ins, outs, context, cb) {
+// completion using the Redis job status notification mechanism
+async function job_status_redis_test(ins, outs, context, cb) {
   var n = Number(ins.number.data[0]);
 
   //console.log("Spawning process...");
 
-  var proc = spawn('node', ['handler.js', context.taskId, context.redis_url]);
+  const executable = context.executor.executable;
+  let jobMessage = JSON.stringify({
+        "executable": executable,
+        "args": context.executor.args,
+        "env": (context.executor.env || {}),
+        "inputs": ins.map(i => i),
+        "outputs": outs.map(o => o),
+        "stdout": context.executor.stdout, // if present, denotes file name to which stdout should be redirected
+      	"redis_url": context.redis_url,
+      	"taskId": context.taskId
+  });
+
+  // "submit" job (start the handler process)
+  var proc = spawn('npm', ['start', context.taskId, context.redis_url], {shell: true});
 
   proc.stderr.on('data', function(data) {
     console.log(data.toString());
@@ -21,17 +34,25 @@ async function task_status_redis_test(ins, outs, context, cb) {
     //console.log('Process exited with code', code);
   });
 
-  // wait for the task to finish (timeout=0 means indefinite)
+  // send message to the job (command to be executed)
+  try {
+      await context.sendMsgToJob(jobMessage);
+  } catch(err) {
+      console.error(err);
+      throw err;
+  }
+
+  // wait for the job to finish (timeout=0 means indefinite)
   try {
-    var taskStatus = await context.taskStatus(0);
-    console.log('Received task status:', taskStatus);
+    var jobStatus = await context.jobStatus(0);
+    console.log('Received job status:', jobStatus);
     setTimeout(function() {
       cb(null, outs);
     }, 5000);
   } catch(err) {
-    console.err(err);
+    console.error(err);
     throw err;
   }
 }
 
-exports.task_status_redis_test = task_status_redis_test;
+exports.job_status_redis_test = job_status_redis_test;

examples/TaskStatusRedis/handler.js

@@ -2,7 +2,7 @@
 var redis = require('redis');
 
 if (process.argv.length < 4) {
-  console.err("Usage: node handler.js <taskId> <redis_url>");
+  console.error("Usage: node handler.js <taskId> <redis_url>");
   process.exit(1);
 }
 
@@ -15,16 +15,58 @@ redis_url = process.argv[3];
 
 var rcl = redis.createClient(redis_url);
 
-// After some delay, do a RPUSH to the Redis list to notify job completion
-var delay=Math.random()*3000;
-// console.log("Delay:", delay);
-setTimeout(function() {
-  rcl.rpush(taskId, "OK", function(err, reply) {
-    if (err) {
-      console.err("Redis notification failed in handler.");
+// get job message from Redis 
+var getJobMessage = async function(timeout) {
+    return new Promise(function(resolve, reject) {
+            const jobMsgKey = taskId + "_msg";
+            rcl.brpop(jobMsgKey, timeout, function(err, reply) {
+                    if (err) reject(err)
+                    else {
+                        resolve(reply);
+                    }
+                });
+            });
+}
+
+// send notification about job completion to Redis
+var notifyJobCompletion = async function() {
+    return new Promise(function(resolve, reject) {
+            rcl.rpush(taskId, "OK", function(err, reply) {
+                if (err) reject(err)
+                else {
+                    resolve(reply);
+                }
+            });
+    });
+}
+
+
+async function executeJob() {
+    
+    // 1. get job message
+    try {
+        var jobMessage = await getJobMessage(10);
+    } catch(err) {
+      console.error(err);
       throw err;
-    } else {
-      process.exit(0);
     }
-  });
-}, delay);
+    console.log("Received job message:", jobMessage);
+
+    // 2. HERE the job would be executed
+
+    // 3. Notify job completion (delay simulates execution time)
+    var delay=Math.random()*3000;
+    // console.log("Delay:", delay);
+    setTimeout(async function() {
+            try {
+                await notifyJobCompletion();
+            } catch(err) {
+                console.error("Redis notification failed", err);
+                throw err;
+            }
+
+            process.exit(0);
+    }, delay);
+}
+
+executeJob()

examples/TaskStatusRedis/workflow.json

@@ -5,12 +5,18 @@
     "type": "dataflow",
     "parlevel": 0,
     "firingLimit": 100,
-    "function": "task_status_redis_test",
+    "function": "job_status_redis_test",
+    "config": {
+        "executor": {
+            "executable": "echo",
+            "args": ["Hello", "world"]
+        }
+    },
     "ins": [ "number" ],
     "outs": [ ]
   } ],
   "signals": [ {
-    "name": "number",
+      "name": "number",
     "data": [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100 ]
   } ],
   "ins": [ "number" ],

functions/awsFargateCommand.js

@@ -203,4 +203,4 @@ class TaskLimitError extends Error {
     }
 }
 
-exports.awsFargateCommand = awsFargateCommand;
+exports.awsFargateCommand = awsFargateCommand;

wflib/index.js

@@ -1575,7 +1575,7 @@ function public_invokeProcFunction(wfId, procId, firingId, insIds_, insValues, o
                 // should be set by the task's executor
                 // Create duplicate redis client for blocking blpop (one per task)
                 var redisCliBlocking = rcl.duplicate();
-                var getTaskStatus = async function(timeout) {
+                var getJobStatus = async function(timeout) {
                   return new Promise(function(resolve, reject) {
                     const taskId = conf.taskId;
                     const redis_cli = redisCliBlocking;
@@ -1589,9 +1589,24 @@ function public_invokeProcFunction(wfId, procId, firingId, insIds_, insValues, o
                   });
                 }
 
-                conf.taskStatus = getTaskStatus;
+                conf.jobStatus = getJobStatus;
                 conf.redis_url = "redis://" + rcl.address;
 
+                var sendMessageToJob = async function(message) {
+                  return new Promise(function(resolve, reject) {
+                          const taskMessageKey=conf.taskId+"_msg";
+                          const redis_cli = redisCliBlocking;
+                          redis_cli.lpush(taskMessageKey, message, function(err, reply) {
+                              if (err) reject(err)
+                              else {
+                                resolve(reply);
+                              }
+                          });
+                  });
+                }
+
+                conf.sendMsgToJob = sendMessageToJob;
+
                 // Pass the workflow working directory
                 if (appConfig.workdir) {
                   conf.workdir = appConfig.workdir;