Merge pull request #128 from kube-HPC/bug_wrapper_algorunner_out_of_m…

…emory

Bug wrapper algorunner out of memory

hkube_python_wrapper/wrapper/algorunner.py

@@ -52,6 +52,7 @@ def __init__(self):
         self._stopped = False
         self._redirectLogs = False
         self._printThread = 0
+        self._done = True
         DaemonThread.__init__(self, "WorkerListener")
 
     @staticmethod
@@ -316,6 +317,16 @@ def _getMethod(self, name):
             return self._algorithm.get(name)
         return None
 
+    def _aliveSignal(self):
+        def routine():
+            while not self._done and not self.stopped:
+                self._sendCommand(messages.outgoing.alive, None)
+                print("stopped is: ", self._stopped)
+                time.sleep(5)
+
+        thread = threading.Thread(target=routine)
+        thread.start()
+
     def _init(self, options):
         redirector = None
         try:
@@ -335,6 +346,8 @@ def _init(self, options):
                     if (method is not None):
                         method(options)
                 self._sendCommand(messages.outgoing.initialized, None)
+                self._done = False
+                self._aliveSignal()
 
         except Exception as e:
             self.sendError(e)
@@ -404,6 +417,7 @@ def _start(self, options):
                     method(self._input)
             method = self._getMethod('start')
             algorithmData = method(self._input, self._hkubeApi)
+            self._done = True
             if not (self._stopped):
                 self._handle_response(algorithmData, jobId, taskId, nodeName, savePaths, span)
 

hkube_python_wrapper/wrapper/messages.py

@@ -22,6 +22,7 @@ def __init__(self):
         self.streamingOutMessage = "streamingOutMessage"
         self.streamingInMessageDone = "streamingInMessageDone"
         self.logData = "logData"
+        self.alive = "alive"
 
 
 class Incoming(object):