#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (c) 2020-2025 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 uuid
from abc import ABC, abstractmethod
from os import path
from logging import Logger
from typing import Any, Dict, List, Optional
from wfcommons.common.file import File
from wfcommons.common.task import Task, TaskType
from wfcommons.common.workflow import Workflow
from wfcommons.utils import generate_rvs
[docs]
class WorkflowRecipe(ABC):
"""
An abstract class of workflow recipes for creating synthetic workflow instances.
:param name: The workflow recipe name.
:type name: str
:param data_footprint: The upper bound for the workflow total data footprint (in bytes).
:type data_footprint: Optional[int]
:param num_tasks: The upper bound for the total number of tasks in the workflow.
:type num_tasks: Optional[int]
:param runtime_factor: The factor of which tasks runtime will be increased/decreased.
:type runtime_factor: Optional[float]
:param input_file_size_factor: The factor of which tasks input files size will be increased/decreased.
:type input_file_size_factor: Optional[float]
:param output_file_size_factor: The factor of which tasks output files size will be increased/decreased.
:type output_file_size_factor: Optional[float]
:param logger: The logger where to log information/warning or errors (optional).
:type logger: Optional[Logger]
"""
def __init__(self, name: str,
data_footprint: Optional[int],
num_tasks: Optional[int],
runtime_factor: Optional[float] = 1.0,
input_file_size_factor: Optional[float] = 1.0,
output_file_size_factor: Optional[float] = 1.0,
logger: Optional[Logger] = None) -> None:
"""Create an object of the workflow recipe."""
# sanity checks
if runtime_factor <= 0.0:
raise ValueError("The runtime factor should be a number higher than 0.0.")
if input_file_size_factor <= 0.0:
raise ValueError("The input file size factor should be a number higher than 0.0.")
if output_file_size_factor <= 0.0:
raise ValueError("The output file size factor should be a number higher than 0.0.")
self.logger: Optional[Logger] = logging.getLogger(__name__) if logger is None else logger
self.name = name
self.data_footprint: Optional[int] = data_footprint
self.num_tasks: Optional[int] = num_tasks
self.runtime_factor: Optional[float] = runtime_factor
self.input_file_size_factor: Optional[float] = input_file_size_factor
self.output_file_size_factor: Optional[float] = output_file_size_factor
self.workflow_recipe = None
self.task_id_counter: int = 1
self.tasks_map = {}
self.tasks_children = {}
self.tasks_parents = {}
self.tasks_input_files: Dict[str, List[File]] = {}
self.tasks_input_files_names: Dict[str, List[str]] = {}
self.tasks_output_files: Dict[str, List[File]] = {}
self.tasks_output_files_names: Dict[str, List[str]] = {}
[docs]
@abstractmethod
def _workflow_recipe(self) -> Dict[str, Any]:
"""
Recipe for generating synthetic instances for a workflow. Recipes can be
generated by using the :class:`~wfcommons.wfinstances.instance_analyzer.InstanceAnalyzer`.
:return: A recipe in the form of a dictionary in which keys are task prefixes.
:rtype: Dict[str, Any]
"""
raise NotImplementedError
[docs]
@classmethod
@abstractmethod
def from_num_tasks(cls,
num_tasks: int,
runtime_factor: Optional[float] = 1.0,
input_file_size_factor: Optional[float] = 1.0,
output_file_size_factor: Optional[float] = 1.0
) -> 'WorkflowRecipe':
"""
Instantiate a workflow recipe that will generate synthetic workflows up to the
total number of tasks provided.
:param num_tasks: The upper bound for the total number of tasks in the workflow.
:type num_tasks: int
:param runtime_factor: The factor of which tasks runtime will be increased/decreased.
:type runtime_factor: Optional[float]
:param input_file_size_factor: The factor of which tasks input files size will be increased/decreased.
:type input_file_size_factor: Optional[float]
:param output_file_size_factor: The factor of which tasks output files size will be increased/decreased.
:type output_file_size_factor: Optional[float]
:return: A workflow recipe object that will generate synthetic workflows up to
the total number of tasks provided.
:rtype: WorkflowRecipe
"""
raise NotImplementedError
[docs]
@abstractmethod
def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
"""
Generate a synthetic workflow instance.
:param workflow_name: The workflow name
:type workflow_name: Optional[str]
:return: A synthetic workflow instance object.
:rtype: Workflow
"""
raise NotImplementedError
[docs]
def _generate_task(self, task_name: str, task_id: str) -> Task:
"""
Generate a synthetic task (mostly blank)
:param task_name: task name.
:type task_name: str
:param task_id: task ID.
:type task_id: str
:return: A task object.
:rtype: task
"""
task_recipe = self._workflow_recipe()[task_name]
# runtime
runtime: float = float(format(
self.runtime_factor * generate_rvs(task_recipe['runtime']['distribution'],
task_recipe['runtime']['min'],
task_recipe['runtime']['max']), '.3f'))
# linking previous generated output files as input files
self.tasks_input_files[task_id] = []
self.tasks_input_files_names[task_id] = []
task = Task(
name=task_name,
# task_id='0{}'.format(task_id.split('_0')[1]),
task_id=task_id,
# category=task_name,
task_type=TaskType.COMPUTE,
runtime=runtime,
machines=[],
program=task_name,
args=[],
cores=1,
avg_cpu=None,
bytes_read=None,
bytes_written=None,
memory=None,
energy=None,
avg_power=None,
priority=None,
input_files=[],
output_files=[]
)
self.tasks_map[task_id] = task
return task
[docs]
def _generate_task_name(self, prefix: str) -> str:
"""
Generate a task name from a prefix appended with an ID.
:param prefix: task prefix.
:type prefix: str
:return: task name from prefix appended with an ID.
:rtype: str
"""
task_name = "{}_{:08d}".format(prefix, self.task_id_counter)
self.task_id_counter += 1
return task_name
[docs]
def _generate_task_files(self, task: Task) -> List[File]:
"""
Generate input and output files for a task. This method is recursive, and is initially called
for each leaf (i.e., exit) tasks of the workflow.
:param task: task object.
:type task: Task
:return: List of files output files.
:rtype: List[File]
"""
# Perhaps we already dit this recursively
if task.task_id in self.tasks_output_files.keys():
return self.tasks_output_files[task.task_id]
# Get the recipe for the task
task_recipe = self._workflow_recipe()[task.name]
# generate the task's output files
output_files_list = self._generate_files(task.task_id, task_recipe['output'], "output")
task.output_files = output_files_list
# obtain input files from parents
input_files = []
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.file_id not in self.tasks_input_files_names[task.task_id]:
self.tasks_input_files[task.task_id].append(File(file_id=input_file.file_id,
size=input_file.size))
self.tasks_input_files_names[task.task_id].append(input_file.file_id)
# generate additional input files (i.e., not coming from parents...)
self._generate_files(task.task_id, task_recipe['input'], "input")
task.input_files = [ifile for ifile in self.tasks_input_files[task.task_id]]
return output_files_list
[docs]
def _generate_files(self, task_id: str, recipe: Dict[str, Any], input_or_output: str) -> List[File]:
"""
Generate files for a specific task ID.
:param task_id: task ID.
:type task_id: str
:param recipe: Recipe for generating the task.
:type recipe: Dict[str, Any]
:param input_of_output: Whether the file is output or input.
:type input_of_output: str
:return: List of files.
:rtype: List[File]
"""
files_list = []
extension_list: List[str] = []
if (input_or_output == "output") and (task_id not in self.tasks_output_files):
for extension in recipe:
if extension not in extension_list:
file = self._generate_file(extension, recipe, input_or_output)
files_list.append(file)
self.tasks_output_files[task_id] = files_list
return files_list
# Setup of references to relevant maps based in input/output
if input_or_output == "input":
task_files = self.tasks_input_files[task_id]
task_files_names = self.tasks_input_files_names[task_id]
elif input_or_output == "output":
task_files = self.tasks_output_files[task_id]
task_files_names = self.tasks_output_files_names[task_id]
else:
raise ValueError(f"Invalid input_or_output value: {input_or_output}")
# Process existing files
for f in task_files:
extension_list.append(path.splitext(f.file_id)[1] if '.' in f.file_id else f.file_id)
files_list.append(f)
# Generate additional files from recipe
for extension in recipe:
if extension not in extension_list:
file = self._generate_file(extension, recipe, input_or_output)
files_list.append(file)
task_files.append(file)
task_files_names.append(file.file_id)
return files_list
[docs]
def _generate_file(self, extension: str, recipe: Dict[str, Any], input_of_output: str) -> File:
"""
Generate a file according to a file recipe.
:param extension:
:type extension: str
:param recipe: Recipe for generating the file.
:type recipe: Dict[str, Any]
:param input_of_output: Whether the file is output or input.
:type input_of_output: str
:return: The generated file.
:rtype: File
"""
size = int((self.input_file_size_factor if input_of_output == "input"
else self.output_file_size_factor) * generate_rvs(recipe[extension]['distribution'],
recipe[extension]['min'],
recipe[extension]['max']))
return File(file_id=str(uuid.uuid4()) + extension,
size=size)