Merge pull request #286 from HSF/dev

add is_processed to workflow
HSF · Mar 14, 2024 · bdcc815 · bdcc815
2 parents 9720992 + c16278a
commit bdcc815
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 0 deletions.
diff --git a/workflow/lib/idds/workflow/workflow.py b/workflow/lib/idds/workflow/workflow.py
@@ -1825,6 +1825,12 @@ def is_subfinished(self, synchronize=True):
         """
         return self.is_terminated() and (self.num_finished_works + self.num_subfinished_works > 0 and self.num_finished_works + self.num_subfinished_works <= self.num_total_works)
 
+    def is_processed(self, synchronize=True):
+        """
+        *** Function called by Transformer agent.
+        """
+        return self.is_terminated(synchronize=synchronize) and (self.num_finished_works + self.num_subfinished_works > 0 and self.num_finished_works + self.num_subfinished_works <= self.num_total_works)
+
     def is_failed(self, synchronize=True):
         """
         *** Function called by Marshaller agent.
@@ -2218,6 +2224,11 @@ def is_subfinished(self, synchronize=True):
             return self.runs[str(self.num_run)].is_subfinished()
         return False
 
+    def is_processed(self, synchronize=True):
+        if self.is_terminated():
+            return self.runs[str(self.num_run)].is_processed()
+        return False
+
     def is_failed(self, synchronize=True):
         if self.is_terminated():
             return self.runs[str(self.num_run)].is_failed()

diff --git a/workflow/lib/idds/workflowv2/workflow.py b/workflow/lib/idds/workflowv2/workflow.py
@@ -1972,6 +1972,12 @@ def is_subfinished(self, synchronize=True):
         """
         return self.is_terminated(synchronize=synchronize) and (self.num_finished_works + self.num_subfinished_works > 0 and self.num_finished_works + self.num_subfinished_works <= self.num_total_works)
 
+    def is_processed(self, synchronize=True):
+        """
+        *** Function called by Transformer agent.
+        """
+        return self.is_terminated(synchronize=synchronize) and (self.num_finished_works + self.num_subfinished_works > 0 and self.num_finished_works + self.num_subfinished_works <= self.num_total_works)
+
     def is_failed(self, synchronize=True):
         """
         *** Function called by Marshaller agent.
@@ -2500,6 +2506,20 @@ def is_subfinished(self, synchronize=True):
             return self.runs[str(self.num_run)].is_subfinished(synchronize=False)
         return False
 
+    def is_processed(self, synchronize=True):
+        if self.is_terminated(synchronize=synchronize):
+            build_work = self.get_build_work()
+            if build_work:
+                if build_work.is_terminated():
+                    if not build_work.is_processed():
+                        return False
+                    else:
+                        pass
+                else:
+                    return False
+            return self.runs[str(self.num_run)].is_processed(synchronize=False)
+        return False
+
     def is_failed(self, synchronize=True):
         if self.is_terminated(synchronize=synchronize):
             build_work = self.get_build_work()