Merge pull request #77 from wfcommons/flowcept_improvements

Swift/T with Flowcept

Author: rafaelfsilva

wfcommons/wfbench/translator/swift_t.py

@@ -217,7 +217,8 @@ def _add_tasks(self, category: str) -> None:
         if num_tasks > 1:
             self.script += f"foreach i in [0:{num_tasks - 1}] {{\n" \
                 f"  string of = sprintf(\"{self.output_folder.absolute()}/data/{category}_%i_output.txt\", i);\n" \
-                f"  string cmd_{self.cmd_counter} = sprintf(command, \"{self.cpu_benchmark}\", \"{category}\", {args});\n" \
+                f"  string task_id = \"{category}_\" + i;\n" \
+                f"  string cmd_{self.cmd_counter} = sprintf(command, \"{self.cpu_benchmark}\", task_id, {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" \
                 f"  {category}__out[i] = string2int(of_{self.cmd_counter});\n" \
@@ -231,7 +232,7 @@ def _add_tasks(self, category: str) -> None:
                 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" \
+            self.script += f"string cmd_{self.cmd_counter} = sprintf(command, \"{self.cpu_benchmark}\", \"{category}_{self.cmd_counter}\", {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" \
                 f"{category}__out[0] = string2int(of_{self.cmd_counter});\n\n"

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

@@ -31,7 +31,7 @@ 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)
+flowcept_agent = Flowcept(workflow_id=workflow_id, workflow_name=workflow_name, bundle_exec_id=workflow_id, start_persistence=False, save_workflow=True)
 
 try:
     flowcept_agent.start()
@@ -51,10 +51,8 @@ while remaining_files:
         break
     time.sleep(1)
     
-time.sleep(180)
 try:
     flowcept_agent.stop()
-    time.sleep(120)
 except Exception:
     import traceback
     traceback.print_exc()
@@ -90,6 +88,7 @@ cpu_threads = int(10 * percent_cpu)
 output_data = {"%s": int(%i)}
 dep = %i
 workflow_id = "%s"
+task_id = f"{workflow_id}_{task_name}"
 
 if 'workflow_id':
     logging.info("Running with Flowcept.")
@@ -98,7 +97,7 @@ if 'workflow_id':
                 bundle_exec_id=workflow_id,
                 start_persistence=False, save_workflow=False)
     fc.start()
-    fc_task = FlowceptTask(workflow_id=workflow_id, used={
+    fc_task = FlowceptTask(workflow_id=workflow_id, task_id=task_id, used={
       'workflow_id': workflow_id,
       'name': task_name,
       'percent-cpu': percent_cpu,