Added timeout property to Node to allow setting a maximum runtime for a given Worker. Update JSON serde to support this.

Author: Nathan Lee

pyrunner/core/node.py

@@ -50,6 +50,7 @@ def __init__(self, id=-1, name=None):
     self._wait_start = 0
     self._start_time = 0
     self._end_time = 0
+    self._timeout = float('inf')
     self._retcode = multiprocessing.Value('i', 0)
     self._thread = None
     self._worker_dir = None
@@ -130,6 +131,14 @@ def poll(self, wait=False):
           self._wait_start = time.time()
           logger.restart_message(self._attempts)
           self._retcode.value = -1
+    elif (time.time() - self._start_time) >= self._timeout:
+      self._thread.terminate()
+      running = False
+      logger = lg.FileLogger(self.logfile)
+      logger.open(False)
+      logger.error('Worker runtime has exceeded the set maximum/timeout of {} seconds.'.format(self._timeout))
+      logger.close()
+      self._retcode.value = 906
 
     return self.retcode if (not running or wait) else None
 
@@ -140,6 +149,7 @@ def terminate(self):
       logger.open(False)
       logger._system_("Keyboard Interrupt (SIGINT) received. Terminating all Worker and exiting.")
       logger.close()
+      self._retcode.value = 907
     return
 
 
@@ -414,6 +424,16 @@ def retry_wait_time(self, value):
     self._retry_wait_time = int(value)
     return self
 
+  @property
+  def timeout(self):
+    return getattr(self, '_timeout', float('inf'))
+  @timeout.setter
+  def timeout(self, value):
+    if int(value) < 1:
+      raise ValueError('timeout must be greater than 0')
+    self._timeout = int(value)
+    return self
+  
   @property
   def parent_nodes(self):
     return self._parent_nodes

pyrunner/core/register.py

@@ -242,5 +242,7 @@ def add_node(self, **kwargs):
       node.max_attempts = kwargs.get('max_attempts')
     if kwargs.get('retry_wait_time'):
       node.retry_wait_time = kwargs.get('retry_wait_time')
+    if kwargs.get('timeout'):
+      node.timeout = kwargs.get('timeout')
 
     return self.add_node_object(node, kwargs.get('status', constants.STATUS_PENDING), kwargs.get('dependencies', ['PyRunnerRootNode']), kwargs.get('named_deps', True))

pyrunner/serde/json.py

@@ -91,5 +91,7 @@ def serialize(self, register):
         obj['tasks'][node.name]['retry_wait_time'] = node.retry_wait_time
       if node.arguments:
         obj['tasks'][node.name]['arguments'] = node.arguments
+      if node.timeout != float('inf'):
+        obj['tasks'][node.name]['timeout'] = node.timeout
 
-    return json.dumps(obj)
+    return json.dumps(obj, indent=4)

pyrunner/version.py

@@ -1 +1 @@
-__version__ = '4.3.0'
+__version__ = '4.4.0'