fix(BuildPipeline): run_llc_parallel 分批步进错误，修复 llc 线程 join 误报 600 秒超时
batch_start 改用 range(task_count)（步进 1）后，后续批次 join 循环 handles.get(batch_start+j) 取到上一批已 CloseHandle 的句柄 → WaitForSingleObject 返回 WAIT_FAILED(-1) 立即返回 → 误报 llc 线程 join 超时/失败（实际未到 600 秒）。 修复：for batch_start in range(0, task_count, max_threads) 保持原分批语义（步进 max_threads）。
This commit is contained in:
TermiNexus committed
1 parent
f7f3bbac02
commit
a088f21bce
1 file changed
+42
-65
@@ -56,13 +56,7 @@ import lib.core.StubMergerCollect as StubMergerCollect
|
||||
_mbuddy: memhub.MemBuddy | t.CPtr
|
||||
|
||||
# ASan 模式（--asan）：对 .ll 先 opt -passes=asan 插桩，链接加 -fsanitize=address
|
||||
_asan_mode: t.CInt = 0
|
||||
|
||||
|
||||
def set_asan_mode(flag: int) -> None:
|
||||
"""设置 ASan 模式（1=启用，0=禁用）"""
|
||||
global _asan_mode
|
||||
_asan_mode = flag
|
||||
_asan_mode: bool = 0
|
||||
|
||||
# 源代码缓冲区大小（1MB）
|
||||
SRC_BUF_SIZE: t.CDefine = 1048576 # 2^20 = 1MB
|
||||
@@ -70,7 +64,6 @@ SRC_BUF_SIZE: t.CDefine = 1048576 # 2^20 = 1MB
|
||||
# .vpi 存根缓冲区大小（256KB，与 Phase1 一致）
|
||||
PYI_BUF_SIZE: t.CSizeT = 262144
|
||||
|
||||
|
||||
# ============================================================
|
||||
# TranslateFileGetTrans - 翻译单个文件，返回 Translator 对象
|
||||
#
|
||||
@@ -100,11 +93,9 @@ def TranslateFileGetTrans(mb: memhub.MemBuddy | t.CPtr, file_path: str,
|
||||
declare_only: 0=全量翻译（默认），1=仅注册 struct/enum/union（不翻译方法体）
|
||||
temp_dir: temp 目录路径（非 None 时在翻译前生成 .vpi 存根文件）
|
||||
"""
|
||||
if file_path is None:
|
||||
return None
|
||||
if file_path is None: return None
|
||||
f: fileio.File | t.CPtr = fileio.File(file_path, fileio.MODE.R)
|
||||
if f.closed:
|
||||
return None
|
||||
if f.closed: return None
|
||||
src_buf: bytes = stdlib.malloc(SRC_BUF_SIZE)
|
||||
if src_buf is None:
|
||||
f.close()
|
||||
@@ -133,10 +124,9 @@ def TranslateFileGetTrans(mb: memhub.MemBuddy | t.CPtr, file_path: str,
|
||||
HandlesType.dump_parse_traceback(file_path)
|
||||
fb_parse: str = VLogger.fmt_buf()
|
||||
if fb_parse is not None:
|
||||
err_line: t.CInt = ast.get_parse_err_line()
|
||||
err_col: t.CInt = ast.get_parse_err_col()
|
||||
viperlib.snprintf(fb_parse, 1024,
|
||||
"解析失败: %s (line %d col %d)", file_path, err_line, err_col)
|
||||
err_line: int = ast.get_parse_err_line()
|
||||
err_col: int = ast.get_parse_err_col()
|
||||
viperlib.snprintf(fb_parse, 1024, "解析失败: %s (line %d col %d)", file_path, err_line, err_col)
|
||||
VLogger.error(fb_parse, "PARSE")
|
||||
return None
|
||||
# 生成 .vpi 存根文件（从 AST 直接生成，在 tr.translate 之前，与 Phase1 一致）
|
||||
@@ -148,7 +138,7 @@ def TranslateFileGetTrans(mb: memhub.MemBuddy | t.CPtr, file_path: str,
|
||||
pyi_pos: t.CSizeT = StubConverter._GeneratePyiFromAst(mb, tree, None, pyi_buf, PYI_BUF_SIZE)
|
||||
if pyi_pos > 0:
|
||||
# 写入 {sha1}.vpi 到 temp_dir（切片路径）
|
||||
td_len_vpi: t.CSizeT = string.strlen(temp_dir)
|
||||
td_len_vpi: t.CSizeT = len(temp_dir)
|
||||
pyi_path: str = StubMergerCollect._sliced_path(temp_dir, td_len_vpi, sha1_val, "vpi")
|
||||
if pyi_path is not None:
|
||||
pf: fileio.File | t.CPtr = fileio.File(pyi_path, fileio.MODE.W)
|
||||
@@ -257,7 +247,7 @@ def ensure_dir(path: str) -> int:
|
||||
if path is None:
|
||||
return 1
|
||||
|
||||
path_len: t.CSizeT = string.strlen(path)
|
||||
path_len: t.CSizeT = len(path)
|
||||
if path_len == 0:
|
||||
return 0
|
||||
|
||||
@@ -269,16 +259,14 @@ def ensure_dir(path: str) -> int:
|
||||
|
||||
# 遇到分隔符时临时截断，创建每一层目录
|
||||
# CreateDirectoryA 在目录已存在时返回 0（失败），忽略即可
|
||||
i: int = 0
|
||||
plen: int = path_len
|
||||
while i < plen:
|
||||
for i in range(plen):
|
||||
ch: int = buf[i]
|
||||
if ch == 47 or ch == 92: # '/' = 47, '\\' = 92
|
||||
saved: int = ch
|
||||
buf[i] = '\0'
|
||||
w32.win32file.CreateDirectoryA(buf, None)
|
||||
buf[i] = saved
|
||||
i += 1
|
||||
|
||||
# 创建最终目录
|
||||
w32.win32file.CreateDirectoryA(buf, None)
|
||||
@@ -309,12 +297,12 @@ def build_sliced_path(base_dir: str, sha1: str, ext: str) -> str:
|
||||
return None
|
||||
|
||||
subdir: str = Config.slice_subdir(sha1, Config.Sha1SliceLevel)
|
||||
base_len: t.CSizeT = string.strlen(base_dir)
|
||||
sha1_len: t.CSizeT = string.strlen(sha1)
|
||||
ext_len: t.CSizeT = string.strlen(ext)
|
||||
base_len: t.CSizeT = len(base_dir)
|
||||
sha1_len: t.CSizeT = len(sha1)
|
||||
ext_len: t.CSizeT = len(ext)
|
||||
|
||||
if subdir is not None:
|
||||
sub_len: t.CSizeT = string.strlen(subdir)
|
||||
sub_len: t.CSizeT = len(subdir)
|
||||
# ensure_dir({base_dir}/{subdir})
|
||||
dir_path: str = _mbuddy.alloc(base_len + sub_len + 2)
|
||||
if dir_path is not None:
|
||||
@@ -323,16 +311,14 @@ def build_sliced_path(base_dir: str, sha1: str, ext: str) -> str:
|
||||
# 文件路径: {base_dir}/{subdir}/{sha1}.{ext}
|
||||
path_len: t.CSizeT = base_len + sub_len + sha1_len + ext_len + 4
|
||||
path: str = _mbuddy.alloc(path_len)
|
||||
if path is None:
|
||||
return None
|
||||
if path is None: return None
|
||||
viperlib.snprintf(path, path_len, "%s/%s/%s.%s", base_dir, subdir, sha1, ext)
|
||||
return path
|
||||
else:
|
||||
# 无切片: {base_dir}/{sha1}.{ext}
|
||||
path_len = base_len + sha1_len + ext_len + 3
|
||||
path = _mbuddy.alloc(path_len)
|
||||
if path is None:
|
||||
return None
|
||||
if path is None: return None
|
||||
viperlib.snprintf(path, path_len, "%s/%s.%s", base_dir, sha1, ext)
|
||||
return path
|
||||
|
||||
@@ -349,23 +335,19 @@ def write_ir_to_file(ir_buf: bytes, ir_len: t.CSizeT, output_dir: str, module_na
|
||||
Returns:
|
||||
0 成功，非 0 失败
|
||||
"""
|
||||
if ir_buf is None or output_dir is None or module_name is None:
|
||||
return 1
|
||||
if ir_buf is None or output_dir is None or module_name is None: return 1
|
||||
|
||||
# 构造切片路径: output_dir/{sha1前缀}/{module_name}.ll
|
||||
path: str = build_sliced_path(output_dir, module_name, "ll")
|
||||
if path is None:
|
||||
return 1
|
||||
if path is None: return 1
|
||||
|
||||
# 打开文件写入
|
||||
f: fileio.File | t.CPtr = fileio.File(path, fileio.MODE.W)
|
||||
if f.closed:
|
||||
return 1
|
||||
if f.closed: return 1
|
||||
|
||||
written: t.CInt64T = f.write(ir_buf, ir_len)
|
||||
f.close()
|
||||
if written < 0:
|
||||
return 1
|
||||
if written < 0: return 1
|
||||
return 0
|
||||
|
||||
|
||||
@@ -387,8 +369,7 @@ def compile_ll_to_obj(ir_path: str, output_dir: str, module_name: str, cc_cmd: s
|
||||
|
||||
# 构造切片 .obj 路径: output_dir/{sha1前缀}/{module_name}.obj
|
||||
obj_path: str = build_sliced_path(output_dir, module_name, "obj")
|
||||
if obj_path is None:
|
||||
return 1
|
||||
if obj_path is None: return 1
|
||||
|
||||
# ASan 模式（_asan_mode=1）：先 opt -passes=asan 插桩再 llc。
|
||||
# -asan-globals=false 关闭全局插桩：TPV 对全局变量（如各模块 _g_gen）
|
||||
@@ -398,7 +379,7 @@ def compile_ll_to_obj(ir_path: str, output_dir: str, module_name: str, cc_cmd: s
|
||||
asan_path: str = build_sliced_path(output_dir, module_name, "asan.ll")
|
||||
if asan_path is None:
|
||||
return 1
|
||||
opt_cmd_len: t.CSizeT = string.strlen(asan_path) + string.strlen(ir_path) + 192
|
||||
opt_cmd_len: t.CSizeT = len(asan_path) + len(ir_path) + 192
|
||||
opt_cmd: bytes = _mbuddy.alloc(opt_cmd_len)
|
||||
if opt_cmd is None:
|
||||
return 1
|
||||
@@ -415,7 +396,7 @@ def compile_ll_to_obj(ir_path: str, output_dir: str, module_name: str, cc_cmd: s
|
||||
compile_input = asan_path
|
||||
|
||||
# 构造命令: llc -filetype=obj -o {obj_path} ir_path
|
||||
cmd_len: t.CSizeT = string.strlen(cc_cmd) + string.strlen(cc_flags) + string.strlen(obj_path) + string.strlen(compile_input) + 64
|
||||
cmd_len: t.CSizeT = len(cc_cmd) + len(cc_flags) + len(obj_path) + len(compile_input) + 64
|
||||
cmd: bytes = _mbuddy.alloc(cmd_len)
|
||||
if cmd is None:
|
||||
return 1
|
||||
@@ -465,14 +446,14 @@ def init_llc_task(task: LlcTask | t.CPtr, ir_path: str, output_dir: str,
|
||||
if obj_path is None:
|
||||
return 1
|
||||
# 模块 SHA1
|
||||
sl: t.CSizeT = string.strlen(module_name)
|
||||
sl: t.CSizeT = len(module_name)
|
||||
task.ModuleSha1 = stdlib.malloc(sl + 1)
|
||||
if task.ModuleSha1 is None:
|
||||
return 1
|
||||
string.strcpy(task.ModuleSha1, module_name)
|
||||
# llc 命令
|
||||
cmd_len: t.CSizeT = string.strlen(cc_cmd) + string.strlen(cc_flags) \
|
||||
+ string.strlen(obj_path) + string.strlen(ir_path) + 64
|
||||
cmd_len: t.CSizeT = len(cc_cmd) + len(cc_flags) \
|
||||
+ len(obj_path) + len(ir_path) + 64
|
||||
task.LlcCmd = stdlib.malloc(cmd_len)
|
||||
if task.LlcCmd is None:
|
||||
return 1
|
||||
@@ -483,7 +464,7 @@ def init_llc_task(task: LlcTask | t.CPtr, ir_path: str, output_dir: str,
|
||||
asan_path: str = build_sliced_path(output_dir, module_name, "asan.ll")
|
||||
if asan_path is None:
|
||||
return 1
|
||||
opt_len: t.CSizeT = string.strlen(asan_path) + string.strlen(ir_path) + 192
|
||||
opt_len: t.CSizeT = len(asan_path) + len(ir_path) + 192
|
||||
task.OptCmd = stdlib.malloc(opt_len)
|
||||
if task.OptCmd is None:
|
||||
return 1
|
||||
@@ -521,7 +502,7 @@ def _run_cmd_wait(cmd: str) -> int:
|
||||
"""
|
||||
if cmd is None:
|
||||
return -1
|
||||
cl_len: t.CSizeT = string.strlen(cmd)
|
||||
cl_len: t.CSizeT = len(cmd)
|
||||
cmd_line: bytes = stdlib.malloc(cl_len + 1)
|
||||
if cmd_line is None:
|
||||
return -1
|
||||
@@ -595,14 +576,16 @@ def run_llc_parallel(tasks: bytes, task_count: int, max_threads: int) -> int:
|
||||
return 1
|
||||
slot_size: t.CSizeT = LlcTask.__sizeof__()
|
||||
ret: int = 0
|
||||
batch_start: int = 0
|
||||
while batch_start < task_count:
|
||||
# 分批执行：batch_start 按 max_threads 步进（每批 spawn+join+close 一批句柄）。
|
||||
# 注意：不能 range(task_count)（步进 1）——join 循环 handles.get(batch_start+j)
|
||||
# 会取到上一批已 CloseHandle 的句柄 → WaitForSingleObject 返回 WAIT_FAILED(-1)
|
||||
# 立即返回 → 误报"600 秒超时"（实际未到 600 秒）。
|
||||
for batch_start in range(0, task_count, max_threads):
|
||||
batch_n: int = max_threads
|
||||
if batch_start + batch_n > task_count:
|
||||
batch_n = task_count - batch_start
|
||||
# 本批 spawn
|
||||
j: int = 0
|
||||
while j < batch_n:
|
||||
for j in range(batch_n):
|
||||
tsk: LlcTask | t.CPtr = (LlcTask | t.CPtr)(t.CVoid(
|
||||
t.CUInt64T(tasks) + t.CSizeT(batch_start + j) * slot_size, t.CPtr))
|
||||
th: vthreading.Thread | t.CPtr = vthreading.Thread(_llc_task_worker, t.CPtr(tsk))
|
||||
@@ -610,18 +593,14 @@ def run_llc_parallel(tasks: bytes, task_count: int, max_threads: int) -> int:
|
||||
# start() 后立即取出句柄（Thread 对象栈地址可能被后续构造复用，
|
||||
# join 时再读 self.handle 可能已被 close 清空/覆盖）
|
||||
handles.append(t.CPtr(th.handle))
|
||||
j += 1
|
||||
j = 0
|
||||
while j < batch_n:
|
||||
for j in range(batch_n):
|
||||
h_j: t.CPtr = t.CPtr(handles.get(batch_start + j))
|
||||
wr: t.CUInt32T = win32sync.WaitForSingleObject(h_j, 600000)
|
||||
if wr != 0:
|
||||
VLogger.error("llc 线程 join 超时/失败（600 秒未完成）", "PhaseB")
|
||||
w32.win32base.CloseHandle(h_j)
|
||||
j += 1
|
||||
# 本批结果检查
|
||||
j = 0
|
||||
while j < batch_n:
|
||||
for j in range(batch_n):
|
||||
tsk2: LlcTask | t.CPtr = (LlcTask | t.CPtr)(t.CVoid(
|
||||
t.CUInt64T(tasks) + t.CSizeT(batch_start + j) * slot_size, t.CPtr))
|
||||
if tsk2 is not None and tsk2.Result != 0:
|
||||
@@ -632,8 +611,6 @@ def run_llc_parallel(tasks: bytes, task_count: int, max_threads: int) -> int:
|
||||
(tsk2.LlcCmd if tsk2.LlcCmd is not None else "?"))
|
||||
VLogger.error(fb, "LLC")
|
||||
ret = 1
|
||||
j += 1
|
||||
batch_start += batch_n
|
||||
return ret
|
||||
|
||||
|
||||
@@ -664,7 +641,7 @@ def _CollectObjFilesImpl(base_dir: str, bd_len: t.CSizeT,
|
||||
while True:
|
||||
fname: str = find_data.cFileName
|
||||
if fname is not None:
|
||||
fname_len: t.CSizeT = string.strlen(fname)
|
||||
fname_len: t.CSizeT = len(fname)
|
||||
if fname_len > 0:
|
||||
# 跳过 . 和 ..
|
||||
is_dot: int = 0
|
||||
@@ -682,7 +659,7 @@ def _CollectObjFilesImpl(base_dir: str, bd_len: t.CSizeT,
|
||||
sub_dir: bytes = _mbuddy.alloc(bd_len + fname_len + 2)
|
||||
if sub_dir is not None:
|
||||
viperlib.snprintf(sub_dir, bd_len + fname_len + 2, "%s/%s", base_dir, fname)
|
||||
sub_len: t.CSizeT = string.strlen(sub_dir)
|
||||
sub_len: t.CSizeT = len(sub_dir)
|
||||
out_pos = _CollectObjFilesImpl(sub_dir, sub_len, out_buf, out_pos, out_size)
|
||||
_mbuddy.free(sub_dir)
|
||||
else:
|
||||
@@ -723,7 +700,7 @@ def collect_obj_files(includes_binary_dir: str, out_buf: bytes, out_size: t.CSiz
|
||||
return 0
|
||||
|
||||
out_buf[0] = '\0'
|
||||
dir_len: t.CSizeT = string.strlen(includes_binary_dir)
|
||||
dir_len: t.CSizeT = len(includes_binary_dir)
|
||||
out_pos: t.CSizeT = _CollectObjFilesImpl(includes_binary_dir, dir_len, out_buf, 0, out_size)
|
||||
out_buf[out_pos] = '\0'
|
||||
return out_pos
|
||||
@@ -778,14 +755,14 @@ def link_obj_to_exe(output_dir: str, module_name: str, linker_cmd: str, linker_f
|
||||
# （DLL 版，运行时需 D:/msys64/clang64/bin 在 PATH）
|
||||
use_flags: str = linker_flags
|
||||
if _asan_mode != 0:
|
||||
fl_len: t.CSizeT = string.strlen(linker_flags) + 256
|
||||
fl_len: t.CSizeT = len(linker_flags) + 256
|
||||
use_flags = _mbuddy.alloc(fl_len)
|
||||
if use_flags is None:
|
||||
return 1
|
||||
viperlib.snprintf(use_flags, fl_len,
|
||||
"%s -L D:/msys64/clang64/lib/clang/22/lib/windows -l:libclang_rt.asan_dynamic-x86_64.dll.a -l:libclang_rt.asan_dynamic_runtime_thunk-x86_64.a",
|
||||
linker_flags)
|
||||
cmd_len: t.CSizeT = string.strlen(linker_cmd) + string.strlen(obj_path) + string.strlen(use_flags) + string.strlen(output_dir) + string.strlen(linker_output) + extra_len + 128
|
||||
cmd_len: t.CSizeT = len(linker_cmd) + len(obj_path) + len(use_flags) + len(output_dir) + len(linker_output) + extra_len + 128
|
||||
cmd: bytes = _mbuddy.alloc(cmd_len)
|
||||
if cmd is None:
|
||||
return 1
|
||||
@@ -899,14 +876,14 @@ def link_objs_to_exe(obj_paths: str, obj_paths_len: t.CSizeT,
|
||||
# ASan 模式：显式链接 clang64 的 libclang_rt（同 link_obj_to_exe）
|
||||
use_flags: str = linker_flags
|
||||
if _asan_mode != 0:
|
||||
fl_len: t.CSizeT = string.strlen(linker_flags) + 256
|
||||
fl_len: t.CSizeT = len(linker_flags) + 256
|
||||
use_flags = _mbuddy.alloc(fl_len)
|
||||
if use_flags is None:
|
||||
return 1
|
||||
viperlib.snprintf(use_flags, fl_len,
|
||||
"%s -L D:/msys64/clang64/lib/clang/22/lib/windows -l:libclang_rt.asan_dynamic-x86_64.dll.a -l:libclang_rt.asan_dynamic_runtime_thunk-x86_64.a",
|
||||
linker_flags)
|
||||
cmd_len: t.CSizeT = string.strlen(linker_cmd) + obj_paths_len + string.strlen(use_flags) + string.strlen(linker_output) + extra_len + 128
|
||||
cmd_len: t.CSizeT = len(linker_cmd) + obj_paths_len + len(use_flags) + len(linker_output) + extra_len + 128
|
||||
cmd: bytes = _mbuddy.alloc(cmd_len)
|
||||
if cmd is None:
|
||||
return 1
|
||||
|
||||
Reference in new issue
Block a user