Allow delay before job completeness checks in remote workflow.

law/contrib/arc/workflow.py

@@ -217,6 +217,9 @@ def arc_job_config(self, config, job_num, branches):
     def arc_check_job_completeness(self):
         return False
 
+    def arc_check_job_completeness_delay(self):
+        return 0.0
+
     def arc_use_local_scheduler(self):
         return True
 

law/contrib/glite/workflow.py

@@ -237,6 +237,9 @@ def glite_job_config(self, config, job_num, branches):
     def glite_check_job_completeness(self):
         return False
 
+    def glite_check_job_completeness_delay(self):
+        return 0.0
+
     def glite_use_local_scheduler(self):
         return True
 

law/contrib/htcondor/workflow.py

@@ -227,6 +227,9 @@ def htcondor_job_config(self, config, job_num, branches):
     def htcondor_check_job_completeness(self):
         return False
 
+    def htcondor_check_job_completeness_delay(self):
+        return 0.0
+
     def htcondor_use_local_scheduler(self):
         return False
 

law/contrib/lsf/workflow.py

@@ -222,6 +222,9 @@ def lsf_job_config(self, config, job_num, branches):
     def lsf_check_job_completeness(self):
         return False
 
+    def lsf_check_job_completeness_delay(self):
+        return 0.0
+
     def lsf_use_local_scheduler(self):
         return True
 

law/contrib/slurm/workflow.py

@@ -219,6 +219,9 @@ def slurm_job_config(self, config, job_num, branches):
     def slurm_check_job_completeness(self):
         return False
 
+    def slurm_check_job_completeness_delay(self):
+        return 0.0
+
     def slurm_use_local_scheduler(self):
         return False
 

law/workflow/remote.py

@@ -954,6 +954,12 @@ def poll(self):
             del unknown_jobs
             del states_by_id
 
+            # get settings from the task for triggering post-finished status checks
+            check_completeness = self._get_task_attribute("check_job_completeness")()
+            check_completeness_delay = self._get_task_attribute("check_job_completeness_delay")()
+            if check_completeness_delay:
+                time.sleep(check_completeness_delay)
+
             # store jobs per status and take further actions depending on the status
             pending_jobs = OrderedDict()
             running_jobs = OrderedDict()
@@ -974,7 +980,6 @@ def poll(self):
 
                 if data["status"] == self.job_manager.FINISHED:
                     # additionally check if the outputs really exist
-                    check_completeness = self._get_task_attribute("check_job_completeness")()
                     if not check_completeness or all(
                         self.task.as_branch(b).complete()
                         for b in sub_data["branches"]