Request grouping added. Closes #18

To prevent requests coming in for the same function from starting too many workers, they are grouped together and one worker is started per group.

dispatch_system/dispatch_daemon/config.json

-{"id":"192.168.1.103","master_node":"10.129.6.5"}
+{"id":"192.168.31.51","master_node":"10.129.6.5"}
\ No newline at end of file

dispatch_system/dispatch_manager/index.js

@@ -205,9 +205,14 @@ app.post('/serverless/execute/:id', (req, res) => {
         libSupport.reverseProxy(req, res, functionToResource, resourceMap)
     } else {
         /**
-         * FIXME: Here, every request even for the same function will be queued up potentially launching multiple 
-         * resource of the same type
+         * Requests are queued up before being dispatched. To prevent requests coming in for the 
+         * same function from starting too many workers, they are grouped together
+         * and one worker is started per group.
          */
+        if (db.has(req.params.id + runtime)) {
+            db.get(req.params.id + runtime).push({ req, res })
+            return;
+        }
         requestQueue.push({ req, res })
         /**
          * We store functions for function placement heuristics purposes. This lets us look into the function
@@ -225,10 +230,12 @@ function dispatch() {
     let lookbackWindow = Math.min(WINDOW_SIZE, requestQueue.length)
     for (let i = 0; i < lookbackWindow; i++) {
         let {req, res} = requestQueue.shift()
+        
 	    // logger.info(req.body)
         let runtime = req.body.runtime
         let functionHash = req.params.id
-        
+        if (!db.has(functionHash + runtime)) {
+            db.set(functionHash + runtime, [])
             let resource_id = libSupport.makeid(20) // each function resource request is associated with an unique ID
             logger.info(`Generated new resource ID: ${resource_id} for runtime: ${runtime}`);
             
@@ -259,7 +266,10 @@ function dispatch() {
             producer.send(payload, () => {
                 logger.info(`Resource Deployment request sent to Dispatch Agent`)
             })
-        db.set(resource_id, { req, res })
+        } else {
+            logger.info("deployment process already started waiting")
+            db.get(functionHash + runtime).push({ req, res })
+        }
         
 
         
@@ -308,9 +318,10 @@ consumer.on('message', function (message) {
         }
         logger.info("Deployed Resource: " + JSON.stringify(message));
         
-        if (db.has(message.resource_id)) {
-            let { req, res } = db.get(message.resource_id)
+        if (db.has(message.functionHash + message.runtime)) {
+            let sendQueue = db.get(message.functionHash + message.runtime)
             if (message.status == false) {
+                // TODO: handle failure
                 res.status(400).json({reason: message.reason})
                 return;
             }
@@ -357,11 +368,15 @@ consumer.on('message', function (message) {
                 logger.info(`Confirmed RM for successful deployment resource_id: ${message.resource_id} deployment`)
             })
             logger.info("forwarding request via reverse proxy to: " + JSON.stringify(resource));
+            while (sendQueue.length != 0) {
+               let {req, res} = sendQueue.shift()
                 libSupport.reverseProxy(req, res, functionToResource, resourceMap)
                     .then(() => {
-                    db.delete(message.resource_id)
+                        
                     })
             }
+            db.delete(message.functionHash + message.runtime)
+        }
 
     } else if (topic == "removeWorker") {
         logger.warn("Worker blown: Removing Metadata " + message);

dispatch_system/dispatch_manager/lib.js

@@ -61,7 +61,7 @@ function reverseProxy(req, res, functionToResource, resourceMap) {
       logger.info("Request received at reverseproxy. Forwarding to: " + url);
       forwardTo.metric += 1
       heap.heapify(functionHeap, compare)
-      console.log(functionHeap);
+      logger.info(functionHeap);
       
       var options = {
          method: 'POST',
@@ -70,6 +70,9 @@ function reverseProxy(req, res, functionToResource, resourceMap) {
          json: true // Automatically stringifies the body to JSON
       };
 
+      // console.log(options);
+      
+      
       rp(options)
          .then(function (parsedBody) {
             
@@ -83,7 +86,7 @@ function reverseProxy(req, res, functionToResource, resourceMap) {
             forwardTo.metric -= 1
             heap.heapify(functionHeap, compare)
             console.log(functionHeap);
-            logger.error("error", err.error.errno);
+            logger.error("error" + err.error.errno);
             res.json(err.message).status(err.statusCode)
             resolve()
          });

dispatch_system/dispatch_manager/repository/worker_env/env.js

@@ -45,7 +45,7 @@ app.listen(port, () => {
 
 function shouldDie() {
     
-    if (Date.now() - lastRequest > 5 * 1000) {
+    if (Date.now() - lastRequest > 30 * 1000) {
         console.log("Idle for too long. Exiting");
         producer.send(
             [{