wfcommons.wfbench¶

wfcommons.wfbench.bench¶

class wfcommons.wfbench.bench.WorkflowBenchmark(recipe: Type[WfChefWorkflowRecipe], num_tasks: int, with_flowcept: bool = False, logger: Logger | None = None)[source]

Bases: object

Generate a workflow benchmark instance based on a workflow recipe (WfChefWorkflowRecipe)

Parameters:

recipe (Type[WfChefWorkflowRecipe]) – A workflow recipe.
num_tasks (int) – Total number of tasks in the benchmark workflow.
with_flowcept (bool)
logger (Optional[Logger]) – The logger where to log information/warning or errors.

create_benchmark_from_input_file(save_dir: Path, input_file: Path, lock_files_folder: Path | None = None, rundir: Path | None = None) → Path[source]

Create a workflow benchmark.

Parameters:

save_dir (pathlib.Path) – Folder to generate the workflow benchmark JSON instance and input data files.
input_file (pathlib.Path)
lock_files_folder (Optional[pathlib.Path])
rundir (Optional[pathlib.Path]) – If you would like for the files to be created/saved in a different directory.

Returns:

The path to the workflow benchmark JSON instance.

Return type:

pathlib.Path

Create a workflow benchmark from a synthetic workflow

Parameters:

save_dir (pathlib.Path) – Folder to generate the workflow benchmark JSON instance and input data files.
workflow (Workflow) – The (synthetic) workflow to use as a benchmark.
percent_cpu (Union[float, Dict[str, float]]) – The maximum percentage of CPU threads.
cpu_work (Union[int, Dict[str, int]]) – Maximum CPU work per workflow task.
gpu_work (Union[int, Dict[str, int]]) – Maximum GPU work per workflow task.
num_chunks (Optional[int]) – Number of chunks for pipelining I/O and computation for each task execution.
time (Optional[int]) – Time limit for running each task (in seconds).
mem (Optional[float]) – Maximum amount of memory consumption per task (in MB).
lock_files_folder (Optional[pathlib.Path])
rundir (Optional[pathlib.Path]) – If you would like for the files to be created/saved in a different directory.

Returns:

The path to the workflow benchmark JSON instance.

Return type:

pathlib.Path

_rename_files_to_wfbench_format() → List[File][source]

Rename the files in the workflow to the wfbench format.

Returns:: A list of the input files that need to be generated (with their new names).
Return type:: List[File]

Create a workflow benchmark.

Parameters:

save_dir (pathlib.Path) – Folder to generate the workflow benchmark JSON instance and input data files.
percent_cpu (Union[float, Dict[str, float]]) – The percentage of CPU threads.
cpu_work (Union[int, Dict[str, int]]) – CPU work per workflow task.
gpu_work (Union[int, Dict[str, int]]) – GPU work per workflow task.
num_chunks (Optional[int]) – Number of chunks for pipelining I/O and computation for each task execution.
time (Optional[int]) – Time limit for running each task (in seconds).
data (Optional[Union[int, Dict[str, str]]]) – Total workflow data footprint (in MB).
mem (Optional[float]) – Maximum amount of memory consumption per task (in MB).
lock_files_folder (Optional[pathlib.Path])
regenerate (Optional[bool]) – Whether to regenerate the workflow tasks
rundir (Optional[pathlib.Path]) – If you would like for the files to be created/saved in a different directory.

Returns:

The path to the workflow benchmark JSON instance.

Return type:

pathlib.Path

_creating_lock_files(lock_files_folder: Path | None) → Tuple[Path | None, Path | None][source]: Creating the lock files

_set_argument_parameters(task: Task, percent_cpu: float | Dict[str, float], cpu_work: int | Dict[str, int], gpu_work: int | Dict[str, int], time: int | None, num_chunks: int | None, mem: float | None, lock_files_folder: Path | None, cores: Path | None, lock: Path | None, rundir: Path | None) → None[source]: Setting the parameters for the arguments section of the JSON

_generate_task_cpu_params(task: Task, percent_cpu: float | Dict[str, float], cpu_work: int | Dict[str, int], lock_files_folder: Path | None, cores: Path | None, lock: Path | None) → List[str][source]: Setting cpu arguments if cpu benchmark requested

_generate_task_gpu_params(task: Task, gpu_work: int | Dict[str, int]) → List[str][source]: Setting gpu arguments if gpu benchmark requested

_create_data_footprint(data: int) → None[source]: task’s data footprint provided as an int

_output_files(data: Dict[str, str]) → Dict[str, Dict[str, int]][source]

Calculate, for each task, total number of output files needed. This method is used when the user is specifying the input file sizes.

Parameters:: data (Dict[str, str])
Returns:
Return type:: Dict[str, Dict[str, int]]

_calculate_input_files()[source]: Calculate total number of files needed. This method is used if the user provides total data footprint.

_add_output_files(output_file_size: int) → None[source]

Add output files when input data was offered by the user.

Parameters:: output_file_size (int) – file size in MB

_add_input_files(input_file_size: int) → None[source]

Add input files when input data was offered by the user.

Parameters:: input_file_size (int) – a file size in MB

generate_input_file(path: Path) → None[source]

Generates input file where customization of cpu percentage, cpu work, gpu work, data size

Parameters:: path (pathlib.Path)

wfcommons.wfbench.bench.generate_sys_data(num_files: int, tasks: Dict[str, int], save_dir: Path) → List[str][source]

Generate workflow’s input data

Parameters:

num_files (int) – number of each file to be generated.
tasks (Dict[str, int]) – Dictionary with the name of the tasks and their data sizes.
save_dir (pathlib.Path) – Folder to generate the workflow benchmark’s input data files.

wfcommons.wfbench.bench.cleanup_sys_files() → None[source]: Remove files already used

wfcommons.wfbench.bench.clean_entry(entry)[source]

wfcommons.wfbench.translator.airflow¶

class wfcommons.wfbench.translator.airflow.AirflowTranslator(workflow: Workflow | Path, logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating Airflow workflow applications.

Parameters:

workflow (Union[Workflow, pathlib.Path],) – Workflow benchmark object or path to the workflow benchmark JSON instance.
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path, name: str | None = None) → None[source]

Translate a workflow benchmark description(WfFormat) into an Airflow workflow application.

Parameters:

output_folder (pathlib.Path) – The name of the output folder.
name (str) – The name of the workflow in the DAG

_sanitize_varname(name: str) → str[source]

Sanitizes string into a valid variable name.

Parameters:: name (str) – The name to sanitize.

_prep_commands(output_folder: Path) → None[source]

Prepares the bash_command strings for the BashOperators.

Parameters:: output_folder (pathlib.Path) – The name of the output folder.

_write_readme_file(output_folder: Path) → None[source]

Write the README file.

Parameters:: output_folder (pathlib.Path) – The path of the output folder.

_abc_impl = <_abc._abc_data object>

wfcommons.wfbench.translator.bash¶

class wfcommons.wfbench.translator.bash.BashTranslator(workflow: Workflow | Path, logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating a sequential bash workflow benchmarks.

Parameters:

workflow (Union[Workflow, pathlib.Path],) – Workflow benchmark object or path to the workflow benchmark JSON instance.
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path) → None[source]

Translate a workflow benchmark description (WfFormat) into an actual workflow application.

Parameters:: output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.

_bash_wftasks_codelines() → list[str][source]

_abc_impl = <_abc._abc_data object>

wfcommons.wfbench.translator.cwl¶

class wfcommons.wfbench.translator.cwl.CWLTranslator(workflow: Workflow | Path, generate_stdout_files: bool | None = True, generate_stderr_files: bool | None = True, logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating CWL workflow benchmarks.

Parameters:

workflow (Union[Workflow, pathlib.Path],) – Workflow benchmark object or path to the workflow benchmark JSON instance.
generate_stdout_files (Optional[bool]) – If true, each step will generate a .out file with stdout from the step’s execution
generate_stderr_files (Optional[bool]) – If true, each step will generate a .err file with stderr from the step’s execution
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path) → None[source]

Translate a workflow benchmark description (WfFormat) into an actual workflow application.

Parameters:: output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.

_parse_steps() → None[source]

_parse_inputs_outputs() → None[source]

_write_cwl_files(output_folder: Path) → None[source]

_write_readme_file(output_folder: Path) → None[source]

_abc_impl = <_abc._abc_data object>

wfcommons.wfbench.translator.dask¶

class wfcommons.wfbench.translator.dask.DaskTranslator(workflow: Workflow | Path, logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating Dask workflow applications.

Parameters:

workflow (Union[Workflow, pathlib.Path],) – Workflow benchmark object or path to the workflow benchmark JSON instance.
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path) → None[source]

Translate a workflow benchmark description (WfFormat) into an actual workflow application.

Parameters:: output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.

_dask_wftasks_codelines(randomizer_varname: str, output_folder: Path, simulate_minimum_execution_time: float = 0.1, simulate_maximum_execution_time: float = 1.1) → list[str][source]

Build the code definining all tasks in the workflow, i.e. WorkflowTask instances.

Parameters:

randomizer_varname (str) – The name of the randomizer.
output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.

Returns:

The non-indented Python lines of code used to instantiate the WorkflowTask instances.

Return type:

list[str]

_parse_tasks(task_name: str) → list[str][source]

Recursively iterates over workflow tasks to generate submit command.

Parameters:: task_name (str) – The name of a task.
Returns:: The
Return type:: list[str]

_abc_impl = <_abc._abc_data object>

wfcommons.wfbench.translator.makeflow¶

class wfcommons.wfbench.translator.makeflow.MakeflowTranslator(workflow: Workflow | Path, logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating Makeflow workflow applications.

Parameters:

workflow (Union[Workflow, pathlib.Path],) – Workflow benchmark object or path to the workflow benchmark JSON instance.
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path) → None[source]

Translate a workflow benchmark description (WfFormat) into an actual workflow application.

Parameters:: output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.

_generate_code()[source]

Generate the Makeflow code

Returns:: the code
Return type:: str

_write_readme_file(output_folder: Path) → None[source]

Write the README file.

Parameters:: output_folder (pathlib.Path) – The path of the output folder.

_abc_impl = <_abc._abc_data object>

wfcommons.wfbench.translator.nextflow¶

class wfcommons.wfbench.translator.nextflow.NextflowTranslator(workflow: Workflow | Path, max_tasks_per_subworkflow: int = 100, max_parents_threshold: int | None = 100, slurm: bool | None = False, logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating Nextflow workflow applications.

This translator can generate either a single-file workflow or split the workflow across multiple module files for better scalability with large workflows.

Parameters:

workflow (Union[Workflow, pathlib.Path]) – Workflow benchmark object or path to the workflow benchmark JSON instance.
max_tasks_per_subworkflow (int) – Maximum number of tasks per module file when using subworkflows (default: 500).
max_parents_threshold (int) – Tasks with more parents than this get their own module (default: 100).
slurm (bool) – Whether to generate a Slurm template script for workflow submission using sbatch.
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path) → None[source]

Translate a workflow benchmark description (WfFormat) into a Nextflow workflow application.

Parameters:: output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.

_get_tasks_in_topological_order() → List[Task][source]

Sort the workflow tasks in topological order.

Returns:: A sorted list of tasks.
Return type:: List[Task]

_create_task_script(task: Task) → None[source]

Generate the bash script for invoking a task.

Parameters:: task (Task) – The task.

_build_spec_arg(name, data, threshold)[source]: Returns (arg_string_for_command_line, setup_code_to_prepend).

_generate_flowcept_code() → str[source]

Generate the Flowcept process code.

Returns:: The code.
Return type:: str

_generate_task_process(task: Task) → tuple[str, str][source]

_write_nextflow_config_file(output_folder: Path)[source]

_write_readme_file(output_folder: Path) → None[source]

Write the README file.

Parameters:: output_folder (pathlib.Path) – The path of the output folder.

_write_nf_config_file(output_folder: Path) → None[source]

Write the plugin config file.

Parameters:: output_folder (pathlib.Path) – The path of the output folder.

_generate_task_call(task: Task, function_name: str, inputs_var: str, results_var: str, include_comment: bool) → str[source]

_write_deps_file(function_name: str, input_files: List) → Path[source]

Write the list of a fan-in task’s dependency keys to a sidecar text file, one raw file_id per line, so the module .nf file can load it with readLines() at runtime instead of embedding thousands of statements.

Parameters:

function_name (str) – The sanitized task/process name (used as the filename).
input_files (List) – The task’s input files.

Returns:

Path to the written deps file.

Return type:

pathlib.Path

_generate_main_workflow() → str[source]

Generate the main workflow file that orchestrates all module functions.

Returns:: The main workflow file content.
Return type:: str

_translate_with_subworkflows(output_folder: Path, sorted_tasks: List[Task]) → None[source]

Generate a multi-file Nextflow workflow with module files.

Parameters:

output_folder (pathlib.Path) – The output folder path.
sorted_tasks (List[Task]) – Tasks in topological order.

_split_into_subworkflows(sorted_tasks: List[Task]) → None[source]

Split the sorted tasks into chunks for separate files.

Tasks with more than max_parents_threshold parents are placed in their own subworkflow to avoid generating overly long channel mixing code.

Parameters:: sorted_tasks (List[Task]) – The topologically sorted list of tasks.

_generate_module_file(module_idx: int, tasks: List[Task]) → str[source]

Generate a module .nf file containing processes and an orchestrating function.

Each module file contains: - Process definitions (private to the file) - A single exported function that orchestrates those processes

Using a function instead of a workflow block allows passing Maps of channels between modules without Nextflow channel type casting issues.

Parameters:

module_idx (int) – The index of this module.
tasks (List[Task]) – The tasks in this module.

Returns:

The complete module file content.

Return type:

str

_generate_module_function(module_idx: int, tasks: List[Task]) → str[source]

Generate a Groovy function that orchestrates a set of tasks.

Using a function instead of a workflow block allows passing Maps of channels between modules without Nextflow channel type casting issues.

Parameters:

module_idx (int) – The index of this module.
tasks (List[Task]) – The tasks in this module.

Returns:

The function code.

Return type:

str

_sanitize_string(string: str) → str[source]

_abc_impl = <_abc._abc_data object>

wfcommons.wfbench.translator.parsl¶

class wfcommons.wfbench.translator.parsl.ParslTranslator(workflow: Workflow | Path, logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating Parsl workflow benchmarks.

Parameters:

workflow (Union[Workflow, pathlib.Path],) – Workflow benchmark object or path to the workflow benchmark JSON instance.
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path) → None[source]

Translate a workflow benchmark description (WfFormat) into an actual workflow application.

Parameters:: output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.

_parsl_wftasks_codelines() → list[str][source]

_write_readme_file(output_folder: Path) → None[source]

Write the README file.

Parameters:: output_folder (pathlib.Path) – The path of the output folder.

_abc_impl = <_abc._abc_data object>

wfcommons.wfbench.translator.pegasus¶

class wfcommons.wfbench.translator.pegasus.PegasusTranslator(workflow: Workflow | Path, logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating Pegasus workflow applications.

Parameters:

workflow (Union[Workflow, pathlib.Path],) – Workflow benchmark object or path to the workflow benchmark JSON instance.
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path, tasks_priorities: Dict[str, int] | None = None) → None[source]

Translate a workflow benchmark description (WfFormat) into a Pegasus workflow application.

Parameters:

output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.
tasks_priorities (Optional[Dict[str, int]]) – Priorities to be assigned to tasks.

_add_task(task_name: str, parent_task: str | None = None, tasks_priorities: Dict[str, int] | None = None) → None[source]

Add a task and its dependencies to the workflow.

Parameters:

task_name (str) – name of the task
parent_task (Optional[str]) – name of the parent task
tasks_priorities (Optional[Dict[str, int]]) – Priorities to be assigned to tasks.

_abc_impl = <_abc._abc_data object>

wfcommons.wfbench.translator.pycompss¶

class wfcommons.wfbench.translator.pycompss.PyCompssTranslator(workflow: Workflow | Path, logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating PyCOMPSs workflow applications.

Parameters:

workflow (Union[Workflow, pathlib.Path],) – Workflow benchmark object or path to the workflow benchmark JSON instance.
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path) → None[source]

Translate a workflow benchmark description (WfFormat) into an actual workflow application.

Parameters:: output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.

_pycompss_code() → None[source]

_abc_impl = <_abc._abc_data object>

wfcommons.wfbench.translator.swift_t¶

class wfcommons.wfbench.translator.swift_t.SwiftTTranslator(workflow: Workflow | Path, stress_path: Path = PosixPath('stress-ng'), logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating Swift/T workflow applications.

Parameters:

workflow (Union[Workflow, pathlib.Path]) – Workflow benchmark object or path to the workflow benchmark JSON instance.
stress_path (pathlib.Path) – Path to the stress-ng command.
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path) → None[source]

Translate a workflow benchmark description (WfFormat) into a Swift/T workflow application.

Parameters:: output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.

_find_categories_list(task_name: str, parent_task: str | None = None) → None[source]

” Find list of task categories ordered by task dependencies.

Parameters:

task_name (str) – name of the task
parent_task (Optional[str]) – name of the parent task

_add_tasks(category: str) → None[source]

Add all tasks for a specific category.

Parameters:: category (str) – category name

_write_readme_file(output_folder: Path) → None[source]

Write the README file.

Parameters:: output_folder (pathlib.Path) – The path of the output folder.

_abc_impl = <_abc._abc_data object>

wfcommons.wfbench.translator.taskvine¶

class wfcommons.wfbench.translator.taskvine.TaskVineTranslator(workflow: Workflow | Path, logger: Logger | None = None)[source]

Bases: Translator

A WfFormat parser for creating TaskVine workflow applications.

Parameters:

workflow (Union[Workflow, pathlib.Path],) – Workflow benchmark object or path to the workflow benchmark JSON instance.
logger (Logger) – The logger where to log information/warning or errors (optional).

translate(output_folder: Path) → None[source]

Translate a workflow benchmark description (WfFormat) into an actual workflow application.

Parameters:: output_folder (pathlib.Path) – The path to the folder in which the workflow benchmark will be generated.

_add_level_tasks(tasks_list: list[str]) → list[str][source]

Add all tasks from a level in the workflow.

Parameters:: tasks_list (list[str]) – list of tasks in the level
Returns:: List of next level tasks
Return type:: list[str]

_add_task(task_name: str, parent_task: str | None = None) → list[str][source]

Add a task and its dependencies to the workflow.

Parameters:

task_name (str) – name of the task
parent_task (Optional[str]) – name of the parent task

Returns:

List of children tasks

Return type:

list[str]

_write_readme_file(output_folder: Path) → None[source]

Write the README file.

Parameters:: output_folder (pathlib.Path) – The path of the output folder.

_abc_impl = <_abc._abc_data object>