Skip to content

X
xanadu
Commit db8d9fb6 authored by Nilanjan Daw's avatar Nilanjan Daw
Browse files
Options

JIT related flags + Idle time instrumentation added 

Added instrumentation to measure worker idle time before it receives its first request. 
Also added flags related to JIT deployment
parent 57e15b96
Hide whitespace changes
Inline Side-by-side
Showing with 75 additions and 59 deletions
dispatch_system/constants.json
View file @ db8d9fb6
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
"autoscalar_metrics": { "autoscalar_metrics": {
"open_request_threshold": 100 "open_request_threshold": 100
}, },
"speculative_deployment": true, "speculative_deployment": false,
"JIT_deployment": false,
"id_size": 20 "id_size": 20
} }
dispatch_system/dispatch_manager/explicit_chain_handler.js
View file @ db8d9fb6
...@@ -201,7 +201,7 @@ router.post('/execute/:id', (req, res) => { ...@@ -201,7 +201,7 @@ router.post('/execute/:id', (req, res) => {
aliases = data aliases = data
let payload = JSON.parse(req.body.data) let payload = JSON.parse(req.body.data)
console.log(payload); console.log(payload);
getTimeLines(aliases, mapPlanner); speculative_deployment(aliases, mapPlanner);
orchestrator(res, payload, map, aliases, {}) orchestrator(res, payload, map, aliases, {})
}) })
} else { } else {
...@@ -214,7 +214,7 @@ router.post('/execute/:id', (req, res) => { ...@@ -214,7 +214,7 @@ router.post('/execute/:id', (req, res) => {
.then(data => { .then(data => {
aliases = data aliases = data
let payload = JSON.parse(req.body.data) let payload = JSON.parse(req.body.data)
getTimeLines(aliases, mapPlanner); speculative_deployment(aliases, mapPlanner);
orchestrator(res, payload, map, aliases, {}) orchestrator(res, payload, map, aliases, {})
}) })
}) })
...@@ -302,71 +302,80 @@ function checkCondition(op1, op2, op, result) { ...@@ -302,71 +302,80 @@ function checkCondition(op1, op2, op, result) {
return (operator[op](data, op2))? "success": "fail" return (operator[op](data, op2))? "success": "fail"
} }
async function getTimeLines(aliases, map) { async function speculative_deployment(aliases, map) {
if (constants.speculative_deployment) { if (constants.speculative_deployment) {
console.log(aliases); console.log(aliases);
let getData = [] let getData = []
for (const [mod, metadata] of Object.entries(map)) { for (const [mod, metadata] of Object.entries(map)) {
console.log(mod, metadata, aliases[mod].alias); if (constants.JIT_deployment) {
console.log(mod, metadata, aliases[mod].alias);
let url = metricsDB + aliases[mod].alias let url = metricsDB + aliases[mod].alias
console.log(url); console.log(url);
let data = libSupport.fetchData(url) let data = libSupport.fetchData(url)
console.log(data); console.log(data);
getData.push(data) getData.push(data)
} else {
let payload = [{
topic: constants.topics.hscale,
messages: JSON.stringify({ "runtime": metadata.runtime, "functionHash": aliases[mod].alias })
}]
notify(payload)
}
} }
if (constants.JIT_deployment) {
Promise.all(getData).then((values) => {
let dataMap = new Map()
for (const data of values) {
dataMap[data._id] = data
}
Promise.all(getData).then((values) => { let done = new Map()
let dataMap = new Map() let toBeDone = new Set()
for (const data of values) { // var plannerMap = new Map(map)
dataMap[data._id] = data do {
} for (const [mod, metadata] of Object.entries(map)) {
if (metadata.wait_for.length == 0 && done[mod] === undefined) {
let done =new Map() done[mod] = dataMap[aliases[mod].alias][metadata.runtime].coldstart // expecting the first ones to run
let toBeDone = new Set() // to be hit by coldstarts
// var plannerMap = new Map(map) // delete plannerMap[mod];
do { } else if (done[mod] === undefined) {
for (const [mod, metadata] of Object.entries(map)) { let flag = true
if (metadata.wait_for.length == 0 && done[mod] === undefined) { let maxWait = 0
done[mod] = dataMap[aliases[mod].alias][metadata.runtime].coldstart // expecting the first ones to run for (const dependency of metadata.wait_for) {
// to be hit by coldstarts console.log(dependency);
// delete plannerMap[mod];
} else if (done[mod] === undefined) { if (done[dependency] === undefined) {
let flag = true flag = false
let maxWait = 0 break
for (const dependency of metadata.wait_for) { } else if (maxWait < done[dependency])
console.log(dependency); maxWait = done[dependency]
}
if (done[dependency] === undefined) { if (flag) {
flag = false console.log("notifying", mod);
break let notifyTime = ((maxWait - dataMap[aliases[mod].alias][metadata.runtime].starttime) > 0) ?
} else if (maxWait < done[dependency]) maxWait - dataMap[aliases[mod].alias][metadata.runtime].starttime : 0
maxWait = done[dependency] console.log(mod, "max wait", maxWait, "notify time:", notifyTime);
} let payload = [{
if (flag) { topic: constants.topics.hscale,
console.log("notifying", mod); messages: JSON.stringify({ "runtime": metadata.runtime, "functionHash": aliases[mod].alias })
let notifyTime = ((maxWait - dataMap[aliases[mod].alias][metadata.runtime].starttime) > 0)? }]
maxWait - dataMap[aliases[mod].alias][metadata.runtime].starttime: 0 setTimeout(notify, notifyTime, payload)
console.log(mod, "max wait", maxWait, "notify time:", notifyTime); done[mod] = maxWait + dataMap[aliases[mod].alias][metadata.runtime].warmstart
let payload = [{ if (toBeDone.has(mod))
topic: constants.topics.hscale, delete toBeDone[mod]
messages: JSON.stringify({ "runtime": metadata.runtime, "functionHash": aliases[mod].alias }) // delete plannerMap[mod]
}] } else {
// setTimeout(notify, notifyTime, payload) toBeDone.add(mod)
done[mod] = maxWait + dataMap[aliases[mod].alias][metadata.runtime].warmstart }
if (toBeDone.has(mod))
delete toBeDone[mod]
// delete plannerMap[mod]
} else {
toBeDone.add(mod)
} }
console.log(done, toBeDone);
} }
console.log(done, toBeDone); } while (toBeDone.size != 0)
} })
} while (toBeDone.size != 0) }
})
} }
} }
......
dispatch_system/dispatch_manager/repository/worker_env/env.js
View file @ db8d9fb6
...@@ -5,7 +5,8 @@ let request = require('request') ...@@ -5,7 +5,8 @@ let request = require('request')
const process = require('process') const process = require('process')
const app = express() const app = express()
let port = 5000, resource_id, functionHash, runtime, idleTime = 30 let port = 5000, resource_id, functionHash, runtime, idleTime = 60, flagFirstRequest = true
let waitTime
resource_id = process.argv[2] resource_id = process.argv[2]
functionHash = process.argv[3] functionHash = process.argv[3]
...@@ -28,6 +29,10 @@ app.use(bodyParser.json()) ...@@ -28,6 +29,10 @@ app.use(bodyParser.json())
let lastRequest = Date.now(), totalRequest = 0 let lastRequest = Date.now(), totalRequest = 0
app.post('/serverless/function/execute/', (req, res) => { app.post('/serverless/function/execute/', (req, res) => {
if (flagFirstRequest) {
waitTime = Date.now() - waitTime
flagFirstRequest = false
}
let payload = req.body let payload = req.body
lastRequest = Date.now() lastRequest = Date.now()
totalRequest++ totalRequest++
...@@ -59,6 +64,7 @@ app.listen(port, () => { ...@@ -59,6 +64,7 @@ app.listen(port, () => {
runtime, resource_id, entity_id: process.pid}), runtime, resource_id, entity_id: process.pid}),
"status": true "status": true
}], () => { }) }], () => { })
waitTime = Date.now()
}) })
function shouldDie() { function shouldDie() {
...@@ -67,7 +73,7 @@ function shouldDie() { ...@@ -67,7 +73,7 @@ function shouldDie() {
let message = JSON.stringify({ let message = JSON.stringify({
functionHash, portExternal: port, functionHash, portExternal: port,
runtime, resource_id, entity_id: process.pid, runtime, resource_id, entity_id: process.pid,
total_request: totalRequest total_request: totalRequest, wait_time: waitTime
}) })
console.log("Idle for too long. Exiting"); console.log("Idle for too long. Exiting");
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment