Feature/gpu support extended

FunctionWorker/python/SessionHelperThread.py

@@ -235,6 +235,7 @@ def _store_message(self, msg):
         self._message_queue.put(msg)
 
     def _handle_special_message(self, msg):
+        #self._logger.debug("[SessionHelperThread] Special message: " + str(msg))
         action = msg["action"]
 
         if action == "--stop":

FunctionWorker/python/StateUtils.py

@@ -419,15 +419,21 @@ def evaluateMapState(self, function_input, key, metadata, sapi):
         self._logger.debug("[StateUtils] evaluateMapState, maxConcurrency: " + str(maxConcurrency))
         self._logger.debug("[StateUtils] evaluateMapState metadata: " + str(metadata))
 
+        self._logger.info("[StateUtils] evaluateMapState, maxConcurrency: " + str(maxConcurrency))
+        self._logger.info("[StateUtils] evaluateMapState metadata: " + str(metadata))
+
         counter_name_topic = self.functionstatename + "-" + self.sandboxid
 
         total_branch_count = len(function_input) # all branches executed concurrently
-
+        #sapi.put(name_prefix + "_" + "mapInputCount", str(len(function_input)))
+
         klist = [total_branch_count]
 
         self.parsedfunctionstateinfo["BranchCount"] = int(total_branch_count) # overwrite parsed BranchCount with new value
         self._logger.debug("[StateUtils] evaluateMapState, total_branch_count: " + str(total_branch_count))
 
+        self._logger.info("[StateUtils] evaluateMapState, total_branch_count: " + str(total_branch_count))
+
         # prepare counter metadata
         counter_metadata = {}
         counter_metadata["__state_action"] = "post_map_processing"
@@ -459,6 +465,7 @@ def evaluateMapState(self, function_input, key, metadata, sapi):
         counter_name_value_metadata["__state_action"] = "post_map_processing"
         counter_name_value_metadata["state_counter"] = metadata["state_counter"]
         self._logger.debug("[StateUtils] evaluateMapState, metadata[state_counter]: " + str(metadata["state_counter"]))
+        self._logger.info("[StateUtils] evaluateMapState, metadata[state_counter]: " + str(metadata["state_counter"]))
         self.mapStateCounter = int(metadata["state_counter"])
 
         counter_name_value = {"__mfnmetadata": counter_name_value_metadata, "__mfnuserdata": '{}'}
@@ -506,6 +513,8 @@ def evaluateMapState(self, function_input, key, metadata, sapi):
         assert py3utils.is_string(workflow_instance_metadata_storage_key)
         self._logger.debug("[StateUtils] full_metadata_encoded put key: " + str(workflow_instance_metadata_storage_key))
 
+        self._logger.info("[StateUtils] full_metadata_encoded put key: " + str(workflow_instance_metadata_storage_key))
+
         sapi.put(workflow_instance_metadata_storage_key, json.dumps(metadata))
 
         # Now provide each branch with its own input
@@ -523,9 +532,14 @@ def evaluateMapState(self, function_input, key, metadata, sapi):
             self._logger.debug("\t Map State StartAt:" + startat)
             self._logger.debug("\t Map State input:" + str(function_input[i]))
 
+            self._logger.info("\t Map State StartAt:" + startat)
+            self._logger.info("\t Map State input:" + str(function_input[i]))
+
         return function_input, metadata
 
     def evaluatePostMap(self, function_input, key, metadata, sapi):
+        self._logger.info("\t inside evaluatePostMap: " + str(function_input)+ " " + str(metadata) + " " + str(sapi))
+
 
         name_prefix = self.functiontopic + "_" + key
 
@@ -542,11 +556,15 @@ def evaluatePostMap(self, function_input, key, metadata, sapi):
 
         self._logger.debug("\t metadata:" + json.dumps(metadata))
 
+        self._logger.info("\t metadata:" + json.dumps(metadata))
+
         workflow_instance_metadata_storage_key = str(function_input["WorkflowInstanceMetadataStorageKey"])
         assert py3utils.is_string(workflow_instance_metadata_storage_key)
         full_metadata_encoded = sapi.get(workflow_instance_metadata_storage_key)
         self._logger.debug("[StateUtils] full_metadata_encoded get: " + str(full_metadata_encoded))
 
+        self._logger.info("[StateUtils] full_metadata_encoded get: " + str(full_metadata_encoded))
+
         full_metadata = json.loads(full_metadata_encoded)
         full_metadata["state_counter"] = state_counter
 
@@ -557,6 +575,8 @@ def evaluatePostMap(self, function_input, key, metadata, sapi):
         branchOutputKeysSet = sapi.retrieveSet(branchOutputKeysSetKey)
         self._logger.debug("\t branchOutputKeysSet: " + str(branchOutputKeysSet))
 
+        self._logger.info("\t branchOutputKeysSet: " + str(branchOutputKeysSet))
+
         if not branchOutputKeysSet:
             self._logger.error("[StateUtils] branchOutputKeysSet is empty")
             raise Exception("[StateUtils] branchOutputKeysSet is empty")
@@ -576,13 +596,17 @@ def evaluatePostMap(self, function_input, key, metadata, sapi):
         NumBranchesFinished = abs(counterValue)
         self._logger.debug("\t NumBranchesFinished:" + str(NumBranchesFinished))
 
+        self._logger.info("\t NumBranchesFinished:" + str(NumBranchesFinished))
+
         do_cleanup = False
 
         if klist[-1] == NumBranchesFinished:
             do_cleanup = True
 
         self._logger.debug("\t do_cleanup:" + str(do_cleanup))
 
+        self._logger.info("\t do_cleanup:" + str(do_cleanup))
+
         counterName = str(mapInfo["CounterName"])
         counter_metadata_key_name = counterName + "_metadata"
         assert py3utils.is_string(counterName)
@@ -610,6 +634,10 @@ def evaluatePostMap(self, function_input, key, metadata, sapi):
 
         self._logger.debug("\t mapInfo_BranchOutputKeys length: " + str(len(mapInfo["BranchOutputKeys"])))
 
+        self._logger.info("\t mapInfo_BranchOutputKeys:" + str(mapInfo["BranchOutputKeys"]))
+
+        self._logger.info("\t mapInfo_BranchOutputKeys length: " + str(len(mapInfo["BranchOutputKeys"])))
+
         for outputkey in mapInfo["BranchOutputKeys"]:
             outputkey = str(outputkey)
             if outputkey in branchOutputKeysSet: # mapInfo["BranchOutputKeys"]:
@@ -623,15 +651,23 @@ def evaluatePostMap(self, function_input, key, metadata, sapi):
                 self._logger.debug("\t branchOutput:" + branchOutput)
                 self._logger.debug("\t branchOutput_decoded(type):" + str(type(branchOutput_decoded)))
                 self._logger.debug("\t branchOutput_decoded:" + str(branchOutput_decoded))
+                self._logger.info("\t branchOutput(type):" + str(type(branchOutput)))
+                self._logger.info("\t branchOutput:" + branchOutput)
+                self._logger.info("\t branchOutput_decoded(type):" + str(type(branchOutput_decoded)))
+                self._logger.info("\t branchOutput_decoded:" + str(branchOutput_decoded))
                 post_map_output_values = post_map_output_values + [branchOutput_decoded]
                 if do_cleanup:
                     sapi.delete(outputkey) # cleanup the key from data layer
                     self._logger.debug("\t cleaned output key:" + outputkey)
+                    self._logger.info("\t cleaned output key:" + outputkey)
             else:
                 post_map_output_values = post_map_output_values + [None]
                 self._logger.debug("\t this_BranchOutputKeys is not contained: " + str(outputkey))
 
+                self._logger.info("\t this_BranchOutputKeys is not contained: " + str(outputkey))
+
         self._logger.debug("\t post_map_output_values:" + str(post_map_output_values))
+        self._logger.info("\t post_map_output_values:" + str(post_map_output_values))
         while (sapi.get(name_prefix + "_" + "mapStatePartialResult")) == "":
             time.sleep(0.1) # wait until value is available
 
@@ -640,18 +676,25 @@ def evaluatePostMap(self, function_input, key, metadata, sapi):
         mapStatePartialResult += post_map_output_values
         sapi.put(name_prefix + "_" + "mapStatePartialResult", str(mapStatePartialResult))
 
+        time.sleep(5.0)
+
         # now apply ResultPath and OutputPath
         if do_cleanup:
 
             sapi.deleteSet(branchOutputKeysSetKey)
 
-        while sapi.get(name_prefix + "_" + "mapStatePartialResult") != str(mapStatePartialResult):
-            time.sleep(0.1)
+        while (sapi.get(name_prefix + "_" + "mapInputCount") == ""):
+            time.sleep(0.1) # wait until value is available
 
         if ast.literal_eval(sapi.get(name_prefix + "_" + "mapInputCount")) == len(mapStatePartialResult):
+            #time.sleep(0.5)
 
             # we are ready to publish  but need to honour ResultPath and OutputPath
+            while (sapi.get(name_prefix + "_" +"mapStatePartialResult") == ""):
+                time.sleep(0.1)
             res_raw = ast.literal_eval(sapi.get(name_prefix + "_" +"mapStatePartialResult"))
+            self._logger.info("[StateUtils] evaluatePostMap: res_raw" + str(res_raw) + " vs. " + sapi.get(name_prefix + "_" + "mapInputCount"))
+
 
             # remove unwanted keys from input before publishing
             function_input = {}
@@ -671,6 +714,9 @@ def evaluatePostMap(self, function_input, key, metadata, sapi):
             sapi.delete(name_prefix + "_" + "mapStatePartialResult")
             sapi.delete(name_prefix + "_" + "tobeProcessedlater")
             post_map_output_values = function_input_post_output
+        else:
+            #raise Exception("mapInputCount" + str(sapi.get(name_prefix + "_" + "mapInputCount")) + " does not match mapStatePartialResult: " + str(mapStatePartialResult))
+            print("mapInputCount" + str(sapi.get(name_prefix + "_" + "mapInputCount")) + " does not match mapStatePartialResult: " + str(mapStatePartialResult))
         return post_map_output_values, full_metadata
 
     def evaluateParallelState(self, function_input, key, metadata, sapi):
@@ -967,7 +1013,7 @@ def evaluatePostParallel(self, function_input, key, metadata, sapi):
 
     def evaluateNonTaskState(self, function_input, key, metadata, sapi):
         # 3. Evaluate Non Task states
-        #self._logger.debug("[StateUtils] NonTask state type: " + str(self.functionstatetype))
+        self._logger.info("[StateUtils] NonTask state type: " + str(self.functionstatetype))
         #self._logger.debug("[StateUtils] Welcome to evaluateNonTaskState! Current key:" + str(key))
         function_output = None
         if self.functionstatetype == StateUtils.choiceStateType:
@@ -1094,6 +1140,9 @@ def evaluateNonTaskState(self, function_input, key, metadata, sapi):
             self._logger.debug("[StateUtils] Map state maxConcurrency: " + str(maxConcurrency))
             self._logger.debug("[StateUtils] Map state handling")
 
+            self._logger.info("[StateUtils] Map state maxConcurrency: " + str(maxConcurrency))
+            self._logger.info("[StateUtils] Map state handling metadata: " + str(metadata) )
+
             if "__state_action" not in metadata or metadata["__state_action"] != "post_map_processing":
                 # here we start the iteration process on a first batch
                 if maxConcurrency != 0:
@@ -1103,32 +1152,41 @@ def evaluateNonTaskState(self, function_input, key, metadata, sapi):
                     tobeProcessednow = function_input
                     tobeProcessedlater = []
                 self._logger.debug("[StateUtils] Map state function_input split:" + str(tobeProcessednow) + " " + str(tobeProcessedlater))
+                self._logger.info("[StateUtils] Map state function_input split:" + str(tobeProcessednow) + " " + str(tobeProcessedlater))
                 sapi.put(name_prefix + "_" + "tobeProcessedlater", str(tobeProcessedlater)) # store elements to be processed on DL
                 sapi.put(name_prefix + "_" + "mapStatePartialResult", "[]") # initialise the collector variable
                 sapi.put(name_prefix + "_" + "mapInputCount", str(len(function_input)))
-                #if "__client_origin_frontend" in metadata and metadata["__client_origin_frontend"] != "":
-                #    sapi.put(name_prefix + "_" + "mapInputOriginFE", str(metadata["__client_origin_frontend"]))
+                #metadata["__state_action"] = ""
 
                 function_output, metadata = self.evaluateMapState(tobeProcessednow, key, metadata, sapi)
+                #metadata["__state_action"] = ""
+
 
             elif metadata["__state_action"] == "post_map_processing":
                 tobeProcessedlater = ast.literal_eval(sapi.get(name_prefix + "_" + "tobeProcessedlater")) # get all elements that have not yet been processed
                 self._logger.debug("[StateUtils] Map state post_map processing input:" + str(tobeProcessedlater))
+                self._logger.info("[StateUtils] Map state post_map processing input:" + str(tobeProcessedlater))
                 # we need to decide at this point if there is a need for more batches. if so:
 
                 if len(tobeProcessedlater) > 0: # we need to start another batch
-                    function_output, metadata = self.evaluatePostMap(function_input, key, metadata, sapi) # take care not to overwrite metadata
-                    metadata["__client_origin_frontend"] =  self._internal_endpoint
-                    time.sleep(5.0) # allow cleanup of previous execution before launching new
-                    #metadata["__client_origin_frontend"] = sapi.get(name_prefix + "_" + "mapInputOriginFE")
-                    self._logger.debug("[StateUtils] Map state metadata between calls: " + str(metadata))
+                    self._logger.info("[StateUtils] tobeProcessedlater: " + str(tobeProcessedlater)+ ", function_input: " +str(function_input))
+                    function_output, metadata2 = self.evaluatePostMap(function_input, key, metadata, sapi) # take care not to overwrite metadata
+                    self._logger.info("[StateUtils] after evaluatPostMap: " + str(function_output))
                     function_output, metadata = self.evaluateMapState(tobeProcessedlater[:maxConcurrency], key, metadata, sapi) # start a new batch
+                    self._logger.info("[StateUtils] after evaluateMapState:" + str(function_output))
+                    self._logger.info("[StateUtils] after evaluateMapState, metadata: " + str(metadata))
                     sapi.put(name_prefix + "_" + "tobeProcessedlater", str(tobeProcessedlater[maxConcurrency:])) # store remaining elements to be processed on DL
+                    self._logger.info("[StateUtils] after sapi.put: " + str(tobeProcessedlater[maxConcurrency:]))
+
 
                 else:# no more batches required. we are at the iteration end, publish the final result
                     self._logger.debug("[StateUtils] Map state input final stage: " + str(function_input))
+                    self._logger.info("[StateUtils] Map state input final stage: " + str(function_input))
                     function_output, metadata = self.evaluatePostMap(function_input, key, metadata, sapi)
 
+            elif metadata["__state_action"] == '':
+                raise Exception("Unkown state action in map state")
+
             else:
                 raise Exception("Unknow action type in map state")
 

GUI/app/pages/functions/FunctionTableCtrl.js

@@ -171,7 +171,6 @@
       return selected.length ? mFunction.runtime : 'Not set';
     };
 
-
     function createTemporaryWorkflow(functionIndex) {
       var req = {
         method: 'POST',
@@ -636,7 +635,7 @@
 
           },
 
-          data:   JSON.stringify({ "action" : "addFunction", "data" : { "user" : { "token" : token } , "function" : { "name" : mFunction.name, "runtime" : mFunction.runtime } } })
+		data:   JSON.stringify({ "action" : "addFunction", "data" : { "user" : { "token" : token } , "function" : { "name" : mFunction.name, "runtime" : mFunction.runtime, "gpu_usage": mFunction.gpu_usage, "gpu_mem_usage": mFunction.gpu_mem_usage } } })
 
         }
         $http(req).then(function successCallback(response) {
@@ -646,7 +645,7 @@
               console.log('new function id:' + response.data.data.function.id);
               toastr.success('Your function has been created successfully!');
               $scope.reloadFunctions();
-              $scope.open('app/pages/functions/modals/codeEditorModal.html', 'lg', mFunction.id, mFunction.name, mFunction.status, mFunction.runtime);
+              $scope.open('app/pages/functions/modals/codeEditorModal.html', 'lg', mFunction.id, mFunction.name, mFunction.status, mFunction.runtime, mFunction.gpu_usage, mFunction.gpu_mem_usage);
 
             } else {
               console.log("Failure status returned by addFunction");
@@ -692,7 +691,7 @@
 
           },
 
-          data:   JSON.stringify({ "action" : "modifyFunction", "data" : { "user" : { "token" : token } , "function" : { "id": mFunction.id, "name" : mFunction.name, "runtime" : mFunction.runtime } } })
+		data:   JSON.stringify({ "action" : "modifyFunction", "data" : { "user" : { "token" : token } , "function" : { "id": mFunction.id, "name" : mFunction.name, "runtime" : mFunction.runtime, "gpu_usage" : mFunction.gpu_usage, "gpu_mem_usage": mFunction.gpu_mem_usage } } })
 
         }
         $http(req).then(function successCallback(response) {
@@ -738,6 +737,8 @@
         name: '',
         status: 'undeployed',
         runtime: 'Python 3.6',
+        gpu_usage: '0',
+        gpu_mem_usage: '0',
         modified: '0'
       };
       $scope.functions.push($scope.inserted);

GUI/app/pages/functions/widgets/editableRowTable.html

@@ -24,6 +24,8 @@
 
     <th st-sort="name" style="text-align:left;border:0;"><b>Name</b></th>
     <th st-sort="runtime" style="border:0;"><b>Runtime</b></th>
+    <th st-sort="gpu_cores" style="border:0;"><b>GPU Cores (%)</b></th>
+    <th st-sort="gpu_memory" style="border:0;"><b>GPU Memory (GB)</b></th>
     <th id="sortHeader" style="border:0;" st-sort="modified" st-sort-default="reverse"><b>Last Modified</b></th>
     <th style="border:0;"><b>Actions</b></th>
   </tr>
@@ -45,6 +47,16 @@
           {{ showRuntime(mFunction) }}
         </span>
     </td>
+    <td style="text-align:left;">
+      <span editable-text="mFunction.gpu_usage" e-name="gpu_usage" e-form="rowform">
+        {{ mFunction.gpu_usage || 'Not set' }}
+      </span>
+    </td>
+     <td style="text-align:left;">
+      <span editable-text="mFunction.gpu_mem_usage" e-name="gpu_mem_usage" e-form="rowform">
+        {{ mFunction.gpu_mem_usage || 'Not set' }}
+      </span>
+    </td>
     <td>
         <span e-name="modified" e-form="rowform">
           {{ showLastModified(mFunction) }}
@@ -66,7 +78,7 @@
         </button>
       </form>
       <div class="buttons" ng-show="!rowform.$visible">
-        <button class="btn btn-primary editable-table-button btn-xs" ng-click="rowform.$show()">Rename</button>
+        <button class="btn btn-primary editable-table-button btn-xs" ng-click="rowform.$show()">Edit</button>
 
 
         <button type="button" ng-disabled="rowform.$waiting" class="btn btn-primary editable-table-button btn-xs" style="width:85px; background-color: #757a74; border-color: #757a74;" data-toggle="modal" ng-click="open('app/pages/functions/modals/codeEditorModal.html', 'lg', functions[getIndex(mFunction)].id, functions[getIndex(mFunction)].name, functions[getIndex(mFunction)].status, functions[getIndex(mFunction)].runtime)">Code Editor</button>

GUI/app/pages/workflows/WorkflowImportCtrl.js

@@ -658,7 +658,7 @@
 
          },
 
-         data:  JSON.stringify({ "action" : "addFunction", "data" : { "user" : { "token" : token } , "function" : { "name" : functionName, "runtime" : functionRuntime } } })
+	       data:  JSON.stringify({ "action" : "addFunction", "data" : { "user" : { "token" : token } , "function" : { "name" : functionName, "runtime" : functionRuntime, "gpu_usage": functionGpuusage, "gpu_mem_usage": functionGpuMemusage  } } })
 
        }
        $http(req).then(function successCallback(response) {

ManagementService/management_init.py

@@ -419,7 +419,7 @@ def printUsage():
         sys.path.append(workflowdir)
         if os.getenv("KUBERNETES_PORT", None) != None:
             import deployWorkflow
-            url, endpoint_key = deployWorkflow.create_k8s_deployment(email, workflow_info, "Python", management=True)
+            url, endpoint_key = deployWorkflow.create_k8s_deployment(email, workflow_info, "Python", 0, 0, management=True)
             DLCLIENT_MANAGEMENT.putMapEntry("Management_workflow_endpoint_map", endpoint_key, url)
             # Kubernetes mode only has one url
             endpoint_list = [url]