Merge branch 'function_chain_predict' into explicit_function_chaining

dispatch_system/constants.json

 {
-    "registry_url" :"10.129.6.5:5000/",
+    "registry_url": "10.129.6.5:5000/",
     "master_port": 8080,
     "master_address": "localhost",
     "grunt_host": "https://www.namandixit.net/lovecraftian_nightmares/grunt",
@@ -37,5 +37,6 @@
     },
     "speculative_deployment": true,
     "JIT_deployment": true,
+    "aggressivity": 0.2,
     "id_size": 20
-}
+}
\ No newline at end of file

dispatch_system/constants_local.json

+{
+    "registry_url": "localhost:5000/",
+    "master_port": 8080,
+    "master_address": "localhost",
+    "grunt_host": "https://www.namandixit.net/lovecraftian_nightmares/grunt",
+    "couchdb_host": "localhost:5984",
+    "db": {
+        "function_meta": "serverless",
+        "metrics": "metrics",
+        "implicit_chain_meta": "implicit_chain",
+        "explicit_chain_meta": "explicit_chain"
+    },
+    "network": {
+        "network_bridge": "hybrid_kafka-serverless",
+        "use_bridge": true,
+        "internal": {
+            "kafka_host": "kafka:9092"
+        },
+        "external": {
+            "kafka_host": "localhost:29092"
+        }
+    },
+    "topics": {
+        "request_dm_2_rm": "request",
+        "heartbeat": "heartbeat",
+        "deployed": "deployed",
+        "remove_worker": "removeWorker",
+        "response_rm_2_dm": "RESPONSE_RM_2_DM_DUMMY",
+        "hscale": "hscale",
+        "log_channel": "LOG_COMMON"
+    },
+    "autoscalar_metrics": {
+        "open_request_threshold": 100
+    },
+    "metrics": {
+        "alpha": 0.7
+    },
+    "speculative_deployment": true,
+    "JIT_deployment": true,
+    "id_size": 20
+}
\ No newline at end of file

dispatch_system/constants_server.json

 {
-    "registry_url" :"10.129.6.5:5000/",
+    "registry_url": "10.129.6.5:5000/",
     "master_port": 8080,
-    "master_address": "10.129.6.5",
+    "master_address": "localhost",
     "grunt_host": "https://www.namandixit.net/lovecraftian_nightmares/grunt",
     "couchdb_host": "10.129.6.5:5984",
     "db": {
@@ -12,8 +12,9 @@
     },
     "network": {
         "network_bridge": "hybrid_kafka-serverless",
+        "use_bridge": false,
         "internal": {
-            "kafka_host": "kafka:9092"
+            "kafka_host": "10.129.6.5:9092"
         },
         "external": {
             "kafka_host": "10.129.6.5:9092"
@@ -34,7 +35,7 @@
     "metrics": {
         "alpha": 0.7
     },
-    "speculative_deployment": false,
+    "speculative_deployment": true,
     "JIT_deployment": true,
     "id_size": 20
-}
+}
\ No newline at end of file

dispatch_system/dispatch_manager/explicit_chain_handler.js

@@ -522,5 +522,5 @@ function createDirectory(path) {
 
 
 module.exports = {
-    router
+    router, notify
 }

dispatch_system/dispatch_manager/index.js

@@ -31,7 +31,8 @@ let usedPort = new Map(), // TODO: remove after integration with RM
     functionBranchTree = sharedMeta.functionBranchTree, // a tree to store function branch predictions
 
     metricsDB = sharedMeta.metricsDB,
-    metadataDB = sharedMeta.metadataDB
+    metadataDB = sharedMeta.metadataDB,
+    implicitChainDB = sharedMeta.implicitChainDB
 
 let kafka = require('kafka-node'),
     Producer = kafka.Producer,
@@ -514,39 +515,105 @@ function autoscalar() {
  * 
  * FIXME: Currently supports homogenous runtime chain i.e takes runtime as a param. 
  * Change it to also profile runtime
+ * FIXME: Hardcoded container as a runtime. Make dynamic.
  */
 async function speculative_deployment(req, runtime) {
     if (constants.speculative_deployment && req.headers['x-resource-id'] === undefined) {
         // console.log(functionBranchTree, req.params.id);
 
+        if (!functionBranchTree.has(req.params.id)) {
+            let data = await libSupport.fetchData(implicitChainDB + req.params.id)
+            if (data.error !== "not_found") {
+                data.branches = new Map(data.branches)
+                functionBranchTree.set(req.params.id, data)
+            }
+        }
+
+        console.log(util.inspect(functionBranchTree, false, null, true /* enable colors */));
+        
+
         if (functionBranchTree.has(req.params.id)) {
             let branchInfo = functionBranchTree.get(req.params.id)
             console.log("mle_path", branchInfo.mle_path);
 
             if (branchInfo.mle_path && branchInfo.mle_path.length > 1) {
-                for (let node of branchInfo.mle_path)
-                    node.id = node.node
-                let metrics = await libSupport.fetchData(metricsDB + "_bulk_get", {
-                    method: 'post',
-                    body: JSON.stringify({
-                        docs: branchInfo.mle_path
-                    }),
-                    headers: { 'Content-Type': 'application/json' },
-                })
-                console.log(util.inspect(metrics, false, null, true /* enable colors */))
-                
-                for (let node of branchInfo.mle_path) {
-                    // console.log(functionToResource);
-
-                    if (!functionToResource.has(node.node + runtime) && !db.has(node.node + runtime)) {
-                        console.log("Deploying according to MLE path: ", node.node);
-
-                        let payload = [{
-                            topic: constants.topics.hscale,
-                            messages: JSON.stringify({ "runtime": "container", "functionHash": node.node })
-                        }]
-                        producer.send(payload, function () { })
-                        db.set(node.node + runtime, [])
+                /**
+                 * calculating the depth upto which speculative deployment will work
+                 */
+                let deployDepth = branchInfo.mle_path.length * constants.aggressivity
+                if (constants.JIT_deployment) {
+                    /**
+                     * Perform Speculation with JIT
+                     */
+                    for (let node of branchInfo.mle_path)
+                        node.id = node.node
+                    let metricsPromise = libSupport.fetchData(metricsDB + "_bulk_get", {
+                        method: 'post',
+                        body: JSON.stringify({
+                            docs: branchInfo.mle_path
+                        }),
+                        headers: { 'Content-Type': 'application/json' },
+                    })
+
+                    let chainDataPromise = libSupport.fetchData(implicitChainDB + "_bulk_get", {
+                        method: 'post',
+                        body: JSON.stringify({
+                            docs: branchInfo.mle_path
+                        }),
+                        headers: { 'Content-Type': 'application/json' },
+                    })
+                    /**
+                     * Get the branch chain and the metrics data related to the MLE path
+                     */
+                    Promise.all([metricsPromise, chainDataPromise])
+                    .then(data => {
+                        let metrics = new Map(), chainData = new Map()
+                        let currentDelay = 0
+                        data[0] = data[0].results, data[1] = data[1].results
+
+                        for (let i = 0; i < deployDepth; i++) {
+                            let id = data[0][i].id
+                            metrics[id] = data[0][i].docs[0].ok
+                            id = data[1][i].id
+                            chainData[id] = data[1][i].docs[0].ok
+                            if (chainData[id])
+                                chainData[id].branches = new Map(chainData[id].branches)
+                        }
+                        
+                        currentDelay = metrics[branchInfo.mle_path[0].id].container.starttime
+                        for (let i = 1; i < deployDepth; i++) {
+                            let parent = chainData[branchInfo.mle_path[i - 1].id]
+                            let self = branchInfo.mle_path[i].id
+                            console.log(self);
+                            currentDelay += parent.branches.get(self)[1]
+                            let invokeTime = currentDelay - metrics[self].container.starttime
+                            invokeTime = (invokeTime < 0)? 0: invokeTime
+                            console.log(self, "current delay", currentDelay, "invoke time:", currentDelay - metrics[self].container.starttime);
+                            setTimeout(chainHandler.notify, invokeTime, "container", self)
+                        }
+                        
+                    })
+
+                } else {
+                    /**
+                     * Perform Speculation without JIT
+                     */
+                    let depthCounter = 0
+                    for (let node of branchInfo.mle_path) {
+                        // console.log(functionToResource);
+                        if (depthCounter > deployDepth)
+                            break
+                        if (!functionToResource.has(node.node + runtime) && !db.has(node.node + runtime)) {
+                            console.log("Deploying according to MLE path: ", node.node);
+
+                            let payload = [{
+                                topic: constants.topics.hscale,
+                                messages: JSON.stringify({ "runtime": "container", "functionHash": node.node })
+                            }]
+                            producer.send(payload, function () { })
+                            db.set(node.node + runtime, [])
+                        }
+                        depthCounter++
                     }
                 }
             }
@@ -554,7 +621,6 @@ async function speculative_deployment(req, runtime) {
     }
 }
 setInterval(libSupport.metrics.broadcastMetrics, 5000)
-// setInterval(libSupport.viterbi, 1000)
 setInterval(autoscalar, 1000);
 setInterval(dispatch, 1000);
 app.listen(port, () => logger.info(`Server listening on port ${port}!`))
\ No newline at end of file

local_experiments/function_chain_explicit.drawio

+<mxfile host="Electron" modified="2020-04-02T13:40:21.628Z" agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) draw.io/12.6.5 Chrome/80.0.3987.86 Electron/8.0.0 Safari/537.36" etag="YQW2D_Ds8ta8mPFvP2tW" version="12.6.5" type="device"><diagram id="_vj1HQFHM5RY5pbnoe9b" name="Page-1">3Vldk5owFP01PupAIuI+rh9r21VnZ+207r44qWQhWyROiAL99Y0SBAx1dfyA7YuTe5LA5dxz7w1Yg91FOGBo6Yyohd0a0KywBns1ANqttvjdAFEMGDqMAZsRK4b0FJiQP1iCmkRXxMJ+biGn1OVkmQfn1PPwnOcwxBgN8sveqJu/6xLZWAEmc+Sq6E9icUc+FjBT/AsmtpPcWW/dxTMLlCyWT+I7yKJBBoL9GuwySnk8WoRd7G64S3iJ9z38Y3bnGMMeP2bD6Pm9P/v2Cuj9eGY5zwMT3JN6S/rGo+SBGV15Ft7s0WqwQxl3qE095A4pXQpQF+A75jySoUIrTgXk8IUrZ3FI+DQzfpGX2ox7YdaIEsPjLJpmje2ehpGY6batlezzOaO/d2ERhHZUTiRNPl2xOT5ARKItxGzMD6yT6sVWTjaS8QGmCyw8FAsYdhEn67yKkBSjvVu32/pEiXAZaDJvQFOqRqaNnqgouUTsqNyVRl0MMm6k0FYLxbrwHzszMAoex7+i+WQ4NYPvENSbt9GFfpwu6lpD00FeHBB8JI+t9YQZEWRg9oFm9gJ6XREZZ2omF+0DoT3k5Bq5K3mnB12JtriOqLDC6AQO4XiyRNsHD0SNz0cV+cu47L6RcKOOU5JyjRnH4cE0krPGXja0pR1kSrKEnEw1bmvnc/jqj/nKBsbX0bD3w2KROYvmSRXIEib0M5GmR70Nc+dnTAXqIjhS0s1L18Wz4qOWr1vGp3XZ0nTNsBllha2w6xi3CdsZB5SGZhr5YJsV7EPHRh+W2YeA2ofAJ+tDUC+5D6mE/RdNB1a96RzyOivognNCtQQNWnlF3/BgVdgB2jftAFLVp72MZFLhJTt3Yl5cpNA31UQppNUsMy+aal4UNPpq5YWh7b1xwJITQ/1QU7Wj0WXaxUXSwjilf5R12L0rL6K7sve5Y1q9Umeopa7gVbRapW7/CADNkkudqZJYUP0uTqLCWAGvR/eLK5IozPRPg/h7b/rPC+z/BQ==</diagram></mxfile>
\ No newline at end of file

local_experiments/speculative_explicit.txt

+vanilla 26.69 24.16 23.74 25.39 24.07
+speculative 22.07 21.94 22.07 21.94 22.14
+jit 23.3 23.25 23.29 23.15 23.79
+
+63 107 68 75 72
+61 111 60 79 80
+61 79 79 67 75
+75 78 63 72 74
+47 49 48 49 46
+53 51 54 48 51
+
+214 154 144 136 173
+125 170 147 188 162
+5048 5026 5031 4971 5618
+5837 5890 5041 5794 5126
+10194 10164 9944 10140 10132
+15119 15460 15023 15183 15146
+
+94 78 81 80 110
+98 87 72 80 81
+81 76 86 145 75
+77 74 96 118 74
+67 63 68 68 64
+181 167 181 89 757
+
+