Merge pull request #410 from int-brain-lab/updateHeld

Update held dependent tasks on complete

Author: k1o0

ibllib/pipes/tasks.py

@@ -118,7 +118,7 @@ def run(self, **kwargs):
                         return self.status
                 self.outputs = self._run(**kwargs)
                 _logger.info(f"Job {self.__class__} complete")
-        except BaseException:
+        except Exception:
             _logger.error(traceback.format_exc())
             _logger.info(f"Job {self.__class__} errored")
             self.status = -1
@@ -410,7 +410,7 @@ def create_alyx_tasks(self, rerun__status__in=None):
         tasks_alyx = []
         # creates all the tasks by iterating through the ordered dict
         for k, t in self.tasks.items():
-            # get the parents alyx ids to reference in the database
+            # get the parents' alyx ids to reference in the database
             if len(t.parents):
                 pnames = [p.name for p in t.parents]
                 parents_ids = [ta['id'] for ta in tasks_alyx if ta['name'] in pnames]
@@ -492,10 +492,10 @@ def run_alyx_task(tdict=None, session_path=None, one=None, job_deck=None,
     :return:
     """
     registered_dsets = []
+    # here we need to check parents' status, get the job_deck if not available
+    if not job_deck:
+        job_deck = one.alyx.rest('tasks', 'list', session=tdict['session'], no_cache=True)
     if len(tdict['parents']):
-        # here we need to check parents status, get the job_deck if not available
-        if not job_deck:
-            job_deck = one.alyx.rest('tasks', 'list', session=tdict['session'], no_cache=True)
         # check the dependencies
         parent_tasks = filter(lambda x: x['id'] in tdict['parents'], job_deck)
         parent_statuses = [j['status'] for j in parent_tasks]
@@ -526,7 +526,7 @@ def run_alyx_task(tdict=None, session_path=None, one=None, job_deck=None,
     else:
         try:
             registered_dsets = task.register_datasets(one=one, max_md5_size=max_md5_size)
-        except BaseException:
+        except Exception:
             _logger.error(traceback.format_exc())
             patch_data['status'] = 'Errored'
         patch_data['status'] = 'Complete'
@@ -541,5 +541,15 @@ def run_alyx_task(tdict=None, session_path=None, one=None, job_deck=None,
         patch_data['status'] = 'Incomplete'
     # update task status on Alyx
     t = one.alyx.rest('tasks', 'partial_update', id=tdict['id'], data=patch_data)
+    # check for dependent held tasks
+    # NB: Assumes dependent tasks are all part of the same session!
+    next(x for x in job_deck if x['id'] == t['id'])['status'] = t['status']  # Update status in job deck
+    dependent_tasks = filter(lambda x: t['id'] in x['parents'] and x['status'] == 'Held', job_deck)
+    for d in dependent_tasks:
+        assert d['id'] != t['id'], 'task its own parent'
+        # if all their parent tasks now complete, set to waiting
+        parent_status = [next(x['status'] for x in job_deck if x['id'] == y) for y in d['parents']]
+        if all(x == 'Complete' for x in parent_status):
+            one.alyx.rest('tasks', 'partial_update', id=d['id'], data={'status': 'Waiting'})
     task.cleanUp()
     return t, registered_dsets

ibllib/tests/test_tasks.py

@@ -37,9 +37,9 @@
 desired_logs_rerun = {
     'Task00': 1,
     'Task01_void': 2,
-    'Task02_error': 2,
+    'Task02_error': 1,
     'Task10': 1,
-    'Task11': None,
+    'Task11': 1,
     'TaskIncomplete': 1,
     'TaskGpuLock': 2
 }
@@ -63,12 +63,17 @@ def _run(self, overwrite=False):
         return out_files
 
 
-# job that raises an error
+# job that raises an error on first run
 class Task02_error(ibllib.pipes.tasks.Task):
-    level = 0
+    run_count = 0
 
     def _run(self, overwrite=False):
-        raise Exception("Something dumb happened")
+        Task02_error.run_count += 1
+        if Task02_error.run_count == 1:
+            raise Exception('Something dumb happened')
+        out_files = self.session_path.joinpath('alf', 'spikes.templates.npy')
+        out_files.touch()
+        return out_files
 
 
 # job that outputs a list of files
@@ -127,6 +132,7 @@ def __init__(self, session_path=None, **kwargs):
         tasks['TaskGpuLock'] = TaskGpuLock(self.session_path)
         tasks['TaskIncomplete'] = TaskIncomplete(self.session_path)
         tasks['Task10'] = Task10(self.session_path, parents=[tasks['Task00']])
+        # When both its parents Complete, this task should be set to Waiting and should finally complete
         tasks['Task11'] = Task11(self.session_path, parents=[tasks['Task02_error'],
                                                              tasks['Task00']])
         self.tasks = tasks
@@ -197,8 +203,11 @@ def test_pipeline_alyx(self):
 
         # test the rerun option
         task_deck, dsets = pipeline.rerun_failed(machine='testmachine')
-        check_statuses = [desired_statuses[t['name']] == t['status'] for t in task_deck]
-        self.assertTrue(all(check_statuses))
+        task_02 = next(t for t in task_deck if t['name'] == 'Task02_error')
+        self.assertEqual('Complete', task_02['status'])
+        dep_task = next(x for x in task_deck if task_02['id'] in x['parents'])
+        assert dep_task['name'] == 'Task11'
+        self.assertEqual('Complete', dep_task['status'], 'Failed to set dependent task from "Held" to "Waiting"')
 
         # check that logs were correctly overwritten
         check_logs = [t['log'].count(desired_logs) == 1 if t['log'] else True for t in task_deck]

release_notes.md

@@ -1,6 +1,7 @@
 ### Develop
 - Setting tasks to Waiting if they encountered lock (status -2)
 - Setting tasks to Incomplete if they return status -3
+- Completed tasks set held dependent tasks to waiting
 
 ## Release Notes 2.3
 ### Release Notes 2.3.1 2021-11-08