Chain probability data now stored on DB

dispatch_system/constants.json

@@ -7,7 +7,8 @@
     "db": {
         "function_meta": "serverless",
         "metrics": "metrics",
-        "implicit_chain_meta": "implicit_chain"
+        "implicit_chain_meta": "implicit_chain",
+        "explicit_chain_meta": "explicit_chain"
     },
     "network": {
         "network_bridge": "hybrid_kafka-serverless",
@@ -34,6 +35,6 @@
         "alpha": 0.7
     },
     "speculative_deployment": true,
-    "JIT_deployment": false,
+    "JIT_deployment": true,
     "id_size": 20
 }

dispatch_system/dispatch_manager/explicit_chain_handler.js

@@ -14,10 +14,11 @@ const logger = libSupport.logger
 const registry_url = constants.registry_url
 let functionToResource = sharedMeta.functionToResource,
     db = sharedMeta.db,
-    conditionProbabilityExpilict = sharedMeta.conditionProbabilityExpilict,
+    conditionProbabilityExplicit = sharedMeta.conditionProbabilityExplicit,
 
     metricsDB = sharedMeta.metricsDB,
-    metadataDB = sharedMeta.metadataDB
+    metadataDB = sharedMeta.metadataDB,
+    explicitChainDB = sharedMeta.explicitChainDB
 
 
 router.post('/deploy', (req, res) => {
@@ -193,6 +194,11 @@ async function deployContainer(path, imageName) {
 router.post('/execute/:id', (req, res) => {
     let map, aliases
     let chain_id = req.params.id
+    libSupport.fetchData(explicitChainDB + chain_id)
+    .then(chainData => {
+        console.log(chainData);
+        if (chainData.error !== "not_found")
+            conditionProbabilityExplicit[chain_id] = chainData
         if (req.files && req.files.map) {
             map = JSON.parse(req.files.map.data.toString());
             let mapPlanner = JSON.parse(req.files.map.data.toString());
@@ -218,6 +224,7 @@ router.post('/execute/:id', (req, res) => {
                     })
             })
         }
+    })
     
 })
 
@@ -226,6 +233,12 @@ async function orchestrator(chain_id, res, payload, map, aliases, result) {
     if (Object.keys(map).length == 0) {
         console.log("time to resolve", result);
         res.json(result)
+        let payload = {
+            method: 'put',
+            body: JSON.stringify(conditionProbabilityExplicit[chain_id]),
+            headers: { 'Content-Type': 'application/json' }
+        }
+        libSupport.fetchData(explicitChainDB + chain_id, payload)
         // return resolve(result)
     }
         
@@ -246,7 +259,7 @@ async function orchestrator(chain_id, res, payload, map, aliases, result) {
                 delete map[functionName]
                 aliases[functionName].status = "running"
                 
-                fetch(url, data).then(res => res.json())
+                libSupport.fetchData(url, data)
                     .then(json => {
                         // console.log(json);
                         result[functionName] = json
@@ -264,21 +277,21 @@ async function orchestrator(chain_id, res, payload, map, aliases, result) {
                             if (metadata.type === "conditional" && metadata.wait_for.length == 0) {
                                 
                                 let conditionResult = checkCondition(metadata.condition.op1, metadata.condition.op2, metadata.condition.op, result)
-                                if (conditionProbabilityExpilict[chain_id] === undefined)
-                                    conditionProbabilityExpilict[chain_id] = {}
-                                if (conditionProbabilityExpilict[chain_id][_key] === undefined)
-                                    conditionProbabilityExpilict[chain_id][_key] = {
+                                if (conditionProbabilityExplicit[chain_id] === undefined)
+                                    conditionProbabilityExplicit[chain_id] = {}
+                                if (conditionProbabilityExplicit[chain_id][_key] === undefined)
+                                    conditionProbabilityExplicit[chain_id][_key] = {
                                         request_count: 0,
                                         probability: 0
                                     }
-                                let oldProbability = conditionProbabilityExpilict[chain_id][_key].probability
+                                let oldProbability = conditionProbabilityExplicit[chain_id][_key].probability
                                 let updateProbability = (conditionResult === 'success') ? 1.0 : 0.0
-                                conditionProbabilityExpilict[chain_id][_key].probability = 
-                                    oldProbability * conditionProbabilityExpilict[chain_id][_key].request_count + updateProbability
-                                conditionProbabilityExpilict[chain_id][_key].request_count++
-                                conditionProbabilityExpilict[chain_id][_key].probability /=
-                                         conditionProbabilityExpilict[chain_id][_key].request_count
-                                console.log(conditionResult, "probability table", conditionProbabilityExpilict);
+                                conditionProbabilityExplicit[chain_id][_key].probability = 
+                                    oldProbability * conditionProbabilityExplicit[chain_id][_key].request_count + updateProbability
+                                conditionProbabilityExplicit[chain_id][_key].request_count++
+                                conditionProbabilityExplicit[chain_id][_key].probability /=
+                                         conditionProbabilityExplicit[chain_id][_key].request_count
+                                console.log(conditionResult, "probability table", conditionProbabilityExplicit);
                                 let branchToTake = metadata[conditionResult]
                                 branchMap = map[branchToTake]
                                 delete map[_key]
@@ -323,7 +336,7 @@ async function speculative_deployment(chain_id, aliases, map, offset, done, toBe
                 if (metadata.type === 'conditional' && !constants.JIT_deployment) {
                     let probability
                     try {
-                        probability = conditionProbabilityExpilict[chain_id][mod].probability
+                        probability = conditionProbabilityExplicit[chain_id][mod].probability
                     } catch (error) {
                         console.log("branch probability not present, random branch taken");
                         
@@ -390,7 +403,7 @@ async function speculative_deployment(chain_id, aliases, map, offset, done, toBe
                                     console.log("setting notification for conditional", mod);
                                     let probability
                                     try {
-                                        probability = conditionProbabilityExpilict[chain_id][mod].probability
+                                        probability = conditionProbabilityExplicit[chain_id][mod].probability
                                     } catch (error) {
                                         console.log("branch probability not present, random branch taken");
                                         probability = Math.random()

dispatch_system/dispatch_manager/lib.js

@@ -6,11 +6,10 @@ const winston = require('winston')
 const constants = require('.././constants.json')
 const secrets = require('./secrets.json')
 const metrics = require('./metrics')
-const sharedMeta = require('./shared_meta')
 const { createLogger, format, transports } = winston;
 const heap = require('heap')
 
-let implicitChainDB = sharedMeta.implicitChainDB
+
 let kafka = require('kafka-node'),
    Producer = kafka.Producer,
    client = new kafka.KafkaClient({
@@ -19,7 +18,8 @@ let kafka = require('kafka-node'),
    }),
    producer = new Producer(client)
 
-
+let implicitChainDB = `http://${secrets.couchdb_username}:${secrets.couchdb_password}@${constants.couchdb_host}`
+implicitChainDB = implicitChainDB + "/" + constants.implicit_chain_db_names + "/"
 
 /**
  * Generates unique IDs of arbitrary length

dispatch_system/dispatch_manager/shared_meta.js

@@ -7,7 +7,7 @@ let db = new Map(), // queue holding request to be dispatched
                                     // resources associated with the function
     workerNodes = new Map(), // list of worker nodes currently known to the DM
     functionBranchTree = new Map(), // a tree to store function branch predictions
-    conditionProbabilityExpilict = new Map() // tree holding conditional probabilities for explicit chains
+    conditionProbabilityExplicit = new Map() // tree holding conditional probabilities for explicit chains
 
 /**
  * URL to the couchdb database server used to store data
@@ -22,9 +22,11 @@ metricsDB = metricsDB + "/" + constants.db.metrics + "/"
 let implicitChainDB = `http://${secrets.couchdb_username}:${secrets.couchdb_password}@${constants.couchdb_host}`
 implicitChainDB = implicitChainDB + "/" + constants.db.implicit_chain_meta + "/"
 
+let explicitChainDB = `http://${secrets.couchdb_username}:${secrets.couchdb_password}@${constants.couchdb_host}`
+explicitChainDB = explicitChainDB + "/" + constants.db.explicit_chain_meta + "/"
 
 module.exports = {
     db, functionBranchTree, functionToResource, workerNodes, resourceMap,
-    conditionProbabilityExpilict,
-    metadataDB, metricsDB, implicitChainDB
+    conditionProbabilityExplicit,
+    metadataDB, metricsDB, implicitChainDB, explicitChainDB
 }
\ No newline at end of file