diff --git a/docs/research/frontier741-concurrent-replay.md b/docs/research/frontier741-concurrent-replay.md new file mode 100644 index 0000000000..a8b1e10a4d --- /dev/null +++ b/docs/research/frontier741-concurrent-replay.md @@ -0,0 +1,140 @@ +# Concurrent bit/complex replay of the #589 + #741 ancestry + +Status: **tooling improvement**. It changes no construction, bound, certificate or +selected record, and claims no new kappa. + +## What changed + +`scripts/verify_frontier741.py` (the cold ancestry replay that `verify_frontier799.py` +runs, and hence the selected `make frontier1042-verify` chain) runs `bit589` to +completion and then `complex741`. Neither reads anything the +other writes: `bit589` writes only under `/bit`, `complex741` only under +`/complex`, and the binding step needs both. The new +`scripts/verify_frontier741_concurrent.py` runs the same two commands at the same +time, then the same binding script. + +`verify_frontier741.py` is untouched because its sha256 is pinned in +`certificates/selected-result.json` and the checkpoint certificates. The new driver +is a sibling that writes the same layout (`bit/`, `complex/`, `bit589.log`, +`complex741.log`, `STEPS.json`, `RESULT.json`) plus a `CONCURRENCY.json` with wall and +summed job times. If either supplier fails, the other is stopped (process group +SIGTERM, then SIGKILL after 20 s), the failing log is named, and nothing is bound. +`--complex-jobs N` (N>1) swaps in the existing dependency-scheduled complex replay +(`verify_complex741_parallel.py`, #853); the default of 1 runs `verify.py` unchanged so +that this change is measured on its own. + +## Reproduce + +```sh +python3 -B scripts/verify_frontier741_concurrent.py --work /absolute/fresh/dir +python3 -B scripts/verify_replay_coverage.py --frontier 741 --work /absolute/fresh/dir +python3 -m unittest tests.test_frontier741_concurrent +``` + +Expected: the first prints `PASS selected finite composition; inherited hypotheses +remain`, and `RESULT.json` has status `PASS_REBUILT_PR589_PR741_COMPOSITION`, kappa +`1661202643272239/1000000000000000000`. The second prints +`PASS_COMPLETE_PR741_REPLAY_COVERAGE`. The tests (8) take about 3 s. + +Developed and measured at base commit `2921874911956718fcaaaf9396fe8267b3cfa9f0`. +The branch is rebased onto `ba82555c` (main when submitted); `verify_frontier741.py`, +its pinned hash, the bindings script and both supplier packages are unchanged between +the two, and the tests were re-run on the rebased tree. Benchmark environment (not the +CI image): WSL2 Ubuntu 24.04, 28 logical CPUs, 7 GB RAM; Python 3.12.15, numpy 2.5.4, +scipy 1.18.1, sympy 1.14.0; the C++17 checkers built with `zig c++` (ziglang 0.17.0, +clang 22.1.8) against Boost 1.86.0 headers. No system compiler was available. + +## Measurement + +Same checkout, same fresh output path each time, one run at a time on an otherwise idle +machine. The first row used different checkout/output paths and is timing only. + +| Driver | Wall seconds | bit589 | complex741 | +| --- | --- | --- | --- | +| original, sequential | 608 | 138 | 469 | +| original, sequential | 599 | – | – | +| original, sequential | 562 | – | – | +| concurrent | 495 | 141 | 495 | +| concurrent | 479 | – | – | +| concurrent, `--complex-jobs 3` | 293 | 150 | 293 | + +Sequential mean 590 s, concurrent mean 487 s: about 17% less wall time, 100 to 120 s +saved. The concurrent run is bounded below by `complex741` (critical path 469 to 495 s), +so the bit supplier is hidden but nothing else is. With `--complex-jobs 3` the complex +chain shortens to the dependency chain of #853 and the bit supplier is again fully +hidden: 293 s, about 50% less than sequential (one sample). That is 3 sequential, 2 +concurrent and 1 combined samples on one machine; treat the ratios as indicative, not +as precise speedups. #853 alone would leave `bit589` (about 140 s) in series with the +complex replay; that exact configuration was not run here. + +## Coverage is unchanged + +* `tests/test_frontier741_concurrent.py` parses `verify_frontier741.py` and requires the + concurrent driver's `bit589`, `complex741` and `bindings` commands, stage order and + environment scrub (`PYTHONOPTIMIZE`, `WITNESS_DISCOVER`, `WITNESS_RECORD`, `SW4_PIN`) + to equal the original's. If the original changes, the test fails. +* Output trees of the original and concurrent drivers (same path) were compared file by + file: 385 common files, 101 differ. A second original run differs from the first in + exactly the same 101 files, so the concurrent schedule adds no new difference. After + removing `seconds`/`elapsed` fields and SHA-256 values (digests of files that embed + timings), every differing JSON file is identical in both comparisons. The remaining + differing files are logs. +* The concurrent output passes `verify_replay_coverage.py --frontier 741` and the + original bindings check. +* Negative controls on a copy of the concurrent output: a changed `bit_lower_decimal` + fails the bindings check (`Rebuilt bit differs`); one extra byte appended to + `complex/global-replay/ASSEMBLY.json` fails coverage (`Required ancestry artifact hash + differs`); a deleted `bit/xcheck.json` fails. +* Mutation controls on the tests: replacing the bit verifier, adding `-O` to the complex + command, removing `SW4_PIN` from the scrub, not stopping the sibling after a failure, + and waiting for each job right after launching it are each caught. (The `SW4_PIN` + mutation first survived; the test now sets the variable before checking the scrub.) + +## Scope and open items + +* Finite replay tooling only. It is not a formal proof, and every inherited assumption + listed in `certificates/selected-result.json` remains. +* The output is accepted as an ancestor by the coverage and bindings checks, so it can + be passed as `--ancestor` to `verify_frontier799.py`. That was run once (WSL, default + concurrent mode, base `2921874`; the 799 scripts are unchanged on later main): + `verify_frontier799.py --work W --ancestor ` passed in 203 s with + `PASS_REBUILT_PR589_PR799_COMPOSITION`, kappa `1661634653669567/10^18`, and + `verify_replay_coverage.py --frontier 799 --ancestor ` printed + `PASS_COMPLETE_PR799_REPLAY_COVERAGE`. It records the ancestor as reused + (`reused_ancestry: true`), as for any externally produced ancestor. Nothing here + changes what `make frontier1042-verify` (or `frontier942-verify`) runs. +* End to end, cold `frontier1042` on the rebased PR tree (WSL, same checkout, one run + each, one after the other): the original `verify_frontier1042.py` took **1560 s** + (`baseline799-cold` 788 s, its own stages 772 s). The new path (concurrent driver 467 s, + then `verify_frontier799.py --ancestor` 189 s, then + `verify_frontier1042.py --baseline799` 756 s, with the ancestor moved to + `/ancestor` as that script expects) took **1412 s**, 148 s or 9.5% less. + Both ended in `PASS_COMPLETE_PR1042_COMPOSITION` with kappa + `0.001672188119205421`; the new path records `reused_completed_baseline: true`. About + 132 s of the saving is the 741 part (788 s to 656 s); the rest is run-to-run noise in + 1042's own stages (chain 479 s vs 464 s). Single runs, so about +/-2% should be + assumed. The #942 replay was not run with it. +* Overlap check at submission: #853 (merged) schedules stages inside `complex741` only and + does not overlap the bit supplier; no open or merged PR found by search touches + `verify_frontier741.py` or overlaps `bit589` with `complex741`. The two are + complementary: with `--complex-jobs 3` the complex chain hides `bit589` entirely (table + above). +* `--complex-jobs 3` was run cold once (293 s). Its output has the same 385 files as the + sequential tree and passes `verify_replay_coverage.py --frontier 741`. Against the + sequential tree it differs in the 101 noise files plus `complex741.log`, whose 45 + lines differ only in the order of `RUN`/`PASS` progress messages; every differing JSON + file is identical after masking `seconds` and SHA-256 values. +* The `u8-native` stage (about 254 s) and the cold #799 `native` stage (about 393 s) + dominate the remaining serial time and were not examined. +* Peak memory with both suppliers running was not measured; the 7 GB benchmark machine + completed it. +* Linux CI was not run; the benchmark used the environment above. + +## Credit and disclosure + +Builds on the pinned `verify_frontier741.py` driver, `verify_frontier741_bindings.py` +(Douglas Colkitt, with OpenAI Codex assistance), the +dependency-scheduled complex replay of #853 (Aaron / navinor), and the `--jobs` pattern +of `verify_frontier596.py`. Track: verification engineering, assignment seed 90, base +commit above. Substantial AI assistance (Claude, in Cursor) in analysis, code, tests +and this note. Contributor: James Reed (@rcfc-jreed). diff --git a/scripts/verify_frontier741_concurrent.py b/scripts/verify_frontier741_concurrent.py new file mode 100644 index 0000000000..d162de9325 --- /dev/null +++ b/scripts/verify_frontier741_concurrent.py @@ -0,0 +1,176 @@ +#!/usr/bin/env python3 +"""Cold replay of PR589 + PR741 with the bit and complex suppliers overlapped. + +Sibling of scripts/verify_frontier741.py, which stays unchanged because its hash +is pinned by the selected record and checkpoint certificates. It executes the same +two supplier commands, in the same environment, into the same output layout +(bit/, complex/, bit589.log, complex741.log, STEPS.json, RESULT.json), then runs +the same scripts/verify_frontier741_bindings.py. The only change is that the two +supplier commands run at the same time: neither reads anything the other writes +(bit589 writes only under /bit, complex741 only under /complex). +If either supplier fails, the other is stopped and nothing is bound. + +The output is a valid --ancestor for scripts/verify_frontier799.py (which then +records it as reused ancestry, as for any externally produced ancestor). +Optional --complex-jobs N (>1) additionally uses the dependency-scheduled +complex replay of scripts/verify_complex741_parallel.py. +""" +import argparse +import hashlib +import json +import os +from pathlib import Path +import signal +import subprocess +import sys +import threading +import time + +ROOT = Path(__file__).resolve().parents[1] + + +def supplier_commands(work, complex_jobs=1): + """The commands scripts/verify_frontier741.py runs, as (name, argv) in its order.""" + if complex_jobs > 1: + complex_command = [sys.executable, '-B', ROOT / 'scripts/verify_complex741_parallel.py', + '--work', work / 'complex', '--jobs', complex_jobs] + else: + complex_command = [sys.executable, '-B', ROOT / 'research/complex-sw4-frame8/verify.py', + '--work', work / 'complex'] + return [('bit589', ['bash', ROOT / 'research/e8-bit-p111-tuned/tune/verify.sh', work / 'bit']), + ('complex741', complex_command)] + + +def binding_command(work): + return [sys.executable, '-B', ROOT / 'scripts/verify_frontier741_bindings.py', + '--bit', work / 'bit', '--complex', work / 'complex', '--out', work / 'RESULT.json'] + + +def supplier_env(): + env = dict(os.environ, PYTHON=sys.executable, PYTHONDONTWRITEBYTECODE='1', + SW4_FRAME8_PYTHON=sys.executable) + for name in ('PYTHONOPTIMIZE', 'WITNESS_DISCOVER', 'WITNESS_RECORD', 'SW4_PIN'): + env.pop(name, None) + return env + + +def _stop(process): + if process.poll() is not None: + return + try: + if hasattr(os, 'killpg'): + os.killpg(process.pid, signal.SIGTERM) + else: + process.terminate() + except ProcessLookupError: + return + try: + process.wait(timeout=20) + except subprocess.TimeoutExpired: + if hasattr(os, 'killpg'): + os.killpg(process.pid, signal.SIGKILL) + else: + process.kill() + process.wait() + + +def run_concurrent(jobs, work, env, cwd=ROOT): + """Run [(name, argv)] at once, one log per job under work. Returns step records + in the given order. On the first non-zero exit the others are stopped and + RuntimeError names the failing job; stopped jobs record their negative exit code. + """ + state = {} + first_failure = [] + lock = threading.Lock() + start = time.monotonic() + for name, command in jobs: + log_path = work / (name + '.log') + stream = log_path.open('w') + print('RUN', name, flush=True) + process = subprocess.Popen(list(map(str, command)), cwd=cwd, env=env, stdout=stream, + stderr=subprocess.STDOUT, + start_new_session=hasattr(os, 'killpg')) + state[name] = dict(process=process, stream=stream, log=log_path, started=time.monotonic()) + + def watch(name): + item = state[name] + code = item['process'].wait() + item['seconds'] = time.monotonic() - item['started'] + with lock: + if code != 0 and not first_failure: + first_failure.append(name) + for other, other_item in state.items(): + if other != name: + _stop(other_item['process']) + if code == 0: + print('PASS', name, flush=True) + + threads = [threading.Thread(target=watch, args=(name,)) for name, _ in jobs] + for thread in threads: + thread.start() + try: + for thread in threads: + thread.join() + except BaseException: + for item in state.values(): + _stop(item['process']) + raise + finally: + for item in state.values(): + item['stream'].close() + records = [dict(name=name, exit_code=state[name]['process'].returncode, + seconds=state[name]['seconds'], + log_sha256=hashlib.sha256(state[name]['log'].read_bytes()).hexdigest()) + for name, _ in jobs] + (work / 'CONCURRENCY.json').write_text(json.dumps(dict( + jobs=[name for name, _ in jobs], wall_seconds=time.monotonic() - start, + sequential_seconds=sum(r['seconds'] for r in records)), indent=2) + '\n') + if first_failure: + raise RuntimeError(f"{first_failure[0]} failed; inspect {state[first_failure[0]]['log']}", + records) + return records + + +def replay(work, complex_jobs=1): + steps = [] + def dump(): + (work / 'STEPS.json').write_text(json.dumps(steps, indent=2) + '\n') + env = supplier_env() + try: + steps.extend(run_concurrent(supplier_commands(work, complex_jobs), work, env)) + except RuntimeError as error: + steps.extend(error.args[1]) + dump() + raise SystemExit(error.args[0]) + dump() + name, command = 'bindings', binding_command(work) + start = time.monotonic() + log = work / (name + '.log') + print('RUN', name, flush=True) + with log.open('w') as stream: + result = subprocess.run(list(map(str, command)), cwd=ROOT, env=env, + stdout=stream, stderr=subprocess.STDOUT) + steps.append(dict(name=name, exit_code=result.returncode, seconds=time.monotonic() - start, + log_sha256=hashlib.sha256(log.read_bytes()).hexdigest())) + dump() + if result.returncode: + raise SystemExit(f'{name} failed; inspect {log}') + print('PASS', name, flush=True) + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument('--work', type=Path, required=True) + p.add_argument('--complex-jobs', type=int, default=1, + help='>1 uses the dependency-scheduled complex replay with this many workers') + a = p.parse_args() + work = a.work.resolve() + if sys.flags.optimize or work.exists() or work.is_relative_to(ROOT) or a.complex_jobs < 1: + raise SystemExit('Use enabled assertions, positive jobs, and a fresh directory outside the repository') + work.mkdir(parents=True) + replay(work, a.complex_jobs) + print('PASS selected finite composition; inherited hypotheses remain') + + +if __name__ == '__main__': + main() diff --git a/tests/test_frontier741_concurrent.py b/tests/test_frontier741_concurrent.py new file mode 100644 index 0000000000..0396f57fec --- /dev/null +++ b/tests/test_frontier741_concurrent.py @@ -0,0 +1,126 @@ +"""Concurrent supplier replay for the PR589 + PR741 ancestry. + +The driver must run exactly the commands of scripts/verify_frontier741.py; only +their schedule differs. These tests tie the command lines and environment to that +script's source, and check overlap and fail-fast behaviour of the scheduler. +""" +import ast +import json +import os +from pathlib import Path +import sys +import tempfile +import time +import unittest +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / 'scripts')) +import verify_frontier741_concurrent as conc + +ORIGINAL = ROOT / 'scripts/verify_frontier741.py' +PY = sys.executable + + +def original_runs(): + """name -> command expression of every run(...) call in verify_frontier741.py.""" + main = next(n for n in ast.parse(ORIGINAL.read_text()).body + if isinstance(n, ast.FunctionDef) and n.name == 'main') + return {call.args[0].value: call.args[1] for call in ast.walk(main) + if isinstance(call, ast.Call) and isinstance(call.func, ast.Name) and call.func.id == 'run' + and call.args and isinstance(call.args[0], ast.Constant)} + + +def original_command(expr, work): + return list(map(str, eval(compile(ast.Expression(expr), 'verify_frontier741.py', 'eval'), + {'ROOT': ROOT, 'work': work, 'sys': sys}))) + + +class SameCommandsAsOriginal(unittest.TestCase): + def setUp(self): + self.work = Path('/replay/work') + self.runs = original_runs() + + def test_original_stage_names_and_order(self): + self.assertEqual(list(self.runs), ['bit589', 'complex741', 'bindings']) + + def test_supplier_commands_equal_original(self): + for name, command in conc.supplier_commands(self.work): + self.assertEqual(list(map(str, command)), original_command(self.runs[name], self.work), name) + + def test_binding_command_equals_original(self): + self.assertEqual(list(map(str, conc.binding_command(self.work))), + original_command(self.runs['bindings'], self.work)) + + def test_environment_matches_original(self): + main = next(n for n in ast.parse(ORIGINAL.read_text()).body + if isinstance(n, ast.FunctionDef) and n.name == 'main') + popped = next(ast.literal_eval(n.iter) for n in ast.walk(main) + if isinstance(n, ast.For) and isinstance(n.iter, ast.Tuple)) + self.assertEqual(set(popped), {'PYTHONOPTIMIZE', 'WITNESS_DISCOVER', 'WITNESS_RECORD', 'SW4_PIN'}) + with mock.patch.dict(os.environ, {name: '1' for name in popped}): + env = conc.supplier_env() + self.assertFalse(set(popped) & set(env)) + self.assertEqual(env['PYTHON'], sys.executable) + self.assertEqual(env['PYTHONDONTWRITEBYTECODE'], '1') + self.assertEqual(env['SW4_FRAME8_PYTHON'], sys.executable) + assigned = next(n.value for n in ast.walk(main) if isinstance(n, ast.Assign) and + isinstance(n.targets[0], ast.Name) and n.targets[0].id == 'env') + self.assertEqual({k.arg for k in assigned.keywords}, + {'PYTHON', 'PYTHONDONTWRITEBYTECODE', 'SW4_FRAME8_PYTHON'}) + + def test_complex_jobs_uses_parallel_scheduler_only_for_complex(self): + commands = dict(conc.supplier_commands(self.work, complex_jobs=3)) + self.assertEqual(list(map(str, commands['bit589'])), + original_command(self.runs['bit589'], self.work)) + self.assertEqual(Path(commands['complex741'][2]).name, 'verify_complex741_parallel.py') + self.assertEqual(list(map(str, commands['complex741']))[-2:], ['--jobs', '3']) + + +class Scheduler(unittest.TestCase): + def run_jobs(self, jobs): + with tempfile.TemporaryDirectory() as directory: + work = Path(directory) + start = time.monotonic() + try: + records = conc.run_concurrent(jobs, work, conc.supplier_env()) + error = None + except RuntimeError as exc: + records, error = exc.args[1], exc + return records, error, time.monotonic() - start, json.loads((work / 'CONCURRENCY.json').read_text()), \ + {n: (work / (n + '.log')).read_text() for n, _ in jobs} + + def test_independent_jobs_overlap_and_keep_order(self): + sleep = [PY, '-c', 'import time;time.sleep(1.5);print("done")'] + records, error, wall, summary, logs = self.run_jobs([('a', sleep), ('b', sleep)]) + self.assertIsNone(error) + self.assertEqual([r['name'] for r in records], ['a', 'b']) + self.assertEqual([r['exit_code'] for r in records], [0, 0]) + self.assertLess(wall, 2.8) + self.assertLess(summary['wall_seconds'], 0.75 * summary['sequential_seconds']) + self.assertEqual(logs['a'], 'done\n') + self.assertEqual(len(records[0]['log_sha256']), 64) + + def test_failure_stops_the_other_job_and_names_the_failure(self): + fail = [PY, '-c', 'import sys;print("bad");sys.exit(3)'] + hang = [PY, '-c', 'import time;time.sleep(120)'] + records, error, wall, _, logs = self.run_jobs([('hang', hang), ('fail', fail)]) + self.assertIsInstance(error, RuntimeError) + self.assertIn('fail failed', error.args[0]) + self.assertLess(wall, 30) + by_name = {r['name']: r for r in records} + self.assertEqual(by_name['fail']['exit_code'], 3) + self.assertNotEqual(by_name['hang']['exit_code'], 0) + self.assertEqual(logs['fail'], 'bad\n') + + def test_success_of_one_job_does_not_stop_the_other(self): + quick = [PY, '-c', 'print(1)'] + slow = [PY, '-c', 'import time;time.sleep(1);print(2)'] + records, error, _, _, logs = self.run_jobs([('quick', quick), ('slow', slow)]) + self.assertIsNone(error) + self.assertEqual([r['exit_code'] for r in records], [0, 0]) + self.assertEqual(logs['slow'], '2\n') + + +if __name__ == '__main__': + unittest.main()