summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorLuke T. Shumaker <lukeshu@lukeshu.com>2025-02-28 18:48:43 -0700
committerLuke T. Shumaker <lukeshu@lukeshu.com>2025-02-28 19:39:29 -0700
commit3d072046d88df970da1938ec2e5f17ce483f1c10 (patch)
treeb14225829ebeb2140d6c07862ad3837e8b906b24
parent58991986652e0a08f919cb30f5302bb1b977ef32 (diff)
stack.c.gen: Refactor to a plugin architecture
-rwxr-xr-xbuild-aux/stack.c.gen699
1 files changed, 423 insertions, 276 deletions
diff --git a/build-aux/stack.c.gen b/build-aux/stack.c.gen
index 968c760..bdac124 100755
--- a/build-aux/stack.c.gen
+++ b/build-aux/stack.c.gen
@@ -143,14 +143,18 @@ class AnalyzeResult(typing.NamedTuple):
dynamic: set[str]
+class Application(typing.Protocol):
+ def extra_nodes(self) -> typing.Collection[Node]: ...
+ def location_xform(self, loc: str) -> str: ...
+ def indirect_callees(self, elem: VCGElem) -> tuple[list[str], bool]: ...
+ def skip_call(self, chain: list[str], funcname: str) -> bool: ...
+
+
def analyze(
*,
ci_fnames: typing.Collection[str],
- extra_nodes: list[Node] = [],
app_func_filters: dict[str, typing.Callable[[str], int]],
- app_location_xform: typing.Callable[[str], str],
- app_indirect_callees: typing.Callable[[VCGElem], tuple[list[str], bool]],
- app_skip_call: typing.Callable[[list[str], str], bool],
+ app: Application,
cfg_max_call_depth: int,
) -> AnalyzeResult:
re_node_label = re.compile(
@@ -221,7 +225,7 @@ def analyze(
if caller not in graph:
raise ValueError(f"unknown caller: {caller}")
if callee == "__indirect_call":
- callees, missing_ok = app_indirect_callees(elem)
+ callees, missing_ok = app.indirect_callees(elem)
for callee in callees:
if callee not in graph[caller].calls:
graph[caller].calls[callee] = missing_ok
@@ -235,7 +239,7 @@ def analyze(
for elem in parse_vcg(fh):
handle_elem(elem)
- for node in extra_nodes:
+ for node in app.extra_nodes():
if node.funcname in graph:
raise ValueError(f"duplicate node {repr(node.funcname)}")
graph[node.funcname] = node
@@ -268,12 +272,12 @@ def analyze(
nonlocal dbg
funcname = resolve_funcname(orig_funcname)
if not funcname:
- if app_skip_call(chain, orig_funcname):
+ if app.skip_call(chain, orig_funcname):
return 0
if not missing_ok:
missing.add(orig_funcname)
return 0
- if app_skip_call(chain, funcname):
+ if app.skip_call(chain, funcname):
return 0
if len(chain) == cfg_max_call_depth:
@@ -283,7 +287,7 @@ def analyze(
if dbg:
print(f"//dbg: {funcname}\t{node.nstatic}")
if node.usage_kind == "dynamic" or node.ndynamic > 0:
- dynamic.add(app_location_xform(funcname))
+ dynamic.add(app.location_xform(funcname))
return node.nstatic + max(
[
0,
@@ -302,7 +306,7 @@ def analyze(
for funcname in graph:
if cnt := grp_filter(funcname):
n = nstatic(funcname)
- rows[app_location_xform(funcname)] = n
+ rows[app.location_xform(funcname)] = n
if n > nmax:
nmax = n
nsum += cnt * n
@@ -312,12 +316,11 @@ def analyze(
################################################################################
-# Application-specific code
-
-re_location = re.compile(r"(?P<filename>.+):(?P<row>[0-9]+):(?P<col>[0-9]+)")
+# Mildly-application-specific code
def read_source(location: str) -> str:
+ re_location = re.compile(r"(?P<filename>.+):(?P<row>[0-9]+):(?P<col>[0-9]+)")
m = re_location.fullmatch(location)
if not m:
raise ValueError(f"unexpected label value {repr(location)}")
@@ -338,165 +341,265 @@ def get_zero_or_one(
return None
-def main(
- *,
- arg_pico_platform: str,
- arg_base_dir: str,
- arg_ci_fnames: typing.Collection[str],
- arg_c_fnames: typing.Collection[str],
-) -> None:
+re_call_other = re.compile(r"(?P<func>[^(]+)\(.*")
- re_call_other = re.compile(r"(?P<func>[^(]+)\(.*")
- all_nodes: list[Node] = []
- hooks_is_intrhandler: list[typing.Callable[[str], bool]] = []
- hooks_indirect_callees: list[
- typing.Callable[[str, str], tuple[list[str], bool] | None]
- ] = []
- hooks_skip_call: list[typing.Callable[[list[str], str], bool]] = []
+class Plugin(typing.Protocol):
+ def extra_nodes(self) -> typing.Collection[Node]: ...
+ def indirect_callees(
+ self, loc: str, line: str
+ ) -> tuple[list[str], bool] | None: ...
+ def skip_call(self, chain: list[str], call: str) -> bool: ...
- # The sbc-harness codebase #######################################
- def _is_config_h(fname: str) -> bool:
- if not fname.startswith(arg_base_dir + "/"):
- return False
- suffix = fname[len(arg_base_dir) + 1 :]
- if suffix.startswith("3rd-party/"):
- return False
- return suffix.endswith("/config.h")
+class PluginApplication:
+ _location_xform: typing.Callable[[str], str]
+ _plugins: list[Plugin]
- config_h_fname = get_zero_or_one(_is_config_h, arg_c_fnames)
+ def __init__(
+ self, location_xform: typing.Callable[[str], str], plugins: list[Plugin]
+ ) -> None:
+ self._location_xform = location_xform
+ self._plugins = plugins
- lib9p_srv_c_fname = get_zero_or_one(
- lambda fname: fname.endswith("lib9p/srv.c"), arg_c_fnames
- )
+ def extra_nodes(self) -> typing.Collection[Node]:
+ ret: list[Node] = []
+ for plugin in self._plugins:
+ ret.extend(plugin.extra_nodes())
+ return ret
- lib9p_generated_c_fname = get_zero_or_one(
- lambda fname: fname.endswith("lib9p/9p.generated.c"), arg_c_fnames
- )
+ def location_xform(self, loc: str) -> str:
+ return self._location_xform(loc)
- def config_h_get(varname: str) -> int | None:
- if config_h_fname:
- with open(config_h_fname, "r") as fh:
- for line in fh:
- line = line.rstrip()
- if line.startswith("#define"):
- parts = line.split()
- if parts[1] == varname:
- return int(parts[2])
- return None
+ def indirect_callees(self, elem: VCGElem) -> tuple[list[str], bool]:
+ loc = elem.attrs.get("label", "")
+ line = read_source(loc)
- objcalls: dict[str, set[str]] = {}
- re_vtable_start = re.compile(r"_vtable\s*=\s*\{")
- re_vtable_entry = re.compile(r"^\s+\.(?P<meth>\S+)\s*=\s*(?P<impl>\S+),.*")
- for fname in arg_c_fnames:
- with open(fname, "r") as fh:
- in_vtable = False
- for line in fh:
- line = line.rstrip()
- if in_vtable:
- if m := re_vtable_entry.fullmatch(line):
- meth = m.group("meth")
- impl = m.group("impl")
- if impl == "NULL":
- continue
- if m.group("meth") not in objcalls:
- objcalls[meth] = set()
- objcalls[meth].add(impl)
- if "}" in line:
- in_vtable = False
- elif re_vtable_start.search(line):
- in_vtable = True
-
- tmessage_handlers: set[str] | None = None
- if lib9p_srv_c_fname:
- re_tmessage_handler = re.compile(
- r"^\s*\[LIB9P_TYP_T[^]]+\]\s*=\s*\(tmessage_handler\)\s*(?P<handler>\S+),\s*$"
- )
- tmessage_handlers = set()
- with open(lib9p_srv_c_fname, "r") as fh:
- for line in fh:
- line = line.rstrip()
- if m := re_tmessage_handler.fullmatch(line):
- tmessage_handlers.add(m.group("handler"))
+ for plugin in self._plugins:
+ ret = plugin.indirect_callees(loc, line)
+ if ret is not None:
+ return ret
- lib9p_msgs: set[str] = set()
- if lib9p_generated_c_fname:
- re_lib9p_msg_entry = re.compile(r"^\s*_MSG_(?:[A-Z]+)\((?P<typ>\S+)\),$")
- with open(lib9p_generated_c_fname, "r") as fh:
- for line in fh:
- line = line.rstrip()
- if m := re_lib9p_msg_entry.fullmatch(line):
- typ = m.group("typ")
- lib9p_msgs.add(typ)
+ placeholder = "__indirect_call"
+ if m := re_call_other.fullmatch(line):
+ placeholder += ":" + m.group("func")
+ placeholder += " at " + self.location_xform(elem.attrs.get("label", ""))
+ return [placeholder], False
+
+ def skip_call(self, chain: list[str], funcname: str) -> bool:
+ for plugin in self._plugins:
+ if plugin.skip_call(chain, funcname):
+ return True
+ return False
+
+
+################################################################################
+# Application-specific code
- re_call_objcall = re.compile(r"LO_CALL\((?P<obj>[^,]+), (?P<meth>[^,)]+)[,)].*")
- def sbc_indirect_callees(loc: str, line: str) -> tuple[list[str], bool] | None:
+class LibObjPlugin:
+ objcalls: dict[str, set[str]]
+
+ def __init__(self, arg_c_fnames: typing.Collection[str]) -> None:
+ objcalls: dict[str, set[str]] = {}
+ re_vtable_start = re.compile(r"_vtable\s*=\s*\{")
+ re_vtable_entry = re.compile(r"^\s+\.(?P<meth>\S+)\s*=\s*(?P<impl>\S+),.*")
+ for fname in arg_c_fnames:
+ with open(fname, "r") as fh:
+ in_vtable = False
+ for line in fh:
+ line = line.rstrip()
+ if in_vtable:
+ if m := re_vtable_entry.fullmatch(line):
+ meth = m.group("meth")
+ impl = m.group("impl")
+ if impl == "NULL":
+ continue
+ if m.group("meth") not in objcalls:
+ objcalls[meth] = set()
+ objcalls[meth].add(impl)
+ if "}" in line:
+ in_vtable = False
+ elif re_vtable_start.search(line):
+ in_vtable = True
+ self.objcalls = objcalls
+
+ def extra_nodes(self) -> typing.Collection[Node]:
+ return []
+
+ def indirect_callees(self, loc: str, line: str) -> tuple[list[str], bool] | None:
+ re_call_objcall = re.compile(r"LO_CALL\((?P<obj>[^,]+), (?P<meth>[^,)]+)[,)].*")
+
if "/3rd-party/" in loc:
return None
if m := re_call_objcall.fullmatch(line):
- if m.group("meth") in objcalls:
- return sorted(objcalls[m.group("meth")]), True
+ if m.group("meth") in self.objcalls:
+ return sorted(self.objcalls[m.group("meth")]), True
return [
f"__indirect_call:{m.group('obj')}.vtable->{m.group('meth')}"
], False
+ return None
+
+ def skip_call(self, chain: list[str], call: str) -> bool:
+ return False
+
+
+class LibHWPlugin:
+ def extra_nodes(self) -> typing.Collection[Node]:
+ return []
+
+ def indirect_callees(self, loc: str, line: str) -> tuple[list[str], bool] | None:
+ if "/3rd-party/" in loc:
+ return None
if "trigger->cb(trigger->cb_arg)" in line:
return [
"alarmclock_sleep_intrhandler",
"w5500_tcp_alarm_handler",
"w5500_udp_alarm_handler",
], True
+ return None
+
+ def skip_call(self, chain: list[str], call: str) -> bool:
+ return False
+
+
+class LibCRIPCPlugin:
+ def extra_nodes(self) -> typing.Collection[Node]:
+ return []
+
+ def indirect_callees(self, loc: str, line: str) -> tuple[list[str], bool] | None:
+ if "/3rd-party/" in loc:
+ return None
if "/chan.h:" in loc and "front->dequeue(" in line:
return [
"_cr_chan_dequeue",
"_cr_select_dequeue",
], True
- if tmessage_handlers and "/srv.c:" in loc and "tmessage_handlers[typ](" in line:
+ return None
+
+ def skip_call(self, chain: list[str], call: str) -> bool:
+ return False
+
+
+class Lib9PPlugin:
+ tmessage_handlers: set[str] | None
+ lib9p_msgs: set[str]
+ _CONFIG_9P_NUM_SOCKS: int | None
+ CONFIG_9P_SRV_MAX_REQS: int | None
+ CONFIG_9P_SRV_MAX_DEPTH: int | None
+
+ def __init__(self, arg_base_dir: str, arg_c_fnames: typing.Collection[str]) -> None:
+ # Find filenames #######################################################
+
+ def _is_config_h(fname: str) -> bool:
+ if not fname.startswith(arg_base_dir + "/"):
+ return False
+ suffix = fname[len(arg_base_dir) + 1 :]
+ if suffix.startswith("3rd-party/"):
+ return False
+ return suffix.endswith("/config.h")
+
+ config_h_fname = get_zero_or_one(_is_config_h, arg_c_fnames)
+
+ lib9p_srv_c_fname = get_zero_or_one(
+ lambda fname: fname.endswith("lib9p/srv.c"), arg_c_fnames
+ )
+
+ lib9p_generated_c_fname = get_zero_or_one(
+ lambda fname: fname.endswith("lib9p/9p.generated.c"), arg_c_fnames
+ )
+
+ # Read config ##########################################################
+
+ def config_h_get(varname: str) -> int | None:
+ if config_h_fname:
+ with open(config_h_fname, "r") as fh:
+ for line in fh:
+ line = line.rstrip()
+ if line.startswith("#define"):
+ parts = line.split()
+ if parts[1] == varname:
+ return int(parts[2])
+ return None
+
+ self._CONFIG_9P_NUM_SOCKS = config_h_get("_CONFIG_9P_NUM_SOCKS")
+ self.CONFIG_9P_SRV_MAX_REQS = config_h_get("CONFIG_9P_SRV_MAX_REQS")
+ self.CONFIG_9P_SRV_MAX_DEPTH = config_h_get("CONFIG_9P_SRV_MAX_DEPTH")
+
+ # Read sources #########################################################
+
+ tmessage_handlers: set[str] | None = None
+ if lib9p_srv_c_fname:
+ re_tmessage_handler = re.compile(
+ r"^\s*\[LIB9P_TYP_T[^]]+\]\s*=\s*\(tmessage_handler\)\s*(?P<handler>\S+),\s*$"
+ )
+ tmessage_handlers = set()
+ with open(lib9p_srv_c_fname, "r") as fh:
+ for line in fh:
+ line = line.rstrip()
+ if m := re_tmessage_handler.fullmatch(line):
+ tmessage_handlers.add(m.group("handler"))
+ self.tmessage_handlers = tmessage_handlers
+
+ lib9p_msgs: set[str] = set()
+ if lib9p_generated_c_fname:
+ re_lib9p_msg_entry = re.compile(r"^\s*_MSG_(?:[A-Z]+)\((?P<typ>\S+)\),$")
+ with open(lib9p_generated_c_fname, "r") as fh:
+ for line in fh:
+ line = line.rstrip()
+ if m := re_lib9p_msg_entry.fullmatch(line):
+ typ = m.group("typ")
+ lib9p_msgs.add(typ)
+ self.lib9p_msgs = lib9p_msgs
+
+ def thread_count(self, name: str) -> int:
+ assert self._CONFIG_9P_NUM_SOCKS
+ assert self.CONFIG_9P_SRV_MAX_REQS
+ if "read" in name:
+ return self._CONFIG_9P_NUM_SOCKS
+ elif "write" in name:
+ return self._CONFIG_9P_NUM_SOCKS * self.CONFIG_9P_SRV_MAX_REQS
+ return 1
+
+ def extra_nodes(self) -> typing.Collection[Node]:
+ return []
+
+ def indirect_callees(self, loc: str, line: str) -> tuple[list[str], bool] | None:
+ if "/3rd-party/" in loc:
+ return None
+ if (
+ self.tmessage_handlers
+ and "/srv.c:" in loc
+ and "tmessage_handlers[typ](" in line
+ ):
# Functions for disabled protocol extensions will be missing.
- return sorted(tmessage_handlers), True
- if lib9p_msgs and "/9p.c:" in loc:
+ return sorted(self.tmessage_handlers), True
+ if self.lib9p_msgs and "/9p.c:" in loc:
for meth in ["validate", "unmarshal", "marshal"]:
if line.startswith(f"tentry.{meth}("):
# Functions for disabled protocol extensions will be missing.
- return sorted(f"{meth}_{msg}" for msg in lib9p_msgs), True
+ return sorted(f"{meth}_{msg}" for msg in self.lib9p_msgs), True
return None
- hooks_indirect_callees += [sbc_indirect_callees]
-
- _CONFIG_9P_NUM_SOCKS = config_h_get("_CONFIG_9P_NUM_SOCKS")
- CONFIG_9P_SRV_MAX_REQS = config_h_get("CONFIG_9P_SRV_MAX_REQS")
- CONFIG_9P_SRV_MAX_DEPTH = config_h_get("CONFIG_9P_SRV_MAX_DEPTH")
-
- def sbc_is_thread(name: str) -> int:
- if name.endswith("_cr") and name != "lib9p_srv_read_cr":
- if "9p" in name:
- assert _CONFIG_9P_NUM_SOCKS
- assert CONFIG_9P_SRV_MAX_REQS
- if "read" in name:
- return _CONFIG_9P_NUM_SOCKS
- elif "write" in name:
- return _CONFIG_9P_NUM_SOCKS * CONFIG_9P_SRV_MAX_REQS
- return 1
- if name == "main":
- return True
+ def skip_call(self, chain: list[str], call: str) -> bool:
+ if "lib9p/srv.c:srv_util_pathfree" in call:
+ assert isinstance(self.CONFIG_9P_SRV_MAX_DEPTH, int)
+ if len(chain) >= self.CONFIG_9P_SRV_MAX_DEPTH and all(
+ ("lib9p/srv.c:srv_util_pathfree" in c)
+ for c in chain[-self.CONFIG_9P_SRV_MAX_DEPTH :]
+ ):
+ return True
return False
- def sbc_is_intrhandler(name: str) -> bool:
- return name in [
- "rp2040_hwtimer_intrhandler",
- "_cr_gdb_intrhandler",
- "hostclock_handle_sig_alarm",
- "hostnet_handle_sig_io",
- ]
- hooks_is_intrhandler += [sbc_is_intrhandler]
+class LibMiscPlugin:
+ def extra_nodes(self) -> typing.Collection[Node]:
+ return []
- sbc_gpio_handlers = [
- "w5500_intrhandler",
- ]
+ def indirect_callees(self, loc: str, line: str) -> tuple[list[str], bool] | None:
+ return None
- def sbc_skip_call(chain: list[str], call: str) -> bool:
+ def skip_call(self, chain: list[str], call: str) -> bool:
if (
len(chain) > 1
and chain[-1] == "__assert_msg_fail"
@@ -504,101 +607,85 @@ def main(
and "__assert_msg_fail" in chain[:-1]
):
return True
- if "lib9p/srv.c:srv_util_pathfree" in call:
- assert isinstance(CONFIG_9P_SRV_MAX_DEPTH, int)
- if len(chain) >= CONFIG_9P_SRV_MAX_DEPTH and all(
- ("lib9p/srv.c:srv_util_pathfree" in c)
- for c in chain[-CONFIG_9P_SRV_MAX_DEPTH:]
- ):
- return True
return False
- hooks_skip_call += [sbc_skip_call]
-
- # pico-sdk #######################################################
-
- if arg_pico_platform == "rp2040":
- def pico_is_intrhandler(name: str) -> bool:
- return name in [
- "gpio_default_irq_handler",
- ]
+class PicoSDKPlugin:
+ app_gpio_handlers: typing.Collection[str]
- hooks_is_intrhandler += [pico_is_intrhandler]
+ def __init__(self, app_gpio_handlers: typing.Collection[str]) -> None:
+ self.app_gpio_handlers = app_gpio_handlers
- def pico_indirect_callees(loc: str, line: str) -> tuple[list[str], bool] | None:
- if "/3rd-party/pico-sdk/" not in loc or "/3rd-party/pico-sdk/lib/" in loc:
- return None
- m = re_call_other.fullmatch(line)
- call: str | None = m.group("func") if m else None
-
- match call:
- case "connect_internal_flash_func":
- return ["rom_func_lookup(ROM_FUNC_CONNECT_INTERNAL_FLASH)"], False
- case "flash_exit_xip_func":
- return ["rom_func_lookup(ROM_FUNC_FLASH_EXIT_XIP)"], False
- case "flash_range_erase_func":
- return ["rom_func_lookup(ROM_FUNC_FLASH_RANGE_ERASE)"], False
- case "flash_flush_cache_func":
- return ["rom_func_lookup(ROM_FUNC_FLASH_FLUSH_CACHE)"], False
- case "rom_table_lookup":
- return ["rom_hword_as_ptr(BOOTROM_TABLE_LOOKUP_OFFSET)"], False
- if "/flash.c:" in loc and "boot2_copyout" in line:
- return ["_stage2_boot"], False
- if "/gpio.c:" in loc and call == "callback":
- return sbc_gpio_handlers, True
- if "/printf.c:" in loc:
- if call == "out":
- return [
- "_out_buffer",
- "_out_null",
- "_out_fct",
- ], False
- if "->fct(" in line:
- return ["stdio_buffered_printer"], False
- if "/stdio.c:" in loc:
- if call == "out_func":
- return [
- "stdio_out_chars_crlf",
- "stdio_out_chars_no_crlf",
- ], False
- if call and (call.startswith("d->") or call.startswith("driver->")):
- _, meth = call.split("->", 1)
- match meth:
- case "out_chars":
- return ["stdio_uart_out_chars"], False
- case "out_flush":
- return ["stdio_uart_out_flush"], False
- case "in_chars":
- return ["stdio_uart_in_chars"], False
+ def indirect_callees(self, loc: str, line: str) -> tuple[list[str], bool] | None:
+ if "/3rd-party/pico-sdk/" not in loc or "/3rd-party/pico-sdk/lib/" in loc:
return None
+ m = re_call_other.fullmatch(line)
+ call: str | None = m.group("func") if m else None
+
+ match call:
+ case "connect_internal_flash_func":
+ return ["rom_func_lookup(ROM_FUNC_CONNECT_INTERNAL_FLASH)"], False
+ case "flash_exit_xip_func":
+ return ["rom_func_lookup(ROM_FUNC_FLASH_EXIT_XIP)"], False
+ case "flash_range_erase_func":
+ return ["rom_func_lookup(ROM_FUNC_FLASH_RANGE_ERASE)"], False
+ case "flash_flush_cache_func":
+ return ["rom_func_lookup(ROM_FUNC_FLASH_FLUSH_CACHE)"], False
+ case "rom_table_lookup":
+ return ["rom_hword_as_ptr(BOOTROM_TABLE_LOOKUP_OFFSET)"], False
+ if "/flash.c:" in loc and "boot2_copyout" in line:
+ return ["_stage2_boot"], False
+ if "/gpio.c:" in loc and call == "callback":
+ return sorted(self.app_gpio_handlers), True
+ if "/printf.c:" in loc:
+ if call == "out":
+ return [
+ "_out_buffer",
+ "_out_null",
+ "_out_fct",
+ ], False
+ if "->fct(" in line:
+ return ["stdio_buffered_printer"], False
+ if "/stdio.c:" in loc:
+ if call == "out_func":
+ return [
+ "stdio_out_chars_crlf",
+ "stdio_out_chars_no_crlf",
+ ], False
+ if call and (call.startswith("d->") or call.startswith("driver->")):
+ _, meth = call.split("->", 1)
+ match meth:
+ case "out_chars":
+ return ["stdio_uart_out_chars"], False
+ case "out_flush":
+ return ["stdio_uart_out_flush"], False
+ case "in_chars":
+ return ["stdio_uart_in_chars"], False
+ return None
- hooks_indirect_callees += [pico_indirect_callees]
-
- def pico_skip_call(chain: list[str], call: str) -> bool:
- if call == "_out_buffer" or call == "_out_fct":
- last = ""
- for pcall in chain:
- if pcall in [
- "__wrap_sprintf",
- "__wrap_snprintf",
- "__wrap_vsnprintf",
- "vfctprintf",
- ]:
- last = pcall
- if last == "vfctprintf":
- return call != "_out_fct"
- else:
- return call == "_out_buffer"
- return False
-
- hooks_skip_call += [pico_skip_call]
+ def skip_call(self, chain: list[str], call: str) -> bool:
+ if call == "_out_buffer" or call == "_out_fct":
+ last = ""
+ for pcall in chain:
+ if pcall in [
+ "__wrap_sprintf",
+ "__wrap_snprintf",
+ "__wrap_vsnprintf",
+ "vfctprintf",
+ ]:
+ last = pcall
+ if last == "vfctprintf":
+ return call != "_out_fct"
+ else:
+ return call == "_out_buffer"
+ return False
+ def extra_nodes(self) -> typing.Collection[Node]:
# src/rp2_common/hardware_divider/include/hardware/divider_helper.S
save_div_state_and_lr = 5 * 4
# src/rp2_common/pico_divider/divider_hardware.S
save_div_state_and_lr_64 = 5 * 4
- all_nodes += [
+ return [
# src/rp2_common/pico_int64_ops/pico_int64_ops_aeabi.S
synthetic_node("__aeabi_lmul", 4),
# src/rp2_common/pico_divider/divider_hardware.S
@@ -668,17 +755,23 @@ def main(
# synthetic_node("rom_hword_as_ptr(BOOTROM_TABLE_LOOKUP_OFFSET)", 0), # TODO
]
- # TinyUSB device #################################################
- usbd_c_fname = get_zero_or_one(
- lambda fname: fname.endswith("/tinyusb/src/device/usbd.c"), arg_c_fnames
- )
+class TinyUSBDevicePlugin:
+ tud_drivers: dict[str, set[str]]
- tusb_config_h_fname = get_zero_or_one(
- lambda fname: fname.endswith("/tusb_config.h"), arg_c_fnames
- )
+ def __init__(self, arg_c_fnames: typing.Collection[str]) -> None:
+ usbd_c_fname = get_zero_or_one(
+ lambda fname: fname.endswith("/tinyusb/src/device/usbd.c"), arg_c_fnames
+ )
+
+ tusb_config_h_fname = get_zero_or_one(
+ lambda fname: fname.endswith("/tusb_config.h"), arg_c_fnames
+ )
+
+ if not usbd_c_fname:
+ self.tud_drivers = {}
+ return
- if usbd_c_fname:
assert tusb_config_h_fname
re_tud_class = re.compile(
r"^\s*#\s*define\s+(?P<k>CFG_TUD_(?:\S{3}|AUDIO|VIDEO|MIDI|VENDOR|USBTMC|DFU_RUNTIME|ECM_RNDIS))\s+(?P<v>\S+).*"
@@ -725,40 +818,41 @@ def main(
in_table = False
elif " _usbd_driver[] = {" in line:
in_table = True
+ self.tud_drivers = tud_drivers
- def tud_indirect_callees(loc: str, line: str) -> tuple[list[str], bool] | None:
- if (
- "/tinyusb/" not in loc
- or "/tinyusb/src/host/" in loc
- or "_host.c:" in loc
- ):
- return None
- m = re_call_other.fullmatch(line)
- assert m
- call = m.group("func")
- if call == "_ctrl_xfer.complete_cb":
- return [
- # "process_test_mode_cb",
- "tud_vendor_control_xfer_cb",
- *sorted(tud_drivers["control_xfer_cb"]),
- ], False
- elif call.startswith("driver->"):
- return sorted(tud_drivers[call[len("driver->") :]]), False
- elif call == "event.func_call.func":
- # callback from usb_defer_func()
- return [], False
+ def extra_nodes(self) -> typing.Collection[Node]:
+ return []
+ def indirect_callees(self, loc: str, line: str) -> tuple[list[str], bool] | None:
+ if "/tinyusb/" not in loc or "/tinyusb/src/host/" in loc or "_host.c:" in loc:
return None
+ m = re_call_other.fullmatch(line)
+ assert m
+ call = m.group("func")
+ if call == "_ctrl_xfer.complete_cb":
+ return [
+ # "process_test_mode_cb",
+ "tud_vendor_control_xfer_cb",
+ *sorted(self.tud_drivers["control_xfer_cb"]),
+ ], False
+ elif call.startswith("driver->"):
+ return sorted(self.tud_drivers[call[len("driver->") :]]), False
+ elif call == "event.func_call.func":
+ # callback from usb_defer_func()
+ return [], False
+
+ return None
- hooks_indirect_callees += [tud_indirect_callees]
+ def skip_call(self, chain: list[str], call: str) -> bool:
+ return False
- # newlib #########################################################
- if arg_pico_platform == "rp2040":
+class NewlibPlugin:
+ def extra_nodes(self) -> typing.Collection[Node]:
# This is accurate to
# /usr/arm-none-eabi/lib/thumb/v6-m/nofp/libg.a as of
# Parabola's arm-none-eabi-newlib 4.5.0.20241231-1.
- all_nodes += [
+ return [
# malloc
synthetic_node("free", 8, {"_free_r"}),
synthetic_node("malloc", 8, {"_malloc_r"}),
@@ -788,18 +882,95 @@ def main(
synthetic_node("random", 8),
]
- # libgcc #########################################################
+ def indirect_callees(self, loc: str, line: str) -> tuple[list[str], bool] | None:
+ return None
- if arg_pico_platform == "rp2040":
+ def skip_call(self, chain: list[str], call: str) -> bool:
+ return False
+
+
+class LibGCCPlugin:
+ def extra_nodes(self) -> typing.Collection[Node]:
# This is accurate to
# /usr/lib/gcc/arm-none-eabi/14.2.0/thumb/v6-m/nofp/libgcc.a
# as of Parabola's arm-none-eabi-gcc 14.2.0-1.
- all_nodes += [
+ return [
synthetic_node("__aeabi_idiv0", 0),
synthetic_node("__aeabi_ldiv0", 0),
synthetic_node("__aeabi_llsr", 0),
]
+ def indirect_callees(self, loc: str, line: str) -> tuple[list[str], bool] | None:
+ return None
+
+ def skip_call(self, chain: list[str], call: str) -> bool:
+ return False
+
+
+def main(
+ *,
+ arg_pico_platform: str,
+ arg_base_dir: str,
+ arg_ci_fnames: typing.Collection[str],
+ arg_c_fnames: typing.Collection[str],
+) -> None:
+
+ plugins: list[Plugin] = []
+ hooks_is_intrhandler: list[typing.Callable[[str], bool]] = []
+
+ # sbc-harness ####################################################
+
+ lib9p_plugin = Lib9PPlugin(arg_base_dir, arg_c_fnames)
+
+ sbc_gpio_handlers = [
+ "w5500_intrhandler",
+ ]
+
+ def sbc_is_thread(name: str) -> int:
+ if name.endswith("_cr") and name != "lib9p_srv_read_cr":
+ if "9p" in name:
+ return lib9p_plugin.thread_count(name)
+ return 1
+ if name == "main":
+ return 1
+ return 0
+
+ def sbc_is_intrhandler(name: str) -> bool:
+ return name in [
+ "rp2040_hwtimer_intrhandler",
+ "_cr_gdb_intrhandler",
+ "hostclock_handle_sig_alarm",
+ "hostnet_handle_sig_io",
+ ]
+
+ hooks_is_intrhandler += [sbc_is_intrhandler]
+
+ plugins += [
+ LibObjPlugin(arg_c_fnames),
+ LibHWPlugin(),
+ LibCRIPCPlugin(),
+ lib9p_plugin,
+ LibMiscPlugin(),
+ ]
+
+ # pico-sdk #######################################################
+
+ if arg_pico_platform == "rp2040":
+
+ def pico_is_intrhandler(name: str) -> bool:
+ return name in [
+ "gpio_default_irq_handler",
+ ]
+
+ hooks_is_intrhandler += [pico_is_intrhandler]
+
+ plugins += [
+ PicoSDKPlugin(sbc_gpio_handlers),
+ TinyUSBDevicePlugin(arg_c_fnames),
+ NewlibPlugin(),
+ LibGCCPlugin(),
+ ]
+
# Tie it all together ############################################
def thread_filter(name: str) -> int:
@@ -824,38 +995,14 @@ def main(
parts[0] = "./" + os.path.relpath(parts[0], arg_base_dir)
return ":".join(parts)
- def indirect_callees(elem: VCGElem) -> tuple[list[str], bool]:
- loc = elem.attrs.get("label", "")
- line = read_source(loc)
-
- for hook in hooks_indirect_callees:
- ret = hook(loc, line)
- if ret is not None:
- return ret
-
- placeholder = "__indirect_call"
- if m := re_call_other.fullmatch(line):
- placeholder += ":" + m.group("func")
- placeholder += " at " + location_xform(elem.attrs.get("label", ""))
- return [placeholder], False
-
- def skip_call(chain: list[str], call: str) -> bool:
- for hook in hooks_skip_call:
- if hook(chain, call):
- return True
- return False
-
result = analyze(
ci_fnames=arg_ci_fnames,
- extra_nodes=all_nodes,
app_func_filters={
"Threads": thread_filter,
"Interrupt handlers": intrhandler_filter,
"Misc": misc_filter,
},
- app_location_xform=location_xform,
- app_indirect_callees=indirect_callees,
- app_skip_call=skip_call,
+ app=PluginApplication(location_xform, plugins),
cfg_max_call_depth=100,
)