Merge branch 'wfperf' of github.com:wfcommons/wfcommons into wfperf

Author: tainagdcoleman

README.md

@@ -75,13 +75,11 @@ When citing WfCommons, please use the following paper. You should also actually
 that paper, as it provides a recent and general overview on the framework.
 
 ```
-@inproceedings{ferreiradasilva2020works,
-  title = {WorkflowHub: Community Framework for Enabling Scientific Workflow Research and Development},
-  author = {Ferreira da Silva, Rafael and Pottier, Lo\"ic and Coleman, Tain\~a and Deelman, Ewa and Casanova, Henri},
-  booktitle = {2020 IEEE/ACM Workflows in Support of Large-Scale Science (WORKS)},
-  year = {2020},
-  pages = {49--56},
-  doi = {10.1109/WORKS51914.2020.00012}
+@article{coleman2021wfcommons,
+  title={WfCommons: A Framework for Enabling Scientific Workflow Research and Development},
+  author={Coleman, Tain{\~a} and Casanova, Henri and Pottier, Lo\"ic and Kaushik, Manav and Deelman, Ewa and Ferreira da Silva, Rafael},
+  journal={arXiv preprint arXiv:2105.14352},
+  year={2021}
 }
 ```
 

docs/source/generating_workflows_recipe.rst

@@ -79,10 +79,8 @@ package for an application in WfCommons: ::
 
 The snippet below shows an example of how to import the recipes: ::
 
-    # creating an Epigenomics workflow recipe
     from wfcommons.wfchef.recipes import EpigenomicsRecipe
 
-
 To check which recipes are installed in a system and how to import them use: ::
     
     $ wfchef ls

setup.py

@@ -22,7 +22,7 @@
     license='GPLv3',
     author='WfCommons team',
     author_email='support@wfcommons.org',
-    description='Community Framework for Enabling Scientific Workflow Research and Education',
+    description='A Framework for Enabling Scientific Workflow Research and Education',
     long_description=long_description,
     long_description_content_type='text/markdown',
     url='https://github.com/wfcommons/wfcommons',

wfcommons/common/task.py

@@ -35,9 +35,15 @@ class Task:
     :param runtime: Task runtime in seconds.
     :type runtime: float
     :param cores: Number of cores required by the task.
-    :type cores: int
+    :type cores: float
+    :param task_id: Job unique ID (e.g., ID0000001).
+    :type task_id: str
+    :param category: Job category (can be used, for example, to define jobs that use the same program).
+    :type category: str
     :param machine: Machine on which is the task has been executed.
     :type machine: Machine
+    :param program: Program name.
+    :type program: str
     :param args: List of task arguments.
     :type args: List[str]
     :param avg_cpu: Average CPU utilization in %.
@@ -64,8 +70,11 @@ def __init__(self,
                  name: str,
                  task_type: TaskType,
                  runtime: float,
-                 cores: int,
+                 cores: float = 1.0,
+                 task_id: Optional[str] = None,
+                 category: Optional[str] = None,
                  machine: Optional[Machine] = None,
+                 program: Optional[str] = None,
                  args: List[str] = [],
                  avg_cpu: Optional[float] = None,
                  bytes_read: Optional[int] = None,
@@ -82,7 +91,10 @@ def __init__(self,
         self.name: str = name
         self.type: TaskType = task_type
         self.runtime: float = runtime
-        self.cores: Optional[int] = cores
+        self.cores: Optional[float] = cores
+        self.task_id: Optional[str] = task_id
+        self.category: Optional[str] = category
+        self.program: Optional[str] = program
         self.args: List[str] = args
         self.avg_cpu: Optional[float] = avg_cpu
         self.bytes_read: Optional[int] = bytes_read
@@ -112,12 +124,17 @@ def as_dict(self) -> Dict:
             'name': self.name,
             'type': self.type.value,
             'runtime': self.runtime,
+            'command': {},
             'parents': [],
             'children': [],
             'files': task_files,
         }
         if self.cores:
             task_obj['cores'] = self.cores
+        if self.task_id:
+            task_obj['id'] = self.task_id
+        if self.category:
+            task_obj['category'] = self.category
         if self.avg_cpu:
             task_obj['avgCPU'] = self.avg_cpu
         if self.bytes_read:
@@ -132,8 +149,10 @@ def as_dict(self) -> Dict:
             task_obj['avgPower'] = self.avg_power
         if self.priority:
             task_obj['priority'] = self.priority
+        if self.program:
+            task_obj['command']['program'] = self.program
         if self.args:
-            task_obj['arguments'] = self.args
+            task_obj['command']['arguments'] = self.args
         if self.machine:
             task_obj['machine'] = self.machine.name
 

wfcommons/common/workflow.py

@@ -17,9 +17,10 @@
 from ..common.task import Task
 from ..version import __version__
 
-from ..wfchef.utils import annotate, create_graph
+from ..wfchef.utils import create_graph
 import tempfile
 
+
 class Workflow(nx.DiGraph):
     """
     Representation of a workflow. The workflow representation is an extension of the
@@ -53,7 +54,7 @@ def __init__(self,
         """Create an object of a workflow representation."""
         self.description = description if description else 'Instance generated with WfCommons - https://wfcommons.org'
         self.created_at = str(datetime.utcnow().isoformat())
-        self.schema_version = '1.0'
+        self.schema_version = '1.2'
         self.wms_name = 'WfCommons' if not wms_name else wms_name
         self.wms_version = str(__version__) if not wms_version else wms_version
         self.wms_url = 'https://docs.wfcommons.org/en/v{}/'.format(__version__) if not wms_url else wms_url
@@ -137,4 +138,4 @@ def write_dot(self, dot_filename: str = None) -> None:
     def to_nx_digraph(self) -> nx.DiGraph:
         with tempfile.NamedTemporaryFile() as temp:
             self.write_json(temp.name)
-            return create_graph(temp.name)
+            return create_graph(temp.name)

wfcommons/version.py

@@ -8,4 +8,4 @@
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 
-__version__ = "0.6-dev"
+__version__ = "0.7-dev"

wfcommons/wfchef/skeletons/__init__.py

@@ -7,5 +7,3 @@
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
-
-

wfcommons/wfgen/abstract_recipe.py

@@ -75,9 +75,13 @@ def __init__(self, name: str,
         self.input_file_size_factor = input_file_size_factor
         self.output_file_size_factor = output_file_size_factor
         self.workflows: List[Workflow] = []
+        self.tasks_map = {}
         self.tasks_files: Dict[str, List[File]] = {}
+        self.tasks_files_names: Dict[str, List[str]] = {}
         self.task_id_counter = 1
         self.this_dir = this_dir
+        self.tasks_children = {}
+        self.tasks_parents = {}
 
     @abstractmethod
     def _workflow_recipe(self) -> Dict[str, Any]:
@@ -142,7 +146,8 @@ def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
         :return: A synthetic workflow instance object.
         :rtype: Workflow
         """
-        workflow = Workflow(name=self.name + "-synthetic-instance" if not workflow_name else workflow_name, makespan=None)
+        workflow = Workflow(name=self.name + "-synthetic-instance" if not workflow_name else workflow_name,
+                            makespan=0)
         graph = self.generate_nx_graph()
 
         task_names = {}
@@ -156,11 +161,30 @@ def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
 
             task_names[node] = task_name
 
+        # tasks dependencies
         for (src, dst) in graph.edges:
             if src in ["SRC", "DST"] or dst in ["SRC", "DST"]:
                 continue
             workflow.add_edge(task_names[src], task_names[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]] = []
+
+            self.tasks_children[task_names[src]].append(task_names[dst])
+            self.tasks_parents[task_names[dst]].append(task_names[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:
+                leaf_tasks.append(task)
+
+        for task in leaf_tasks:
+            self._generate_task_files(task)
+
         workflow.nxgraph = graph
         self.workflows.append(workflow)
         return workflow
@@ -171,21 +195,13 @@ def _load_base_graph(self) -> nx.DiGraph:
     def _load_microstructures(self) -> Dict:
         return json.loads(self.this_dir.joinpath("microstructures.json").read_text())
 
-    def _generate_task(self, task_name: str,
-                       task_id: str,
-                       input_files: Optional[List[File]] = None,
-                       files_recipe: Optional[Dict[FileLink, Dict[str, int]]] = None
-                       ) -> Task:
+    def _generate_task(self, task_name: str, task_id: str) -> Task:
         """Generate a synthetic task.
 
         :param task_name: task name.
         :type task_name: str
         :param task_id: task ID.
         :type task_id: str
-        :param input_files: List of input files to be included.
-        :type input_files: List[File]
-        :param files_recipe: Recipe for generating task files.
-        :type files_recipe: Dict[FileLink, Dict[str, int]]
 
         :return: A task object.
         :rtype: task
@@ -197,22 +213,17 @@ def _generate_task(self, task_name: str,
                                                task_recipe['runtime']['min'],
                                                task_recipe['runtime']['max']), '.3f'))
 
-        # linking previous generated output files as input files
+        # # linking previous generated output files as input files
         self.tasks_files[task_id] = []
-        if input_files:
-            for f in input_files:
-                if f.link == FileLink.OUTPUT:
-                    self.tasks_files[task_id].append(File(name=f.name, size=f.size, link=FileLink.INPUT))
-
-        # generate additional in/output files
-        self._generate_files(task_id, task_recipe['input'], FileLink.INPUT, files_recipe)
-        self._generate_files(task_id, task_recipe['output'], FileLink.OUTPUT, files_recipe)
-
-        return Task(
+        self.tasks_files_names[task_id] = []
+        task = Task(
             name=task_id,
+            task_id='0{}'.format(task_id.split('_0')[1]),
+            category=task_name,
             task_type=TaskType.COMPUTE,
             runtime=runtime,
             machine=None,
+            program=task_name,
             args=[],
             cores=1,
             avg_cpu=None,
@@ -222,8 +233,11 @@ def _generate_task(self, task_name: str,
             energy=None,
             avg_power=None,
             priority=None,
-            files=self.tasks_files[task_id]
+            files=[]
         )
+        self.tasks_map[task_id] = task
+
+        return task
 
     def _generate_task_name(self, prefix: str) -> str:
         """Generate a task name from a prefix appended with an ID.
@@ -238,8 +252,40 @@ def _generate_task_name(self, prefix: str) -> str:
         self.task_id_counter += 1
         return task_name
 
-    def _generate_files(self, task_id: str, recipe: Dict[str, Any], link: FileLink,
-                        files_recipe: Optional[Dict[FileLink, Dict[str, int]]] = None) -> None:
+    def _generate_task_files(self, task: Task) -> List[File]:
+        """Generate input and output files for a task.
+
+        :param task: task object.
+        :type task: Task
+
+        :return: List of files output files.
+        :rtype: List[File]
+        """
+        task_recipe = self._workflow_recipe()[task.category]
+
+        # generate output files
+        output_files_list = self._generate_files(task.name, task_recipe['output'], FileLink.OUTPUT)
+        task.files = self.tasks_files[task.name]
+
+        # 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]:
+                input_files.extend(self._generate_task_files(self.tasks_map[parent_task_name]))
+
+        for input_file in input_files:
+            if input_file.name not in self.tasks_files_names[task.name]:
+                self.tasks_files[task.name].append(File(name=input_file.name,
+                                                        link=FileLink.INPUT,
+                                                        size=input_file.size))
+                self.tasks_files_names[task.name].append(input_file.name)
+
+        # generate additional input files
+        self._generate_files(task.name, task_recipe['input'], FileLink.INPUT)
+
+        return output_files_list
+
+    def _generate_files(self, task_id: str, recipe: Dict[str, Any], link: FileLink) -> List[File]:
         """Generate files for a specific task ID.
 
         :param task_id: task ID.
@@ -248,21 +294,27 @@ def _generate_files(self, task_id: str, recipe: Dict[str, Any], link: FileLink,
         :type recipe: Dict[str, Any]
         :param link: Type of file link.
         :type link: FileLink
-        :param files_recipe: Recipe for generating task files.
-        :type files_recipe: Dict[FileLink, Dict[str, int]]
+
+        :return: List of files.
+        :rtype: List[File]
         """
+        files_list = []
         extension_list: List[str] = []
         for f in self.tasks_files[task_id]:
             if f.link == link:
+                files_list.append(f)
                 extension_list.append(path.splitext(f.name)[1] if '.' in f.name else f.name)
 
         for extension in recipe:
             if extension not in extension_list:
                 num_files = 1
-                if files_recipe and link in files_recipe and extension in files_recipe[link]:
-                    num_files = files_recipe[link][extension]
                 for _ in range(0, num_files):
-                    self.tasks_files[task_id].append(self._generate_file(extension, recipe, link))
+                    file = self._generate_file(extension, recipe, link)
+                    files_list.append(file)
+                    self.tasks_files[task_id].append(file)
+                    self.tasks_files_names[task_id].append(file.name)
+
+        return files_list
 
     def _generate_file(self, extension: str, recipe: Dict[str, Any], link: FileLink) -> File:
         """Generate a file according to a file recipe.

wfcommons/wfinstances/instance.py

@@ -117,14 +117,20 @@ def __init__(self, input_instance: str, schema_file: Optional[str] = None, logge
             machine = task.get('machine', None)
             machine = None if machine is None else self.machines[machine]
 
+            # Fetch the command associated to this task
+            command = task.get('command', None)
+
             self.workflow.add_node(
                 task['name'],
                 task=Task(
                     name=task['name'],
+                    task_id=task.get('id', None),
+                    category=task.get('category', None),
                     task_type=TaskType(task['type']),
                     runtime=task['runtime'],
                     machine=machine,
-                    args=task.get('arguments', None),
+                    program=command.get('program', None) if command else None,
+                    args=command.get('arguments', None) if command else None,
                     cores=task.get('cores', None),
                     avg_cpu=task.get('avgCPU', None),
                     bytes_read=task.get('bytesRead', None),