perf(phase2): llc 编译改多线程异步并行
Phase B 主循环从'每文件同步跑 llc'改为'循环内写 .ll + 构造 LlcTask （命令串 stdlib.malloc，避免主池 reset_to(mark2) 回收悬空），循环后 run_llc_parallel 分批发 LLC_MAX_THREADS(8) 个线程并发跑 llc，join 后 校验结果。worker 只调 stdlib.system()，不做任何池分配（MemBuddy 非线程 安全）；ASan 模式 opt 插桩也在 worker 内并行。includes 编译路径与 单文件 run_pipeline 保持同步不变。
This commit is contained in:
TermiNexus committed
1 parent
eb0398eae3
commit
18b9b8bfa8
2 files changed
+198
-8
No files matched your search
@@ -4,6 +4,8 @@ import stdio
|
||||
import string
|
||||
import stdlib
|
||||
import memhub
|
||||
import viperlib
|
||||
import vthreading
|
||||
import w32.fileio as fileio
|
||||
import w32.win32file
|
||||
import w32.win32base
|
||||
@@ -427,6 +429,155 @@ def compile_ll_to_obj(ir_path: str, output_dir: str, module_name: str, cc_cmd: s
|
||||
return 0
|
||||
|
||||
|
||||
# ============================================================
|
||||
# llc 并行编译（线程异步，Phase B 多模块同时跑 llc）
|
||||
#
|
||||
# 线程安全约束：
|
||||
# - LlcTask 及命令串一律 stdlib.malloc：主池 mb 会被 reset_to(mark2)
|
||||
# 回收，线程运行期间字符串悬空 → 概率崩溃
|
||||
# - worker 只调 stdlib.system()，不做任何池分配（MemBuddy 分配非线程安全）
|
||||
# - 依赖 CRT system() 的多线程安全性（ThreadingTest 已验证）
|
||||
# ============================================================
|
||||
LLC_MAX_THREADS: t.CDefine = 8
|
||||
|
||||
|
||||
class LlcTask:
|
||||
"""llc 异步任务：命令串（stdlib.malloc）+ 模块 SHA1 + 结果"""
|
||||
OptCmd: bytes # ASan 插桩命令（非 ASan 为 None）
|
||||
LlcCmd: bytes # llc 命令
|
||||
ModuleSha1: bytes # 模块 SHA1（报错用）
|
||||
Result: int # worker 写：0 成功，非 0 失败
|
||||
|
||||
|
||||
def init_llc_task(task: LlcTask | t.CPtr, ir_path: str, output_dir: str,
|
||||
module_name: str, cc_cmd: str, cc_flags: str) -> int:
|
||||
"""填充 llc 任务（全部 stdlib.malloc，线程运行期不被回收），0 成功"""
|
||||
if task is None or ir_path is None or output_dir is None or module_name is None:
|
||||
return 1
|
||||
string.memset(task, 0, LlcTask.__sizeof__())
|
||||
task.Result = -1
|
||||
obj_path: str = build_sliced_path(output_dir, module_name, "obj")
|
||||
if obj_path is None:
|
||||
return 1
|
||||
# 模块 SHA1
|
||||
sl: t.CSizeT = string.strlen(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
|
||||
task.LlcCmd = stdlib.malloc(cmd_len)
|
||||
if task.LlcCmd is None:
|
||||
return 1
|
||||
viperlib.snprintf(task.LlcCmd, cmd_len, "%s %s -o %s %s",
|
||||
cc_cmd, cc_flags, obj_path, ir_path)
|
||||
# ASan 模式：opt 插桩命令（worker 先跑，失败即置 Result）
|
||||
if _asan_mode != 0:
|
||||
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
|
||||
task.OptCmd = stdlib.malloc(opt_len)
|
||||
if task.OptCmd is None:
|
||||
return 1
|
||||
viperlib.snprintf(task.OptCmd, opt_len,
|
||||
"opt -passes=forceattrs,asan -force-attribute=sanitize_address"
|
||||
" -asan-globals=false -o %s %s",
|
||||
asan_path, ir_path)
|
||||
return 0
|
||||
|
||||
|
||||
def free_llc_task(task: LlcTask | t.CPtr):
|
||||
"""释放任务（命令串 + 记录）"""
|
||||
if task is None:
|
||||
return
|
||||
if task.OptCmd is not None:
|
||||
stdlib.free(task.OptCmd)
|
||||
if task.LlcCmd is not None:
|
||||
stdlib.free(task.LlcCmd)
|
||||
if task.ModuleSha1 is not None:
|
||||
stdlib.free(task.ModuleSha1)
|
||||
stdlib.free(task)
|
||||
|
||||
|
||||
def _llc_task_worker(arg: t.CPtr) -> t.CPtr:
|
||||
"""llc 线程入口：只调 stdlib.system()，不做任何分配（线程安全）
|
||||
Result 仅在命令链执行完后写入（初始 -1；join 超时保持 -1 → 报失败）"""
|
||||
task: LlcTask | t.CPtr = (LlcTask | t.CPtr)(arg)
|
||||
if task is None:
|
||||
return t.CPtr(0)
|
||||
if task.OptCmd is not None:
|
||||
if stdlib.system(task.OptCmd) != 0:
|
||||
task.Result = 1
|
||||
return t.CPtr(0)
|
||||
if stdlib.system(task.LlcCmd) != 0:
|
||||
task.Result = 1
|
||||
else:
|
||||
task.Result = 0
|
||||
return t.CPtr(0)
|
||||
|
||||
|
||||
def run_llc_parallel(tasks: bytes, task_count: int, max_threads: int) -> int:
|
||||
"""并行执行 llc：分批 max_threads 个线程，全部 join 后检查结果
|
||||
|
||||
tasks: LlcTask 记录连续数组（stdlib.malloc），每个记录由 init_llc_task 填充。
|
||||
Returns: 0 全部成功；非 0 有失败（已打印失败模块与命令）
|
||||
"""
|
||||
if tasks is None or task_count <= 0:
|
||||
return 0
|
||||
if max_threads <= 0:
|
||||
max_threads = 1
|
||||
# Thread 对象指针数组（stdlib，保证线程对象存活到 join）
|
||||
th_ptrs: bytes = stdlib.malloc(t.CSizeT(max_threads) * 8)
|
||||
if th_ptrs is None:
|
||||
VLogger.error("run_llc_parallel: 线程数组分配失败", "PhaseB")
|
||||
return 1
|
||||
slot_size: t.CSizeT = LlcTask.__sizeof__()
|
||||
ret: int = 0
|
||||
batch_start: int = 0
|
||||
while batch_start < task_count:
|
||||
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:
|
||||
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))
|
||||
th.start()
|
||||
# u64 槽存 Thread*（转 u64 存储，读取时再转回指针）
|
||||
c.DerefAs(t.CUInt64T(t.CUInt64T(th_ptrs) + t.CSizeT(j) * 8, t.CPtr), t.CUInt64T(th))
|
||||
j += 1
|
||||
# 本批 join
|
||||
j = 0
|
||||
while j < batch_n:
|
||||
th_j: vthreading.Thread | t.CPtr = (vthreading.Thread | t.CPtr)(t.CPtr(c.Deref(
|
||||
t.CUInt64T(t.CUInt64T(th_ptrs) + t.CSizeT(j) * 8, t.CPtr))))
|
||||
th_j.join(600000)
|
||||
th_j.close()
|
||||
j += 1
|
||||
# 本批结果检查
|
||||
j = 0
|
||||
while j < 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:
|
||||
fb: str = VLogger.fmt_buf()
|
||||
if fb is not None:
|
||||
viperlib.snprintf(fb, 1024, "llc 编译失败: %s (cmd=%s)",
|
||||
(tsk2.ModuleSha1 if tsk2.ModuleSha1 is not None else "?"),
|
||||
(tsk2.LlcCmd if tsk2.LlcCmd is not None else "?"))
|
||||
VLogger.error(fb, "LLC")
|
||||
ret = 1
|
||||
j += 1
|
||||
batch_start += batch_n
|
||||
stdlib.free(th_ptrs)
|
||||
return ret
|
||||
|
||||
|
||||
def _CollectObjFilesImpl(base_dir: str, bd_len: t.CSizeT,
|
||||
out_buf: bytes, out_pos: t.CSizeT, out_size: t.CSizeT) -> t.CSizeT:
|
||||
"""递归扫描 base_dir，收集 .obj 文件路径到 out_buf，返回新的 out_pos
|
||||
|
||||
+47
-8
@@ -725,6 +725,14 @@ def RunMultiFileProject(mb: memhub.MemBuddy | t.CPtr,
|
||||
# 组合 IR 缓冲区大小（4MB，足够容纳本地 stub + 所有依赖 stub + 本地 text）
|
||||
COMBINED_IR_SIZE: t.CSizeT = 4194304
|
||||
|
||||
# llc 任务表（stdlib.malloc，跨线程存活）：连续 LlcTask 记录数组，
|
||||
# 循环内只写 .ll + 构造任务，循环后 run_llc_parallel 分批并行跑 llc
|
||||
llc_tasks: bytes = stdlib.malloc(t.CSizeT(file_count) * BuildPipeline.LlcTask.__sizeof__())
|
||||
if llc_tasks is None:
|
||||
VLogger.error("llc 任务表分配失败", "PhaseB")
|
||||
return 1
|
||||
llc_task_count: int = 0
|
||||
|
||||
for i in range(file_count):
|
||||
ea: t.CUInt64T = t.CUInt64T(entries) + i * entry_size
|
||||
ent: SrcFileEntry | t.CPtr = (SrcFileEntry | t.CPtr)(t.CVoid(ea, t.CPtr))
|
||||
@@ -747,23 +755,37 @@ def RunMultiFileProject(mb: memhub.MemBuddy | t.CPtr,
|
||||
stdlib.free(combined_ir)
|
||||
continue
|
||||
|
||||
# 编译为 .obj
|
||||
cret: int = BuildPipeline.compile_module_to_obj(
|
||||
combined_ir, combined_len, temp_dir, output_dir, ent.Sha1,
|
||||
cc_cmd, cc_flags)
|
||||
# 写 .ll（llc 线程只读此文件）；llc 本身改为循环后批量并行
|
||||
ret_w: int = BuildPipeline.write_ir_to_file(combined_ir, combined_len, temp_dir, ent.Sha1)
|
||||
stdlib.free(combined_ir)
|
||||
if cret != 0:
|
||||
if ret_w != 0:
|
||||
fb: str = VLogger.fmt_buf()
|
||||
if fb is not None:
|
||||
viperlib.snprintf(fb, 1024, "llc 编译失败，终止编译: %s", ent.Path)
|
||||
viperlib.snprintf(fb, 1024, "写 .ll 失败: %s", ent.Path)
|
||||
VLogger.error(fb, "PhaseB")
|
||||
sys.exit(1)
|
||||
|
||||
# 构造 llc 任务（命令串 stdlib.malloc，线程运行期不被 reset 回收）
|
||||
llc_ir_path: str = BuildPipeline.build_sliced_path(temp_dir, ent.Sha1, "ll")
|
||||
if llc_ir_path is None:
|
||||
VLogger.error("llc 输入路径构造失败", "PhaseB")
|
||||
sys.exit(1)
|
||||
tslot: BuildPipeline.LlcTask | t.CPtr = (BuildPipeline.LlcTask | t.CPtr)(t.CVoid(
|
||||
t.CUInt64T(llc_tasks) + t.CSizeT(llc_task_count) * BuildPipeline.LlcTask.__sizeof__(), t.CPtr))
|
||||
if BuildPipeline.init_llc_task(tslot, llc_ir_path, output_dir, ent.Sha1,
|
||||
cc_cmd, cc_flags) != 0:
|
||||
fb: str = VLogger.fmt_buf()
|
||||
if fb is not None:
|
||||
viperlib.snprintf(fb, 1024, "llc 任务构造失败: %s", ent.Path)
|
||||
VLogger.error(fb, "PhaseB")
|
||||
sys.exit(1)
|
||||
llc_task_count += 1
|
||||
|
||||
compiled_count += 1
|
||||
# 显示编译完成的文件
|
||||
# 显示已提交 llc 的文件
|
||||
fb_comp: str = VLogger.fmt_buf()
|
||||
if fb_comp is not None:
|
||||
viperlib.snprintf(fb_comp, 1024, "[%d/%d] 编译完成: %s", i + 1, file_count, ent.Path)
|
||||
viperlib.snprintf(fb_comp, 1024, "[%d/%d] llc 已提交: %s", i + 1, file_count, ent.Path)
|
||||
VLogger.info(fb_comp, "PhaseB")
|
||||
# 构造 .obj 路径，检测是否是 main 模块（test_main.vp/.py 或 main.vp/.py）
|
||||
od_len: t.CSizeT = string.strlen(output_dir)
|
||||
@@ -795,6 +817,23 @@ def RunMultiFileProject(mb: memhub.MemBuddy | t.CPtr,
|
||||
else:
|
||||
VLogger.error(".obj 路径缓冲区不足", "PhaseB")
|
||||
stdlib.free(obj_path_sliced)
|
||||
|
||||
# 并行执行全部 llc（分批 LLC_MAX_THREADS 线程）
|
||||
ret_p: int = BuildPipeline.run_llc_parallel(llc_tasks, llc_task_count,
|
||||
BuildPipeline.LLC_MAX_THREADS)
|
||||
if ret_p != 0:
|
||||
VLogger.error("llc 编译失败，终止编译", "PhaseB")
|
||||
sys.exit(1)
|
||||
|
||||
# 释放 llc 任务（命令串 + 记录）
|
||||
ti: int = 0
|
||||
while ti < llc_task_count:
|
||||
tfree: BuildPipeline.LlcTask | t.CPtr = (BuildPipeline.LlcTask | t.CPtr)(t.CVoid(
|
||||
t.CUInt64T(llc_tasks) + t.CSizeT(ti) * BuildPipeline.LlcTask.__sizeof__(), t.CPtr))
|
||||
BuildPipeline.free_llc_task(tfree)
|
||||
ti += 1
|
||||
stdlib.free(llc_tasks)
|
||||
|
||||
fb: str = VLogger.fmt_buf()
|
||||
if fb is not None:
|
||||
viperlib.snprintf(fb, 1024, "编译完成: %d/%d", compiled_count, file_count)
|
||||
|
||||
Reference in new issue
Block a user