Select Git revision
Template.py
Template.py 13.99 KiB
#! /usr/bin/python3
from __future__ import annotations
import typing
from scripts.Infrastructure.InstructionBlock import InstructionBlock
from scripts.Infrastructure.Instruction import Instruction
from scripts.Infrastructure.MPICall import MPICall
template = """// @{generatedby}@
/* ///////////////////////// The MPI Bug Bench ////////////////////////
Description: @{desc}@
Version of MPI: @{version}@
THIS BLOCK IS CURRENTLY NOT USED:
BEGIN_MPI_FEATURES
P2P!basic: @{p2pfeature}@
P2P!nonblocking: @{ip2pfeature}@
P2P!persistent: @{persfeature}@
COLL!basic: Lacking
COLL!nonblocking: Lacking
COLL!persistent: Lacking
COLL!tools: Yes
RMA: Lacking
END_MPI_FEATURES
BEGIN_MBB_TESTS
$ mpirun -np @{min_num_ranks}@ ${EXE}
| @{outcome}@
| @{errormsg}@
END_MBB_TESTS
////////////////////// End of MBI headers /////////////////// */
#include <mpi.h>
#include <stdio.h>
#include <stdlib.h>
int main(int argc, char **argv) {
int nprocs = -1;
int rank = -1;
@{mpi_init}@
MPI_Comm_size(MPI_COMM_WORLD, &nprocs);
MPI_Comm_rank(MPI_COMM_WORLD, &rank);
if (nprocs < @{min_num_ranks}@)
printf("MBB ERROR: This test needs at least @{min_num_ranks}@ processes to produce a bug!\\n");
@{test_code}@
@{mpi_finalize}@
printf("Rank %d finished normally\\n", rank);
return 0;
}
"""
init_thread_template = """int provided;
MPI_Init_thread(&argc, &argv, @{thread_level}@, &provided);
if (@{thread_level}@ < provided)
printf("MBI ERROR: The MPI Implementation does not provide the required thread level!\\n");
"""
class TemplateManager:
"""
Class Overview:
The `TemplateManager` class is responsible for managing MPI error case templates.
Attributes:
_descr_full (str): The full description of the template.
_descr_short (str): The short description of the template.
_instructions (list): List of instruction blocks.
_thread_level (str): The MPI thread level to use.
_min_ranks (int): The minimum number of MPI ranks.
_has_finalize (bool): Whether to include a call to MPI_Finalize.
_has_init (bool): Whether to include a call to MPI_Init.
Methods:
- __init__(self, min_ranks: int = 2, thread_level: str = None, has_finalize: bool = True,
has_init: bool = True): Initializes a new instance of the TemplateManager class.
- __str__(self): Converts the TemplateManager instance to a string, replacing placeholders.
- register_instruction(self, inst: Instruction | typing.List[Instruction]): Registers an instruction with the template.
- get_version(self) -> str: Retrieves the minimum required MPI version.
- set_description(self, descr_short: str, descr_full: str): Sets the short and full descriptions for the template.
- get_short_descr(self) -> str: Retrieves the short description (to use as a filename).
- get_instruction(self, identifier: str = None, return_list=False, idx: int = None) -> Instruction | typing.List[Instruction]: Retrieves an instruction or list of them based on its identifier or index.
- insert_instruction(self, new_instruction: Instruction, after_instruction: str | int = None, before_instruction: str | int = None): Inserts a new instruction into the template.
- remove_instruction(self, identifier: str = None, idx: int | typing.List[int] = None): Removes an instruction from the template.
- replace_instruction(self, new_instruction=Instruction, identifier: str = None, idx: int | typing.List[int] = None): Replaces an instruction in the template with a new one.
"""
def __init__(self, min_ranks: int = 2, thread_level: str = None, has_finalize: bool = True, has_init: bool = True):
"""
Initialize the TemplateManager
Parameters:
min_ranks (int): the number of MPI ranks to use for this case
thread_level : the MPI Thread Level to use (None means use MPI_Init instead of Init_Thread)
has_finalize (bool) : if call to MPI Finalize should be included
has_init (bool) : if call to MPI Init should be included
"""
self._descr_full = ""
self._descr_short = ""
self._instructions = []
self._thread_level = thread_level
self._min_ranks = min_ranks
self._has_finalize = has_finalize
self._has_init = has_init
def __str__(self):
"""
Converts the TemplateManager to a string, replacing placeholders.
"""
version = self.get_version()
code_string = ""
current_rank = 'all'
for inst in self._instructions:
if inst.get_ranks_executing != current_rank:
if current_rank != 'all':
code_string = code_string + "}\n" # end previous if
current_rank = inst.get_ranks_executing()
if current_rank == 'not0':
code_string = code_string + "if (rank!=0){\n"
elif current_rank != 'all':
code_string = code_string + "if (rank==%D){\n" % current_rank
code_string += str(inst) + "\n"
# end for inst
if current_rank != 'all':
code_string = code_string + "}\n" # end previous if
init_string = ""
if self._has_init:
if self._thread_level == None:
init_string = "MPI_Init(&argc, &argv);"
else:
init_string = init_thread_template.replace("@{thread_level}@", self._thread_level)
finalize_string = ""
if self._has_finalize:
finalize_string = "MPI_Finalize();"
return (template
.replace("@{min_num_ranks}@", str(self._min_ranks))
.replace("@{mpi_init}@", init_string)
.replace("@{mpi_finalize}@", finalize_string)
.replace("@{desc}@", self._descr_full)
.replace("@{version}@", version)
.replace("@{test_code}@", code_string))
def register_instruction(self, inst: Instruction | typing.List[Instruction]):
"""
Registers an instruction block with the template. inserting it at the end, before the mpi finalize
Parameters:
- block: The instruction block to register.
"""
if isinstance(inst, list):
self._instructions.extend(inst)
else:
self._instructions.append(inst)
def get_version(self) -> str:
"""
Retrieves the minimum required MPI version.
Returns:
str: The MPI version used.
"""
max_v = "0.0"
for block in self._blocks:
assert isinstance(block, InstructionBlock)
max_v = max(block.get_version(), max_v)
return max_v
def set_description(self, descr_short: str, descr_full: str):
"""
Sets the short and full descriptions for the template.
Parameters:
- descr_short: The short description for use as the filename.
- descr_full: The full description for use in the header.
"""
self._descr_full = descr_full
self._descr_short = descr_short
# TODO one could write a function to check if short desc = filename is conforming with the naming sceme
def get_short_descr(self) -> str:
"""
Retrieves the short description to use as a filename.
Returns:
str: The short description.
"""
assert self._descr_short != ""
return self._descr_short
def get_instruction(self, identifier: str = None, return_list=False, idx: int = None) -> Instruction | typing.List[
Instruction]:
"""
Retrieves an instruction or list fh them based on its identifier or index.
Parameters:
identifier (str): The identifier of the instruction.
return_list (bool): Whether to return a list of instruction.
idx (int): The index of the instruction block.
Returns:
Instruction | List[Instruction]: The instruction block(s).
Raises:
ValueError: if both index and identifier are provided
IndexError: if return_list is False and not exactely one instruction is found by index
"""
# assert only one param is not None
parameters = [identifier, idx]
if parameters.count(None) != 1:
raise ValueError("Only one parameter is allowed to be specified")
if identifier is not None:
to_return = [i for i in self._instructions if i.get_identifier() == identifier]
if return_list:
return to_return
if len(to_return) == 1:
return to_return[0]
if len(to_return) == 0:
raise IndexError("Found no matching Instruction")
raise IndexError("Found too many elements")
if idx is not None:
if return_list:
return [self._instructions[idx]]
return self._instructions[idx]
raise ValueError("Neither Both block name nor index is given")
def __get_instruction_index(self, identifier: str) -> typing.List[int]:
"""
internal helper function to receive the indices of instructions with ghe given identifier
"""
return [idx for inst, idx in enumerate(self._instructions) if inst.get_identifier() == identifier]
def insert_instruction(self, new_instruction: Instruction, after_instruction: str | int = None,
before_instruction: str | int = None):
"""
Inserts a new instruction into the template.
Parameters:
new_instruction (Instruction): The new instruction to insert.
after_instruction (str | int): The instruction after which to insert the new one (identifier or index).
before_instruction (str | int): The instruction before which to insert the new one (identifier or index).
Raises:
ValueError: if both before and after are provided
IndexError: if it finds multiple places to insert by identifier
"""
# assert only one param is not None
parameters = [after_instruction, before_instruction]
if parameters.count(None) != 1:
raise ValueError("Only one parameter is allowed to be specified")
idx_to_use = 0
if after_instruction is not None:
if isinstance(after_instruction, int):
idx_to_use = 1 + after_instruction
else:
assert isinstance(after_instruction, str)
inst_idx_list = self.__get_instruction_index(after_instruction)
if len(inst_idx_list) != 1:
raise IndexError("Did not find place to insert")
idx_to_use = 1 + inst_idx_list[0]
if before_instruction is not None:
if isinstance(before_instruction, int):
idx_to_use = before_instruction
else:
assert isinstance(before_instruction, str)
inst_idx_list = self.__get_instruction_index(before_instruction)
if len(inst_idx_list) != 1:
raise IndexError("Did not find place to insert")
idx_to_use = inst_idx_list[0]
self._instructions.insert(idx_to_use, new_instruction)
def remove_instruction(self, identifier: str = None, idx: int | typing.List[int] = None):
"""
Removes an instruction from the template.
Parameters:
identifier (str): The identifier of the instruction to remove.
idx (int | List[int]): The index or list of indices of the instruction(s) to remove.
"""
# assert only one param is not None
parameters = [identifier, idx]
if parameters.count(None) != 1:
raise ValueError("Only one parameter is allowed to be specified")
idxs_to_remove = []
if idx is not None:
if isinstance(idx, int):
idxs_to_remove = [idx]
else:
idxs_to_remove = idx
if identifier is not None:
idxs_to_remove = self.__get_instruction_index(identifier)
if len(idxs_to_remove) == 0:
# TODO
# may also be a silen No-Op?
raise ValueError("Nothing to remove")
self._instructions = [elem for idx, elem in enumerate(self._instructions) if idx not in idxs_to_remove]
def replace_instruction(self, new_instruction=Instruction, identifier: str = None,
idx: int | typing.List[int] = None):
"""
Replaces an instruction in the template with a new one.
Parameters:
new_instruction (Instruction | List[Instruction]): The new instruction(s) to replace with.
identifier (str): The identifier of the instruction to replace.
idx (int | List[int]): The index or list of indices of the instruction(s) to replace.
Raises
ValueError: if the number of instructions to replace does not match the number of instructions provided
or if both idx and identifier are provided
Notes: The instructions to be replaced must not be in contiguous order
"""
parameters = [identifier, idx]
if parameters.count(None) != 1:
raise ValueError("Only one parameter is allowed to be specified")
new_instruction_list = []
if isinstance(new_instruction, Instruction):
new_instruction_list = [new_instruction]
else:
new_instruction_list = new_instruction
idxs_to_replace = []
if idx is not None:
if isinstance(idx, int):
idxs_to_replace = [idx]
else:
idxs_to_replace = idx
if identifier is not None:
idxs_to_replace = self.__get_instruction_index(identifier)
if len(idxs_to_replace) == len(new_instruction_list):
raise ValueError("Number of instructions to Replace does not match number of given instructions")
for (index, replacement) in zip(idxs_to_replace, new_instruction_list):
self._instructions[index] = replacement