fix(phase2): llc 线程 join 改用句柄数组替代 Thread 对象存取

插桩证实 list[vthreading.Thread|t.CPtr] 的 get(0)/get(1) 返回同一
对象指针(栈对象地址复用),j=0 的 close() 把 handle 置 NULL 后 j=1
的 join 用 NULL 句柄 → WaitForSingleObject 立即 WAIT_FAILED → alive=0
→ e5fb 线程句柄从未被 join(失败槽 Result=-1 误报)。改为 start()
后立即取出 handle 存入 list[t.CPtr](8 字节指针存取稳定),join/close
直接对句柄 WaitForSingleObject/CloseHandle,不再依赖 Thread 对象存续;
join 索引用全局 batch_start+j(handles 跨批次累积)。
This commit is contained in:
TermiNexus committed 2026-08-19 22:19:33 +08:00
1 parent 88778dc422
commit 0810f9f9e8
1 file changed
+22 -28
+22 -28
View File
@@ -603,20 +603,16 @@ def run_llc_parallel(tasks: bytes, task_count: int, max_threads: int) -> int:
return 0
if max_threads <= 0:
max_threads = 1
# Thread 对象数组(list 保类型):c.Deref 对 i8* 无法推断加载类型时
# 只读 1 字节 → 垃圾指针 → join/is_alive 立即失败(ClassArrayTest llc
# "5 秒误报失败"根因);list[Thread].get 返回类型明确,无推断问题。
# 注意:list[T](pool, N) 第二参数是 elem_size(元素大小)不是容量!
# 此前传 max_threads=16 → 槽间距 16 字节 < Thread 结构体大小,set 时
# 元素重叠覆盖(get 返回混合垃圾),且初始容量仅 8,set(8..15) 越界写
# 破坏堆 → llc 成功后随机 ACCESS_VIOLATION。
# 且元素类型必须用 Thread | t.CPtr(指针):list[vthreading.Thread]
# 的 append 存 8 字节(th 指针)但 get 按 Thread 结构体返回槽地址 →
# join 把对象地址当句柄 → WAIT_FAILED 立即返回 alive=0(误报失败)。
# list[Thread | t.CPtr] 与 list[ast.AST | t.CPtr] 同理,store/load 一致。
th_list: list[vthreading.Thread | t.CPtr] | t.CPtr = list[vthreading.Thread | t.CPtr](_mbuddy)
if th_list is None:
VLogger.error("run_llc_parallel: 线程数组分配失败", "PhaseB")
# 线程句柄数组(list[t.CPtr] 8 字节指针存取稳定):插桩证实
# list[vthreading.Thread|t.CPtr] 的 get(0)/get(1) 返回同一对象指针
# (栈对象地址复用),且 j=0 的 close() 把 handle 置 NULL 后 j=1 的
# join 用 NULL 句柄 → WaitForSingleObject 立即 WAIT_FAILED → alive=0
# → e5fb 线程句柄从未被 join(失败槽 Result=-1)。改为 start() 后
# 立即取出 handle 值存入句柄数组,join/close 直接对句柄操作,
# 不再依赖 Thread 对象的存续。
handles: list[t.CPtr] | t.CPtr = list[t.CPtr](_mbuddy)
if handles is None:
VLogger.error("run_llc_parallel: 线程句柄数组分配失败", "PhaseB")
return 1
slot_size: t.CSizeT = LlcTask.__sizeof__()
ret: int = 0
@@ -632,27 +628,25 @@ def run_llc_parallel(tasks: bytes, task_count: int, max_threads: int) -> int:
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()
th_list.append(th)
# start() 后立即取出句柄(Thread 对象栈地址可能被后续构造复用,
# join 时再读 self.handle 可能已被 close 清空/覆盖)
handles.append(th.handle)
j += 1
# 本批 join
# 本批 join(直接对句柄操作,不依赖 Thread 对象;handles 跨批次
# 累积,全局索引 = batch_start + j)
j = 0
while j < batch_n:
th_j: vthreading.Thread | t.CPtr = th_list.get(j)
#region debug-point LLC9
stdio.printf("[LLC-DBG2] j=%d ptr=%llx handle=%llx started=%d\n", j,
t.CUInt64T(th_j), t.CUInt64T(th_j.handle), th_j.started)
#endregion
h_j: t.CPtr = handles.get(batch_start + j)
#region debug-point LLC7
stdio.printf("[LLC-DBG] join 前 %d\n", j)
#endregion
th_j.join(600000)
#region debug-point LLC10
stdio.printf("[LLC-DBG2] j=%d alive=%d\n", j, th_j.is_alive())
stdio.printf("[LLC-DBG] join 前 %d handle=%llx\n", j, t.CUInt64T(h_j))
#endregion
wr: t.CUInt32T = win32sync.WaitForSingleObject(h_j, 600000)
#region debug-point LLC8
stdio.printf("[LLC-DBG] join 后 %d alive=%d\n", j, th_j.is_alive())
stdio.printf("[LLC-DBG] join 后 %d wait=%u\n", j, wr)
#endregion
th_j.close()
if wr != 0:
VLogger.error("llc 线程 join 超时/失败(600 秒未完成)", "PhaseB")
w32.win32base.CloseHandle(h_j)
j += 1
# 本批结果检查
j = 0