Merge pull request #407 from int-brain-lab/task_statuses

Task statuses

Author: juhuntenburg

ibllib/pipes/tasks.py

@@ -72,9 +72,10 @@ def run(self, **kwargs):
         -   logging to variable
         -   writing a lock file if the GPU is used
         -   labels the status property of the object. The status value is labeled as:
-            0: Complete
+             0: Complete
             -1: Errored
             -2: Didn't run as a lock was encountered
+            -3: Incomplete
         """
         # if taskid of one properties are not available, local run only without alyx
         use_alyx = self.one is not None and self.taskid is not None
@@ -110,7 +111,11 @@ def run(self, **kwargs):
                     if not self._creates_lock():
                         self.status = -2
                         _logger.info(f"Job {self.__class__} exited as a lock was found")
-                        return
+                        new_log = log_capture_string.getvalue()
+                        self.log = new_log if self.clobber else self.log + new_log
+                        log_capture_string.close()
+                        _logger.removeHandler(ch)
+                        return self.status
                 self.outputs = self._run(**kwargs)
                 _logger.info(f"Job {self.__class__} complete")
         except BaseException:
@@ -215,6 +220,7 @@ def setUp(self, **kwargs):
     def tearDown(self):
         """
         Function after runs()
+        Does not run if a lock is encountered by the task (status -2)
         """
         if self.gpu >= 1:
             self._lock_file_path().unlink()
@@ -459,7 +465,7 @@ def rerun_failed(self, **kwargs):
         return self.run(status__in=['Waiting', 'Held', 'Started', 'Errored', 'Empty'], **kwargs)
 
     def rerun(self, **kwargs):
-        return self.run(status__in=['Waiting', 'Held', 'Started', 'Errored', 'Empty', 'Complete'],
+        return self.run(status__in=['Waiting', 'Held', 'Started', 'Errored', 'Empty', 'Complete', 'Incomplete'],
                         **kwargs)
 
     @property
@@ -527,6 +533,12 @@ def run_alyx_task(tdict=None, session_path=None, one=None, job_deck=None,
     # overwrite status to errored
     if status == -1:
         patch_data['status'] = 'Errored'
+    # Status -2 means a lock was encountered during run, should be rerun
+    if status == -2:
+        patch_data['status'] = 'Waiting'
+    # Status -3 should be returned if a task is Incomplete
+    if status == -3:
+        patch_data['status'] = 'Incomplete'
     # update task status on Alyx
     t = one.alyx.rest('tasks', 'partial_update', id=tdict['id'], data=patch_data)
     task.cleanUp()

ibllib/tests/test_tasks.py

@@ -24,7 +24,9 @@
     'Task01_void': 'Empty',
     'Task02_error': 'Errored',
     'Task10': 'Complete',
-    'Task11': 'Held'
+    'Task11': 'Held',
+    'TaskIncomplete': 'Incomplete',
+    'TaskGpuLock': 'Waiting'
 }
 
 desired_datasets = ['spikes.times.npy', 'spikes.amps.npy', 'spikes.clusters.npy']
@@ -37,7 +39,9 @@
     'Task01_void': 2,
     'Task02_error': 2,
     'Task10': 1,
-    'Task11': None
+    'Task11': None,
+    'TaskIncomplete': 1,
+    'TaskGpuLock': 2
 }
 
 
@@ -89,6 +93,27 @@ def _run(self, overwrite=False):
         return out_files
 
 
+#  Job that encounters a GPU lock and is set to Waiting
+class TaskGpuLock(ibllib.pipes.tasks.Task):
+    gpu = 1
+
+    # Overwrite setUp to create a lock file before running the task and remove it after
+    def setUp(self):
+        self.make_lock_file()
+        self.data_handler = self.get_data_handler()
+        return True
+
+    def _run(self, overwrite=False):
+        pass
+
+
+#  Job that encounters a GPU lock and is set to Waiting
+class TaskIncomplete(ibllib.pipes.tasks.Task):
+
+    def _run(self, overwrite=False):
+        self.status = -3
+
+
 class SomePipeline(ibllib.pipes.tasks.Pipeline):
 
     def __init__(self, session_path=None, **kwargs):
@@ -99,46 +124,14 @@ def __init__(self, session_path=None, **kwargs):
         tasks['Task00'] = Task00(self.session_path)
         tasks['Task01_void'] = Task01_void(self.session_path)
         tasks['Task02_error'] = Task02_error(self.session_path)
+        tasks['TaskGpuLock'] = TaskGpuLock(self.session_path)
+        tasks['TaskIncomplete'] = TaskIncomplete(self.session_path)
         tasks['Task10'] = Task10(self.session_path, parents=[tasks['Task00']])
         tasks['Task11'] = Task11(self.session_path, parents=[tasks['Task02_error'],
                                                              tasks['Task00']])
         self.tasks = tasks
 
 
-#  job to output a single file (pathlib.Path)
-class GpuTask(ibllib.pipes.tasks.Task):
-    gpu = 1
-
-    def _run(self, overwrite=False):
-        out_files = self.session_path.joinpath('alf', 'gpu.times.npy')
-        out_files.touch()
-        return out_files
-
-
-class TestLocks(unittest.TestCase):
-
-    def test_gpu_lock_and_local_data_handler(self) -> None:
-        with tempfile.TemporaryDirectory() as td:
-            session_path = Path(td).joinpath('algernon', '2021/02/12', '001')
-            session_path.joinpath('alf').mkdir(parents=True)
-            task = GpuTask(session_path, one=None, location='local')
-            assert task.is_locked() is False
-            task.run()
-            assert task.status == 0
-            assert task.is_locked() is False
-            # then make a lock file and make sure it fails and is still locked afterwards
-            task._make_lock_file()
-            task.run()
-            assert task.status == - 2
-            assert task.is_locked()
-            # test the time out feature
-            task.time_out_secs = - 1
-            task._make_lock_file()
-            assert not task.is_locked()
-            task.run()
-            assert task.status == 0
-
-
 class TestPipelineAlyx(unittest.TestCase):
 
     def setUp(self) -> None:
@@ -213,7 +206,7 @@ def test_pipeline_alyx(self):
         self.assertTrue(all(check_logs))
         self.assertTrue(all(check_rerun))
 
-        # Rerun without clobber and check that logs are overwritten
+        # Rerun without clobber and check that logs are not overwritten
         task_deck, dsets = pipeline.rerun_failed(machine='testmachine', clobber=False)
         check_logs = [t['log'].count(desired_logs) == desired_logs_rerun[t['name']] if t['log']
                       else t['log'] == desired_logs_rerun[t['name']] for t in task_deck]
@@ -222,6 +215,45 @@ def test_pipeline_alyx(self):
         self.assertTrue(all(check_logs))
         self.assertTrue(all(check_rerun))
 
+        # Remove the lock file
+        Path.home().joinpath('.one', 'gpu.lock').unlink()
+
+
+class GpuTask(ibllib.pipes.tasks.Task):
+    gpu = 1
+
+    def _run(self, overwrite=False):
+        out_files = self.session_path.joinpath('alf', 'gpu.times.npy')
+        out_files.touch()
+        return out_files
+
+
+class TestLocks(unittest.TestCase):
+
+    def test_gpu_lock_and_local_data_handler(self) -> None:
+        # Remove any existing locks first
+        if Path.home().joinpath('.one', 'gpu.lock').exists():
+            Path.home().joinpath('.one', 'gpu.lock').unlink()
+        with tempfile.TemporaryDirectory() as td:
+            session_path = Path(td).joinpath('algernon', '2021/02/12', '001')
+            session_path.joinpath('alf').mkdir(parents=True)
+            task = GpuTask(session_path, one=None, location='local')
+            assert task.is_locked() is False
+            task.run()
+            assert task.status == 0
+            assert task.is_locked() is False
+            # then make a lock file and make sure it fails and is still locked afterwards
+            task._make_lock_file()
+            task.run()
+            assert task.status == - 2
+            assert task.is_locked()
+            # test the time out feature
+            task.time_out_secs = - 1
+            task._make_lock_file()
+            assert not task.is_locked()
+            task.run()
+            assert task.status == 0
+
 
 if __name__ == "__main__":
     unittest.main(exit=False, verbosity=2)

release_notes.md

@@ -1,3 +1,7 @@
+### Develop
+- Setting tasks to Waiting if they encountered lock (status -2)
+- Setting tasks to Incomplete if they return status -3
+
 ## Release Notes 2.3
 ### Release Notes 2.3.1 2021-11-08
 - Trial wheel extraction: use alternative sync method when first on fails