basic plugin cluster handling

libs/plugins.js

@@ -72,10 +72,60 @@ module.exports = function(s,config,lang,io){
         s.debugLog(`resetDetectorPluginArray : ${JSON.stringify(pluginArray)}`)
         s.detectorPluginArray = pluginArray
     }
-    s.sendToAllDetectors = function(data){
+    if(config.detectorPluginsCluster){
-        s.detectorPluginArray.forEach(function(name){
+        s.debugLog(`Detector Plugins running in Cluster Mode`)
-            s.connectedPlugins[name].tx(data)
+        var currentPluginCpuUsage = {}
-        })
+        var currentPluginGpuUsage = {}
+        if(config.clusterBasedOnGpu){
+            const getPluginWithLowestUtilization = () => {
+                var selectedPluginServer = null
+                var lowestUsed = 1000
+                s.detectorPluginArray.forEach((pluginName) => {
+                    var overAllPercent = 0
+                    var gpus = currentPluginGpuUsage[pluginName]
+                    gpus.forEach((gpu) => {
+                        console.log(gpu)
+                        const percent = gpu.utilization
+                        overAllPercent += percent
+                    })
+                    if((overAllPercent / gpus.length) < lowestUsed){
+                        selectedPluginServer = pluginName
+                        lowestUsed = overAllPercent
+                    }
+                })
+                if(selectedPluginServer){
+                    return s.connectedPlugins[selectedPluginServer]
+                }else{
+                    return {tx: () => {}}
+                }
+            }
+        }else{
+            const getPluginWithLowestUtilization = () => {
+                var selectedPluginServer = null
+                var lowestUsed = 1000
+                s.detectorPluginArray.forEach((pluginName) => {
+                    const percent = currentPluginCpuUsage[pluginName]
+                    if(percent < lowestUsed){
+                        selectedPluginServer = pluginName
+                        lowestUsed = percent
+                    }
+                })
+                if(selectedPluginServer){
+                    return s.connectedPlugins[selectedPluginServer]
+                }else{
+                    return {tx: () => {}}
+                }
+            }
+        }
+        s.sendToAllDetectors = function(data){
+            getPluginWithLowestUtilization().tx(data)
+        }
+    }else{
+        s.sendToAllDetectors = function(data){
+            s.detectorPluginArray.forEach(function(name){
+                s.connectedPlugins[name].tx(data)
+            })
+        }
     }
     s.sendDetectorInfoToClient = function(data,txFunction){
         s.detectorPluginArray.forEach(function(name){
@@ -224,16 +274,28 @@ module.exports = function(s,config,lang,io){
         }
     }
     var onWebSocketConnection = function(cn){
+        if(config.detectorPluginsCluster){
+            const addCpuUsageHandler = (pluginName) => {
+                cn.on('cpuUsage',function(percent){
+                    currentPluginCpuUsage[pluginName] = percent
+                })
+                cn.on('gpuUsage',function(gpus){
+                    currentPluginGpuUsage[pluginName] = gpus
+                })
+            }
+        }
         cn.on('ocv',function(d){
             if(!cn.pluginEngine && d.f === 'init'){
                 if(config.pluginKeys[d.plug] === d.pluginKey){
                     s.pluginInitiatorSuccess("client",d,cn)
+                    if(config.detectorPluginsCluster)addCpuUsageHandler(d.plug)
                 }else{
                     s.pluginInitiatorFail("client",d,cn)
                 }
             }else{
                 if(config.pluginKeys[d.plug] === d.pluginKey){
                     s.pluginEventController(d)
+                    if(config.detectorPluginsCluster)addCpuUsageHandler(d.plug)
                 }else{
                     cn.disconnect()
                 }

plugins/pluginBase.js

@@ -114,6 +114,75 @@ module.exports = function(__dirname, config){
     s.detectObject = (buffer,d,tx,frameLocation) => {
         console.log('detectObject handler not set')
     }
+    const getCpuUsage = (callback) => {
+        var k = {}
+        switch(s.platform){
+            case'win32':
+                k.cmd = "@for /f \"skip=1\" %p in ('wmic cpu get loadpercentage') do @echo %p%"
+            break;
+            case'darwin':
+                k.cmd = "ps -A -o %cpu | awk '{s+=$1} END {print s}'";
+            break;
+            case'linux':
+                k.cmd = 'top -b -n 2 | awk \'toupper($0) ~ /^.?CPU/ {gsub("id,","100",$8); gsub("%","",$8); print 100-$8}\' | tail -n 1';
+            break;
+            case'freebsd':
+                k.cmd = 'vmstat 1 2 | awk \'END{print 100-$19}\''
+            break;
+	    case'openbsd':
+                k.cmd = 'vmstat 1 2 | awk \'END{print 100-$18}\''
+            break;
+        }
+        if(config.customCpuCommand){
+          exec(config.customCpuCommand,{encoding:'utf8',detached: true},function(err,d){
+              if(s.isWin===true) {
+                  d = d.replace(/(\r\n|\n|\r)/gm, "").replace(/%/g, "")
+              }
+              callback(d)
+              s.onGetCpuUsageExtensions.forEach(function(extender){
+                  extender(d)
+              })
+          })
+        } else if(k.cmd){
+             exec(k.cmd,{encoding:'utf8',detached: true},function(err,d){
+                 if(s.isWin===true){
+                     d=d.replace(/(\r\n|\n|\r)/gm,"").replace(/%/g,"")
+                 }
+                 callback(d)
+                 s.onGetCpuUsageExtensions.forEach(function(extender){
+                     extender(d)
+                 })
+             })
+        } else {
+            callback(0)
+        }
+    }
+    const parseNvidiaSmi = function(callback){
+        var response = {
+            ok: true,
+        }
+        exec('nvidia-smi -x -q',function(err,data){
+            var response = xmlParser.toJson(data)
+            var newArray = []
+            try{
+                JSON.parse(response).nvidia_smi_log.gpu.forEach((gpu)=>{
+                    newArray.push({
+                        id: gpu.minor_number,
+                        name: gpu.product_name,
+                        brand: gpu.product_brand,
+                        fan_speed: gpu.fan_speed,
+                        temperature: gpu.temperature,
+                        power: gpu.power_readings,
+                        utilization: gpu.utilization,
+                        maxClocks: gpu.max_clocks,
+                    })
+                })
+            }catch(err){
+
+            }
+            if(callback)callback(newArray)
+        })
+    }
     s.onCameraInitExtensions = []
     s.onCameraInit = (extender) => {
         s.onCameraInitExtensions.push(extender)
@@ -317,6 +386,22 @@ module.exports = function(__dirname, config){
     s.getWebsocket = () => {
         return io
     }
+    if(config.clusterMode){
+        plugLog('Plugin enabling Cluster Mode...')
+        if(config.clusterBasedOnGpu){
+            setTimeout(() => {
+                parseNvidiaSmi((gpus)=>{
+                    io.emit('gpuUsage',gpus)
+                })
+            },1000 * 10)
+        }else{
+            setTimeout(() => {
+                getCpuUsage((percent) => {
+                    io.emit('cpuUsage',percent)
+                })
+            },1000 * 10)
+        }
+    }
     s.createPythonScriptDaemon = () => {
         if(!config.pythonScript){config.pythonScript = config.dirname + '/pumpkin.py'}
         if(!config.pythonPort){config.pythonPort = 7990}