File size: 3,992 Bytes
e1acc15 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """Read-only, bounded Spark cleanup proof. Dockerd children outlive client scopes.
Wait for anonymous expert/Engram JIT containers even before their first CUDA
context. Never stop a compiler or remove an unrelated container. Any failed
enumeration, inspection or GPU query refuses restoration and retains HOLD.
The exact stdlib source is sent inline; no remote checkout dependency.
"""
from __future__ import annotations
import json
import math
from pathlib import Path
import re
import shlex
import subprocess
import sys
import time
from typing import cast
def checked(argv: list[str], timeout: float) -> str:
return subprocess.check_output(argv, text=True, timeout=timeout)
def snapshot(timeout: float) -> bool:
ids = checked(['docker', 'ps', '-a', '--no-trunc', '--format', '{{.ID}}'], timeout).split()
if len(ids) != len(set(ids)) or any(re.fullmatch('[0-9a-f]{64}', cid) is None for cid in ids):
raise ValueError('Docker container identity enumeration invalid; HOLD stays open')
jit = False
for cid in ids:
value: object = json.loads(checked(['docker', 'inspect', cid], timeout))
if not isinstance(value, list) or len(value) != 1 or not isinstance(value[0], dict):
raise ValueError('Docker inspection invalid; HOLD stays open')
row = cast(dict[str, object], value[0])
config, state = row.get('Config'), row.get('State')
if row.get('Id') != cid or not isinstance(config, dict) or not isinstance(state, dict):
raise ValueError('Docker identity/config/state unproved; HOLD stays open')
name = row.get('Name')
if not isinstance(name, str) or not isinstance(row.get('Path'), str) or not isinstance(row.get('Args'), list):
raise ValueError('Docker name/command unproved; HOLD stays open')
status = state.get('Status')
if status not in ('created', 'running', 'paused', 'restarting', 'removing', 'exited', 'dead'):
raise ValueError('Docker lifecycle unproved; HOLD stays open')
# A stopped (exited/dead) container cannot act; old renamed production containers such as
# dsv41-serve-stopped-20261022 left DEP1-W-20261027T020407Z's restore refused with production down.
if status not in ('exited', 'dead') and re.search(
r'^(?:dsv41-serve(?:$|-)|dsv41-exp)|experiment|nbw|tv1|pp4|up3|dep1', name.lstrip('/')):
raise ValueError('experiment/serve container remains: ' + name + '; HOLD stays open')
# Inspect full argv AND config/env/mount identity. Random --rm names and
# import/build-lock pauses have no relationship to compute-app presence.
identity = json.dumps({key: row.get(key) for key in ('Path', 'Args', 'Config', 'Mounts')})
is_jit = any(key in identity for key in
('tf_experts', 'vllm_exl3_moe', '_load_engram_remote_ext',
'DSV41_ENGRAM_OVERLAY', 'offload/engram/spark', 'prebuild_jit.sh'))
jit |= is_jit and status not in ('exited', 'dead')
apps = checked(['nvidia-smi', '--query-compute-apps=pid,process_name', '--format=csv,noheader'], timeout)
return not jit and not apps.strip()
def wait(deadline: float, marker: str) -> None:
if not math.isfinite(deadline):
raise ValueError('finite quiet-proof deadline required')
while True:
left = deadline - time.time()
if left <= 0:
raise ValueError('JIT/compute quiet wait exhausted; HOLD stays open')
if snapshot(min(10, left)):
if time.time() >= deadline:
raise ValueError('quiet proof arrived after deadline; HOLD stays open')
print(marker, flush=True)
return
time.sleep(min(5, max(0, deadline - time.time())))
def command(marker: str, *, deadline: float) -> str:
return shlex.join(['python3', '-c', Path(__file__).read_text(), str(deadline), marker])
if __name__ == '__main__':
wait(float(sys.argv[1]), sys.argv[2])
|