219 lines
11 KiB
Python
219 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Build a clean source revision and run a serial, fully recorded numerical campaign."""
|
|
import argparse
|
|
import csv
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import platform
|
|
import resource
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
|
|
|
|
def digest(path):
|
|
result = hashlib.sha256()
|
|
with path.open('rb') as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b''):
|
|
result.update(chunk)
|
|
return result.hexdigest()
|
|
|
|
|
|
def utc():
|
|
return datetime.datetime.now(datetime.timezone.utc).isoformat()
|
|
|
|
|
|
def json_safe(value):
|
|
if isinstance(value, float) and not math.isfinite(value):
|
|
return None
|
|
if isinstance(value, dict):
|
|
return {key: json_safe(item) for key, item in value.items()}
|
|
if isinstance(value, list):
|
|
return [json_safe(item) for item in value]
|
|
return value
|
|
|
|
|
|
def write_json(path, value):
|
|
path.write_text(json.dumps(json_safe(value), indent=2, allow_nan=False) + '\n')
|
|
|
|
|
|
def read_rows(path):
|
|
if not path.exists():
|
|
return []
|
|
with path.open() as stream:
|
|
return list(csv.DictReader(stream))
|
|
|
|
|
|
def summarize(folder):
|
|
result = {'newton': read_rows(folder / 'newton.csv'),
|
|
'frozen_linear': read_rows(folder / 'frozen_linear.csv'),
|
|
'preconditioner_benchmark': read_rows(folder / 'preconditioner_summary.csv'),
|
|
'trials': read_rows(folder / 'trials.csv')}
|
|
meta = folder / 'metadata.txt'
|
|
if meta.exists():
|
|
result['metadata'] = dict(line.split('=', 1) for line in meta.read_text().splitlines() if '=' in line)
|
|
for stage in ['initial', 'final']:
|
|
rows = read_rows(folder / f'{stage}_physical.csv')
|
|
result[stage + '_physical'] = {
|
|
q: {row['metric']: float(row['value']) for row in rows if row['quadrature'] == q}
|
|
for q in sorted({row['quadrature'] for row in rows})}
|
|
return result
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--source', required=True, type=Path)
|
|
parser.add_argument('--build', required=True, type=Path)
|
|
parser.add_argument('--mesh', required=True, type=Path)
|
|
parser.add_argument('--cases', required=True, type=Path)
|
|
parser.add_argument('--output', required=True, type=Path)
|
|
parser.add_argument('--cmake', default='cmake')
|
|
parser.add_argument('--jobs', type=int, default=2)
|
|
parser.add_argument('--timeout', type=float, default=1800)
|
|
args = parser.parse_args()
|
|
source, build, mesh, output = [p.resolve() for p in [args.source, args.build, args.mesh, args.output]]
|
|
if output.exists():
|
|
parser.error('Output directory already exists; use a new directory for every run.')
|
|
def git(*words):
|
|
return subprocess.check_output(['git', *words], cwd=source, text=True).strip()
|
|
if git('status', '--porcelain', '--untracked-files=no'):
|
|
parser.error('Tracked source must be committed before running a campaign.')
|
|
cache = dict(line.split('=', 1) for line in (build / 'CMakeCache.txt').read_text().splitlines()
|
|
if '=' in line and not line.startswith(('#', '//')))
|
|
if Path(cache['CMAKE_HOME_DIRECTORY:INTERNAL']).resolve() != source:
|
|
parser.error('Build directory belongs to a different source checkout.')
|
|
cases = json.loads(args.cases.read_text())
|
|
seen = set()
|
|
for case in cases:
|
|
name = case['id']
|
|
if not name or any(c not in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-' for c in name):
|
|
parser.error('Case IDs may contain only letters, digits, underscore and hyphen.')
|
|
if name in seen:
|
|
parser.error('Duplicate case ID.')
|
|
seen.add(name)
|
|
if not isinstance(case['arguments'], list) or not all(isinstance(a, str) for a in case['arguments']):
|
|
parser.error('Case arguments must be a list of strings.')
|
|
if '--mesh' in case['arguments'] or '--output' in case['arguments']:
|
|
parser.error('Mesh and output are controlled by the campaign runner.')
|
|
output.mkdir(parents=True)
|
|
build_command = [args.cmake, '--build', str(build), '--target', 'plateau_diagnostic', '-j', str(args.jobs)]
|
|
provenance = {'started_utc': utc(), 'source': str(source), 'commit': git('rev-parse', 'HEAD'),
|
|
'branch': git('branch', '--show-current'), 'platform': platform.platform(),
|
|
'build_command': build_command, 'build_settings': cache,
|
|
'mesh': str(mesh), 'mesh_sha256': digest(mesh), 'cases': cases,
|
|
'runtime_environment': {'OMPI_MCA_btl': 'self,sm'}}
|
|
write_json(output / 'provenance.json', provenance)
|
|
shutil.copy2(args.cases, output / 'cases.json')
|
|
print('BUILD', provenance['commit'], flush=True)
|
|
with (output / 'build.log').open('w') as stream:
|
|
subprocess.run(build_command, cwd=source, stdout=stream, stderr=subprocess.STDOUT, check=True)
|
|
if git('rev-parse', 'HEAD') != provenance['commit'] or git('status', '--porcelain', '--untracked-files=no'):
|
|
raise RuntimeError('Source changed during build.')
|
|
# A private immutable executable prevents later branch/build operations from changing queued cases.
|
|
executable = output / 'plateau_diagnostic'
|
|
shutil.copy2(build / 'plateau_diagnostic', executable)
|
|
provenance['binary_sha256'] = digest(executable)
|
|
provenance['build_finished_utc'] = utc()
|
|
write_json(output / 'provenance.json', provenance)
|
|
summaries = []
|
|
for case in cases:
|
|
name = case['id']
|
|
folder = output / name
|
|
command = [str(executable), *case['arguments'], '--mesh', str(mesh), '--output', str(folder)]
|
|
state_files = {}
|
|
for i, word in enumerate(case['arguments']):
|
|
if word in ('--probe-state', '--probe-direction', '--replay-candidate'):
|
|
path = Path(case['arguments'][i + 1])
|
|
if not path.is_absolute():
|
|
path = source / path
|
|
path = path.resolve()
|
|
state_files[str(path)] = digest(path)
|
|
elif word in ('--density-snapshot', '--restart-snapshot'):
|
|
path = Path(case['arguments'][i + 1])
|
|
path = (path if path.is_absolute() else source / path).resolve()
|
|
if word == '--restart-snapshot':
|
|
snapshot_file = path / 'accepted_state.txt' if path.is_dir() else path
|
|
prior_folder = snapshot_file.parent
|
|
if not (prior_folder / 'metadata.txt').is_file():
|
|
prior_folder = prior_folder.parent
|
|
prior_manifest = prior_folder.parent / 'provenance.json'
|
|
prior_run = prior_folder.parent / (prior_folder.name + '_run.json')
|
|
prior = json.loads(prior_manifest.read_text())
|
|
saved_run = json.loads(prior_run.read_text())
|
|
if prior['mesh_sha256'] != provenance['mesh_sha256']:
|
|
raise RuntimeError('Restart mesh differs from the snapshot campaign.')
|
|
for saved_file in (snapshot_file, prior_folder / 'metadata.txt', prior_folder / 'layout.csv'):
|
|
value = digest(saved_file)
|
|
key = str(saved_file.relative_to(prior_folder))
|
|
if value != saved_run['output_hashes'][key]:
|
|
raise RuntimeError('Restart input no longer matches its original run: ' + key)
|
|
state_files[str(saved_file)] = value
|
|
state_files[str(prior_manifest)] = digest(prior_manifest)
|
|
state_files[str(prior_run)] = digest(prior_run)
|
|
continue
|
|
# Modal coordinates also depend on the mesh and named block
|
|
# layout. Bind replay to its original recorded campaign rather
|
|
# than trusting a coincidentally matching vector length.
|
|
prior_manifest = path.parent / 'provenance.json'
|
|
prior_run = path.parent / (path.name + '_run.json')
|
|
prior = json.loads(prior_manifest.read_text())
|
|
saved_run = json.loads(prior_run.read_text())
|
|
if prior['mesh_sha256'] != provenance['mesh_sha256']:
|
|
raise RuntimeError('Density replay mesh differs from the snapshot campaign.')
|
|
for filename in ('accepted_state.txt', 'metadata.txt', 'layout.csv'):
|
|
snapshot_file = path / filename
|
|
value = digest(snapshot_file)
|
|
if value != saved_run['output_hashes'][filename]:
|
|
raise RuntimeError('Density replay input no longer matches its original run: ' + filename)
|
|
state_files[str(snapshot_file)] = value
|
|
state_files[str(prior_manifest)] = digest(prior_manifest)
|
|
state_files[str(prior_run)] = digest(prior_run)
|
|
print('START', name, flush=True)
|
|
start = time.monotonic()
|
|
cpu_start = resource.getrusage(resource.RUSAGE_CHILDREN)
|
|
record = {'id': name, 'command': command, 'started_utc': utc(), 'input_state_hashes': state_files}
|
|
write_json(output / (name + '_run.json'), record)
|
|
with (output / (name + '.log')).open('w') as stream:
|
|
try:
|
|
run = subprocess.run(command, cwd=source, stdout=stream, stderr=subprocess.STDOUT,
|
|
env={**os.environ, 'OMPI_MCA_btl': 'self,sm'}, timeout=args.timeout)
|
|
record['exit_code'] = run.returncode
|
|
except subprocess.TimeoutExpired:
|
|
record['timed_out'] = True
|
|
record['exit_code'] = None
|
|
record.update(seconds=time.monotonic() - start, finished_utc=utc())
|
|
cpu_end = resource.getrusage(resource.RUSAGE_CHILDREN)
|
|
# Keep independent clocks: suspend can affect elapsed clocks differently,
|
|
# and rescaling solver phase timings would hide that uncertainty.
|
|
record['utc_elapsed_seconds'] = (
|
|
datetime.datetime.fromisoformat(record['finished_utc']) -
|
|
datetime.datetime.fromisoformat(record['started_utc'])).total_seconds()
|
|
record['utc_minus_monotonic_seconds'] = record['utc_elapsed_seconds'] - record['seconds']
|
|
record['child_user_cpu_seconds'] = cpu_end.ru_utime - cpu_start.ru_utime
|
|
record['child_system_cpu_seconds'] = cpu_end.ru_stime - cpu_start.ru_stime
|
|
record['child_cpu_seconds'] = record['child_user_cpu_seconds'] + record['child_system_cpu_seconds']
|
|
record['output_hashes'] = {
|
|
str(path.relative_to(folder)): digest(path)
|
|
for path in sorted(folder.rglob('*'))
|
|
if path.is_file() and path.suffix in ('.txt', '.csv', '.json')
|
|
}
|
|
record['inputs_unchanged'] = all(digest(Path(path)) == value for path, value in state_files.items())
|
|
write_json(output / (name + '_run.json'), record)
|
|
summaries.append({**record, **summarize(folder)})
|
|
write_json(output / 'summary.json', summaries)
|
|
print('FINISH', name, record['exit_code'], round(record['seconds'], 2), flush=True)
|
|
if record['exit_code'] not in (0, 3):
|
|
raise RuntimeError('Case failed to execute; see its retained log and run record.')
|
|
if not record['inputs_unchanged']:
|
|
raise RuntimeError('A frozen input changed during the diagnostic run.')
|
|
# Exit 0 denotes completed observations; exit 3 from a model remains explicitly recorded as nonconvergence.
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|