swift/t with flowcept

Author: rafaelfsilva

wfcommons/wfbench/translator/swift_t.py

@@ -43,7 +43,7 @@ def __init__(self,
         self.files_map = {}
         self.tasks_map = {}
         self.cmd_counter = 1
-        self.last_file = ""
+        self.out_files = []
 
         # find applications
         self.apps = []
@@ -103,7 +103,9 @@ def translate(self, output_folder: pathlib.Path) -> None:
 
         # flowcept end
         if self.workflow.workflow_id:
-            self.script += f"string fc = sprintf(flowcept, \"{self.workflow.workflow_id}\", \"{self.workflow.name}\", \"{self.last_file}\");\n" \
+            out_files = ", ".join(f'"{item}"' for item in self.out_files)
+            self.script += f"string out_files[] = [{out_files}];\n"
+            self.script += f"string fc = sprintf(flowcept, \"{self.workflow.workflow_id}\", \"{self.workflow.name}\", out_files);\n" \
                             "python_persist(fc);\n"
 
         run_workflow_code = self._merge_codelines("templates/swift_t_templates/workflow.swift", self.script)
@@ -223,11 +225,12 @@ def _add_tasks(self, category: str) -> None:
                 f"  string of_{self.cmd_counter} = sprintf(\"0%s\", co_{self.cmd_counter});\n" \
                 f"  {category}__out[i] = string2int(of_{self.cmd_counter});\n" \
                 "}\n\n"
-            self.last_file = f"{self.output_folder.absolute()}/data/{category}_{num_tasks - 1}_output.txt"
+            self.out_files.append(f"{self.output_folder.absolute()}/data/{category}_{num_tasks - 1}_output.txt")
         else:
-            self.last_file = f"{self.output_folder.absolute()}/data/{category}_0_output.txt"
+            out_file = f"{self.output_folder.absolute()}/data/{category}_0_output.txt"
+            self.out_files.append(out_file)
             args = args.replace(
-                ", of", f", \"{self.last_file}\"").replace("[i]", "[0]")
+                ", of", f", \"{out_file}\"").replace("[i]", "[0]")
             self.script += f"string cmd_{self.cmd_counter} = sprintf(command, \"{self.cpu_benchmark}\", \"{category}\", {args});\n" \
                 f"string co_{self.cmd_counter} = python_persist(cmd_{self.cmd_counter});\n" \
                 f"string of_{self.cmd_counter} = sprintf(\"0%s\", co_{self.cmd_counter});\n" \

wfcommons/wfbench/translator/templates/swift_t_templates/workflow.swift

@@ -14,10 +14,10 @@ time.sleep(30)
 global const string flowcept =
 """
 import logging
-import os.path
 import subprocess
 import time
 from flowcept.flowcept_api.flowcept_controller import Flowcept
+from pathlib import Path
 
 logging.basicConfig(
     level=logging.INFO,
@@ -28,7 +28,7 @@ logging.basicConfig(
 
 workflow_id = "%s"
 workflow_name = "%s"
-file_path = "%s"
+out_files = "%A"
 
 logging.info("Flowcept Starting")
 flowcept_agent = Flowcept(workflow_id=workflow_id, workflow_name=workflow_name, bundle_exec_id=workflow_id)
@@ -39,18 +39,22 @@ except Exception:
     import traceback
     traceback.print_exc()
 
-while not os.path.exists(file_path):
+remaining_files = set(out_files)
+
+while remaining_files:
+    found_files = {f for f in remaining_files if Path(f).exists()}
+    if found_files:
+        remaining_files -= found_files
+    if not remaining_files:
+        break
     time.sleep(1)
 
-if os.path.isfile(file_path):
-    time.sleep(240)
-    try:
-        flowcept_agent.stop()
-    except Exception:
-        import traceback
-        traceback.print_exc()
-else:
-    raise ValueError(f"{file_path} is not a file!")
+time.sleep(240)
+try:
+    flowcept_agent.stop()
+except Exception:
+    import traceback
+    traceback.print_exc()
 
 logging.info("Flowcept Completed")
 """;