#!/usr/bin/env python3 """Mos-1599: eight isolated T9 loops, 50 rounds each, M1-50 then fixed then whole file.""" import argparse from concurrent.futures import ThreadPoolExecutor, as_completed import json import os from pathlib import Path import re import signal import subprocess import sys import threading import time from stress import ROOT, TEST, ANCHOR, BASE_COMMAND, PATTERN, digest, sample_nice WORKERS, ROUNDS = 8, 50 PROOF = ('runtime.mjs:597 writes library-sync-held after sync returns; sync.mjs:203 finally clears running before that. ' 'T9 sees the line after the startup tick releases the mirror. TIMEOUT_MAX prevents another tick during the verb.') # A child sets its own absolute priority immediately before exec; subsequent drift is recorded only. CHILD_PRIORITY = ('import os,sys; os.setpriority(os.PRIO_PROCESS,0,19); ' 'print("initial-child-nice="+str(os.getpriority(os.PRIO_PROCESS,0)),flush=True); ' 'os.execvp(sys.argv[1],sys.argv[1:])') def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--clear-key', required=True) parser.add_argument('--output', type=Path, required=True) args = parser.parse_args() os.setpriority(os.PRIO_PROCESS, 0, 19) initial_nice = os.getpriority(os.PRIO_PROCESS, 0) path = ROOT / TEST fixed = path.read_bytes() assert fixed == subprocess.check_output(['git', 'show', f'HEAD:{TEST}'], cwd=ROOT) assert fixed.count(ANCHOR) == 1 report = {'clear': args.clear_key, 'source': subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=ROOT, text=True).strip(), 'workers': WORKERS, 'rounds': ROUNDS, 'fixedSHA256': digest(fixed), 'runnerInitialNice': initial_nice, 'loadStart': os.getloadavg(), 'runs': [], 'status': 'running'} output = args.output.resolve() output.mkdir(parents=True, exist_ok=False) lock, stopped, active = threading.Lock(), threading.Event(), {} def save(): (output / 'summary.json').write_text(json.dumps(report, indent=2) + '\n') def clean(): record = {'SHA256': digest(path.read_bytes()), 'gitDiffStat': subprocess.check_output(['git', 'diff', '--stat', 'HEAD'], cwd=ROOT, text=True), 'gitStatusPorcelain': subprocess.check_output(['git', 'status', '--porcelain'], cwd=ROOT, text=True)} assert record['SHA256'] == report['fixedSHA256'] and not record['gitDiffStat'] and not record['gitStatusPorcelain'], 'Dirty restore: stop' return record def stop_all(): stopped.set() with lock: groups = list(active) for pgid in groups: try: os.killpg(pgid, signal.SIGTERM) except ProcessLookupError: pass def attempt(phase, worker, number, whole=False): temporary = subprocess.check_output(['mktemp', '-d', '/mnt/storage/scratch/tmp/gXXX'], text=True).strip() assert len(temporary.encode()) == 29 and re.fullmatch(r'/mnt/storage/scratch/tmp/g[A-Za-z0-9]{3}', temporary) command = ['timeout', '-k', '10', '120', *BASE_COMMAND] if not whole: command += ['--test-name-pattern', PATTERN] command += [str(TEST)] command = [sys.executable, '-B', '-c', CHILD_PRIORITY, *command] env = {**os.environ, 'TMPDIR': temporary} env.pop('MOSAIC_LIVE_PROVIDER', None) env.pop('NODE_TEST_CONTEXT', None) stem = f'{phase}-w{worker:02}-{number:03}' log = output / (stem + '.tap') started, signals = time.monotonic(), [] with log.open('w') as stream: child = subprocess.Popen(command, cwd=ROOT, env=env, stdout=stream, stderr=subprocess.STDOUT, start_new_session=True) with lock: active[child.pid] = stem nice_samples, nice_stop = [], threading.Event() sampler = threading.Thread(target=sample_nice, args=(child.pid, nice_samples, nice_stop), daemon=True) sampler.start() try: if stopped.is_set(): os.killpg(child.pid, signal.SIGTERM) code = child.wait(timeout=135) except subprocess.TimeoutExpired: os.killpg(child.pid, signal.SIGKILL) signals.append('KILL') code = child.wait() finally: nice_stop.set() sampler.join() try: os.killpg(child.pid, signal.SIGTERM) signals.append('TERM') time.sleep(.2) try: os.killpg(child.pid, signal.SIGKILL) signals.append('KILL') except ProcessLookupError: pass except ProcessLookupError: pass with lock: active.pop(child.pid, None) text = log.read_text() counts = {name: int(value) for name, value in re.findall(r'^# (tests|pass|fail|cancelled|skipped) (\d+)$', text, re.M)} green = code == 0 and counts.get('pass') == (6 if whole else 1) and all(counts.get(k) == 0 for k in ['fail', 'cancelled', 'skipped']) race = (phase == 'M1-50' and code == 1 and counts.get('fail') == 1 and counts.get('cancelled') == 0 and 'ERR_ASSERTION' in text and re.search(r'^not ok \d+ - C2-03b T9 runtime:', text, re.M) and "+ code: 'library-sync-busy'" in text and "- code: 'library-fetch-no-grant'" in text) entry = {'phase': phase, 'worker': worker, 'round': number, 'TMPDIR': temporary, 'dataRootPattern': temporary + '/library-*/data', 'command': command, 'exit': code, 'counts': counts, 'green': green, 'race': bool(race), 'load': os.getloadavg(), 'seconds': round(time.monotonic() - started, 3), 'niceSamples': nice_samples, 'cleanupSignals': signals, 'TAP': log.name, 'TAPSHA256': digest(log.read_bytes())} (output / (stem + '.json')).write_text(json.dumps(entry, indent=2) + '\n') with lock: report['runs'].append({k: v for k, v in entry.items() if k not in ['command', 'niceSamples']}) print(json.dumps({k: entry[k] for k in ['phase', 'worker', 'round', 'exit', 'green', 'race', 'load', 'seconds']}), flush=True) if not green and not race: stop_all() raise RuntimeError(f'Red/uncertain run {log}; stop for Sage, no rerun') return entry def phase(label): barrier = threading.Barrier(WORKERS) def worker(number): try: barrier.wait() for round_number in range(1, ROUNDS + 1): if stopped.is_set(): return attempt(label, number, round_number) except BaseException: stop_all() raise pool = ThreadPoolExecutor(max_workers=WORKERS) try: futures = [pool.submit(worker, n) for n in range(1, WORKERS + 1)] for future in as_completed(futures): future.result() except BaseException: stop_all() raise finally: pool.shutdown(wait=True, cancel_futures=True) rows = [r for r in report['runs'] if r['phase'] == label] assert len(rows) == WORKERS * ROUNDS and len({r['TMPDIR'] for r in rows}) == len(rows) report[label] = {'attempts': len(rows), 'passes': sum(r['green'] for r in rows), 'raceFailures': sum(r['race'] for r in rows), 'loadEnd': os.getloadavg()} save() try: report['preflight'] = clean() path.write_bytes(fixed.replace(ANCHOR, b' runtime = await start(50);', 1)) try: phase('M1-50') finally: path.write_bytes(fixed) report['mutantRestore'] = clean() save() report['mos1599Outcome'] = 'A' if report['M1-50']['raceFailures'] else 'B' if report['mos1599Outcome'] == 'B': report['constructionProof'] = PROOF phase('fixed') attempt('whole-file', 0, 1, whole=True) report['status'] = 'green' except BaseException as error: report['status'], report['stopReason'] = 'stopped', str(error) stop_all() raise finally: path.write_bytes(fixed) report['finalRestore'] = clean() report['loadEnd'] = os.getloadavg() save() if __name__ == '__main__': main()