From 61d41707e948954d5e41bf152ff4163ca93c923f Mon Sep 17 00:00:00 2001 From: hoelee Date: Mon, 12 Oct 2026 18:42:02 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE:=20=E5=AD=90=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=E9=87=8D=E9=92=89=20+=20=E8=AE=A4=E8=AF=86=20Windows=20?= =?UTF-8?q?=E4=B8=8D=E7=BB=A7=E6=89=BF=20affinity=20=E7=9A=84=E4=BA=8B?= =?UTF-8?q?=E5=AE=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pin_child()/run_pinned() 在每次 spawn 后尽力钉住子进程（含 SSIM/VMAF 闸门解码）， 但实测 ffmpeg/x265 会拒绝外部 SetProcessAffinityMask(ACCESS_DENIED)，真正生效的是控制面 给引擎所在 Job Object 设的亲和限（后代在创建时继承 job）。注释写明这条事实，别再指望继承。 --- scripts/library_run.py | 75 +++++++++++++++++++++++++++++++++++++----- 1 file changed, 66 insertions(+), 9 deletions(-) diff --git a/scripts/library_run.py b/scripts/library_run.py index 0f1a83c..79618f8 100644 --- a/scripts/library_run.py +++ b/scripts/library_run.py @@ -43,6 +43,54 @@ FFPROBE = T.FFPROBE # Windows Terminal window per spawned child while two encodes ran. CREATE_NO_WINDOW suppresses that # (0 on POSIX, where the flag does not exist). NOWIN = getattr(subprocess, "CREATE_NO_WINDOW", 0) + +# The cores THIS run may use for its heavy children (0 = do not restrict). Set from --affinity-mask. +CHILD_MASK = 0 + + +def pin_child(pid): + """Restrict a freshly spawned child to CHILD_MASK. MUST be called after every spawn. + + Windows does NOT carry an affinity mask across CreateProcess: the child starts with the full + system mask, unlike Linux where sched_setaffinity IS inherited. The engine's own pinning + therefore only ever covered itself - measured 2026-10-12, the engine sat at 0xF0F0 while all + three of its ffmpeg children ran at 0x3FFFFFF (all 26 CPUs): every per-task core selection was + cosmetic for the encoder, and the gate decodes (SSIM/VMAF) competed with all three encodes. + Verified by reading ProcessorAffinity back off the child, not by trusting this call's return. + """ + m = CHILD_MASK + if not m: + return False + try: + k32 = ctypes.windll.kernel32 + k32.OpenProcess.restype = ctypes.c_void_p + k32.OpenProcess.argtypes = [ctypes.c_uint32, ctypes.c_int, ctypes.c_uint32] + k32.SetProcessAffinityMask.argtypes = [ctypes.c_void_p, ctypes.c_size_t] + k32.SetProcessAffinityMask.restype = ctypes.c_int + k32.CloseHandle.argtypes = [ctypes.c_void_p] + h = k32.OpenProcess(0x0200 | 0x0400, 0, int(pid)) # SET_INFORMATION | QUERY_INFORMATION + if not h: + return False # child already gone - nothing to pin + try: + return bool(k32.SetProcessAffinityMask(h, ctypes.c_size_t(m))) + finally: + k32.CloseHandle(h) + except Exception: # noqa: BLE001 + return False + + +def run_pinned(cmd, timeout=None, **kw): + """subprocess.run() with the child pinned to CHILD_MASK (see pin_child).""" + kw.setdefault("creationflags", NOWIN) + p = subprocess.Popen(cmd, **kw) + pin_child(p.pid) + try: + out, err = p.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + p.kill() + out, err = p.communicate() + raise subprocess.TimeoutExpired(cmd, timeout, output=out, stderr=err) + return subprocess.CompletedProcess(cmd, p.returncode, out, err) ROOT = r"Z:\Class\樊登读书\樊登读书会（每周更新）" WORK = r"C:\AI\transcode-run" STAGE = os.path.join(WORK, "stage") @@ -295,8 +343,8 @@ def probe(p, tries=3): def run(cmd, timeout=86400, **kw): kw.setdefault("creationflags", NOWIN) # never a visible console per call - return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", - errors="replace", timeout=timeout, **kw) + return run_pinned(cmd, timeout=timeout, capture_output=True, text=True, + encoding="utf-8", errors="replace", **kw) STALL_S = 900 # an encode with no output growth for this long is wedged, not slow @@ -322,6 +370,7 @@ def run_watched(cmd, watch, stall_s=STALL_S, timeout=86400): err = "" with open(errf, "wb") as ferr: p = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=ferr, creationflags=NOWIN) + pin_child(p.pid) # the encoder itself: Windows children do not inherit the mask last, last_t, t0 = -1, time.time(), time.time() while p.poll() is None: time.sleep(30) @@ -633,7 +682,7 @@ def ssim_vs_source(src_local, out_local, info, workdir, tag, dur=SSIM_WINDOW): "frame-by-frame comparison is not applicable" % (n_src, n_out, 100.0 * (n_out - n_src) / n_src)), {} try: - subprocess.run([FFMPEG, "-y", "-v", "error", "-f", "rawvideo", "-pix_fmt", "yuv420p", "-s", cmp_scale, + run_pinned([FFMPEG, "-y", "-v", "error", "-f", "rawvideo", "-pix_fmt", "yuv420p", "-s", cmp_scale, "-i", ref, "-f", "rawvideo", "-pix_fmt", "yuv420p", "-s", cmp_scale, "-i", out, "-lavfi", "ssim=stats_file=ssim_%s.log" % tag, "-f", "null", "-"], capture_output=True, cwd=workdir, timeout=900, @@ -770,7 +819,7 @@ def _ssim_series(ref_raw, out_raw, off, nf, workdir, tag): _copy_frames(out_raw, pb, max(0, -off), k, fs) log = os.path.join(workdir, "pair_%s.log" % tag) try: - subprocess.run([FFMPEG, "-y", "-v", "error", "-f", "rawvideo", "-pix_fmt", "yuv420p", "-s", "480:270", + run_pinned([FFMPEG, "-y", "-v", "error", "-f", "rawvideo", "-pix_fmt", "yuv420p", "-s", "480:270", "-i", pa, "-f", "rawvideo", "-pix_fmt", "yuv420p", "-s", "480:270", "-i", pb, "-lavfi", "ssim=stats_file=%s" % os.path.basename(log), "-f", "null", "-"], capture_output=True, cwd=workdir, timeout=900, @@ -822,7 +871,7 @@ def vmaf_of_pair(ref_raw, out_raw, off, nf, workdir, tag): try: _copy_frames(ref_raw, pa, max(0, off), k, fs) _copy_frames(out_raw, pb, max(0, -off), k, fs) - subprocess.run([FFMPEG, "-y", "-v", "error"] + raw + ["-i", pb] + raw + ["-i", pa, + run_pinned([FFMPEG, "-y", "-v", "error"] + raw + ["-i", pb] + raw + ["-i", pa, "-lavfi", "libvmaf=log_path=%s:log_fmt=json" % os.path.basename(log), "-f", "null", "-"], capture_output=True, cwd=workdir, timeout=900, creationflags=NOWIN) @@ -1311,12 +1360,20 @@ def main(): if args.limit: files = files[:args.limit] if args.tier == "x265": - # x265 is CPU: pin it to the cores the control plane chose so the desktop keeps the rest - # (children inherit the mask). Default = the E-core mask of the host this was measured on. + # x265 is CPU: pin it to the cores the control plane chose so the desktop keeps the rest. + # NB pin_to_e_cores only covers THIS process - Windows does not inherit an affinity mask, so + # every heavy child is pinned separately (detached=False, pin_child right after spawn). + global CHILD_MASK mask = int(str(args.affinity_mask), 0) pinned = pin_to_e_cores(mask) - log("=== affinity pinning: mask=0x%X pools=%d -> %s (read back 0x%X)" % ( - mask, args.pools, "on" if pinned else "FAILED (running unpinned)", affinity_mask())) + CHILD_MASK = mask if pinned else 0 + # The child pin is best effort and rarely lands: ffmpeg/x265 refuse SetProcessAffinityMask + # from outside (ERROR_ACCESS_DENIED - x265 pins its own threads for NUMA), which is why the + # control plane also puts the engine in a Job Object WITH an affinity limit: children inherit + # job membership, and the scheduler enforces the limit on all of them. + log("=== affinity pinning: mask=0x%X pools=%d -> %s (read back 0x%X; children inherit the " + "control plane's job affinity, per-child pin is best effort)" % ( + mask, args.pools, "on" if pinned else "FAILED (running unpinned)", affinity_mask())) # One lock PER TIER, not one global lock: the copy tier uses no encoder, the x265 tier is CPU # and the nvenc tier is the GPU ASIC, so they are safe to run side by side whereas two runs of # the same tier would race on dst and starve a shared unit.