from __future__ import annotations import sys import os import re import shutil import subprocess import vast as ast import traceback import json import pickle import time import cProfile from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed, wait, FIRST_COMPLETED from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode from lib.core.Handles.HandlesBase import CTypeInfo from lib.includes import t from lib.core.SymbolUtils import AnnotationContainsName from lib.Projectrans.Utils import compute_sha1, _check_annotation_for_export, find_reachable_files_from_entries, topological_sort_files, parse_python_file from lib.Projectrans.Phase1Generator import _TryGlobalCache, _SaveToGlobalCache, _GLOBAL_CACHE_DIR import TransPyC from StubGen import PythonToStubConverter def _ExtractCDefineConstantsFromPyi(pyi_path: str, all_dc: dict[str, object]) -> None: """从 .pyi 文件中提取 CDefine 常量到 all_dc。文件不存在时静默跳过。 治本修复:原实现只支持 ast.Constant 和 float() 调用, 无法处理 t.CUnsignedLong(-11)(ast.Call)和 FOREGROUND_GREEN | FOREGROUND_INTENSITY(ast.BinOp)等复杂表达式。 现使用 ConstEvaluator.eval_full 统一求值,支持: - ast.Constant: 简单常量 (0x0002) - ast.Call: 类型构造 (t.CUnsignedLong(-11)) - ast.BinOp: 位运算 (FOREGROUND_RED | FOREGROUND_GREEN) - ast.UnaryOp: 一元运算 (-11) - ast.Name: 引用其他 CDefine 常量 """ if not os.path.exists(pyi_path): return try: # 复用 parse_python_file 的 AST 缓存(.pyi 文件被多处解析) _, tree = parse_python_file(pyi_path) if tree is None: return # 治本修复:使用 ConstEvaluator 处理复杂表达式 # 创建 mock Gen 对象,其 _define_constants 指向 all_dc, # 使 ConstEvaluator 能查找同文件中已提取的常量(如 FOREGROUND_RED 引用 FOREGROUND_BLUE) from lib.core.ConstEvaluator import ConstEvaluator, EvalContext class _MockGen: """轻量 mock,仅提供 _define_constants 供 ConstEvaluator 查找""" def __init__(self) -> None: self._define_constants: dict[str, object] = all_dc mock_gen: _MockGen = _MockGen() ctx: EvalContext = EvalContext(Gen=mock_gen) for node in ast.iter_child_nodes(tree): if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): if AnnotationContainsName(node.annotation, 'CDefine') and node.value: val: object = ConstEvaluator.eval_full(node.value, ctx) if val is not None and isinstance(val, (int, float, str)): all_dc[node.target.id] = val except Exception as e: _vlog().warning(f"Phase2Translator: 忽略异常 {e}", exc_info=e) def _parallel_translate_worker(src_path: str, out_path: str, src_root: str, temp_dir: str, output_dir: str, include_dirs: list[str], triple: str | None, datalayout: str | None, sha1_map: dict[str, str], sig_files: dict[str, str], stub_files: dict[str, str], include_py_map: dict[str, str], slice_level: int, shared_sym_pickle_path: str, struct_sha1_map: dict[str, str], prebuilt_ModuleSha1Map: dict[str, str] | None = None) -> tuple[str, str, str]: # 注意:不可跨任务复用 Phase2Translator/共享符号表!_translate_file 编译过程中 # 会向共享 SymbolTable 写入符号(import insert/类注册等),复用导致前文件的 # 符号污染后文件 → 跨模块 struct 类型前缀错位(如 7cadfe07e983fd90.CompilerFactory # 混入 _list 前缀,Type of #1 arg mismatch)。每任务必须从 pickle 反序列化出 # 独立副本(原实现;worker 复用省掉的 pickle.load 时间仅 ~0.5s/次,远小于 # worker 内翻译计算,不值得以正确性为代价)。 try: trans = Phase2Translator(src_root, temp_dir, output_dir, compile_cmd='llc', include_dirs=include_dirs, target_triple=triple, target_datalayout=datalayout, build_pyi_cache=False) trans.sha1_map = sha1_map trans.sig_files = sig_files trans.stub_files = stub_files trans.include_py_map = include_py_map trans.slice_level = slice_level trans.struct_sha1_map = struct_sha1_map if shared_sym_pickle_path and os.path.exists(shared_sym_pickle_path): # 使用 pickle.load(f) 流式加载,避免 pickle.loads(f.read()) 一次性 # 将整个 pickle 文件读入内存(磁盘副本 + 反序列化对象同时占用内存)导致 MemoryError。 # 同时提高递归限制,防止深层嵌套对象反序列化失败。 sys.setrecursionlimit(max(sys.getrecursionlimit(), 100000)) with open(shared_sym_pickle_path, 'rb') as f: shared_data = pickle.load(f) _sym_bytes: bytes | None = shared_data.get('symbol_table_bytes') if _sym_bytes is not None: trans._shared_symbol_table = pickle.loads(_sym_bytes) trans._shared_source_module_sig_files = shared_data['source_module_sig_files'] trans._shared_all_dc = shared_data['all_dc'] trans._shared_export_extern_funcs = shared_data['export_extern_funcs'] trans.inline_func_symbols = shared_data['inline_func_symbols'] trans.function_default_args = shared_data['function_default_args'] trans.function_param_names = shared_data.get('function_param_names', {}) trans._shared_generic_class_templates = shared_data.get('generic_class_templates', {}) trans._shared_doc_meta = shared_data.get('shared_doc_meta', {}) trans._shared_class_vtable = shared_data.get('class_vtable', set()) trans._shared_cross_module_vtable = shared_data.get('cross_module_vtable', set()) trans._shared_class_methods = shared_data.get('class_methods', {}) trans._shared_cross_module_novtable = shared_data.get('cross_module_novtable', set()) else: trans._precollect_inline_symbols() trans._build_shared_symbol_data() trans._translate_file(src_path, out_path, prebuilt_ModuleSha1Map=prebuilt_ModuleSha1Map) return ('ok', src_path, '') except Exception as e: # fail-fast(2026-08-18):include 文件编译出错立即终止整个编译进程 # (主进程收到 BrokenProcessPool 后取消剩余任务退出)。 sys.stderr.write( f"[FATAL] include 翻译失败,立即终止编译: {src_path}\n" f"{traceback.format_exc()}\n") sys.stderr.flush() os._exit(88) def _parallel_translate_batch(batch: list, src_root: str, temp_dir: str, output_dir: str, include_dirs: list[str], triple: str | None, datalayout: str | None, sha1_map: dict[str, str], sig_files: dict[str, str], stub_files: dict[str, str], include_py_map: dict[str, str], slice_level: int, shared_sym_pickle_path: str, struct_sha1_map: dict[str, str]) -> tuple[list, list]: """批次子任务:worker 内创建一个 Phase2Translator + pickle.load 一次,顺序翻译批内全部文件。 batch 每项: (i, src_path, out_path, sha1, prebuilt_ModuleSha1Map) 防污染:SymbolTable 是编译期唯一被写的共享对象(import insert/类注册), 批内每个文件翻译前从 symbol_table_bytes 反序列化出干净副本,前文件的 符号写入不会残留到后文件(否则跨模块 struct 类型前缀错位)。 返回 (错误列表[(src_path, err)], batch)。 """ try: trans = Phase2Translator(src_root, temp_dir, output_dir, compile_cmd='llc', include_dirs=include_dirs, target_triple=triple, target_datalayout=datalayout, build_pyi_cache=False) trans.sha1_map = sha1_map trans.sig_files = sig_files trans.stub_files = stub_files trans.include_py_map = include_py_map trans.slice_level = slice_level trans.struct_sha1_map = struct_sha1_map symbol_table_bytes: bytes | None = None if shared_sym_pickle_path and os.path.exists(shared_sym_pickle_path): sys.setrecursionlimit(max(sys.getrecursionlimit(), 100000)) with open(shared_sym_pickle_path, 'rb') as f: shared_data = pickle.load(f) symbol_table_bytes = shared_data.get('symbol_table_bytes') if symbol_table_bytes is not None: trans._shared_symbol_table = pickle.loads(symbol_table_bytes) trans._shared_source_module_sig_files = shared_data['source_module_sig_files'] trans._shared_all_dc = shared_data['all_dc'] trans._shared_export_extern_funcs = shared_data['export_extern_funcs'] trans.inline_func_symbols = shared_data['inline_func_symbols'] trans.function_default_args = shared_data['function_default_args'] trans.function_param_names = shared_data.get('function_param_names', {}) trans._shared_generic_class_templates = shared_data.get('generic_class_templates', {}) trans._shared_doc_meta = shared_data.get('shared_doc_meta', {}) trans._shared_class_vtable = shared_data.get('class_vtable', set()) trans._shared_cross_module_vtable = shared_data.get('cross_module_vtable', set()) trans._shared_class_methods = shared_data.get('class_methods', {}) trans._shared_cross_module_novtable = shared_data.get('cross_module_novtable', set()) else: trans._precollect_inline_symbols() trans._build_shared_symbol_data() batch_errors: list = [] for _i, src_path, out_path, _sha1, msm in batch: try: # 每文件前重置 SymbolTable 为干净副本(防跨文件污染) if symbol_table_bytes is not None: trans._shared_symbol_table = pickle.loads(symbol_table_bytes) trans._shared_symbol_table.translator = None trans._translate_file(src_path, out_path, prebuilt_ModuleSha1Map=msm) except Exception as e: # fail-fast(2026-08-18):任意文件编译出错立即终止整个编译进程, # 不再收集错误继续翻译批内剩余文件(避免浪费大量时间)。主进程 # 收到 BrokenProcessPool/future 异常后取消全部剩余批次退出。 sys.stderr.write( f"[FATAL] 文件翻译失败,立即终止编译: {src_path}\n" f"{traceback.format_exc()}\n") sys.stderr.flush() os._exit(88) return (batch_errors, batch) except Exception as e: return ([(item[1], f'{e}\n{traceback.format_exc()}') for item in batch], batch) # ================================================================== # 子进程性能剖析:主进程 cProfile 只能看到进程池等待(WaitForMultipleObjects/ # as_completed),看不到 worker 进程内部 CPU 分布(vast parse 等热点)。 # 固定启用(不设环境变量开关),worker 每个任务导出 cProfile dump # (worker_prof_{pid}_{seq}.prof)到 PROF_DIR(D:\Users\TermiNexus\Desktop\TransPyC\PROF), # 可用 pstats 合并分析。每次编译开始时由 _CleanProfFiles 清理该目录 *.prof。 # ================================================================== import functools as _functools # 硬编码 PROF 输出目录(2026-08-18 起不再用 TPC_WORKER_PROF 环境变量控制) PROF_DIR: str = r'D:\Users\TermiNexus\Desktop\TransPyC\PROF' def _CleanProfFiles() -> None: """清理 PROF_DIR 下全部 *.prof(每次编译开始调用)。""" try: if os.path.isdir(PROF_DIR): for _n in os.listdir(PROF_DIR): if _n.endswith('.prof'): try: os.remove(os.path.join(PROF_DIR, _n)) except OSError: pass except OSError: pass def _ProfileWorker(_fn): @_functools.wraps(_fn) def _wrapper(*args, **kwargs): _prof: cProfile.Profile = cProfile.Profile() _prof.enable() try: return _fn(*args, **kwargs) finally: _prof.disable() _seq: int = getattr(_wrapper, '_seq', 0) + 1 _wrapper._seq = _seq try: if not os.path.isdir(PROF_DIR): os.makedirs(PROF_DIR, exist_ok=True) _out: str = os.path.join(PROF_DIR, f'worker_prof_{os.getpid()}_{_seq}.prof') _prof.dump_stats(_out) except Exception: pass return _wrapper _parallel_translate_worker = _ProfileWorker(_parallel_translate_worker) _parallel_translate_batch = _ProfileWorker(_parallel_translate_batch) # ================================================================== # worker pyi 缓存预热:ProcessPoolExecutor(spawn) 下每个 worker 重新 # import 模块后,HandlesImports.ImportHandle 的类级 _pyi_cache 全空, # 每进程首次任务重新 vast.parse 全部依赖 .vpi/.pyi # (cProfile: 4479 次 parse / 245s)。 # 方案:主进程把 temp 下全部 .vpi 的 vast AST 树 pickle 落盘 # (temp/_pyi_cache_v1.pkl,任务 B),worker initializer 启动时 # pickle.load 一次填充 _pyi_cache(任务 A);load 失败回退逐文件 # vast.parse。文件系统共享即为"共用缓存",不做跨进程共享内存(任务 C)。 # 环境变量 TPC_WORKER_WARMUP=0 可关闭(回归对比基线)。 # ================================================================== _PYI_CACHE_PKL_NAME: str = '_pyi_cache_v1.pkl' _PYI_CACHE_SIG_NAME: str = '_pyi_cache_v1.sig.json' _PYI_CACHE_VERSION: int = 1 def _ScanPyiFiles(temp_dir: str) -> list[str]: """扫描 temp 目录下全部 .vpi/.pyi 文件,返回绝对路径列表(与 _TryLoadClassMembersFromPyi 的 _pyi_cache 键格式一致)。""" try: names: list[str] = os.listdir(temp_dir) except OSError: return [] paths: list[str] = [] for fname in sorted(names): if fname.startswith('_'): continue if fname.endswith('.vpi') or fname.endswith('.pyi'): paths.append(os.path.join(temp_dir, fname)) return paths def _PyiCacheSignature(paths: list[str]) -> list[tuple[str, int, float]]: """vpi 清单签名:(basename, size, mtime) 元组列表,用于缓存失效判断。""" sig: list[tuple[str, int, float]] = [] for p in paths: try: st = os.stat(p) sig.append((os.path.basename(p), st.st_size, st.st_mtime)) except OSError: sig.append((os.path.basename(p), -1, -1.0)) return sig def _LoadPyiCachePkl(temp_dir: str) -> dict | None: """pickle.load temp/_pyi_cache_v1.pkl,返回 {abs_path: vast树|None}。 文件不存在/损坏/版本不符时返回 None(调用方回退逐文件解析)。""" pkl_path: str = os.path.join(temp_dir, _PYI_CACHE_PKL_NAME) try: with open(pkl_path, 'rb') as f: data: dict = pickle.load(f) if not isinstance(data, dict) or data.get('version') != _PYI_CACHE_VERSION: return None cache = data.get('cache') if not isinstance(cache, dict): return None return cache except Exception: return None def _BuildPyiCacheFile(temp_dir: str) -> str | None: """构建/刷新 temp/_pyi_cache_v1.pkl:遍历全部 .vpi 做 vast.parse 后 pickle.dump(含版本号与签名)。签名(basename,size,mtime)未变时跳过。 返回 pkl 路径;失败(无 vpi / pickle 不兼容等)返回 None。 注意:vast AST 可 pickle 性由 Nodes.py 侧保证;若实际 pickle.dump 失败,此处静默回退,worker 的 initializer 将改为逐文件 vast.parse。""" paths: list[str] = _ScanPyiFiles(temp_dir) if not paths: return None sig: list[tuple[str, int, float]] = _PyiCacheSignature(paths) pkl_path: str = os.path.join(temp_dir, _PYI_CACHE_PKL_NAME) sig_path: str = os.path.join(temp_dir, _PYI_CACHE_SIG_NAME) try: if os.path.isfile(pkl_path) and os.path.isfile(sig_path): with open(sig_path, 'r', encoding='utf-8') as f: stored = json.load(f) # json 往返后 tuple→list,需规范化再比较 if stored == [[n, s, m] for n, s, m in sig]: return pkl_path except Exception: pass cache: dict[str, object] = {} # 增量复用:Phase2 过程会新增 include .vpi(签名每次变 → 全量重建 12.2s)。 # 加载旧 pkl,按 (basename, size, mtime) 匹配未变化的文件复用 vast 树, # 只对新增/变化的文件重新 parse。 old_by_name: dict[str, tuple[int, float, object | None]] = {} try: old = _LoadPyiCachePkl(temp_dir) if isinstance(old, dict): old_sig = old.get('signature') or [] old_trees = list((old.get('cache') or {}).values()) for _i, (_name, _size, _mtime) in enumerate(old_sig): _tree = old_trees[_i] if _i < len(old_trees) else None old_by_name[_name] = (_size, _mtime, _tree) except Exception: pass for p, (name, size, mtime) in zip(paths, sig): old_entry = old_by_name.get(name) if (old_entry is not None and old_entry[0] == size and old_entry[1] == mtime and old_entry[2] is not None): cache[p] = old_entry[2] continue try: with open(p, 'r', encoding='utf-8') as f: cache[p] = ast.parse(f.read()) except Exception: cache[p] = None try: with open(pkl_path, 'wb') as f: pickle.dump({'version': _PYI_CACHE_VERSION, 'signature': sig, 'cache': cache}, f) with open(sig_path, 'w', encoding='utf-8') as f: json.dump(sig, f) return pkl_path except Exception as e: _vlog().warning(f"Phase2Translator: pyi 缓存落盘失败({e}),worker 将回退逐文件解析") return None def _PyiCacheWorkerInitializer(temp_dir: str) -> None: """worker 进程启动时预热 ImportHandle._pyi_cache(类级缓存)。 spawn 下每 worker 重新 import 模块后 _pyi_cache 为空,首次任务会 重新 vast.parse 全部依赖 .vpi(4479 次 parse / 245s)。此处启动时 一次性填充:优先 pickle.load 主进程落盘的 _pyi_cache_v1.pkl; 失败(文件不存在/损坏/版本不符/pickle 不兼容)回退逐文件 vast.parse。 预热只解析 .vpi(pyi 树小);stub 索引保留 _build_stub_index 懒加载。 TPC_WORKER_WARMUP=0 时关闭(便于回归对比)。 """ if os.environ.get('TPC_WORKER_WARMUP', '1') != '1': return if not os.path.isdir(temp_dir): return from lib.core.Handles.HandlesImports import ImportHandle cache: dict | None = _LoadPyiCachePkl(temp_dir) if cache is not None: ImportHandle._pyi_cache = cache ImportHandle._pyi_cache_dir = os.path.abspath(temp_dir) return # 回退:逐文件 vast.parse 预热(worker 单线程,无竞争) try: new_cache: dict[str, object] = {} for p in _ScanPyiFiles(temp_dir): try: with open(p, 'r', encoding='utf-8') as f: new_cache[p] = ast.parse(f.read()) except Exception: new_cache[p] = None if new_cache: ImportHandle._pyi_cache = new_cache ImportHandle._pyi_cache_dir = os.path.abspath(temp_dir) except Exception: pass def _count_braces_outside_strings(text: str) -> int: """统计花括号差值,跳过 c"..." 字符串字面量内部的括号""" count = 0 i = 0 while i < len(text): if text[i:i+2] == 'c"' and (i == 0 or text[i-1] != '\\'): # 跳过 c"..." 字面量 i += 2 while i < len(text): if text[i] == '"' and (i == 0 or text[i-1] != '\\'): i += 1 break i += 1 continue if text[i] == '{': count += 1 elif text[i] == '}': count -= 1 i += 1 return count class Phase2Translator: """阶段二:使用声明接口翻译源文件""" INCLUDE_LIB_EXTENSIONS: tuple[str, ...] = ('.dll', '.ll', '.so', '.o', '.a', '.lib') INCLUDE_SRC_EXTENSIONS: tuple[str, ...] = ('.c', '.cpp', '.cc', '.cxx', '.m', '.mm') INCLUDE_DIRS: list[str] = [ os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), 'includes'), ] # include 源文件 sha1/存在性缓存(_build_current_ModuleSha1Map): # 编译期间 include .vp/.py 源文件不变,避免每文件重复 isfile + open + # compute_sha1(PROF 显示 9306 isfile + 4653 open + TextIOWrapper.read # 35.9s 的大头来源,99 次调用 × 全量 include 重算)。 _include_sha1_cache: dict[str, str] = {} _include_exist_cache: dict[str, bool] = {} def __init__(self, src_root: str, temp_dir: str, output_dir: str, compile_cmd: str = 'llc', compile_flags: list[str] | None = None, linker_cmd: str | None = None, linker_flags: list[str] | None = None, linker_output: str | None = None, include_dirs: list[str] | None = None, entry_files: list[str] | None = None, target_triple: str | None = None, target_datalayout: str | None = None, force_recompile_includes: bool = False, build_pyi_cache: bool = True) -> None: self.src_root = os.path.abspath(src_root) self.temp_dir = os.path.abspath(temp_dir) self.output_dir = os.path.abspath(output_dir) self.compile_cmd = compile_cmd self.compile_flags = compile_flags or ['-filetype=obj'] self.triple = target_triple self.datalayout = target_datalayout if not self.triple: for i, flag in enumerate(self.compile_flags): if flag.startswith('-mtriple='): self.triple = flag[len('-mtriple='):] elif flag == '-mtriple' and i + 1 < len(self.compile_flags): self.triple = self.compile_flags[i + 1] self.linker_cmd = linker_cmd self.linker_flags = linker_flags or [] self.linker_output = linker_output # ASan 模式:TPC_ASAN=1 时对全部 .ll 做 opt -passes=asan 插桩并链接 libasan。 # 检测 stdlib 堆/栈/全局越界与垃圾指针(MemBuddy pool 内部由 canary 覆盖)。 # 注意:mingw64 的 clang 不提供 ASan runtime(libclang_rt.asan 缺失), # 必须使用 clang64 环境(自带 asan_dynamic runtime),PATH 前置 clang64/bin。 self.asan_mode = os.environ.get('TPC_ASAN') == '1' if self.asan_mode: _clang64_bin = r'D:\msys64\clang64\bin' if os.path.isdir(_clang64_bin): os.environ['PATH'] = _clang64_bin + os.pathsep + os.environ.get('PATH', '') self.slice_level = 3 self.sha1_map: dict[str, str] = {} self.sig_files: dict[str, str] = {} self.stub_files: dict[str, str] = {} self.include_py_map: dict[str, str] = {} self.used_includes: set[str] = set() self._last_ErrorStack: list | None = None self.function_default_args: dict[str, list] = {} self.function_param_names: dict[str, list] = {} self.inline_func_symbols: dict[str, CTypeInfo] = {} self.extra_link_files: list[str] = [] self.extra_compile_files: list[str] = [] self.extra_py_files: list[str] = [] self._include_sha1s: set[str] = set() # 持有 ProcessPoolExecutor 引用,避免局部变量 GC 时 finalizer 阻塞 ~99s # (join_executor_internals 在 GC 时同步等待 worker 退出队列) self._executor_pool: ProcessPoolExecutor | None = None self.entry_files: list[str] | None = entry_files self.force_recompile_includes: bool = force_recompile_includes # pyi cache 构建开关:仅主进程编译启动时构建一次(_LoadSha1Map 内), # worker 子进程只读主进程落盘的 temp/_pyi_cache_v1.pkl(_PyiCacheWorkerInitializer), # 不重复构建——worker 每任务重建 pkl 的 pickle.load/dump(43 次 × ~0.5s ≈ 23s # + pickle.loads 10.4s)纯属浪费;worker 缺 includes vpi 时运行期按需 parse 回退。 self._build_pyi_cache: bool = build_pyi_cache self._shared_symbol_table: object | None = None self._shared_source_module_sig_files: dict[str, str] = {} self._shared_all_dc: dict[str, dict] = {} self._shared_export_extern_funcs: dict[str, str] = {} self.include_dirs: list[str] = [] self.struct_sha1_map: dict[str, str] = {} self._shared_generic_class_templates: dict[str, dict] = {} self._shared_doc_meta: dict[str, dict[str, str]] = {} # 共享 vtable 状态跨模块(修复 pool.alloc(48) 虚表分派失败导致段错误) self._shared_class_vtable: set[str] = set() self._shared_cross_module_vtable: set[str] = set() self._shared_class_methods: dict[str, list[str]] = {} self._shared_cross_module_novtable: set[str] = set() default_dirs = [d for d in self.INCLUDE_DIRS if os.path.isdir(d)] if include_dirs: seen = set() self.include_dirs = [] for d in list(include_dirs) + default_dirs: d = os.path.abspath(d) if d not in seen and os.path.isdir(d): self.include_dirs.append(d) seen.add(d) else: self.include_dirs = default_dirs self._LoadSha1Map() def _LoadSha1Map(self) -> None: """从 temp_dir 加载 SHA1 映射、签名文件和 stub 声明文件列表。 temp_dir 中缺失的文件会从全局缓存补充。 """ if not os.path.exists(self.temp_dir): _vlog().error(f"声明目录不存在: {self.temp_dir}") return for fname in os.listdir(self.temp_dir): fpath = os.path.join(self.temp_dir, fname) if fname.startswith('_'): continue if fname.endswith('.vpi') or fname.endswith('.pyi'): sha1 = fname[:-4] self.sig_files[sha1] = fpath elif fname.endswith('.stub.ll'): sha1 = fname[:-8] self.stub_files[sha1] = fpath # 从内存存储器读取 SHA1 映射(_sha1_map.txt 仅人类可读,不再作为机器数据源) # 延迟导入避免循环依赖(Config 导入 Phase2Translator) from lib.Projectrans.Config import get_sha1_map_store mem_store: dict[str, str] = get_sha1_map_store() if mem_store: for sha1, rel in mem_store.items(): self.sha1_map[sha1] = rel if rel.startswith('includes/'): module_name = os.path.splitext(os.path.basename(rel))[0] self.include_py_map[module_name] = sha1 else: # 内存存储器为空(独立运行 Phase2 的回退):从 stub 文件名推断 for sha1, stub_path in self.stub_files.items(): self.sha1_map[sha1] = sha1 # 从全局缓存补充 temp_dir 中缺失的 .vpi / .stub.ll。 # 只补充"当前编译有效"的 sha1(在 sha1_map 中):.transpyc_cache 会残留 # 旧版源码的缓存(如旧 __builder.vp 的 9026a429f5e2d0e0.pyi/.stub.ll 与新 # d9b3d06934cd1898 共存),不加过滤会把旧版 .vpi/.stub.ll 也加载进 # sig_files/stub_files → 同名类(IRBuilder)按旧 sha1 前缀注册 → # ld.lld undefined symbol: 9026a429f5e2d0e0.new_builder。 if os.path.isdir(_GLOBAL_CACHE_DIR): for fname in os.listdir(_GLOBAL_CACHE_DIR): if fname.startswith('_'): continue if fname.endswith('.vpi') or fname.endswith('.pyi'): sha1 = fname[:-4] if sha1 in self.sha1_map and sha1 not in self.sig_files: self.sig_files[sha1] = os.path.join(_GLOBAL_CACHE_DIR, fname) elif fname.endswith('.stub.ll'): sha1 = fname[:-8] if sha1 in self.sha1_map and sha1 not in self.stub_files: self.stub_files[sha1] = os.path.join(_GLOBAL_CACHE_DIR, fname) # 构建 pyi AST pickle 缓存(temp/_pyi_cache_v1.pkl)供 worker 预热: # Phase1 已生成全部 .vpi,此处把 vast 树落盘,避免每个 worker 首次任务 # 重新 vast.parse(4479 次 parse / 245s)。签名未变时零开销跳过。 # 仅主进程构建(build_pyi_cache=True);worker 只读 pkl,不重复构建。 if self._build_pyi_cache and os.environ.get('TPC_WORKER_WARMUP', '1') == '1': _BuildPyiCacheFile(self.temp_dir) def _build_struct_sha1_map(self) -> dict[str, str]: struct_sha1_map = {} valid_sha1_keys = set(self.sha1_map.keys()) # 收集 temp_dir 和全局缓存中的 .vpi/.pyi 文件 pyi_dirs: list[str] = [self.temp_dir] if os.path.isdir(_GLOBAL_CACHE_DIR): pyi_dirs.append(_GLOBAL_CACHE_DIR) for scan_dir in pyi_dirs: if not os.path.isdir(scan_dir): continue for fname in os.listdir(scan_dir): if not (fname.endswith('.vpi') or fname.endswith('.pyi')): continue sha1_key = fname.replace('.vpi', '').replace('.pyi', '') if sha1_key not in valid_sha1_keys: continue if sha1_key in struct_sha1_map: continue pyi_path = os.path.join(scan_dir, fname) # 复用 parse_python_file 的 AST 缓存(.pyi 文件可能被多处解析) _, tree = parse_python_file(pyi_path) if tree is None: continue for node in ast.walk(tree): if isinstance(node, ast.ClassDef): struct_sha1_map[node.name] = sha1_key return struct_sha1_map def run(self) -> None: """扫描源目录,翻译并生成 .ll 含代码文件(导入遍历)""" import sys as _sys if self.entry_files: py_files = list(self.entry_files) else: main_py = os.path.join(self.src_root, 'main.vp') if not os.path.exists(main_py): # 回退到 main.py(向后兼容) main_py = os.path.join(self.src_root, 'main.py') if os.path.exists(main_py): py_files = [main_py] else: py_files = [] for root, dirs, files in os.walk(self.src_root): dirs[:] = [d for d in dirs if d not in ('__pycache__',)] for file in files: if file.endswith('.vp') or file.endswith('.py'): py_files.append(os.path.join(root, file)) if not py_files: _vlog().info("未找到入口文件或源文件") return reachable = find_reachable_files_from_entries(self.src_root, py_files) py_files = list(reachable) py_files = topological_sort_files(py_files, self.src_root) _vlog().info(f"找到 {len(py_files)} 个可达源文件(已按依赖排序)") _vlog().info("处理顺序:") for i, f in enumerate(py_files[:10], 1): rel = os.path.relpath(f, self.src_root) _vlog().info(f" {i}. {rel}") os.makedirs(self.output_dir, exist_ok=True) valid_sha1s = set(self.sha1_map.keys()) old_stub_count = len(self.stub_files) old_sig_count = len(self.sig_files) self.stub_files = {k: v for k, v in self.stub_files.items() if k in valid_sha1s} self.sig_files = {k: v for k, v in self.sig_files.items() if k in valid_sha1s} if old_stub_count != len(self.stub_files) or old_sig_count != len(self.sig_files): _vlog().info(f"过滤旧文件: stub {old_stub_count}->{len(self.stub_files)}, sig {old_sig_count}->{len(self.sig_files)}") active_sha1s = set() self._precollect_inline_symbols() self._build_shared_symbol_data() self._precollect_vtable_state() self.struct_sha1_map = self._build_struct_sha1_map() need_translate = [] for i, src_path in enumerate(py_files, 1): rel = os.path.relpath(src_path, self.src_root) # 复用 parse_python_file 的 AST 缓存(get_file_dependencies 已 parse 过) content, tree = parse_python_file(src_path) sha1 = compute_sha1(content) active_sha1s.add(sha1) out_path = os.path.join(self.output_dir, f"{sha1}.ll") current_ModuleSha1Map = self._build_current_ModuleSha1Map(sha1) if tree is not None: try: for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: self.used_includes.add(alias.name.split('.')[0]) elif isinstance(node, ast.ImportFrom): if node.module: self.used_includes.add(node.module.split('.')[0]) # 检测 list/dict 容器使用,自动添加 _list/_dict/json 到 used_includes # 以便 _scan_include_libraries 能发现并编译这些库文件 if isinstance(node, ast.Subscript) and isinstance(node.value, ast.Name): if node.value.id == 'dict': self.used_includes.add('_dict') self.used_includes.add('json') elif node.value.id == 'list': self.used_includes.add('_list') if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): if node.func.id == 'dict': self.used_includes.add('_dict') self.used_includes.add('json') elif node.func.id == 'list': self.used_includes.add('_list') # 检测类型注解 d: dict / l: list (非泛型形式) if isinstance(node, ast.AnnAssign) and isinstance(node.annotation, ast.Name): if node.annotation.id == 'dict': self.used_includes.add('_dict') self.used_includes.add('json') elif node.annotation.id == 'list': self.used_includes.add('_list') except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"收集源文件导入的 include 模块失败: {_e}", "Exception") if os.path.isfile(out_path): deps_changed = self._check_deps_changed(sha1, current_ModuleSha1Map) if not deps_changed: _vlog().info(f"[{i}/{len(py_files)}] 跳过(缓存): {rel} -> {sha1}.ll") continue else: if self._recombine_ll(sha1, current_ModuleSha1Map): _vlog().info(f"[{i}/{len(py_files)}] 重组(依赖变化): {rel} -> {sha1}.ll") continue _vlog().info(f"[{i}/{len(py_files)}] 重译(依赖变化): {rel} -> {sha1}.ll") else: stub_cache = os.path.join(self.output_dir, f"{sha1}.stub.ll") text_cache = os.path.join(self.output_dir, f"{sha1}.text.ll") if os.path.isfile(stub_cache) and os.path.isfile(text_cache): if self._recombine_ll(sha1, current_ModuleSha1Map): _vlog().info(f"[{i}/{len(py_files)}] 重组(缓存恢复): {rel} -> {sha1}.ll") continue need_translate.append((i, src_path, out_path, sha1, current_ModuleSha1Map)) if need_translate: # 提前从 include 文件源码中提取函数参数名,供主项目编译时 keyword 参数匹配使用 self._extract_include_param_names_from_dirs() # 提前从 App 源文件中提取函数默认参数,供跨模块调用参数检查使用 # (避免并行编译时,被调用函数的文件尚未编译完,导致默认参数信息缺失) self._extract_app_default_args(need_translate) _vlog().info(f"\n需要翻译 {len(need_translate)} 个文件") t_start = time.time() n_workers = min(os.cpu_count() or 4, len(need_translate)) if n_workers > 1 and len(need_translate) > 1: _vlog().info(f" 并行翻译 (workers={n_workers})") shared_pickle_path = os.path.join(self.temp_dir, '_shared_sym.json') try: # symbol_table_bytes:SymbolTable 独立 pickle 字节,供批次 worker # 每文件前反序列化出干净副本(SymbolTable 是编译期唯一被写的共享 # 对象,批次内复用 trans 时必须在每个文件前重置,防止前文件 import # 写入的符号污染后文件 → 跨模块 struct 类型前缀错位)。 symbol_table_bytes: bytes = pickle.dumps(self._shared_symbol_table) shared_data = { 'symbol_table_bytes': symbol_table_bytes, 'source_module_sig_files': self._shared_source_module_sig_files, 'all_dc': self._shared_all_dc, 'export_extern_funcs': self._shared_export_extern_funcs, 'inline_func_symbols': self.inline_func_symbols, 'function_default_args': self.function_default_args, 'function_param_names': self.function_param_names, 'generic_class_templates': self._shared_generic_class_templates, 'shared_doc_meta': self._shared_doc_meta, 'class_vtable': self._shared_class_vtable, 'cross_module_vtable': self._shared_cross_module_vtable, 'class_methods': self._shared_class_methods, 'cross_module_novtable': self._shared_cross_module_novtable, } with open(shared_pickle_path, 'wb') as f: f.write(pickle.dumps(shared_data)) _vlog().info(f" 共享符号表已序列化 ({os.path.getsize(shared_pickle_path)//1024}KB)") except Exception as e: _vlog().warning(f"符号表序列化失败({e}),worker将自行构建") shared_pickle_path = '' errors = [] # 复用 self._executor_pool:避免局部变量 GC 时 finalizer 阻塞 ~99s # (join_executor_internals 在 GC 时同步等待 worker 退出队列) if self._executor_pool is None: # initializer 预热 ImportHandle._pyi_cache(pickle.load 落盘缓存), # 避免每个 worker 首次任务重新 vast.parse 全部依赖 .vpi self._executor_pool = ProcessPoolExecutor(max_workers=n_workers, initializer=_PyiCacheWorkerInitializer, initargs=(self.temp_dir,)) executor = self._executor_pool try: # 批次子任务模型:多个文件包装为一个子任务(batch), # 每 worker 处理一批——Phase2Translator 创建 + pickle.load 只做 # 一次,批内文件顺序翻译(每文件前重置 SymbolTable 防污染)。 # 相比"一文件一任务":减少进程池调度/通信/构造开销 # (200+ 文件 → n_workers 个批次任务)。 batches: list[list] = [] # 按源文件大小加权分桶(LPT 贪心:最大文件优先放入当前总成本最小的桶), # 使各批次总成本均匀。原按文件数均分(batch_size=ceil(N/n))导致大文件 # 集中的批次拖长尾,其他 worker 提前完成空等(任务颗粒度不均匀)。 if len(need_translate) <= n_workers: batches = [[item] for item in need_translate] else: sized_items: list = [] for item in need_translate: try: _size: int = os.path.getsize(item[1]) except OSError: _size = 0 sized_items.append((item, _size)) buckets: list[list] = [[] for _ in range(n_workers)] bucket_cost: list[int] = [0] * n_workers for item, _size in sorted(sized_items, key=lambda t: t[1], reverse=True): target: int = min(range(n_workers), key=lambda w: bucket_cost[w]) buckets[target].append(item) bucket_cost[target] += _size batches = [b for b in buckets if b] futures = {} for bidx, batch in enumerate(batches): rel_first = os.path.relpath(batch[0][1], self.src_root) future = executor.submit( _parallel_translate_batch, batch, self.src_root, self.temp_dir, self.output_dir, self.include_dirs, self.triple, self.datalayout, self.sha1_map, self.sig_files, self.stub_files, self.include_py_map, self.slice_level, shared_pickle_path, self.struct_sha1_map, ) futures[future] = (bidx, rel_first, len(batch)) done_files = 0 total_files = len(need_translate) for future in as_completed(futures): bidx, rel_first, bcount = futures[future] try: batch_errors, batch = future.result() done_files += len(batch) for src, err in batch_errors: rel = os.path.relpath(src, self.src_root) _vlog().error(f" {rel}: {err}") errors.append((rel, err)) _vlog().info(f" [{done_files}/{total_files}] 批次完成({bcount} 文件, 首个 {rel_first})") except Exception as e: # fail-fast:worker 已 os._exit(文件编译出错)或进程池 # 已 broken → 立即取消全部剩余批次并终止编译,不空等。 _vlog().error(f" 批次 {bidx} 异常(worker 已终止): {e}") for fut in futures: if not fut.done(): fut.cancel() if self._executor_pool is not None: self._executor_pool.shutdown(wait=False, cancel_futures=True) self._executor_pool = None raise RuntimeError( f"批次 {bidx} worker 异常终止,编译立即停止: {e}") except Exception: # 异常时立即清理,正常流程不 shutdown(留给 includes 阶段复用) if self._executor_pool is not None: self._executor_pool.shutdown(wait=False, cancel_futures=True) self._executor_pool = None raise if errors: _vlog().error(f"\n{len(errors)} 个文件翻译失败,立即终止编译") for rel, err in errors: _vlog().error(f" {rel}: {err}") raise RuntimeError(f"{len(errors)} 个文件翻译失败,已终止编译") else: for idx, (i, src_path, out_path, sha1, msm) in enumerate(need_translate): rel = os.path.relpath(src_path, self.src_root) t0 = time.time() try: self._translate_file(src_path, out_path, prebuilt_ModuleSha1Map=msm) t1 = time.time() _vlog().info(f" [{idx+1}/{len(need_translate)}] {rel} ({t1-t0:.1f}s)") except Exception as e: _vlog().error(f" {rel}: {e}") traceback.print_exc() raise RuntimeError(f"翻译 {rel} 失败,已终止编译: {e}") from e t_end = time.time() _vlog().info(f" 翻译总耗时: {t_end-t_start:.1f}s") _vlog().info(f"\n[阶段二完成] .ll 文件生成到: {self.output_dir}") self._scan_include_libraries() self._compile_include_py_files(active_sha1s) self._compile_ll_files(active_sha1s) self._compile_include_sources() if self.linker_cmd: self._link_obj_files(active_sha1s) # 清理 ProcessPoolExecutor:所有翻译已完成,shutdown(wait=False) 让 worker # 后台退出。避免 GC finalizer 在程序末尾阻塞 ~99s(join_executor_internals) if self._executor_pool is not None: self._executor_pool.shutdown(wait=False, cancel_futures=True) self._executor_pool = None def _build_current_ModuleSha1Map(self, self_sha1: str | None) -> dict[str, str]: """构建当前模块的依赖 SHA1 映射(与 _translate_file 中逻辑一致)""" ModuleSha1Map = {} for sha1_key, sig_path in self.sig_files.items(): if sha1_key == self_sha1: continue rel_path = self.sha1_map.get(sha1_key, sha1_key) if rel_path.startswith('includes/'): inc_mod_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') inc_mod_name = inc_mod_name[len('includes/'):] inc_short_name = inc_mod_name.split('.')[-1] if '.' in inc_mod_name else inc_mod_name actual_sha1 = sha1_key for inc_dir in self.include_dirs: if os.path.isdir(inc_dir): py_path = os.path.join(inc_dir, rel_path[len('includes/'):].replace(os.sep, '/')) py_path = os.path.splitext(py_path)[0] + '.vp' # 存在性/内容缓存:编译期 include 源文件不变,命中跳过磁盘 I/O exists = Phase2Translator._include_exist_cache.get(py_path) if exists is None: exists = os.path.isfile(py_path) Phase2Translator._include_exist_cache[py_path] = exists if not exists: # 回退到 .py(向后兼容) py_path = os.path.splitext(py_path)[0] + '.py' exists = Phase2Translator._include_exist_cache.get(py_path) if exists is None: exists = os.path.isfile(py_path) Phase2Translator._include_exist_cache[py_path] = exists if exists: py_sha1 = Phase2Translator._include_sha1_cache.get(py_path) if py_sha1 is None: with open(py_path, 'r', encoding='utf-8') as f: py_sha1 = compute_sha1(f.read()) Phase2Translator._include_sha1_cache[py_path] = py_sha1 actual_sha1 = py_sha1 break ModuleSha1Map[inc_mod_name] = actual_sha1 ModuleSha1Map[inc_short_name] = actual_sha1 # Also add parent package name (e.g., 'hashlib' for 'hashlib.__init__') if inc_short_name == '__init__' and '.' in inc_mod_name: parent_pkg = inc_mod_name.rsplit('.', 1)[0] ModuleSha1Map[parent_pkg] = actual_sha1 continue module_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') ModuleSha1Map[module_name] = sha1_key short_name = module_name.split('.')[-1] if '.' in module_name else module_name ModuleSha1Map[short_name] = sha1_key # 包入口文件:注册父包名(如 'os.__init__' -> 'os'),使 import os 能解析到包级符号 if short_name == '__init__' and '.' in module_name: parent_pkg = module_name.rsplit('.', 1)[0] ModuleSha1Map[parent_pkg] = sha1_key return ModuleSha1Map @staticmethod def _extract_llvm_type_name(line: str) -> str | None: """从 LLVM IR 类型定义行提取类型名(字符串操作,不使用正则) 支持 list[str]、dict[str, int] 等含特殊字符的类型名。 例如: %"18e6990aa81cc5b6.list[str]" = type { ... } → "18e6990aa81cc5b6.list[str]" %struct.foo = type { ... } → "struct.foo" """ stripped = line.strip() if not stripped.startswith('%'): return None # 处理带引号的情况: %"name" = type ... if stripped.startswith('%"'): end_quote = stripped.find('"', 2) if end_quote > 0: return stripped[2:end_quote] return None # 处理不带引号的情况: %name = type ... eq_pos = stripped.find('=') if eq_pos > 1: return stripped[1:eq_pos].strip() return None @staticmethod def _find_all_type_refs(text: str) -> set[str]: """查找文本中所有 %"..."/%name 类型引用(字符串操作,不使用正则) 支持 list[str]、dict[str, int] 等含特殊字符的类型名。 """ refs: set[str] = set() i: int = 0 n: int = len(text) while i < n: if text[i] == '%': if i + 1 < n and text[i + 1] == '"': # 带引号: %"name" end: int = text.find('"', i + 2) if end > 0: name: str = text[i + 2:end] if name: refs.add(name) i = end + 1 continue else: # 不带引号: %name (仅匹配字母、数字、下划线、点) j: int = i + 1 while j < n and (text[j].isalnum() or text[j] in '._'): j += 1 if j > i + 1: refs.add(text[i + 1:j]) i = j continue i += 1 return refs @staticmethod def _split_ll(ll_content: str) -> tuple[str, str]: """将 .ll 内容拆分为 stub(声明)和 text(代码)两部分 stub: target, 注释, %type = type, declare, @xxx = external global text: define ... { ... }, @xxx = global/constant (带初始化器) """ stub_lines = [] text_lines = [] in_define = False brace_depth = 0 module_sha1 = None defined_names = set() for line in ll_content.splitlines(True): stripped = line.strip() if stripped.startswith('define '): dm = re.match(r'define\s+[^@]*@"?([a-f0-9]+)\.', stripped) if dm and module_sha1 is None: module_sha1 = dm.group(1) dm2 = re.match(r'define\s+[^@]*@"?([^"\s(]+)"?', stripped) if dm2: defined_names.add(dm2.group(1)) in_global_def = False global_var_name = None global_type_parts = [] global_brace_depth = 0 for line in ll_content.splitlines(True): stripped = line.strip() if in_global_def: text_lines.append(line) global_type_parts.append(stripped) global_brace_depth += _count_braces_outside_strings(stripped) if global_brace_depth <= 0: full_type = ' '.join(global_type_parts).strip() depth = 0 type_end = len(full_type) for ci, cc in enumerate(full_type): if cc in ('{', '['): depth += 1 elif cc in ('}', ']'): depth -= 1 elif depth == 0 and cc == ' ': type_end = ci break var_type = full_type[:type_end].strip() if var_type: stub_lines.append(f'{global_var_name} = external global {var_type}\n') in_global_def = False global_var_name = None global_type_parts = [] global_brace_depth = 0 continue if in_define: text_lines.append(line) brace_depth += _count_braces_outside_strings(stripped) if brace_depth <= 0: in_define = False continue if stripped.startswith('define '): text_lines.append(line) in_define = True brace_depth = _count_braces_outside_strings(stripped) decl_line = re.sub(r'^define\s+', 'declare ', stripped) decl_line = re.sub(r'\b(linkonce_odr|weak_odr|linkonce|weak|common|appending|internal|private)\s+', '', decl_line) decl_line = re.sub(r'\balwaysinline\s+', '', decl_line) paren_pos = decl_line.rfind(')') if paren_pos >= 0: decl_line = decl_line[:paren_pos + 1] stub_lines.append(decl_line + '\n') continue if stripped == '{' and text_lines and not in_define: prev_stripped = text_lines[-1].strip() if text_lines else '' if prev_stripped.startswith('define ') or prev_stripped.endswith('noredzone') or prev_stripped.endswith(')'): text_lines.append(line) brace_depth = 1 in_define = True if prev_stripped.startswith('define '): decl_line = re.sub(r'^define\s+', 'declare ', prev_stripped) decl_line = re.sub(r'\b(linkonce_odr|weak_odr|linkonce|weak|common|appending|internal|private)\s+', '', decl_line) decl_line = re.sub(r'\balwaysinline\s+', '', decl_line) paren_pos = decl_line.rfind(')') if paren_pos >= 0: decl_line = decl_line[:paren_pos + 1] stub_lines.append(decl_line + '\n') continue if stripped.startswith('@') and ' global ' in stripped and 'external global' not in stripped: text_lines.append(line) if ' internal ' not in stripped and ' private ' not in stripped: name_match = re.match(r'(@\S+)', stripped) if name_match: var_name = name_match.group(1) global_pos = stripped.find(' global ') after_global = stripped[global_pos + 8:] depth = 0 type_end = len(after_global) for ci, cc in enumerate(after_global): if cc in ('{', '['): depth += 1 elif cc in ('}', ']'): depth -= 1 elif depth == 0 and cc == ' ': type_end = ci break var_type = after_global[:type_end].strip() brace_depth_gv = 0 for cc in var_type: if cc == '{': brace_depth_gv += 1 elif cc == '}': brace_depth_gv -= 1 if brace_depth_gv > 0: in_global_def = True global_var_name = var_name global_type_parts = [var_type] global_brace_depth = brace_depth_gv elif var_type: stub_lines.append(f'{var_name} = external global {var_type}\n') continue if stripped.startswith('@') and ' constant ' in stripped: text_lines.append(line) continue if stripped.startswith('declare'): dm = re.match(r'declare\s+[^@]*@"?([^"\s(]+)"?', stripped) if dm: fname = dm.group(1) if fname in defined_names: stub_lines.append(line) continue dot_pos = fname.find('.') if dot_pos > 0: fname_sha1 = fname[:dot_pos] if module_sha1 and fname_sha1 != module_sha1 and len(fname_sha1) >= 16: continue if re.match(r'^[a-f0-9]{16}\.', fname): continue stub_lines.append(line) continue stub_lines.append(line) return ''.join(stub_lines), ''.join(text_lines) def _save_deps(self, sha1: str, ModuleSha1Map: dict[str, str]) -> None: """保存依赖指纹到 .deps.json""" deps_path = os.path.join(self.output_dir, f"{sha1}.deps.json") deps = {} for mod_name, dep_sha1 in ModuleSha1Map.items(): deps[mod_name] = dep_sha1 with open(deps_path, 'w', encoding='utf-8') as f: json.dump(deps, f) def _LoadDeps(self, sha1: str) -> dict[str, str] | None: """加载依赖指纹""" deps_path = os.path.join(self.output_dir, f"{sha1}.deps.json") if os.path.isfile(deps_path): try: with open(deps_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"加载依赖指纹 JSON 失败: {_e}", "Exception") return None def _check_deps_changed(self, sha1: str, current_ModuleSha1Map: dict[str, str]) -> bool: """检查依赖指纹是否变化""" saved_deps = self._LoadDeps(sha1) if saved_deps is None: return True for mod_name, dep_sha1 in current_ModuleSha1Map.items(): if saved_deps.get(mod_name) != dep_sha1: return True for mod_name, dep_sha1 in saved_deps.items(): if current_ModuleSha1Map.get(mod_name) != dep_sha1: return True return False def _recombine_ll(self, sha1: str, current_ModuleSha1Map: dict[str, str]) -> bool: """依赖变化时重新组合 .ll:更新 .stub.ll 中的 SHA1 前缀,拼接 .stub.ll + .text.ll""" saved_deps = self._LoadDeps(sha1) if saved_deps is None: return False stub_path = os.path.join(self.output_dir, f"{sha1}.stub.ll") text_path = os.path.join(self.output_dir, f"{sha1}.text.ll") ll_path = os.path.join(self.output_dir, f"{sha1}.ll") if not os.path.isfile(stub_path) or not os.path.isfile(text_path): return False with open(stub_path, 'r', encoding='utf-8') as f: stub_content = f.read() with open(text_path, 'r', encoding='utf-8') as f: text_content = f.read() sha1_replacements = {} for mod_name, old_sha1 in saved_deps.items(): new_sha1 = current_ModuleSha1Map.get(mod_name) if new_sha1 and old_sha1 != new_sha1: sha1_replacements[old_sha1] = new_sha1 if sha1_replacements: for old_sha1, new_sha1 in sha1_replacements.items(): stub_content = stub_content.replace(old_sha1, new_sha1) text_content = text_content.replace(old_sha1, new_sha1) combined = stub_content if not combined.endswith('\n'): combined += '\n' combined += text_content with open(ll_path, 'w', encoding='utf-8', newline='\n') as f: f.write(combined) with open(stub_path, 'w', encoding='utf-8', newline='\n') as f: f.write(stub_content) with open(text_path, 'w', encoding='utf-8', newline='\n') as f: f.write(text_content) self._save_deps(sha1, current_ModuleSha1Map) obj_path = os.path.join(self.output_dir, f"{sha1}.obj") if os.path.isfile(obj_path): os.remove(obj_path) return True def _inject_auto_imports(self, code: str, src_path: str) -> str: """检测源码中对 list/dict 容器的使用,自动注入 import _list / import _dict。 当使用 dict 时,同时注入 import json (用于 dict 的 JSON 处理功能)。 检测模式: - list[T] / dict[V] 作为类型注解 (ast.Subscript, value=Name(id='list'/'dict')) - list[T](...) / dict[V](...) 作为构造函数调用 (ast.Call, func=Subscript) - list(...) / dict(...) 作为构造函数调用 (ast.Call, func=Name) """ try: tree = ast.parse(code) except SyntaxError: return code # 检查是否已经导入了 _list / _dict / json has_list_import: bool = False has_dict_import: bool = False has_json_import: bool = False for node in ast.iter_child_nodes(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name == '_list': has_list_import = True elif alias.name == '_dict': has_dict_import = True elif alias.name == 'json': has_json_import = True elif isinstance(node, ast.ImportFrom): if node.module == '_list': has_list_import = True elif node.module == '_dict': has_dict_import = True elif node.module == 'json': has_json_import = True if has_list_import and has_dict_import and has_json_import: break # 检查是否使用了 list / dict 容器 uses_list: bool = False uses_dict: bool = False for node in ast.walk(tree): if isinstance(node, ast.Subscript): if isinstance(node.value, ast.Name): if node.value.id == 'list': uses_list = True elif node.value.id == 'dict': uses_dict = True if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): if node.func.id == 'list': uses_list = True elif node.func.id == 'dict': uses_dict = True # 检测类型注解 d: dict / l: list (非泛型形式) if isinstance(node, ast.AnnAssign) and isinstance(node.annotation, ast.Name): if node.annotation.id == 'dict': uses_dict = True elif node.annotation.id == 'list': uses_list = True # 注入 import 到源码开头 prefix: str = '' if uses_list and not has_list_import: prefix += 'import _list\n' if uses_dict and not has_dict_import: prefix += 'import _dict\n' if uses_dict and not has_json_import: prefix += 'import json\n' if prefix: return prefix + code return code def _translate_file(self, src_path: str, out_path: str, prebuilt_ModuleSha1Map: dict[str, str] | None = None) -> None: """翻译单个源文件,嵌入相关 .stub.ll 声明""" with open(src_path, 'r', encoding='utf-8') as f: code = f.read() # SHA1 必须在自动导入注入之前计算,以保持与 Phase1 生成的文件名/函数名前缀一致 sha1 = compute_sha1(code) # 自动导入: 检测 list 容器使用,自动注入 import _list code = self._inject_auto_imports(code, src_path) if prebuilt_ModuleSha1Map is not None: ModuleSha1Map = prebuilt_ModuleSha1Map else: ModuleSha1Map = self._build_current_ModuleSha1Map(sha1) trans = TransPyC.TransPyC(code=code, triple=self.triple, datalayout=self.datalayout) trans.translator.CurrentFile = src_path trans.SliceLevel = self.slice_level trans.translator.SliceLevel = self.slice_level trans.translator.SliceCount = 0 trans.translator.SliceInfos = [] trans.translator.LlvmGen = None trans.translator._module_sha1 = sha1 trans.translator._ModuleSha1Map = ModuleSha1Map trans.translator._struct_sha1_map = self.struct_sha1_map trans.translator._global_function_default_args = self.function_default_args trans.translator._global_function_param_names = self.function_param_names trans.translator._temp_dir = self.temp_dir trans.translator._shared_doc_meta = self._shared_doc_meta # 传递共享 vtable 状态 trans.translator._shared_class_vtable = self._shared_class_vtable trans.translator._shared_cross_module_vtable = self._shared_cross_module_vtable trans.translator._shared_class_methods = self._shared_class_methods trans.translator._shared_cross_module_novtable = self._shared_cross_module_novtable if self._shared_symbol_table is not None: trans.translator.SymbolTable = self._shared_symbol_table # 确保共享符号表的 translator 指向当前 translator,以便 import 别名解析正确 if hasattr(self._shared_symbol_table, 'translator'): self._shared_symbol_table.translator = trans.translator trans.translator._source_module_sig_files = self._shared_source_module_sig_files trans.translator._all_define_constants = self._shared_all_dc trans.translator._export_extern_funcs = self._shared_export_extern_funcs if self._shared_generic_class_templates: if not hasattr(trans.translator.ClassHandler, '_generic_class_templates'): trans.translator.ClassHandler._generic_class_templates = {} trans.translator.ClassHandler._generic_class_templates.update(self._shared_generic_class_templates) else: export_extern_funcs = {} for sha1_key, sig_path in self.sig_files.items(): if sha1_key == sha1: continue try: with open(sig_path, 'r', encoding='utf-8') as f: sig_content = f.read() sig_tree = ast.parse(sig_content) for node in ast.iter_child_nodes(sig_tree): if isinstance(node, ast.FunctionDef): if _check_annotation_for_export(node.returns): export_extern_funcs[node.name] = sha1_key except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"收集导出外部函数声明失败: {_e}", "Exception") if export_extern_funcs: trans.translator._export_extern_funcs = export_extern_funcs for includes_dir in self.include_dirs: if os.path.isdir(includes_dir): for pyi_file in os.listdir(includes_dir): if pyi_file.endswith('.vpi') or pyi_file.endswith('.pyi'): pyi_path = os.path.join(includes_dir, pyi_file) module_name = os.path.splitext(pyi_file)[0] trans.translator.SymbolTable.LoadModuleSymbols(pyi_path, module_name, lineno=0) for py_file in os.listdir(includes_dir): if (py_file.endswith('.vp') or py_file.endswith('.py')) and (not py_file.startswith('_') or py_file == '_list.vp' or py_file == '_list.py' or py_file == '_dict.vp' or py_file == '_dict.py'): module_name = os.path.splitext(py_file)[0] sha1_key = self.include_py_map.get(module_name) if sha1_key and sha1_key in self.sig_files: sig_path = self.sig_files[sha1_key] trans.translator.SymbolTable.LoadModuleSymbols(sig_path, module_name, lineno=0) # 收集需要重导出的包,等所有模块符号加载完后再处理 _pending_reexports = [] for sha1_key, sig_path in self.sig_files.items(): rel_path = self.sha1_map.get(sha1_key, sha1_key) if rel_path.startswith('includes/'): mod_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') mod_name = mod_name[len('includes/'):] trans.translator.SymbolTable.LoadModuleSymbols(sig_path, mod_name, lineno=0) if mod_name.endswith('.__init__'): package_name = mod_name[:-len('.__init__')] _pending_reexports.append((sig_path, package_name)) # 所有 includes 模块符号已加载,现在处理包重导出 for sig_path, package_name in _pending_reexports: self._register_package_reexports(trans, sig_path, package_name) for sha1_key, sig_path in self.sig_files.items(): if sha1_key == sha1: continue rel_path = self.sha1_map.get(sha1_key, sha1_key) if rel_path.startswith('includes/'): continue module_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') trans.translator.SymbolTable.LoadModuleSymbols(sig_path, module_name, lineno=0) trans.translator._source_module_sig_files[module_name] = sig_path short_name = module_name.split('.')[-1] if '.' in module_name else module_name if short_name != module_name: trans.translator._source_module_sig_files[short_name] = sig_path all_dc = {} for sha1_key, stub_path in self.stub_files.items(): if sha1_key == sha1: continue if os.path.exists(stub_path): try: with open(stub_path, 'r', encoding='utf-8') as f: stub_content = f.read() for line in stub_content.splitlines(): line = line.strip() if line.startswith('@') and '= global' in line and 'i32' in line and not line.startswith('@llvm'): var_name = line.split('=')[0].strip().lstrip('@') val_part = line.split('i32')[-1].strip().rstrip(']').lstrip('[').lstrip('i32') try: val = int(val_part) all_dc[var_name] = val except Exception as e: _vlog().warning(f"Phase2Translator: 忽略异常 {e}", exc_info=e) except Exception as e: _vlog().warning(f"Phase2Translator: 忽略异常 {e}", exc_info=e) for sha1_key, pyi_path in self.sig_files.items(): if sha1_key == sha1: continue _ExtractCDefineConstantsFromPyi(pyi_path, all_dc) if all_dc: trans.translator._all_define_constants = all_dc for sym_name, sym_info in self.inline_func_symbols.items(): if sym_name not in trans.translator.SymbolTable: trans.translator.SymbolTable[sym_name] = sym_info else: existing = trans.translator.SymbolTable[sym_name] if not getattr(existing, 'InlineBody', None) and getattr(sym_info, 'InlineBody', None): existing.IsInline = sym_info.IsInline existing.InlineBody = sym_info.InlineBody existing.InlineParams = sym_info.InlineParams try: result = trans.Convert( OutputFilename=out_path, SourceFilename=src_path, target='llvm' ) except Exception as e: error_stack = getattr(trans.translator, '_ErrorStack', []) if error_stack: self._last_ErrorStack = error_stack chain_lines = [] for entry in reversed(error_stack): if len(entry) >= 3: exc_msg, line_info = entry[1], entry[2] chain_lines.append(" %s\n %s" % (line_info, exc_msg)) if chain_lines: enriched = str(e) + "\n" + "\n".join(chain_lines) raise type(e)(enriched) from e node_info = "" if hasattr(trans.translator, 'LlvmGen') and trans.translator.LlvmGen: node_info = trans.translator.LlvmGen._get_node_info() if node_info: enriched = "%s\n %s" % (str(e), node_info.strip()) raise type(e)(enriched) from e raise if result and isinstance(result, str): for func_name, func_def in trans.translator.FunctionDefCache.items(): if hasattr(func_def, 'args'): mangled = trans.translator.LlvmGen._mangle_func_name(func_name) if trans.translator.LlvmGen else func_name # 提取参数名 if hasattr(func_def.args, 'args'): param_names = [arg.arg for arg in func_def.args.args] self.function_param_names[mangled] = param_names self.function_param_names[func_name] = param_names # 提取默认值 if hasattr(func_def.args, 'defaults') and func_def.args.defaults: defaults = [] for d in func_def.args.defaults: if isinstance(d, ast.Constant): defaults.append(d.value) else: defaults.append(None) self.function_default_args[mangled] = defaults self.function_default_args[func_name] = defaults for sym_name, sym_info in trans.translator.SymbolTable.items(): if isinstance(sym_info, CTypeInfo) and getattr(sym_info, 'IsInline', False) and getattr(sym_info, 'InlineBody', None): self.inline_func_symbols[sym_name] = sym_info out_dir = os.path.dirname(out_path) if out_dir: os.makedirs(out_dir, exist_ok=True) # 修复 LLVM 22+: external global 不能有初值,改为 linkonce result = re.sub(r'^(@[\w.]+)\s*=\s*external\s+global\s+(\w[\w*]*)\s+(-?\d+|null|zeroinitializer|true|false)$', r'\1 = linkonce global \2 \3', result, flags=re.MULTILINE) # 后处理:添加 llvmlite 不支持的 LLVM 属性(如 willreturn, mustprogress) pending_str_attrs = getattr(trans.translator.LlvmGen, '_pending_str_attrs', None) if pending_str_attrs: for func_name, attrs in pending_str_attrs.items(): attrs_str = ' '.join(attrs) # 匹配 define ... @func_name(...) ... { 并在 { 之前添加属性 pattern = rf'(define\s+[^@]*@{re.escape(func_name)}\s*\([^)]*\))([^{{]*)(\{{)' result, n = re.subn(pattern, rf'\1\2 {attrs_str}\3', result, count=1) if n == 0: _vlog().warning(f"后处理添加属性失败: 未找到函数 {func_name} 的定义") with open(out_path, 'w', encoding='utf-8', newline='\n') as f: f.write(result) stub_content, text_content = self._split_ll(result) base = os.path.splitext(out_path)[0] with open(base + '.stub.ll', 'w', encoding='utf-8', newline='\n') as f: f.write(stub_content) with open(base + '.text.ll', 'w', encoding='utf-8', newline='\n') as f: f.write(text_content) self._save_deps(sha1, ModuleSha1Map) _vlog().info(f" -> {os.path.basename(out_path)}") else: stub_path = self.stub_files.get(sha1) if stub_path and os.path.exists(stub_path): out_dir = os.path.dirname(out_path) if out_dir: os.makedirs(out_dir, exist_ok=True) with open(stub_path, 'r', encoding='utf-8') as f: stub_content = f.read() with open(out_path, 'w', encoding='utf-8', newline='\n') as f: f.write(stub_content) _vlog().info(f" -> {os.path.basename(out_path)} (仅声明)") def _register_package_reexports(self, trans, init_pyi_path: str, package_name: str) -> None: """解析 __init__.pyi 中的 from .xxx import yyy 或 from __xxx import yyy,在包级别注册导出符号""" try: # 复用 parse_python_file 的 AST 缓存 _, tree = parse_python_file(init_pyi_path) if tree is None: return for node in ast.walk(tree): if isinstance(node, ast.ImportFrom) and node.module: sub_module = node.module.lstrip('.') for alias in node.names: symbol_name = alias.name exported_name = alias.asname if alias.asname else symbol_name source_keys = [ f"{package_name}.{sub_module}.{symbol_name}", f"{package_name}.__{sub_module}.{symbol_name}" if not sub_module.startswith('__') else None, symbol_name, ] target_key = f"{package_name}.{exported_name}" if target_key not in trans.translator.SymbolTable: for source_key in source_keys: if source_key and source_key in trans.translator.SymbolTable: trans.translator.SymbolTable[target_key] = trans.translator.SymbolTable[source_key] break if exported_name not in trans.translator.SymbolTable: for source_key in source_keys: if source_key and source_key in trans.translator.SymbolTable: trans.translator.SymbolTable[exported_name] = trans.translator.SymbolTable[source_key] break except Exception as e: _vlog().warning(f" [警告] 解析包重导出失败 {init_pyi_path}: {e}") def _llc_compile(self, ll_path: str, obj_path: str, cwd: str | None = None) -> subprocess.CompletedProcess: """编译单个 .ll 为 obj。 ASan 模式(TPC_ASAN=1)下先对 IR 执行 opt 插桩: 检测 stdlib 堆/栈/全局越界写与垃圾指针解引用(MemBuddy pool 内部 由块尾 canary 覆盖,两者互补)。opt 失败返回其结果(returncode!=0)。 LLVM >= 20 起 opt -passes=asan 只对带 sanitize_address 函数属性的 函数插桩(clang -fsanitize=address 自动加;手写 IR 没有)—— 先 forceattrs 批量加属性再 asan,否则只链接 runtime 不插桩。 """ compile_input = ll_path if self.asan_mode: asan_ll = ll_path[:-3] + '_asan.ll' # -asan-globals=false:TPC 对全局变量(如各模块 _g_gen)用直接地址寻址, # ASan 全局 redzone 重定位后引用错位会导致读到垃圾指针(如 # register_struct 的 self=0xbebebebe...)。先关闭全局插桩排除该因素。 opt_cmd = ['opt', '-passes=forceattrs,asan', '-force-attribute=sanitize_address', '-asan-globals=false', ll_path, '-o', asan_ll] opt_r = subprocess.run(opt_cmd, capture_output=True, text=True, cwd=cwd) if opt_r.returncode != 0: _vlog().error(f" opt ASan 插桩失败: {opt_r.stderr.strip()}") return opt_r compile_input = asan_ll cmd = [self.compile_cmd] + self.compile_flags + ['-o', obj_path, compile_input] return subprocess.run(cmd, capture_output=True, text=True, cwd=cwd) def _compile_ll_files(self, active_sha1s: set[str]) -> None: """将 output_dir 中属于 active_sha1s 的 .ll 文件编译为 .obj(跳过已编译过的 include 文件)""" # 第一次 walk:清理 stale 文件(.obj/.deps.json/纯 .ll,排除 .stub.ll/.text.ll) stale_files = [] for root, dirs, files in os.walk(self.output_dir): for file in files: if file.endswith('.obj'): sha1 = file[:-4] if sha1 not in active_sha1s: stale_files.append(os.path.join(root, file)) elif file.endswith('.deps.json'): sha1 = file[:-10] if sha1 not in active_sha1s: stale_files.append(os.path.join(root, file)) elif file.endswith('.ll') and not file.endswith('.stub.ll') and not file.endswith('.text.ll'): sha1 = file[:-3] if sha1 not in active_sha1s: stale_files.append(os.path.join(root, file)) if stale_files: for fpath in stale_files: try: os.remove(fpath) except Exception as e: _vlog().warning(f"Phase2Translator: 忽略异常 {e}", exc_info=e) _vlog().info(f"[清理] 删除 {len(stale_files)} 个过时文件") # 第二次 walk:收集需要编译的 .ll 文件(排除 .stub.ll/.text.ll) ll_files = [] for root, dirs, files in os.walk(self.output_dir): for file in files: if file.endswith('.ll') and not file.endswith('.stub.ll') and not file.endswith('.text.ll'): sha1 = file[:-3] if sha1 in active_sha1s: fpath = os.path.join(root, file) obj_path = os.path.join(root, file[:-3] + '.obj') if not os.path.isfile(obj_path): ll_files.append(fpath) if not ll_files: _vlog().info("[编译] 无 .ll 文件需要重新编译") else: _vlog().info(f"\n[编译] 找到 {len(ll_files)} 个 .ll 文件需要编译") # 第三次 walk:收集 all_ll_files(用于 alwaysinline 检测,排除 include sha1s) all_ll_files = [] for root, dirs, files in os.walk(self.output_dir): for file in files: if file.endswith('.ll') and not file.endswith('.stub.ll') and not file.endswith('.text.ll'): sha1 = file[:-3] if sha1 in active_sha1s and sha1 not in self._include_sha1s: all_ll_files.append(os.path.join(root, file)) has_inline = False for ll_path in all_ll_files: try: with open(ll_path, 'r', encoding='utf-8') as f: content = f.read() if 'alwaysinline' in content: has_inline = True break except Exception as e: _vlog().warning(f"Phase2Translator: 忽略异常 {e}", exc_info=e) if has_inline: llvm_link = shutil.which('llvm-link') or shutil.which('llvm-link.exe') opt_cmd = shutil.which('opt') or shutil.which('opt.exe') if llvm_link and opt_cmd: _vlog().info(" [内联] 检测到 alwaysinline 函数,执行跨模块内联优化...") for ll_path in all_ll_files: try: with open(ll_path, 'r', encoding='utf-8') as f: content = f.read() type_defs = [] type_def_indices = set() for i, line in enumerate(content.splitlines()): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: type_defs.append(line) type_def_indices.add(i) if type_def_indices: lines = content.splitlines(True) content_no_types = ''.join(line for i, line in enumerate(lines) if i not in type_def_indices) header_end = 0 for i, line in enumerate(content_no_types.splitlines()): stripped = line.strip() if stripped.startswith(';') or stripped.startswith('target ') or stripped.startswith('source_filename') or not stripped: header_end = i + 1 continue break lines2 = content_no_types.splitlines(True) fixed = ''.join(lines2[:header_end]) + '\n'.join(type_defs) + '\n' + ''.join(lines2[header_end:]) with open(ll_path, 'w', encoding='utf-8') as f: f.write(fixed) except Exception as e: _vlog().warning(f"Phase2Translator: 忽略异常 {e}", exc_info=e) merged_ll = os.path.join(self.output_dir, '_merged.ll') optimized_ll = os.path.join(self.output_dir, '_merged_opt.ll') try: link_cmd = [llvm_link] + all_ll_files + ['-o', merged_ll] result = subprocess.run(link_cmd, capture_output=True, text=True) if result.returncode != 0: _vlog().warning(f" llvm-link 失败: {result.stderr.strip()},回退到单独编译") has_inline = False else: opt_result_cmd = [opt_cmd, '-passes=always-inline', merged_ll, '-S', '-o', optimized_ll] result = subprocess.run(opt_result_cmd, capture_output=True, text=True) if result.returncode != 0: _vlog().warning(f" opt 失败: {result.stderr.strip()},回退到单独编译") has_inline = False else: _vlog().info(" [内联] 跨模块内联优化完成") with open(optimized_ll, 'r', encoding='utf-8') as f: opt_content = f.read() opt_content = re.sub(r'^(define|declare)\s+(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external)\s+', r'\1 \2 dso_local ', opt_content, flags=re.MULTILINE) opt_content = re.sub(r'^(define|declare)\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external|dso_local|@)', r'\1 dso_local ', opt_content, flags=re.MULTILINE) # LLVM 22+: external 不允许在带初值的全局定义上显式写出,改为 linkonce 保留多模块合并语义 opt_content = re.sub(r'(@[\w.]+)\s*=\s*external\s+global\s+(\w[\w*]*)\s+(-?\d+|null|zeroinitializer|true|false)$', r'\1 = linkonce global \2 \3', opt_content, flags=re.MULTILINE) opt_content = re.sub(r'(@[\w.]+)\s*=\s*(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr)\s+(global|constant)', r'\1 = \2 dso_local \3', opt_content) opt_content = re.sub(r'(@[\w.]+)\s*=\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|dso_local)(global|constant)', r'\1 = dso_local \2', opt_content) with open(optimized_ll, 'w', encoding='utf-8') as f: f.write(opt_content) merged_obj = os.path.join(self.output_dir, '_merged.obj') try: result = self._llc_compile(optimized_ll, merged_obj, self.output_dir) if result.returncode == 0: _vlog().success(f" 合并模块编译完成") for ll_path in ll_files: obj_path = ll_path[:-3] + '.obj' sha1_obj = os.path.join(self.output_dir, os.path.basename(ll_path)[:-3] + '.obj') if os.path.isfile(sha1_obj): os.remove(sha1_obj) return else: _vlog().warning(f" 合并模块编译失败: {result.stderr.strip()},回退到单独编译") has_inline = False except Exception as e: _vlog().warning(f" 合并模块编译异常: {e},回退到单独编译") has_inline = False except FileNotFoundError: _vlog().warning(f" 找不到 llvm-link 或 opt,回退到单独编译") has_inline = False else: _vlog().warning(" 未找到 llvm-link/opt 工具,回退到单独编译(alwaysinline 可能不生效)") def _compile_one(ll_path: str) -> tuple[str, str, str]: rel = os.path.relpath(ll_path, self.output_dir) obj_path = ll_path[:-3] + '.obj' if os.path.isfile(obj_path): return (rel, 'cached', '') try: with open(ll_path, 'r', encoding='utf-8') as f: ll_content = f.read() sha1 = os.path.basename(ll_path)[:-3] deps = self._LoadDeps(sha1) stub_prefix = '' if deps: # 去重:同一个全局变量如果同时有 external 声明和实际定义, # 移除 external 声明,避免 "redefinition of global" 错误 # 注意:此去重必须在收集现有符号之前执行,否则行号偏移会导致 # lines_to_remove 误删其他行(如全局变量定义) ll_lines_for_dedup = ll_content.splitlines() external_globals: dict[str, int] = {} non_external_globals: dict[str, int] = {} for i, line in enumerate(ll_lines_for_dedup): stripped = line.strip() if stripped.startswith('@') and (' global ' in stripped or ' constant ' in stripped): gm = re.match(r'(@\S+)', stripped) if gm: gname = gm.group(1) is_external = re.match(r'@\S+\s*=\s*external\s+', stripped) is not None if is_external: external_globals[gname] = i else: non_external_globals[gname] = i lines_to_remove_global_dedup: set[int] = set() for gname in external_globals: if gname in non_external_globals: lines_to_remove_global_dedup.add(external_globals[gname]) if lines_to_remove_global_dedup: ll_content = '\n'.join( line for i, line in enumerate(ll_lines_for_dedup) if i not in lines_to_remove_global_dedup ) + '\n' existing_symbols: dict[str, str] = {} opaque_line_indices: dict[str, int] = {} existing_declares: set[str] = set() existing_declare_lines: dict[str, set[int]] = {} for i, line in enumerate(ll_content.splitlines()): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: name = Phase2Translator._extract_llvm_type_name(stripped) if name: if '= type opaque' in stripped: existing_symbols[name] = 'opaque' opaque_line_indices[name] = i else: existing_symbols[name] = 'full' elif stripped.startswith('declare'): dm = re.match(r'declare\s+[^@]*@"?([^"\s(]+)"?', stripped) if dm: fname = dm.group(1) existing_declares.add(fname) existing_declare_lines.setdefault(fname, set()).add(i) seen_stub_sha1: set[str] = set() replaced_opaque_names: set[str] = set() replaced_declare_names: set[str] = set() all_stub_lines: list[str | None] = [] stub_type_map: dict[str, tuple[bool, int]] = {} stub_declare_names: set[str] = set() stub_declare_lines: dict[str, str] = {} existing_defines: set[str] = set() existing_global_defs: set[str] = set() stub_global_defs: set[str] = set() for line in ll_content.splitlines(): stripped = line.strip() if stripped.startswith('define'): dm = re.match(r'define\s+[^@]*@"?([^"\s(]+)"?', stripped) if dm: existing_defines.add(dm.group(1)) elif stripped.startswith('@') and (' global ' in stripped or ' constant ' in stripped): gm = re.match(r'(@\S+)', stripped) if gm: existing_global_defs.add(gm.group(1)) for dep_name, dep_sha1 in deps.items(): if dep_sha1 in seen_stub_sha1: continue seen_stub_sha1.add(dep_sha1) stub_path = os.path.join(self.output_dir, f"{dep_sha1}.stub.ll") if not os.path.isfile(stub_path): stub_path = os.path.join(self.temp_dir, f"{dep_sha1}.stub.ll") if os.path.isfile(stub_path): with open(stub_path, 'r', encoding='utf-8') as sf: stub_content = sf.read() for line in stub_content.splitlines(): stripped = line.strip() if stripped.startswith(';') or stripped.startswith('target ') or stripped.startswith('source_filename') or not stripped: continue if stripped.startswith('declare'): dm = re.match(r'declare\s+[^@]*@"?([^"\s(]+)"?', stripped) if dm: fname = dm.group(1) if 'zdef_alloc' in fname: _vlog().debug(f"[DEBUG] stub declare for {fname}: existing_declares={fname in existing_declares}, stub_declare_names={fname in stub_declare_names}, has_struct={'%' in stripped}, existing={existing_declare_lines.get(fname)}") if fname in existing_defines: continue if fname in existing_declares or fname in stub_declare_names: existing_decl_line = stub_declare_lines.get(fname) if existing_decl_line is not None: has_struct = '%' in stripped existing_has_struct = '%' in existing_decl_line if existing_decl_line else False if has_struct and not existing_has_struct: for i, l in enumerate(all_stub_lines): if l is not None and l.strip().startswith('declare'): em = re.match(r'declare\s+[^@]*@"?([^"\s(]+)"?', l.strip()) if em and em.group(1) == fname: all_stub_lines[i] = line break continue elif fname in existing_declares: has_struct = '%' in stripped if has_struct: replaced_declare_names.add(fname) stub_declare_names.add(fname) stub_declare_lines[fname] = line all_stub_lines.append(line) continue continue stub_declare_names.add(fname) stub_declare_lines[fname] = line all_stub_lines.append(line) continue if stripped.startswith('%') and '= type' in stripped: name = Phase2Translator._extract_llvm_type_name(stripped) if name: is_opaque = '= type opaque' in stripped if name in existing_symbols: if existing_symbols[name] == 'opaque' and not is_opaque: replaced_opaque_names.add(name) existing_symbols[name] = 'full' idx = len(all_stub_lines) all_stub_lines.append(line) stub_type_map[name] = (is_opaque, idx) continue if name in stub_type_map: prev_opaque, prev_idx = stub_type_map[name] if prev_opaque and not is_opaque: all_stub_lines[prev_idx] = None idx = len(all_stub_lines) all_stub_lines.append(line) stub_type_map[name] = (is_opaque, idx) continue idx = len(all_stub_lines) all_stub_lines.append(line) stub_type_map[name] = (is_opaque, idx) else: all_stub_lines.append(line) elif stripped.startswith('@') and (' global ' in stripped or ' constant ' in stripped): gm = re.match(r'(@\S+)', stripped) if gm and (gm.group(1) in existing_global_defs or gm.group(1) in stub_global_defs): continue if gm: stub_global_defs.add(gm.group(1)) all_stub_lines.append(line) else: all_stub_lines.append(line) filtered = [l for l in all_stub_lines if l is not None] if filtered: stub_prefix = '\n'.join(filtered) + '\n' lines_to_remove = set() if replaced_opaque_names: for name in replaced_opaque_names: if name in opaque_line_indices: lines_to_remove.add(opaque_line_indices[name]) if replaced_declare_names: _vlog().debug(f"[DEBUG] replaced_declare_names: {replaced_declare_names}") for fname in replaced_declare_names: if fname in existing_declare_lines: lines_to_remove.update(existing_declare_lines[fname]) _vlog().debug(f"[DEBUG] removing lines {existing_declare_lines[fname]} for {fname}") else: _vlog().debug(f"[DEBUG] fname {fname} not in existing_declare_lines: {list(existing_declare_lines.keys())}") if lines_to_remove: ll_lines = ll_content.splitlines(True) ll_content = ''.join(line for i, line in enumerate(ll_lines) if i not in lines_to_remove) type_defs_in_ll = [] type_def_line_indices = set() for i, line in enumerate(ll_content.splitlines()): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: type_defs_in_ll.append(line) type_def_line_indices.add(i) if type_def_line_indices: ll_lines = ll_content.splitlines(True) ll_content = ''.join(line for i, line in enumerate(ll_lines) if i not in type_def_line_indices) if type_defs_in_ll: stub_prefix += '\n'.join(type_defs_in_ll) + '\n' if stub_prefix: header_end = 0 for i, line in enumerate(ll_content.splitlines()): stripped = line.strip() if stripped.startswith(';') or stripped.startswith('target ') or stripped.startswith('source_filename') or not stripped: header_end = i + 1 continue break ll_lines = ll_content.splitlines(True) ll_content = ''.join(ll_lines[:header_end]) + stub_prefix + ''.join(ll_lines[header_end:]) ll_content = re.sub(r'^(define|declare)\s+(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external)\s+', r'\1 \2 dso_local ', ll_content, flags=re.MULTILINE) ll_content = re.sub(r'^(define|declare)\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external|dso_local|@)', r'\1 dso_local ', ll_content, flags=re.MULTILINE) # LLVM 22+: external 不允许在带初值的全局定义上显式写出,改为 linkonce 保留多模块合并语义 ll_content = re.sub(r'^(@[\w.]+)\s*=\s*external\s+global\s+(\w[\w*]*)\s+(-?\d+|null|zeroinitializer|true|false)$', r'\1 = linkonce global \2 \3', ll_content, flags=re.MULTILINE) ll_content = re.sub(r'^(@[\w.]+)\s*=\s*(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr)\s+(global|constant)', r'\1 = \2 dso_local \3', ll_content, flags=re.MULTILINE) ll_content = re.sub(r'^(@[\w.]+)\s*=\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|dso_local)(global|constant)', r'\1 = dso_local \2', ll_content, flags=re.MULTILINE) # LLVM 22+: 移除同一模块中已有 define 的 declare 语句(LLVM 22 将 declare+define 视为重复定义) defined_funcs: set[str] = set() for m in re.finditer(r'^define\s+(?:dso_local\s+)?[^@]*@\"?([^"\s\(]+)\"?\s*\(', ll_content, re.MULTILINE): defined_funcs.add(m.group(1)) if defined_funcs: lines: list[str] = ll_content.splitlines(True) kept_lines: list[str] = [] for line in lines: m = re.match(r'^declare\s+(?:dso_local\s+)?[^@]*@\"?([^"\s\(]+)\"?\s*\(', line) if m and m.group(1) in defined_funcs: continue kept_lines.append(line) ll_content = ''.join(kept_lines) # 移除重复的 declare 语句(同一函数多次 declare 但没有 define) seen_declares: set[str] = set() dedup_lines: list[str] = ll_content.splitlines(True) dedup_kept: list[str] = [] for line in dedup_lines: dm = re.match(r'^declare\s+(?:dso_local\s+)?[^@]*@\"?([^"\s\(]+)\"?\s*\(', line) if dm: fname = dm.group(1) if fname in seen_declares: continue seen_declares.add(fname) dedup_kept.append(line) ll_content = ''.join(dedup_kept) final_type_defs = [] final_type_def_indices = set() seen_type_names_main = set() for i, line in enumerate(ll_content.splitlines()): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: type_name_main = stripped.split('=')[0].strip() final_type_def_indices.add(i) if type_name_main in seen_type_names_main: continue seen_type_names_main.add(type_name_main) final_type_defs.append(line) if final_type_def_indices: final_lines = ll_content.splitlines(True) ll_content_no_types = ''.join(line for i, line in enumerate(final_lines) if i not in final_type_def_indices) header_end = 0 for i, line in enumerate(ll_content_no_types.splitlines()): stripped = line.strip() if stripped.startswith(';') or stripped.startswith('target ') or stripped.startswith('source_filename') or not stripped: header_end = i + 1 continue break final_lines2 = ll_content_no_types.splitlines(True) ll_content = ''.join(final_lines2[:header_end]) + '\n'.join(final_type_defs) + '\n' + ''.join(final_lines2[header_end:]) dso_ll_path = ll_path[:-3] + '_dso.ll' with open(dso_ll_path, 'w', encoding='utf-8') as f: f.write(ll_content) result = self._llc_compile(dso_ll_path, obj_path, os.path.dirname(ll_path) or '.') if result.returncode == 0: return (rel, 'ok', '') else: return (rel, 'error', result.stderr.strip()) except FileNotFoundError: return (rel, 'error', f'找不到编译器: {self.compile_cmd}') except Exception as e: return (rel, 'error', str(e)) need_compile = [] for ll_path in ll_files: obj_path = ll_path[:-3] + '.obj' if not os.path.isfile(obj_path): need_compile.append(ll_path) if need_compile: n_workers = min(os.cpu_count() or 4, len(need_compile)) _vlog().info(f" 并行编译 {len(need_compile)} 个文件 (workers={n_workers})") errors = [] with ThreadPoolExecutor(max_workers=n_workers) as executor: futures = {executor.submit(_compile_one, ll_path): ll_path for ll_path in need_compile} for future in as_completed(futures): rel, status, err = future.result() if status == 'ok': _vlog().success(f" {rel}") elif status == 'error': _vlog().error(f" {rel}: {err}") errors.append((rel, err)) elif status == 'cached': pass if errors: _vlog().error(f"\n{len(errors)} 个文件编译失败") sys.exit(1) def _precollect_vtable_state(self) -> None: """预扫描 includes .py 文件,收集 @t.CVTable 类的虚表信息。 并行编译(ProcessPoolExecutor)创建独立内存的 worker 进程, vtable 状态必须预收集并通过 pickle 共享。 复用 parse_python_file 的 AST 缓存。 """ for includes_dir in self.include_dirs: if not os.path.isdir(includes_dir): continue for root, dirs, files in os.walk(includes_dir): dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__'] for fname in files: if not (fname.endswith('.vp') or fname.endswith('.py')): continue py_path: str = os.path.join(root, fname) code, tree = parse_python_file(py_path) # 字符串预过滤:不含 CVTable 的文件跳过 AST 遍历 if tree is None or 'CVTable' not in code: continue try: for node in ast.iter_child_nodes(tree): if not isinstance(node, ast.ClassDef): continue has_cvtable: bool = False for dec in node.decorator_list: if isinstance(dec, ast.Attribute): if getattr(dec.value, 'id', None) == 't' and dec.attr == 'CVTable': has_cvtable = True elif isinstance(dec, ast.Name): if dec.id == 'CVTable': has_cvtable = True if not has_cvtable: continue class_name: str = node.name has_methods: bool = False if class_name not in self._shared_class_methods: self._shared_class_methods[class_name] = [] for item in node.body: if isinstance(item, ast.FunctionDef): has_methods = True method_name: str = item.name # 构造函数不放入 vtable(避免子类与基类 vtable 大小不一致破坏多态分派) if method_name in ('__new__', '__init__', '__before_init__'): continue if method_name == '__init__': func_full_name: str = f"{class_name}.__init__" elif method_name == '__call__': func_full_name = f"{class_name}.__call__" else: func_full_name = f"{class_name}.{method_name}" if func_full_name not in self._shared_class_methods[class_name]: self._shared_class_methods[class_name].append(func_full_name) if has_methods: self._shared_class_vtable.add(class_name) self._shared_cross_module_vtable.add(class_name) except Exception: pass def _build_include_py_index(self, includes_dir: str) -> set[str]: """预扫描 includes 目录,构建所有 .py 文件路径集合 用 set 查询替代 os.path.isfile,避免 3490 次磁盘 stat(83s→<0.1s)。 同时规范化路径为小写(Windows 不区分大小写)以加速查找。 """ py_index: set[str] = set() for root, dirs, files in os.walk(includes_dir): dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__'] for fname in files: if fname.endswith('.vp') or fname.endswith('.py'): py_index.add(os.path.normcase(os.path.join(root, fname))) return py_index def _scan_include_libraries(self) -> None: """扫描 includes 目录,检测被使用的模块对应的库文件和源文件""" if not self.used_includes: return for includes_dir in self.include_dirs: if not os.path.isdir(includes_dir): continue # 预构建 .py 文件索引,避免递归发现时反复 os.path.isfile py_index = self._build_include_py_index(includes_dir) for module_name in self.used_includes: module_dir = os.path.join(includes_dir, module_name) if os.path.isdir(module_dir): self._scan_dir_for_libs(module_dir, module_name) for root, dirs, files in os.walk(module_dir): dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__'] for fname in files: if fname.endswith('.vp') or fname.endswith('.py'): py_path = os.path.join(root, fname) if py_path not in self.extra_py_files: self.extra_py_files.append(py_path) _vlog().info(f" [includes] 发现 Python 包文件: {os.path.relpath(py_path, includes_dir)}") for ext in self.INCLUDE_LIB_EXTENSIONS: lib_path = os.path.join(includes_dir, module_name + ext) if os.path.isfile(lib_path): self.extra_link_files.append(lib_path) _vlog().info(f" [includes] 发现库文件: {os.path.relpath(lib_path, includes_dir)}") for ext in self.INCLUDE_SRC_EXTENSIONS: src_path = os.path.join(includes_dir, module_name + ext) if os.path.isfile(src_path): self.extra_compile_files.append(src_path) _vlog().info(f" [includes] 发现源文件: {os.path.relpath(src_path, includes_dir)}") py_path = os.path.join(includes_dir, module_name + '.vp') if os.path.normcase(py_path) not in py_index: # 回退到 .py(向后兼容) py_path = os.path.join(includes_dir, module_name + '.py') if os.path.normcase(py_path) in py_index: self.extra_py_files.append(py_path) _vlog().info(f" [includes] 发现 Python 源文件: {os.path.relpath(py_path, includes_dir)}") discovered = set() for ef in list(self.extra_py_files): self._discover_transitive_deps(ef, includes_dir, discovered, py_index) if self.extra_link_files: _vlog().info(f"\n[includes] 需要链接的库文件: {len(self.extra_link_files)} 个") if self.extra_compile_files: _vlog().info(f"[includes] 需要编译的源文件: {len(self.extra_compile_files)} 个") def _scan_dir_for_libs(self, directory: str, module_name: str) -> None: """递归扫描目录中的库文件和源文件""" for root, dirs, files in os.walk(directory): for fname in files: fpath = os.path.join(root, fname) _, ext = os.path.splitext(fname) ext = ext.lower() if ext in self.INCLUDE_LIB_EXTENSIONS: self.extra_link_files.append(fpath) _vlog().info(f" [includes] 发现库文件: {os.path.relpath(fpath, directory)}") elif ext in self.INCLUDE_SRC_EXTENSIONS: self.extra_compile_files.append(fpath) _vlog().info(f" [includes] 发现源文件: {os.path.relpath(fpath, directory)}") def _discover_transitive_deps(self, py_path: str, includes_dir: str, discovered: set[str], py_index: set[str] | None = None) -> None: """递归发现 Python 文件的间接依赖 复用 parse_python_file 的 AST 缓存,避免重复 open+read+ast.parse。 py_index 为预构建的 .py 文件路径集合,避免 os.path.isfile 磁盘 stat。 """ if py_path in discovered: return discovered.add(py_path) _, tree = parse_python_file(py_path) if tree is None: return try: for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: self._try_add_include_dep(alias.name, includes_dir, discovered, py_index) elif isinstance(node, ast.ImportFrom): if node.module and node.level == 0: self._try_add_include_dep(node.module, includes_dir, discovered, py_index) except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"发现 include 传递依赖失败: {_e}", "Exception") def _try_add_include_dep(self, module_name: str, includes_dir: str, discovered: set, py_index: set[str] | None = None): """尝试将模块添加为 include 依赖(支持子包,如 w32.win32base) py_index 非空时用 set 查询替代 os.path.isfile(O(1) vs 24ms/次)。 """ # 候选路径:子包路径(w32/win32base.vp/.py)、顶层模块(w32.vp/.py)、子包 __init__.vp/__init__.py candidates: list[str] = [] # 子包路径:w32.win32base → w32/win32base.vp / .py candidates.append(os.path.join(includes_dir, module_name.replace('.', os.sep) + '.vp')) candidates.append(os.path.join(includes_dir, module_name.replace('.', os.sep) + '.py')) # 子包 __init__.vp/__init__.py:w32.win32base → w32/win32base/__init__.vp / .py candidates.append(os.path.join(includes_dir, module_name.replace('.', os.sep), '__init__.vp')) candidates.append(os.path.join(includes_dir, module_name.replace('.', os.sep), '__init__.py')) # 顶层模块:取第一个组件 → w32.vp / .py top_module = module_name.split('.')[0] if top_module != module_name: candidates.append(os.path.join(includes_dir, top_module + '.vp')) candidates.append(os.path.join(includes_dir, top_module + '.py')) for py_path in candidates: exists: bool if py_index is not None: exists = os.path.normcase(py_path) in py_index else: exists = os.path.isfile(py_path) if exists and py_path not in self.extra_py_files: self.extra_py_files.append(py_path) _vlog().info(f" [includes] 发现间接依赖: {os.path.relpath(py_path, includes_dir)}") self._discover_transitive_deps(py_path, includes_dir, discovered, py_index) return # 找到第一个匹配即可 def _is_decl_only_file(self, src_path: str) -> bool: """复用 parse_python_file 的 AST 缓存""" _, tree = parse_python_file(src_path) if tree is None: return False try: for node in ast.walk(tree): if isinstance(node, ast.ClassDef): return False if isinstance(node, ast.FunctionDef): body = node.body if len(body) == 1: stmt = body[0] if isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Constant) and stmt.value.value is ...: continue if isinstance(stmt, ast.Pass): continue return False return True except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"判断是否为声明文件失败: {_e}", "Exception") return False def _sort_include_files_by_deps(self, file_list: list[str]) -> tuple[list[str], dict[str, set[str]]]: """按依赖关系对 include 文件进行拓扑排序,确保被依赖的文件先编译。 返回 (排序后的文件列表, 依赖字典)。 依赖字典: {fpath: set(依赖的fpath)},仅包含 file_list 中的依赖。 """ # 构建文件名到路径的映射 name_to_path = {} for fpath in file_list: basename = os.path.splitext(os.path.basename(fpath))[0] name_to_path[basename] = fpath # 也支持包内模块: vpsdk/window -> path for fpath in file_list: # 尝试从 includes 目录推断模块全名 for includes_dir in self.include_dirs: try: rel = os.path.relpath(fpath, includes_dir) mod_name = os.path.splitext(rel)[0].replace(os.sep, '.').replace('/', '.') name_to_path[mod_name] = fpath except ValueError: pass # 分析每个文件的 import 依赖(复用 parse_python_file 的 AST 缓存) deps = {} # path -> set of paths it depends on for fpath in file_list: deps[fpath] = set() try: _, tree = parse_python_file(fpath) if tree is None: continue for node in ast.iter_child_nodes(tree): if isinstance(node, ast.Import): for alias in node.names: # 尝试完整路径和各级前缀,确保同包内子模块依赖能被正确解析 # 例如 import zlib.zhuff -> 应先匹配 zlib.zhuff,再回退到 zlib parts = alias.name.split('.') for i in range(len(parts), 0, -1): mod = '.'.join(parts[:i]) if mod in name_to_path and name_to_path[mod] != fpath: deps[fpath].add(name_to_path[mod]) break elif isinstance(node, ast.ImportFrom): if node.module and node.level == 0: # 绝对导入:同样尝试完整路径和各级前缀 parts = node.module.split('.') for i in range(len(parts), 0, -1): mod = '.'.join(parts[:i]) if mod in name_to_path and name_to_path[mod] != fpath: deps[fpath].add(name_to_path[mod]) break elif node.level > 0: # 相对导入:from .xxx import yyy 或 from ..xxx import yyy # 根据当前文件位置推断目标模块的完整路径 current_dir = os.path.dirname(fpath) # node.level == 1 表示当前包,node.level == 2 表示上一级包 package_dir = current_dir for _ in range(node.level - 1): package_dir = os.path.dirname(package_dir) # 推断当前包名(从 includes 目录的相对路径) package_name: str | None = None for includes_dir in self.include_dirs: try: rel = os.path.relpath(package_dir, includes_dir) if rel and rel != '.': package_name = rel.replace(os.sep, '.').replace('/', '.') break except ValueError: pass # 构建目标模块名并查找 target_mod: str | None = None if node.module: target_mod = f"{package_name}.{node.module}" if package_name else node.module else: # from . import yyy:yyy 是子模块,在 names 中 target_mod = package_name if target_mod: parts = target_mod.split('.') for i in range(len(parts), 0, -1): mod = '.'.join(parts[:i]) if mod in name_to_path and name_to_path[mod] != fpath: deps[fpath].add(name_to_path[mod]) break # from . import yyy:yyy 可能是子模块 if node.level > 0 and not node.module: for alias in node.names: sub_mod = f"{package_name}.{alias.name}" if package_name else alias.name if sub_mod in name_to_path and name_to_path[sub_mod] != fpath: deps[fpath].add(name_to_path[sub_mod]) except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"分析 include 文件导入依赖失败: {_e}", "Exception") # 拓扑排序 (Kahn's algorithm) # in_degree 只计入 file_list 中的依赖——decl-only 文件(如 stdint.py) # 不在 file_list 中,不应计入入度,否则依赖它们的文件入度永远不为 0, # 被误判为循环依赖通过 fallback 添加,导致编译顺序错误。 file_set = set(file_list) in_degree = {fpath: sum(1 for dep in deps[fpath] if dep in file_set) for fpath in file_list} # 反向邻接表:如果 A 依赖 B,则 B -> A reverse_adj = {fpath: [] for fpath in file_list} for fpath in file_list: for dep in deps[fpath]: if dep in reverse_adj: reverse_adj[dep].append(fpath) result = [] queue = [fpath for fpath in file_list if in_degree[fpath] == 0] # 对队列排序以保持确定性 queue.sort(key=lambda x: x) while queue: current = queue.pop(0) result.append(current) for neighbor in reverse_adj[current]: in_degree[neighbor] -= 1 if in_degree[neighbor] == 0: queue.append(neighbor) queue.sort(key=lambda x: x) # 如果有循环依赖,把剩余文件也加入 for fpath in file_list: if fpath not in result: result.append(fpath) # 仅保留 file_list 内的依赖(用于波次并行翻译) file_set_result = set(file_list) deps_in_set = {fpath: (deps[fpath] & file_set_result) for fpath in file_list} return result, deps_in_set def _get_precompiled_obj_path(self, src_path: str, sha1: str) -> str | None: """计算 include 文件在 includes.binary/ 中的预编译 .obj 路径。 预编译目录位于 include 目录同级,命名为 {basename}.binary/。 文件命名为 {sha1}.{triple_safe}.{rel_model}.obj,区分目标平台和重定位模型。 """ AbsSrcPath: str = os.path.abspath(src_path) RelModel: str = 'static' for _flag in self.compile_flags: if _flag.startswith('-relocation-model='): RelModel = _flag[len('-relocation-model='):] break for inc_dir in self.include_dirs: abs_inc: str = os.path.abspath(inc_dir) try: rel: str = os.path.relpath(AbsSrcPath, abs_inc) if rel.startswith('..'): continue inc_parent: str = os.path.dirname(abs_inc) inc_base: str = os.path.basename(abs_inc) precompiled_dir: str = os.path.join(inc_parent, inc_base + '.binary') triple_safe: str = (self.triple or 'default').replace(os.sep, '_').replace('/', '_').replace('\\', '_').replace(':', '_') return os.path.join(precompiled_dir, f"{sha1}.{triple_safe}.{RelModel}.obj") except ValueError: continue return None def _compile_include_py_files(self, active_sha1s: set[str]) -> None: """通过 TransPyC 翻译器编译 includes 目录中发现的 .py 文件 优化流程:预注册符号 → 并行翻译 → 串行后处理 → 并行编译 → 注册+缓存 """ if not self.extra_py_files: return # 按依赖关系排序 include 文件,确保被依赖的文件先编译 sorted_files, file_deps = self._sort_include_files_by_deps(self.extra_py_files) _vlog().info(f"\n[includes 编译] 找到 {len(sorted_files)} 个 Python 源文件") # 构建 include 文件的模块 SHA1 映射(预计算所有文件,构建完整映射) include_ModuleSha1Map = self._build_current_ModuleSha1Map(None) need_translate = [] # [(src_path, sha1, module_name, ll_path, obj_path, precompiled_obj)] for src_path in sorted_files: module_name = os.path.splitext(os.path.basename(src_path))[0] # 复用 parse_python_file 缓存的 content,避免重复 open+read py_content, _ = parse_python_file(src_path) sha1 = compute_sha1(py_content) self.include_py_map[module_name] = sha1 active_sha1s.add(sha1) self._include_sha1s.add(sha1) for includes_dir in self.include_dirs: try: rel = os.path.relpath(src_path, includes_dir) mod_full = os.path.splitext(rel)[0].replace(os.sep, '.').replace('/', '.') include_ModuleSha1Map[mod_full] = sha1 include_ModuleSha1Map[module_name] = sha1 # 包入口文件:注册父包名(如 'os.__init__' -> 'os') if module_name == '__init__' and '.' in mod_full: parent_pkg = mod_full.rsplit('.', 1)[0] include_ModuleSha1Map[parent_pkg] = sha1 except ValueError: pass include_ModuleSha1Map[module_name] = sha1 ll_path = os.path.join(self.output_dir, f"{sha1}.ll") obj_path = os.path.join(self.output_dir, f"{sha1}.obj") precompiled_obj = self._get_precompiled_obj_path(src_path, sha1) if self._is_decl_only_file(src_path): _vlog().info(f" 跳过(声明文件): {module_name}.py") # 声明文件(如 stdint.py)虽不编译为 .obj,但其 typedef/类型别名 # 必须注册到共享符号表,否则依赖它的 include 文件编译时 # 类型解析失败(如 UINT32PTR 被当作命名结构体而非 i32*) self._register_compiled_include(src_path, sha1, None, include_ModuleSha1Map) self._collect_inline_symbols(src_path) continue # 优先检查 includes.binary 预编译缓存(跨项目共享) if not self.force_recompile_includes and precompiled_obj and os.path.isfile(precompiled_obj): _vlog().info(f" 跳过(预编译): {module_name}.py -> {os.path.basename(precompiled_obj)}") # 预编译跳过的文件仍需注册 .pyi/.stub.ll 声明和符号, # 否则依赖它的 include 文件(如 vpui.py import vpsdk.window) # 在 _EmitModuleDeclarationsLlvm 中找不到声明,导致 Undefined symbol self._register_compiled_include(src_path, sha1, None, include_ModuleSha1Map) self._collect_inline_symbols(src_path) self.extra_link_files.append(precompiled_obj) continue if not self.force_recompile_includes and os.path.isfile(obj_path): _vlog().info(f" 跳过(缓存): {module_name}.py -> {sha1}.obj") self._register_compiled_include(src_path, sha1, None, include_ModuleSha1Map) self._collect_inline_symbols(src_path) self.extra_link_files.append(obj_path) continue need_translate.append((src_path, sha1, module_name, ll_path, obj_path, precompiled_obj)) if not need_translate: return # 预注册所有待翻译文件的符号(从 Phase1 缓存加载 .pyi / .stub.ll) # 这样并行翻译时所有跨模块符号都已可用 for src_path, sha1, module_name, ll_path, obj_path, precompiled_obj in need_translate: self._register_compiled_include(src_path, sha1, None, include_ModuleSha1Map) # 预收集 stub_decls(所有文件共用,避免重复扫描 .stub.ll) stub_decls = [] for other_sha1, other_stub in self.stub_files.items(): if os.path.exists(other_stub): try: with open(other_stub, 'r', encoding='utf-8') as sf: for sline in sf: sl = sline.strip() if sl.startswith('%') and '= type' in sl and 'opaque' not in sl: struct_name = Phase2Translator._extract_llvm_type_name(sl) if struct_name: stub_decls.append(sl) except Exception as e: _vlog().warning(f"Phase2Translator: 忽略异常 {e}", exc_info=e) all_type_defs = {} for sdecl in stub_decls: tname = Phase2Translator._extract_llvm_type_name(sdecl) if tname: all_type_defs[tname] = sdecl # ========== 阶段1:并行翻译 ========== _vlog().info(f" 翻译 {len(need_translate)} 个文件") if len(need_translate) > 1: # 序列化共享符号表供 worker 使用 shared_pickle_path = os.path.join(self.output_dir, '_shared_sym_includes.pickle') try: # symbol_table 只以 pickle 字节传递(symbol_table_bytes),不重复序列化 # 对象本身(原双重序列化使文件体积/时间翻倍,4.65MB 中近半冗余)。 shared_data = { 'symbol_table_bytes': pickle.dumps(self._shared_symbol_table), 'source_module_sig_files': self._shared_source_module_sig_files, 'all_dc': self._shared_all_dc, 'export_extern_funcs': self._shared_export_extern_funcs, 'inline_func_symbols': self.inline_func_symbols, 'function_default_args': self.function_default_args, 'function_param_names': self.function_param_names, 'generic_class_templates': self._shared_generic_class_templates, 'shared_doc_meta': self._shared_doc_meta, 'class_vtable': self._shared_class_vtable, 'cross_module_vtable': self._shared_cross_module_vtable, 'class_methods': self._shared_class_methods, 'cross_module_novtable': self._shared_cross_module_novtable, } with open(shared_pickle_path, 'wb') as f: f.write(pickle.dumps(shared_data)) except Exception as e: _vlog().warning(f"符号表序列化失败({e}),include将串行翻译") shared_pickle_path = '' if shared_pickle_path and os.path.exists(shared_pickle_path): n_workers = min(os.cpu_count() or 4, len(need_translate)) errors = [] # 波次并行翻译:按依赖关系分批提交,确保被依赖的文件先翻译完成。 # 根因:_TryLoadStructFromStub 扫描 output 目录加载跨模块 struct 定义, # 若依赖文件尚未完成翻译(output stub 未写入),只能取到 Phase1 temp # 中的不完整 stub(缺少跨模块继承字段),导致运行时崩溃。 item_by_path = {item[0]: item for item in need_translate} translate_paths = set(item_by_path.keys()) # 每个文件的依赖(仅限本轮待翻译集合内) translate_deps = { p: (file_deps.get(p, set()) & translate_paths) for p in translate_paths } # 复用 self._executor_pool(与 App 源文件阶段共享),避免 GC 阻塞 if self._executor_pool is None: # 新池:刷新 pyi pickle 缓存,把 _register_compiled_include 新增的 # include .vpi 一并落盘,使 worker 预热包含它们(签名未变时零开销) if os.environ.get('TPC_WORKER_WARMUP', '1') == '1': _BuildPyiCacheFile(self.temp_dir) self._executor_pool = ProcessPoolExecutor(max_workers=n_workers, initializer=_PyiCacheWorkerInitializer, initargs=(self.temp_dir,)) executor = self._executor_pool try: completed_paths: set[str] = set() remaining = list(need_translate) # 保持拓扑序 in_progress: dict = {} # future -> item done_count = 0 while remaining or in_progress: # 提交所有依赖已完成的文件 ready = [ item for item in remaining if translate_deps[item[0]] <= completed_paths ] for item in ready: remaining.remove(item) src_path, sha1, module_name, ll_path, obj_path, precompiled_obj = item future = executor.submit( _parallel_translate_worker, src_path, ll_path, self.src_root, self.temp_dir, self.output_dir, self.include_dirs, self.triple, self.datalayout, self.sha1_map, self.sig_files, self.stub_files, self.include_py_map, self.slice_level, shared_pickle_path, self.struct_sha1_map, include_ModuleSha1Map ) in_progress[future] = item if not in_progress: # 无可提交且无进行中:循环依赖 fallback for item in remaining: src_path, sha1, module_name, ll_path, obj_path, precompiled_obj = item future = executor.submit( _parallel_translate_worker, src_path, ll_path, self.src_root, self.temp_dir, self.output_dir, self.include_dirs, self.triple, self.datalayout, self.sha1_map, self.sig_files, self.stub_files, self.include_py_map, self.slice_level, shared_pickle_path, self.struct_sha1_map, include_ModuleSha1Map ) in_progress[future] = item remaining.clear() # 等待至少一个完成 done_set, _ = wait(in_progress, return_when=FIRST_COMPLETED) for future in done_set: item = in_progress.pop(future) src_path, sha1, module_name, ll_path, obj_path, precompiled_obj = item completed_paths.add(src_path) done_count += 1 try: status, _, err = future.result() if status == 'ok': _vlog().info(f" [{done_count}/{len(need_translate)}] 翻译完成: {module_name}.py") else: # fail-fast:文件编译出错立即终止(worker 已 os._exit, # 此处兜底处理残留的 error 返回) raise RuntimeError( f"include 翻译失败,立即终止: {module_name}.py: {err}") except Exception as e: _vlog().error(f" {module_name}.py: {e}") for f2 in in_progress: f2.cancel() if self._executor_pool is not None: self._executor_pool.shutdown(wait=False, cancel_futures=True) self._executor_pool = None raise RuntimeError( f"include worker 异常终止,编译立即停止: {e}") except Exception: if self._executor_pool is not None: self._executor_pool.shutdown(wait=False, cancel_futures=True) self._executor_pool = None raise if errors: _vlog().error(f"\n{len(errors)} 个 include 文件翻译失败") sys.exit(1) else: # 串行回退 for item in need_translate: src_path, sha1, module_name, ll_path, obj_path, precompiled_obj = item _vlog().info(f" 翻译: {module_name}.py -> {sha1}.ll") self._translate_file(src_path, ll_path, prebuilt_ModuleSha1Map=include_ModuleSha1Map) else: item = need_translate[0] src_path, sha1, module_name, ll_path, obj_path, precompiled_obj = item _vlog().info(f" 翻译: {module_name}.py -> {sha1}.ll") self._translate_file(src_path, ll_path, prebuilt_ModuleSha1Map=include_ModuleSha1Map) # ========== 阶段2:串行后处理(dso_local + stub_decls 注入)========== compile_tasks = [] # [(dso_ll_path, obj_path, src_path, sha1, module_name, ll_path, precompiled_obj, ll_content)] for item in need_translate: src_path, sha1, module_name, ll_path, obj_path, precompiled_obj = item if not os.path.isfile(ll_path): _vlog().error(f" 翻译未生成 .ll 文件: {module_name}.py") continue with open(ll_path, 'r', encoding='utf-8') as f: ll_content = f.read() # dso_local 正则标注(使用 ^ + MULTILINE 避免匹配字符串常量内容中的 define/declare) ll_content = re.sub(r'^(define|declare)\s+(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external)\s+', r'\1 \2 dso_local ', ll_content, flags=re.MULTILINE) ll_content = re.sub(r'^(define|declare)\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external|dso_local)', r'\1 dso_local ', ll_content, flags=re.MULTILINE) # LLVM 22+: external 不允许在带初值的全局定义上显式写出,改为 linkonce 保留多模块合并语义 ll_content = re.sub(r'^(@[\w.]+)\s*=\s*external\s+global\s+(\w[\w*]*)\s+(-?\d+|null|zeroinitializer|true|false)$', r'\1 = linkonce global \2 \3', ll_content, flags=re.MULTILINE) ll_content = re.sub(r'^(@[\w.]+)\s*=\s*(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr)\s+(global|constant)', r'\1 = \2 dso_local \3', ll_content, flags=re.MULTILINE) ll_content = re.sub(r'^(@[\w.]+)\s*=\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|dso_local)(global|constant)', r'\1 = dso_local \2', ll_content, flags=re.MULTILINE) # stub_decls 注入(复用预收集的 stub_decls) if stub_decls: for sdecl in stub_decls: sname = Phase2Translator._extract_llvm_type_name(sdecl) if sname: short = sname.split('.')[-1] if '.' in sname else sname ll_content = re.sub( r'%"?' + re.escape(sname) + r'"?\s*=\s*type\s*opaque', sdecl, ll_content) if short != sname: ll_content = re.sub( r'%"?' + re.escape(short) + r'"?\s*=\s*type\s*opaque', sdecl, ll_content) max_iterations = 10 for _ in range(max_iterations): defined_types = set() for line in ll_content.splitlines(): sl = line.strip() if sl.startswith('%') and '= type' in sl: tname = Phase2Translator._extract_llvm_type_name(sl) if tname: defined_types.add(tname) # 同时添加短名,防止 missing 循环因全名/短名不匹配而重复添加 if '.' in tname: parts = tname.split('.', 1) if len(parts[0]) == 16 and all(c in '0123456789abcdef' for c in parts[0]): defined_types.add(parts[1]) referenced_types = Phase2Translator._find_all_type_refs(ll_content) missing = referenced_types - defined_types new_defs = [] for tname in sorted(missing): if tname in all_type_defs: new_defs.append(all_type_defs[tname]) else: short_key = tname.split('.')[-1] if '.' in tname else tname for full_name, defn in all_type_defs.items(): if full_name.split('.')[-1] == short_key: if short_key not in defined_types: new_defs.append(defn) break else: # 未在 stub 中找到定义(如泛型类 list[T]), # 声明为 opaque 以满足 llc 的类型前向声明要求 new_defs.append(f'%"{tname}" = type opaque') if not new_defs: break inject_block = '\n'.join(new_defs) + '\n' lines = ll_content.splitlines(True) insert_pos = 0 for i, line in enumerate(lines): stripped = line.strip() if stripped and not stripped.startswith(';') and not stripped.startswith('target ') and not stripped.startswith('source_filename') and not stripped.startswith('declare ') and not stripped.startswith('@') and not stripped.startswith('define ') and not stripped.startswith('attributes ') and not stripped.startswith('!') and not stripped.startswith('module '): insert_pos = i break if insert_pos == 0: insert_pos = len(lines) ll_content = ''.join(lines[:insert_pos]) + inject_block + ''.join(lines[insert_pos:]) # 将所有类型定义移到 declares 之前 # (llc 要求 by-value struct 参数的类型在 declare 前定义,否则报 invalid type for function argument) lines = ll_content.splitlines(True) type_def_lines = [] other_lines = [] seen_type_names_inc = set() for line in lines: stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: type_name_inc = stripped.split('=')[0].strip() if type_name_inc in seen_type_names_inc: continue seen_type_names_inc.add(type_name_inc) type_def_lines.append(line) else: other_lines.append(line) if type_def_lines: insert_pos = 0 for i, line in enumerate(other_lines): stripped = line.strip() if stripped and not stripped.startswith(';') and not stripped.startswith('target ') and not stripped.startswith('source_filename') and not stripped.startswith('!'): insert_pos = i break if insert_pos == 0: insert_pos = len(other_lines) ll_content = ''.join(other_lines[:insert_pos]) + ''.join(type_def_lines) + ''.join(other_lines[insert_pos:]) dso_ll_path = ll_path[:-3] + '_dso.ll' with open(dso_ll_path, 'w', encoding='utf-8') as f: f.write(ll_content) compile_tasks.append((dso_ll_path, obj_path, src_path, sha1, module_name, ll_path, precompiled_obj, ll_content)) if not compile_tasks: return # ========== 阶段3:并行编译(llc)========== _vlog().info(f" 编译 {len(compile_tasks)} 个 .ll 文件") compile_results = {} # task -> result def _compile_one_include(task): dso_ll_path, obj_path = task[0], task[1] result = self._llc_compile(dso_ll_path, obj_path, self.output_dir) return (task, result) if len(compile_tasks) > 1: n_workers = min(os.cpu_count() or 4, len(compile_tasks)) with ThreadPoolExecutor(max_workers=n_workers) as executor: futures = {executor.submit(_compile_one_include, task): task for task in compile_tasks} done_count = 0 for future in as_completed(futures): done_count += 1 try: task, result = future.result() module_name = task[4] if result.returncode == 0: _vlog().success(f" [{done_count}/{len(compile_tasks)}] 编译成功: {module_name}.py") else: _vlog().warning(f" [{done_count}/{len(compile_tasks)}] 编译失败: {module_name}.py: {result.stderr.strip()}") compile_results[task] = result except Exception as e: task = futures[future] module_name = task[4] _vlog().error(f" [{done_count}/{len(compile_tasks)}] 编译异常: {module_name}.py: {e}") compile_results[task] = None else: task = compile_tasks[0] task_result, result = _compile_one_include(task) module_name = task[4] if result.returncode == 0: _vlog().success(f" 编译成功: {module_name}.py") else: _vlog().warning(f" 编译失败: {module_name}.py: {result.stderr.strip()}") compile_results[task] = result # ========== 阶段4:注册 + 缓存 ========== has_error = False for task in compile_tasks: dso_ll_path, obj_path, src_path, sha1, module_name, ll_path, precompiled_obj, ll_content = task result = compile_results.get(task) if result and result.returncode == 0: self.extra_link_files.append(obj_path) # 复制到 includes.binary 预编译目录 if precompiled_obj: precompiled_dir = os.path.dirname(precompiled_obj) try: if not os.path.exists(precompiled_dir): os.makedirs(precompiled_dir, exist_ok=True) shutil.copy2(obj_path, precompiled_obj) except Exception as _e: if _config_mode == "strict": _vlog().warning(f"预编译缓存写入失败: {_e}") # 注册符号 self._register_compiled_include(src_path, sha1, ll_content, include_ModuleSha1Map) # 清理临时文件(仅成功时清理,失败时保留以便调试) # DEBUG: 保留 .ll 文件以便静态分析 # for ext in ['.ll', '_dso.ll']: # p = ll_path[:-3] + ext # if os.path.isfile(p): # os.remove(p) else: has_error = True _vlog().warning(f" 保留失败文件以便调试: {ll_path}") if has_error: _vlog().error(f"\nincludes 文件编译失败") sys.exit(1) # 从 include 文件源码中提取函数参数名,供主项目编译时 keyword 参数匹配使用 self._extract_include_param_names(sorted_files) def _extract_include_param_names_from_dirs(self) -> None: """从 include 目录的 Python 源码中提取函数参数名,存入 function_param_names 优化:单次递归遍历 AST,维护当前类上下文,O(N) 复杂度。 复用 parse_python_file 的 AST 缓存。 """ for includes_dir in self.include_dirs: if not os.path.isdir(includes_dir): continue for filename in os.listdir(includes_dir): if not (filename.endswith('.vp') or filename.endswith('.py')): continue src_path = os.path.join(includes_dir, filename) try: source, tree = parse_python_file(src_path) if tree is None: continue sha1 = compute_sha1(source) self._collect_function_params(tree, sha1) except Exception as e: _vlog().warning(f" 提取参数名失败 {src_path}: {e}") def _collect_function_params(self, tree: ast.AST, sha1: str | None) -> None: """单次递归遍历 AST 收集函数参数名(含类方法),O(N) 复杂度 遍历时维护 current_class 上下文:遇到 ClassDef 进入类作用域, 遇到 FunctionDef 记录参数名并标注所属类,函数内部的嵌套定义 不再属于外层类。 """ def visit(node: ast.AST, current_class: str | None) -> None: for child in ast.iter_child_nodes(node): if isinstance(child, ast.ClassDef): # 进入类作用域,递归处理类体 visit(child, child.name) elif isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)): func_name = child.name param_names = [arg.arg for arg in child.args.args] self.function_param_names[func_name] = param_names if sha1: self.function_param_names[f'{sha1}.{func_name}'] = param_names if current_class: self.function_param_names[f'{current_class}.{func_name}'] = param_names if sha1: self.function_param_names[f'{sha1}.{current_class}.{func_name}'] = param_names # 函数内部的嵌套定义不属于外层类 visit(child, None) else: visit(child, current_class) visit(tree, None) def _extract_app_default_args(self, need_translate: list) -> None: """从 App 源文件中提取函数默认参数,存入 function_default_args 在 Phase 2 并行编译开始前调用,确保跨模块调用参数检查能正确识别默认参数。 need_translate 中每个元素是 (i, src_path, out_path, sha1, current_ModuleSha1Map) 复用 parse_python_file 的 AST 缓存(run() 循环已 parse 过)。 """ for entry in need_translate: src_path = entry[1] sha1 = entry[3] _, tree = parse_python_file(src_path) if tree is None: continue try: for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): func_name = node.name defaults = getattr(node.args, 'defaults', []) if defaults: default_vals = [] for d in defaults: if isinstance(d, ast.Constant): default_vals.append(d.value) else: default_vals.append(None) # 存储多种键格式,确保跨模块调用检查能找到 self.function_default_args[func_name] = default_vals self.function_default_args[f'{sha1}.{func_name}'] = default_vals # 也更新参数名(预扫描时补充) param_names = [arg.arg for arg in node.args.args] if func_name not in self.function_param_names: self.function_param_names[func_name] = param_names if f'{sha1}.{func_name}' not in self.function_param_names: self.function_param_names[f'{sha1}.{func_name}'] = param_names except Exception as e: _vlog().warning(f" 提取默认参数失败 {src_path}: {e}") def _extract_include_param_names(self, sorted_files: list[str]) -> None: """从 include 文件的 Python 源码中提取函数参数名,存入 function_param_names 优化:复用 _collect_function_params 的 O(N) 单次遍历 + parse_python_file 缓存。 """ for src_path in sorted_files: if self._is_decl_only_file(src_path): continue try: _, tree = parse_python_file(src_path) if tree is None: continue module_name = os.path.splitext(os.path.basename(src_path))[0] sha1 = self.include_py_map.get(module_name) self._collect_function_params(tree, sha1) except Exception as e: _vlog().warning(f" 提取参数名失败 {src_path}: {e}") def _register_compiled_include(self, src_path: str, sha1: str, ll_content: str, include_ModuleSha1Map: dict[str, str]) -> None: """编译 include 文件成功后,生成签名和 stub 文件并注册到符号表中""" module_name = os.path.splitext(os.path.basename(src_path))[0] # 确定完整模块名(如 vpsdk.window) mod_full = module_name for includes_dir in self.include_dirs: try: rel = os.path.relpath(src_path, includes_dir) mod_full = os.path.splitext(rel)[0].replace(os.sep, '.').replace('/', '.') break except ValueError: pass # 1. 生成 .pyi 签名文件(如果还没有的话) sig_path = os.path.join(self.temp_dir, f"{sha1}.vpi") if not os.path.isfile(sig_path): # 全局缓存检查(不被 --clean 清除) _TryGlobalCache(sha1, 'pyi', sig_path) if not os.path.isfile(sig_path): try: with open(src_path, 'r', encoding='utf-8') as f: py_content = f.read() sig_content = PythonToStubConverter.convert(py_content, mod_full) os.makedirs(self.temp_dir, exist_ok=True) with open(sig_path, 'w', encoding='utf-8', newline='\n') as f: f.write(sig_content) _SaveToGlobalCache(sha1, 'pyi', sig_path) except Exception as e: if _config_mode == "strict": raise _vlog().warning(f"签名生成失败不阻塞编译: {e}", "Exception") # 注册到 sig_files 和 _source_module_sig_files if os.path.isfile(sig_path): self.sig_files[sha1] = sig_path self.sha1_map[sha1] = f"includes/{mod_full.replace('.', '/')}.vp" if self._shared_source_module_sig_files is not None: self._shared_source_module_sig_files[mod_full] = sig_path self._shared_source_module_sig_files[module_name] = sig_path # 将签名信息加载到共享符号表中 if self._shared_symbol_table is not None: try: self._shared_symbol_table.LoadModuleSymbols(sig_path, mod_full, lineno=0) except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"加载模块签名到共享符号表失败: {_e}", "Exception") # 2. 生成 .stub.ll 文件 stub_path = os.path.join(self.output_dir, f"{sha1}.stub.ll") if not os.path.isfile(stub_path): # 全局缓存检查(不被 --clean 清除) _TryGlobalCache(sha1, 'stub.ll', stub_path) if not os.path.isfile(stub_path) and ll_content: try: stub_content, _ = self._split_ll(ll_content) with open(stub_path, 'w', encoding='utf-8', newline='\n') as f: f.write(stub_content) _SaveToGlobalCache(sha1, 'stub.ll', stub_path) except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"生成 .stub.ll 文件失败: {_e}", "Exception") # 注册到 stub_files if os.path.isfile(stub_path): self.stub_files[sha1] = stub_path elif os.path.isfile(sig_path): # 如果 stub 文件不存在,也尝试从 temp 目录查找 temp_stub = os.path.join(self.temp_dir, f"{sha1}.stub.ll") if os.path.isfile(temp_stub): self.stub_files[sha1] = temp_stub else: # 最后尝试从全局缓存直接读取 _TryGlobalCache(sha1, 'stub.ll', temp_stub) if os.path.isfile(temp_stub): self.stub_files[sha1] = temp_stub def _precollect_inline_symbols(self) -> None: """在主翻译循环之前,扫描includes目录中used_includes相关的模块收集内联函数AST""" if not self.used_includes: return for includes_dir in self.include_dirs: if not os.path.isdir(includes_dir): continue for module_name in self.used_includes: module_dir = os.path.join(includes_dir, module_name) if os.path.isdir(module_dir): for root, dirs, files in os.walk(module_dir): dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__'] for fname in files: if fname.endswith('.vp') or fname.endswith('.py'): src_path = os.path.join(root, fname) self._collect_inline_symbols(src_path) py_path = os.path.join(includes_dir, module_name + '.vp') if not os.path.isfile(py_path): # 回退到 .py(向后兼容) py_path = os.path.join(includes_dir, module_name + '.py') if os.path.isfile(py_path): self._collect_inline_symbols(py_path) for ef in self.extra_py_files: self._collect_inline_symbols(ef) def _collect_inline_symbols(self, src_path: str): """解析源文件,收集内联函数的AST信息到inline_func_symbols 复用 parse_python_file 的 AST 缓存,避免重复 open+read+ast.parse。 """ try: _, tree = parse_python_file(src_path) if tree is None: return module_name = os.path.splitext(os.path.basename(src_path))[0] for node in ast.iter_child_nodes(tree): if isinstance(node, ast.FunctionDef) and node.returns: is_inline = False if isinstance(node.returns, ast.BinOp) and isinstance(node.returns.op, ast.BitOr): for part in [node.returns.left, node.returns.right]: if isinstance(part, ast.Attribute) and part.attr == 'CInline': is_inline = True break if isinstance(part, ast.Name) and part.id == 'CInline': is_inline = True break elif isinstance(node.returns, ast.Attribute) and node.returns.attr == 'CInline': is_inline = True elif isinstance(node.returns, ast.Name) and node.returns.id == 'CInline': is_inline = True if is_inline: func_name = node.name sym_key = f"{module_name}.{func_name}" info = CTypeInfo() info.Name = func_name info.IsFunction = True info.IsInline = True info.InlineBody = node.body info.InlineParams = [arg.arg for arg in node.args.args] info.Storage = t.CInline() self.inline_func_symbols[func_name] = info self.inline_func_symbols[sym_key] = info except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"预收集内联函数符号失败: {_e}", "Exception") def _build_shared_symbol_data(self) -> None: """一次性构建所有文件共享的符号表数据,避免每个文件重复加载""" t0 = time.time() _vlog().info("[缓存] 构建共享符号表数据...") base_trans = TransPyC.TransPyC(code="pass", triple=self.triple, datalayout=self.datalayout) base_trans.translator.LlvmGen = None base_trans.translator._ModuleSha1Map = {} base_trans.translator._global_function_default_args = self.function_default_args base_trans.translator._global_function_param_names = self.function_param_names for includes_dir in self.include_dirs: if os.path.isdir(includes_dir): for pyi_file in os.listdir(includes_dir): if pyi_file.endswith('.vpi') or pyi_file.endswith('.pyi'): pyi_path = os.path.join(includes_dir, pyi_file) module_name = os.path.splitext(pyi_file)[0] base_trans.translator.SymbolTable.LoadModuleSymbols(pyi_path, module_name, lineno=0) for py_file in os.listdir(includes_dir): if (py_file.endswith('.vp') or py_file.endswith('.py')) and (not py_file.startswith('_') or py_file == '_list.vp' or py_file == '_list.py' or py_file == '_dict.vp' or py_file == '_dict.py'): module_name = os.path.splitext(py_file)[0] sha1_key = self.include_py_map.get(module_name) if sha1_key and sha1_key in self.sig_files: sig_path = self.sig_files[sha1_key] base_trans.translator.SymbolTable.LoadModuleSymbols(sig_path, module_name, lineno=0) # 收集需要重导出的包,等所有模块符号加载完后再处理 _pending_reexports = [] for sha1_key, sig_path in self.sig_files.items(): rel_path = self.sha1_map.get(sha1_key, sha1_key) if rel_path.startswith('includes/'): mod_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') mod_name = mod_name[len('includes/'):] base_trans.translator.SymbolTable.LoadModuleSymbols(sig_path, mod_name, lineno=0) if mod_name.endswith('.__init__'): package_name = mod_name[:-len('.__init__')] _pending_reexports.append((sig_path, package_name)) # 所有 includes 模块符号已加载,现在处理包重导出 for sig_path, package_name in _pending_reexports: self._register_package_reexports(base_trans, sig_path, package_name) for sha1_key, sig_path in self.sig_files.items(): rel_path = self.sha1_map.get(sha1_key, sha1_key) if rel_path.startswith('includes/'): continue module_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') base_trans.translator.SymbolTable.LoadModuleSymbols(sig_path, module_name, lineno=0) base_trans.translator._source_module_sig_files[module_name] = sig_path short_name = module_name.split('.')[-1] if '.' in module_name else module_name if short_name != module_name: base_trans.translator._source_module_sig_files[short_name] = sig_path all_dc = {} for sha1_key, stub_path in self.stub_files.items(): if os.path.exists(stub_path): try: with open(stub_path, 'r', encoding='utf-8') as f: stub_content = f.read() for line in stub_content.splitlines(): line = line.strip() if line.startswith('@') and '= global' in line and 'i32' in line and not line.startswith('@llvm'): var_name = line.split('=')[0].strip().lstrip('@') val_part = line.split('i32')[-1].strip().rstrip(']').lstrip('[').lstrip('i32') try: val = int(val_part) all_dc[var_name] = val except Exception as e: _vlog().warning(f"Phase2Translator: 忽略异常 {e}", exc_info=e) except Exception as e: _vlog().warning(f"Phase2Translator: 忽略异常 {e}", exc_info=e) for sha1_key, pyi_path in self.sig_files.items(): _ExtractCDefineConstantsFromPyi(pyi_path, all_dc) export_extern_funcs = {} for sha1_key, sig_path in self.sig_files.items(): try: # 复用 parse_python_file 的 AST 缓存(sig 文件被多处解析) _, sig_tree = parse_python_file(sig_path) if sig_tree is None: continue for node in ast.iter_child_nodes(sig_tree): if isinstance(node, ast.FunctionDef): if _check_annotation_for_export(node.returns): export_extern_funcs[node.name] = sha1_key except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"收集共享导出外部函数失败: {_e}", "Exception") for sym_name, sym_info in self.inline_func_symbols.items(): if sym_name not in base_trans.translator.SymbolTable: base_trans.translator.SymbolTable[sym_name] = sym_info else: existing = base_trans.translator.SymbolTable[sym_name] if not getattr(existing, 'InlineBody', None) and getattr(sym_info, 'InlineBody', None): existing.IsInline = sym_info.IsInline existing.InlineBody = sym_info.InlineBody existing.InlineParams = sym_info.InlineParams self._shared_symbol_table = base_trans.translator.SymbolTable self._shared_source_module_sig_files = dict(base_trans.translator._source_module_sig_files) self._shared_all_dc = all_dc self._shared_export_extern_funcs = export_extern_funcs generic_class_templates = {} for includes_dir in self.include_dirs: if os.path.isdir(includes_dir): abs_includes = os.path.join(self.src_root, includes_dir) if not os.path.isabs(includes_dir) else includes_dir if not os.path.isdir(abs_includes): abs_includes = includes_dir for py_file in os.listdir(abs_includes): if (py_file.endswith('.vp') or py_file.endswith('.py')) and (not py_file.startswith('_') or py_file == '_list.vp' or py_file == '_list.py' or py_file == '_dict.vp' or py_file == '_dict.py'): py_path = os.path.join(abs_includes, py_file) try: # 复用 parse_python_file 的 AST 缓存 py_content, py_tree = parse_python_file(py_path) if py_tree is None: continue # 计算 include 文件的 SHA1,供特化时统一命名空间 py_sha1: str = compute_sha1(py_content) for node in py_tree.body: if isinstance(node, ast.ClassDef) and hasattr(node, 'type_params') and node.type_params: type_params = [tp.name for tp in node.type_params] generic_class_templates[node.name] = { 'node': node, 'type_params': type_params, 'sha1': py_sha1, } except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"收集泛型类模板失败: {_e}", "Exception") self._shared_generic_class_templates = generic_class_templates # 加载所有 {sha1}.doc.json 到 _shared_doc_meta,供跨模块 __doc__ 查询 shared_doc_meta: dict[str, dict[str, str]] = {} if os.path.isdir(self.temp_dir): for fname in os.listdir(self.temp_dir): if fname.endswith('.doc.json'): sha1_key = fname[:-len('.doc.json')] try: with open(os.path.join(self.temp_dir, fname), 'r', encoding='utf-8') as f: data = json.load(f) if isinstance(data, dict): shared_doc_meta[sha1_key] = {str(k): str(v) for k, v in data.items()} except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"加载 {fname} 失败: {_e}", "Exception") # 也从全局缓存加载 .doc.json(temp_dir 被 --clean 清空时补充) if os.path.isdir(_GLOBAL_CACHE_DIR): for fname in os.listdir(_GLOBAL_CACHE_DIR): if fname.endswith('.doc.json'): sha1_key = fname[:-len('.doc.json')] if sha1_key in shared_doc_meta: continue try: with open(os.path.join(_GLOBAL_CACHE_DIR, fname), 'r', encoding='utf-8') as f: data = json.load(f) if isinstance(data, dict): shared_doc_meta[sha1_key] = {str(k): str(v) for k, v in data.items()} except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"加载全局缓存 {fname} 失败: {_e}", "Exception") self._shared_doc_meta = shared_doc_meta t1 = time.time() _vlog().info(f"[缓存] 共享符号表构建完成 ({t1-t0:.2f}s, {len(self._shared_symbol_table)} 符号, {len(generic_class_templates)} 泛型模板, {len(shared_doc_meta)} doc 元数据)") def _compile_include_sources(self): """扫描项目目录中的汇编存根,编译 includes 目录中发现的 C/C++ 源文件""" extra_sources = list(self.extra_compile_files) scan_dirs = [] if self.src_root and os.path.isdir(self.src_root): scan_dirs.append(self.src_root) # 不再扫描 output_dir 的父目录:当 output_dir 位于顶层时, # 父目录是整个仓库根,会误扫到 ViperOS/includes/vqt6 等无关 C/C++ 源文件。 # 项目特定的 C/汇编文件应放在 src_root 下,或通过 extra_compile_files 指定。 for scan_dir in scan_dirs: for root, dirs, files in os.walk(scan_dir): dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__'] for fname in files: if fname.endswith(('.S', '.s', '.c', '.cpp')): if fname.startswith('test_'): continue fpath = os.path.join(root, fname) if fpath not in extra_sources: extra_sources.append(fpath) if not extra_sources: return cc_map = { '.c': 'gcc', '.cpp': 'g++', '.cc': 'g++', '.cxx': 'g++', '.m': 'clang', '.mm': 'clang++', '.s': 'gcc', '.S': 'gcc', } _vlog().info(f"\n[includes 编译] 找到 {len(extra_sources)} 个源文件") for src_path in extra_sources: _, ext = os.path.splitext(src_path) ext = ext.lower() obj_path = os.path.join(self.output_dir, os.path.splitext(os.path.basename(src_path))[0] + '.obj') if ext in ('.s', '.S') and any('oformat' in f for f in self.linker_flags): cc = 'clang' compile_src = src_path extra_flags = ['-c', '-target', 'x86_64-none-elf', '-o', obj_path, compile_src] else: cc = cc_map.get(ext, 'gcc') extra_flags = ['-c', '-o', obj_path, src_path] _vlog().info(f" 编译: {os.path.basename(src_path)} -> {os.path.basename(obj_path)}") try: cmd = [cc] + extra_flags result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: _vlog().success(" 编译成功") self.extra_link_files.append(obj_path) else: _vlog().error(f" 编译失败: {result.stderr.strip()}") _vlog().error(f"\nC/C++ 源文件编译失败,立即终止编译。") sys.exit(1) except FileNotFoundError: _vlog().error(f" 找不到编译器: {cc}") _vlog().error(f"\n找不到编译器,立即终止编译。") sys.exit(1) except Exception as e: _vlog().error(f" {e}") _vlog().error(f"\n编译异常,立即终止编译。") sys.exit(1) def _strip_debug_sections(self, exe_path: str) -> None: """剥离 .exe 中的调试段(跳过裸机二进制)""" if not os.path.exists(exe_path): return ext = os.path.splitext(exe_path)[1].lower() if ext not in ('.exe', '.dll', '.obj', '.o', '.elf'): _vlog().info(f" [剥离] 跳过({ext} 格式无需剥离)") return for tool in ['llvm-strip', 'strip']: strip_cmd = shutil.which(tool) if strip_cmd: try: result = subprocess.run( [strip_cmd, '--strip-debug', exe_path], capture_output=True, text=True ) if result.returncode == 0: _vlog().info(f" [剥离] 调试段已移除 ({tool})") else: _vlog().warning(f" [剥离] 警告: {result.stderr.strip()}") except Exception as _e: if _config_mode == "strict": raise _vlog().warning(f"剥离可执行文件调试段失败: {_e}", "Exception") break def _link_obj_files(self, active_sha1s: set[str]) -> None: """将 active_sha1s 对应的 .obj 文件和 includes 库文件链接为可执行文件""" obj_files = [] merged_obj = os.path.join(self.output_dir, '_merged.obj') if os.path.isfile(merged_obj): obj_files.append(merged_obj) else: for root, dirs, files in os.walk(self.output_dir): for file in files: if file.endswith('.obj'): sha1 = file[:-4] if sha1 in active_sha1s and sha1 not in self._include_sha1s: obj_files.append(os.path.join(root, file)) if not obj_files and not self.extra_link_files: _vlog().info("[链接] 无文件可链接") return _vlog().info(f"\n[链接] 找到 {len(obj_files)} 个 .obj 文件") all_link_files = obj_files + self.extra_link_files seen = set() all_link_files = [f for f in all_link_files if not (f in seen or seen.add(f))] if self.extra_link_files: _vlog().info(f"[链接] 额外库文件: {len(self.extra_link_files)} 个") for lib in self.extra_link_files: _vlog().info(f" - {os.path.basename(lib)}") if not self.linker_cmd: _vlog().info("[链接] 未配置链接器") return exe_path = os.path.join(self.output_dir, self.linker_output) if self.linker_output else os.path.join(self.output_dir, 'a.exe') is_binary = self.linker_output and self.linker_output.endswith('.bin') # 检查是否直接输出二进制(使用 --oformat binary / -oformat binary 标志) is_direct_binary = is_binary and any('oformat' in f for f in self.linker_flags) if is_binary and not is_direct_binary: link_output = exe_path[:-4] + '_pe.exe' else: link_output = exe_path # 自动复制或生成 linker.ld 到输出目录 linker_ld_src = os.path.join(os.path.dirname(self.output_dir), 'linker.ld') linker_ld_dst = os.path.join(self.output_dir, 'linker.ld') if os.path.isfile(linker_ld_src) and not os.path.isfile(linker_ld_dst): shutil.copy2(linker_ld_src, linker_ld_dst) _vlog().info(f" [链接] 使用链接脚本: linker.ld") elif not os.path.isfile(linker_ld_dst) and any('-T' in f or '-T' in f.replace(',', ' ') for f in self.linker_flags): default_ld = """ENTRY(_start) SECTIONS { . = 0x100000; .text : { *(.text) *(.text.*) } .rodata : { *(.rodata) *(.rodata.*) } .data : { *(.data) *(.data.*) } .bss : { *(.bss) *(.bss.*) } /DISCARD/ : { *(.comment) *(.note) *(.eh_frame) *(.eh_frame_hdr) *(.reloc) *(.rela) *(.debug*) } } """ with open(linker_ld_dst, 'w', encoding='utf-8', newline='\n') as f: f.write(default_ld) _vlog().info(f" [链接] 自动创建默认链接脚本: linker.ld") try: # 将 linker_flags 拆分为非库标志和库标志(-l/-L),库标志放在目标文件之后 # 这样 Unix 风格链接器才能正确解析库中的符号 # -Wl, 选项是链接器全局选项,放在目标文件之前 non_lib_flags = [] lib_flags = [] i = 0 while i < len(self.linker_flags): flag = self.linker_flags[i] if flag in ('-l', '-L') and i + 1 < len(self.linker_flags): # -l xxx / -L xxx 分开的形式 lib_flags.extend([flag, self.linker_flags[i + 1]]) i += 2 elif flag.startswith('-l') or flag.startswith('-L'): # -lxxx / -Lxxx 合并的形式 lib_flags.append(flag) i += 1 else: # 其他所有标志(包括 -Wl, 选项)放在目标文件之前 non_lib_flags.append(flag) i += 1 # ASan 模式:链接 libasan(配合各 .ll 的 opt -passes=asan 插桩) if self.asan_mode: non_lib_flags.append('-fsanitize=address') cmd = [self.linker_cmd] + non_lib_flags + ['-o', link_output] + all_link_files + lib_flags _vlog().info(f" 执行: {' '.join(cmd)}") result = subprocess.run( cmd, capture_output=True, text=True, cwd=self.output_dir or '.' ) if result.returncode == 0: if is_binary and not is_direct_binary: objcopy = shutil.which('llvm-objcopy') or shutil.which('objcopy') if objcopy: conv = subprocess.run( [objcopy, '-O', 'binary', link_output, exe_path], capture_output=True, text=True ) if conv.returncode == 0: os.remove(link_output) _vlog().success(f" 生成裸机二进制: {exe_path}") file_size = os.path.getsize(exe_path) _vlog().info(f" [大小] {file_size} 字节") else: _vlog().error(f" 二进制转换失败: {conv.stderr.strip()}") sys.exit(1) else: _vlog().error(" 找不到 llvm-objcopy,无法生成裸机二进制") sys.exit(1) else: _vlog().success(f" 生成: {exe_path}") file_size = os.path.getsize(exe_path) _vlog().info(f" [大小] {file_size} 字节") if not is_binary: self._strip_debug_sections(exe_path) else: _vlog().error(f" 链接失败: {result.stderr.strip()}") _vlog().error(f"\n链接失败,立即终止编译。") sys.exit(1) except FileNotFoundError: _vlog().error(f" 找不到链接器: {self.linker_cmd}") _vlog().error(f"\n找不到链接器,立即终止编译。") sys.exit(1) except Exception as e: _vlog().error(f" {e}") _vlog().error(f"\n链接异常,立即终止编译。") sys.exit(1)