Merge pull request #73 from wfcommons/flowcept_improvements

Enabling Swift/T support for Flowcept

Author: rafaelfsilva

wfcommons/wfbench/translator/swift_t.py

@@ -39,18 +39,17 @@ def __init__(self,
         self.stress_path = stress_path
         self.categories_list = []
         self.categories_input = {}
-        self.parsed_tasks = []
+        self.parsed_tasks = set()
         self.files_map = {}
         self.tasks_map = {}
         self.cmd_counter = 1
+        self.out_files = set()
 
         # find applications
-        self.apps = []
+        self.apps = set()
         for task in self.tasks.values():
             self.tasks_map[task.task_id] = task.name
-
-            if task.name not in self.apps:
-                self.apps.append(task.name)
+            self.apps.add(task.name)
 
             out_count = 0
             for file in task.output_files:
@@ -76,7 +75,7 @@ def translate(self, output_folder: pathlib.Path) -> None:
         in_count = 0
         self.output_folder = output_folder
         self.cpu_benchmark = output_folder.joinpath("./bin/cpu-benchmark").absolute()
-        self.script = f"string fs = sprintf(flowcept_start, \"{self.workflow.workflow_id}\", \"{self.workflow.name}\");\nstring fss = python_persist(fs);\n\n" if self.workflow.workflow_id else ""
+        self.script = f"string fs = sprintf(flowcept_start, \"{self.workflow.workflow_id}\");\nstring fss = python_persist(fs);\n\n" if self.workflow.workflow_id else ""
         self.script += "string root_in_files[];\n"
 
         for task_name in self.root_task_names:
@@ -100,9 +99,11 @@ def translate(self, output_folder: pathlib.Path) -> None:
         for category in self.categories_list:
             self._add_tasks(category)
 
-        # flowcept stop
-        # if self.workflow.workflow_id:
-        #     self.script += "string fss = sprintf(flowcept_stop);\npython_persist(fss);"
+        # flowcept end
+        if self.workflow.workflow_id:
+            out_files = ", ".join(f"'{item}'" for item in self.out_files)
+            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)
 
@@ -134,7 +135,7 @@ def _find_categories_list(self, task_name: str, parent_task: Optional[str] = Non
             if parent_task.name not in self.categories_list:
                 return
 
-        self.parsed_tasks.append(task_name)
+        self.parsed_tasks.add(task_name)
         category = self.tasks_map[task_name]
         if category not in self.categories_list:
             self.categories_list.append(category)
@@ -153,29 +154,28 @@ def _add_tasks(self, category: str) -> None:
         :type category: str
         """
         num_tasks = 0
+        num_children = 0
         input_files_cat = {}
-        parsed_input_files = []
+        parsed_input_files = set()
         self.script += f"int {category}__out[];\n"
 
-        for task_name in self.tasks:
-            task = self.tasks[task_name]
-
+        for task in self.tasks.values():
             if task.name == category:
                 # in/output files
-                input_files = []
+                num_children = len(self.task_children[task.task_id])
+                input_files = set()
                 prefix = ""
 
                 for file in task.output_files:
-                    out_file = file.file_id
                     file_size = file.size
 
                 for file in task.input_files:
                     cat_prefix = self.files_map[file.file_id].split("__out")[0]
                     if file.file_id not in parsed_input_files:
                         input_files_cat.setdefault(cat_prefix, 0)
                         input_files_cat[cat_prefix] += 1
-                        parsed_input_files.append(file.file_id)
-                    input_files.append(self.files_map[file.file_id])
+                        parsed_input_files.add(file.file_id)
+                    input_files.add(self.files_map[file.file_id])
                     if not prefix:
                         prefix = cat_prefix
 
@@ -222,10 +222,15 @@ 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"
-            
+            if not num_children:
+                for i in range(num_tasks):
+                    self.out_files.add(f"{self.output_folder.absolute()}/data/{category}_{i}_output.txt")
         else:
+            out_file = f"{self.output_folder.absolute()}/data/{category}_0_output.txt"
+            if not num_children:
+                self.out_files.add(out_file)
             args = args.replace(
-                ", of", f", \"{self.output_folder.absolute()}/data/{category}_0_output.txt\"").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

@@ -4,17 +4,62 @@ import python;
 import string;
 import unix;
 
-global const string flowcept_start = 
+global const string flowcept_start =
 """
+import time
 workflow_id = "%s"
+time.sleep(30)
+""";
+
+global const string flowcept =
+"""
+import logging
+import pathlib
+import subprocess
+import time
 from flowcept.flowcept_api.flowcept_controller import Flowcept
-flowcept_agent = Flowcept(workflow_id=workflow_id, workflow_name="%s", bundle_exec_id=workflow_id)
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="[WfBench][%%(asctime)s][%%(levelname)s] %%(message)s",
+    datefmt="%%H:%%M:%%S",
+    handlers=[logging.StreamHandler()]
+)
+
+workflow_id = "%s"  
+workflow_name = "%s"
+out_files = [%s]
+
+logging.info("Flowcept Starting")
+flowcept_agent = Flowcept(workflow_id=workflow_id, workflow_name=workflow_name, bundle_exec_id=workflow_id)
 
 try:
     flowcept_agent.start()
 except Exception:
     import traceback
     traceback.print_exc()
+
+remaining_files = set(out_files)
+
+while remaining_files:
+    found_files = set()
+    for f in remaining_files:
+        if pathlib.Path(f).exists():
+            found_files.add(f)
+    remaining_files -= found_files
+    if not remaining_files:
+        break
+    time.sleep(1)
+    
+time.sleep(180)
+try:
+    flowcept_agent.stop()
+    time.sleep(120)
+except Exception:
+    import traceback
+    traceback.print_exc()
+
+logging.info("Flowcept Completed")
 """;
 
 string command = 
@@ -188,7 +233,6 @@ logging.info(f"Benchmark {task_name} completed!")
 if 'workflow_id':
     fc_task.end()
     fc.stop()
-    time.sleep(1)
 """;
 
 # Generated code goes here