Skip to content
Snippets Groups Projects
Commit ed2c1cc8 authored by Jammer, Tim's avatar Jammer, Tim
Browse files

TemplateManager dealing with instructions directly

parent 91e29056
No related branches found
No related tags found
1 merge request!14Infrastructure: Remove Instructionblock
Show whitespace changes
Inline Side-by-side
scripts/Infrastructure/InstructionBlock.py deleted 100644 → 0
+ 0
− 202
View file @ 91e29056
from __future__ import annotations
import typing
from scripts.Infrastructure.Instruction import Instruction
from scripts.Infrastructure.MPICall import MPICall
class InstructionBlock:
"""
Class Overview:
The `InstructionBlock` class represents a block of instructions in a Testcase (to be registered for a template).
First, the Instructions for all ranks are executed (in the order they are registered)
Then each thread executes the instructions registered to this specific rank
If one need a different Order: use multiple Instruction Blocks
Methods:
- `__init__(self)`: Initializes a new instance of the InstructionBlock class.
- `register_operation(self, op, kind='all')`: Registers an operation based on rank.
- `get_version(self)`: Retrieves required MPI version
- `__str__(self)`: Converts the InstructionBlock instance to a string, replacing placeholders.
"""
def __init__(self, name: str = None):
"""
Initialize an empty InstructionBlock
Parameters:
- name (str): The name of the block (for referencing this block with the template Manager)
May be None, does not influence the code generated
"""
self.operations = {'all': [], 'not0': [], }
assert not isinstance(name, int)
self.name = name
def register_instruction(self, op: str | Instruction | typing.List[Instruction], kind: str | int = 'all'):
"""
Registers an operation based on rank.
Parameters:
- op: The operation (or list of Operations) to register.
- kind: Rank to execute the operation ('all', 'not0', or integer).
- all: all Ranks execute this operation
- not0: all Ranks but the Root (rank 0) execute
- Or the integer of the rank that should execute
Note: if a str is passed as the operation, it will create a new Instruction from the given string
"""
if isinstance(op, str):
op = Instruction(op)
if kind == 'all':
if isinstance(op, list):
self.operations['all'].extend(op)
else:
self.operations['all'].append(op)
elif kind == 'not0':
if isinstance(op, list):
self.operations['not0'].extend(op)
else:
self.operations['not0'].append(op)
else:
as_int = int(kind) # will Raise ValueError if not integer
if as_int not in self.operations:
self.operations[as_int] = []
if isinstance(op, list):
self.operations[as_int].extend(op)
else:
self.operations[as_int].append(op)
def get_version(self) -> str:
"""
Retrieves the minimum required MPI version.
Returns:
str: The MPI version used.
"""
max_v = "0.0"
for k, v in self.operations.items():
for op in v:
if isinstance(op, MPICall):
max_v = max(op.get_version(), max_v)
return max_v
def __str__(self):
"""
Converts the InstructionBlock instance to a string, replacing placeholders.
Returns:
str: The string representation of the InstructionBlock.
"""
result_str = ""
for key, val in self.operations.items():
if key == 'all':
for op in val:
result_str += str(op) + "\n"
elif key == 'not0':
if len(val) > 0:
result_str += "if (rank != 0) {\n"
for op in val:
result_str += str(op) + "\n"
result_str += "}\n"
else:
if len(val) > 0:
result_str += "if (rank == %d) {\n" % int(key)
for op in val:
result_str += str(op) + "\n"
result_str += "}\n"
return result_str
def has_instruction(self, kind: int | str = 'all', index: int = 0) -> bool:
"""
Checks if the Block has an operation with the given index and kind
Parameters:
- kind ('all','not0' or integer): which ranks should execute the operation
- index (int ): the index of the operation within the given kind
Returns:
boolean
"""
try:
result = self.operations[kind][index]
return True
except (KeyError, IndexError) as e:
return False
def get_instruction(self, kind: int | str = 'all', index: str | int = 0) -> Instruction | typing.List[Instruction]:
"""
Retrieve the operation registered. will Raise IndexError if not present
Parameters:
- kind ('all','not0' or integer): which ranks should execute the operation
- index ('all' or int): the index of the operation within the given kind; 'all' means that the list of all operations for the kind is returned
Returns:
str: The operation specified by kind and index
"""
if index == 'all':
if kind not in self.operations:
return []
return self.operations[kind]
else:
as_int = int(index) # will Raise ValueError if not integer
return self.operations[kind][as_int]
def replace_instruction(self, op: str | Instruction | typing.List[Instruction], kind: str | int = 'all',
index: str | int = 0):
"""
Replace the operation registered. will Raise IndexError if not present
Parameters:
- op the new operation or list of operations
- kind ('all','not0' or integer): which ranks should execute the operation
- index ('all' or int): the index of the operation within the given kind; 'all' means all operations will be replaced with the given list
Notes : if one wants to replace all operations one needs to provide a list
if one only wants to replace one operation: no list of operations is allowed
if a string is passed as the operation, it will create a new Instruction
"""
if isinstance(op, str):
op = Instruction(op)
if index == 'all':
if not isinstance(op, list):
raise ValueError('Provide List for replacement')
self.operations[kind] = op
else:
as_int = int(index) # will Raise ValueError if not integer
if not isinstance(op, Instruction):
raise ValueError('Provide Instruction')
if len(self.operations[kind]) < as_int:
raise IndexError("Operation Not Found")
self.operations[kind][as_int] = op
def insert_instruction(self, op: str | Instruction | typing.List[Instruction], kind: str | int = 'all',
before_index: int = 0):
"""
Inserts an operation before the specified one. will Raise IndexError if not present
Parameters:
- op the new operation or list of operations
- kind ('all','not0' or integer): which ranks should execute the operation
- index (int): the index of the operation within the given kind
note: if str is passed as the operation, it will Create a New Instruction
"""
if isinstance(op, str):
op = Instruction(op)
as_int = int(before_index) # will Raise ValueError if not integer
if len(self.operations[kind]) < before_index:
raise IndexError("Operation Not Found")
if isinstance(op, list):
self.operations[kind] = (
self.operations[kind][0:before_index - 1] + op + self.operations[kind][before_index:])
else:
self.operations[kind].insert(before_index, op)
def remove_instruction(self, kind: str | int = 'all', index: str | int = 0):
"""
Removes the operation registered. will Raise IndexError if not present
Parameters:
- kind ('all','not0' or integer): which ranks should execute the operation
- index ('all' or int): the index of the operation within the given kind
"""
if index == 'all':
self.operations[kind] = []
else:
as_int = int(index) # will Raise ValueError if not integer
if len(self.operations[kind]) < index:
raise IndexError("Operation Not Found")
del self.operations[kind][index]
scripts/Infrastructure/Template.py
+ 105
− 106
View file @ ed2c1cc8
#! /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}@
......@@ -83,7 +89,7 @@ class TemplateManager:
"""
self._descr_full = ""
self._descr_short = ""
self._blocks = []
self._instructions = []
self._thread_level = thread_level
self._min_ranks = min_ranks
self._has_finalize = has_finalize
......@@ -117,13 +123,16 @@ class TemplateManager:
.replace("@{version}@", version)
.replace("@{test_code}@", block_string))
def register_instruction_block(self, block: InstructionBlock):
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.
"""
self._blocks.append(block)
if isinstance(inst, list):
self._instructions.extend(inst)
else:
self._instructions.append(inst)
def get_version(self) -> str:
"""
......@@ -157,132 +166,122 @@ class TemplateManager:
assert self._descr_short != ""
return self._descr_short
def get_block(self, block_name: str = None, idx: int = None) -> InstructionBlock:
def get_instruction(self, identifier: str = None, return_list=False, idx: int = None) -> Instruction | typing.List[
Instruction]:
"""
Retrieves the given Instruction Block Either by name or by index
Raises IndexError if the specified block is not found
Raises IndexError if multiple Blocks with the given name are found
Raises ValueError if Both a block name and index are given (or none is given)
Args:
block_name (str): The name of the InstructionBlock to receive
idx (int): index of the InstructionBlock to retrieve
Returns:
the specified Block
TODO DOCUMENTATION
"""
if block_name is not None:
if idx is not None:
raise ValueError("Both block name and index are given")
to_return = [b for b in self._blocks if b.name == block_name]
if len(to_return) == 0:
raise IndexError("Block Not Found")
if len(to_return) > 1:
raise IndexError("Multiple Blocks Found")
# 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 block_name is not None:
raise ValueError("Both block name and index are given")
return self._blocks[idx]
if return_list:
return [self._instructions[idx]]
return self._instructions[idx]
raise ValueError("Neither Both block name nor index is given")
def insert_block(self, new_block: InstructionBlock, after_block_name: str = None, after_idx: int = None,
before_block_name: str = None, before_idx: int = None):
def __get_instruction_index(self, identifier: str) -> typing.List[int]:
"""
inserts the given Instruction Block after/before the one specified Either by name or by index
Raises IndexError if the specified block is not found
Raises IndexError if multiple Blocks with the given name are found
Raises ValueError if Both a block name and index are given (or none is given)
Raises ValueError if Both a before and an after block is given (or none is given)
Args:
new_block (InstructionBlock): the block to insert
after_block_name (str): The name of the InstructionBlock to receive
after_idx (int): index of the InstructionBlock to retrieve
TODO: documentation
"""
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):
"""
TODO: documentation
"""
# assert only one param is not None
parameters = [after_block_name, after_idx, before_block_name, before_idx]
if parameters.count(None) != 3:
parameters = [after_instruction, before_instruction]
if parameters.count(None) != 1:
raise ValueError("Only one parameter is allowed to be specified")
if after_block_name is not None:
to_return = [b for b in self._blocks if b.name == after_block_name]
if len(to_return) == 0:
raise IndexError("Block Not Found")
if len(to_return) > 1:
raise IndexError("Multiple Blocks Found")
self._blocks.insert(self._blocks.index(to_return[0]) + 1, new_block)
return
if before_block_name is not None:
to_return = [b for b in self._blocks if b.name == before_block_name]
if len(to_return) == 0:
raise IndexError("Block Not Found")
if len(to_return) > 1:
raise IndexError("Multiple Blocks Found")
self._blocks.insert(self._blocks.index(to_return[0]), new_block)
return
if after_idx is not None:
self._blocks.insert(after_idx + 1, new_block)
return
if before_idx is not None:
self._blocks.insert(before_idx, new_block)
return
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]
raise ValueError("Neither Both block name nor index is given")
self._instructions.insert(idx_to_use, new_instruction)
def remove_block(self, block_name: str = None, idx: int = None):
def remove_instruction(self, identifier: str = None, idx: int | typing.List[int] = None):
"""
Removes the given Instruction Block Either by name or by index
Raises IndexError if the specified block is not found
Raises IndexError if multiple Blocks with the given name are found
Raises ValueError if Both a block name and index are given (or none is given)
Args:
block_name (str): The name of the InstructionBlock to receive
idx (int): index of the InstructionBlock to retrieve
TODO: documentation
"""
if block_name is not None:
if idx is not None:
raise ValueError("Both block name and index are given")
to_return = [b for b in self._blocks if b.name == block_name]
if len(to_return) == 0:
raise IndexError("Block Not Found")
if len(to_return) > 1:
raise IndexError("Multiple Blocks Found")
self._blocks.remove(to_return[0])
# 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 block_name is not None:
raise ValueError("Both block name and index are given")
del self._blocks[idx]
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)
raise ValueError("Neither Both block name nor index is given")
if len(idxs_to_remove) == 0:
# TODO
# may also be a silen No-Op?
raise ValueError("Nothing to remove")
def replace_block(self, new_block: InstructionBlock, block_name: str = None, idx: int = None):
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):
"""
Removes the given Instruction Block Either by name or by index
Raises IndexError if the specified block is not found
Raises IndexError if multiple Blocks with the given name are found
Raises ValueError if Both a block name and index are given (or none is given)
Args:
new_block (InstructionBlock): The new Block to replace the old one (does not need to have the same name)
block_name (str): The name of the InstructionBlock to receive
idx (int): index of the InstructionBlock to retrieve
TODO: documentation
"""
assert isinstance(new_block, InstructionBlock)
if block_name is not None:
if idx is not None:
raise ValueError("Both block name and index are given")
to_return = [b for b in self._blocks if b.name == block_name]
if len(to_return) == 0:
raise IndexError("Block Not Found")
if len(to_return) > 1:
raise IndexError("Multiple Blocks Found")
self._blocks[self._blocks.index(to_return[0])] = new_block
# assert only one param is not None
parameters = [identifier, idx]
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 block_name is not None:
raise ValueError("Both block name and index are given")
self._blocks[idx] = new_block
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)
raise ValueError("Neither Both block name nor index is given")
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
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment