TermiNexus
2021c0edb7
为匹配白名单的直接 call 点插桩 internal i64 __fhs_site_<K>（call 前 +1），计数天然包含循环内放大的真实执行次数，可把热点定位到具体调用点。 新增 --hotspot-callees 覆盖白名单、--hotspot-sites-all 全量；按去 16 位 sha1 前缀后的基名匹配。Ctrl+C 落盘独立文件 hotspot_sites.txt（# sites 分段，含调用方/被调方/计数/源码位置），按计数降序；不破坏既有 hotspot.txt 格式。 在 AST->IR 生成期由 _HotspotSiteBuilder 为每条新生成的 call 指令轻量挂载 _src_file/_src_line，插桩时作为可选尾字段输出；取不到则留空不伪造。开关关闭时改用原生 ir.IRBuilder，零影响。
1519 lines
83 KiB
Python
1519 lines
83 KiB
Python
from __future__ import annotations
|
||
from typing import Any
|
||
import vast as ast
|
||
import os
|
||
import llvmlite.ir as ir
|
||
from lib.includes import t as _t
|
||
from lib.includes.t import CTypeRegistry
|
||
from lib.core.Handles.HandlesBase import CTypeInfo as _CTypeInfo, BuiltinTypeMap
|
||
from lib.constants.config import DEFAULT_HOTSPOT_CALLEES
|
||
|
||
|
||
class VaArgInstruction(ir.Instruction):
|
||
def descr(self, buf: list[str]) -> None:
|
||
ptr: ir.Value = self.operands[0]
|
||
buf.append("va_arg {0} {1}, {2}\n".format(
|
||
ptr.type, ptr.get_reference(), self.type))
|
||
|
||
|
||
class _HotspotSiteBuilder(ir.IRBuilder):
|
||
"""--hotspot-sites 专用 IRBuilder：为每条新生成的 call 指令挂源码位置。
|
||
|
||
仅当调用点热点统计开关打开时才被实例化（见 BaseGenMixin._NewIrBuilder）；
|
||
未启用时一律使用原生 ir.IRBuilder，不产生任何额外属性、行为与既有完全一致。
|
||
|
||
重写 call() 只在返回的 ir.CallInstr 对象上追加两个普通 Python 属性
|
||
（_src_file / _src_line），不改变指令语义、操作数，也不参与 IR 渲染
|
||
（llvmlite 仅按指令自身类型输出文本，忽略这些自定义属性）。取不到位置时
|
||
留空（None / 0），不伪造。
|
||
"""
|
||
|
||
def __init__(self, block: ir.Block | None, gen: Any) -> None:
|
||
super().__init__(block)
|
||
self._HotspotGen: Any = gen
|
||
|
||
def call(self, fn: Any, args: Any, name: str = '', cconv: Any = None,
|
||
tail: Any = False, fastmath: Any = (), attrs: Any = (),
|
||
arg_attrs: Any = None) -> ir.CallInstr:
|
||
instr: ir.CallInstr = super().call(
|
||
fn, args, name=name, cconv=cconv, tail=tail,
|
||
fastmath=fastmath, attrs=attrs, arg_attrs=arg_attrs)
|
||
gen: Any = self._HotspotGen
|
||
# 记录生成期当前源码位置（语句级 lineno + 当前源文件名）；取不到则留空
|
||
instr._src_file = getattr(gen, '_current_source_file', None)
|
||
instr._src_line = getattr(gen, '_current_lineno', 0) or 0
|
||
return instr
|
||
|
||
|
||
class BaseGenMixin:
|
||
|
||
@staticmethod
|
||
def _parse_triple(triple: str | None) -> dict[str, bool | int]:
|
||
"""解析目标三元组，返回平台信息字典"""
|
||
info: dict[str, bool | int] = {
|
||
'is_windows': False, 'is_linux': False, 'is_macos': False,
|
||
'is_x86_64': False, 'is_arm64': False, 'is_32bit': False,
|
||
'ptr_size': 8, 'long_size': 8, 'wchar_t_size': 32,
|
||
}
|
||
if not triple:
|
||
return info
|
||
triple_lower: str = triple.lower()
|
||
# 检测操作系统
|
||
if 'windows' in triple_lower or 'win32' in triple_lower or 'mingw' in triple_lower or 'msvc' in triple_lower or 'cygwin' in triple_lower:
|
||
info['is_windows'] = True
|
||
# Windows LLP64: long = 32 位, wchar_t = 16 位
|
||
info['long_size'] = 32
|
||
info['wchar_t_size'] = 16
|
||
elif 'linux' in triple_lower:
|
||
info['is_linux'] = True
|
||
elif 'darwin' in triple_lower or 'macos' in triple_lower or 'apple' in triple_lower:
|
||
info['is_macos'] = True
|
||
# 检测架构
|
||
if 'x86_64' in triple_lower or 'amd64' in triple_lower or 'x64' in triple_lower:
|
||
info['is_x86_64'] = True
|
||
elif 'aarch64' in triple_lower or 'arm64' in triple_lower:
|
||
info['is_arm64'] = True
|
||
elif 'i386' in triple_lower or 'i686' in triple_lower or 'arm' in triple_lower:
|
||
info['is_32bit'] = True
|
||
info['ptr_size'] = 4
|
||
info['long_size'] = 32
|
||
if info['is_windows']:
|
||
info['wchar_t_size'] = 16
|
||
else:
|
||
info['wchar_t_size'] = 32
|
||
return info
|
||
|
||
def __init__(self, triple: str | None = None, datalayout: str | None = None) -> None:
|
||
self.module: ir.Module = ir.Module(name="transpyc_module")
|
||
if triple:
|
||
self.module.triple = triple
|
||
else:
|
||
self.module.triple = "x86_64-none-elf"
|
||
if datalayout:
|
||
self.module.data_layout = datalayout
|
||
else:
|
||
self.module.data_layout = "e-m:e-p270:32:32-p271:32:32-p272:64:64-i64:64-f80:128-n8:16:32:64-S128"
|
||
|
||
# 解析目标平台信息
|
||
self._platform_info: dict[str, bool | int] = self._parse_triple(self.module.triple)
|
||
self.ptr_size: int = self._platform_info['ptr_size']
|
||
# 根据目标平台配置 t 模块中的平台相关类型大小
|
||
_t.configure_platform(self.module.triple)
|
||
self.builder: ir.IRBuilder | None = None
|
||
self.func: ir.Function | None = None
|
||
self.functions: dict[str, ir.Function] = {}
|
||
self.variables: dict[str, ir.Value | None] = {}
|
||
self._direct_values: dict[str, ir.Value] = {}
|
||
self._reg_values: dict[str, ir.Value] = {}
|
||
self.var_signedness: dict[str, bool] = {}
|
||
self._unsigned_results: set[int] = {}
|
||
self.string_const_counter: int = 0
|
||
self.Vtables: dict[str, ir.GlobalVariable] = {}
|
||
self.class_methods: dict[str, list[str]] = {}
|
||
self.class_members: dict[str, list[tuple[str, ir.Type]]] = {}
|
||
self.class_member_defaults: dict[str, dict[str, Any]] = {}
|
||
self.class_member_signeds: dict[str, dict[str, bool]] = {}
|
||
self.class_member_bitfields: dict[str, dict[str, int]] = {}
|
||
self.class_member_byteorders: dict[str, dict[str, str]] = {}
|
||
self.local_var_byteorders: dict[str, str] = {}
|
||
self.class_member_element_class: dict[str, dict[str, str]] = {}
|
||
self.class_member_bitoffsets: dict[str, dict[str, int]] = {}
|
||
self.structs: dict[str, ir.IdentifiedStructType] = {}
|
||
self.typedef_int_widths: dict[str, int] = {} # typedef名称 -> 底层整数位宽 (如 'UINT' -> 32, 'BYTE' -> 8)
|
||
self._cross_module_vtable_classes: set[str] = set()
|
||
self._cross_module_novtable: set[str] = set()
|
||
self.class_vtable: set[str] = set()
|
||
self._class_graph: dict[str, Any] = {}
|
||
self.class_parent: dict[str, str] = {}
|
||
# 多继承：父类名数组（声明顺序）。class_parent 兼容保留为第一个父类。
|
||
self.class_parents: dict[str, list[str]] = {}
|
||
# Store member annotations for deferred type resolution (cross-module class references)
|
||
self.class_member_annotations: dict[str, dict[str, Any]] = {}
|
||
self.loop_break_targets: list[ir.Block] = []
|
||
self.loop_continue_targets: list[ir.Block] = []
|
||
self.eh_jmp_buf_stack: list[ir.Value] = []
|
||
self.eh_except_block_stack: list[ir.Block] = []
|
||
self.eh_finally_stack: list[Any] = []
|
||
self.global_vars: set[str] = set()
|
||
self.nonlocal_params: dict[str, Any] = {}
|
||
self._variable_scope_stack: list[dict[str, Any]] = [] # Stack of outer scope variables for nested function nonlocal lookup
|
||
self.var_struct_class: dict[str, str] = {}
|
||
self.var_ptr_element: dict[str, bool] = {}
|
||
# 全局变量的指针元素标志（持久，不受函数重置影响）
|
||
# 用于全局变量 _argv: bytes | t.CPtr 等，使 _argv[index] 使用 8 字节指针步长
|
||
self.global_var_ptr_element: dict[str, bool] = {}
|
||
self.class_provides: dict[str, list[str]] = {}
|
||
self.class_requires: dict[str, list[str]] = {}
|
||
self.class_require_must: dict[str, list[str]] = {}
|
||
# 编译期 with 上下文栈：记录当前嵌套 with 链中所有可用的 provides 字段名
|
||
# 用于 __require_must__ 的纯编译期静态可达性检查
|
||
self._with_provides_stack: list[list[str]] = []
|
||
# 函数参数自动取地址模式: func_name -> [None, None, 'need', 'always', ...]
|
||
# 'need': t.CNeedPtr — 仅对非指针值取地址
|
||
# 'always': t.CAutoPtr — 无条件取地址
|
||
self._auto_addr_params: dict[str, list[str | None]] = {}
|
||
self.global_struct_class: dict[str, str] = {}
|
||
self.var_type_info: dict[str, Any] = {}
|
||
self.var_const_flags: dict[str, bool] = {}
|
||
self._current_source_file: str | None = None
|
||
self._current_source_lines: list[str] = []
|
||
self._ns_cache: dict[str, str] = {}
|
||
self._type_cache: dict[tuple[str, bool], ir.Type] = {}
|
||
self._typedef_cache: dict[str, Any] = {}
|
||
self._func_sig_cache: dict[str, Any] = {}
|
||
self.var_type_assignments: dict[str, Any] = {}
|
||
self._temp_struct_ptrs: list[ir.Value] = []
|
||
self._stop_iter_flag_param: ir.Value | None = None
|
||
self._local_heap_ptrs: list[tuple[ir.Value, str, str | None]] = []
|
||
self._var_to_heap_ptr: dict[str, ir.Value] = {}
|
||
self.known_return_types: dict[str, ir.Type] = {}
|
||
self._current_lineno: int = 0
|
||
self._current_node_info: str = ""
|
||
self.module_sha1: str | None = None
|
||
self.ModuleSha1Map: dict[str, str] = {}
|
||
self._temp_dir: str | None = None
|
||
self._export_funcs: set[str] = set()
|
||
self._export_extern_funcs: dict[str, str] = {}
|
||
self._current_module_func_names: set[str] = set()
|
||
# __doc__ 魔法字符串注册表：symbol_name -> docstring 文本（None 表示空/优化剥离）
|
||
self._doc_strings: dict[str, str | None] = {}
|
||
self.class_packed: set[str] = set()
|
||
self._define_constants: dict[str, int] = {}
|
||
# 以下属性原先在外部懒注入，此处显式初始化以避免 hasattr/getattr 动态反射
|
||
self._struct_sha1_map: dict[str, str] = {}
|
||
self.SymbolTable: Any = None
|
||
self._Trans: Any = None
|
||
self.class_sha1_map: dict[str, str] = {}
|
||
# 以下属性原先懒注入，此处显式初始化以消除 hasattr/getattr 动态反射
|
||
self._import_handler_ref: Any = None
|
||
self._classes_in_progress: set[str] = set()
|
||
# 字段收集期间添加当前类名，阻止 CollectAnnAssignMember 触发的嵌套
|
||
# _generate_structs 用不完整 class_members 调用 set_body（set_body 只能调用一次）
|
||
# 用集合而非单值，因为嵌套特化类 _EmitClassLlvm 会清除单值，导致外层类被错误设置 body
|
||
self._collecting_members_set: set[str] = set()
|
||
self._last_var_name: str | None = None
|
||
self._variadic_info: dict | None = None
|
||
self._vtable_copy_counter: int = 0
|
||
self._decorated_funcs: dict = {}
|
||
self._current_tree: Any = None
|
||
self._DefineConstants: dict = {}
|
||
self._all_define_constants: dict = {}
|
||
# 闭包追踪：记录返回闭包的函数和变量
|
||
# _closure_return_types: func_name -> (fn_return_type, nonlocal_arg_types)
|
||
self._closure_return_types: dict[str, tuple[Any, list[Any]]] = {}
|
||
# _closure_var_types: var_name -> (fn_return_type, nonlocal_arg_types)
|
||
self._closure_var_types: dict[str, tuple[Any, list[Any]]] = {}
|
||
# 函数热点统计（--func-hotspot）：None=未启用（零影响）；str=计数器落盘文件路径
|
||
self.func_hotspot_file: str | None = None
|
||
# 注册策略：False=种子 CLI 单模块（在 main 入口注册）；True=Projectrans
|
||
# 多模块（每模块用 @llvm.global_ctors 自注册，不依赖 main 所在模块）
|
||
self.func_hotspot_module_local: bool = False
|
||
# 调用点热点统计（--hotspot-sites）：None=未启用（零影响）；str=调用点计数落盘文件路径
|
||
self.hotspot_sites_file: str | None = None
|
||
# 是否对全部直接 call 点插桩（忽略白名单；体积/开销大）
|
||
self.hotspot_sites_all: bool = False
|
||
# 白名单被调函数基名列表；None=使用默认内存/字符串族白名单
|
||
self.hotspot_callees: list[str] | None = None
|
||
# 注册策略：与 func_hotspot_module_local 同理
|
||
self.hotspot_sites_module_local: bool = False
|
||
pi: dict[str, bool | int] = self._platform_info
|
||
# 根据目标平台设置宏（声明式，从三元组推导）
|
||
if pi['is_windows']:
|
||
self._define_constants['_WIN32'] = 1
|
||
if not pi['is_32bit']:
|
||
self._define_constants['_WIN64'] = 1
|
||
self._define_constants['WIN64'] = 1
|
||
self._define_constants['WIN32'] = 1
|
||
if pi['is_linux']:
|
||
self._define_constants['__linux__'] = 1
|
||
if pi['is_macos']:
|
||
self._define_constants['__APPLE__'] = 1
|
||
self._define_constants['__MACH__'] = 1
|
||
if pi['is_x86_64']:
|
||
self._define_constants['__x86_64__'] = 1
|
||
self._define_constants['__x86_64'] = 1
|
||
self._define_constants['_M_X64'] = 1
|
||
elif pi['is_arm64']:
|
||
self._define_constants['__aarch64__'] = 1
|
||
self._define_constants['_M_ARM64'] = 1
|
||
if not pi['is_32bit'] and pi['is_linux']:
|
||
self._define_constants['__LP64__'] = 1
|
||
elif pi['is_32bit']:
|
||
self._define_constants['_ILP32'] = 1
|
||
self._define_constants['SIZEOF_VOID_P'] = pi['ptr_size']
|
||
self._define_constants['__SIZEOF_POINTER__'] = pi['ptr_size']
|
||
|
||
def find_struct_by_pointee(self, pointee: ir.Type) -> tuple[str, ir.IdentifiedStructType] | None:
|
||
"""根据 pointee 类型查找匹配的结构体，返回 (class_name, struct_type) 或 None"""
|
||
for class_name, struct_type in self.structs.items():
|
||
if pointee == struct_type:
|
||
return (class_name, struct_type)
|
||
return None
|
||
|
||
@staticmethod
|
||
def _extract_short_name(name: str) -> str:
|
||
"""从结构体类型名中提取短名（去掉 sha1 前缀）。
|
||
如 "6bc91356c652f8ca.Function" → "Function"
|
||
如 "Function" → "Function"（无 sha1 前缀时原样返回）
|
||
"""
|
||
if '.' in name:
|
||
prefix: str = name.split('.', 1)[0]
|
||
if len(prefix) == 16 and all(c in '0123456789abcdef' for c in prefix):
|
||
return name.split('.', 1)[1]
|
||
return name
|
||
|
||
@staticmethod
|
||
def _extract_struct_sha1(st: ir.IdentifiedStructType) -> str:
|
||
"""从 IdentifiedStructType 的 name 中提取 sha1 前缀（16 位十六进制）。
|
||
返回空字符串表示无 sha1 前缀。用于同名类冲突检测。
|
||
"""
|
||
if not isinstance(st, ir.IdentifiedStructType):
|
||
return ''
|
||
try:
|
||
name: str = st.name
|
||
except (AssertionError, AttributeError):
|
||
return ''
|
||
if '.' in name:
|
||
prefix: str = name.split('.', 1)[0]
|
||
if len(prefix) == 16 and all(c in '0123456789abcdef' for c in prefix):
|
||
return prefix
|
||
return ''
|
||
|
||
def _get_or_create_struct(self, name: str, source_sha1: str | None = None, packed: bool = False) -> ir.IdentifiedStructType:
|
||
clean_name: str = name.replace(' *', '').replace('*', '').replace(' ', '_').replace('const', 'const_').replace('volatile', 'volatile_')
|
||
# 如果名称包含模块前缀（如 "a9fa0f6200c09e65.zbit_writer"），
|
||
# 先尝试用短名称（如 "zbit_writer"）查找已注册的 struct，
|
||
# 避免为同一 struct 创建多个不同的类型对象导致跨模块引用类型不一致
|
||
if '.' in clean_name and clean_name not in self.structs:
|
||
parts: list[str] = clean_name.split('.', 1)
|
||
if len(parts[0]) == 16 and all(c in '0123456789abcdef' for c in parts[0]):
|
||
short_name: str = parts[1]
|
||
if short_name in self.structs:
|
||
existing_short: ir.IdentifiedStructType = self.structs[short_name]
|
||
# 治本修复：同名类 sha1 冲突检测。
|
||
# 当请求的 sha1 前缀与现有 struct 的 sha1 前缀不一致时
|
||
# （如 stmts.py:Module 与 __module.py:Module 同名冲突），
|
||
# 不能复用已有 struct（set_body 已锁定为错误布局），
|
||
# 需创建独立 struct 并以 sha1 前缀全名存储。
|
||
_existing_sha1: str = self._extract_struct_sha1(existing_short)
|
||
if _existing_sha1 and _existing_sha1 != parts[0]:
|
||
if clean_name in self.structs:
|
||
return self.structs[clean_name]
|
||
st_new: ir.IdentifiedStructType = ir.IdentifiedStructType(self.module, name=clean_name, packed=packed)
|
||
self.structs[clean_name] = st_new
|
||
return st_new
|
||
return existing_short
|
||
# 优先解析 typedef 整数类型（如 INT/UINT/BYTE），避免将其注册为 opaque struct。
|
||
# 必须在 `if clean_name in self.structs` 缓存检查之前执行，
|
||
# 否则首次调用（SymbolTable 未就绪时）会将 typedef 名缓存为 opaque struct，
|
||
# 后续调用即使 SymbolTable 就绪也只会返回缓存的错误类型。
|
||
# 此修复同时解决两个问题：
|
||
# 1. alloca unsized type：变量声明 `argc: INT = 6` 不再生成 `alloca %"...INT"`
|
||
# 2. 下标访问：FindStructNameInAnnotation 不再匹配 typedef 名（因不在 Gen.structs）
|
||
td_width: int | None = self._resolve_typedef_int_width(clean_name)
|
||
if td_width is not None:
|
||
if td_width > 0:
|
||
self.typedef_int_widths[clean_name] = td_width
|
||
return ir.IntType(td_width)
|
||
# td_width == 0: 特殊 typedef 名称（str/bytes/list 等），返回 i8
|
||
self.typedef_int_widths[clean_name] = 0
|
||
return ir.IntType(8)
|
||
if clean_name in self.structs:
|
||
existing: ir.IdentifiedStructType = self.structs[clean_name]
|
||
# 治本修复：同名类 sha1 冲突检测（短名调用路径）。
|
||
# 当 source_sha1 与现有 struct 的 sha1 前缀不一致时（跨模块同名类冲突），
|
||
# 不能复用已有 struct（set_body 已锁定为错误布局），
|
||
# 需创建独立 struct 并以 sha1 前缀全名存储，不覆盖短名条目。
|
||
if source_sha1:
|
||
_existing_sha1: str = self._extract_struct_sha1(existing)
|
||
if _existing_sha1 and _existing_sha1 != source_sha1:
|
||
full_key: str = f"{source_sha1}.{clean_name}"
|
||
if full_key in self.structs:
|
||
st_full: ir.IdentifiedStructType = self.structs[full_key]
|
||
if packed and isinstance(st_full, ir.IdentifiedStructType) and not st_full.packed:
|
||
st_full.packed = True
|
||
return st_full
|
||
ns_name: str = f"{source_sha1}.{clean_name}"
|
||
st_new: ir.IdentifiedStructType = ir.IdentifiedStructType(self.module, name=ns_name, packed=packed)
|
||
self.structs[full_key] = st_new
|
||
return st_new
|
||
if packed and isinstance(existing, ir.IdentifiedStructType) and not existing.packed:
|
||
existing.packed = True
|
||
return existing
|
||
if clean_name in {'typedef', 'CType', 'CVolatile', 'CEnum', 'CUnion', 'CStruct', 'enum', 'REnum', 'renum', 'Void', 'void'}:
|
||
return ir.IntType(8)
|
||
# typing.Any 是"任意类型"标记，应解析为 i8*（通用指针），
|
||
# 而非创建为 opaque 结构体（否则包含 Any 字段的结构体会变成 unsized）。
|
||
if clean_name == 'Any':
|
||
return ir.IntType(8).as_pointer()
|
||
# t.CType 的基本类型子类（CInt/CChar/CSizeT/CLong/CFloat/CDouble 等，
|
||
# position 含 BASE 且 Size>0）有具体 LLVM 类型，应直接返回对应的
|
||
# ir.IntType/ir.FloatType/ir.DoubleType，而非创建结构体。
|
||
# _is_ctype_marker_name 对这些类型返回 False（不跳过），需要在此处拦截。
|
||
_basic_llvm_type: ir.Type | None = self._try_resolve_ctype_basic(clean_name)
|
||
if _basic_llvm_type is not None:
|
||
return _basic_llvm_type
|
||
# t.CType 及其子类是"类型强转"标记，不应被编译为真实结构体。
|
||
# 动态检查 t 模块中所有 CType 子类名（排除 CStruct/CEnum/CUnion/REnum 子类，
|
||
# 因为用户定义的结构体类继承自它们，需要被创建为结构体）。
|
||
if self._is_ctype_marker_name(clean_name):
|
||
return ir.IntType(8)
|
||
if self.SymbolTable:
|
||
Entry: _CTypeInfo | None = self.SymbolTable.lookup(clean_name)
|
||
if Entry and Entry.IsTypedef:
|
||
resolved_str: str = self._resolve_typedef(clean_name)
|
||
if resolved_str != clean_name:
|
||
return self._type_str_to_llvm(resolved_str)
|
||
if Entry.BaseType:
|
||
if not isinstance(Entry.BaseType, (_t._CTypedef,)) or Entry.PtrCount > 0:
|
||
resolved: str = self._resolve_ctype_to_str(Entry)
|
||
if resolved:
|
||
return self._type_str_to_llvm(resolved)
|
||
if Entry and Entry.IsRenum:
|
||
if clean_name in self.structs:
|
||
return self.structs[clean_name]
|
||
# REnum 结构体尚未创建时不能 fall through 到 IsEnum（会返回 i32），
|
||
# 需要继续到下方正常结构体创建路径，创建 opaque 结构体。
|
||
# 这对自引用 REnum（如 LLVMType 包含 LLVMType*）至关重要：
|
||
# _parse_llvm_type 先通过此路径创建 opaque 结构体，_TryLoadStructFromStub 再 set_body。
|
||
elif Entry and Entry.IsEnum and clean_name not in self.class_methods:
|
||
return ir.IntType(32)
|
||
# 治本修复：当 clean_name 是已注册的类（在 class_methods 或 class_members 中）时，
|
||
# 不应走 IsExceptionClass/IsFunction/IsVariable 分支返回非结构体类型。
|
||
# 否则 _get_or_create_struct 返回指针类型（如 _TypedPointerType）而非 IdentifiedStructType，
|
||
# 导致 _TryLoadStructFromStub 的 set_body 被跳过（isinstance 检查失败），
|
||
# struct 保持不完整字段数，引发跨模块 GEP 越界崩溃。
|
||
_is_class_name: bool = clean_name in self.class_methods or clean_name in self.class_members
|
||
if Entry and Entry.IsExceptionClass and not _is_class_name:
|
||
return ir.IntType(32)
|
||
if Entry and Entry.IsFunction and not _is_class_name:
|
||
return ir.PointerType(ir.IntType(8))
|
||
if Entry and Entry.IsVariable and not _is_class_name:
|
||
return ir.PointerType(ir.IntType(8))
|
||
if Entry and Entry.BaseType:
|
||
if isinstance(Entry.BaseType, type) and issubclass(Entry.BaseType, _t.CEnum):
|
||
return ir.IntType(32)
|
||
if Entry.BaseType is _t.CEnum:
|
||
return ir.IntType(32)
|
||
ns_name: str
|
||
if source_sha1:
|
||
ns_name = f"{source_sha1}.{clean_name}"
|
||
elif clean_name in self._struct_sha1_map:
|
||
ns_name = f"{self._struct_sha1_map[clean_name]}.{clean_name}"
|
||
else:
|
||
ns_name = self._mangle_name(clean_name)
|
||
st: ir.IdentifiedStructType = ir.IdentifiedStructType(self.module, name=ns_name, packed=packed)
|
||
self.structs[clean_name] = st
|
||
# 尝试解析 typedef 的底层整数位宽
|
||
width: int | None = self._resolve_typedef_int_width(clean_name)
|
||
if width is not None:
|
||
self.typedef_int_widths[clean_name] = width
|
||
return st
|
||
|
||
def _resolve_typedef_int_width(self, name: str) -> int | None:
|
||
"""解析 typedef 名称的底层整数位宽
|
||
|
||
通过 CTypeRegistry 和 BuiltinTypeMap 动态解析，
|
||
替代硬编码的 TYPEDEF_NAMES 映射表。
|
||
返回 int 位宽或 None（无法解析）。
|
||
"""
|
||
# 1. 先查 BuiltinTypeMap（覆盖 BYTEPTR 等指针别名）
|
||
bm: Any = BuiltinTypeMap.Get(name)
|
||
if bm:
|
||
ctype_cls: Any
|
||
ptr_count: int
|
||
ctype_cls, ptr_count = bm
|
||
if ptr_count > 0:
|
||
return None # 指针类型，不是整数 typedef
|
||
llvm_str: str = CTypeRegistry.CTypeToLLVM(ctype_cls)
|
||
if llvm_str and llvm_str.startswith('i') and llvm_str[1:].isdigit():
|
||
return int(llvm_str[1:])
|
||
|
||
# 2. 再查 SymbolTable（用户定义的 typedef）
|
||
if self.SymbolTable:
|
||
Entry: _CTypeInfo | None = self.SymbolTable.lookup(name)
|
||
if Entry and Entry.IsTypedef:
|
||
if Entry.BaseType:
|
||
if not isinstance(Entry.BaseType, (_t._CTypedef,)) and Entry.PtrCount == 0:
|
||
llvm_str: str = CTypeRegistry.CTypeToLLVM(type(Entry.BaseType) if isinstance(Entry.BaseType, _t.CType) else Entry.BaseType)
|
||
if llvm_str and llvm_str.startswith('i') and llvm_str[1:].isdigit():
|
||
return int(llvm_str[1:])
|
||
if Entry.OriginalType:
|
||
if isinstance(Entry.OriginalType, _CTypeInfo) and Entry.OriginalType.BaseType:
|
||
if not isinstance(Entry.OriginalType.BaseType, (_t._CTypedef,)) and Entry.OriginalType.PtrCount == 0:
|
||
llvm_str: str = CTypeRegistry.CTypeToLLVM(type(Entry.OriginalType.BaseType) if isinstance(Entry.OriginalType.BaseType, _t.CType) else Entry.OriginalType.BaseType)
|
||
if llvm_str and llvm_str.startswith('i') and llvm_str[1:].isdigit():
|
||
return int(llvm_str[1:])
|
||
elif isinstance(Entry.OriginalType, str):
|
||
# OriginalType 是类型名字符串（如 'CInt', 'CUnsignedInt'），
|
||
# 通过 CTypeRegistry 解析为整数位宽
|
||
cls: Any = CTypeRegistry.GetClassByName(Entry.OriginalType)
|
||
if cls:
|
||
llvm_str: str = CTypeRegistry.CTypeToLLVM(cls)
|
||
if llvm_str and llvm_str.startswith('i') and llvm_str[1:].isdigit():
|
||
return int(llvm_str[1:])
|
||
|
||
# 3. 尝试 CTypeRegistry 直接按名称查找
|
||
ctype_cls: Any = CTypeRegistry.GetClassByName(name)
|
||
if ctype_cls:
|
||
llvm_str: str = CTypeRegistry.CTypeToLLVM(ctype_cls)
|
||
if llvm_str and llvm_str.startswith('i') and llvm_str[1:].isdigit():
|
||
return int(llvm_str[1:])
|
||
|
||
# 4. 特殊名称（非整数类型，但需要标记为 typedef 以跳过 opaque 定义）
|
||
# 注意：如果 name 已注册为 struct（用户定义了同名 class，如 includes/_dict.py 中的 dict），
|
||
# 则不应视为 typedef，应返回 None 让调用方走 struct 创建路径。
|
||
# 解决时序问题：_parse_llvm_declare 从 stub.ll 解析时 dict 可能尚未注册到 Gen.structs，
|
||
# 但 SymbolTable 在 .pyi 解析阶段已就绪，可通过 is_struct 检查区分 class 和 typedef。
|
||
_SPECIAL_TYPEDEFS: set[str] = {
|
||
'typedef', 'CType', 'CVolatile', 'CEnum', 'CUnion',
|
||
'CStruct', 'enum', 'REnum', 'renum',
|
||
'list', 'dict', 'tuple', 'set', 'str', 'bytes', 'array', 'type', 'CArray',
|
||
}
|
||
if name in _SPECIAL_TYPEDEFS:
|
||
if name in self.structs:
|
||
return None # 已注册为 struct，不视为 typedef
|
||
if self.SymbolTable is not None:
|
||
try:
|
||
if self.SymbolTable.is_struct(name):
|
||
return None # SymbolTable 中标记为 struct，不视为 typedef
|
||
except Exception:
|
||
pass
|
||
return 0 # 标记为 typedef，但位宽为 0（非整数）
|
||
|
||
# t.CType 及其子类是"类型强转"标记，不应被创建为结构体。
|
||
# 动态检查 t 模块中所有 CType 子类名（排除 CStruct/CEnum/CUnion/REnum 子类）。
|
||
# 返回 0 标记为 typedef，使 _get_or_create_struct 走 typedef 路径返回 i8。
|
||
if self._is_ctype_marker_name(name):
|
||
return 0
|
||
|
||
return None
|
||
|
||
@staticmethod
|
||
def _is_ctype_marker_name(name: str) -> bool:
|
||
"""检查名称是否是 t.CType 的"类型强转"标记子类名。
|
||
|
||
只有 position 不含 BASE 的纯修饰符/标记类型（CPtr/CVolatile/CConst/
|
||
CInline/CStatic/CExtern/CExport/CRegister/CAuto/_CTypedef/CTypeDefault 等）
|
||
才应该跳过结构体创建。有 BASE position 的基本类型（CInt/CChar/CSizeT/
|
||
CLong/CFloat/CDouble 等）有具体大小，应被正常解析为对应宽度的整数/浮点。
|
||
|
||
CStruct/CEnum/CUnion/REnum 的用户子类是合法结构体，需要被创建。
|
||
|
||
注意：CTypeRegistry.GetClassByName 会跳过 CType 本身和 CTypeDefault，
|
||
所以这里直接用 getattr(t, name) 检查 t 模块属性。
|
||
|
||
Returns:
|
||
True 如果 name 是 CType 标记子类名（应跳过结构体创建）
|
||
"""
|
||
if not name:
|
||
return False
|
||
# 去除可能的 SHA1 前缀（如 "abc123.CPtr" -> "CPtr"）
|
||
short_name: str = name.split('.')[-1] if '.' in name else name
|
||
# 直接查 t 模块属性（CTypeRegistry._build 会跳过 CType/CTypeDefault/BigEndian/LittleEndian）
|
||
t_cls: Any = getattr(_t, short_name, None)
|
||
if not isinstance(t_cls, type) or not issubclass(t_cls, _t.CType):
|
||
return False
|
||
# 排除 CStruct/CEnum/CUnion/REnum 子类（它们是命名类型的基类，
|
||
# 用户定义的结构体继承自它们，需要被创建为结构体）
|
||
if issubclass(t_cls, (_t.CStruct, _t.CEnum, _t.REnum, _t.CUnion)):
|
||
return False
|
||
# 基本类型（position 含 BASE 且 Size > 0）有具体大小，不应跳过
|
||
# （CInt→i32, CChar→i8, CSizeT→i64, CLong→i64, CFloat→f32 等）
|
||
# CVoid 虽含 BASE 但 Size=0，CType Size=None，仍需跳过
|
||
if _t.CType.BASE in t_cls.position:
|
||
# 检查实例的 Size（类属性可能未设置，用实例获取）
|
||
try:
|
||
inst = t_cls()
|
||
size_val = getattr(inst, 'Size', None)
|
||
if size_val is not None and size_val > 0:
|
||
return False
|
||
except Exception:
|
||
pass
|
||
return True
|
||
|
||
def _try_resolve_ctype_basic(self, name: str) -> ir.Type | None:
|
||
"""尝试将 t.CType 的基本类型子类名解析为对应的 LLVM 基本类型。
|
||
|
||
仅对 t 模块内的 CType 基本类型子类（CInt/CChar/CSizeT/CLong/CFloat/
|
||
CDouble 等，即 position 含 BASE 且 Size>0）生效，返回对应的
|
||
ir.IntType/ir.FloatType/ir.DoubleType。
|
||
|
||
CStruct/CEnum/CUnion/REnum 子类不在此处理（用户自定义结构体需要被
|
||
创建为结构体）。CVoid/CType/CPtr 等无 Size 或 Size=0 的类型也不在此
|
||
处理（应通过 _is_ctype_marker_name 跳过结构体创建）。
|
||
|
||
Returns:
|
||
对应的 ir.Type，或 None 表示不是基本类型，需要继续走结构体创建路径。
|
||
"""
|
||
if not name:
|
||
return None
|
||
short_name: str = name.split('.')[-1] if '.' in name else name
|
||
cls: Any = getattr(_t, short_name, None)
|
||
if not isinstance(cls, type) or not issubclass(cls, _t.CType):
|
||
return None
|
||
if issubclass(cls, (_t.CStruct, _t.CEnum, _t.REnum, _t.CUnion)):
|
||
return None
|
||
if _t.CType.BASE not in getattr(cls, 'position', frozenset()):
|
||
return None
|
||
try:
|
||
inst = cls()
|
||
size_val = getattr(inst, 'Size', None)
|
||
if size_val is None or size_val <= 0:
|
||
return None
|
||
except Exception:
|
||
return None
|
||
llvm_str: str | None = CTypeRegistry.NameToLLVM(short_name)
|
||
if not llvm_str:
|
||
return None
|
||
return self._type_str_to_llvm(llvm_str)
|
||
|
||
def _set_node_info(self, node: ast.AST, extra_info: str = "") -> None:
|
||
if hasattr(node, 'lineno'):
|
||
self._current_lineno = node.lineno
|
||
else:
|
||
self._current_lineno = 0
|
||
if hasattr(node, 'col_offset'):
|
||
self._current_node_info = f"line {self._current_lineno}, col {node.col_offset}"
|
||
else:
|
||
self._current_node_info = f"line {self._current_lineno}"
|
||
if extra_info:
|
||
self._current_node_info += f" ({extra_info})"
|
||
|
||
def _set_source_info(self, filepath: str) -> None:
|
||
self._current_source_file = filepath
|
||
try:
|
||
with open(filepath, 'r', encoding='utf-8') as f:
|
||
self._current_source_lines = f.readlines()
|
||
except Exception: # 文件读取失败时回退为空行列表
|
||
self._current_source_lines = []
|
||
|
||
def _skip_mangle(self, name: str) -> bool:
|
||
skip_prefixes: tuple[str, ...] = ('llvm.', 'str_const_', 'printf', 'memset_pattern')
|
||
if any(name.startswith(p) for p in skip_prefixes):
|
||
return True
|
||
if '.' in name:
|
||
parts: list[str] = name.split('.', 1)
|
||
if len(parts[0]) >= 8 and all(c in '0123456789abcdef' for c in parts[0]):
|
||
return True
|
||
return False
|
||
|
||
def _mangle_name(self, name: str) -> str:
|
||
if self._skip_mangle(name):
|
||
return name
|
||
if name in self._export_funcs:
|
||
return name
|
||
if self.module_sha1:
|
||
return f"{self.module_sha1}.{name}"
|
||
return name
|
||
|
||
def _find_function(self, name: str) -> ir.Function | None:
|
||
if name in self.functions:
|
||
return self.functions[name]
|
||
g: Any = self.module.globals.get(name)
|
||
if g and isinstance(g, ir.Function):
|
||
self.functions[name] = g
|
||
return g
|
||
mangled: str = self._mangle_func_name(name)
|
||
if mangled != name:
|
||
if mangled in self.functions:
|
||
self.functions[name] = self.functions[mangled]
|
||
return self.functions[mangled]
|
||
g = self.module.globals.get(mangled)
|
||
if g and isinstance(g, ir.Function):
|
||
self.functions[mangled] = g
|
||
self.functions[name] = g
|
||
return g
|
||
for sha1 in self.ModuleSha1Map.values():
|
||
prefixed: str = f"{sha1}.{name}"
|
||
if prefixed in self.functions:
|
||
self.functions[name] = self.functions[prefixed]
|
||
return self.functions[prefixed]
|
||
g = self.module.globals.get(prefixed)
|
||
if g and isinstance(g, ir.Function):
|
||
self.functions[prefixed] = g
|
||
self.functions[name] = g
|
||
return g
|
||
return None
|
||
|
||
def _has_function(self, name: str) -> bool:
|
||
return self._find_function(name) is not None
|
||
|
||
def _get_mro_chain(self, ClassName: str) -> list[str]:
|
||
"""MRO 线性化类名列表（深度优先 + 声明顺序 + 去重，含自身）。
|
||
|
||
顺序: 自身 → 父类1 及祖先 → 父类2 及祖先（先声明父类优先，菱形去重）。
|
||
与 TPV HandlesStruct.get_mro_chain 的语义一致。
|
||
|
||
支持带 sha1 前缀的类名（如 "c9b44d17cd621a75.CallGenLib"，跨模块
|
||
structs 表键）：查 class_parents 前剥离前缀，否则表键（裸类名）不匹配
|
||
→ MRO 链只剩自身 → 继承方法调用 fail-fast "Undefined method
|
||
'find_func_in_module' in class '...CallGenLib'"（CallGenLib 根因）。
|
||
"""
|
||
chain: list[str] = []
|
||
visited: set[str] = set()
|
||
stack: list[str] = [ClassName]
|
||
parents_map: dict[str, list[str]] = getattr(self, 'class_parents', {})
|
||
while stack:
|
||
cur: str = stack.pop()
|
||
if not cur or cur in visited:
|
||
continue
|
||
visited.add(cur)
|
||
chain.append(cur)
|
||
# 归一化：剥离 "16位hex." 前缀后用裸类名查父类表
|
||
lookup: str = cur
|
||
if len(cur) > 17 and cur[16] == '.':
|
||
lookup = cur[17:]
|
||
parents: list[str] = parents_map.get(lookup, parents_map.get(cur, []))
|
||
# 父类逆序压栈，保证声明顺序弹出（先访问父类1 的整条链）
|
||
for p in reversed(parents):
|
||
if p and p not in visited:
|
||
stack.append(p)
|
||
return chain
|
||
|
||
def _get_parent_classes(self, ClassName: str) -> list[str]:
|
||
"""获取类的所有父类名（声明顺序），兼容旧 class_parent 单链。"""
|
||
parents_map: dict[str, list[str]] = getattr(self, 'class_parents', {})
|
||
parents: list[str] = parents_map.get(ClassName)
|
||
if parents:
|
||
return parents
|
||
p: str | None = self.class_parent.get(ClassName)
|
||
if p:
|
||
return [p]
|
||
return []
|
||
|
||
def _get_or_declare_function(self, mangled_name: str, func_type: ir.FunctionType) -> ir.Function:
|
||
# DDEBUG: 检测 mangled_name 含双点（不应到达这里）
|
||
if '..' in mangled_name:
|
||
import sys
|
||
sys.stderr.write(f"[DDEBUG] _get_or_declare_function DOUBLE-DOT mangled_name='{mangled_name}'\n")
|
||
import traceback
|
||
traceback.print_stack(file=sys.stderr, limit=10)
|
||
sys.stderr.flush()
|
||
sys.exit(1)
|
||
g = self.module.globals.get(mangled_name)
|
||
if g and isinstance(g, ir.Function):
|
||
return g
|
||
return ir.Function(self.module, func_type, name=mangled_name)
|
||
|
||
def _get_function(self, name: str) -> ir.Function | None:
|
||
func: ir.Function | None = self._find_function(name)
|
||
return func
|
||
|
||
def _mangle_func_name(self, name: str, module_name: str | None = None) -> str:
|
||
# DDEBUG: 检测即将产生双点的情况，立即打印并终止
|
||
if name.startswith('.') or (module_name and module_name.endswith('.') and len(module_name) > 0):
|
||
import sys
|
||
sys.stderr.write(f"[DDEBUG] _mangle_func_name LEADING-DOT name='{name}' module_name='{module_name}'\n")
|
||
import traceback
|
||
traceback.print_stack(file=sys.stderr, limit=8)
|
||
sys.stderr.flush()
|
||
sys.exit(1)
|
||
if name in self._export_funcs and not module_name:
|
||
return name
|
||
if name in self._export_extern_funcs:
|
||
target_sha1: str = self._export_extern_funcs[name]
|
||
if module_name:
|
||
# 显式带 module_name 调用（如 stdlib.system）：即使当前模块定义了同名函数，
|
||
# 也应按模块匹配 extern，返回裸名。snprintf 之所以工作而 system 不工作，
|
||
# 就是因为 os._win32 定义了自己的 system 包装，导致旧逻辑跳过了 extern 匹配。
|
||
if module_name == target_sha1:
|
||
return name
|
||
if module_name in self.ModuleSha1Map and self.ModuleSha1Map[module_name] == target_sha1:
|
||
return name
|
||
init_name: str = f"{module_name}.__init__"
|
||
if init_name in self.ModuleSha1Map and self.ModuleSha1Map[init_name] == target_sha1:
|
||
return name
|
||
# 兜底：module_name 不在 ModuleSha1Map 中或 sha1 不匹配，
|
||
# 但函数在 _export_extern_funcs 中（C extern 不应混淆），
|
||
# 仅当当前模块未定义同名函数时才返回裸名。
|
||
if name not in self._current_module_func_names:
|
||
return name
|
||
elif name not in self._current_module_func_names:
|
||
# 无 module_name 调用（如 from stdlib import system; system()）：
|
||
# 仅当当前模块未定义同名函数时才返回 extern 裸名，
|
||
# 否则让本地定义走正常 mangle 路径（本地 def 遮蔽 import 语义）。
|
||
return name
|
||
# main 函数作为程序入口点，永远不混淆
|
||
if name == 'main':
|
||
return name
|
||
# Try class_sha1_map first for class methods (e.g., 'md5.__init__' -> class 'md5')
|
||
class_sha1_map: dict[str, str] = self.class_sha1_map
|
||
if class_sha1_map:
|
||
if '.' in name:
|
||
base_class: str = name.split('.')[0]
|
||
if base_class in class_sha1_map:
|
||
sha1: str = class_sha1_map[base_class]
|
||
return f"{sha1}.{name}"
|
||
elif name in class_sha1_map:
|
||
sha1: str = class_sha1_map[name]
|
||
return f"{sha1}.{name}"
|
||
# If module_name is already a SHA1 hash (16 hex chars), use it directly
|
||
if module_name and len(module_name) == 16 and all(c in '0123456789abcdef' for c in module_name):
|
||
return f"{module_name}.{name}"
|
||
if module_name and module_name in self.ModuleSha1Map:
|
||
sha1: str = self.ModuleSha1Map[module_name]
|
||
return f"{sha1}.{name}"
|
||
# Try parent package name (e.g., 'hashlib' -> 'hashlib.__init__')
|
||
if module_name:
|
||
init_name: str = f"{module_name}.__init__"
|
||
if init_name in self.ModuleSha1Map:
|
||
sha1: str = self.ModuleSha1Map[init_name]
|
||
return f"{sha1}.{name}"
|
||
if self._skip_mangle(name):
|
||
return name
|
||
if self.module_sha1:
|
||
result = f"{self.module_sha1}.{name}"
|
||
else:
|
||
result = name
|
||
# DDEBUG: 检测返回结果中是否含双点
|
||
if '..' in result:
|
||
import sys
|
||
sys.stderr.write(f"[DDEBUG] _mangle_func_name RESULT-DOUBLE-DOT name='{name}' module_name='{module_name}' result='{result}'\n")
|
||
import traceback
|
||
traceback.print_stack(file=sys.stderr, limit=10)
|
||
sys.stderr.flush()
|
||
sys.exit(1)
|
||
return result
|
||
|
||
def _get_source_line(self, lineno: int) -> str | None:
|
||
if 0 < lineno <= len(self._current_source_lines):
|
||
return self._current_source_lines[lineno - 1].rstrip('\r\n')
|
||
return None
|
||
|
||
def _get_node_info(self) -> str:
|
||
info: str = ""
|
||
if self._current_node_info:
|
||
info = f" [{self._current_node_info}]"
|
||
if self._current_source_file:
|
||
info += f" in {self._current_source_file}"
|
||
lineno: int = self._current_lineno
|
||
if lineno > 0:
|
||
info += f", line {lineno}:"
|
||
source_line: str | None = self._get_source_line(lineno)
|
||
if source_line:
|
||
info += f"\n → {source_line}"
|
||
return info
|
||
|
||
# ========== 统一工具方法 ==========
|
||
|
||
def _create_string_global(self, s: str, name: str | None = None) -> ir.Constant:
|
||
"""创建字符串全局变量并返回 i8* 指针。
|
||
|
||
统一字符串全局变量创建模式：编码为 utf-8 + '\\x00' 终止符，
|
||
创建 internal linkage 的 GlobalVariable，返回 i8* 指针常量。
|
||
"""
|
||
str_val: str = s + '\x00'
|
||
str_bytes: bytes = str_val.encode('utf-8')
|
||
arr_type: ir.ArrayType = ir.ArrayType(ir.IntType(8), len(str_bytes))
|
||
gv_name: str = name or f"str_const_{self.string_const_counter}"
|
||
self.string_const_counter += 1
|
||
gv: ir.GlobalVariable = ir.GlobalVariable(self.module, arr_type, name=gv_name)
|
||
gv.initializer = ir.Constant(arr_type, bytearray(str_bytes))
|
||
gv.linkage = 'internal'
|
||
return ir.Constant(ir.PointerType(ir.IntType(8)), gv.get_reference())
|
||
|
||
def _create_doc_global(self, symbol_name: str, docstring: str | None) -> None:
|
||
"""为符号创建 __doc__ 全局常量。
|
||
|
||
- docstring 为 None 或空字符串时，不创建常量（访问时内联空指针）
|
||
- slice_level != 0（优化开启）时，内容视为空
|
||
- 非空时创建 internal linkage、global_constant=True 的只读字符串常量
|
||
- 常量名格式：{module_sha1}.{symbol_name}.__doc__
|
||
"""
|
||
slice_level: int = getattr(self._Trans, 'SliceLevel', 3)
|
||
if slice_level != 0:
|
||
docstring = None
|
||
if docstring is None or docstring == '':
|
||
self._doc_strings[symbol_name] = None
|
||
return
|
||
self._doc_strings[symbol_name] = docstring
|
||
mangled: str = self._mangle_name(f"{symbol_name}.__doc__")
|
||
self._create_string_global(docstring, name=mangled)
|
||
gv: ir.GlobalVariable | None = self.module.globals.get(mangled)
|
||
if gv is not None:
|
||
gv.global_constant = True
|
||
|
||
def _get_doc_ptr(self, symbol_name: str) -> ir.Value | None:
|
||
"""获取符号 __doc__ 的 i8* 指针。
|
||
|
||
- 若符号未注册 __doc__，返回 None（调用方回退到其他解析）
|
||
- 若 __doc__ 为空（无 docstring 或优化剥离），返回 i8* null 内联空指针
|
||
- 若 __doc__ 非空，返回指向字符串全局常量的 i8* 指针
|
||
- 跨模块 docstring（在 _doc_strings 中有文本但 module.globals 无对应常量）首次访问时创建本地 internal linkage 只读副本
|
||
"""
|
||
if symbol_name not in self._doc_strings:
|
||
return None
|
||
docstring: str | None = self._doc_strings[symbol_name]
|
||
if docstring is None or docstring == '':
|
||
return ir.Constant(ir.PointerType(ir.IntType(8)), None)
|
||
mangled: str = self._mangle_name(f"{symbol_name}.__doc__")
|
||
gv: ir.GlobalVariable | None = self.module.globals.get(mangled)
|
||
if gv is None:
|
||
# 跨模块 docstring：创建本地 internal linkage 只读字符串常量副本
|
||
self._create_string_global(docstring, name=mangled)
|
||
gv = self.module.globals.get(mangled)
|
||
if gv is None:
|
||
return ir.Constant(ir.PointerType(ir.IntType(8)), None)
|
||
gv.global_constant = True
|
||
zero: ir.Constant = ir.Constant(ir.IntType(32), 0)
|
||
return self.builder.gep(gv, [zero, zero], name=f"{symbol_name}_doc_ptr")
|
||
|
||
def _apply_bswap_if_big(self, val: ir.Value, byte_order: str, name: str) -> ir.Value:
|
||
"""大端字节序时对整数应用 bswap，否则原值返回。
|
||
|
||
仅对 16/32/64 位整数生效，统一 bswap 逻辑。
|
||
"""
|
||
if byte_order == 'big' and isinstance(val.type, ir.IntType) and val.type.width in (16, 32, 64):
|
||
return self.builder.bswap(val, name=name)
|
||
return val
|
||
|
||
def _zero_value_for_type(self, llvm_type: ir.Type) -> ir.Constant | None:
|
||
"""为指定 LLVM 类型生成零值常量。
|
||
|
||
合并原 HandlesAssign._zero_value 与 StructGen._zero_value_for_type 的逻辑。
|
||
返回 ir.Constant 或 None（无法生成零值的类型，如 opaque struct）。
|
||
"""
|
||
if isinstance(llvm_type, ir.IntType):
|
||
return ir.Constant(llvm_type, 0)
|
||
elif isinstance(llvm_type, (ir.FloatType, ir.DoubleType)):
|
||
return ir.Constant(llvm_type, 0.0)
|
||
elif isinstance(llvm_type, ir.PointerType):
|
||
return ir.Constant(llvm_type, None)
|
||
elif isinstance(llvm_type, ir.ArrayType):
|
||
elem_zv: ir.Constant | None = self._zero_value_for_type(llvm_type.element)
|
||
if elem_zv is None:
|
||
return None
|
||
return ir.Constant(llvm_type, [elem_zv] * llvm_type.count)
|
||
elif isinstance(llvm_type, ir.BaseStructType):
|
||
if llvm_type.elements is not None and len(llvm_type.elements) > 0:
|
||
elem_zvs: list[ir.Constant | None] = [self._zero_value_for_type(m) for m in llvm_type.elements]
|
||
if any(zv is None for zv in elem_zvs):
|
||
return None
|
||
return ir.Constant(llvm_type, elem_zvs)
|
||
return None
|
||
return None
|
||
|
||
# ========== 函数热点统计（--func-hotspot） ==========
|
||
|
||
def _UniqueGlobalName(self, base: str) -> str:
|
||
"""返回本模块内未被占用的全局符号名（追加下划线直到不冲突）。
|
||
|
||
仅用于 --func-hotspot 自动生成的内部符号，避免与用户全局/函数重名。
|
||
"""
|
||
name: str = base
|
||
while name in self.module.globals:
|
||
name += '_'
|
||
return name
|
||
|
||
def _DeclareExternalFunc(self, name: str, ftype: ir.FunctionType) -> ir.Function:
|
||
"""获取或声明外部 C 函数（用于 --func-hotspot 所需的 Win32/CRT 接口）。
|
||
|
||
已存在同名函数时直接复用；若已存在符号类型不一致则显式报错（不静默降级）。
|
||
"""
|
||
existing: Any = self.module.globals.get(name)
|
||
if existing is not None:
|
||
if not isinstance(existing, ir.Function):
|
||
raise RuntimeError(
|
||
f'--func-hotspot: 全局符号 {name} 已存在且不是函数，无法声明外部接口')
|
||
if existing.ftype != ftype:
|
||
raise RuntimeError(
|
||
f'--func-hotspot: 外部函数 {name} 已存在但签名不一致'
|
||
f'（期望 {ftype}，实际 {existing.ftype}）')
|
||
return existing
|
||
return ir.Function(self.module, ftype, name=name)
|
||
|
||
def _EmitFuncHotspotCounters(self) -> tuple[list[ir.Function], list[ir.GlobalVariable], list[ir.Constant]]:
|
||
"""为模块内每个函数定义安装 internal i64 计数器并在入口递增。
|
||
|
||
返回 (被计数的函数定义列表, 计数器列表, 函数名常量指针列表)。
|
||
|
||
抽出自原 _InstallFuncHotspot，供两种注册策略复用：
|
||
- 种子 CLI 单模块：在 main 入口注册处理器（原行为）；
|
||
- Projectrans 多模块：每模块用 @llvm.global_ctors 自注册。
|
||
计数器一律 internal，跨模块（独立翻译单元）链接不产生重复符号。
|
||
"""
|
||
i64: ir.IntType = ir.IntType(64)
|
||
# 收集模块内所有函数定义（只有声明没有基本块的属于 declare，跳过）
|
||
defs: list[ir.Function] = [f for f in self.module.functions if getattr(f, 'blocks', None)]
|
||
if not defs:
|
||
raise RuntimeError('--func-hotspot: 模块内没有任何函数定义，无法安装计数')
|
||
counters: list[ir.GlobalVariable] = []
|
||
name_ptrs: list[ir.Constant] = []
|
||
for idx, func in enumerate(defs):
|
||
# 计数器全局：唯一命名的 internal i64，初值 0
|
||
counter: ir.GlobalVariable = ir.GlobalVariable(
|
||
self.module, i64, name=self._UniqueGlobalName(f'__fhs_cnt_{idx}'))
|
||
counter.linkage = 'internal'
|
||
counter.initializer = ir.Constant(i64, 0)
|
||
counters.append(counter)
|
||
# 可读函数名（IR 中真实符号名）写入只读字符串常量，供处理器打印
|
||
name_ptrs.append(self._create_string_global(
|
||
func.name, name=self._UniqueGlobalName(f'__fhs_name_{idx}')))
|
||
# 入口块最前递增：load -> add 1 -> store（置于最前，不破坏原有跳转）
|
||
entry: ir.Block = func.entry_basic_block
|
||
b: ir.IRBuilder = ir.IRBuilder(entry)
|
||
b.position_at_start(entry)
|
||
cur: ir.Value = b.load(counter, name=f'fhs_load_{idx}')
|
||
inc: ir.Value = b.add(cur, ir.Constant(i64, 1), name=f'fhs_inc_{idx}')
|
||
b.store(inc, counter)
|
||
return defs, counters, name_ptrs
|
||
|
||
def _CreateFuncHotspotHandler(
|
||
self,
|
||
path: str,
|
||
counters: list[ir.GlobalVariable],
|
||
name_ptrs: list[ir.Constant],
|
||
module_id: str | None = None,
|
||
) -> ir.Function:
|
||
"""生成 Win32 控制台 Ctrl+C 处理器：把所有计数器追加写入文件。
|
||
|
||
- 处理器签名为 Win32 PHANDLER_ROUTINE（i32 (i32) -> BOOL）
|
||
- 用 CRT fopen("a")/fprintf/fclose 落盘：多模块各自持有处理器，运行时各
|
||
自以追加方式写同一个 <path>，互不覆盖（见 _InstallFuncHotspotModuleLocal）
|
||
- module_id 非空时首行先写 "# module <module_id>\n"，用于区分不同模块的计数段
|
||
- 每行格式为 "函数名 计数\n"
|
||
- 返回 FALSE(0)，让默认处理继续，使 Ctrl+C 仍能终止程序
|
||
"""
|
||
i32: ir.IntType = ir.IntType(32)
|
||
i8p: ir.PointerType = ir.PointerType(ir.IntType(8))
|
||
fopen: ir.Function = self._DeclareExternalFunc('fopen', ir.FunctionType(i8p, [i8p, i8p]))
|
||
fprintf: ir.Function = self._DeclareExternalFunc(
|
||
'fprintf', ir.FunctionType(i32, [i8p, i8p], var_arg=True))
|
||
fclose: ir.Function = self._DeclareExternalFunc('fclose', ir.FunctionType(i32, [i8p]))
|
||
handler: ir.Function = ir.Function(
|
||
self.module, ir.FunctionType(i32, [i32]),
|
||
name=self._UniqueGlobalName('__fhs_handler'))
|
||
handler.linkage = 'internal'
|
||
b: ir.IRBuilder = ir.IRBuilder(handler.append_basic_block(name='entry'))
|
||
b.position_at_start(handler.entry_basic_block)
|
||
path_ptr: ir.Constant = self._create_string_global(
|
||
path, name=self._UniqueGlobalName('__fhs_path'))
|
||
# 追加模式 "a"：多模块多处理器各自追加到同一个 <path>，互不覆盖
|
||
mode_ptr: ir.Constant = self._create_string_global(
|
||
'a', name=self._UniqueGlobalName('__fhs_mode'))
|
||
fmt_ptr: ir.Constant = self._create_string_global(
|
||
'%s %llu\n', name=self._UniqueGlobalName('__fhs_fmt'))
|
||
fp: ir.Value = b.call(fopen, [path_ptr, mode_ptr], name='fhs_fp')
|
||
if module_id:
|
||
hdr_fmt_ptr: ir.Constant = self._create_string_global(
|
||
'# module %s\n', name=self._UniqueGlobalName('__fhs_hdr_fmt'))
|
||
mid_ptr: ir.Constant = self._create_string_global(
|
||
module_id, name=self._UniqueGlobalName('__fhs_module_id'))
|
||
b.call(fprintf, [fp, hdr_fmt_ptr, mid_ptr], name='fhs_write_module')
|
||
for gv, name_ptr in zip(counters, name_ptrs):
|
||
b.call(fprintf, [fp, fmt_ptr, name_ptr, b.load(gv)], name='fhs_write')
|
||
b.call(fclose, [fp])
|
||
b.ret(ir.Constant(i32, 0))
|
||
return handler
|
||
|
||
def _RegisterFuncHotspotViaCtor(self, handler: ir.Function) -> None:
|
||
"""用 @llvm.global_ctors 在进程启动期为本模块注册 Ctrl+C 处理器。
|
||
|
||
生成 internal void @__fhs_ctor()（内部调用 SetConsoleCtrlHandler），
|
||
并把它写入 appending linkage 的 @llvm.global_ctors —— 每个模块一份，
|
||
链接器按 appending 语义把各模块的构造器拼接进 .init_array，进程启动时
|
||
各执行一次。相比"在 main 入口注册"，本方案不依赖 main 所在模块，
|
||
适配 Projectrans 的"每源文件独立翻译单元"模型。
|
||
|
||
模块内若已存在 @llvm.global_ctors 视为冲突，显式报错（不静默降级）。
|
||
"""
|
||
i32: ir.IntType = ir.IntType(32)
|
||
i8p: ir.PointerType = ir.PointerType(ir.IntType(8))
|
||
# 构造器元素类型 { i32, void ()*, i8* }：优先级、函数指针、关联数据(NULL)
|
||
void_fnptr: ir.PointerType = ir.PointerType(ir.FunctionType(ir.VoidType(), []))
|
||
set_handler: ir.Function = self._DeclareExternalFunc(
|
||
'SetConsoleCtrlHandler', ir.FunctionType(i32, [i8p, i32]))
|
||
ctor: ir.Function = ir.Function(
|
||
self.module, ir.FunctionType(ir.VoidType(), []),
|
||
name=self._UniqueGlobalName('__fhs_ctor'))
|
||
ctor.linkage = 'internal'
|
||
b: ir.IRBuilder = ir.IRBuilder(ctor.append_basic_block(name='entry'))
|
||
b.call(set_handler, [b.bitcast(handler, i8p), ir.Constant(i32, 1)])
|
||
b.ret_void()
|
||
if 'llvm.global_ctors' in self.module.globals:
|
||
raise RuntimeError(
|
||
'--func-hotspot: 模块内已存在 @llvm.global_ctors，无法注册热点处理器（不静默降级）')
|
||
elem_ty: ir.LiteralStructType = ir.LiteralStructType([i32, void_fnptr, i8p])
|
||
arr_ty: ir.ArrayType = ir.ArrayType(elem_ty, 1)
|
||
entry_val: ir.Constant = ir.Constant(elem_ty, [
|
||
ir.Constant(i32, 65535),
|
||
# 传函数引用字符串（非 Function 对象）：与 _create_string_global 同法，
|
||
# 否则 llvmlite 会以 str(Function) 渲染整个 define 体
|
||
ir.Constant(void_fnptr, ctor.get_reference()),
|
||
ir.Constant(i8p, None),
|
||
])
|
||
gv: ir.GlobalVariable = ir.GlobalVariable(self.module, arr_ty, name='llvm.global_ctors')
|
||
gv.linkage = 'appending'
|
||
gv.initializer = ir.Constant(arr_ty, [entry_val])
|
||
|
||
def _InstallFuncHotspotTruncate(
|
||
self,
|
||
defs: list[ir.Function],
|
||
path: str,
|
||
require_main: bool = False,
|
||
flag: str = '--func-hotspot',
|
||
) -> ir.Function | None:
|
||
"""在含 main 的模块的 main 入口插入一次 "w" 打开→关闭，清空计数文件。
|
||
|
||
--func-hotspot/--hotspot-sites 的处理器以追加（"a"）方式写 <path>，若不清空会
|
||
累积历史运行的数据。清空必须满足两点：
|
||
- 只在含 main 的模块执行一次（其它模块无 main，返回 None 且不截断）；
|
||
- 必须在 main 入口而非构造器：@llvm.global_ctors 的跨模块执行顺序无保证，
|
||
放构造器里可能在其它模块已写入后清空 → 丢数据。
|
||
main 入口在程序启动、任何 Ctrl+C 处理器写入之前执行，故每次运行都以空文件
|
||
开始，配合追加写保证"一次运行一份结果"。
|
||
|
||
flag 仅用于报错信息，调用方可传 '--hotspot-sites'。
|
||
|
||
返回找到的 main 定义（供调用方继续注册处理器）；未找到时返回 None。
|
||
require_main=True（种子 CLI 单模块）时未找到 main 直接报错（不静默降级）。
|
||
"""
|
||
main_func: ir.Function | None = None
|
||
for func in defs:
|
||
if func.name == 'main':
|
||
main_func = func
|
||
break
|
||
if main_func is None:
|
||
if require_main:
|
||
raise RuntimeError(
|
||
f'{flag}: 找不到 main 函数定义，无法清空计数文件')
|
||
return None
|
||
i32: ir.IntType = ir.IntType(32)
|
||
i8p: ir.PointerType = ir.PointerType(ir.IntType(8))
|
||
fopen: ir.Function = self._DeclareExternalFunc('fopen', ir.FunctionType(i8p, [i8p, i8p]))
|
||
fclose: ir.Function = self._DeclareExternalFunc('fclose', ir.FunctionType(i32, [i8p]))
|
||
path_ptr: ir.Constant = self._create_string_global(
|
||
path, name=self._UniqueGlobalName('__fhs_trunc_path'))
|
||
mode_ptr: ir.Constant = self._create_string_global(
|
||
'w', name=self._UniqueGlobalName('__fhs_trunc_mode'))
|
||
b: ir.IRBuilder = ir.IRBuilder(main_func.entry_basic_block)
|
||
b.position_at_start(main_func.entry_basic_block)
|
||
fp: ir.Value = b.call(fopen, [path_ptr, mode_ptr], name='fhs_trunc_fp')
|
||
b.call(fclose, [fp])
|
||
return main_func
|
||
|
||
def _InstallFuncHotspot(self) -> None:
|
||
"""安装函数热点统计（--func-hotspot）：种子 CLI 单模块路径。
|
||
|
||
在 main 入口最前：先用 "w" 打开一次 <path> 清空历史计数，再注册 Ctrl+C
|
||
处理器（程序启动期各执行一次）。仅在 func_hotspot_file 非 None（用户传入
|
||
--func-hotspot）时被调用；未启用时该方法不会执行，IR 输出与既有行为一致。
|
||
"""
|
||
path: str | None = self.func_hotspot_file
|
||
if not path:
|
||
raise RuntimeError('--func-hotspot: 需要提供计数器输出文件路径')
|
||
i32: ir.IntType = ir.IntType(32)
|
||
i8p: ir.PointerType = ir.PointerType(ir.IntType(8))
|
||
defs, counters, name_ptrs = self._EmitFuncHotspotCounters()
|
||
handler: ir.Function = self._CreateFuncHotspotHandler(
|
||
path, counters, name_ptrs, module_id=self.module_sha1 or None)
|
||
# main 入口清空历史计数文件（require_main=True → 无 main 显式报错）
|
||
main_func: ir.Function | None = self._InstallFuncHotspotTruncate(
|
||
defs, path, require_main=True)
|
||
if main_func is None:
|
||
raise RuntimeError(
|
||
'--func-hotspot: 找不到 main 函数定义，无法注册 Ctrl+C 处理器')
|
||
# 在 main 入口注册处理器（清空之后、程序 body 之前执行）
|
||
mb: ir.IRBuilder = ir.IRBuilder(main_func.entry_basic_block)
|
||
mb.position_at_start(main_func.entry_basic_block)
|
||
set_handler: ir.Function = self._DeclareExternalFunc(
|
||
'SetConsoleCtrlHandler', ir.FunctionType(i32, [i8p, i32]))
|
||
handler_ptr: ir.Value = mb.bitcast(handler, i8p)
|
||
mb.call(set_handler, [handler_ptr, ir.Constant(i32, 1)])
|
||
|
||
def _InstallFuncHotspotModuleLocal(self) -> None:
|
||
"""安装函数热点统计（--func-hotspot）：Projectrans 多模块路径。
|
||
|
||
每个模块是独立翻译单元、各自 llc 后链接，无法跨模块共享同一处理器，
|
||
故每模块自带 internal 计数器 + internal 处理器（统一追加写最终 <path>），
|
||
并各自通过 @llvm.global_ctors 在进程启动期注册 SetConsoleCtrlHandler。
|
||
所有新增符号均 internal，跨模块不产生重复符号；多个处理器在 Ctrl+C 时
|
||
由系统 LIFO 依次调用、各自以追加（"a"）方式写同一文件，互不覆盖。
|
||
仅含 main 的模块额外在 main 入口以 "w" 清空一次文件（见
|
||
_InstallFuncHotspotTruncate），保证"一次运行一份结果"。未启用
|
||
（func_hotspot_file 为 None）时本方法不会被调用。
|
||
"""
|
||
path: str | None = self.func_hotspot_file
|
||
if not path:
|
||
raise RuntimeError('--func-hotspot: 需要提供计数器输出文件路径')
|
||
module_id: str = getattr(self, 'module_sha1', None) or ''
|
||
defs, counters, name_ptrs = self._EmitFuncHotspotCounters()
|
||
handler: ir.Function = self._CreateFuncHotspotHandler(
|
||
path, counters, name_ptrs, module_id=module_id or None)
|
||
self._RegisterFuncHotspotViaCtor(handler)
|
||
# 非 main 模块无 main、不需要截断（返回 None 不报错）：其它模块的写入
|
||
# 依赖含 main 模块在启动期清空，不会互相覆盖。
|
||
self._InstallFuncHotspotTruncate(defs, path)
|
||
|
||
# ========== 调用点热点统计（--hotspot-sites） ==========
|
||
|
||
def _NewIrBuilder(self, block: ir.Block) -> ir.IRBuilder:
|
||
"""创建 IR 构建器：仅在 --hotspot-sites 启用时挂钩源码位置。
|
||
|
||
未启用（hotspot_sites_file 为 None）时返回原生 ir.IRBuilder，行为与既有
|
||
完全一致（零影响，不挂任何属性）；启用时返回 _HotspotSiteBuilder，仅在
|
||
每条新生成的 call 指令对象上追加 _src_file/_src_line 两个普通属性，供
|
||
调用点热点插桩读取源码位置。
|
||
"""
|
||
if self.hotspot_sites_file:
|
||
return _HotspotSiteBuilder(block, self)
|
||
return ir.IRBuilder(block)
|
||
|
||
@staticmethod
|
||
def _HotspotCalleeBase(name: str) -> str:
|
||
"""返回被调函数名的基名：去掉 16 位 sha1 模块前缀。
|
||
|
||
仅当形如 "<16位十六进制>.xxx" 时才剥离，避免误伤普通函数名；
|
||
例如 "23499a69276862b2.strcmp" -> "strcmp"，"strlen" -> "strlen"。
|
||
"""
|
||
if len(name) > 17 and name[16] == '.':
|
||
prefix: str = name[:16]
|
||
if all(ch in '0123456789abcdefABCDEF' for ch in prefix):
|
||
return name[17:]
|
||
return name
|
||
|
||
def _EmitHotspotSiteCounters(
|
||
self,
|
||
) -> tuple[list[ir.Function], list[tuple[ir.GlobalVariable, str, str, str, int]]]:
|
||
"""为匹配白名单的每个直接 call 调用点安装 internal i64 计数器并在 call 之前递增。
|
||
|
||
返回 (模块内函数定义列表, 记录列表)；记录为
|
||
(计数器全局, 调用方函数名, 被调方基名, 源码文件名, 源码行号)，
|
||
记录顺序即计数器编号 __fhs_site_<K> 的顺序。源码位置来自生成期挂在
|
||
call 指令对象上的 _src_file/_src_line（见 _HotspotSiteBuilder）；生成期
|
||
无法关联到 AST 的合成调用点取不到位置，则文件名留空、行号记 0。
|
||
|
||
插桩规则：
|
||
- 白名单：hotspot_sites_all=True 时不过滤（全部直接调用点）；否则仅对
|
||
"被调名去 16 位 sha1 前缀后的基名" 命中白名单的调用点插桩。
|
||
- 跳过间接调用（callee 不是具名 ir.Function）：无法归属被调方，故不插桩。
|
||
- 跳过 __fhs_ 前缀的辅助函数自身，避免在插桩生成的代码里计数
|
||
（--func-hotspot 的处理器/构造器也在此列）。
|
||
- 只在 call 之前插入 load/add/store，不改变返回值/参数/控制流。
|
||
- 计数器一律 internal，跨模块（独立翻译单元）链接不产生重复符号，初值 0。
|
||
"""
|
||
i64: ir.IntType = ir.IntType(64)
|
||
if self.hotspot_sites_all:
|
||
whitelist: set[str] | None = None
|
||
else:
|
||
names: list[str] = (
|
||
list(self.hotspot_callees) if self.hotspot_callees is not None
|
||
else list(DEFAULT_HOTSPOT_CALLEES)
|
||
)
|
||
whitelist = {self._HotspotCalleeBase(n) for n in names}
|
||
defs: list[ir.Function] = [f for f in self.module.functions if getattr(f, 'blocks', None)]
|
||
records: list[tuple[ir.GlobalVariable, str, str, str, int]] = []
|
||
for func in defs:
|
||
if func.name.startswith('__fhs_'):
|
||
continue
|
||
for block in list(func.blocks):
|
||
for instr in list(block.instructions):
|
||
if not isinstance(instr, ir.CallInstr):
|
||
continue
|
||
callee: Any = instr.callee
|
||
if not isinstance(callee, ir.Function) or not callee.name:
|
||
continue
|
||
base: str = self._HotspotCalleeBase(callee.name)
|
||
if whitelist is not None and base not in whitelist:
|
||
continue
|
||
idx: int = len(records)
|
||
counter: ir.GlobalVariable = ir.GlobalVariable(
|
||
self.module, i64, name=self._UniqueGlobalName(f'__fhs_site_{idx}'))
|
||
counter.linkage = 'internal'
|
||
counter.initializer = ir.Constant(i64, 0)
|
||
# 在 call 之前递增计数：load -> add 1 -> store（不改变 call 本身）
|
||
b: ir.IRBuilder = ir.IRBuilder(block)
|
||
b.position_before(instr)
|
||
cur: ir.Value = b.load(counter, name=f'fhs_site_load_{idx}')
|
||
inc: ir.Value = b.add(cur, ir.Constant(i64, 1), name=f'fhs_site_inc_{idx}')
|
||
b.store(inc, counter)
|
||
# 源码位置：生成期 _HotspotSiteBuilder 挂在 call 指令对象上；缺失则留空
|
||
src_file: str = getattr(instr, '_src_file', None) or ''
|
||
src_line: int = getattr(instr, '_src_line', 0) or 0
|
||
records.append((counter, func.name, base, src_file, src_line))
|
||
return defs, records
|
||
|
||
def _CreateHotspotSitesHandler(
|
||
self,
|
||
path: str,
|
||
records: list[tuple[ir.GlobalVariable, str, str, str, int]],
|
||
module_id: str | None = None,
|
||
) -> ir.Function:
|
||
"""生成 Win32 控制台 Ctrl+C 处理器：按调用次数降序把各调用点追加写入 <path>。
|
||
|
||
落盘格式（追加）：
|
||
# sites <module_sha1>\\n
|
||
<module_sha1> <caller_func> -> <callee_base> <count> <file>:<line>\\n
|
||
module_id 为空时省略行首 sha1 与头部 sha1（留空，不伪造）。
|
||
|
||
- 排序：落盘前用选择排序（两层运行期循环，IR 规模 O(N)、运行期 O(N^2) 仅
|
||
在 Ctrl+C 触发一次）按计数降序输出，只输出 count>0 的调用点；计数相同时
|
||
按计数器编号 K 升序（稳定、可复现）。
|
||
- 源码位置：作为可选尾字段，格式 "<file>:<line>"（含前置单个空格）；该位置
|
||
来自生成期挂在 call 指令对象上的 _src_file/_src_line（见 _HotspotSiteBuilder）。
|
||
生成期无法关联到 AST 的合成调用点取不到位置，则此字段整体省略（不伪造、不填默认值）。
|
||
- 处理器签名为 Win32 PHANDLER_ROUTINE（i32 (i32) -> BOOL），用 CRT
|
||
fopen("a")/fprintf/fclose 落盘（沿用既有做法），返回 FALSE(0) 让默认
|
||
处理继续，使 Ctrl+C 仍能终止程序。
|
||
"""
|
||
n: int = len(records)
|
||
i8: ir.IntType = ir.IntType(8)
|
||
i32: ir.IntType = ir.IntType(32)
|
||
i64: ir.IntType = ir.IntType(64)
|
||
i8p: ir.PointerType = ir.PointerType(i8)
|
||
i64p: ir.PointerType = ir.PointerType(i64)
|
||
c0_8: ir.Constant = ir.Constant(i8, 0)
|
||
c1_8: ir.Constant = ir.Constant(i8, 1)
|
||
c0_32: ir.Constant = ir.Constant(i32, 0)
|
||
c0_64: ir.Constant = ir.Constant(i64, 0)
|
||
c1_64: ir.Constant = ir.Constant(i64, 1)
|
||
fopen: ir.Function = self._DeclareExternalFunc('fopen', ir.FunctionType(i8p, [i8p, i8p]))
|
||
fprintf: ir.Function = self._DeclareExternalFunc(
|
||
'fprintf', ir.FunctionType(i32, [i8p, i8p], var_arg=True))
|
||
fclose: ir.Function = self._DeclareExternalFunc('fclose', ir.FunctionType(i32, [i8p]))
|
||
handler: ir.Function = ir.Function(
|
||
self.module, ir.FunctionType(i32, [i32]),
|
||
name=self._UniqueGlobalName('__fhs_sites_handler'))
|
||
handler.linkage = 'internal'
|
||
b: ir.IRBuilder = ir.IRBuilder(handler.append_basic_block(name='entry'))
|
||
b.position_at_start(handler.entry_basic_block)
|
||
# 无匹配调用点：生成空处理器（不落盘、不建数组），仅占位以保持注册链路一致
|
||
if n == 0:
|
||
b.ret(ir.Constant(i32, 0))
|
||
return handler
|
||
path_ptr: ir.Constant = self._create_string_global(
|
||
path, name=self._UniqueGlobalName('__fhs_sites_path'))
|
||
mode_ptr: ir.Constant = self._create_string_global(
|
||
'a', name=self._UniqueGlobalName('__fhs_sites_mode'))
|
||
mid_ptr: ir.Constant | None = None
|
||
hdr_fmt_ptr: ir.Constant | None = None
|
||
if module_id:
|
||
fmt_ptr: ir.Constant = self._create_string_global(
|
||
'%s %s -> %s %llu%s\n', name=self._UniqueGlobalName('__fhs_sites_fmt'))
|
||
hdr_fmt_ptr = self._create_string_global(
|
||
'# sites %s\n', name=self._UniqueGlobalName('__fhs_sites_hdr_fmt'))
|
||
mid_ptr = self._create_string_global(
|
||
module_id, name=self._UniqueGlobalName('__fhs_sites_module_id'))
|
||
else:
|
||
fmt_ptr = self._create_string_global(
|
||
'%s -> %s %llu%s\n', name=self._UniqueGlobalName('__fhs_sites_fmt'))
|
||
fp: ir.Value = b.call(fopen, [path_ptr, mode_ptr], name='fhs_sites_fp')
|
||
if module_id and hdr_fmt_ptr is not None:
|
||
b.call(fprintf, [fp, hdr_fmt_ptr, mid_ptr], name='fhs_sites_write_hdr')
|
||
# 归属元数据全局数组：运行期按索引取 "调用方名 / 被调方基名 / 计数器地址"
|
||
caller_ty: ir.ArrayType = ir.ArrayType(i8p, n)
|
||
callee_ty: ir.ArrayType = ir.ArrayType(i8p, n)
|
||
cntptr_ty: ir.ArrayType = ir.ArrayType(i64p, n)
|
||
caller_gv: ir.GlobalVariable = ir.GlobalVariable(
|
||
self.module, caller_ty, name=self._UniqueGlobalName('__fhs_sites_caller'))
|
||
caller_gv.linkage = 'internal'
|
||
caller_gv.global_constant = True
|
||
caller_gv.initializer = ir.Constant(caller_ty, [
|
||
self._create_string_global(rec[1], name=self._UniqueGlobalName(f'__fhs_sites_caller_{k}'))
|
||
for k, rec in enumerate(records)
|
||
])
|
||
callee_gv: ir.GlobalVariable = ir.GlobalVariable(
|
||
self.module, callee_ty, name=self._UniqueGlobalName('__fhs_sites_callee'))
|
||
callee_gv.linkage = 'internal'
|
||
callee_gv.global_constant = True
|
||
callee_gv.initializer = ir.Constant(callee_ty, [
|
||
self._create_string_global(rec[2], name=self._UniqueGlobalName(f'__fhs_sites_callee_{k}'))
|
||
for k, rec in enumerate(records)
|
||
])
|
||
# 源码位置全局数组：每项形如 " <file>:<line>"（含前置空格）；取不到位置则为空串
|
||
loc_ty: ir.ArrayType = ir.ArrayType(i8p, n)
|
||
loc_gv: ir.GlobalVariable = ir.GlobalVariable(
|
||
self.module, loc_ty, name=self._UniqueGlobalName('__fhs_sites_loc'))
|
||
loc_gv.linkage = 'internal'
|
||
loc_gv.global_constant = True
|
||
loc_gv.initializer = ir.Constant(loc_ty, [
|
||
self._create_string_global(
|
||
f' {rec[3]}:{rec[4]}' if (rec[3] and rec[4] > 0) else '',
|
||
name=self._UniqueGlobalName(f'__fhs_sites_loc_{k}'))
|
||
for k, rec in enumerate(records)
|
||
])
|
||
cnt_gv: ir.GlobalVariable = ir.GlobalVariable(
|
||
self.module, cntptr_ty, name=self._UniqueGlobalName('__fhs_sites_cnt_ptrs'))
|
||
cnt_gv.linkage = 'internal'
|
||
cnt_gv.global_constant = True
|
||
cnt_gv.initializer = ir.Constant(cntptr_ty, [
|
||
ir.Constant(i64p, rec[0].get_reference()) for rec in records
|
||
])
|
||
# 全部 alloca 置于 entry 块：避免在循环体内 alloca 导致每次迭代爆栈
|
||
used: ir.Value = b.alloca(ir.ArrayType(i8, n), name='fhs_sites_used')
|
||
for i in range(n):
|
||
b.store(c0_8, b.gep(used, [c0_32, ir.Constant(i32, i)]))
|
||
k_ptr: ir.Value = b.alloca(i64, name='fhs_sites_k')
|
||
best_ptr: ir.Value = b.alloca(i64, name='fhs_sites_best')
|
||
bestv_ptr: ir.Value = b.alloca(i64, name='fhs_sites_bestv')
|
||
i_ptr: ir.Value = b.alloca(i64, name='fhs_sites_i')
|
||
b.store(c0_64, k_ptr)
|
||
# 选择排序：外层 k 走 0..n-1，每轮在未输出项里选计数最大者（计数相同取 K 小者）
|
||
loop_k: ir.Block = handler.append_basic_block(name='fhs_sites_sort_k')
|
||
body_k: ir.Block = handler.append_basic_block(name='fhs_sites_sort_k_body')
|
||
loop_i: ir.Block = handler.append_basic_block(name='fhs_sites_sort_i')
|
||
body_i: ir.Block = handler.append_basic_block(name='fhs_sites_sort_i_body')
|
||
scan_i: ir.Block = handler.append_basic_block(name='fhs_sites_sort_i_scan')
|
||
upd_i: ir.Block = handler.append_basic_block(name='fhs_sites_sort_i_upd')
|
||
next_i: ir.Block = handler.append_basic_block(name='fhs_sites_sort_i_next')
|
||
end_i: ir.Block = handler.append_basic_block(name='fhs_sites_sort_i_end')
|
||
print_k: ir.Block = handler.append_basic_block(name='fhs_sites_print_k')
|
||
after_k: ir.Block = handler.append_basic_block(name='fhs_sites_sort_end')
|
||
b.branch(loop_k)
|
||
# loop_k: if k < n -> body_k else after_k
|
||
b.position_at_end(loop_k)
|
||
k_val: ir.Value = b.load(k_ptr)
|
||
b.cbranch(b.icmp_signed('<', k_val, ir.Constant(i64, n)), body_k, after_k)
|
||
# body_k: best=-1, bestv=0, i=0; -> loop_i
|
||
b.position_at_end(body_k)
|
||
b.store(ir.Constant(i64, -1), best_ptr)
|
||
b.store(c0_64, bestv_ptr)
|
||
b.store(c0_64, i_ptr)
|
||
b.branch(loop_i)
|
||
# loop_i: if i < n -> body_i else end_i
|
||
b.position_at_end(loop_i)
|
||
i_val: ir.Value = b.load(i_ptr)
|
||
b.cbranch(b.icmp_signed('<', i_val, ir.Constant(i64, n)), body_i, end_i)
|
||
# body_i: if used[i]==0 -> scan_i else next_i
|
||
b.position_at_end(body_i)
|
||
used_i: ir.Value = b.load(b.gep(used, [c0_32, i_val]))
|
||
b.cbranch(b.icmp_signed('==', used_i, c0_8), scan_i, next_i)
|
||
# scan_i: v = *cnt_ptrs[i]; if v > bestv -> upd_i else next_i
|
||
b.position_at_end(scan_i)
|
||
cnt_p: ir.Value = b.load(b.gep(cnt_gv, [c0_32, i_val]))
|
||
v_val: ir.Value = b.load(cnt_p)
|
||
b.cbranch(b.icmp_signed('>', v_val, b.load(bestv_ptr)), upd_i, next_i)
|
||
# upd_i: bestv=v, best=i; -> next_i
|
||
b.position_at_end(upd_i)
|
||
b.store(v_val, bestv_ptr)
|
||
b.store(i_val, best_ptr)
|
||
b.branch(next_i)
|
||
# next_i: i++; -> loop_i
|
||
b.position_at_end(next_i)
|
||
b.store(b.add(b.load(i_ptr), c1_64), i_ptr)
|
||
b.branch(loop_i)
|
||
# end_i: best<0（剩余全为 0 计数）-> after_k；否则 print_k
|
||
b.position_at_end(end_i)
|
||
best_val: ir.Value = b.load(best_ptr)
|
||
b.cbranch(b.icmp_signed('<', best_val, c0_64), after_k, print_k)
|
||
# print_k: used[best]=1; fprintf(...); k++; -> loop_k
|
||
b.position_at_end(print_k)
|
||
idx: ir.Value = b.load(best_ptr)
|
||
b.store(c1_8, b.gep(used, [c0_32, idx]))
|
||
caller_ptr: ir.Value = b.load(b.gep(caller_gv, [c0_32, idx]))
|
||
callee_ptr: ir.Value = b.load(b.gep(callee_gv, [c0_32, idx]))
|
||
loc_ptr: ir.Value = b.load(b.gep(loc_gv, [c0_32, idx]))
|
||
if module_id and mid_ptr is not None:
|
||
b.call(fprintf, [fp, fmt_ptr, mid_ptr, caller_ptr, callee_ptr,
|
||
b.load(bestv_ptr), loc_ptr],
|
||
name='fhs_sites_write')
|
||
else:
|
||
b.call(fprintf, [fp, fmt_ptr, caller_ptr, callee_ptr,
|
||
b.load(bestv_ptr), loc_ptr],
|
||
name='fhs_sites_write')
|
||
b.store(b.add(b.load(k_ptr), c1_64), k_ptr)
|
||
b.branch(loop_k)
|
||
# after_k: fclose; return 0
|
||
b.position_at_end(after_k)
|
||
b.call(fclose, [fp])
|
||
b.ret(ir.Constant(i32, 0))
|
||
return handler
|
||
|
||
def _RegisterHotspotSitesViaCtor(self, handler: ir.Function) -> None:
|
||
"""为本模块的调用点热点处理器注册 Ctrl+C（复用/追加 @llvm.global_ctors）。
|
||
|
||
生成 internal void @__fhs_sites_ctor()（内部调用 SetConsoleCtrlHandler），
|
||
并写入 appending linkage 的 @llvm.global_ctors：
|
||
- 模块内尚无该全局时新建（单模块/纯 --hotspot-sites 场景）；
|
||
- 已存在且为本功能兼容的 appending 结构数组时追加一条（常见于同时启用
|
||
--func-hotspot：其构造器已建好 @llvm.global_ctors）；
|
||
- 已存在但不兼容时显式报错（不静默降级）。
|
||
"""
|
||
i32: ir.IntType = ir.IntType(32)
|
||
i8p: ir.PointerType = ir.PointerType(ir.IntType(8))
|
||
void_fnptr: ir.PointerType = ir.PointerType(ir.FunctionType(ir.VoidType(), []))
|
||
set_handler: ir.Function = self._DeclareExternalFunc(
|
||
'SetConsoleCtrlHandler', ir.FunctionType(i32, [i8p, i32]))
|
||
ctor: ir.Function = ir.Function(
|
||
self.module, ir.FunctionType(ir.VoidType(), []),
|
||
name=self._UniqueGlobalName('__fhs_sites_ctor'))
|
||
ctor.linkage = 'internal'
|
||
cb: ir.IRBuilder = ir.IRBuilder(ctor.append_basic_block(name='entry'))
|
||
cb.call(set_handler, [cb.bitcast(handler, i8p), ir.Constant(i32, 1)])
|
||
cb.ret_void()
|
||
elem_ty: ir.LiteralStructType = ir.LiteralStructType([i32, void_fnptr, i8p])
|
||
entry_val: ir.Constant = ir.Constant(elem_ty, [
|
||
ir.Constant(i32, 65535),
|
||
ir.Constant(void_fnptr, ctor.get_reference()),
|
||
ir.Constant(i8p, None),
|
||
])
|
||
existing: Any = self.module.globals.get('llvm.global_ctors')
|
||
if existing is None:
|
||
arr_ty: ir.ArrayType = ir.ArrayType(elem_ty, 1)
|
||
gv: ir.GlobalVariable = ir.GlobalVariable(self.module, arr_ty, name='llvm.global_ctors')
|
||
gv.linkage = 'appending'
|
||
gv.initializer = ir.Constant(arr_ty, [entry_val])
|
||
return
|
||
value_ty: Any = getattr(existing, 'value_type', None)
|
||
if (not isinstance(existing, ir.GlobalVariable)
|
||
or existing.linkage != 'appending'
|
||
or not isinstance(value_ty, ir.ArrayType)
|
||
or value_ty.element != elem_ty
|
||
or existing.initializer is None):
|
||
raise RuntimeError(
|
||
'--hotspot-sites: 模块内已存在不兼容的 @llvm.global_ctors，'
|
||
'无法注册调用点热点处理器（不静默降级）')
|
||
old_entries: list = list(existing.initializer.constant)
|
||
new_ty: ir.ArrayType = ir.ArrayType(elem_ty, len(old_entries) + 1)
|
||
existing.type = ir.PointerType(new_ty)
|
||
existing.value_type = new_ty
|
||
existing.initializer = ir.Constant(new_ty, old_entries + [entry_val])
|
||
|
||
def _InstallHotspotSites(self) -> None:
|
||
"""安装调用点热点统计（--hotspot-sites）：种子 CLI 单模块路径。
|
||
|
||
在 main 入口最前先用 "w" 打开一次 <path> 清空历史计数，再注册 Ctrl+C
|
||
处理器（程序启动期各执行一次），与 _InstallFuncHotspot 同法。仅在
|
||
hotspot_sites_file 非 None（用户传入 --hotspot-sites）时被调用；未启用时
|
||
该方法不会执行，IR 输出与既有行为一致（零影响）。
|
||
"""
|
||
path: str | None = self.hotspot_sites_file
|
||
if not path:
|
||
raise RuntimeError('--hotspot-sites: 需要提供调用点计数输出文件路径')
|
||
i32: ir.IntType = ir.IntType(32)
|
||
i8p: ir.PointerType = ir.PointerType(ir.IntType(8))
|
||
defs, records = self._EmitHotspotSiteCounters()
|
||
handler: ir.Function = self._CreateHotspotSitesHandler(
|
||
path, records, module_id=self.module_sha1 or None)
|
||
main_func: ir.Function | None = self._InstallFuncHotspotTruncate(
|
||
defs, path, require_main=True, flag='--hotspot-sites')
|
||
if main_func is None:
|
||
raise RuntimeError(
|
||
'--hotspot-sites: 找不到 main 函数定义，无法注册 Ctrl+C 处理器')
|
||
mb: ir.IRBuilder = ir.IRBuilder(main_func.entry_basic_block)
|
||
mb.position_at_start(main_func.entry_basic_block)
|
||
set_handler: ir.Function = self._DeclareExternalFunc(
|
||
'SetConsoleCtrlHandler', ir.FunctionType(i32, [i8p, i32]))
|
||
mb.call(set_handler, [mb.bitcast(handler, i8p), ir.Constant(i32, 1)])
|
||
|
||
def _InstallHotspotSitesModuleLocal(self) -> None:
|
||
"""安装调用点热点统计（--hotspot-sites）：Projectrans 多模块路径。
|
||
|
||
每模块自带 internal 调用点计数器 + internal 处理器（统一追加写最终 <path>），
|
||
并各自通过 @llvm.global_ctors 在进程启动期注册 SetConsoleCtrlHandler
|
||
（见 _RegisterHotspotSitesViaCtor；若 --func-hotspot 已建该全局则追加复用）。
|
||
仅含 main 的模块额外在 main 入口以 "w" 清空一次文件，保证"一次运行一份结果"。
|
||
未启用（hotspot_sites_file 为 None）时本方法不会被调用。
|
||
"""
|
||
path: str | None = self.hotspot_sites_file
|
||
if not path:
|
||
raise RuntimeError('--hotspot-sites: 需要提供调用点计数输出文件路径')
|
||
module_id: str = getattr(self, 'module_sha1', None) or ''
|
||
defs, records = self._EmitHotspotSiteCounters()
|
||
handler: ir.Function = self._CreateHotspotSitesHandler(
|
||
path, records, module_id=module_id or None)
|
||
self._RegisterHotspotSitesViaCtor(handler)
|
||
# 非 main 模块无 main、不需要截断（返回 None 不报错）：其它模块的写入
|
||
# 依赖含 main 模块在启动期清空，不会互相覆盖。
|
||
self._InstallFuncHotspotTruncate(defs, path, flag='--hotspot-sites') |