From 0287dcd2f65ee41fa88e856d6ed80dfde0b531f1 Mon Sep 17 00:00:00 2001 From: hoelee Date: Mon, 12 Oct 2026 11:23:47 +0800 Subject: [PATCH] library_run: fix the affinity pin (it never applied), parametrise mask/pools/stage, name the real refusal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three changes, each from a measured failure on 2026-10-12: * pin_to_e_cores(mask) - the old code called ctypes' GetCurrentProcess() without declaring a restype, so the pseudo-handle -1 was truncated to 32 bits and SetProcessAffinityMask failed with ERROR_INVALID_HANDLE; since the call's return value was never inspected it logged "E-core pinning: on" while the process kept every CPU (read back: 0x3FFFFFF = all 26 CPUs). Now the prototype is declared, the mask is a parameter (--affinity-mask, default unchanged) and the log prints the mask READ BACK from the kernel. * --stage-dir: scratch can live on another drive; the ledger, the locks and gate_policy.json stay in WORK (records, not cache). * the quality gate now says WHICH criterion failed. Both tests shared one "mean X < bar Y" template, so a frame-share refusal was filed as a mean one - ledger example: "ssim min 0.9917 mean 0.9972 (1199 frames) ... systematic: mean 0.9972 < bar 0.995, 1.67% of frames below 0.85" (0912 洞见), a healthy mean refused by the share test. The distinction decides the fix: a mean shortfall answers to a higher CRF, a frame-share defect does not. Engine defaults are unchanged: a standalone run behaves exactly as before. --- scripts/library_run.py | 89 ++++++++++++++++++++++++++++++++++-------- 1 file changed, 72 insertions(+), 17 deletions(-) diff --git a/scripts/library_run.py b/scripts/library_run.py index ebf7134..c4f918f 100644 --- a/scripts/library_run.py +++ b/scripts/library_run.py @@ -872,16 +872,47 @@ def _perframe_best(ref_raw, out_raw, nf, workdir, tag): return len(s), min(s), sum(s) / len(s) -def pin_to_e_cores(): - """x265 tiers: keep the P-cores free for the desktop (measured 3.51x vs 4.37x realtime).""" +def pin_to_e_cores(mask=None): + """x265 tiers: restrict this process - and every child it spawns, because a process affinity + mask is inherited - to the cores the control plane picked. Default = the E-core mask of the + physical host (measured there: 3.51x vs 4.37x realtime when the desktop keeps the P-cores). + + The mask is a parameter, not a constant, because the control plane runs on hosts whose core + layout differs from the one this constant was measured on (2026-10-12: the Windows VM has 26 + logical CPUs and 0x3FF0000 = CPUs 16-25 is 10 of them, i.e. 16 vCPUs were idle), and because + the dashboard's per-core checkboxes have to reach the encoder. `--pools` must be kept in step + with the mask by the caller: x265 will not use more threads than it is given. + """ + m = E_CORE_MASK if mask in (None, "") else int(mask) try: - ctypes.windll.kernel32.SetProcessAffinityMask( - ctypes.windll.kernel32.GetCurrentProcess(), ctypes.c_size_t(E_CORE_MASK)) - return True - except Exception: + k32 = ctypes.windll.kernel32 + # GetCurrentProcess() returns the PSEUDO-handle -1 (0xFFFFFFFFFFFFFFFF). With no declared + # restype ctypes truncates it to a 32-bit int, SetProcessAffinityMask fails with + # ERROR_INVALID_HANDLE (6), and because the old code never inspected the return value it + # logged "E-core pinning: on" while applying NOTHING. Verified 2026-10-12 by reading the + # mask back: the process stayed at 0x3FFFFFF (all 26 CPUs). Declare the prototype. + k32.GetCurrentProcess.restype = ctypes.c_void_p + k32.SetProcessAffinityMask.argtypes = [ctypes.c_void_p, ctypes.c_size_t] + k32.SetProcessAffinityMask.restype = ctypes.c_int + return bool(k32.SetProcessAffinityMask(k32.GetCurrentProcess(), ctypes.c_size_t(m))) + except Exception: # noqa: BLE001 return False +def affinity_mask() -> int: + """The mask this process is actually restricted to (0 when it cannot be read).""" + try: + k32 = ctypes.windll.kernel32 + k32.GetCurrentProcess.restype = ctypes.c_void_p + k32.GetProcessAffinityMask.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_size_t), + ctypes.POINTER(ctypes.c_size_t)] + p, s = ctypes.c_size_t(0), ctypes.c_size_t(0) + k32.GetProcessAffinityMask(k32.GetCurrentProcess(), ctypes.byref(p), ctypes.byref(s)) + return int(p.value) + except Exception: # noqa: BLE001 + return 0 + + def encode(src_local, dst_local, info, args, tier, vbps, q_off=0): cmd = [FFMPEG, "-y", "-hide_banner", "-loglevel", "warning", "-stats", "-i", src_local, "-map", "0:v:0", "-map", "0:a:0?", "-map_metadata", "0"] @@ -1086,17 +1117,27 @@ def process(src, args, dest=None, origin=None): mean_min = SSIM_MEAN_MIN_FALLBACK systematic = bad_share > MAX_BAD_SHARE or ( mean is not None and mean < mean_min) + # Name the criterion that ACTUALLY failed, at the bar that applied to this item. + # Both tests used to print the same "mean X < bar Y" template, so a frame-share + # refusal was reported as a mean one - and "mean 0.9972 < bar 0.995" was written into + # the ledgers and then read as fact (measured 2026-10-12 on 0912 洞见: 20 of 1199 + # frames below 0.85 at a healthy mean, refused by the SHARE test). The distinction + # decides the fix: a mean shortfall answers to a higher CRF, a frame-share defect + # does not, so a wrong label sends the owner to the wrong knob. + why = "" + if mean is not None and mean < mean_min: + why = "mean %.4f < bar %.3f" % (mean, mean_min) + if bad_share > MAX_BAD_SHARE: + why += "%s%.2f%% of frames below %.2f (%d of %d)" % ( + ", " if why else "", 100.0 * bad_share, BAD_FRAME_SSIM, + sstat.get("below", 0), nfr) if systematic: if q_off < QUALITY_STEPS and (attempt == 1): q_off = QUALITY_STEPS - log(" quality gate: mean %.4f < bar %.3f, %d of %d frames below %.2f " - "(%.2f%%) - retrying at quality+%d" % ( - mean, mean_min, sstat.get("below", 0), nfr, - BAD_FRAME_SSIM, 100.0 * bad_share, q_off)) + log(" quality gate: %s - retrying at quality+%d" % (why, q_off)) continue - return ("FAIL quality: %s - systematic: mean %.4f < bar %.3f, %.2f%% of frames " - "below %.2f%s" - % (ssim_note, mean, mean_min, 100.0 * bad_share, BAD_FRAME_SSIM, + return ("FAIL quality: %s - systematic: %s%s" + % (ssim_note, why, " (re-do of an already playable file)" if redo_playable else "")), None if lo is not None and lo < SSIM_TARGET: log(" %d scattered frame(s) below %.2f (%.2f%% of the window) with a healthy " @@ -1198,12 +1239,19 @@ def collect(root): def main(): - global DECODE_CHECK + global DECODE_CHECK, STAGE ap = argparse.ArgumentParser() ap.add_argument("--cq", type=int, default=CQ_NVENC, help="nvenc tier quality (measured default 30)") ap.add_argument("--crf", type=int, default=CRF_X265, help="x265 tier quality (measured default 24)") ap.add_argument("--x265-preset", default=X265_PRESET) - ap.add_argument("--pools", type=int, default=X265_POOLS, help="x265 threads (match the E-core mask)") + ap.add_argument("--pools", type=int, default=X265_POOLS, help="x265 threads (match the affinity mask)") + ap.add_argument("--affinity-mask", default="0x%X" % E_CORE_MASK, + help="hex CPU mask for the x265 tier (default = the E-core mask); the mask is " + "inherited by the ffmpeg children and must agree with --pools") + ap.add_argument("--stage-dir", default=None, + help="scratch directory for this run (default %s). The ledger, the locks and " + "the policy file stay in WORK - only the disposable per-item scratch " + "moves, so the dashboard can point it at a roomier drive." % STAGE) ap.add_argument("--tier", choices=["copy", "nvenc", "x265"], help="only touch files classified into this tier (run tiers separately)") ap.add_argument("--force-tier", choices=["copy", "nvenc", "x265"], default=None, @@ -1238,6 +1286,9 @@ def main(): "THROW THE OUTPUT AWAY - nothing is uploaded and nothing is replaced") args = ap.parse_args() DECODE_CHECK = not args.no_decode_check + if args.stage_dir: + STAGE = os.path.abspath(args.stage_dir) + log("=== stage dir: %s (ledger + locks stay in %s)" % (STAGE, WORK)) os.makedirs(STAGE, exist_ok=True) swept = sweep_stage() @@ -1257,8 +1308,12 @@ def main(): if args.limit: files = files[:args.limit] if args.tier == "x265": - # x265 is CPU: pin it to the E-cores so the P-cores stay free (children inherit the mask). - log("=== E-core pinning: %s" % ("on" if pin_to_e_cores() else "FAILED (running unpinned)")) + # 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. + 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())) # 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.