stage cleanup: purge each item's gate artefacts, sweep dead leftovers, state bug 13
Every item left its SSIM windows (291 MB each), per-offset pairs, VMAF pairs and stats logs behind; anything killed mid-item left them for ever. Measured 2026-10-12: 6 windows + 123 gate logs, work drive at 3.9 GB, both lanes idle on the guard. process() now purges its own tag/pid artefacts in the finally, and sweep_stage() clears leftovers older than STAGE_MAX_AGE_H while skipping any tag that has a fresh <tag>_out.mp4.err (i.e. is encoding right now). Regression: scripts/test_stage_cleanup.py.
This commit is contained in:
@@ -160,6 +160,84 @@ def disp(p):
|
||||
POLICY = os.path.join(WORK, "gate_policy.json")
|
||||
PARK_AFTER = 2
|
||||
|
||||
# A leftover in the stage dir older than this was left by a run that died (pause, reboot, the stall
|
||||
# watchdog, a crash): reclaim it. Measured 2026-10-12: 6 x 291 MB SSIM windows + 123 gate logs had
|
||||
# piled up, and the work drive got full enough that the guards refused every new item.
|
||||
STAGE_MAX_AGE_H = float(os.environ.get("RT_STAGE_MAX_AGE_H", "6"))
|
||||
|
||||
|
||||
def purge_item_artifacts(tag, workdir=None):
|
||||
"""Delete THIS item's gate artefacts (window raws, per-offset pairs, VMAF pairs, stats logs).
|
||||
|
||||
Names carry the item tag (the stage hash) or this process's pid, so a lane running at the same
|
||||
time - it shares the stage dir - can never match. Called from process()'s finally, so a clean
|
||||
item leaves nothing behind even before the supervisor's janitor looks.
|
||||
"""
|
||||
workdir = workdir or STAGE
|
||||
keys = (tag, str(os.getpid()))
|
||||
removed = 0
|
||||
try:
|
||||
names = os.listdir(workdir)
|
||||
except OSError:
|
||||
return 0
|
||||
for name in names:
|
||||
if not name.startswith(("ssim_", "pair_", "vmaf_")):
|
||||
continue
|
||||
if not any(k and k in name for k in keys):
|
||||
continue
|
||||
try:
|
||||
os.remove(os.path.join(workdir, name))
|
||||
removed += 1
|
||||
except OSError:
|
||||
pass
|
||||
return removed
|
||||
|
||||
|
||||
def sweep_stage(workdir=None, max_age_h=None):
|
||||
"""Reclaim stage files left by runs that died. Never touches a tag that is encoding right now.
|
||||
|
||||
A live item is recognised by ITS OWN progress file (`<tag>_out.mp4.err`, rewritten every 0.5 s by
|
||||
ffmpeg and removed when the encode ends), so an in-flight tag is skipped wholesale.
|
||||
"""
|
||||
workdir = workdir or STAGE
|
||||
max_age_h = STAGE_MAX_AGE_H if max_age_h is None else max_age_h
|
||||
if not os.path.isdir(workdir):
|
||||
return {"files": 0, "bytes": 0}
|
||||
now = time.time()
|
||||
cutoff = now - max_age_h * 3600
|
||||
try:
|
||||
entries = os.listdir(workdir)
|
||||
except OSError:
|
||||
return {"files": 0, "bytes": 0}
|
||||
live_tags = set()
|
||||
for name in entries:
|
||||
if name.endswith(".err"):
|
||||
try:
|
||||
if now - os.path.getmtime(os.path.join(workdir, name)) < max_age_h * 3600:
|
||||
live_tags.add(name.split("_")[0])
|
||||
except OSError:
|
||||
pass
|
||||
files = size = 0
|
||||
for name in entries:
|
||||
p = os.path.join(workdir, name)
|
||||
try:
|
||||
st = os.stat(p)
|
||||
except OSError:
|
||||
continue
|
||||
if not os.path.isfile(p):
|
||||
continue
|
||||
if not (name.startswith(("ssim_", "pair_", "vmaf_")) or "_src." in name or "_out." in name):
|
||||
continue
|
||||
if st.st_mtime >= cutoff or any(t and t in name for t in live_tags):
|
||||
continue
|
||||
try:
|
||||
os.remove(p)
|
||||
files += 1
|
||||
size += st.st_size
|
||||
except OSError:
|
||||
pass
|
||||
return {"files": files, "bytes": size}
|
||||
|
||||
|
||||
def load_policy():
|
||||
"""{"accept": set(disp keys), "park": {disp key: failed rounds}} - missing file means neither."""
|
||||
@@ -1104,6 +1182,9 @@ def process(src, args, dest=None, origin=None):
|
||||
os.remove(p)
|
||||
except OSError:
|
||||
pass
|
||||
# ... and the gate's own artefacts (window raws, pairs, VMAF, stats logs): they are named
|
||||
# with this item's tag or this pid, and leaving them behind is what filled the stage dir.
|
||||
purge_item_artifacts(tag)
|
||||
|
||||
|
||||
def collect(root):
|
||||
@@ -1159,6 +1240,9 @@ def main():
|
||||
DECODE_CHECK = not args.no_decode_check
|
||||
|
||||
os.makedirs(STAGE, exist_ok=True)
|
||||
swept = sweep_stage()
|
||||
if swept["files"]:
|
||||
log(" stage sweep: reclaimed %d dead file(s), %.0f MB" % (swept["files"], swept["bytes"] / 1e6))
|
||||
if args.path:
|
||||
files = [args.path]
|
||||
if not os.path.exists(args.path):
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
"""Regression for the engine's own stage cleanup (library_run.purge_item_artifacts / sweep_stage).
|
||||
|
||||
The engine shares the stage dir between lanes, so:
|
||||
* purge_item_artifacts may only delete files carrying THIS item's tag or THIS process's pid,
|
||||
* sweep_stage may only delete leftovers older than max_age_h, and must skip every file of a tag
|
||||
that still has a fresh `<tag>_out.mp4.err` progress file (that tag is encoding right now).
|
||||
|
||||
python scripts/test_stage_cleanup.py
|
||||
"""
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
import library_run as LR # noqa: E402
|
||||
|
||||
MINE, OTHER = "aa11bb22cc33", "998877665544"
|
||||
root = tempfile.mkdtemp(prefix="stage_cleanup_")
|
||||
OLD = 12 * 3600
|
||||
|
||||
|
||||
def touch(name, age_s=0.0, size=1024):
|
||||
p = os.path.join(root, name)
|
||||
with open(p, "wb") as fh:
|
||||
fh.write(b"\0" * size)
|
||||
if age_s:
|
||||
t = time.time() - age_s
|
||||
os.utime(p, (t, t))
|
||||
return p
|
||||
|
||||
|
||||
# purge: only MINE / this pid
|
||||
touch("ssim_%s.log" % MINE)
|
||||
touch("ssim_%d_%s_ref.yuv" % (os.getpid(), MINE))
|
||||
touch("pair_a_%s.yuv" % MINE)
|
||||
touch("vmaf_%s.json" % os.getpid())
|
||||
touch("ssim_%s.log" % OTHER) # another lane's item -> keep
|
||||
touch("pair_a_%s.yuv" % OTHER) # another lane's item -> keep
|
||||
n = LR.purge_item_artifacts(MINE, workdir=root)
|
||||
left = sorted(os.listdir(root))
|
||||
purge_checks = {
|
||||
"purged_mine": n == 4,
|
||||
"other_lane_untouched": all(("ssim_%s.log" % OTHER) in left and ("pair_a_%s.yuv" % OTHER) in left
|
||||
for _ in [0]),
|
||||
}
|
||||
|
||||
# sweep: old leftovers go, a live tag (fresh .err) stays whole
|
||||
root2 = tempfile.mkdtemp(prefix="stage_sweep_")
|
||||
root = root2
|
||||
touch("ssim_1_dead1_ref.yuv", OLD, 2048)
|
||||
touch("dead99_src.mp4", OLD, 4096)
|
||||
touch("ssim_2_live42_ref.yuv", OLD, 2048) # same tag as the fresh .err below -> keep
|
||||
touch("live42_src.mp4", OLD, 4096)
|
||||
touch("live42_out.mp4.err", 30, 256) # fresh -> tag live42 is encoding now
|
||||
touch("ledger.jsonl", OLD, 8192)
|
||||
st = LR.sweep_stage(workdir=root2, max_age_h=6)
|
||||
left2 = sorted(os.listdir(root2))
|
||||
sweep_checks = {
|
||||
"dead_leftovers_removed": "ssim_1_dead1_ref.yuv" not in left2 and "dead99_src.mp4" not in left2,
|
||||
"live_tag_kept_whole": "ssim_2_live42_ref.yuv" in left2 and "live42_src.mp4" in left2,
|
||||
"ledger_never_touched": "ledger.jsonl" in left2,
|
||||
"freed_bytes_reported": st["bytes"] == 2048 + 4096,
|
||||
}
|
||||
|
||||
ok = True
|
||||
for k, v in list(purge_checks.items()) + list(sweep_checks.items()):
|
||||
print("%-26s %s" % (k, "PASS" if v else "FAIL"))
|
||||
ok = ok and v
|
||||
print("purge removed %d, sweep %s" % (n, st))
|
||||
shutil.rmtree(tempfile.gettempdir() + "/" + os.path.basename(root), ignore_errors=True)
|
||||
sys.exit(0 if ok else 1)
|
||||
Reference in New Issue
Block a user