206 lines
13 KiB
Python
206 lines
13 KiB
Python
"""Host checks or separately admitted tiny GPU ladder capability; no trajectories."""
|
|
import argparse
|
|
from contextlib import nullcontext
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import Mock
|
|
|
|
import numpy as np
|
|
import run_spacetime_gpu_ladder as ladder
|
|
|
|
gpu,ref,tg=ladder.gpu,ladder.ref,ladder.tg
|
|
RECORDS=[]
|
|
ADMITTED=False
|
|
|
|
|
|
def config(n):
|
|
return ladder.old.LBMConfig(nx=n,ny=n,viscosity=.0003*n,velocity=0.,ddf_shifting=True)
|
|
|
|
|
|
class Host(unittest.TestCase):
|
|
def test_scaling_and_admission(self):
|
|
for _,n,refined in ladder.CASES:
|
|
s=ladder.setup(n,refined)
|
|
self.assertEqual(s['finest_n'],n*(2 if refined else 1))
|
|
self.assertEqual(len(s['early_steps']),65)
|
|
np.testing.assert_allclose(np.array(s['early_steps'])*.03/n,s['early_times'],rtol=0,atol=3e-17)
|
|
self.assertEqual(s['intervals']*.03/n,.48)
|
|
gpu.validate_config(config(n),uniform=not refined,allow_n128=refined)
|
|
with self.assertRaises(ValueError): gpu.validate_config(config(128))
|
|
with self.assertRaises(ValueError): gpu.validate_config(config(256),allow_n128=True)
|
|
with self.assertRaises(ValueError): gpu.validate_config(config(512),uniform=True)
|
|
|
|
def test_n128_tables_and_mutations(self):
|
|
plan=ref.Plan(128,(16,16,112,96)); tables=gpu.pack_plan(plan)
|
|
out=ladder.old.table_guard.validate_tables(plan,tables,allow_n128=True)
|
|
self.assertEqual(out['structure_sha256'],'ee4ecb645674ae04eb87220a5217851405ccc0f45fbb0fe210499dcea5cd93b1')
|
|
self.assertEqual(len(plan.groups),1060)
|
|
with self.assertRaises(ValueError): ladder.old.table_guard.validate_tables(plan,tables)
|
|
for kind in ('topology','balanced_weights'):
|
|
bad={k:v.copy() for k,v in tables.items()}
|
|
if kind=='topology': bad['sources'][0,0]+=1
|
|
else: bad['weights'][0,:2,:2]+=1e-5*np.array([[1,-1],[-1,1]])
|
|
with self.assertRaises(ValueError): ladder.old.table_guard.validate_tables(plan,bad,allow_n128=True)
|
|
|
|
def test_projection_manufactured_nonuniform_leaves(self):
|
|
n=16; plan=ref.Plan(n,(3,3,13,12)); masks=(plan.maps.coarse_active,plan.maps.fine_active)
|
|
coords=[tg.centers(m.shape,l) for l,m in enumerate(masks)]
|
|
x=np.concatenate([v[0][m] for v,m in zip(coords,masks)])
|
|
y=np.concatenate([v[1][m] for v,m in zip(coords,masks)])
|
|
w=np.concatenate([np.full(m.sum(),.25**l) for l,m in enumerate(masks)])
|
|
b=np.column_stack((np.cos(4*np.pi*x/n),np.sin(4*np.pi*x/n),np.cos(4*np.pi*y/n),np.sin(4*np.pi*y/n)))
|
|
b-=w@b/w.sum(); gram=b.T@(w[:,None]*b)
|
|
residual=np.cos(6*np.pi*x/n)+np.sin(6*np.pi*y/n); residual-=w@residual/w.sum()
|
|
residual-=b@np.linalg.solve(gram,b.T@(w*residual)); residual*=1e-5
|
|
coefficients=np.array([2,-3,4,-1])*1e-5
|
|
pressure=tg.exact(x,y,0,n,.0003*n)[2]+b@coefficients+residual+.017
|
|
arrays=[]; offset=0
|
|
for m in masks:
|
|
a=np.full((9,*m.shape),np.nan); count=m.sum()
|
|
a[:,m]=ref.cpu.W[:,None]*(1+3*pressure[offset:offset+count]); arrays.append(a); offset+=count
|
|
result=ladder.pressure_projection(arrays,masks,0,n,.0003*n)
|
|
np.testing.assert_allclose(result['coefficients'],coefficients,rtol=0,atol=3e-17)
|
|
self.assertAlmostEqual(result['residual_rms'],np.sqrt(w@(residual*residual)/w.sum()),places=15)
|
|
np.testing.assert_allclose(result['orthogonality'],0,rtol=0,atol=1e-18)
|
|
self.assertLess(abs(result['energy_closure']),1e-20)
|
|
expected=tg.metrics(arrays,masks,0,n,.0003*n)['pressure_l2']
|
|
self.assertAlmostEqual(result['relative_rms'],expected,places=12)
|
|
|
|
def test_uniform_budget_and_negative_control(self):
|
|
rng=np.random.default_rng(12); a=rng.normal(0,.001,(9,12,12)).astype(np.float32)
|
|
g=(a*.99).astype(np.float32)
|
|
b=np.stack([np.roll(g[q],(int(ref.cpu.CY[q]),int(ref.cpu.CX[q])),axis=(0,1)) for q in range(9)])
|
|
ledger=gpu.uniform_ledger(a,b,g,shifted=True)
|
|
self.assertLess(max(abs(np.array(ledger['unexplained']))),ledger['unexplained_bound'])
|
|
b[1,0,0]+=1e-4
|
|
with self.assertRaises(AssertionError): gpu.uniform_ledger(a,b,g,shifted=True)
|
|
|
|
def test_uniform_failed_io_poisoning(self):
|
|
run=object.__new__(gpu.UniformGPU); run.configs=(config(12),); run.status='ready'
|
|
run.ctx=SimpleNamespace(activate=nullcontext); run.buffers=['f']; run.cuda=SimpleNamespace(memcpy_htod=Mock(side_effect=RuntimeError('copy')))
|
|
with self.assertRaisesRegex(RuntimeError,'copy'): run.initialize(np.broadcast_to(ref.cpu.W[:,None,None],(9,12,12)))
|
|
self.assertEqual(run.status,'failed'); run.status='ready'; run.read=Mock(side_effect=RuntimeError('read'))
|
|
with self.assertRaisesRegex(RuntimeError,'read'): run._download_stored()
|
|
self.assertEqual(run.status,'failed')
|
|
|
|
def test_uniform_health_mask_reaches_both_kernels(self):
|
|
run=object.__new__(gpu.UniformGPU); run.configs=(config(16),); run.status='ready'; run.identity=False
|
|
run.ctx=SimpleNamespace(activate=nullcontext); run.buffers=['f','g','out']; run.active=object(); run.counts='counts'
|
|
run.stream=object(); run.cuda=SimpleNamespace(memset_d32_async=Mock()); run.launch=Mock()
|
|
run.read=Mock(return_value=np.zeros(6,np.uint32)); run.health_counts=np.zeros(6,np.uint64); run.step_count=0
|
|
run.step(audit=False)
|
|
self.assertEqual([call.args[1] for call in run.launch.call_args_list],['Collide','Pull'])
|
|
self.assertTrue(all(call.args[4] is run.active for call in run.launch.call_args_list))
|
|
|
|
def test_zero_error_comparisons(self):
|
|
rows=[dict(tstar=t,velocity_l2=0.,pressure_l2=0.,energy_relative=0.) for t in ladder.old.TIMES]
|
|
result=ladder.comparisons([dict(name=name,rows=rows) for name,_,_ in ladder.CASES],dict(rows=rows))
|
|
self.assertIsNone(result['trends'][0]['samples'][0]['pressure_l2']['ratio'])
|
|
self.assertNotIn('two_point_slope',result['controls'][0]['samples'][0]['pressure_l2'])
|
|
self.assertIn('two_point_slope',result['trends'][0]['samples'][0]['pressure_l2'])
|
|
json.dumps(result,allow_nan=False)
|
|
|
|
def test_frozen_screen_and_json_receipt(self):
|
|
rows=[dict(tstar=t,minimum_population=.02,**{key:np.float64(limit*.5) for key,limit in ladder.LIMITS.items()}) for t in ladder.old.TIMES]
|
|
self.assertTrue(ladder.assess(rows,[0]*6)['campaign_pass'])
|
|
rows[0]['mass_relative']=2e-5
|
|
failed=ladder.assess(rows,[0]*6); self.assertFalse(failed['campaign_pass'])
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path=Path(tmp)/'receipt.json'; ref.write_receipt(path,dict(completed=True,passed=np.bool_(False),assessment=failed))
|
|
saved=json.loads(path.read_text()); self.assertFalse(saved['passed']); self.assertTrue(saved['completed'])
|
|
|
|
|
|
class Device(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
if not ADMITTED: raise unittest.SkipTest('explicit tiny GPU admission required')
|
|
from CelerisLab.cuda.context import CudaContext
|
|
cls.context=CudaContext(0)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls): cls.context.close()
|
|
|
|
def test_uniform_actual_upload_oracle_and_reset(self):
|
|
for n in (16,64,128,256):
|
|
with gpu.UniformGPU(config(n),context=self.context) as run:
|
|
analytic=tg.initial((n,n),0,n,.0003*n); run.initialize(analytic); start=run._download_stored()
|
|
actual_start=run.download(); mask=np.ones((n,n),bool)
|
|
spec=ref.Spec(n_coarse=n,patch_lo=1,halo=2,tau_coarse=.5+.0009*n,tau_fine=.5+.0018*n,collision='srt')
|
|
collided=ref.cpu._collide(spec,actual_start,np.isfinite(actual_start),spec.tau_coarse,mask,np.zeros((n,n),int))
|
|
expected=np.stack([np.roll(collided[q],(int(ref.cpu.CY[q]),int(ref.cpu.CX[q])),axis=(0,1)) for q in range(9)])
|
|
ledger=run.step(); actual=run.download()
|
|
np.testing.assert_allclose(actual,expected,rtol=0,atol=3e-6)
|
|
audited=run._download_stored(); run.initialize(analytic); run.step(audit=False)
|
|
np.testing.assert_array_equal(audited.view(np.uint32),run._download_stored().view(np.uint32))
|
|
run.initialize(analytic); np.testing.assert_array_equal(start.view(np.uint32),run._download_stored().view(np.uint32))
|
|
self.assertEqual(run.step_count,0); self.assertFalse(run.health_counts.any())
|
|
RECORDS.append(dict(case=f'uniform{n}_one_interval',error=float(abs(actual-expected).max()),ledger=ledger,
|
|
build_metadata=run.build_metadata,device_bytes=run.device_bytes,raw_reset_verified=True))
|
|
|
|
def test_uniform_tagged_phase_and_health_controls(self):
|
|
n=16; rng=np.random.default_rng(n)
|
|
analytic=np.broadcast_to(ref.cpu.W[:,None,None],(9,n,n))*(1+1e-3*rng.standard_normal((9,n,n)))
|
|
for identity in (True,False):
|
|
with gpu.UniformGPU(config(n),context=self.context,identity=identity) as run:
|
|
run.initialize(analytic); start=run.download(); raw_start=run._download_stored(); mask=np.ones((n,n),bool)
|
|
spec=ref.Spec(tau_coarse=.5+.0009*n,collision='identity' if identity else 'srt')
|
|
collided=ref.cpu._collide(spec,start,np.isfinite(start),spec.tau_coarse,mask,np.zeros((n,n),int))
|
|
expected=np.stack([np.roll(collided[q],(int(ref.cpu.CY[q]),int(ref.cpu.CX[q])),axis=(0,1)) for q in range(9)])
|
|
ledger=run.step(); actual=run.download(); np.testing.assert_allclose(actual,expected,rtol=0,atol=3e-6)
|
|
if identity:
|
|
rolled=np.stack([np.roll(raw_start[q],(int(ref.cpu.CY[q]),int(ref.cpu.CX[q])),axis=(0,1)) for q in range(9)])
|
|
np.testing.assert_array_equal(run._download_stored().view(np.uint32),rolled.view(np.uint32))
|
|
if not identity:
|
|
wrong=ref.cpu._collide(spec,expected,np.isfinite(expected),spec.tau_coarse,mask,np.zeros((n,n),int))
|
|
self.assertGreater(float(abs(actual-wrong).max()),3e-6)
|
|
RECORDS.append(dict(case='uniform_tagged',identity=identity,error=float(abs(actual-expected).max()),ledger=ledger))
|
|
with gpu.UniformGPU(config(n),context=self.context) as run:
|
|
run.initialize(analytic)
|
|
with self.context.activate():
|
|
bad=np.broadcast_to(-gpu._WEIGHTS-.01,(9,n,n)).astype(np.float32).copy()
|
|
run.cuda.memcpy_htod(run.buffers[0],bad)
|
|
with self.assertRaises(AssertionError): run.step()
|
|
self.assertTrue(run.health_counts.any()); self.assertEqual(run.status,'failed')
|
|
RECORDS.append(dict(case='uniform_health_negative_control',health_counts=run.health_counts.tolist()))
|
|
|
|
def test_refined_n128_actual_upload_oracle(self):
|
|
with gpu.SpaceTimeGPU(config(128),(16,16,112,96),context=self.context,allow_n128=True) as run:
|
|
tables=ladder.old.table_guard.validate_tables(run.plan,run.tables,allow_n128=True)
|
|
run.initialize(ref.cpu.State(tg.initial((128,128),0,128,.0384),tg.initial((256,256),1,128,.0384)))
|
|
spec=ref.Spec(n_coarse=128,patch_lo=16,halo=32,tau_coarse=.6152,tau_fine=.7304,collision='srt')
|
|
expected,_=ref.advance(spec,run.plan,run.download()); ledger=run.step(); actual=run.download(); errors=[]
|
|
for a,b in zip((actual.coarse_f,actual.fine_f),(expected.coarse_f,expected.fine_f)):
|
|
np.testing.assert_allclose(a,b,rtol=0,atol=3e-6,equal_nan=True); errors.append(float(np.nanmax(abs(a-b))))
|
|
RECORDS.append(dict(case='refined128_one_interval',errors=errors,ledger=ledger,tables=tables,
|
|
build_metadata=run.build_metadata,device_bytes=run.device_bytes))
|
|
|
|
|
|
def main():
|
|
global ADMITTED
|
|
p=argparse.ArgumentParser(description=__doc__); modes=p.add_mutually_exclusive_group(required=True)
|
|
modes.add_argument('--host-only',action='store_true'); modes.add_argument('--gpu-admitted',action='store_true')
|
|
p.add_argument('--output',type=Path,required=True); args=p.parse_args(); ADMITTED=args.gpu_admitted
|
|
if args.output.exists(): p.error('output exists; preserve evidence')
|
|
if ADMITTED and not os.environ.get('CUDA_VISIBLE_DEVICES'): p.error('explicit CUDA_VISIBLE_DEVICES required')
|
|
paths=(Path(__file__),Path(ladder.__file__),Path(gpu.__file__),Path(gpu.__file__).with_suffix('.cu'),Path(ladder.old.table_guard.__file__))
|
|
report=dict(completed=False,passed=False,mode='tiny_gpu' if ADMITTED else 'host_only',records=RECORDS,
|
|
source=dict(commit=subprocess.check_output(['git','rev-parse','HEAD'],cwd=ladder.ROOT,text=True).strip(),
|
|
dirty=bool(subprocess.check_output(['git','status','--porcelain'],cwd=ladder.ROOT,text=True)),
|
|
hashes={p.name:ladder.old.sha(p) for p in paths}))
|
|
ref.write_receipt(args.output,report); started=time.monotonic()
|
|
suite=unittest.TestLoader().loadTestsFromTestCase(Host)
|
|
if ADMITTED: suite.addTests(unittest.TestLoader().loadTestsFromTestCase(Device))
|
|
result=unittest.TextTestRunner(verbosity=2).run(suite)
|
|
report.update(completed=True,passed=result.wasSuccessful(),tests=result.testsRun,seconds=time.monotonic()-started,
|
|
failures=[dict(test=str(t),error=e) for t,e in result.errors+result.failures])
|
|
ref.write_receipt(args.output,report); return 0 if report['passed'] else 1
|
|
|
|
|
|
if __name__=='__main__': raise SystemExit(main())
|