Select Git revision
InstructionBlock.py
InstructionBlock.py 8.45 KiB
from __future__ import annotations
import typing
from scripts.Infrastructure.Instruction import Instruction
from scripts.Infrastructure.MPICall import MPI_Call
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, MPI_Call):
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:
assert 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':
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
print(op)
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]