Digital Drosophila Archive
81 lines
3.5 KiB
Python
81 lines
3.5 KiB
Python
"""New pipeline experiment process receipts, with exclusive output and no retries."""
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import time
|
|
|
|
ROOT = Path('/home/qishizhe/phase4ar-pipeline-acceleration-20261001')
|
|
|
|
def output_bytes():
|
|
total = 0
|
|
for f in ROOT.rglob('*'):
|
|
try:
|
|
if f.is_file() and not f.is_symlink(): total += f.stat().st_size
|
|
except FileNotFoundError:
|
|
pass # Concurrent compiler temporary file removal, not a payload.
|
|
return total
|
|
|
|
def stop_group(proc):
|
|
try: os.killpg(proc.pid, signal.SIGTERM)
|
|
except ProcessLookupError: pass
|
|
try: proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired: pass
|
|
try: os.killpg(proc.pid, signal.SIGKILL)
|
|
except ProcessLookupError: pass
|
|
return proc.wait()
|
|
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument('--label', required=True)
|
|
p.add_argument('--timeout', type=int, default=900)
|
|
p.add_argument('command', nargs=argparse.REMAINDER)
|
|
a = p.parse_args()
|
|
assert re.fullmatch(r'[a-z0-9-]+', a.label) and a.command and 0 < a.timeout <= 900
|
|
assert ROOT.is_dir() and not ROOT.is_symlink()
|
|
assert shutil.disk_usage(ROOT).free >= 100 * (1 << 30)
|
|
out = ROOT / 'jobs'
|
|
out.mkdir(exist_ok=True)
|
|
start = time.perf_counter()
|
|
record = dict(label=a.label, command=a.command, scope='independent engineering replay only',
|
|
started_utc=datetime.now(timezone.utc).isoformat(), scientific_AR_resumed=False)
|
|
with (out / (a.label + '.start.json')).open('x') as f: json.dump(record, f, indent=2)
|
|
with (out / (a.label + '.stdout.log')).open('x') as stdout, (out / (a.label + '.stderr.log')).open('x') as stderr:
|
|
proc = None
|
|
try:
|
|
# All children belong to a dedicated group created by THIS invocation.
|
|
# On stop, terminate this group only, never unrelated services/science.
|
|
proc = subprocess.Popen(a.command, stdout=stdout, stderr=stderr, start_new_session=True,
|
|
env={**os.environ, 'PYTHONDONTWRITEBYTECODE':'1'})
|
|
record.update(pid=proc.pid, process_group=proc.pid)
|
|
stop_reason = None
|
|
while True:
|
|
try:
|
|
proc.wait(timeout=5)
|
|
break
|
|
except subprocess.TimeoutExpired:
|
|
if time.perf_counter() - start > a.timeout: stop_reason = 'wall_timeout'
|
|
elif shutil.disk_usage(ROOT).free < 100 * (1 << 30): stop_reason = 'minimum_disk_reserve'
|
|
elif output_bytes() > 2 * (1 << 30):
|
|
stop_reason = 'output_cap'
|
|
if stop_reason:
|
|
break
|
|
if not stop_reason and output_bytes() > 2 * (1 << 30): stop_reason = 'output_cap'
|
|
if stop_reason or proc.returncode != 0: stop_group(proc)
|
|
record.update(exit_code=proc.returncode, observed_exit=True,
|
|
timed_out=stop_reason == 'wall_timeout', stop_reason=stop_reason)
|
|
except OSError as error:
|
|
rc = None if proc is None else stop_group(proc)
|
|
record.update(exit_code=rc, observed_exit=proc is not None, timed_out=False, launch_error=str(error))
|
|
except BaseException:
|
|
if proc is not None: stop_group(proc)
|
|
raise
|
|
record.update(wall_seconds=time.perf_counter()-start, finished_utc=datetime.now(timezone.utc).isoformat())
|
|
with (out / (a.label + '.exit.json')).open('x') as f: json.dump(record, f, indent=2)
|
|
print(json.dumps(record), flush=True)
|
|
raise SystemExit(record['exit_code'] if record['observed_exit'] and not record.get('stop_reason') else 124)
|