perf(compile): pyi 树 pickle 落盘 + worker initializer 预热，消除进程间重复 parse
43 个 worker 任务分 ~16 个 spawn 进程，每进程重新 parse 全部依赖 .vpi（4479 次 vast.parse、245s）。主进程把 temp 下全部 .vpi 的 vast AST 树 pickle 落盘 temp/_pyi_cache_v1.pkl（签名=(名,size,mtime)， 变化才重建），两处 ProcessPoolExecutor 加 initializer 预热 ImportHandle 类级 _pyi_cache（优先 pickle.load，失败回退逐文件 parse）。 TPC_WORKER_WARMUP=0 可关闭对比。预计进程内首次解析尖峰降为一次 load。
This commit is contained in:
TermiNexus committed
1 parent
9605af8ea1
commit
b760082be8
1 file changed
+154
-2
@@ -222,6 +222,142 @@ _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] = {}
|
||||
for p in paths:
|
||||
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
|
||||
@@ -377,6 +513,12 @@ class Phase2Translator:
|
||||
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）。签名未变时零开销跳过。
|
||||
if 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())
|
||||
@@ -571,7 +713,11 @@ class Phase2Translator:
|
||||
# 复用 self._executor_pool：避免局部变量 GC 时 finalizer 阻塞 ~99s
|
||||
# （join_executor_internals 在 GC 时同步等待 worker 退出队列）
|
||||
if self._executor_pool is None:
|
||||
self._executor_pool = ProcessPoolExecutor(max_workers=n_workers)
|
||||
# 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），
|
||||
@@ -2312,7 +2458,13 @@ class Phase2Translator:
|
||||
}
|
||||
# 复用 self._executor_pool（与 App 源文件阶段共享），避免 GC 阻塞
|
||||
if self._executor_pool is None:
|
||||
self._executor_pool = ProcessPoolExecutor(max_workers=n_workers)
|
||||
# 新池：刷新 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()
|
||||
|
||||
Reference in new issue
Block a user