mirror of
https://github.com/linux-msm/laptops-kernel.git
synced 2026-08-13 14:19:53 -07:00
verification/rvgen: Add support for Hybrid Automata
Add the possibility to parse dot files as hybrid automata and generate the necessary code from rvgen. Hybrid automata are very similar to deterministic ones and most functionality is shared, the dot files include also constraints together with event names (separated by ;) and state names (separated by \n). The tool can now generate the appropriate code to validate constraints at runtime according to the dot specification. Reviewed-by: Nam Cao <namcao@linutronix.de> Link: https://lore.kernel.org/r/20260330111010.153663-5-gmonaco@redhat.com Signed-off-by: Gabriele Monaco <gmonaco@redhat.com>
This commit is contained in:
Gabriele Monaco
@@ -9,7 +9,7 @@
|
||||
# Documentation/trace/rv/da_monitor_synthesis.rst
|
||||
|
||||
if __name__ == '__main__':
|
||||
from rvgen.dot2k import dot2k
|
||||
from rvgen.dot2k import da2k, ha2k
|
||||
from rvgen.generator import Monitor
|
||||
from rvgen.container import Container
|
||||
from rvgen.ltl2k import ltl2k
|
||||
@@ -29,7 +29,7 @@ if __name__ == '__main__':
|
||||
monitor_parser.add_argument("-p", "--parent", dest="parent",
|
||||
required=False, help="Create a monitor nested to parent")
|
||||
monitor_parser.add_argument('-c', "--class", dest="monitor_class",
|
||||
help="Monitor class, either \"da\" or \"ltl\"")
|
||||
help="Monitor class, either \"da\", \"ha\" or \"ltl\"")
|
||||
monitor_parser.add_argument('-s', "--spec", dest="spec", help="Monitor specification file")
|
||||
monitor_parser.add_argument('-t', "--monitor_type", dest="monitor_type",
|
||||
help=f"Available options: {', '.join(Monitor.monitor_types.keys())}")
|
||||
@@ -43,7 +43,9 @@ if __name__ == '__main__':
|
||||
if params.subcmd == "monitor":
|
||||
print("Opening and parsing the specification file %s" % params.spec)
|
||||
if params.monitor_class == "da":
|
||||
monitor = dot2k(params.spec, params.monitor_type, vars(params))
|
||||
monitor = da2k(params.spec, params.monitor_type, vars(params))
|
||||
elif params.monitor_class == "ha":
|
||||
monitor = ha2k(params.spec, params.monitor_type, vars(params))
|
||||
elif params.monitor_class == "ltl":
|
||||
monitor = ltl2k(params.spec, params.monitor_type, vars(params))
|
||||
else:
|
||||
|
||||
@@ -9,24 +9,64 @@
|
||||
# Documentation/trace/rv/deterministic_automata.rst
|
||||
|
||||
import ntpath
|
||||
import re
|
||||
from typing import Iterator
|
||||
|
||||
class _ConstraintKey:
|
||||
"""Base class for constraint keys."""
|
||||
|
||||
class _StateConstraintKey(_ConstraintKey, int):
|
||||
"""Key for a state constraint. Under the hood just state_id."""
|
||||
def __new__(cls, state_id: int):
|
||||
return super().__new__(cls, state_id)
|
||||
|
||||
class _EventConstraintKey(_ConstraintKey, tuple):
|
||||
"""Key for an event constraint. Under the hood just tuple(state_id,event_id)."""
|
||||
def __new__(cls, state_id: int, event_id: int):
|
||||
return super().__new__(cls, (state_id, event_id))
|
||||
|
||||
class Automata:
|
||||
"""Automata class: Reads a dot file and part it as an automata.
|
||||
|
||||
It supports both deterministic and hybrid automata.
|
||||
|
||||
Attributes:
|
||||
dot_file: A dot file with an state_automaton definition.
|
||||
"""
|
||||
|
||||
invalid_state_str = "INVALID_STATE"
|
||||
# val can be numerical, uppercase (constant or macro), lowercase (parameter or function)
|
||||
# only numerical values should have units
|
||||
constraint_rule = re.compile(r"""
|
||||
^
|
||||
(?P<env>[a-zA-Z_][a-zA-Z0-9_]+) # C-like identifier for the env var
|
||||
(?P<op>[!<=>]{1,2}) # operator
|
||||
(?P<val>
|
||||
[0-9]+ | # numerical value
|
||||
[A-Z_]+\(\) | # macro
|
||||
[A-Z_]+ | # constant
|
||||
[a-z_]+\(\) | # function
|
||||
[a-z_]+ # parameter
|
||||
)
|
||||
(?P<unit>[a-z]{1,2})? # optional unit for numerical values
|
||||
""", re.VERBOSE)
|
||||
constraint_reset = re.compile(r"^reset\((?P<env>[a-zA-Z_][a-zA-Z0-9_]+)\)")
|
||||
|
||||
def __init__(self, file_path, model_name=None):
|
||||
self.__dot_path = file_path
|
||||
self.name = model_name or self.__get_model_name()
|
||||
self.__dot_lines = self.__open_dot()
|
||||
self.states, self.initial_state, self.final_states = self.__get_state_variables()
|
||||
self.events = self.__get_event_variables()
|
||||
self.function = self.__create_matrix()
|
||||
self.env_types = {}
|
||||
self.env_stored = set()
|
||||
self.constraint_vars = set()
|
||||
self.self_loop_reset_events = set()
|
||||
self.events, self.envs = self.__get_event_variables()
|
||||
self.function, self.constraints = self.__create_matrix()
|
||||
self.events_start, self.events_start_run = self.__store_init_events()
|
||||
self.env_stored = sorted(self.env_stored)
|
||||
self.constraint_vars = sorted(self.constraint_vars)
|
||||
self.self_loop_reset_events = sorted(self.self_loop_reset_events)
|
||||
|
||||
def __get_model_name(self) -> str:
|
||||
basename = ntpath.basename(self.__dot_path)
|
||||
@@ -116,30 +156,93 @@ class Automata:
|
||||
|
||||
return states, initial_state, final_states
|
||||
|
||||
def __get_event_variables(self) -> list[str]:
|
||||
def __get_event_variables(self) -> tuple[list[str], list[str]]:
|
||||
# here we are at the begin of transitions, take a note, we will return later.
|
||||
cursor = self.__get_cursor_begin_events()
|
||||
|
||||
events = []
|
||||
envs = []
|
||||
while self.__dot_lines[cursor].lstrip()[0] == '"':
|
||||
# transitions have the format:
|
||||
# "all_fired" -> "both_fired" [ label = "disable_irq" ];
|
||||
# ------------ event is here ------------^^^^^
|
||||
if self.__dot_lines[cursor].split()[1] == "->":
|
||||
line = self.__dot_lines[cursor].split()
|
||||
event = "".join(line[line.index("label")+2:-1]).replace('"', '')
|
||||
event = "".join(line[line.index("label") + 2:-1]).replace('"', '')
|
||||
|
||||
# when a transition has more than one lables, they are like this
|
||||
# "local_irq_enable\nhw_local_irq_enable_n"
|
||||
# so split them.
|
||||
|
||||
for i in event.split("\\n"):
|
||||
events.append(i)
|
||||
# if the event contains a constraint (hybrid automata),
|
||||
# it will be separated by a ";":
|
||||
# "sched_switch;x<1000;reset(x)"
|
||||
ev, *constr = i.split(";")
|
||||
if constr:
|
||||
if len(constr) > 2:
|
||||
raise ValueError("Only 1 constraint and 1 reset are supported")
|
||||
envs += self.__extract_env_var(constr)
|
||||
events.append(ev)
|
||||
else:
|
||||
# state labels have the format:
|
||||
# "enable_fired" [label = "enable_fired\ncondition"];
|
||||
# ----- label is here -----^^^^^
|
||||
# label and node name must be the same, condition is optional
|
||||
state = self.__dot_lines[cursor].split("label")[1].split('"')[1]
|
||||
_, *constr = state.split("\\n")
|
||||
if constr:
|
||||
if len(constr) > 1:
|
||||
raise ValueError("Only 1 constraint is supported in the state")
|
||||
envs += self.__extract_env_var([constr[0].replace(" ", "")])
|
||||
cursor += 1
|
||||
|
||||
return sorted(set(events))
|
||||
return sorted(set(events)), sorted(set(envs))
|
||||
|
||||
def __create_matrix(self) -> list[list[str]]:
|
||||
def _split_constraint_expr(self, constr: list[str]) -> Iterator[tuple[str,
|
||||
str | None]]:
|
||||
"""
|
||||
Get a list of strings of the type constr1 && constr2 and returns a list of
|
||||
constraints and separators: [[constr1,"&&"],[constr2,None]]
|
||||
"""
|
||||
exprs = []
|
||||
seps = []
|
||||
for c in constr:
|
||||
while "&&" in c or "||" in c:
|
||||
a = c.find("&&")
|
||||
o = c.find("||")
|
||||
pos = a if o < 0 or 0 < a < o else o
|
||||
exprs.append(c[:pos].replace(" ", ""))
|
||||
seps.append(c[pos:pos + 2].replace(" ", ""))
|
||||
c = c[pos + 2:].replace(" ", "")
|
||||
exprs.append(c)
|
||||
seps.append(None)
|
||||
return zip(exprs, seps)
|
||||
|
||||
def __extract_env_var(self, constraint: list[str]) -> list[str]:
|
||||
env = []
|
||||
for c, _ in self._split_constraint_expr(constraint):
|
||||
rule = self.constraint_rule.search(c)
|
||||
reset = self.constraint_reset.search(c)
|
||||
if rule:
|
||||
env.append(rule["env"])
|
||||
if rule.groupdict().get("unit"):
|
||||
self.env_types[rule["env"]] = rule["unit"]
|
||||
if rule["val"][0].isalpha():
|
||||
self.constraint_vars.add(rule["val"])
|
||||
# try to infer unit from constants or parameters
|
||||
val_for_unit = rule["val"].lower().replace("()", "")
|
||||
if val_for_unit.endswith("_ns"):
|
||||
self.env_types[rule["env"]] = "ns"
|
||||
if val_for_unit.endswith("_jiffies"):
|
||||
self.env_types[rule["env"]] = "j"
|
||||
if reset:
|
||||
env.append(reset["env"])
|
||||
# environment variables that are reset need a storage
|
||||
self.env_stored.add(reset["env"])
|
||||
return env
|
||||
|
||||
def __create_matrix(self) -> tuple[list[list[str]], dict[_ConstraintKey, list[str]]]:
|
||||
# transform the array into a dictionary
|
||||
events = self.events
|
||||
states = self.states
|
||||
@@ -157,6 +260,7 @@ class Automata:
|
||||
|
||||
# declare the matrix....
|
||||
matrix = [[ self.invalid_state_str for x in range(nr_event)] for y in range(nr_state)]
|
||||
constraints: dict[_ConstraintKey, list[str]] = {}
|
||||
|
||||
# and we are back! Let's fill the matrix
|
||||
cursor = self.__get_cursor_begin_events()
|
||||
@@ -166,12 +270,24 @@ class Automata:
|
||||
line = self.__dot_lines[cursor].split()
|
||||
origin_state = line[0].replace('"','').replace(',','_')
|
||||
dest_state = line[2].replace('"','').replace(',','_')
|
||||
possible_events = "".join(line[line.index("label")+2:-1]).replace('"', '')
|
||||
possible_events = "".join(line[line.index("label") + 2:-1]).replace('"', '')
|
||||
for event in possible_events.split("\\n"):
|
||||
event, *constr = event.split(";")
|
||||
if constr:
|
||||
key = _EventConstraintKey(states_dict[origin_state], events_dict[event])
|
||||
constraints[key] = constr
|
||||
# those events reset also on self loops
|
||||
if origin_state == dest_state and "reset" in "".join(constr):
|
||||
self.self_loop_reset_events.add(event)
|
||||
matrix[states_dict[origin_state]][events_dict[event]] = dest_state
|
||||
else:
|
||||
state = self.__dot_lines[cursor].split("label")[1].split('"')[1]
|
||||
state, *constr = state.replace(" ", "").split("\\n")
|
||||
if constr:
|
||||
constraints[_StateConstraintKey(states_dict[state])] = constr
|
||||
cursor += 1
|
||||
|
||||
return matrix
|
||||
return matrix, constraints
|
||||
|
||||
def __store_init_events(self) -> tuple[list[bool], list[bool]]:
|
||||
events_start = [False] * len(self.events)
|
||||
@@ -203,3 +319,13 @@ class Automata:
|
||||
if any(self.events_start):
|
||||
return False
|
||||
return self.events_start_run[self.events.index(event)]
|
||||
|
||||
def is_hybrid_automata(self) -> bool:
|
||||
return bool(self.envs)
|
||||
|
||||
def is_event_constraint(self, key: _ConstraintKey) -> bool:
|
||||
"""
|
||||
Given the key in self.constraints return true if it is an event
|
||||
constraint, false if it is a state constraint
|
||||
"""
|
||||
return isinstance(key, _EventConstraintKey)
|
||||
|
||||
@@ -19,6 +19,7 @@ class Dot2c(Automata):
|
||||
enum_suffix = ""
|
||||
enum_states_def = "states"
|
||||
enum_events_def = "events"
|
||||
enum_envs_def = "envs"
|
||||
struct_automaton_def = "automaton"
|
||||
var_automaton_def = "aut"
|
||||
|
||||
@@ -61,6 +62,37 @@ class Dot2c(Automata):
|
||||
|
||||
return buff
|
||||
|
||||
def __get_non_stored_envs(self) -> list[str]:
|
||||
return [e for e in self.envs if e not in self.env_stored]
|
||||
|
||||
def __get_enum_envs_content(self) -> list[str]:
|
||||
buff = []
|
||||
# We first place env variables that have a u64 storage.
|
||||
# Those are limited by MAX_HA_ENV_LEN, other variables
|
||||
# are read only and don't require a storage.
|
||||
unstored = self.__get_non_stored_envs()
|
||||
for env in list(self.env_stored) + unstored:
|
||||
buff.append(f"\t{env}{self.enum_suffix},")
|
||||
|
||||
buff.append(f"\tenv_max{self.enum_suffix},")
|
||||
max_stored = unstored[0] if len(unstored) else "env_max"
|
||||
buff.append(f"\tenv_max_stored{self.enum_suffix} = {max_stored}{self.enum_suffix},")
|
||||
|
||||
return buff
|
||||
|
||||
def format_envs_enum(self) -> list[str]:
|
||||
buff = []
|
||||
if self.is_hybrid_automata():
|
||||
buff.append(f"enum {self.enum_envs_def} {{")
|
||||
buff += self.__get_enum_envs_content()
|
||||
buff.append("};\n")
|
||||
buff.append(f"_Static_assert(env_max_stored{self.enum_suffix} <= MAX_HA_ENV_LEN,"
|
||||
' "Not enough slots");')
|
||||
if {"ns", "us", "ms", "s"}.intersection(self.env_types.values()):
|
||||
buff.append("#define HA_CLK_NS")
|
||||
buff.append("")
|
||||
return buff
|
||||
|
||||
def get_minimun_type(self) -> str:
|
||||
min_type = "unsigned char"
|
||||
|
||||
@@ -81,6 +113,8 @@ class Dot2c(Automata):
|
||||
buff.append("struct %s {" % self.struct_automaton_def)
|
||||
buff.append("\tchar *state_names[state_max%s];" % (self.enum_suffix))
|
||||
buff.append("\tchar *event_names[event_max%s];" % (self.enum_suffix))
|
||||
if self.is_hybrid_automata():
|
||||
buff.append(f"\tchar *env_names[env_max{self.enum_suffix}];")
|
||||
buff.append("\t%s function[state_max%s][event_max%s];" % (min_type, self.enum_suffix, self.enum_suffix))
|
||||
buff.append("\t%s initial_state;" % min_type)
|
||||
buff.append("\tbool final_states[state_max%s];" % (self.enum_suffix))
|
||||
@@ -113,6 +147,17 @@ class Dot2c(Automata):
|
||||
|
||||
return buff
|
||||
|
||||
def format_aut_init_envs_string(self) -> list[str]:
|
||||
buff = []
|
||||
if self.is_hybrid_automata():
|
||||
buff.append("\t.env_names = {")
|
||||
# maintain consistent order with the enum
|
||||
ordered_envs = list(self.env_stored) + self.__get_non_stored_envs()
|
||||
buff.append(self.__get_string_vector_per_line_content(ordered_envs))
|
||||
buff.append("\t},")
|
||||
|
||||
return buff
|
||||
|
||||
def __get_max_strlen_of_states(self) -> int:
|
||||
max_state_name = max(self.states, key = len).__len__()
|
||||
return max(max_state_name, self.invalid_state_str.__len__())
|
||||
@@ -205,10 +250,12 @@ class Dot2c(Automata):
|
||||
buff += self.format_states_enum()
|
||||
buff += self.format_invalid_state()
|
||||
buff += self.format_events_enum()
|
||||
buff += self.format_envs_enum()
|
||||
buff += self.format_automaton_definition()
|
||||
buff += self.format_aut_init_header()
|
||||
buff += self.format_aut_init_states_string()
|
||||
buff += self.format_aut_init_events_string()
|
||||
buff += self.format_aut_init_envs_string()
|
||||
buff += self.format_aut_init_function()
|
||||
buff += self.format_aut_init_initial_state()
|
||||
buff += self.format_aut_init_final_states()
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -255,12 +255,14 @@ class Monitor(RVGenerator):
|
||||
monitor_class_type = self.fill_monitor_class_type()
|
||||
tracepoint_args_skel_event = self.fill_tracepoint_args_skel("event")
|
||||
tracepoint_args_skel_error = self.fill_tracepoint_args_skel("error")
|
||||
tracepoint_args_skel_error_env = self.fill_tracepoint_args_skel("error_env")
|
||||
trace_h = trace_h.replace("%%MODEL_NAME%%", self.name)
|
||||
trace_h = trace_h.replace("%%MODEL_NAME_UP%%", self.name.upper())
|
||||
trace_h = trace_h.replace("%%MONITOR_CLASS%%", monitor_class)
|
||||
trace_h = trace_h.replace("%%MONITOR_CLASS_TYPE%%", monitor_class_type)
|
||||
trace_h = trace_h.replace("%%TRACEPOINT_ARGS_SKEL_EVENT%%", tracepoint_args_skel_event)
|
||||
trace_h = trace_h.replace("%%TRACEPOINT_ARGS_SKEL_ERROR%%", tracepoint_args_skel_error)
|
||||
trace_h = trace_h.replace("%%TRACEPOINT_ARGS_SKEL_ERROR_ENV%%", tracepoint_args_skel_error_env)
|
||||
return trace_h
|
||||
|
||||
def print_files(self):
|
||||
|
||||
@@ -21,7 +21,7 @@
|
||||
*/
|
||||
#define RV_MON_TYPE RV_MON_%%MONITOR_TYPE%%
|
||||
#include "%%MODEL_NAME%%.h"
|
||||
#include <rv/da_monitor.h>
|
||||
#include <rv/%%MONITOR_CLASS%%_monitor.h>
|
||||
|
||||
/*
|
||||
* This is the instrumentation part of the monitor.
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
/* SPDX-License-Identifier: GPL-2.0 */
|
||||
|
||||
/*
|
||||
* Snippet to be included in rv_trace.h
|
||||
*/
|
||||
|
||||
#ifdef CONFIG_RV_MON_%%MODEL_NAME_UP%%
|
||||
DEFINE_EVENT(event_%%MONITOR_CLASS%%, event_%%MODEL_NAME%%,
|
||||
%%TRACEPOINT_ARGS_SKEL_EVENT%%);
|
||||
|
||||
DEFINE_EVENT(error_%%MONITOR_CLASS%%, error_%%MODEL_NAME%%,
|
||||
%%TRACEPOINT_ARGS_SKEL_ERROR%%);
|
||||
|
||||
DEFINE_EVENT(error_env_%%MONITOR_CLASS%%, error_env_%%MODEL_NAME%%,
|
||||
%%TRACEPOINT_ARGS_SKEL_ERROR_ENV%%);
|
||||
#endif /* CONFIG_RV_MON_%%MODEL_NAME_UP%% */
|
||||
Reference in New Issue
Block a user