148 lines
5.0 KiB
Python
148 lines
5.0 KiB
Python
import logging
|
|
|
|
from ScEpTIC.AST.elements.instructions.other_operations import CallOperation
|
|
from ScEpTIC.AST.elements.instructions.termination_instructions import ReturnOperation
|
|
from ScEpTIC.AST.elements.types import BaseType
|
|
|
|
class InterruptsManager:
|
|
"""
|
|
Hardware interrupts handler.
|
|
"""
|
|
|
|
def __init__(self, vmstate):
|
|
self._isr = {}
|
|
self._vmstate = vmstate
|
|
|
|
self._nested_interrupts_enabled = vmstate.config.interrupts.nested_interrupts
|
|
self._enabled = vmstate.config.interrupts.enabled
|
|
self._firing_enabled = True
|
|
self._isr_running = False
|
|
|
|
self._firing_interrupts = []
|
|
self._interrupts_queue_length = vmstate.config.interrupts.interrupts_queue_length
|
|
|
|
def enable(self):
|
|
"""
|
|
Enables interrupts
|
|
"""
|
|
self._enabled = True
|
|
|
|
def disable(self, flush_queue=True):
|
|
"""
|
|
Disables interrupts (_firing_interrupts flushed if required + no queue of firing interrupts)
|
|
"""
|
|
self._enabled = False
|
|
if flush_queue:
|
|
self._firing_interrupts.clear()
|
|
|
|
def enable_firing(self):
|
|
"""
|
|
Enable interrupt firing
|
|
"""
|
|
self._firing_enabled = True
|
|
|
|
def disable_firing(self):
|
|
"""
|
|
Disable interrupt firing (i.e. pause interrupt firing, but collect the interrupt triggered)
|
|
"""
|
|
self._firing_enabled = False
|
|
|
|
def interrupt_firing(self):
|
|
"""
|
|
Returns if an interrupt is firing
|
|
"""
|
|
# firing enabled
|
|
# len(firing interrupt) > 0
|
|
# another interrupt is not running OR nested interrupts enabled
|
|
return self._firing_enabled and len(self._firing_interrupts) > 0 and (not self._isr_running or self._nested_interrupts_enabled)
|
|
|
|
def is_isr_running(self):
|
|
return self._isr_running
|
|
|
|
def trigger_interrupt(self, interrupt_signal):
|
|
"""
|
|
Triggers an interrupt
|
|
"""
|
|
logging.debug(f"Interrupt {interrupt_signal} triggered")
|
|
# Interrupt disabled
|
|
if not self._enabled:
|
|
return
|
|
|
|
if interrupt_signal not in self._isr:
|
|
raise Exception(f'Interrupt signal {interrupt_signal} not registered')
|
|
|
|
# No interrupt signal queue for the same signal
|
|
if interrupt_signal not in self._firing_interrupts and len(self._firing_interrupts) < self._interrupts_queue_length:
|
|
self._firing_interrupts.append(interrupt_signal)
|
|
|
|
def get_firing_interrupt(self, return_isr_call=True):
|
|
"""
|
|
Returns the first interrupt that is firing
|
|
"""
|
|
# Interrupt firing disabled
|
|
if not self._enabled or not self._firing_enabled:
|
|
return None
|
|
|
|
try:
|
|
interrupt_signal = self._firing_interrupts[0]
|
|
except IndexError:
|
|
return None
|
|
|
|
if return_isr_call:
|
|
return self.get_isr_call(interrupt_signal)
|
|
|
|
return interrupt_signal
|
|
|
|
def _isr_call_callback(self, interrupt_signal):
|
|
"""
|
|
Removes the interrupt signal
|
|
"""
|
|
self._isr_running = True
|
|
self._firing_interrupts.remove(interrupt_signal)
|
|
|
|
def _isr_return_callback(self):
|
|
self._isr_running = False
|
|
|
|
def register_isr(self, interrupt_signal, function_name):
|
|
"""
|
|
Sets the ISR associated to a given signal
|
|
"""
|
|
self._isr[interrupt_signal] = f"{self._vmstate.config.program.ir_function_prefix}{function_name}"
|
|
|
|
def get_isr_call(self, interrupt_signal):
|
|
"""
|
|
Returns the call operation to the ISR of the given interrupt_signal
|
|
"""
|
|
if interrupt_signal not in self._isr:
|
|
raise Exception(f'Interrupt signal {interrupt_signal} not registered')
|
|
|
|
isr = self._isr[interrupt_signal]
|
|
# If it's a string, process it
|
|
if isinstance(isr, str):
|
|
isr = self._register_isr(interrupt_signal, isr)
|
|
|
|
return isr
|
|
|
|
def _register_isr(self, interrupt_signal, function_name):
|
|
"""
|
|
Creates a function call to the ISR given an interrupt signal
|
|
"""
|
|
if function_name not in self._vmstate.functions:
|
|
raise Exception(f"ISR {interrupt_signal} cannot be registered: unknown function {function_name}")
|
|
|
|
# Call Operation to be injected + callback to isr_callback
|
|
isr_call = CallOperation(None, function_name, BaseType("int", 32), [], [], {})
|
|
isr_call.custom_callback = lambda: self._isr_call_callback(interrupt_signal)
|
|
isr_call.is_isr_call = True
|
|
self._isr[interrupt_signal] = isr_call
|
|
|
|
# Update return operations to ignore PC update, as interrupts inject a call operation
|
|
for instruction in self._vmstate.functions[function_name].body:
|
|
if isinstance(instruction, ReturnOperation):
|
|
instruction.update_pc = False
|
|
instruction.custom_callback = lambda: self._isr_return_callback()
|
|
instruction.is_isr_return = True
|
|
instruction.isr_call = isr_call
|
|
|
|
return isr_call
|