2026-07-10 10:38:57 +02:00

577 lines
29 KiB
Python

import json
import math
import time
from ScEpTIC.AST.transformations.base.instructions.enter_mcu_lpm import EnterMCULPM
from ScEpTIC.analysis.utils.saved_state_calculator import SavedStateCalculator
from ScEpTIC.analysis.utils.voltage_thresholds_identifier import VoltageThresholdsIdentifier
from ScEpTIC.emulator.energy.options import OpModeName
from ScEpTIC.exceptions import ConfigurationException
from ScEpTIC.analysis.base_analysis import ScEpTICAnalysis
from ScEpTIC.AST.elements.instructions.memory_operations import StoreOperation, LoadOperation
from ScEpTIC.AST.elements.instructions.other_operations import CallOperation
from ScEpTIC.AST.elements.instructions.termination_instructions import ReturnOperation
from ScEpTIC.AST.transformations.base.instructions.simulate_clock_cycles import SimulateClockCycles
from ScEpTIC.analysis.options import AnalysisResultFormat
from ScEpTIC.emulator.energy.mcu import MCUClockCycleAction, MCUPowerState
from ScEpTIC.tools import fancy_dict_to_str, fancy_list_to_str
from ScEpTIC.emulator.energy import energy_utils
from ScEpTIC.AST.transformations.base.instructions.printf import Printf
class EnergyAnalysis(ScEpTICAnalysis):
"""
Energy analysis
"""
def __init__(self, vm):
super().__init__(vm)
self._check_unsupported_conf()
# Configuration params
self.system_model = self.vm.state.config.analysis.energy.system_model
self.power_failures_during_state_save = not self.vm.state.config.analysis.energy.ignore_power_failures_during_state_save
self.max_simulation_time = self.vm.state.config.analysis.energy.max_simulation_time
# State-save params
self.use_checkpoints = self.vm.state_retention.use_checkpoints
self.static_placement = self.vm.state_retention.static_placement
self.probe_energy_buffer = self.vm.state_retention.probe_energy_buffer
self.probe_with_adc = self.vm.state_retention.probe_with_adc
# Hibernate after save
self.hibernation_enabled = self.vm.state_retention.hibernate_after_save
# State save calculator
self.state_calc = SavedStateCalculator(self.vm)
self.stdout_enabled = Printf.stdout_enabled
# Threshold voltage for saving the state
if self.use_checkpoints and (not self.static_placement or self.probe_energy_buffer):
# Identify optimal voltage
if self.vm.state_retention.calculate_v_save_state:
v_identifier = VoltageThresholdsIdentifier(self.vm, self.system_model)
v_identifier.identify()
self.v_save_state = v_identifier.v_save_state
self.v_resume = v_identifier.v_resume
# Fixed in configuration
else:
self.v_save_state = self.vm.state_retention.v_save_state
self.v_resume = self.vm.state_retention.v_resume
# Probe with ADC
if self.probe_energy_buffer and self.probe_with_adc:
if not self.system_model.mcu.has_adc:
raise Exception("MCU does not have an ADC (probe set to use ADC)")
# ADC min voltage may be higher than optimal v_state_save
# -> ADC must be on to probe energy buffer
self.v_save_state = max(self.v_save_state, self.system_model.mcu.adc_min_v)
self.v_resume = max(self.v_resume, self.system_model.mcu.adc_min_v)
custom_signals = [('probe', 'Probe Energy Buffer'), ('save', 'Save State'), ('restore', 'Restore State')]
self.system_model.init_custom_signals(custom_signals)
if self.vm.state.config.analysis.energy.calculate_cache_only:
import sys
sys.exit(0)
def _check_unsupported_conf(self):
"""
Cheks for unsupported configurations
"""
if self.vm.state_retention.restore_non_volatile_gst:
raise Exception(f"Unsupported configuration: restore non-volatile gst")
if self.vm.state_retention.restore_stack and self.vm.state.memory.memory_positions['stack'] == 'non_volatile':
raise Exception(f"Unsupported configuration: restore non-volatile stack")
if self.vm.state_retention.restore_heap and self.vm.state.memory.memory_positions['heap'] == 'non_volatile':
raise Exception(f"Unsupported configuration: restore non-volatile heap")
def mmu_log_energy_failure(self):
for mmu in self.vm.state.memory.mmus.values():
mmu.log_data('ENERGY_FAILURE', None, None, None, None, None)
def get_result(self, result_format):
"""
Returns the analysis result
:param result_format: the result format
"""
names, result = self.system_model.get_stats()
names['global_clock'] = 'Global CLock (ScEpTIC VM)'
result['global_clock'] = self.vm.state.global_clock
signals = self.system_model.get_collected_signals()
results = {}
if result_format == AnalysisResultFormat.JSON:
results['analysis'] = json.dumps(result)
if len(signals) > 1:
results['signals'] = json.dumps(signals)
elif result_format == AnalysisResultFormat.TEXT:
retstr = '[General Stats]\n'
for name, val in result.items():
if not isinstance(val, dict) and not isinstance(val, list):
retstr += f' {names[name]}: {val}\n'
retstr += '\n'
for name, val in result.items():
if isinstance(val, dict):
retstr += f'[{names[name]}]\n'
retstr += fancy_dict_to_str(val)
retstr += '\n\n'
elif isinstance(val, list):
retstr += f'[{names[name]}]\n'
retstr += fancy_list_to_str(val)
retstr += '\n\n'
results['analysis'] = retstr
if len(signals) > 1:
retstr = ''
for data in signals:
data = ', '.join([str(x) for x in data])
retstr += f"{data}\n"
results['signals'] = retstr
else:
raise ConfigurationException(f"Invalid result format {result_format}")
return results
def run(self):
"""
Runs the energy analysis simulation
"""
state_save_function_name = self.vm.state_retention.routine_names['save']
op_mode_name = OpModeName.PROGRAM_EXECUTION
adc_op_name = OpModeName.PROBE_ENERGY_BUFFER
self.vm.reset()
self.system_model.reset()
self.system_model.energy_buffer.set_voltage(self.system_model.mcu.v_on)
self.system_model.mcu.set_mcu_state(MCUPowerState.ON)
# init MCU and custom devices; set clock to custom devices clock cycles
self.vm.state.global_clock += self.system_model.init()
save_id = 0
restore_id = 0
# Flag to always save the state (V < v_adc_min)
probe_adc_low_voltage_flag = False
self.system_model.set_custom_signal('probe', False)
self.system_model.set_custom_signal('save', False)
self.system_model.set_custom_signal('restore', False)
print("Starting simulation")
t = time.time()
last_lpm_print_tick = 0
print_lpm_out = False
self.vm.set_termination_reason("SIMULATION_COMPLETED")
while not self.vm.state.program_end_reached:
# current instruction
current_instruction = self.vm.state.current_instruction
#print(current_instruction)
#print(self.vm.state.register_file.pc.pc_tree())
op_ticks = current_instruction.tick_count
sysmodel_ticks = 0
power_failure_occurred = False
save_state = False
# Function name
function_name = None
if isinstance(current_instruction, CallOperation):
function_name = current_instruction.resolve_function_name()
# Execute instruction
self.vm.state.run_step()
ticks_elapsed = self.system_model.elapsed_ticks['total']['total']
if ticks_elapsed % 1500000 == 0:
elapsed = time.time() - t
print(f"Still running {self.vm.state.register_file.pc} - TICKS={energy_utils.float_to_str(ticks_elapsed, omit_dot_zero=True)} - elapsed {math.floor(elapsed)}s ({math.floor(ticks_elapsed/elapsed)} op/s) - Cap={self.system_model.energy_buffer.get_voltage()}V; Harvester={self.system_model.energy_harvester.get_voltage()}V; Source={self.system_model.energy_source.get_voltage()}V; T={round(self.system_model.energy_source.elapsed_time, 3)}s")
if self.max_simulation_time > 0:
if self.system_model.get_simulation_time() >= self.max_simulation_time:
self.vm.set_termination_reason("MAX_SIMULATION_TIME_REACHED")
print(f"MAX SIMULATION TIME REACHED")
break
# Simulate cycles
if isinstance(current_instruction, SimulateClockCycles):
op_type = current_instruction.get_instruction_type()
sysmodel_ticks += self.system_model.run_step(op_type, OpModeName.SIMULATE_CYCLES)
### COSTS
# Operations not corresponding to a machine-code instruction has op_tick = 0 and need not to execute an
# energy-simulation cycle
# e.g., a "useless branch" (targets next instruction only; used by llvm to split the program into basic blocks)
elif op_ticks == 0:
sysmodel_ticks += 0
# Load / Store
elif isinstance(current_instruction, LoadOperation) or isinstance(current_instruction, StoreOperation):
op_type, additional_op_mode_name = self.state_calc.get_memory_access_mcu_action(current_instruction)
sysmodel_ticks += self.system_model.run_step(op_type, op_mode_name, additional_op_mode_name=additional_op_mode_name)
# Call / Return
elif isinstance(current_instruction, CallOperation) or isinstance(current_instruction, ReturnOperation):
mem_op_type, additional_op_mode_name = self.state_calc.get_call_ret_mcu_action(current_instruction)
memory_ticks = current_instruction.memory_tick_count
normal_ticks = op_ticks - memory_ticks
for _ in range(memory_ticks):
sysmodel_ticks += self.system_model.run_step(mem_op_type, op_mode_name, additional_op_mode_name=additional_op_mode_name)
for _ in range(normal_ticks):
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.NO_MEMORY_ACCESS, op_mode_name)
# All the other instructions
else:
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.NO_MEMORY_ACCESS, op_mode_name)
### CUSTOM ACTIONS
# Enter LPM
if isinstance(current_instruction, EnterMCULPM):
lpm_time = current_instruction.get_lpm_time()
# - If power failure occurs -> stop (power failure simulated after the else branch)
# - If voltage raises above v_resume -> resume from LPM
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.LPM_ENTER, OpModeName.LPM_ENTER)
if self.stdout_enabled:
ticks_elapsed = self.system_model.elapsed_ticks['total']['total']
if last_lpm_print_tick < (ticks_elapsed - 50000):
print(f"Entering LPM at {self.system_model.energy_buffer.get_voltage()}V")
print_lpm_out = True
last_lpm_print_tick = ticks_elapsed
# Start timekeeper
if self.system_model.timekeeper is not None:
self.system_model.timekeeper.start_tracking()
lpm_start_time = self.system_model.get_simulation_time()
lpm_end_time = lpm_start_time + lpm_time
lpm_wakeup_using_time = lpm_time > 0
i = 0
while True:
if i % 50000 == 0:
if self.stdout_enabled:
print(f"LPM Recharging - Cap={self.system_model.energy_buffer.get_voltage()}V; Harvester={self.system_model.energy_harvester.get_voltage()}V; Source={self.system_model.energy_source.get_voltage()}V; T={round(self.system_model.energy_source.elapsed_time, 3)}s")
if self.max_simulation_time > 0:
if self.system_model.get_simulation_time() >= self.max_simulation_time:
self.vm.set_termination_reason("MAX_SIMULATION_TIME_REACHED", "LPM_RECHARGING")
print(f"MAX SIMULATION TIME REACHED DURING LPM RECHARGING")
break
i += 1
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.LPM_NOP, OpModeName.LPM)
# Energy failure (do not stop timekeeper)
if self.system_model.power_failure_occurring():
power_failure_occurred = True
break
# Resume when:
# - Time lpm_time elapsed (if lpm_time > 0)
# - v_on reached (if lpm_time = 0)
if (lpm_wakeup_using_time and lpm_end_time < self.system_model.get_simulation_time()) or (not lpm_wakeup_using_time and self.system_model.energy_buffer.get_voltage() >= self.system_model.mcu.v_on):
# Exit from LPM (automatically wait for t_wakeup + re-initialize custom devices)
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.LPM_EXIT, OpModeName.LPM_EXIT)
if print_lpm_out:
print(f"LPM out {self.system_model.harvested_energy['total']['total']}")
print_lpm_out = False
# Stop timekeeper, as we are exiting LPM
if self.system_model.timekeeper is not None:
self.system_model.timekeeper.stop_tracking()
# Exit from while loop and continue execution
break
if print_lpm_out:
print(f"Exiting LPM at {self.system_model.energy_buffer.get_voltage()}V")
print_lpm_out = False
if self.use_checkpoints:
# Save state if static placement and a call to a state-save operation is being executed
# Note: the cost of executing the function call is already considered
if self.static_placement and function_name == state_save_function_name:
save_state = True
# set automatic check for power failures
# if self.power_failures_during_state_save is set to False, the analysis considers the state-saving
# operation to always complete, even if in a real-world scenario the device would turn off.
# Note that the additional energy is considered.
self.system_model.set_power_failures_automatic_check(self.power_failures_during_state_save)
# Probe
if self.probe_energy_buffer:
self.system_model.set_custom_signal('probe', True)
sysmodel_ticks += self._simulate_save_restore_call_ret(0, OpModeName.PROBE_ENERGY_BUFFER)
energy_buffer_voltage = self.system_model.energy_buffer.get_voltage()
# Assumption: when voltage < minimum operating voltage for the ADC, the MCU always save the state
if self.probe_with_adc and energy_buffer_voltage <= self.system_model.mcu.adc_min_v:
# Set flag
if not probe_adc_low_voltage_flag:
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.VOLATILE_MEMORY_ACCESS, adc_op_name)
probe_adc_low_voltage_flag = True
# Read flag
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.VOLATILE_MEMORY_ACCESS, adc_op_name)
# Flag not set
if not probe_adc_low_voltage_flag:
# OR operation on flag (flag is false)
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.NO_MEMORY_ACCESS, adc_op_name)
# Probe the energy buffer
sysmodel_ticks += self.system_model.state_retention.execute_probe_energy_buffer()
# Check if state must be saved
save_state = self.system_model.energy_buffer.get_voltage() <= self.v_save_state
if not save_state:
# unset automatic check for power failures
self.system_model.set_power_failures_automatic_check(False)
self.system_model.set_custom_signal('probe', False)
# print(f"Probed{'-forced' if probe_adc_low_voltage_flag else ''}: {save_state} - {self.system_model.energy_buffer.get_voltage()}V vs {self.v_save_state}")
#else:
# print(f"Not Probed: {save_state} - {self.system_model.energy_buffer.get_voltage()}V vs {self.v_save_state}")
# Save state if interrupt-based state-saving and energy buffer voltage lower than the v_save_state threshold
if not self.static_placement and self.system_model.energy_buffer.get_voltage() <= self.v_save_state:
save_state = True
# set automatic check for power failures
# if self.power_failures_during_state_save is set to False, the analysis considers the state-saving
# operation to always complete, even if in a real-world scenario the device would turn off.
# Note that the additional energy is considered.
self.system_model.set_power_failures_automatic_check(self.power_failures_during_state_save)
# Save the state
if save_state:
self.system_model.set_custom_signal('save', True)
#print(f"Going to save the state at {self.system_model.energy_buffer.get_voltage()}")
# get state size
save_only_pc, n_registers = self.state_calc.get_saved_registers(current_instruction)
n_memory_cells = self.state_calc.get_saved_memory_cells(current_instruction)
state_save_ticks = self.system_model.state_retention.execute_save_state(save_only_pc, n_registers, n_memory_cells)
# simulate call/ret + update global clock
self._simulate_save_restore_call_ret(state_save_ticks, OpModeName.STATE_SAVE)
# unset automatic check for power failures
self.system_model.set_power_failures_automatic_check(False)
#print(f"Reached {self.system_model.energy_buffer.get_voltage()}")
if not self.system_model.power_failure_occurred():
self.system_model.set_custom_signal('save', False)
save_id += 1
t = time.time()
self.vm.state_retention.execute_state_save()
# Account for physical memory energy consumption
if self.vm.state_retention.physical_mmu is not None:
self.system_model.run_step(MCUClockCycleAction.PHYSICAL_MEMORY_ACCESS, OpModeName.STATE_SAVE, additional_op_mode_name='STATE_RETENTION_MEMORY')
#print(f"State-save {self.system_model.elapsed_ticks['total']['total']} - saved in {time.time() - t}s")
if self.vm.state_retention.hibernate_after_save:
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.LPM_ENTER, OpModeName.LPM_ENTER)
if self.stdout_enabled:
print(f"Entering LPM at {self.system_model.energy_buffer.get_voltage()}V")
# Start timekeeper
if self.system_model.timekeeper is not None:
self.system_model.timekeeper.start_tracking()
# Simulate hibernation ticks
# - If power failure occurs -> stop (power failure simulated after the else branch)
# - If voltage raises above v_resume -> resume from LPM
i = 0
while True:
if i % 50000 == 0:
if self.stdout_enabled:
print(f"LPM Recharging - Cap={self.system_model.energy_buffer.get_voltage()}V; Harvester={self.system_model.energy_harvester.get_voltage()}V; Source={self.system_model.energy_source.get_voltage()}V; T={round(self.system_model.energy_source.elapsed_time, 3)}s")
if self.max_simulation_time > 0:
if self.system_model.get_simulation_time() >= self.max_simulation_time:
self.vm.set_termination_reason("MAX_SIMULATION_TIME_REACHED", "LPM_RECHARGING")
print(f"MAX SIMULATION TIME REACHED DURING LPM RECHARGING")
break
i += 1
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.LPM_NOP, OpModeName.LPM)
if self.system_model.power_failure_occurring():
power_failure_occurred = True
break
# Check if voltage >= V_resume
if self.system_model.energy_buffer.get_voltage() >= self.v_resume:
# Exit from LPM (automatically wait for v_wakeup + re-initialize custom devices)
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.LPM_EXIT, OpModeName.LPM_EXIT)
if self.stdout_enabled:
print("LPM out")
# Stop timekeeper, as we are exiting from LPM
if self.system_model.timekeeper is not None:
self.system_model.timekeeper.stop_tracking()
# Reset ADC flag if probe set
if self.probe_energy_buffer and self.probe_with_adc:
probe_adc_low_voltage_flag = False
sysmodel_ticks += self.system_model.run_step(MCUClockCycleAction.VOLATILE_MEMORY_ACCESS, adc_op_name)
# Exit from while loop and continue execution
break
if self.stdout_enabled:
print(f"Exiting LPM at {self.system_model.energy_buffer.get_voltage()}V")
self.system_model.set_custom_signal('save', False)
# additional ticks due to wait cycles / state-save / etc
additional_ticks = max(0, sysmodel_ticks - op_ticks)
self.vm.state.global_clock += additional_ticks
# Simulate a power failure
if power_failure_occurred or self.system_model.power_failure_occurring():
if self.stdout_enabled:
print(f"Energy failure: {self.system_model.elapsed_ticks['total']['total']} {self.system_model.energy_buffer.get_voltage()}V")
self.system_model.record_power_failure()
self.mmu_log_energy_failure()
# Start timekeeper (automatically ignored if previously started)
if self.system_model.timekeeper is not None:
self.system_model.timekeeper.start_tracking()
# reset volatile state
self.vm.state.reset()
# On power failure -> full recharge
if self.system_model.energy_source.power_off_full_recharge:
v_target = self.system_model.mcu.v_on
v_supply = self.system_model.energy_source.full_recharge_voltage
self.system_model.execute_full_recharge(MCUClockCycleAction.NOP_OFF_RECHARGE, OpModeName.ENERGY_BUFFER_RECHARGE, v_target, v_supply)
# On power failure -> simulate energy source recharge
else:
i = 0
while self.system_model.energy_buffer.get_voltage() < self.system_model.mcu.v_on:
if i % 1000000 == 0:
if self.stdout_enabled:
print(f"Recharging to {self.system_model.mcu.v_on}V - Cap={self.system_model.energy_buffer.get_voltage()}V; Harvester={self.system_model.energy_harvester.get_voltage()}V; Source={self.system_model.energy_source.get_voltage()}V; T={round(self.system_model.energy_source.elapsed_time, 3)}s")
if self.max_simulation_time > 0:
if self.system_model.get_simulation_time() >= self.max_simulation_time:
self.vm.set_termination_reason("MAX_SIMULATION_TIME_REACHED", "POWER_OFF_RECHARGE")
self.system_model.run_step(MCUClockCycleAction.NOP_OFF_RECHARGE, OpModeName.ENERGY_BUFFER_RECHARGE)
if self.stdout_enabled:
print(f"Recharged to {self.system_model.energy_buffer.get_voltage()}V")
# Stop timekeeper tracking
if self.system_model.timekeeper is not None:
self.system_model.timekeeper.stop_tracking()
# restore saved state
self.vm.state_retention.execute_state_restore()
self.system_model.set_custom_signal('restore', True)
# Detect non-terminating path bugs
if self.use_checkpoints and restore_id == save_id:
# TODO: add a way to check non-termination for task-based systemds
raise Exception("Error: unable to reach next state-saving operation (non-terminating path bug)!")
restore_id = save_id
# re-initialize components (turn on MCU / reset custom devices)
restore_ticks = self.system_model.init()
# run restore
# Note: here we do not check for power failures, as a power failure happening during restore triggers
# the non-terminating path bug check during the subsequent iteration of the simulation.
# In fact, when a power failure occurs during restore, the next iteration executes the instruction and
# re-enters this portion of code. As restore_id == save_id, the simulation stops.
restore_ticks += self.system_model.state_retention.execute_restore_state()
# Account for physical memory energy consumption
if self.vm.state_retention.physical_mmu is not None:
self.system_model.run_step(MCUClockCycleAction.PHYSICAL_MEMORY_ACCESS, OpModeName.STATE_RESTORE, additional_op_mode_name='STATE_RETENTION_MEMORY')
# simulate call/ret + update global clock
n_ticks = self._simulate_save_restore_call_ret(restore_ticks, OpModeName.STATE_RESTORE)
# restore ADC probe flag
probe_adc_low_voltage_flag = False
self.system_model.set_custom_signal('restore', False)
if self.stdout_enabled:
print(f"Restore: {self.system_model.elapsed_ticks['total']['total']} {self.system_model.energy_buffer.get_voltage()}V")
self.system_model._collect_signals(0.0)
def _simulate_save_restore_call_ret(self, n_ticks, op_mode_name):
"""
Simulates the execution of a function call for state-saving and state-restoring operations.
:param n_ticks: number of executed clock cycles to save/restore the state
"""
# state-saving/state-restoring call/ret execute on volatile stack
ticks = CallOperation.tick_count + ReturnOperation.tick_count
memory_ticks = CallOperation.memory_tick_count + ReturnOperation.memory_tick_count
normal_ticks = ticks - memory_ticks
for _ in range(memory_ticks):
n_ticks += self.system_model.run_step(MCUClockCycleAction.VOLATILE_MEMORY_ACCESS, op_mode_name)
for _ in range(normal_ticks):
n_ticks += self.system_model.run_step(MCUClockCycleAction.NO_MEMORY_ACCESS, op_mode_name)
self.vm.state.global_clock += n_ticks
return n_ticks