swift/t with flowcept

Author: rafaelfsilva

wfcommons/wfbench/translator/swift_t.py

@@ -39,19 +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 = []
+        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:
@@ -137,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)
@@ -156,16 +154,16 @@ 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:
@@ -176,8 +174,8 @@ def _add_tasks(self, category: str) -> None:
                     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
 
@@ -224,11 +222,13 @@ 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"
-            for i in range(num_tasks):
-                self.out_files.append(f"{self.output_folder.absolute()}/data/{category}_{i}_output.txt")
+            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"
-            self.out_files.append(out_file)
+            if not num_children:
+                self.out_files.add(out_file)
             args = args.replace(
                 ", of", f", \"{out_file}\"").replace("[i]", "[0]")
             self.script += f"string cmd_{self.cmd_counter} = sprintf(command, \"{self.cpu_benchmark}\", \"{category}\", {args});\n" \

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

@@ -51,10 +51,10 @@ while remaining_files:
         break
     time.sleep(1)
     
-time.sleep(120)
+time.sleep(180)
 try:
     flowcept_agent.stop()
-    time.sleep(60)
+    time.sleep(120)
 except Exception:
     import traceback
     traceback.print_exc()