Make fetch_input count as a heartbeat, and ensure that functions that return None on purpose work.

Author: dan-blanchard

gridmap/job.py

@@ -79,9 +79,12 @@
 
 
 # Set of "not running" job statuses
-SLEEP_STATUSES = {psutil.STATUS_SLEEPING, psutil.STATUS_DEAD,
-                  psutil.STATUS_IDLE, psutil.STATUS_STOPPED,
-                  psutil.STATUS_ZOMBIE}
+_SLEEP_STATUSES = {psutil.STATUS_SLEEPING, psutil.STATUS_DEAD,
+                   psutil.STATUS_IDLE, psutil.STATUS_STOPPED,
+                   psutil.STATUS_ZOMBIE}
+
+# Placeholder string, since a job could potentially return None on purpose
+_JOB_NOT_FINISHED = '*@#%$*@#___GRIDMAP___NOT___DONE___@#%**#*$&*%'
 
 
 class JobException(Exception):
@@ -152,7 +155,7 @@ def __init__(self, f, args, kwlist=None, cleanup=True, mem_free="1G",
         self.jobid = -1
         self.kwlist = kwlist if kwlist is not None else {}
         self.cleanup = cleanup
-        self.ret = None
+        self.ret = _JOB_NOT_FINISHED
         self.num_slots = num_slots
         self.mem_free = mem_free
         self.white_list = []
@@ -327,6 +330,7 @@ def check(self, session_id, jobs):
 
                     if msg["command"] == "fetch_input":
                         return_msg = self.jobid_to_job[job_id]
+                        job.timestamp = datetime.now()
 
                     if msg["command"] == "store_output":
                         # be nice
@@ -386,14 +390,14 @@ def check_if_alive(self):
         check if jobs are alive and determine cause of death if not
         """
         logger = logging.getLogger(__name__)
+        logger.debug('Checking if jobs are alive')
         for job in self.jobs:
 
             # noting was returned yet
-            if job.ret is None:
+            if job.ret == _JOB_NOT_FINISHED:
 
                 # exclude first-timers
                 if job.timestamp is not None:
-
                     # check heart-beats if there was a long delay
                     current_time = datetime.now()
                     time_delta = current_time - job.timestamp
@@ -402,7 +406,7 @@ def check_if_alive(self):
                         job.cause_of_death = "unknown"
                     elif (len(job.track_cpu) > MAX_IDLE_HEARTBEATS and
                           all((cpu_load <= IDLE_THRESHOLD and
-                               state in SLEEP_STATUSES) for cpu_load, state in
+                               state in _SLEEP_STATUSES) for cpu_load, state in
                               job.track_cpu[-MAX_IDLE_HEARTBEATS:])):
                         logger.error('Job stalled for unknown reason.')
                         job.cause_of_death = 'stalled'
@@ -416,7 +420,7 @@ def check_if_alive(self):
 
             # attempt to resubmit
             if job.cause_of_death:
-                logger.info("creating error report")
+                logger.info("Creating error report")
 
                 # send report
                 send_error_mail(job)