Merge branch 'main' into main

rafaelfsilva · web-flow · commit b70ba0814540 · 2024-01-09T10:36:10.000-05:00
diff --git a/.readthedocs.yaml b/.readthedocs.yaml
@@ -0,0 +1,19 @@
+# Read the Docs configuration file for Sphinx projects
+# See https://docs.readthedocs.io/en/stable/config-file/v2.html for details
+
+# Required
+version: 2
+
+# Set the OS, Python version and other tools you might need
+build:
+  os: ubuntu-22.04
+  tools:
+    python: "3.11"
+
+# Build documentation in the "docs/" directory with Sphinx
+sphinx:
+  configuration: docs/source/conf.py
+
+python:
+   install:
+   - requirements: docs/requirements.txt
diff --git a/docs/requirements.txt b/docs/requirements.txt
@@ -1,3 +1,15 @@
 sphinx>=5.3.0
 sphinx_rtd_theme>=1.2.0
 recommonmark>=0.7.1
+jsonschema~=3.2.0
+matplotlib>=3.3.0
+networkx>=2.4
+numpy>=1.19.1
+python-dateutil>=2.8.1
+requests>=2.24.0
+scipy>=1.5.2
+setuptools>=49.3.1
+pyyaml>=5.3.1
+pandas>=1.2.4
+stringcase>=1.2.0
+filelock>=3.6.0
diff --git a/docs/source/dev_api_wfbench.rst b/docs/source/dev_api_wfbench.rst
@@ -13,6 +13,16 @@ wfcommons.wfbench.bench
    :show-inheritance:
    :private-members:
    :noindex:
+ 
+wfcommons.wfbench.translator.nextflow
+------------------------------------
+
+.. automodule:: wfcommons.wfbench.translator.nextflow
+   :members:
+   :undoc-members:
+   :show-inheritance:
+   :private-members:
+   :noindex:
 
 wfcommons.wfbench.translator.pegasus
 ------------------------------------
diff --git a/docs/source/dev_api_wfcommons.rst b/docs/source/dev_api_wfcommons.rst
@@ -1,5 +1,5 @@
 wfcommons.utils
------------------
+---------------
 
 .. automodule:: wfcommons.utils
    :members:
diff --git a/docs/source/generating_workflow_benchmarks.rst b/docs/source/generating_workflow_benchmarks.rst
@@ -67,6 +67,35 @@ description in :ref:`json-format-label`.
     to execute memory-intensive threads. Therefore, it is crucial to ensure that 
     :code:`stress-ng` is installed on all worker nodes.
 
+Nextflow
+++++++++
+`Nextflow <https://www.nextflow.io/>`_ is a workflow management system that enables
+the development of portable and reproducible workflows. It supports deploying workflows
+on a variety of execution platforms including local, HPC schedulers, and cloud-based
+and container-based environments. Below, we provide an example on how to generate
+workflow benchmark for running with Nextflow:
+
+    import pathlib
+
+    from wfcommons import BlastRecipe
+    from wfcommons.wfbench import WorkflowBenchmark, NextflowTranslator
+
+    # create a workflow benchmark object to generate specifications based on a recipe
+    benchmark = WorkflowBenchmark(recipe=BlastRecipe, num_tasks=500)
+
+    # generate a specification based on performance characteristics
+    benchmark.create_benchmark(pathlib.Path("/tmp/"), cpu_work=100, data=10, percent_cpu=0.6)
+
+    # generate a Nextflow workflow
+    translator = NextflowTranslator(benchmark.workflow)
+    translator.translate(output_file_name=pathlib.Path("/tmp/benchmark-workflow.nf"))
+
+.. warning::
+
+    Nextflow's way of defining workflows does not support tasks with iterations i.e. tasks 
+    that depend on another instance of the same abstract task. Thus, the translator
+    fails when you try to translate a workflow with iterations.
+
 Pegasus
 +++++++
 
diff --git a/docs/source/user_api_common.rst b/docs/source/user_api_common.rst
@@ -1,8 +1,8 @@
 wfcommons.common
-==================
+================
 
 wfcommons.common.file
------------------------
+---------------------
 
 .. automodule:: wfcommons.common.file
    :members:
@@ -11,7 +11,7 @@ wfcommons.common.file
    :private-members:
 
 wfcommons.common.task
------------------------
+---------------------
 
 .. automodule:: wfcommons.common.task
    :members:
@@ -20,7 +20,7 @@ wfcommons.common.task
    :private-members:
 
 wfcommons.common.machine
---------------------------
+------------------------
 
 .. automodule:: wfcommons.common.machine
    :members:
@@ -29,7 +29,7 @@ wfcommons.common.machine
    :private-members:
 
 wfcommons.common.workflow
----------------------------
+-------------------------
 
 .. automodule:: wfcommons.common.workflow
    :members:
diff --git a/docs/source/user_api_wfbench.rst b/docs/source/user_api_wfbench.rst
@@ -1,7 +1,7 @@
 .. _wfcommons-wfbench-label:
 
 wfcommons.wfbench
-================
+=================
 
 .. toctree::
    :maxdepth: 4
@@ -14,6 +14,13 @@ wfcommons.wfbench.bench
    :undoc-members:
    :show-inheritance:
 
+wfcommons.wfbench.translator.nextflow
+-------------------------------------
+
+.. automodule:: wfcommons.wfbench.translator.nextflow
+   :members:
+   :undoc-members:
+   :show-inheritance:
 
 wfcommons.wfbench.translator.pegasus
 ------------------------------------
diff --git a/wfcommons/common/task.py b/wfcommons/common/task.py
@@ -132,33 +132,33 @@ def as_dict(self) -> Dict:
             'children': [],
             'files': task_files,
         }
-        if self.runtime:
+        if self.runtime is not None:
             task_obj['runtimeInSeconds'] = self.runtime
-        if self.cores:
+        if self.cores is not None:
             task_obj['cores'] = self.cores
-        if self.task_id:
+        if self.task_id is not None:
             task_obj['id'] = self.task_id
-        if self.category:
+        if self.category is not None:
             task_obj['category'] = self.category
-        if self.avg_cpu:
+        if self.avg_cpu is not None:
             task_obj['avgCPU'] = self.avg_cpu
-        if self.bytes_read:
+        if self.bytes_read is not None:
             task_obj['readBytes'] = self.bytes_read
-        if self.bytes_written:
+        if self.bytes_written is not None:
             task_obj['writtenBytes'] = self.bytes_written
-        if self.memory:
+        if self.memory is not None:
             task_obj['memoryInBytes'] = self.memory
-        if self.energy:
+        if self.energy is not None:
             task_obj['energy'] = self.energy
-        if self.avg_power:
+        if self.avg_power is not None:
             task_obj['avgPower'] = self.avg_power
-        if self.priority:
+        if self.priority is not None:
             task_obj['priority'] = self.priority
-        if self.program:
+        if self.program is not None:
             task_obj['command']['program'] = self.program
-        if self.args:
+        if self.args is not None:
             task_obj['command']['arguments'] = self.args
-        if self.machine:
+        if self.machine is not None:
             task_obj['machine'] = self.machine.name
         if self.launch_dir:
             task_obj['launchDir'] = self.launch_dir
diff --git a/wfcommons/wfbench/bench.py b/wfcommons/wfbench/bench.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 #
-# Copyright (c) 2021-2022 The WfCommons Team.
+# Copyright (c) 2021-2023 The WfCommons Team.
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -79,7 +79,9 @@ def create_benchmark(self,
                          percent_cpu: Union[float, Dict[str, float]] = 0.6,
                          cpu_work: Union[int, Dict[str, int]] = None,
                          gpu_work: Union[int, Dict[str, int]] = None,
+                         time: Optional[int] = None,
                          data: Optional[Union[int, Dict[str, str]]] = None,
+                         mem: Optional[float] = None,
                          lock_files_folder: Optional[pathlib.Path] = None,
                          regenerate: Optional[bool] = True) -> pathlib.Path:
         """Create a workflow benchmark.
@@ -90,8 +92,12 @@ def create_benchmark(self,
         :type percent_cpu: Union[float, Dict[str, float]]
         :param cpu_work: CPU work per workflow task.
         :type cpu_work: Union[int, Dict[str, int]]
+        :param time: Time limit for running each task (in seconds).
+        :type time: Optional[int]
         :param data: Dictionary of input size files per workflow task type or total workflow data footprint (in MB).
         :type data: Optional[Union[int, Dict[str, str]]]
+        :param mem: Maximum amount of memory consumption per task (in MB).
+        :type mem: Optional[float]
         :param lock_files_folder:
         :type lock_files_folder: Optional[pathlib.Path]
         :param regenerate: Whether to regenerate the workflow tasks
@@ -150,9 +156,15 @@ def create_benchmark(self,
 
                 params.extend([f"--gpu-work {_gpu_work}"])
 
+            if mem:
+                params.extend([f"--mem {mem}"])
+            
+            if time:
+                params.extend([f"--time {time}"])
+
             task.runtime = 0
             task.files = []
-            task.program = f"{this_dir.joinpath('wfbench.py')}"
+            task.program = "wfbench.py"
             task.args = [task.name]
             task.args.extend(params)
 
diff --git a/wfcommons/wfbench/translator/__init__.py b/wfcommons/wfbench/translator/__init__.py
@@ -11,3 +11,4 @@
 from .dask import DaskTranslator
 from .pegasus import PegasusTranslator
 from .swift_t import SwiftTTranslator
+from .nextflow import NextflowTranslator
diff --git a/wfcommons/wfbench/translator/abstract_translator.py b/wfcommons/wfbench/translator/abstract_translator.py
@@ -1,23 +1,27 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 #
-# Copyright (c) 2021-2022 The WfCommons Team.
+# Copyright (c) 2021-2023 The WfCommons Team.
 #
 # This program is free software: you can redistribute it and/or modify
 # 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.
 
 import logging
+import os
 import pathlib
+import shutil
 
 from abc import ABC, abstractmethod
-from typing import Dict, List, Optional, Union
+from typing import Optional, Union
 
-from ...common import Task, Workflow
+from ...common import FileLink, Task, Workflow
 from ...wfinstances.instance import Instance
 
 
+this_dir = pathlib.Path(__file__).resolve().parent
+
 class Translator(ABC):
     """
     An abstract class of WfFormat parser for creating workflow benchmark applications.
@@ -69,13 +73,43 @@ def __init__(self,
                     self.task_children[task['name']].append(child)
 
     @abstractmethod
-    def translate(self, output_file_path: pathlib.Path) -> None:
+    def translate(self, output_folder: pathlib.Path) -> None:
         """
         Translate a workflow benchmark description (WfFormat) into an actual workflow application.
 
-        :param output_file_path: The path of the output file.
-        :type output_file_path: pathlib.Path
+        :param output_folder: The path to the folder in which the workflow benchmark will be generated.
+        :type output_folder: pathlib.Path
+        """
+
+    def _copy_binary_files(self, output_folder: pathlib.Path) -> None:
+        """
+        Copy binary files to workflow benchmark's bin folder.
+
+        :param output_folder: The path to the folder in which the workflow benchmark will be generated.
+        :type output_folder: pathlib.Path
+        """
+        bin_folder = output_folder.joinpath("bin")
+        bin_folder.mkdir()
+        shutil.copy(this_dir.joinpath("../wfbench.py"), bin_folder)
+        shutil.copy(shutil.which("cpu-benchmark"), bin_folder)
+
+    def _generate_input_files(self, output_folder: pathlib.Path) -> None:
+        """
+        Generate workflow input files into workflow benchmark's data folder.
+
+        :param output_folder: The path to the folder in which the workflow benchmark will be generated.
+        :type output_folder: pathlib.Path
         """
+        generated_files = []
+        data_folder = output_folder.joinpath("data")
+        data_folder.mkdir()
+        for task_name in self.root_task_names:
+            task = self.tasks[task_name]
+            for file in task.files:
+                if file.name not in generated_files and file.link == FileLink.INPUT:
+                    generated_files.append(file.name)
+                    with open(data_folder.joinpath(file.name), "wb") as fp:
+                        fp.write(os.urandom(int(file.size)))
 
     def _write_output_file(self, contents: str, output_file_path: pathlib.Path) -> None:
         """
@@ -91,32 +125,33 @@ def _write_output_file(self, contents: str, output_file_path: pathlib.Path) -> N
             out.write(contents)
         self.logger.info(f"Translated content written to '{output_file_path}'")
 
-    def _find_children(self, task_name: str) -> List[Task]:
+    def _find_children(self, task_name: str) -> list[Task]:
         """
         Find the children for a specific task.
 
         :param task_name: The task name.
         :type task_name: str
 
         :return: List of task's children.
-        :rtype: List[Task]
+        :rtype: list[Task]
         """
+        self.logger.debug(f"Finding children for task '{task_name}'")
         children = None
         for node in self.instance.instance["workflow"]["tasks"]:
             if node["name"] == task_name:
                 children = node["children"]
 
         return children
 
-    def _find_parents(self, task_name: str) -> List[Task]:
+    def _find_parents(self, task_name: str) -> list[Task]:
         """
         Find the parents for a specific task.
 
         :param task_name: The task name.
         :type task_name: str
 
         :return: List of task's parents.
-        :rtype: List[Task]
+        :rtype: list[Task]
         """
         self.logger.debug(f"Finding parents for task '{task_name}'")
         parents = None
diff --git a/wfcommons/wfbench/translator/dask.py b/wfcommons/wfbench/translator/dask.py
diff --git a/wfcommons/wfbench/translator/nextflow.py b/wfcommons/wfbench/translator/nextflow.py
diff --git a/wfcommons/wfbench/wfbench.py b/wfcommons/wfbench/wfbench.py
