| """Coordinator-only pair subprocess sequence. Default is a no-action preview.""" |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import os |
| import subprocess |
| import sys |
| import time |
| from pathlib import Path |
|
|
| from offload.launch.windows import pf1_window as pf |
| from offload.launch.windows import up2_checks as ck |
| from offload.prefill_maxlen.evidence import Json |
|
|
| from .pair_contract import SCHEMA, checked_output, guard, validate |
|
|
| SOURCES = ('offload/dspark/probabilistic/pp4/pair_qualify.py', |
| 'offload/dspark/probabilistic/pp4/pair_probe.py', 'offload/dspark/probabilistic/pp4/pair_diagnostics.py', |
| 'offload/dspark/probabilistic/pp4/pair_contract.py', |
| 'offload/dspark/probabilistic/pp4/full_worker.py', 'offload/dspark/probabilistic/pp4/n4-greedy-reference.json', |
| 'offload/dspark/worker/sampled_fused.py', 'offload/dspark/worker/tp2.py', 'offload/dspark/worker/server.py') |
|
|
|
|
| def execute(out: Path, commit: str, run: str, deadline: float) -> dict[str, Json]: |
| checked_output(out); guard() |
| ck.require(sys.executable == '/home/operator/miniconda3/bin/python3', 'qualified interpreter required') |
| ck.require(subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip() == commit, |
| 'pair source commit differs') |
| out.mkdir(parents=True, exist_ok=False) |
| result: dict[str, Json] = {'schema': SCHEMA, 'source_commit': commit, 'run': run, |
| 'complete': False, 'classification': 'MEASURED physical pair', 'gpus': pf.WORKER_GPUS, |
| 'source_sha256': {name: pf.digest(pf.ROOT/name) for name in SOURCES}} |
| locks = tuple(int(os.environ[key]) for key in ('PP4_PAIR_LOCK_FD0', 'PP4_PAIR_LOCK_FD1')) |
| started = time.monotonic() |
| for mode, extra, limit in ( |
| ('correctness', ['-m', 'offload.dspark.probabilistic.pp4.pair_probe', '--mode', 'correctness'], 240), |
| ('timeout0', ['-m', 'offload.dspark.probabilistic.pp4.pair_probe', '--mode', 'timeout0'], 60), |
| ('timeout1', ['-m', 'offload.dspark.probabilistic.pp4.pair_probe', '--mode', 'timeout1'], 60), |
| ('whole_worker', ['-m', 'offload.dspark.probabilistic.pp4.full_worker', '--physical-pair', '--qualification-only'], 600)): |
| remaining = min(limit, deadline-time.time()) |
| ck.require(remaining > 5, 'pair qualification budget exhausted; restore without measurement') |
| path = out/(mode+'.json') |
| with (out/(mode+'.log')).open('x') as log: |
| |
| |
| |
| subprocess.run([sys.executable, *extra, '--out', str(path)], stdout=log, stderr=subprocess.STDOUT, |
| timeout=remaining, check=True, pass_fds=locks) |
| child = ck.read(path) |
| if mode == 'correctness': |
| result['cases'] = child['cases'] |
| else: |
| if mode.startswith('timeout'): |
| child['process_exited'] = True |
| result[mode] = child |
| (out/'receipt.json').write_text(json.dumps(result, allow_nan=False)+'\n') |
| result['seconds'] = time.monotonic()-started |
| result['complete'] = True |
| validate(result, commit, run) |
| (out/'receipt.json').write_text(json.dumps(result, allow_nan=False)+'\n') |
| return result |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument('--execute', action='store_true') |
| parser.add_argument('--out', type=Path) |
| parser.add_argument('--commit', default='') |
| parser.add_argument('--run', default='') |
| parser.add_argument('--deadline', type=float, default=0) |
| args = parser.parse_args() |
| if not args.execute: |
| print('DRY RUN: fenced HOLD only; physical pair B1-B4/depth4-5, changed-input graphs, mixed T0, ' |
| 'fresh solo q/ID receipt, isolated peer-timeout/sticky-error processes, whole-K2 archived N4 identity; restore.') |
| return |
| ck.require(args.out is not None, 'pair output required') |
| assert isinstance(args.out, Path) |
| execute(args.out, args.commit, args.run, args.deadline) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
