235 lines
12 KiB
Python
235 lines
12 KiB
Python
"""Opt-in, frozen N64 space-time GPU trajectory; no convergence certification."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import resource
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
import numpy as np
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
sys.path.insert(0, str(ROOT/'src'))
|
|
sys.path.insert(0, str(Path(__file__).parents[1]/'taylor_green'))
|
|
from CelerisLab.config import LBMConfig
|
|
import oracle as tg
|
|
import spacetime_gpu as gpu
|
|
import spacetime_table_guard as table_guard
|
|
ref = gpu.ref
|
|
|
|
CPU_SHA = '180eef90a1c3f43113fe18cc54981ee4723cfc0cc2724334907ff188022fa770'
|
|
UNIFORM_SHA = '01e7415ae39456f2de7f0c3d331afb971c49d19353b43038650e32e80737f1e7'
|
|
CUDA_SHA = 'c47869688f7ac2ec93411c0af233887fe8b77405838299e4ae52c8963fd97fe4'
|
|
STEPS = (256, 512, 768, 1024)
|
|
TIMES = (.12, .24, .36, .48)
|
|
FACTORS = dict(velocity_l2=.8, pressure_l2=1., energy_relative=1.)
|
|
LIMITS = dict(velocity_l2=.05, pressure_l2=.15, energy_relative=.05, mass_relative=1e-5)
|
|
HEALTH = ('boundary_pre_nonfinite', 'boundary_pre_nonpositive', 'boundary_stored_nonfinite',
|
|
'boundary_stored_nonpositive', 'physical_population_nonfinite', 'physical_population_nonpositive')
|
|
|
|
|
|
def sha(path):
|
|
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
|
|
|
|
|
def load_references(cpu_path, uniform_path):
|
|
for path, expected in ((cpu_path, CPU_SHA), (uniform_path, UNIFORM_SHA)):
|
|
if sha(path) != expected:
|
|
raise ValueError(f'pinned reference receipt differs: {path}')
|
|
cpu, uniform = [json.loads(Path(p).read_text()) for p in (cpu_path, uniform_path)]
|
|
assert cpu['completed'] and cpu['experimental_pass'] and not cpu['source']['dirty']
|
|
assert cpu['source']['commit'] == '07879bc52bfef8d95503146c55d3823c8b6f8d0a'
|
|
assert [r['mode'] for r in cpu['runs']] == ['constant_ivp', 'spacetime']
|
|
assert all(tuple(row['tstar'] for row in r['rows']) == TIMES for r in cpu['runs'])
|
|
assert uniform['completed'] and not uniform['source']['dirty']
|
|
assert uniform['source']['revision'] == '9805383f9c50d867e8555471baf5dbda8333a652'
|
|
assert (uniform['n'], uniform['depth'], uniform['viscosity']) == (64, 1, .0192)
|
|
assert uniform['limits'] == tg.LIMITS == LIMITS
|
|
return cpu, uniform
|
|
|
|
|
|
def total(state, maps):
|
|
arrays = [a.astype(np.float64) for a in (state.coarse_f, state.fine_f)]
|
|
return ref.cpu.MOMENTS.T @ ref.directional(ref.cpu.State(*arrays), maps)
|
|
|
|
|
|
def warm_and_reset(run, initial, check_cap):
|
|
"""Time 32 synchronized intervals, then restore exact uploaded arrival F."""
|
|
run.initialize(initial)
|
|
before = run.download()
|
|
started = time.monotonic()
|
|
for _ in range(32):
|
|
check_cap()
|
|
run.step(audit=False)
|
|
seconds = time.monotonic()-started
|
|
warm_health = run.health_counts.tolist()
|
|
run.initialize(initial)
|
|
after = run.download()
|
|
for a, b in zip((before.coarse_f, before.fine_f), (after.coarse_f, after.fine_f)):
|
|
np.testing.assert_array_equal(a, b)
|
|
if run.step_count != 0 or after.tick != 0 or run.health_counts.any():
|
|
raise AssertionError('warm-up reset did not restore tick/counters')
|
|
return after, dict(intervals=32, seconds=seconds, estimated_1024_seconds=32*seconds,
|
|
health_counts=warm_health, reset_verified=True)
|
|
|
|
|
|
def assess(rows, cpu, uniform, health_counts):
|
|
if tuple(row['tstar'] for row in rows) != TIMES:
|
|
raise ValueError('four frozen sample times required')
|
|
worst = {key: max(row[key] for row in rows) for key in LIMITS}
|
|
campaign = {key: all(np.isfinite(row[key]) and 0 <= row[key] <= limit for row in rows)
|
|
for key, limit in LIMITS.items()}
|
|
campaign['population_health'] = not any(health_counts)
|
|
campaign['sampled_positive_finite'] = all(
|
|
np.isfinite(row[key]) and row[key] > 0 for row in rows
|
|
for key in ('minimum_population', 'minimum_boundary_pre', 'minimum_boundary_stored'))
|
|
baseline, candidate = cpu['runs']
|
|
thresholds = {key: factor*baseline['worst'][key] for key, factor in FACTORS.items()}
|
|
experimental = {key: all(np.isfinite(row[key]) and row[key] >= 0 for row in rows)
|
|
and worst[key] <= threshold for key, threshold in thresholds.items()}
|
|
differences = []
|
|
for actual, expected in zip(rows, candidate['rows']):
|
|
delta = {}
|
|
for key, value in expected.items():
|
|
d = np.asarray(actual[key])-np.asarray(value)
|
|
delta[key] = dict(cpu=value, signed=d.tolist(), absolute=np.abs(d).tolist())
|
|
differences.append(dict(tstar=actual['tstar'], fields=delta))
|
|
historical_worst = max(row['pressure_l2'] for row in uniform['rows'] if row['tstar'] > 0)
|
|
historical_limit = max(2*historical_worst, .02)
|
|
assert historical_limit == .024506347612699403
|
|
return dict(worst=worst, campaign_checks=campaign, campaign_pass=all(campaign.values()),
|
|
experimental_thresholds=thresholds, experimental_checks=experimental,
|
|
experimental_pass=all(experimental.values()),
|
|
gpu_minus_cpu=dict(status='descriptive_only_no_equivalence_tolerance', rows=differences),
|
|
historical_pressure_gate=dict(status='historical_shifted_FP32_uniform_vs_unshifted_FP32_candidate',
|
|
matched_storage=False, uniform_worst=historical_worst, threshold=historical_limit,
|
|
candidate_worst=worst['pressure_l2'], passed=worst['pressure_l2'] <= historical_limit))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--gpu-admitted', action='store_true')
|
|
parser.add_argument('--cpu-receipt', type=Path, required=True)
|
|
parser.add_argument('--uniform-receipt', type=Path, required=True)
|
|
parser.add_argument('--output', type=Path, required=True)
|
|
args = parser.parse_args()
|
|
if not args.gpu_admitted or not os.environ.get('CUDA_VISIBLE_DEVICES'):
|
|
parser.error('explicit --gpu-admitted and CUDA_VISIBLE_DEVICES required')
|
|
if args.output.exists():
|
|
parser.error('output exists; preserve evidence and choose a new path')
|
|
cpu, uniform = load_references(args.cpu_receipt, args.uniform_receipt)
|
|
gpu.check_sources()
|
|
if sha(Path(gpu.__file__).with_suffix('.cu')) != CUDA_SHA:
|
|
raise RuntimeError('reviewed CUDA source changed')
|
|
source = dict(commit=subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=ROOT, text=True).strip(),
|
|
dirty=bool(subprocess.check_output(['git', 'status', '--porcelain'], cwd=ROOT, text=True)),
|
|
hashes={p.name: sha(p) for p in (Path(__file__), Path(gpu.__file__), Path(gpu.__file__).with_suffix('.cu'),
|
|
Path(__file__).with_name('test_spacetime_gpu.py'), Path(table_guard.__file__))},
|
|
reference_hashes=gpu.PINS, numpy=np.__version__, checkout=str(ROOT))
|
|
if source['dirty']:
|
|
raise RuntimeError('clean committed checkout required')
|
|
started = time.monotonic()
|
|
report = dict(scope='N64 experimental space-time GPU; no grid-convergence qualification',
|
|
completed=False, passed=False, source=source, rows=[], health_counts=[0]*6, health_labels=HEALTH,
|
|
setup=dict(n=64, bounds=[8,8,56,48], re_L=100, U=.03, viscosity=.0192, tau=[.5576,.6152],
|
|
phase='arrival_f', storage='unshifted_FP32', collision='SRT', interval_count=1024,
|
|
initialization='same physical FP64 equilibrium as CPU, rounded to FP32 on upload',
|
|
cpu_storage='FP64', sample_steps=STEPS, sample_times=TIMES),
|
|
references=dict(cpu_receipt_sha256=CPU_SHA, cpu_source=cpu['source'],
|
|
cpu_archive_drive_id='197F9nwDOGf_YROK1a46YpVvNrSldnH_w',
|
|
cpu_archive_sha256='850ddce3f5203ef38e76e5153e718a9b92e137d1fc388ba7cd12f5d47eab83b4',
|
|
uniform_receipt_sha256=UNIFORM_SHA, uniform_source=uniform['source'],
|
|
uniform_storage='historical_shifted_FP32'),
|
|
frozen_screen=dict(campaign_limits=LIMITS, experimental_factors=FACTORS,
|
|
gpu_cpu_equivalence_tolerance=None, historical_pressure_threshold=.024506347612699403),
|
|
boundary_budget_status='not_measured', max_directional_residual=None,
|
|
minima_scope='initial and four sampled physical arrival states; boundary minima at four samples',
|
|
limits=dict(seconds=180., peak_rss_mib=512.), cuda_visible_devices=os.environ['CUDA_VISIBLE_DEVICES'])
|
|
run = context = None
|
|
phase = 'setup'
|
|
|
|
def save():
|
|
report['resources'] = dict(seconds=time.monotonic()-started,
|
|
peak_rss_mib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/1024,
|
|
device_allocations_bytes=None if run is None else run.device_bytes,
|
|
device_memory_scope='owner allocations; excludes compiler, driver and context')
|
|
if run is not None:
|
|
report['health_counts'] = run.health_counts.tolist()
|
|
report['completed_intervals'] = run.step_count
|
|
ref.write_receipt(args.output, report)
|
|
|
|
def check_cap():
|
|
if time.monotonic()-started > 180 or resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/1024 > 512:
|
|
raise RuntimeError('frozen resource cap exceeded')
|
|
|
|
save()
|
|
try:
|
|
from CelerisLab.cuda.context import CudaContext
|
|
context = CudaContext(0)
|
|
config = LBMConfig(nx=64, ny=64, viscosity=.0192, velocity=0., ddf_shifting=False)
|
|
run = gpu.SpaceTimeGPU(config, (8,8,56,48), context=context)
|
|
report.update(device=context.name, sm_arch=context.sm_arch, build_metadata=run.build_metadata,
|
|
table_sha256=gpu.table_hash(run.tables))
|
|
report['table_validation'] = table_guard.validate_tables(run.plan,run.tables)
|
|
initial = ref.cpu.State(tg.initial((64,64),0,64,.0192), tg.initial((128,128),1,64,.0192))
|
|
phase = 'warm_reset'
|
|
initial, report['warmup'] = warm_and_reset(run, initial, check_cap)
|
|
maps = run.plan.maps
|
|
masks = (maps.coarse_active, maps.fine_active)
|
|
arrays = (initial.coarse_f, initial.fine_f)
|
|
mass0 = tg.metrics(arrays, masks, 0, 64, .0192)['mass']
|
|
moment0 = total(initial, maps)
|
|
report['initial'] = dict(mass=mass0, momentum=moment0[1:].tolist(),
|
|
minimum_population=min(float(a[:,m].min()) for a,m in zip(arrays,masks)))
|
|
save()
|
|
check_cap()
|
|
phase = 'trajectory'
|
|
trajectory_started = time.monotonic()
|
|
for step in range(1,1025):
|
|
check_cap()
|
|
run.step(audit=False)
|
|
if step in STEPS:
|
|
state = run.download()
|
|
if state.tick != 2*step:
|
|
raise AssertionError('sample is not synchronized arrival state')
|
|
arrays = (state.coarse_f, state.fine_f)
|
|
row = tg.metrics(arrays, masks, step, 64, .0192, mass0)
|
|
row.update(step=step, momentum_change=(total(state,maps)-moment0)[1:].tolist(),
|
|
mass_change=row['mass']-mass0,
|
|
minimum_population=min(float(a[:,m].min()) for a,m in zip(arrays,masks)))
|
|
with context.activate():
|
|
row['minimum_boundary_pre'] = float(run.read(run.pre,(len(run.plan.groups),4),np.float64).min())
|
|
row['minimum_boundary_stored'] = float(run.read(run.h,(len(run.plan.groups),4)).min())
|
|
if not all(np.isfinite(value).all() for value in row.values()):
|
|
report['failed_sample'] = {key: repr(value) for key,value in row.items()}
|
|
raise FloatingPointError('nonfinite sample diagnostic')
|
|
report['rows'].append(row)
|
|
save()
|
|
print(json.dumps(row, allow_nan=False), flush=True)
|
|
report['trajectory_seconds'] = time.monotonic()-trajectory_started
|
|
check_cap()
|
|
report.update(assess(report['rows'], cpu, uniform, run.health_counts))
|
|
report['completed'] = True
|
|
report['passed'] = report['campaign_pass'] and report['experimental_pass']
|
|
except Exception as exc:
|
|
report['failure'] = dict(phase=phase, error=repr(exc))
|
|
raise
|
|
finally:
|
|
# Preserve the primary failure and every completed sample even if cleanup fails.
|
|
for owner in (run, context):
|
|
if owner is not None:
|
|
try:
|
|
owner.close()
|
|
except Exception as exc:
|
|
report.setdefault('cleanup_errors', []).append(repr(exc))
|
|
report['passed'] = False
|
|
save()
|
|
return 0 if report['passed'] else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|