Merge pull request #78 from ftschirpke/fix-abstract-recipe-generation

fix: used task's id instead of name in abstract recipe generation

Author: rafaelfsilva

wfcommons/wfchef/wfchef_abstract_recipe.py

@@ -177,36 +177,36 @@ def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
                             makespan=0)
         graph = self.generate_nx_graph()
 
-        task_names = {}
+        task_ids = {}
         for node in graph.nodes:
             if node in ["SRC", "DST"]:
                 continue
             node_type = graph.nodes[node]["type"]
-            task_name = self._generate_task_name(node_type)
-            task = self._generate_task(node_type, task_name)
+            task_id = self._generate_task_name(node_type)
+            task = self._generate_task(node_type, task_id)
             workflow.add_task(task)
 
-            task_names[node] = task_name
+            task_ids[node] = task_id
 
         # tasks dependencies
         for (src, dst) in graph.edges:
             if src in ["SRC", "DST"] or dst in ["SRC", "DST"]:
                 continue
-            workflow.add_dependency(task_names[src], task_names[dst])
+            workflow.add_dependency(task_ids[src], task_ids[dst])
 
-            if task_names[src] not in self.tasks_children:
-                self.tasks_children[task_names[src]] = []
-            if task_names[dst] not in self.tasks_parents:
-                self.tasks_parents[task_names[dst]] = []
+            if task_ids[src] not in self.tasks_children:
+                self.tasks_children[task_ids[src]] = []
+            if task_ids[dst] not in self.tasks_parents:
+                self.tasks_parents[task_ids[dst]] = []
 
-            self.tasks_children[task_names[src]].append(task_names[dst])
-            self.tasks_parents[task_names[dst]].append(task_names[src])
+            self.tasks_children[task_ids[src]].append(task_ids[dst])
+            self.tasks_parents[task_ids[dst]].append(task_ids[src])
 
         # find leaf tasks
         leaf_tasks = []
         for node_name in workflow.nodes:
             task: Task = workflow.nodes[node_name]['task']
-            if task.name not in self.tasks_children:
+            if task.task_id not in self.tasks_children:
                 leaf_tasks.append(task)
 
         for task in leaf_tasks:

wfcommons/wfgen/abstract_recipe.py

@@ -196,8 +196,8 @@ def _generate_task_files(self, task: Task) -> List[File]:
         :return: List of files output files.
         :rtype: List[File]
         """
-        if task.name in self.tasks_output_files.keys():
-            return self.tasks_output_files[task.name]
+        if task.task_id in self.tasks_output_files.keys():
+            return self.tasks_output_files[task.task_id]
 
         task_recipe = self._workflow_recipe()[task.name]
 
@@ -208,19 +208,19 @@ def _generate_task_files(self, task: Task) -> List[File]:
         # obtain input files from parents
         input_files = []
 
-        if task.name in self.tasks_parents.keys():
-            for parent_task_name in self.tasks_parents[task.name]:
+        if task.task_id in self.tasks_parents.keys():
+            for parent_task_name in self.tasks_parents[task.task_id]:
                 output_files = self._generate_task_files(self.tasks_map[parent_task_name])
                 self.tasks_output_files.setdefault(parent_task_name, [])
                 self.tasks_output_files[parent_task_name] = output_files
                 input_files.extend(output_files)
 
         for input_file in input_files:
-            if input_file not in self.tasks_files_names[task.task_id]:
-                self.tasks_files[task.task_id].append(File(name=input_file,
+            if input_file.file_id not in self.tasks_files_names[task.task_id]:
+                self.tasks_files[task.task_id].append(File(file_id=input_file.file_id,
                                                            link=FileLink.INPUT,
                                                            size=input_file.size))
-                self.tasks_files_names[task.task_id].append(input_file)
+                self.tasks_files_names[task.task_id].append(input_file.file_id)
 
         # generate additional input files
         self._generate_files(task.task_id, task_recipe['input'], FileLink.INPUT)