修: 子进程重钉 + 认识 Windows 不继承 affinity 的事实
pin_child()/run_pinned() 在每次 spawn 后尽力钉住子进程（含 SSIM/VMAF 闸门解码）， 但实测 ffmpeg/x265 会拒绝外部 SetProcessAffinityMask(ACCESS_DENIED)，真正生效的是控制面 给引擎所在 Job Object 设的亲和限（后代在创建时继承 job）。注释写明这条事实，别再指望继承。
This commit is contained in:
+66
-9
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user