Merge pull request #89 from hyperflow-wms/master

Merge updates and fixes to redis-v4

Author: balis

.github/workflows/main.yml

@@ -20,7 +20,7 @@ on:
 
 jobs:
   build:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
     steps:
       - uses: actions/checkout@v3
         with:
@@ -30,7 +30,7 @@ jobs:
         run: make image
   deploy:
     needs: [ "build" ]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
     if: github.event_name == 'release' && github.event.action == 'created'
     steps:
       - name: Login to Docker Hub

examples/RemoteJobs/README.md

@@ -1,12 +1,32 @@
-## Communication with remote Jobs using Redis
-This example demonstrates how HyperFlow can communicate with remote job executors using Redis. 
+## Distributed execution of workflow Jobs using Redis
+This example explains the distributed execution model of Hyperflow. It demonstrates how HyperFlow can communicate with remote job executors using Redis. It is also useful for testing the implementation of the [Hyperflow job executor](https://github.com/hyperflow-wms/hyperflow-job-executor).
 
- - The workflow invokes function `submitRemoteJob` from `functions.js` 100 times. This function simulates submission of jobs by starting 100 parallel processes of `node handler.js  <taskId> <redis_url>`.
- - `handler.js` represents a remote job executor which is passed two parameters: `taskId` and `redis_url`. 
- - `handler.js` gets a `jobMessage` from HyperFlow, and then sends back a notification that the job has completed; `taskId` is used to construct appropriate Redis keys for this communication.
- - On the HyperFlow engine side, the Process Function can use two functions: `context.sendMsgToJob` to send a message to the job, and `context.jobResult` to wait for the notification. These functions return a [`Promise`](https://javascript.info/promise-basics), so the async/await syntax can be used as shown in the example.
+The distributed execution architecture consists of:
+1. Master components:
+   - **The Hyperflow engine** - executes the workflow graph; for each workflow task it invokes the **Job invoker** function
+   - **Job invoker** - Javascript function which creates jobs on a (remote) infrastructure to execute workflow tasks
+   - **Redis server** - used for communication between the Hyperflow engine and Job executors on remote workers
+ 1. Worker components:
+    - **Hyperflow job executor** - receives the job command from the Hyperflow engine and spawns application software
+    - **Application software** - software that actually performs workflow tasks
+
+In this example:
+ - The workflow has two tasks (see `workflow.json`): one that executes `job.js`, the other which simply runs `ls -l`. Note that the commands to be executed are specified in `workflow.json`. 
+ - The engine invokes the function `submitRemoteJob` (Job invoker) from `functions.js`. This function simulates submission of jobs by starting the Hyperflow job executor and communicating with it via Redis to run jobs.
+ - `../../../hyperflow-job-executor/handler.js` represents a remote job executor which is passed two parameters: `taskId` and `redis_url`. The executor gets a `jobMessage` from HyperFlow, executes the command in a separate process, and then sends back a notification that the job has completed; `taskId` is used to construct appropriate Redis keys for this communication.
+ - On the HyperFlow engine side, the Job invoker can use two functions (provided by Hyperflow): `context.sendMsgToJob` to send a message to the job executor, and `context.jobResult` to wait for the notification from the executor. These functions return a [`Promise`](https://javascript.info/promise-basics), so the async/await syntax can be used as shown in the example.
  - The parameter to the `context.jobResult` function is a timeout in seconds (0 denotes infinity). One can use a retry library, such as [promise-retry](https://www.npmjs.com/package/promise-retry), to implement an exponential retry strategy.
- - The Process Function also gets the Redis URL in `context.redis_url` which can be passed to the remote job executors.
+ - The Job invoker also gets the Redis URL in `context.redis_url` which can be passed to the remote job executors.
 
- To run the workflow, simply do `hflow run .` in this directory. You might need to run once `npm install` to install dependencies.
+ To run the workflow, execute the following commands:
+ 1. First, clone the Hyperflow engine and the Hyperflow job executor:
+    - `git clone https://github.com/hyperflow`
+    - `git clone https://github.com/hyperflow-wms/hyperflow-job-executor`
+    - `cd hyperflow; npm install`
+    - `cd ../hyperflow-job-executor; npm install`
+ 1. Start the redis server
+ 1. To run the workflow:
+    - `cd ../hyperflow/examples/RemoteJobs`
+    - `npm install` (once)
+    - `hflow run .`
 

functions/kubernetes/amqpConnector.js

@@ -1,17 +1,42 @@
+
 const amqplib = require('amqplib'),
     createJobMessage = require('../../common/jobMessage').createJobMessage;
-let channels = {};
+
 let conn = null;
+let connPromise = null;
+let channels = {};
+let channelPromises = {};
+
+async function getConnection() {
+    if (conn) return conn;
+    if (!connPromise) {
+        console.log("[AMQP] Creating new connection...");
+        connPromise = amqplib.connect(`amqp://${process.env.RABBIT_HOSTNAME}`, "heartbeat=60");
+    }
+    conn = await connPromise;
+    return conn;
+}
 
 async function initialize(queue_name) {
+    const connection = await getConnection();
+
+    if (channels[queue_name]) return;
 
-    if (conn === null) {
-        conn = await amqplib.connect(`amqp://${process.env.RABBIT_HOSTNAME}`, "heartbeat=60");
+    if (!channelPromises[queue_name]) {
+        channelPromises[queue_name] = (async () => {
+            try {
+                console.log(`[AMQP] Creating channel for queue ${queue_name}`);
+                const ch = await connection.createChannel();
+                await ch.assertQueue(queue_name, { durable: false, expires: 6000000 });
+                channels[queue_name] = ch;
+            } catch (err) {
+                delete channelPromises[queue_name]; // retry logic
+                throw err;
+            }
+        })();
     }
-    let ch = await conn.createChannel()
-    await ch.assertQueue(queue_name, {durable: false, expires: 6000000});
-    channels[queue_name] = ch
 
+    await channelPromises[queue_name];
 }
 
 function getQueueName(context) {
@@ -30,28 +55,26 @@ function getQueueName(context) {
 
 async function enqueueJobs(jobArr, taskIdArr, contextArr, customParams) {
     let context = contextArr[0];
-    let queue_name = getQueueName(context)
-    if (conn === null || !(queue_name in channels)) {
-        await initialize(queue_name)
-    }
-    let ch = channels[queue_name]
+    let queue_name = getQueueName(context);
     try {
+        await initialize(queue_name);
+        let ch = channels[queue_name];
 
-        console.log(`jobArr: ${JSON.stringify(jobArr)}, taskIdArr: ${JSON.stringify(taskIdArr)}, contextArr: ${JSON.stringify(contextArr)}, customParams: ${JSON.stringify(customParams)}`)
+        console.log(`jobArr: ${JSON.stringify(jobArr)}, taskIdArr: ${JSON.stringify(taskIdArr)}, contextArr: ${JSON.stringify(contextArr)}, customParams: ${JSON.stringify(customParams)}`);
         let tasks = [];
 
         for (let i = 0; i < jobArr.length; i++) {
             let job = jobArr[i];
             let taskId = taskIdArr[i];
             let jobMessage = createJobMessage(job.ins, job.outs, contextArr[i], taskId);
-            await context.sendMsgToJob(JSON.stringify(jobMessage), taskId) // TODO remove
-            tasks.push({"id": taskId, "message": jobMessage});
+            await context.sendMsgToJob(JSON.stringify(jobMessage), taskId); // TODO remove
+            tasks.push({ "id": taskId, "message": jobMessage });
         }
 
-        await ch.publish('', queue_name, Buffer.from(JSON.stringify({'tasks': tasks})));
+        ch.sendToQueue(queue_name, Buffer.from(JSON.stringify({ 'tasks': tasks })));
     } catch (error) {
-        console.log(error)
+        console.log(error);
     }
 }
 
-exports.enqueueJobs = enqueueJobs
+exports.enqueueJobs = enqueueJobs