Compare commits

...
11 changed files with 3101 additions and 0 deletions
@@ -0,0 +1,263 @@
"""Four opt-in shifted GPU controls: U64/U128/U256 and refined base128/fine256."""
import argparse
import hashlib
import inspect
import json
import os
from pathlib import Path
import resource
import subprocess
import time
import numpy as np
import run_spacetime_gpu_n64 as old
gpu,ref,tg,ROOT = old.gpu,old.ref,old.tg,old.ROOT
BASE='26d0d3853c9d4d3cd6ae3f27653dced56f8c5fec'
REFERENCE_SHA='9d154c989b087bf288f86d413db0f3067b73767b618c329d4b9e53c4eee00032'
CASES=(('uniform64',64,False),('uniform128',128,False),('uniform256',256,False),('refined128',128,True))
LIMITS=old.LIMITS
SECONDS,RSS_MIB=600.,512.
def setup(n, refined):
return dict(base_n=n,finest_n=2*n if refined else n,normalized_domain=[0.,1.,0.,1.],
re_L=100,U=.03,mach=.03*np.sqrt(3),viscosity=.0003*n,
tau=[.5+.0009*n]+([.5+.0018*n] if refined else []),
bounds=[n//8,n//8,7*n//8,3*n//4] if refined else None,
normalized_patch=[.125,.125,.875,.75] if refined else None,
intervals=16*n,sample_steps=[4*n,8*n,12*n,16*n],sample_times=list(old.TIMES),
early_steps=list(range(0,4*n+1,n//16)),early_times=(np.arange(65)*.001875).tolist(),
phase='arrival_f',storage='shifted_FP32',collision='SRT',
initialization='common analytic FP64 equilibrium at actual cell centers, production shifted codec')
def pressure_projection(arrays,masks,t,n,viscosity):
"""One global gauge and a weighted Gram fit, including unequal leaf areas."""
rows=[]
for level,(a,mask) in enumerate(zip(arrays,masks)):
rho=np.asarray(a[:,mask],np.float64).sum(axis=0)
x,y=tg.centers(mask.shape,level); x,y=x[mask],y[mask]
exact=tg.exact(x,y,t,n,viscosity)[2]
angle=4*np.pi/n
basis=np.column_stack((np.cos(angle*x),np.sin(angle*x),np.cos(angle*y),np.sin(angle*y)))
rows.append(((rho-1)/3,exact,np.full(len(rho),.25**level),basis))
p,exact,w,b=[np.concatenate(x) for x in zip(*rows)]
volume=w.sum(); p-=w@p/volume; exact-=w@exact/volume; e=p-exact
b-=w@b/volume
gram=b.T@(w[:,None]*b); rhs=b.T@(w*e)
coefficients=np.linalg.solve(gram,rhs); residual=e-b@coefficients
norm=float(w@(e*e)); exact_norm=float(w@(exact*exact)); remaining=float(w@(residual*residual))
projected=float(coefficients@gram@coefficients)
return dict(step=t,tstar=t*.03/n,absolute_rms=np.sqrt(norm/volume),relative_rms=np.sqrt(norm/exact_norm),
coefficients=coefficients.tolist(),basis=['cos_4pi_x','sin_4pi_x','cos_4pi_y','sin_4pi_y'],
residual_rms=np.sqrt(remaining/volume),residual_relative=np.sqrt(remaining/exact_norm),
gram_condition=float(np.linalg.cond(gram)),
orthogonality=(b.T@(w*residual)/volume).tolist(),energy_closure=(norm-projected-remaining)/volume)
def load_reference(path):
if old.sha(path)!=REFERENCE_SHA:
raise ValueError('verified shifted N64 receipt differs')
data=json.loads(Path(path).read_text())
if (not data['completed'] or not data['passed'] or data['source']['commit']!=BASE
or data['setup']['storage']!='shifted_FP32' or data['setup']['bounds']!=[8,8,56,48]
or data['setup']['interval_count']!=1024 or len(data['rows'])!=4):
raise ValueError('N64 reference identity differs')
if tuple(row['tstar'] for row in data['rows'])!=old.TIMES:
raise ValueError('N64 reference sample times differ')
return data
def source_identity(reference):
git=lambda *args:subprocess.check_output(['git',*args],cwd=ROOT,text=True).strip()
if git('status','--porcelain'):
raise RuntimeError('clean committed checkout required')
unchanged=['src','tests/validation/taylor_green/oracle.py']+[f'tests/validation/multigrid/{name}' for name in
('spacetime_gpu.cu','spacetime_reference.py','cpu_reference.py','run_spacetime_gpu_n64.py','test_spacetime_gpu.py')]
subprocess.check_call(['git','diff','--exit-code',BASE,'--',*unchanged],cwd=ROOT)
previous=subprocess.check_output(['git','show',f'{BASE}:tests/validation/multigrid/spacetime_gpu.py'],cwd=ROOT,text=True)
# The raw interval schedule remains byte-for-byte unchanged despite the new owner.
method=inspect.getsource(gpu.SpaceTimeGPU.step)
if method not in previous or np.__version__!=reference['source']['numpy']:
raise RuntimeError('raw schedule or NumPy identity changed')
files=[Path(__file__),Path(gpu.__file__),Path(old.table_guard.__file__),Path(__file__).with_name('test_spacetime_gpu_ladder.py')]
return dict(commit=git('rev-parse','HEAD'),dirty=False,numpy=np.__version__,production_tree=git('rev-parse','HEAD:src'),
hashes={p.name:old.sha(p) for p in files},cuda_sha256=old.sha(Path(gpu.__file__).with_suffix('.cu')),
raw_schedule_sha256=hashlib.sha256(method.encode()).hexdigest(),unchanged_against=BASE,
reference_hashes=gpu.PINS)
def directional(arrays,masks):
return sum(.25**level*np.asarray(a[:,m],np.float64).sum(axis=1) for level,(a,m) in enumerate(zip(arrays,masks)))
def populations(run,refined,*,raw=False):
state=run._download_stored() if raw else run.download()
if not refined:
return (state,)
if state.phase!='arrival_f' or state.tick!=2*run.step_count:
raise AssertionError('refined sample is not synchronized arrival F')
return state.coarse_f,state.fine_f
def assess(rows,health):
checks={key:all(np.isfinite(row[key]) and 0<=row[key]<=limit for row in rows) for key,limit in LIMITS.items()}
checks['population_health']=not any(health)
checks['sampled_population_positive']=all(row['minimum_population']>0 for row in rows)
return dict(worst={key:max(row[key] for row in rows) for key in LIMITS},campaign_checks=checks,
campaign_pass=all(checks.values()),accuracy_comparison_status='descriptive_only_no_new_threshold')
def ratios(a,b,*,trend=False):
out=dict(ratio=a/b if b>0 else None,ratio_status='defined' if b>0 else 'zero_denominator')
if trend:
out.update(two_point_slope=np.log2(a/b) if a>0 and b>0 else None,
slope_status='defined' if a>0 and b>0 else 'zero_error_log_undefined')
return out
def comparisons(runs,reference):
data={r['name']:r for r in runs}; data['refined64']=reference
out=[]
for refined,base,fine in (('refined64','uniform64','uniform128'),('refined128','uniform128','uniform256')):
for control in (base,fine):
out.append(dict(candidate=refined,control=control,control_spacing='base' if control==base else 'finest',
samples=[dict(tstar=a['tstar'],**{key:dict(candidate=a[key],control=b[key],**ratios(a[key],b[key]))
for key in ('velocity_l2','pressure_l2','energy_relative')})
for a,b in zip(data[refined]['rows'],data[control]['rows'])]))
trends=[]
for coarse,fine in (('uniform64','uniform128'),('uniform128','uniform256'),('refined64','refined128')):
trends.append(dict(coarse=coarse,fine=fine,samples=[dict(tstar=a['tstar'],**{
key:ratios(a[key],b[key],trend=True)
for key in ('velocity_l2','pressure_l2','energy_relative')})
for a,b in zip(data[coarse]['rows'],data[fine]['rows'])]))
return dict(controls=out,trends=trends,caveat='norm differences are not interface error; fixed Mach and two refined sizes do not certify convergence')
def main():
p=argparse.ArgumentParser(description=__doc__)
p.add_argument('--gpu-admitted',action='store_true'); p.add_argument('--reference-n64',type=Path,required=True)
p.add_argument('--output',type=Path,required=True); args=p.parse_args()
if not args.gpu_admitted or not os.environ.get('CUDA_VISIBLE_DEVICES'):
p.error('explicit GPU admission and CUDA_VISIBLE_DEVICES required')
if args.output.exists() or args.output.with_suffix(args.output.suffix+'.tmp').exists(): p.error('output or partial receipt exists; preserve it')
reference=load_reference(args.reference_n64); source=source_identity(reference); gpu.check_sources()
started=time.monotonic()
report=dict(completed=False,passed=False,source=source,runs=[],limits=dict(seconds=SECONDS,peak_rss_mib=RSS_MIB),
campaign_limits=LIMITS,reference=dict(sha256=REFERENCE_SHA,source=reference['source'],
coverage='verified refined base64/fine128 four-time rows only; early pressure trace unavailable'),
historical_pressure_gate=dict(scope='reused_refined64_only',**reference['historical_pressure_gate']),
n64_experimental_screen=dict(scope='reused_refined64_only',checks=reference['experimental_checks']),
frozen_cases=[setup(n,r) for _,n,r in CASES],early_trace_scope='65 predeclared rows per new run, no field snapshots',
cuda_visible_devices=os.environ['CUDA_VISIBLE_DEVICES'])
run=context=ledger=None
case_started=None
def save():
report['resources']=dict(seconds=time.monotonic()-started,peak_rss_mib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/1024,
rss_scope='Python process peak; external wrapper monitors process group and wall time')
report['resources']['within_limits']=report['resources']['seconds']<=SECONDS and report['resources']['peak_rss_mib']<=RSS_MIB
if report['completed'] and not report['resources']['within_limits']: report['passed']=False
if run is not None:
report['runs'][-1].update(completed_intervals=run.step_count,health_counts=run.health_counts.tolist(),
seconds=time.monotonic()-case_started)
if ledger is not None:
report['runs'][-1].update(ledger=ledger.snapshot(),audited_intervals=ledger.steps)
temporary=args.output.with_suffix(args.output.suffix+'.tmp')
ref.write_receipt(temporary,report); os.replace(temporary,args.output)
def cap():
if time.monotonic()-started>SECONDS or resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/1024>RSS_MIB:
raise RuntimeError('frozen aggregate resource cap exceeded')
save()
try:
from CelerisLab.cuda.context import CudaContext
context=CudaContext(0)
if context.name!=reference['device'] or context.sm_arch!=reference['sm_arch']:
raise RuntimeError('matched reference device identity differs')
report.update(device=context.name,sm_arch=context.sm_arch)
for name,n,refined in CASES:
cap(); ledger=None; case_started=time.monotonic(); cfg=old.LBMConfig(nx=n,ny=n,viscosity=.0003*n,velocity=0.,ddf_shifting=True)
case=dict(name=name,setup=setup(n,refined),completed=False,phase='setup',rows=[],early_pressure=[],prefixes=[],health_counts=[0]*6,
ledger_scope='all intervals; fine/transfer components are mathematically absent for uniform runs')
report['runs'].append(case); save()
run=gpu.SpaceTimeGPU(cfg,tuple(case['setup']['bounds']),context=context,allow_n128=True) if refined else gpu.UniformGPU(cfg,context=context)
case.update(build_metadata=run.build_metadata,device_allocations_bytes=run.device_bytes,
device_memory_scope='owner allocations; excludes driver/context/compiler')
expected_modules=None
if not refined and n in (64,128): expected_modules=[reference['build_metadata'][0 if n==64 else 1]]
if refined: expected_modules=[report['runs'][1]['build_metadata'][0],report['runs'][2]['build_metadata'][0]]
if expected_modules is not None:
if run.build_metadata!=expected_modules: raise RuntimeError('matched source/config/PTX metadata differs')
case['matched_build_metadata']=True
if any(m['flags']!=reference['build_metadata'][0]['flags'] for m in run.build_metadata):
raise RuntimeError('compiler flags differ from reference')
if refined:
case['tables']=old.table_guard.validate_tables(run.plan,run.tables,allow_n128=True)
masks=(run.plan.maps.coarse_active,run.plan.maps.fine_active)
else: masks=(np.ones((n,n),bool),)
analytic=tuple(tg.initial(m.shape,l,n,cfg.viscosity) for l,m in enumerate(masks))
initial=ref.cpu.State(*analytic) if refined else analytic[0]
physical=lambda:populations(run,refined)
stored=lambda:populations(run,refined,raw=True)
run.initialize(initial); raw0=stored(); direction0=directional(raw0,masks)
case['phase']='warm_reset'; warm_started=time.monotonic()
for _ in range(32): cap(); run.step()
case['warmup_seconds']=time.monotonic()-warm_started
run.initialize(initial)
for a,b in zip(raw0,stored()): np.testing.assert_array_equal(a.view(np.uint32),b.view(np.uint32))
if run.step_count or run.health_counts.any(): raise AssertionError('warm reset failed')
case['raw_reset_verified']=True; arrays=physical()
first=tg.metrics(arrays,masks,0,n,cfg.viscosity); mass0=first['mass']; moment0=ref.cpu.MOMENTS.T@directional(arrays,masks)
case['initial']=dict(mass=mass0,momentum=moment0[1:].tolist(),minimum_population=min(float(a[:,m].min()) for a,m in zip(arrays,masks)))
case['initial_quantization']=[float(abs(a[:,m]-b[:,m]).max()) for a,b,m in zip(arrays,analytic,masks)]
case['early_pressure'].append(pressure_projection(arrays,masks,0,n,cfg.viscosity))
ledger=old.LedgerTotals(); case['phase']='trajectory'; trajectory_started=time.monotonic()
case['all_interval_mass']=dict(maximum_relative=0.,step=0,signed_change_at_maximum=0.,limit=LIMITS['mass_relative'],passed=True)
for step in range(1,16*n+1):
cap(); measured=run.step(); ledger.add(measured)
if run.step_count!=step: raise AssertionError('interval count differs')
mass_change=float(ledger.sums[0].sum()); drift=abs(mass_change)/mass0
if drift>case['all_interval_mass']['maximum_relative']:
case['all_interval_mass'].update(maximum_relative=drift,step=step,signed_change_at_maximum=mass_change,passed=drift<=LIMITS['mass_relative'])
if step in case['setup']['early_steps'] or step in case['setup']['sample_steps']:
arrays=physical()
if step in case['setup']['early_steps']:
case['early_pressure'].append(pressure_projection(arrays,masks,step,n,cfg.viscosity)); save()
if step in case['setup']['sample_steps']:
row=tg.metrics(arrays,masks,step,n,cfg.viscosity,mass0)
row.update(step=step,momentum_change=(ref.cpu.MOMENTS.T@directional(arrays,masks)-moment0)[1:].tolist(),
mass_change=row['mass']-mass0,minimum_population=min(float(a[:,m].min()) for a,m in zip(arrays,masks)))
if not all(np.isfinite(v).all() for v in row.values()): raise FloatingPointError('nonfinite sampled metric')
case['rows'].append(row); prefix=ledger.snapshot(directional(stored(),masks)-direction0)
prefix['physical_minus_raw_endpoint_moments']=(ref.cpu.MOMENTS.T@directional(arrays,masks)-moment0-ref.cpu.MOMENTS.T@np.array(prefix['endpoint_directional_change'])).tolist()
case['prefixes'].append(prefix)
if not prefix['closure_pass'] or not prefix['component_split_pass']: raise AssertionError('cumulative budget failed')
save(); print(json.dumps(dict(case=name,**row),allow_nan=False),flush=True)
case.update(assess(case['rows'],run.health_counts))
cap(); case.update(completed=True,phase='completed',health_counts=run.health_counts.tolist(),audited_intervals=ledger.steps,
ledger=ledger.snapshot(),trajectory_seconds=time.monotonic()-trajectory_started,seconds=time.monotonic()-case_started)
case['passed']=case['campaign_pass'] and case['all_interval_mass']['passed'] and ledger.steps==16*n and len(case['early_pressure'])==65
run.close(); run=None; save()
cap(); report.update(completed=True,passed=all(c['passed'] for c in report['runs']),comparisons=comparisons(report['runs'],reference))
except Exception as exc:
report['failure']=repr(exc)
raise
finally:
if run is not None and ledger is not None:
current=report['runs'][-1]
current['failure_endpoint_status']='unavailable_owner_not_ready'
if run.status=='ready':
try:
current['failure_endpoint']=ledger.snapshot(directional(populations(run,refined,raw=True),masks)-direction0)
current['failure_endpoint_status']='measured'
except Exception as exc: current['failure_endpoint_status']=repr(exc)
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())
@@ -0,0 +1,392 @@
"""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 = 'afea11ea4a1c2c3800b2f6a7a8e81f5568989fc9e1cc02fb52d7e1ea3864e003'
AUDIT_SHA = 'fba340b1533bec308b666ee84666df0ac9d08e2872be31ee379e2d1447565158'
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 differing_fields(actual, expected):
"""Compare serialized sample numerics exactly, including the sign of zero."""
if actual.keys() != expected.keys():
return ['field_set']
return [key for key in actual if np.asarray(actual[key]).dtype != np.asarray(expected[key]).dtype
or np.asarray(actual[key]).shape != np.asarray(expected[key]).shape
or np.asarray(actual[key]).tobytes() != np.asarray(expected[key]).tobytes()]
def sample_difference(actual, expected):
"""Descriptive comparison only; never an equivalence gate for different codecs."""
return {key:dict(reference=value,signed=(np.asarray(actual[key])-value).tolist(),
absolute=np.abs(np.asarray(actual[key])-value).tolist()) for key,value in expected.items()}
class LedgerTotals:
"""Compensated directional sums; retain absolute step sums to expose cancellation."""
names = ('net_change','physical_collision','coarse_collision','fine0_collision','fine1_collision',
'child_rounding','child_interpolation','child_cast','register_rounding','receiver_rounding','unexplained')
def __init__(self):
self.sums = np.zeros((len(self.names),9)); self.correction = self.sums.copy()
self.absolute = self.sums.copy(); self.absolute_moments = np.zeros((len(self.names),3))
self.maxima = self.sums.copy(); self.max_steps = np.zeros(self.sums.shape,np.int64)
self.cells = np.zeros((3,4)); self.cell_correction = self.cells.copy()
self.steps = 0; self.bound = 0.; self.maximum = 0.
def add(self, ledger):
values = np.array([ledger[name] for name in self.names],np.float64)
cells = np.asarray(ledger['collision_cell_mass'],np.float64)
if (values.shape != self.sums.shape or not np.isfinite(values).all()
or cells.shape != (3,4) or not np.isfinite(cells).all()
or not np.isfinite(ledger['unexplained_bound']) or ledger['unexplained_bound'] <= 0):
raise AssertionError('invalid diagnostic ledger')
delta = values-self.correction
updated = self.sums+delta
self.correction = (updated-self.sums)-delta
self.sums = updated
self.absolute += abs(values)
self.absolute_moments += abs(values @ ref.cpu.MOMENTS)
larger = abs(values) > self.maxima
self.max_steps[larger] = self.steps+1; self.maxima = np.maximum(self.maxima,abs(values))
delta = cells-self.cell_correction
updated = self.cells+delta
self.cell_correction = (updated-self.cells)-delta; self.cells = updated
self.steps += 1; self.bound += ledger['unexplained_bound']
self.maximum = max(self.maximum,float(abs(values[-1]).max()))
def snapshot(self, endpoint_change=None):
values = dict(zip(self.names,self.sums))
accounted = sum(values[key] for key in ('physical_collision','child_rounding','register_rounding','receiver_rounding'))
closure = None if endpoint_change is None else endpoint_change-accounted
telescoping = None if endpoint_change is None else endpoint_change-values['net_change']
closure_pass = None if closure is None else max(float(abs(closure).max()),float(abs(telescoping).max())) <= self.bound
collision_split = values['physical_collision']-sum(values[key] for key in ('coarse_collision','fine0_collision','fine1_collision'))
child_split = values['child_rounding']-values['child_interpolation']-values['child_cast']
split_pass = max(float(abs(collision_split).max()),float(abs(child_split).max())) <= self.bound
return dict(intervals=self.steps, summed_per_step_bound=self.bound, max_step_unexplained=self.maximum,
closure_pass=closure_pass,component_split_pass=split_pass,
endpoint_status='not_measured' if endpoint_change is None else 'measured',
endpoint_directional_change=None if endpoint_change is None else endpoint_change.tolist(),
closure_residual=None if closure is None else closure.tolist(),
telescoping_residual=None if telescoping is None else telescoping.tolist(),
closure_mass_momentum=None if closure is None else (ref.cpu.MOMENTS.T@closure).tolist(),
collision_split_residual=collision_split.tolist(),child_split_residual=child_split.tolist(),
collision_cell_mass=dict(levels=['coarse','fine0','fine1'],columns=['net','positive','negative','l1'],
applied_volume_weights=[1.,.25,.25],values=self.cells.tolist()),
components={key:dict(directional=self.sums[i].tolist(),mass_momentum=(self.sums[i]@ref.cpu.MOMENTS).tolist(),
absolute_step_directional_sum=self.absolute[i].tolist(),
absolute_step_mass_momentum_sum=self.absolute_moments[i].tolist(),
max_absolute_interval=self.maxima[i].tolist(),max_absolute_interval_step=self.max_steps[i].tolist())
for i,key in enumerate(self.names)})
def warm_and_reset(run, initial, check_cap, *, audit=False):
"""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=audit)
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, ledger_audit=audit)
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('--audit-against', type=Path, help='audit every interval and compare exact saved 1779cf2 samples')
parser.add_argument('--shifted', action='store_true', help='opt-in shifted storage with all interval ledgers')
parser.add_argument('--unshifted-receipt', type=Path, help='pinned1779 comparison, descriptive only')
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')
if args.shifted != (args.unshifted_receipt is not None) or (args.shifted and args.audit_against):
parser.error('--shifted requires --unshifted-receipt and cannot use exact --audit-against')
exact_repeat = args.audit_against is not None
cpu, uniform = load_references(args.cpu_receipt, args.uniform_receipt)
previous = None
comparison_path = args.audit_against or args.unshifted_receipt
if comparison_path is not None:
if sha(comparison_path) != AUDIT_SHA:
raise ValueError('pinned GPU trajectory receipt differs')
previous = json.loads(comparison_path.read_text())
if previous['source']['numpy'] != np.__version__:
raise RuntimeError('audited repeat requires unchanged NumPy environment')
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='shifted_FP32' if args.shifted else 'unshifted_FP32', collision='SRT', interval_count=1024,
initialization='common analytic FP64 field, encoded using production FP64 weights before FP32 upload' if args.shifted
else '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='pending_measurement' if previous else '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
ledger_totals = LedgerTotals() if previous else None
if previous:
report['diagnostic'] = dict(completed=False, passed=False, reference_sha256=AUDIT_SHA,
reference_source=previous['source']['commit'], sample_checks=[], prefixes=[],
scope='net drift attribution; exact saved diagnostics only, not full-field trajectory certification' if exact_repeat
else 'raw stored-value ledger; unshifted and CPU comparisons descriptive, initial codecs differ')
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
if ledger_totals is not None:
report['audited_intervals'] = ledger_totals.steps
if ledger_totals.steps:
report['boundary_budget_status'] = 'measured' if ledger_totals.steps==1024 else 'measured_prefix'
report['diagnostic']['accumulated'] = ledger_totals.snapshot()
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=args.shifted)
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)
if previous:
for key in ('device','sm_arch','table_sha256') + (('build_metadata',) if exact_repeat else ()):
if report[key] != previous[key]:
raise AssertionError(f'audited repeat identity differs: {key}')
analytic = ref.cpu.State(tg.initial((64,64),0,64,.0192), tg.initial((128,128),1,64,.0192))
initial = analytic
phase = 'warm_reset'
initial, report['warmup'] = warm_and_reset(run, initial, check_cap, audit=previous is not None)
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)))
if exact_repeat and differing_fields(report['initial'],previous['initial']):
raise AssertionError('audited repeat initial state diagnostics differ')
raw_initial = run._download_stored() if args.shifted else initial
direction0 = ref.directional(ref.cpu.State(*(a.astype(np.float64) for a in
(raw_initial.coarse_f,raw_initial.fine_f))),maps)
if args.shifted:
errors=[]; unshifted_errors=[]
for raw,actual,wanted,mask in zip((raw_initial.coarse_f,raw_initial.fine_f),arrays,
(analytic.coarse_f,analytic.fine_f),masks):
encoded=gpu._encode_physical(wanted,True)
np.testing.assert_array_equal(raw[:,mask].view(np.uint32),encoded[:,mask].view(np.uint32))
errors.append(float(abs(actual[:,mask]-wanted[:,mask]).max()))
unshifted_errors.append(float(abs(gpu._encode_physical(wanted,False)[:,mask]-wanted[:,mask]).max()))
report['warmup']['raw_reset_verified']=True
report['initial_quantization']=dict(maximum_population_error_by_level=errors,
unshifted_codec_error_by_level=unshifted_errors,
moment_change_vs_analytic=(moment0-total(analytic,maps)).tolist(),
cpu_start='pinned analytic FP64 trajectory, not reinitialized to decoded GPU upload',
matched_cpu_initial_bits=False)
report['unshifted_comparison']=dict(status='descriptive_only_codecs_have_different_initial_bits',
reference_sha256=AUDIT_SHA,initial=sample_difference(report['initial'],previous['initial']),rows=[])
save()
check_cap()
phase = 'trajectory'
trajectory_started = time.monotonic()
for step in range(1,1025):
check_cap()
ledger = run.step(audit=previous is not None)
if ledger_totals is not None:
ledger_totals.add(ledger)
report['max_directional_residual'] = ledger_totals.maximum
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)))
if ledger_totals is not None:
row['minimum_boundary_pre'] = ledger['minimum_pre']
row['minimum_boundary_stored'] = ledger['minimum_stored']
else:
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)
if previous:
expected=previous['rows'][len(report['rows'])-1]
changed = differing_fields(row,expected) if exact_repeat else []
report['diagnostic']['sample_checks'].append(dict(step=step,
status='exact_repeat' if exact_repeat else 'descriptive_only',differing_fields=changed if exact_repeat else None))
if args.shifted:
report['unshifted_comparison']['rows'].append(dict(step=step,fields=sample_difference(row,expected)))
raw_state=run._download_stored() if args.shifted else state
endpoint = ref.directional(ref.cpu.State(*(a.astype(np.float64) for a in
(raw_state.coarse_f,raw_state.fine_f))),maps)-direction0
prefix = ledger_totals.snapshot(endpoint)
if args.shifted:
prefix['physical_minus_raw_endpoint_moments'] = (
total(state,maps)-moment0-ref.cpu.MOMENTS.T@endpoint).tolist()
report['diagnostic']['prefixes'].append(prefix)
if not prefix['closure_pass'] or not prefix['component_split_pass']:
raise AssertionError('cumulative ledger does not close within summed existing per-step bounds')
if changed:
raise AssertionError(f'audited repeat sample differs: {changed}')
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))
if args.shifted:
report['historical_pressure_gate'].update(status='historical_shifted_FP32_uniform_vs_shifted_FP32_candidate',
matched_storage=True,matched_control=False)
report['completed'] = True
report['passed'] = report['campaign_pass'] and report['experimental_pass']
if previous:
report['diagnostic'].update(completed=True,passed=ledger_totals.steps==1024 and len(report['rows'])==4)
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())
+165
View File
@@ -0,0 +1,165 @@
// Test-only arrival-f space-time coupling. No production scheduler changes.
#include <stdint.h>
#include <math.h>
#include "config.h"
#include "core/precision.cuh"
#include "core/descriptors.cuh"
#include "core/layout.cuh"
#include "operators/macro.cuh"
#include "operators/collision_srt.cuh"
#include "streaming/pull_double_buffer.cuh"
#if DIM != 2 || NQ != 9 || COLLISION_MODEL != 0 || STORE_PRECISION != 0 || USE_LES || STREAMING_MODEL != 0
#error "Space-time experiment requires FP32 D2Q9 SRT double buffers without LES"
#endif
__device__ __forceinline__ void collide(float* f, float omega) {
float rho, ux, uy, feq[NQ];
compute_rho_u(f, rho, ux, uy);
#if USE_DDF_SHIFTING
// Same perturbation-preserving SRT branch as production BlockStep.
const float drho = f[0]+f[1]+f[2]+f[3]+f[4]+f[5]+f[6]+f[7]+f[8];
const float jx = f[1]-f[2]+f[5]-f[6]+f[7]-f[8];
const float jy = f[3]-f[4]+f[5]-f[6]-f[7]+f[8];
compute_feq_from_momentum(drho, jx, jy, ux, uy, feq);
#else
compute_feq(rho, ux, uy, feq);
#endif
collide_srt_no_force(f, feq, omega);
}
__device__ __forceinline__ double physical(double value, unsigned int q) {
#if USE_DDF_SHIFTING
// Match population_storage.py's FP64 weights, not promoted FP32 d_w.
return value + (q == 0 ? 4.0/9.0 : (q < 5 ? 1.0/9.0 : 1.0/36.0));
#else
return value;
#endif
}
__device__ __forceinline__ void health(float stored, unsigned int q, unsigned int* counts) {
const double value = physical((double)stored, q);
if (!isfinite(value)) atomicAdd(counts + 4, 1u);
if (value <= 0.0f) atomicAdd(counts + 5, 1u);
}
extern "C" __global__ void CoarseStart(
const float* coarse, const float* fine, float* out, float omega,
unsigned int x0, unsigned int y0, unsigned int x1, unsigned int y1,
unsigned int* counts)
{
unsigned int x, y; unsigned long k;
index_from_thread(x, y, k);
if (x >= NX || y >= NY) return;
const bool covered = x >= x0 && x < x1 && y >= y0 && y < y1;
float f[NQ];
for (unsigned int q = 0; q < NQ; ++q) {
const unsigned long s = q * 4ul * TOTAL_CELLS + 2ul * y * (2u * NX) + 2u * x;
f[q] = covered ? 0.25f * (fine[s] + fine[s+1] + fine[s+2u*NX] + fine[s+2u*NX+1])
: coarse[index_f(k, q)];
}
collide(f, omega);
for (int q = 0; q < NQ; ++q) if (!covered) health(f[q], q, counts);
stream_pull_store(k, f, out);
}
extern "C" __global__ void Collide(
const float* in, float* out, const unsigned char* active, float omega,
unsigned int* counts)
{
unsigned int x, y; unsigned long k;
index_from_thread(x, y, k);
if (x >= NX || y >= NY) return;
float f[NQ];
if (active == nullptr || active[k]) {
for (int q = 0; q < NQ; ++q) f[q] = in[index_f(k, q)];
collide(f, omega);
if (active != nullptr) for (int q = 0; q < NQ; ++q) health(f[q], q, counts);
} else {
for (int q = 0; q < NQ; ++q) f[q] = NAN;
}
stream_pull_store(k, f, out);
}
extern "C" __global__ void Pull(
const float* in, float* out, const unsigned char* active, unsigned int* counts)
{
unsigned int x, y; unsigned long k;
index_from_thread(x, y, k);
if (x >= NX || y >= NY) return;
float f[NQ];
if (active == nullptr || active[k]) {
unsigned long j[NQ];
compute_neighbors(x, y, j);
stream_pull_load(k, f, in, j);
if (active != nullptr) for (int q = 0; q < NQ; ++q) health(f[q], q, counts);
} else {
for (int q = 0; q < NQ; ++q) f[q] = NAN;
}
stream_pull_store(k, f, out);
}
extern "C" __global__ void Interpolate(
const float* g0, const float* gp, const uint32_t* samples,
const uint32_t* origins, const int* sizes, const double* weights,
double* pre, float* h, unsigned int groups, unsigned int frozen,
unsigned int* counts)
{
unsigned int group = blockIdx.x * blockDim.x + threadIdx.x;
if (group >= groups) return;
const int size = sizes[group];
const unsigned int q = origins[group] / TOTAL_CELLS;
for (int child = 0; child < 4; ++child) {
double value = frozen ? (double)g0[origins[group]] : 0.0;
if (!frozen) for (int j = 0; j < size; ++j) {
const float* source = j < size/2 ? g0 : gp;
value += weights[(4u*group+child)*10u+j] * (double)source[samples[10u*group+j]];
}
const float stored = __double2float_rn(value);
pre[4u*group+child] = value;
h[4u*group+child] = stored;
if (!isfinite(physical(value, q))) atomicAdd(counts, 1u);
if (physical(value, q) <= 0.0) atomicAdd(counts+1, 1u);
if (!isfinite(physical((double)stored, q))) atomicAdd(counts+2, 1u);
if (physical((double)stored, q) <= 0.0) atomicAdd(counts+3, 1u);
}
}
extern "C" __global__ void Inject(
float* fine, const float* h, const uint32_t* rows, unsigned int count,
unsigned int omit_first)
{
unsigned int row = blockIdx.x * blockDim.x + threadIdx.x;
if (row >= count || (omit_first && row == 0)) return;
fine[rows[2u*row]] = h[rows[2u*row+1]];
}
extern "C" __global__ void Register(
const float* predicted, const float* fine, const float* h,
const uint32_t* targets, const uint32_t* sources, const int* times,
double* registers, unsigned int count, int substep, int mutation)
{
unsigned int row = blockIdx.x * blockDim.x + threadIdx.x;
if (row >= count) return;
double value = substep == 0 ? -(double)predicted[targets[row]] : registers[row];
for (int j = 0; j < 4; ++j) {
const unsigned int slot = 4u*row+j;
const int when = times[slot];
if (when != substep && !(substep == 0 && when == -1)) continue;
if ((mutation == 1 && when == -1) || (mutation == 2 && when == 0)
|| (mutation == 3 && when == 1)) continue;
value += 0.25 * (double)(when == -1 ? h[sources[slot]] : fine[sources[slot]]);
}
registers[row] = value;
}
extern "C" __global__ void ApplyRegisters(
float* predicted, const double* registers, const uint32_t* targets,
unsigned int count, unsigned int* counts)
{
unsigned int row = blockIdx.x * blockDim.x + threadIdx.x;
if (row >= count) return;
const unsigned int slot = targets[row];
predicted[slot] = __double2float_rn((double)predicted[slot] + registers[row]);
health(predicted[slot], slot / TOTAL_CELLS, counts);
}
+472
View File
@@ -0,0 +1,472 @@
"""Test-only arrival-f GPU owner; CUDA compilation/execution requires admission."""
from dataclasses import replace
import hashlib
from numbers import Integral
from pathlib import Path
import numpy as np
import spacetime_reference as ref
from CelerisLab.lbm.population_storage import _encode_physical, _decode_stored, _WEIGHTS
PINS = {"spacetime_reference.py": "1fe13242b9a41062a1227db4a6f9b695165aefab7f4c0b3b2adff0c9023552bf",
"cpu_reference.py": "bba2a06a9f6517b32079ffff15e4060d8c3b7ed35f16ca134e1877226c9e4e1e",
"../taylor_green/oracle.py": "83e4d7425d45538278e9fb181af450a36358bd9a5ed6be57ff611dd9ae5eb35b"}
def check_sources():
for name, expected in PINS.items():
if hashlib.sha256((Path(__file__).parent/name).read_bytes()).hexdigest() != expected:
raise RuntimeError(f"reference source changed: {name}")
def validate_config(config, *, uniform=False, allow_n128=False):
if type(uniform) is not bool or type(allow_n128) is not bool:
raise ValueError("admission flags must be bool")
cfg = replace(config)
cfg.validate()
if (cfg.nx != cfg.ny or not 8 <= cfg.nx <= (256 if uniform else (128 if allow_n128 else 64)) or cfg.nz != 1
or cfg.collision != "SRT" or not cfg.is_d2q9
or cfg.streaming != "double_buffer" or cfg.store_precision != "FP32"
or cfg.data_type != "FP32" or type(cfg.ddf_shifting) is not bool or cfg.les_enabled
or cfg.velocity != 0 or cfg.wall_slot_width or cfg.wall_normal_velocity):
raise ValueError("experiment requires periodic unforced square FP32 D2Q9/SRT within admitted size")
if uniform:
return (cfg,)
fine = replace(cfg, nx=2*cfg.nx, ny=2*cfg.ny, viscosity=2*cfg.viscosity)
fine.validate()
return cfg, fine
def pack_plan(plan):
"""Pack reviewed coordinate events; no production transfer builder imports."""
n, nf = plan.n, plan.nf
groups, receivers = plan.groups, plan.receivers
data = dict(samples=np.zeros((len(groups), 10), np.uint32),
origins=np.empty(len(groups), np.uint32), sizes=np.empty(len(groups), np.int32),
weights=np.zeros((len(groups), 4, 10), np.float64),
targets=(receivers[:, 0]*n*n+receivers[:, 1]).astype(np.uint32))
lookup = {tuple(row): i for i, row in enumerate(receivers)}
packets = [[] for _ in receivers]; incoming = [[], []]
for i, g in enumerate(groups):
size = len(g.samples)
if size not in (6, 10):
raise ValueError("unexpected interpolation stencil")
data['samples'][i, :size] = g.q*n*n+g.samples
data['origins'][i] = g.q*n*n+g.source
data['sizes'][i] = size
data['weights'][i, :, :size] = g.weights
for k, event in enumerate(g.events):
if event is None:
packets[lookup[(g.q, g.receiver)]].append((-1, 4*i+k))
else:
m, slot = event
incoming[m].append((g.q*nf*nf+slot, 4*i+k))
for m, outgoing in enumerate(plan.outgoing):
for q, source, target in outgoing:
packets[lookup[(int(q), int(target))]].append((m, int(q*nf*nf+source)))
if any(len(row) != 4 or len(set(row)) != 4 for row in packets):
raise ValueError("each receiver needs four distinct quarter-packets")
if len({packet for row in packets for packet in row}) != 4*len(receivers):
raise ValueError("packet has multiple receiver owners")
data['times'] = np.array([[t for t, _ in row] for row in packets], np.int32)
data['sources'] = np.array([[s for _, s in row] for row in packets], np.uint32)
for m, rows in enumerate(incoming):
data[f'incoming{m}'] = np.array(rows, np.uint32)
if len({slot for slot, _ in rows}) != len(rows):
raise ValueError("duplicate incoming writer")
return {key: np.ascontiguousarray(value) for key, value in data.items()}
def table_hash(data):
digest = hashlib.sha256()
for key, value in sorted(data.items()):
digest.update(f"{key}:{value.dtype}:{value.shape}".encode()+value.tobytes())
return digest.hexdigest()
def rounding_ledger(plan, tables, before, after, g0, predicted, fine_phases, pre, h, registers, *, shifted=False):
"""Audit actual stored FP32 phases; never call a missing packet 'rounding'."""
maps, n = plan.maps, plan.n
f64 = lambda a: np.asarray(a, dtype=np.float64)
difference = f64(g0[:, maps.coarse_active])-f64(before.coarse_f[:, maps.coarse_active])
coarse_collision = difference.sum(axis=1)
cell_mass = [difference.sum(axis=0)]
collision = coarse_collision.copy(); fine_collision = []
for f, g in fine_phases:
difference = f64(g[:, maps.fine_active])-f64(f[:, maps.fine_active])
fine_collision.append(.25*difference.sum(axis=1))
cell_mass.append(.25*difference.sum(axis=0))
collision += fine_collision[-1]
np.testing.assert_array_equal(h, pre.astype(np.float32))
origins = f64(g0).ravel()[tables['origins']]
np.testing.assert_allclose(pre.mean(axis=1), origins, rtol=0, atol=2e-14)
group = np.bincount(tables['origins']//(n*n), weights=f64(h).mean(axis=1)-origins, minlength=9)
interpolation = np.bincount(tables['origins']//(n*n), weights=pre.mean(axis=1)-origins, minlength=9)
child_cast = np.bincount(tables['origins']//(n*n), weights=f64(h).mean(axis=1)-pre.mean(axis=1), minlength=9)
packet = np.empty(tables['times'].shape, np.float64)
for when, source in ((-1, h), (0, fine_phases[0][1]), (1, fine_phases[1][1])):
mask = tables['times'] == when
packet[mask] = f64(source).ravel()[tables['sources'][mask]]
p = f64(predicted).ravel()[tables['targets']]
ideal = -p + .25*packet.sum(axis=1)
reg_error = registers-ideal
bound = 16*np.finfo(float).eps*(abs(p)+.25*abs(packet).sum(axis=1))
if np.any(abs(reg_error) > bound):
raise AssertionError("receiver packet register mismatch (not roundoff)")
actual = after.coarse_f.ravel()[tables['targets']]
np.testing.assert_array_equal(actual, (p+registers).astype(np.float32))
q = tables['targets']//(n*n)
store_error = np.bincount(q, weights=f64(actual)-(p+registers), minlength=9)
register_error = np.bincount(q, weights=reg_error, minlength=9)
# directional() must sum converted arrays, not accumulate in FP32.
total = lambda s: ref.directional(ref.cpu.State(f64(s.coarse_f), f64(s.fine_f)), maps)
net = total(after)-total(before)
raw = net-collision
unexplained = raw-group-store_error-register_error
limit = float(64*np.finfo(float).eps*(total(before).sum()+int(shifted)*n*n))
np.testing.assert_allclose(unexplained, 0, rtol=0, atol=limit)
offset = _WEIGHTS.ravel()[tables['origins']//(n*n),None] if shifted else 0.
return dict(raw=raw.tolist(), net_change=net.tolist(), physical_collision=collision.tolist(),
coarse_collision=coarse_collision.tolist(), fine0_collision=fine_collision[0].tolist(),
fine1_collision=fine_collision[1].tolist(), child_interpolation=interpolation.tolist(),
child_cast=child_cast.tolist(),
collision_cell_mass=[[float(a.sum()),float(a[a>0].sum()),float(a[a<0].sum()),float(abs(a).sum())]
for a in cell_mass],
child_rounding=group.tolist(), receiver_rounding=store_error.tolist(),
register_rounding=register_error.tolist(), unexplained=unexplained.tolist(),
unexplained_bound=limit, minimum_pre=float((pre+offset).min()),
minimum_stored=float((f64(h)+offset).min()))
class SpaceTimeGPU:
"""Caller-owned context, one stream, explicit synchronized arrival state."""
def __init__(self, config, bounds, *, context, mode='spacetime', identity=False, allow_n128=False):
check_sources()
self.configs = validate_config(config,allow_n128=allow_n128)
if mode not in ('spacetime', 'frozen') or type(identity) is not bool:
raise ValueError("unsupported interpolation or debug collision mode")
self.bounds = tuple(bounds)
if len(self.bounds) != 4 or any(isinstance(v, bool) or not isinstance(v, Integral) for v in self.bounds):
raise ValueError("rectangle bounds must be four integers")
self.plan = ref.Plan(config.nx, self.bounds)
self.tables = pack_plan(self.plan)
self.ctx, self.mode, self.identity = context, mode, identity
self.allocations, self.modules, self.build_metadata = [], [], []
self.status, self.step_count, self.device_bytes = 'new', 0, 0
self.health_counts = np.zeros(6, np.uint64)
import pycuda.driver as cuda
from CelerisLab.cuda import compiler_v2 as compiler
root = Path(__file__).resolve().parents[3]
if Path(compiler._KERNEL_DIR).resolve() != root/'src/CelerisLab/lbm/kernels':
raise RuntimeError("wrong production kernel checkout")
self.cuda = cuda
with context.activate():
self.stream = cuda.Stream()
try:
source = Path(__file__).with_suffix('.cu')
for cfg in self.configs:
build = compiler.generate_config(cfg)
try:
Path(build, source.name).write_bytes(source.read_bytes())
ptx = compiler.compile_kernel(context.sm_arch, source=source.name, build_dir=build)
self.modules.append(compiler.load_module(ptx))
import json
metadata = json.loads(Path(ptx+'.json').read_text())
metadata['test_entry_sha256'] = hashlib.sha256(source.read_bytes()).hexdigest()
self.build_metadata.append(metadata)
finally:
compiler.cleanup_build_dir(build)
self.device_tables = {key: self.upload(value) for key, value in self.tables.items()}
self.active = self.upload(self.plan.maps.fine_active.astype(np.uint8))
self.coarse = [self.allocate(9*config.nx**2*4) for _ in range(4)]
self.fine = [self.allocate(9*(2*config.nx)**2*4) for _ in range(3)]
self.pre = self.allocate(len(self.plan.groups)*4*8)
self.h = self.allocate(len(self.plan.groups)*4*4)
self.registers = self.allocate(len(self.plan.receivers)*8)
self.counts = self.allocate(6*4)
except BaseException:
try:
self.close()
except Exception:
pass
raise
def allocate(self, size):
pointer = self.cuda.mem_alloc(size)
self.allocations.append(pointer); self.device_bytes += size
return pointer
def upload(self, array):
pointer = self.allocate(array.nbytes)
self.cuda.memcpy_htod(pointer, array)
return pointer
def read(self, pointer, shape, dtype=np.float32):
self.stream.synchronize()
out = np.empty(shape, dtype)
self.cuda.memcpy_dtoh(out, pointer)
return out
def launch(self, level, name, *args, count=None):
cfg = self.configs[level]
nt = cfg.threads_per_block
grid = ((cfg.nx+nt-1)//nt, cfg.ny) if count is None else ((count+nt-1)//nt, 1)
self.modules[level].get_function(name)(*args, block=(nt,1,1), grid=grid, stream=self.stream)
def initialize(self, state):
if (self.status not in ('new', 'ready') or state.phase != 'arrival_f'
or isinstance(state.tick, bool) or not isinstance(state.tick, Integral) or state.tick != 0):
raise ValueError("initial synchronized arrival state with tick=0 required")
shifted = self.configs[0].ddf_shifting
arrays = []
for values, mask in zip((state.coarse_f,state.fine_f),
(self.plan.maps.coarse_active,self.plan.maps.fine_active)):
if np.shape(values) != (9,*mask.shape):
raise ValueError("positive finite leaf populations and exact shapes required")
a = _encode_physical(values,shifted)
physical = _decode_stored(a,shifted)
if not np.isfinite(physical[:,mask]).all() or (physical[:,mask] <= 0).any():
raise ValueError("positive finite leaf populations and exact shapes required")
a[:,~mask] = np.nan
arrays.append(a)
with self.ctx.activate():
try:
self.cuda.memcpy_htod(self.coarse[0], arrays[0])
self.cuda.memcpy_htod(self.fine[0], arrays[1])
except BaseException:
self.status = 'failed'
raise
self.status, self.step_count = 'ready', 0
self.health_counts.fill(0)
def _download_stored(self):
"""Private audit/reset snapshot; values remain in their storage representation."""
if self.status != 'ready':
raise RuntimeError(f"GPU owner is {self.status}")
try:
with self.ctx.activate():
arrays = [self.read(self.coarse[0], (9,self.plan.n,self.plan.n)),
self.read(self.fine[0], (9,self.plan.nf,self.plan.nf))]
except BaseException:
self.status = 'failed'
raise
for a, mask in zip(arrays, (self.plan.maps.coarse_active,self.plan.maps.fine_active)):
a[:,~mask] = np.nan
return ref.cpu.State(*arrays, tick=2*self.step_count)
def download(self):
"""Physical arrival populations; shifted values are reconstructed in FP64."""
stored = self._download_stored()
return ref.cpu.State(*(_decode_stored(a,self.configs[0].ddf_shifting)
for a in (stored.coarse_f,stored.fine_f)),tick=stored.tick)
def step(self, *, mutation=0, audit=True):
"""One interval; audit=False skips only host ledger snapshots, never health checks."""
if (self.status != 'ready' or type(mutation) is not int or mutation not in range(5)
or type(audit) is not bool or (mutation and not audit)):
raise RuntimeError("ready owner and known test mutation required")
before = self._download_stored() if audit else None
n, nf, t = self.plan.n, self.plan.nf, self.device_tables
ng, nr = len(self.plan.groups), len(self.plan.receivers)
u32, null = np.uint32, np.uintp(0)
omega = [np.float32(0 if self.identity else cfg.omega) for cfg in self.configs]
with self.ctx.activate():
try:
self.status = 'advancing'
self.cuda.memset_d32_async(self.counts, 0, 6, self.stream)
fc, g0, predicted, gp = self.coarse
self.launch(0, 'CoarseStart', fc,self.fine[0],g0,omega[0],
*(u32(v) for v in self.bounds),self.counts)
self.launch(0, 'Pull', g0,predicted,null,self.counts)
self.launch(0, 'Collide', predicted,gp,null,omega[0],self.counts)
self.launch(0, 'Interpolate', g0,gp,t['samples'],t['origins'],t['sizes'],t['weights'],
self.pre,self.h,u32(ng),u32(self.mode=='frozen'),self.counts,count=ng)
if audit:
coarse_g = self.read(g0,(9,n,n)); prediction = self.read(predicted,(9,n,n))
fine_phases = []
for m in (0,1):
f, out, g = self.fine
if audit:
old_f = self.read(f,(9,nf,nf))
self.launch(1, 'Collide', f,g,self.active,omega[1],self.counts)
if audit:
fine_phases.append((old_f,self.read(g,(9,nf,nf))))
self.launch(0, 'Register', predicted,g,self.h,t['targets'],t['sources'],t['times'],
self.registers,u32(nr),np.int32(m),np.int32(mutation),count=nr)
rows = self.tables[f'incoming{m}']
self.launch(1, 'Inject', g,self.h,t[f'incoming{m}'],u32(len(rows)),u32(mutation==4),count=len(rows))
self.launch(1, 'Pull', g,out,self.active,self.counts)
self.fine[0], self.fine[1] = out, f
self.launch(0, 'ApplyRegisters', predicted,self.registers,t['targets'],u32(nr),self.counts,count=nr)
counts = self.read(self.counts,(6,),np.uint32)
self.health_counts += counts.astype(np.uint64)
if counts.any():
raise AssertionError(f"boundary/leaf health failed: {counts.tolist()}")
ledger = dict(boundary_budget_status='not_measured',unexplained=None)
if audit:
coarse_after = self.read(predicted,(9,n,n))
coarse_after[:,~self.plan.maps.coarse_active] = np.nan
after = ref.cpu.State(coarse_after,self.read(self.fine[0],(9,nf,nf)),tick=2*(self.step_count+1))
ledger = rounding_ledger(self.plan,self.tables,before,after,coarse_g,prediction,fine_phases,
self.read(self.pre,(ng,4),np.float64),self.read(self.h,(ng,4)),
self.read(self.registers,(nr,),np.float64),
shifted=self.configs[0].ddf_shifting)
ledger['boundary_budget_status'] = 'measured'
ledger['health_counts'] = counts.tolist()
self.coarse[0], self.coarse[2] = predicted, fc
self.step_count += 1
self.status = 'ready'
return ledger
except BaseException:
self.status = 'failed'
raise
def close(self):
if self.status == 'closed':
return
with self.ctx.activate():
error = None
try:
self.stream.synchronize()
except Exception as exc:
error = exc
failed = []
for pointer in self.allocations:
try:
pointer.free()
except Exception as exc:
error = error or exc
failed.append(pointer)
self.allocations = failed
self.status = 'cleanup_failed' if failed else 'closed'
if not failed:
self.modules.clear()
if error:
raise error
def __enter__(self):
return self
def __exit__(self, kind, value, traceback):
try:
self.close()
except Exception:
if kind is None:
raise
def uniform_ledger(before, after, collided, *, shifted):
"""Periodic streaming permutes each direction; only collision changes its sum."""
a,b,g = [np.asarray(x,np.float64) for x in (before,after,collided)]
collision=(g-a).sum(axis=(1,2)); net=b.sum(axis=(1,2))-a.sum(axis=(1,2))
residual=net-collision
bound=float(64*np.finfo(float).eps*(a.sum()+int(shifted)*a.shape[1]*a.shape[2]))
np.testing.assert_allclose(residual,0,rtol=0,atol=bound)
cells=(g-a).sum(axis=0)
out={key:[0.]*9 for key in ('fine0_collision','fine1_collision','child_rounding',
'child_interpolation','child_cast','register_rounding','receiver_rounding')}
out.update(net_change=net.tolist(),physical_collision=collision.tolist(),coarse_collision=collision.tolist(),
unexplained=residual.tolist(),unexplained_bound=bound,
collision_cell_mass=[[float(cells.sum()),float(cells[cells>0].sum()),
float(cells[cells<0].sum()),float(abs(cells).sum())],[0.]*4,[0.]*4],
boundary_budget_status='not_applicable_uniform',inactive_components='fine, child, register, receiver')
return out
class UniformGPU(SpaceTimeGPU):
"""One explicit arrival-F level; reuse the reviewed kernels and owner helpers."""
def __init__(self, config, *, context, identity=False):
check_sources()
self.configs=validate_config(config,uniform=True)
if type(identity) is not bool:
raise ValueError('identity must be bool')
self.ctx,self.identity=context,identity
self.allocations,self.modules,self.build_metadata=[],[],[]
self.status,self.step_count,self.device_bytes='new',0,0
self.health_counts=np.zeros(6,np.uint64)
import pycuda.driver as cuda
from CelerisLab.cuda import compiler_v2 as compiler
if Path(compiler._KERNEL_DIR).resolve()!=Path(__file__).resolve().parents[3]/'src/CelerisLab/lbm/kernels':
raise RuntimeError('wrong production kernel checkout')
self.cuda=cuda
with context.activate():
self.stream=cuda.Stream()
try:
source=Path(__file__).with_suffix('.cu')
build=compiler.generate_config(self.configs[0])
try:
Path(build,source.name).write_bytes(source.read_bytes())
ptx=compiler.compile_kernel(context.sm_arch,source=source.name,build_dir=build)
self.modules.append(compiler.load_module(ptx))
import json
metadata=json.loads(Path(ptx+'.json').read_text())
metadata['test_entry_sha256']=hashlib.sha256(source.read_bytes()).hexdigest()
self.build_metadata.append(metadata)
finally:
compiler.cleanup_build_dir(build)
n=config.nx
self.buffers=[self.allocate(9*n*n*4) for _ in range(3)]
self.active=self.upload(np.ones((n,n),np.uint8))
self.counts=self.allocate(6*4)
except BaseException:
try:
self.close()
except BaseException:
pass
raise
def initialize(self, physical):
cfg=self.configs[0]
if self.status not in ('new','ready') or np.shape(physical)!=(9,cfg.ny,cfg.nx):
raise ValueError('ready owner and exact uniform population shape required')
stored=_encode_physical(physical,cfg.ddf_shifting)
decoded=_decode_stored(stored,cfg.ddf_shifting)
if not np.isfinite(decoded).all() or (decoded<=0).any():
raise ValueError('positive finite physical populations required')
try:
with self.ctx.activate():
self.cuda.memcpy_htod(self.buffers[0],stored)
self.health_counts[:]=0; self.step_count=0; self.status='ready'
except BaseException:
self.status='failed'
raise
def _download_stored(self):
if self.status!='ready':
raise RuntimeError('uniform owner is not ready')
try:
with self.ctx.activate():
n=self.configs[0].nx
return self.read(self.buffers[0],(9,n,n))
except BaseException:
self.status='failed'
raise
def download(self):
return _decode_stored(self._download_stored(),self.configs[0].ddf_shifting)
def step(self, *, audit=True):
if self.status!='ready' or type(audit) is not bool:
raise RuntimeError('ready owner and bool audit required')
cfg=self.configs[0]; shape=(9,cfg.ny,cfg.nx)
before=self._download_stored() if audit else None
with self.ctx.activate():
try:
self.status='advancing'
self.cuda.memset_d32_async(self.counts,0,6,self.stream)
f,g,out=self.buffers
self.launch(0,'Collide',f,g,self.active,np.float32(0 if self.identity else cfg.omega),self.counts)
self.launch(0,'Pull',g,out,self.active,self.counts)
counts=self.read(self.counts,(6,),np.uint32)
self.health_counts+=counts.astype(np.uint64)
if counts.any():
raise AssertionError(f'uniform population health failed: {counts.tolist()}')
ledger=uniform_ledger(before,self.read(out,shape),self.read(g,shape),shifted=cfg.ddf_shifting) if audit else dict(
boundary_budget_status='not_measured',unexplained=None)
ledger['health_counts']=counts.tolist()
self.buffers[0],self.buffers[2]=out,f
self.step_count+=1; self.status='ready'
return ledger
except BaseException:
self.status='failed'
raise
@@ -0,0 +1,334 @@
"""Experimental FP64 BVP coupling: one periodic rectangle, D2Q9, ratio two.
Guzik et al., https://www.osti.gov/servlets/purl/1226215, sections 3.2--3.5.
Coordinate-generated stream events; CPUREF collision only, no production imports.
This is a mathematical prototype, not a certified accuracy implementation.
"""
from dataclasses import dataclass
from functools import lru_cache
import argparse
import hashlib
import json
from pathlib import Path
import resource
import sys
import subprocess
import time
import numpy as np
import cpu_reference as cpu
class Spec(cpu.Spec):
@property
def n_fine(self):
return 2 * self.n_coarse
def geometry(n, bounds):
x0, y0, x1, y1 = bounds
if not (2 <= x0 < x1 <= n - 2 and 2 <= y0 < y1 <= n - 2):
raise ValueError("one interior rectangle with two coarse cells of clearance required")
y, x = np.indices((n, n))
covered = (x >= x0) & (x < x1) & (y >= y0) & (y < y1)
y, x = np.indices((2*n, 2*n))
active = (x >= 2*x0) & (x < 2*x1) & (y >= 2*y0) & (y < 2*y1)
inner = (x >= 2*x0-2) & (x < 2*x1+2) & (y >= 2*y0-2) & (y < 2*y1+2) & ~active
rows, links = [], []
for ky, kx in zip(*np.nonzero(~covered)):
for q, (cy, cx) in enumerate(zip(cpu.CY, cpu.CX)):
if any(active[2*ky+dy-cy, 2*kx+dx-cx]
for dy in (0, 1) for dx in (0, 1)
if 0 <= 2*ky+dy-cy < 2*n and 0 <= 2*kx+dx-cx < 2*n):
rows.append((q, ky, kx))
sy, sx = (ky-cy) % n, (kx-cx) % n
if not covered[sy, sx]:
links.append(cpu.CornerLink(f"{kx}_{ky}_{q}", q, (sx, sy), (kx, ky)))
kind = np.where(active, 0, np.where(inner, 1, 2))
return cpu.Maps(~covered, covered, active, inner, kind == 2, kind,
np.arange(2*n)//2, (np.arange(2*n)+.5)/2,
np.array(rows, dtype=int), tuple(links))
@lru_cache(None)
def interpolation(kind, points):
"""Return data coordinates, evaluation matrix, and constrained coefficient map."""
if kind == "face":
samples = np.array([(s, t) for t in (0., 1.) for s in (-1., 0., 1.)])
a = np.array([[1, s, t, s*t] for s, t in samples])
e = np.array([[1, s, t, s*t] for s, t in points])
b = e.mean(axis=0, keepdims=True)
d = np.zeros((1, 6)); d[0, 1] = 1
else:
cross = ((0, 0), (-1, 0), (1, 0), (0, -1), (0, 1))
samples = np.array([(x, y, t) for t in (0., 1.) for x, y in cross])
a = np.array([[1, x, y, t] for x, y, t in samples])
e = np.array([[1, x, y, t] for x, y, t in points])
b = np.stack((e.mean(axis=0), [0, 0, 0, 1]))
d = np.zeros((2, 10)); d[0, 0] = 1; d[1, 0] = -1; d[1, 5] = 1
kkt = np.block([[a.T @ a, b.T], [b, np.zeros((len(b), len(b)))]])
coeff = np.linalg.solve(kkt, np.vstack((a.T, d)))[:4]
if not np.allclose(b @ coeff, d, rtol=0, atol=2e-14):
raise AssertionError("invalid constrained interpolation matrix")
return samples, a, e, coeff
@dataclass
class Group:
q: int
source: int
receiver: int
kind: str
points: tuple
samples: np.ndarray
weights: np.ndarray
events: tuple # (substep, fine ghost slot), or None for a leftover
class Plan:
def __init__(self, n, bounds):
self.n, self.nf = n, 2*n
self.maps = geometry(n, bounds)
active = self.maps.fine_active
lo, hi = np.array(bounds[1::-1]), np.array(bounds[3:1:-1])
self.groups = []
for iy, ix in zip(*np.nonzero(self.maps.coarse_active)):
origin = np.array((iy, ix))
for q in range(1, 9):
c = np.array((cpu.CY[q], cpu.CX[q]))
starts = [2*origin + (dy, dx) for dy in (0, 1) for dx in (0, 1)]
events = []
for p in starts:
event = None
for m in (0, 1):
old, new = p+m*c, p+(m+1)*c
inside = lambda z: bool(np.all(z >= 0) and np.all(z < 2*n) and active[tuple(z)])
if not inside(old) and inside(new):
event = (m, int(old[0]*2*n+old[1]))
break
events.append(event)
if all(event is None for event in events):
continue
axes = [j for j in range(2) if origin[j] < lo[j] or origin[j] >= hi[j]]
kind = "corner" if len(axes) == 2 else "face"
points = []
for p in starts:
x = (p+.5)/2
times = [(float(lo[j] if c[j] > 0 else hi[j])-x[j])/c[j] for j in axes]
time = max(times)
offset = x+c*time-(origin+.5+c*.5)
points.append((float(offset[1]), float(offset[0]), float(time-.5))
if kind == "corner" else (float(offset[1-axes[0]]), float(time-.5)))
points = tuple(points)
samples, _, e, coeff = interpolation(kind, points)
indices = []
for row in samples:
shift = np.array((row[1], row[0]), dtype=int) if kind == "corner" else np.eye(2, dtype=int)[1-axes[0]]*int(row[0])
pos = (origin+shift) % n
indices.append(int(pos[0]*n+pos[1]))
target = (origin+c) % n
self.groups.append(Group(q, int(iy*n+ix), int(target[0]*n+target[1]), kind,
points, np.array(indices), e @ coeff, tuple(events)))
self.outgoing = []
# m is zero-based: a substep-0 exit travels two fine spacings by
# synchronization; a substep-1 exit travels one.
for m in (0, 1):
records = []
for q in range(1, 9):
c = np.array((cpu.CY[q], cpu.CX[q]))
for y, x in zip(*np.nonzero(active)):
p = np.array((y, x)); endpoint = p+c
if not active[tuple(endpoint)]:
target = (p+(2-m)*c)//2
assert self.maps.coarse_active[tuple(target)]
records.append((q, int(y*2*n+x), int(target[0]*n+target[1])))
self.outgoing.append(np.array(records, dtype=int))
incoming = [(m, g.q, slot) for g in self.groups for event in g.events
if event is not None for m, slot in [event]]
assert len(incoming) == len(set(incoming)), "duplicate incoming writer"
expected = {(m, int(q), int((y-cpu.CY[q])*2*n+x-cpu.CX[q]))
for m in (0, 1) for q in range(1, 9)
for y, x in zip(*np.nonzero(active)) if not active[y-cpu.CY[q], x-cpu.CX[q]]}
assert set(incoming) == expected, "incoming coverage mismatch"
receivers = {(int(q), int(target)) for out in self.outgoing for q, _, target in out}
receivers |= {(g.q, g.receiver) for g in self.groups if None in g.events}
assert receivers == {(int(q), int(y*n+x)) for q, y, x in self.maps.coalesce}
self.receivers = np.array(sorted(receivers), dtype=int)
def directional(state, maps):
return state.coarse_f[:, maps.coarse_active].sum(axis=1) + .25*state.fine_f[:, maps.fine_active].sum(axis=1)
def advance(spec, plan, state, mode="spacetime", mutation=None):
"""One arrival-to-arrival interval. 'frozen' disables space AND time variation."""
if mode not in ("frozen", "spacetime"):
raise ValueError(mode)
n, nf, maps = plan.n, plan.nf, plan.maps
if state.phase != "arrival_f" or state.tick % 2:
raise ValueError("synchronized arrival state required")
fc, ff = state.coarse_f.copy(), state.fine_f.copy()
covered_mean = ff.reshape(9, n, 2, n, 2).mean(axis=(2, 4))
fc[:, maps.coarse_covered] = covered_mean[:, maps.coarse_covered]
all_cells = np.ones((n, n), dtype=bool)
def collide(f, mask, tau):
return cpu._collide(spec, f, np.isfinite(f), tau, mask, np.zeros(mask.shape, dtype=int))
gc = collide(fc, all_cells, spec.tau_coarse)
collision = (gc[:, maps.coarse_active]-fc[:, maps.coarse_active]).sum(axis=1)
predicted = cpu._stream_periodic(gc)
gp = collide(predicted, all_cells, spec.tau_coarse) if mode == "spacetime" else gc
registers = np.zeros((9, n*n))
q, target = plan.receivers.T
registers[q, target] = -predicted.reshape(9, -1)[q, target]
incoming = [[], []]
group_error = 0.
min_boundary = float("inf")
negative_boundary = nonfinite_boundary = 0
dropped = False
for group in plan.groups:
q = group.q
count = len(group.samples)//2
data = np.concatenate((gc.reshape(9, -1)[q, group.samples[:count]], gp.reshape(9, -1)[q, group.samples[count:]]))
value = (group.weights @ data if mode == "spacetime"
else np.full(4, gc.reshape(9, -1)[q, group.source]))
min_boundary = min(min_boundary, float(value.min()))
negative_boundary += int(np.count_nonzero(value < 0))
nonfinite_boundary += int(np.count_nonzero(~np.isfinite(value)))
group_error = max(group_error, abs(value.mean()-gc.reshape(9, -1)[q, group.source]))
for k, event in enumerate(group.events):
if event is None:
if mutation == "omit_leftover" and not dropped:
dropped = True
else:
registers[q, group.receiver] += .25*value[k]
else:
m, slot = event
incoming[m].append((q, slot, value[k]))
for m in (0, 1):
gf = collide(ff, maps.fine_active, spec.tau_fine)
collision += .25*(gf[:, maps.fine_active]-ff[:, maps.fine_active]).sum(axis=1)
out = plan.outgoing[m]
if mutation != f"omit_outgoing_{m}":
np.add.at(registers, (out[:, 0], out[:, 2]), .25*gf.reshape(9, -1)[out[:, 0], out[:, 1]])
for q, slot, value in incoming[m]:
gf.reshape(9, -1)[q, slot] = value
fresh = np.full_like(ff, np.nan)
for q in range(9):
moved = cpu._shift(gf[q], int(cpu.CY[q]), int(cpu.CX[q]), np.nan)
fresh[q, maps.fine_active] = moved[maps.fine_active]
if not np.isfinite(fresh[:, maps.fine_active]).all():
raise AssertionError("invalid fine incoming population")
ff = fresh
final_c = predicted + registers.reshape(9, n, n)
final_c[:, maps.coarse_covered] = np.nan
final = cpu.State(final_c, ff, tick=state.tick+2)
residual = directional(final, maps)-directional(state, maps)-collision
minimum = min(float(final_c[:, maps.coarse_active].min()), float(ff[:, maps.fine_active].min()))
return final, dict(directional_residual=residual, collision=collision,
group_error=group_error, min_boundary=min_boundary, min_population=minimum,
negative_boundary_count=negative_boundary, nonfinite_boundary_count=nonfinite_boundary)
def write_receipt(path, result):
"""Normalize NumPy scalars at the JSON boundary; reject other unknown types."""
def scalar(value):
if isinstance(value, np.generic):
return value.item()
raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable")
path.write_text(json.dumps(result, indent=2, allow_nan=False, default=scalar)+"\n")
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--n64", action="store_true", help="explicitly run the frozen 1024-interval experiment")
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--max-seconds",type=float,default=180.,help="wall-time cap, including plan generation")
parser.add_argument("--max-rss-mib",type=float,default=256.,help="Linux peak-RSS screening cap")
args = parser.parse_args()
if not args.n64:
parser.error("full trajectory requires --n64")
if args.output.exists():
parser.error("output exists; preserve prior evidence and choose a new path")
if not (np.isfinite(args.max_seconds) and args.max_seconds>0 and np.isfinite(args.max_rss_mib) and args.max_rss_mib>0):
parser.error("positive finite resource caps required")
started = time.monotonic()
sys.path.insert(0, str(Path(__file__).parents[1]/"taylor_green"))
import oracle as tg
plan = Plan(64, (8, 8, 56, 48))
spec = Spec(n_coarse=64, patch_lo=8, halo=16, tau_coarse=.5576, tau_fine=.6152, collision="srt")
arrays = [tg.initial((64,64), 0, 64, .0192), tg.initial((128,128), 1, 64, .0192)]
for a, mask in zip(arrays, (plan.maps.coarse_active, plan.maps.fine_active)):
a[:, ~mask] = np.nan
initial = cpu.State(*arrays)
mass0 = tg.metrics(arrays, (plan.maps.coarse_active, plan.maps.fine_active), 0, 64, .0192)["mass"]
metrics = ("velocity_l2", "pressure_l2", "energy_relative")
frozen_baseline = dict(zip(metrics, (.01131134574353476, .03592961319244149, .012514764591008386)))
budget_bound = float(64*np.finfo(float).eps*mass0)
root = Path(__file__).resolve().parents[3]
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:hashlib.sha256(p.read_bytes()).hexdigest()
for p in (Path(__file__),Path(cpu.__file__),Path(tg.__file__))},numpy=np.__version__)
result = dict(scope="experimental CPU FP64; no GPU/accuracy certification", source=source,
screen=dict(velocity_factor=.8, pressure_factor=1., energy_factor=1.,
directional_step_bound=budget_bound, mass_relative=1e-11,
momentum_absolute=1e-9, baseline_metric_atol=1e-10),
limits=dict(seconds=args.max_seconds,rss_mib=args.max_rss_mib),runs=[],completed=False)
def save():
result["resources"] = dict(seconds=time.monotonic()-started,peak_rss_mib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/1024)
write_receipt(args.output, result)
save()
try:
for mode in ("constant_ivp", "spacetime"):
measured = mode == "spacetime"
state = initial; record = dict(mode=mode,rows=[],
boundary_budget_status="measured" if measured else "not_measured",
max_directional_residual=0. if measured else None,
minimum_boundary=None,negative_boundary_count=0 if measured else None,
nonfinite_boundary_count=0 if measured else None)
result["runs"].append(record)
for k in range(1, 1025):
if time.monotonic()-started>args.max_seconds or resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/1024>args.max_rss_mib:
raise RuntimeError("experiment resource cap exceeded")
if mode == "constant_ivp":
state,_ = cpu.advance_coarse_once(spec,plan.maps,state)
else:
state,diagnostic = advance(spec,plan,state)
error = float(np.max(np.abs(diagnostic["directional_residual"])))
record["max_directional_residual"] = max(record["max_directional_residual"],error)
if np.isfinite(diagnostic["min_boundary"]):
record["minimum_boundary"] = (diagnostic["min_boundary"] if record["minimum_boundary"] is None else min(record["minimum_boundary"],diagnostic["min_boundary"]))
for name in ("negative_boundary_count","nonfinite_boundary_count"):
record[name] += diagnostic[name]
if (not np.isfinite(error) or error>budget_bound or diagnostic["group_error"]>2e-14
or diagnostic["negative_boundary_count"] or diagnostic["nonfinite_boundary_count"]
or not np.isfinite(diagnostic["min_boundary"]) or diagnostic["min_boundary"]<=0
or not np.isfinite(diagnostic["min_population"]) or diagnostic["min_population"]<=0):
record["failed_diagnostic"] = {name: repr(value) for name,value in diagnostic.items()}
raise AssertionError("candidate directional/health invariant failed")
if k in (256,512,768,1024):
row = tg.metrics((state.coarse_f,state.fine_f),(plan.maps.coarse_active,plan.maps.fine_active),k,64,.0192,mass0)
row["momentum_change"] = (cpu.MOMENTS.T@(directional(state,plan.maps)-directional(initial,plan.maps)))[1:].tolist()
record["rows"].append(row)
save(); print(mode,k,json.dumps(row),flush=True)
record["worst"] = {key:max(row[key] for row in record["rows"]) for key in metrics}
if mode == "constant_ivp":
for key,value in frozen_baseline.items():
if abs(record["worst"][key]-value)>1e-10:
raise AssertionError(f"frozen baseline mismatch: {key}")
baseline,candidate = result["runs"]
result["checks"] = {key:candidate["worst"][key]<=factor*baseline["worst"][key]
for key,factor in zip(metrics,(.8,1.,1.))}
result["checks"]["mass_momentum"] = all(row["mass_relative"]<=1e-11 and max(map(abs,row["momentum_change"]))<=1e-9 for row in candidate["rows"])
result["completed"] = True
result["experimental_pass"] = all(result["checks"].values())
save()
return 0 if result["experimental_pass"] else 1
except Exception as exc:
result["failure"] = dict(mode=mode,step=k,error=repr(exc))
save()
raise
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,92 @@
"""N64/N128 admission check: exact topology and independently rational interpolation.
Raw FP64 bytes are diagnostic: different LAPACK builds need not produce them
identically. The 2e-14 ceiling is the existing interpolation constraint policy,
not a universal LAPACK error theorem. This module never changes runtime weights.
"""
from fractions import Fraction as F
from functools import lru_cache
import numpy as np
from spacetime_gpu import table_hash
STRUCTURE_SHA = 'cd1842952c4dd4df82da8015454c25da9be12b983270236cc65b4044fa38e6e4'
STRUCTURES = {64: (STRUCTURE_SHA, 532),
128: ('ee4ecb645674ae04eb87220a5217851405ccc0f45fbb0fe210499dcea5cd93b1', 1060)}
ATOL = 2e-14
def _transpose(a):
return list(map(list, zip(*a)))
def _multiply(a, b):
return [[sum(x*y for x,y in zip(row,column)) for column in zip(*b)] for row in a]
@lru_cache(None)
def exact_weights(kind, points):
"""Solve the small documented constrained fit using exact rational arithmetic."""
if kind == 'face':
a = [[F(1),F(s),F(t),F(s*t)] for t in (0,1) for s in (-1,0,1)]
e = [[F(1),F(s),F(t),F(s)*F(t)] for s,t in points]
b = [[sum(column)/4 for column in zip(*e)]]
d = [[F(int(j==1)) for j in range(6)]]
elif kind == 'corner':
a = [[F(1),F(x),F(y),F(t)] for t in (0,1) for x,y in ((0,0),(-1,0),(1,0),(0,-1),(0,1))]
e = [[F(1),F(x),F(y),F(t)] for x,y,t in points]
b = [[sum(column)/4 for column in zip(*e)], [F(0),F(0),F(0),F(1)]]
d = [[F(int(j==0)) for j in range(10)], [F(int(j==5)-int(j==0)) for j in range(10)]]
else:
raise ValueError('unsupported interpolation kind')
kkt = [r+s for r,s in zip(_multiply(_transpose(a),a),_transpose(b))]
kkt += [row+[F(0)]*len(b) for row in b]
rhs = _transpose(a)+d
n = len(kkt)
augmented = [row+data for row,data in zip(kkt,rhs)]
for j in range(n):
pivot = next(i for i in range(j,n) if augmented[i][j])
augmented[j],augmented[pivot] = augmented[pivot],augmented[j]
divisor = augmented[j][j]
augmented[j] = [v/divisor for v in augmented[j]]
for i in range(n):
if i != j:
factor = augmented[i][j]
augmented[i] = [x-factor*y for x,y in zip(augmented[i],augmented[j])]
solution = [row[n:] for row in augmented]
if _multiply(kkt,solution) != rhs:
raise AssertionError('exact interpolation solve failed')
return np.array(_multiply(e,solution[:4]), dtype=np.float64)
def validate_tables(plan, tables, *, allow_n128=False):
"""Reject topology, layout, padding, operator, or constraint changes separately."""
structure = table_hash({key:value for key,value in tables.items() if key != 'weights'})
admitted = (64,128) if allow_n128 is True else (64,)
if plan.n not in admitted or structure != STRUCTURES[plan.n][0]:
raise ValueError('frozen integer topology changed')
w = tables['weights']
if (w.dtype != np.dtype(np.float64) or w.shape != (STRUCTURES[plan.n][1],4,10) or len(plan.groups) != w.shape[0]
or not all(a.flags.c_contiguous for a in tables.values()) or not np.isfinite(w).all()):
raise ValueError('weight dtype/shape/layout/finiteness changed')
errors = dict(coefficient=0., constant=0., mean=0., temporal=0.)
for i,g in enumerate(plan.groups):
size = len(g.samples)
if size != tables['sizes'][i] or size != {'face':6,'corner':10}.get(g.kind) or len(g.points) != 4:
raise ValueError('interpolation group coverage/size changed')
if w[i,:,size:].tobytes() != np.zeros((4,10-size),np.float64).tobytes():
raise ValueError('weight padding is not exact positive zero')
actual = w[i,:,:size]
errors['coefficient'] = max(errors['coefficient'],float(abs(actual-exact_weights(g.kind,g.points)).max()))
errors['constant'] = max(errors['constant'],float(abs(actual.sum(axis=1)-1).max()))
central = np.zeros(size); central[1 if g.kind=='face' else 0] = 1
errors['mean'] = max(errors['mean'],float(abs(actual.mean(axis=0)-central).max()))
if g.kind == 'corner':
early,late = (g.points.index((0.,0.,t)) for t in (-.25,.25))
slope = np.zeros(size); slope[0],slope[5] = -1,1
errors['temporal'] = max(errors['temporal'],float(abs(2*(actual[late]-actual[early])-slope).max()))
if any(error > ATOL for error in errors.values()):
raise ValueError(f'exact interpolation/constraint check failed: {errors}; bound={ATOL}')
return dict(status='passed', structure_sha256=structure, raw_table_sha256=table_hash(tables),
coefficient_oracle='exact rational KKT, rounded once to FP64', atol=ATOL, rtol=0.,
maximum_absolute_errors=errors, positive_zero_padding=True)
@@ -0,0 +1,239 @@
"""Host-only experimental union event contract; no solver or runtime admission.
Consumes existing level-wide union geometry, never its legacy transfer tables.
The frozen 707afe6 rectangle oracle remains independent and unchanged. Topology
acceptance proves packet accounting, not interpolation accuracy or GPU support.
"""
from types import SimpleNamespace
import numpy as np
import spacetime_reference as ref
class UnresolvedTopology(ValueError):
"""An experimental coupling limitation, not a production geometry rejection."""
def __init__(self, code, **detail):
self.code, self.detail = code, detail
super().__init__(f"experimental space-time topology unresolved: {code}: {detail}")
def owned_slots(layout, directions, cells):
"""Map dense logical reads/writes to existing owner-core storage, not halos."""
cells = np.asarray(cells)
if not np.issubdtype(cells.dtype, np.integer) or np.any((cells < 0) | (cells >= np.prod(layout.shape))):
raise ValueError('invalid dense cell index')
owner = layout.owner.ravel()[cells]
if np.any(owner == layout.hole):
raise UnresolvedTopology('missing_block_owner', cells=cells[owner == layout.hole].tolist())
return layout.slot(owner, directions)
def _first_entry(covered, origin, child, c):
"""First continuous union entry of one characteristic, in exact quarter times.
Probe the cells immediately before/after each crossed grid line using integer
eighth-cell coordinates. Return only actual exterior faces of the UNION.
"""
start4 = 4*origin + 2*child + 1
for t4 in (1, 3):
before = (2*start4 + c*(2*t4-1))//8
after = (2*start4 + c*(2*t4+1))//8
if not covered[tuple(before)] and covered[tuple(after)]:
axes = []
for j in range(2):
if before[j] != after[j]:
neighbor = after.copy(); neighbor[j] = before[j]
if not covered[tuple(neighbor)]: axes.append(j)
return t4, axes
return None
def _crossing_time(covered, origin, position, c, axes):
"""Fig.11 convex T uses the last plane; Fig.13 concave T uses the first.
Both use the same KKT formulation on different boundary geometry.
DOI/primary PDF: https://www.osti.gov/servlets/purl/1226215, §3.5.1.
"""
times=[(origin[j]+int(c[j]>0)-position[j])/c[j] for j in axes]
if len(axes)==1: return times[0]
sides=[bool(covered[tuple(origin+np.eye(2,dtype=int)[j]*c[j])]) for j in axes]
if sides[0]!=sides[1]:
raise UnresolvedTopology('mixed_corner_support',source_yx=origin.tolist())
return min(times) if all(sides) else max(times)
class UnionPlan:
"""Two-level, level-wide square geometry; all coordinates are level-local.
Assumes a complete parent domain; does not validate nested parent coverage.
``pair`` is existing FixedPatchSet2D geometry, even for one rectangle. Scalar
window origins and nested state coverage need their own contract; they are
explicitly unadmitted here. No global rectangle-gap rule is imposed.
"""
phase = 'arrival_f'
runtime_supported = False
interpolation_accuracy = 'unqualified_for_new_unions'
def __init__(self, pair):
ny, nx = pair.coarse_shape
self.origin = tuple(pair.fine_origin)
if ny != nx or pair.fine_shape != (2*ny, 2*nx) or self.origin != (0., 0.):
raise UnresolvedTopology('requires_level_wide_square_pair',
coarse_shape=pair.coarse_shape, fine_shape=pair.fine_shape, origin=self.origin)
self.n, self.nf = nx, 2*nx
covered = np.array(pair.covered, dtype=bool, copy=True)
active = np.kron(covered, np.ones((2, 2), bool))
if not np.array_equal(active, pair.fine_kind == 0):
raise UnresolvedTopology('non_pair_active_coverage')
self.maps = SimpleNamespace(coarse_active=~covered, coarse_covered=covered, fine_active=active)
self.outgoing, transit = [[], []], []
# Guzik et al. §3.5.4, Fig.19 (https://www.osti.gov/servlets/purl/1226215):
# retain substep-0 outgoing G at source, then fill the substep-1 ghost.
# Rows are (q, fine source, next ghost); no covered coarse receiver credit.
for q in range(1, 9):
c = np.array((ref.cpu.CY[q], ref.cpu.CX[q]))
boundary = active & ~np.roll(active, tuple(-c), axis=(0, 1))
for y, x in zip(*np.nonzero(boundary)):
p = np.array((y, x))
for m in (0, 1):
target = (p+(2-m)*c)//2
if covered[tuple(target)]:
if m != 0 or not active[tuple(p+2*c)]:
raise UnresolvedTopology('unsupported_transit', q=q, substep=m, source_yx=p.tolist())
ghost=p+c
transit.append((q,int(y*self.nf+x),int(ghost[0]*self.nf+ghost[1])))
else:
self.outgoing[m].append((q, int(y*self.nf+x), int(target[0]*nx+target[1])))
self.outgoing = [np.array(x, dtype=np.int64).reshape(-1, 3) for x in self.outgoing]
self.transit = np.array(transit,dtype=np.int64).reshape(-1,3)
self.groups = []
for iy, ix in zip(*np.nonzero(~covered)):
origin = np.array((iy, ix))
# Two-cell production margins make all nearby probes in bounds.
if not covered[max(0,iy-1):iy+2,max(0,ix-1):ix+2].any(): continue
for q in range(1, 9):
c = np.array((ref.cpu.CY[q], ref.cpu.CX[q]))
children = [np.array((dy, dx)) for dy in (0, 1) for dx in (0, 1)]
starts = [2*origin+child for child in children]
events = []
for p in starts:
event = None
for m in (0, 1):
old, new = p+m*c, p+(m+1)*c
if not active[tuple(old)] and active[tuple(new)]:
event = (m, int(old[0]*self.nf+old[1])); break
events.append(event)
if all(e is None for e in events): continue
entries = [_first_entry(covered, origin, child, c) for child in children]
axes = sorted({j for entry in entries if entry is not None for j in entry[1]})
if not axes:
raise UnresolvedTopology('missing_exterior_support', source_yx=origin.tolist(), q=q)
points = []
for p, entry, event in zip(starts, entries, events):
position = (p+.5)/2
time = _crossing_time(covered,origin,position,c,axes)
if entry is not None and (4*time != entry[0] or event is None):
raise UnresolvedTopology('nonconvex_entry_support', source_yx=origin.tolist(), q=q,
quarter_time=entry[0], support_quarter_time=float(4*time))
offset = position+c*time-(origin+.5+c*.5)
points.append((float(offset[1]), float(offset[0]), float(time-.5)) if len(axes)==2
else (float(offset[1-axes[0]]), float(time-.5)))
kind = 'corner' if len(axes)==2 else 'face'; points = tuple(points)
samples, _, evaluation, coefficients = ref.interpolation(kind, points)
indices = []
for row in samples:
shift = np.array((row[1],row[0]), int) if kind=='corner' else np.eye(2,dtype=int)[1-axes[0]]*int(row[0])
pos = (origin+shift) % nx; indices.append(int(pos[0]*nx+pos[1]))
target = (origin+c) % nx
self.groups.append(ref.Group(q,int(iy*nx+ix),int(target[0]*nx+target[1]),kind,
points,np.array(indices),evaluation@coefficients,tuple(events)))
receivers = {(int(q),int(target)) for rows in self.outgoing for q,_,target in rows}
receivers |= {(g.q,g.receiver) for g in self.groups if None in g.events}
self.receivers = np.array(sorted(receivers),dtype=np.int64).reshape(-1,2)
self.proof = validate_ownership(self)
def validate_ownership(plan):
"""Independent mask-based event coverage and four-quarter receiver accounting."""
n, nf, active = plan.n, plan.nf, plan.maps.fine_active
actual = [(m,g.q,slot) for g in plan.groups for e in g.events if e is not None for m,slot in [e]]
expected_transit={(q,int(y*nf+x),int((y+ref.cpu.CY[q])*nf+x+ref.cpu.CX[q]))
for q in range(1,9) for y,x in zip(*np.nonzero(active))
if not active[y+ref.cpu.CY[q],x+ref.cpu.CX[q]]
and active[y+2*ref.cpu.CY[q],x+2*ref.cpu.CX[q]]}
routes=[tuple(map(int,row)) for row in plan.transit]
if len(routes)!=len(set(routes)) or set(routes)!=expected_transit:
raise ValueError('transit provenance/uniqueness/coverage failed')
actual += [(1,q,ghost) for q,source,ghost in routes]
expected = {(m,q,int((y-ref.cpu.CY[q])*nf+x-ref.cpu.CX[q]))
for m in (0,1) for q in range(1,9) for y,x in zip(*np.nonzero(active))
if not active[y-ref.cpu.CY[q],x-ref.cpu.CX[q]]}
if len(actual)!=len(set(actual)) or set(actual)!=expected:
raise ValueError('incoming writer uniqueness/coverage failed')
packets = {tuple(map(int,row)):[] for row in plan.receivers}
if len(packets)!=len(plan.receivers): raise ValueError('duplicate coarse receiver')
anchors = set()
for i,g in enumerate(plan.groups):
if not 1<=g.q<9 or not 0<=g.source<n*n or not plan.maps.coarse_active.ravel()[g.source]:
raise ValueError('invalid coarse source anchor')
if (g.q,g.source) in anchors: raise ValueError('duplicate coarse source group')
anchors.add((g.q,g.source))
origin=np.array(divmod(g.source,n)); c=np.array((ref.cpu.CY[g.q],ref.cpu.CX[g.q]))
target=(origin+c)%n
if g.receiver!=target[0]*n+target[1]: raise ValueError('group receiver provenance failed')
ordered=[]; entries=[]
for dy,dx in ((0,0),(0,1),(1,0),(1,1)):
p=2*origin+(dy,dx); event=None
for m in (0,1):
old,new=p+m*c,p+(m+1)*c
if not active[tuple(old)] and active[tuple(new)]:
event=(m,int(old[0]*nf+old[1])); break
ordered.append(event)
entries.append(_first_entry(plan.maps.coarse_covered,origin,np.array((dy,dx)),c))
if tuple(ordered)!=g.events or all(e is None for e in ordered):
raise ValueError('ordered child characteristic provenance failed')
axes=sorted({j for e in entries if e is not None for j in e[1]})
if not axes or g.kind!=('corner' if len(axes)==2 else 'face'):
raise ValueError('group supporting axes differ')
samples,_,evaluation,coefficients=ref.interpolation(g.kind,g.points)
indices=[]
for row in samples:
shift=np.array((row[1],row[0]),int) if g.kind=='corner' else np.eye(2,dtype=int)[1-axes[0]]*int(row[0])
pos=(origin+shift)%n; indices.append(int(pos[0]*n+pos[1]))
if not np.array_equal(g.samples,indices): raise ValueError('sample anchor provenance failed')
for k,(dy,dx) in enumerate(((0,0),(0,1),(1,0),(1,1))):
position=origin+(np.array((dy,dx))+.5)/2
time=_crossing_time(plan.maps.coarse_covered,origin,position,c,axes)
offset=position+c*time-(origin+.5+c*.5)
point=(float(offset[1]),float(offset[0]),float(time-.5)) if g.kind=='corner' else (float(offset[1-axes[0]]),float(time-.5))
if g.points[k]!=point: raise ValueError('evaluation point provenance failed')
source=np.zeros(len(g.samples)); source[0 if g.kind=='corner' else 1]=1
if (not np.allclose(g.weights,evaluation@coefficients,rtol=0,atol=2e-14)
or not np.allclose(g.weights.mean(axis=0),source,rtol=0,atol=2e-14)):
raise ValueError('group coefficients/directional budget failed')
for k,event in enumerate(g.events):
if event is None:
if (g.q,g.receiver) not in packets: raise ValueError('missing leftover receiver')
packets[(g.q,g.receiver)].append(('leftover',i,k))
for m,rows in enumerate(plan.outgoing):
expected_out = {(q,int(y*nf+x),int(((y+(2-m)*ref.cpu.CY[q])//2)*n+(x+(2-m)*ref.cpu.CX[q])//2))
for q in range(1,9) for y,x in zip(*np.nonzero(active))
if not active[y+ref.cpu.CY[q],x+ref.cpu.CX[q]]
and plan.maps.coarse_active[(y+(2-m)*ref.cpu.CY[q])//2,(x+(2-m)*ref.cpu.CX[q])//2]}
actual_out = [tuple(map(int,row)) for row in rows]
if len(actual_out)!=len(set(actual_out)) or set(actual_out)!=expected_out:
raise ValueError('outgoing uniqueness/coverage failed')
for q,slot,target in actual_out: packets[(q,target)].append(('outgoing',m,q,slot))
expected_receivers = {(q,int(y*n+x)) for y,x in zip(*np.nonzero(plan.maps.coarse_active))
for q in range(1,9) if any(active[2*y+dy-ref.cpu.CY[q],2*x+dx-ref.cpu.CX[q]]
for dy in (0,1) for dx in (0,1)
if 0<=2*y+dy-ref.cpu.CY[q]<nf and 0<=2*x+dx-ref.cpu.CX[q]<nf)}
if set(packets)!=expected_receivers: raise ValueError('coarse receiver coverage failed')
tokens=[token for row in packets.values() for token in row]
if any(len(row)!=4 for row in packets.values()) or len(tokens)!=len(set(tokens)):
raise ValueError('receiver needs four distinct quarter-packets')
return dict(topology='supported',incoming_events=len(actual),outgoing_events=sum(map(len,plan.outgoing)),
transit_events=len(routes),
groups=len(plan.groups),receivers=len(packets),receiver_quarter_packets=len(tokens),
interpolation_accuracy=plan.interpolation_accuracy,runtime_supported=False,
scope='pair-local ownership/coefficients; complete parent domain assumed, nested coverage unvalidated')
@@ -0,0 +1,497 @@
"""Explicit host-only checks or admitted tiny GPU audit; never a long trajectory."""
import argparse
from contextlib import nullcontext
from dataclasses import replace
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import time
from types import SimpleNamespace
import unittest
from unittest.mock import Mock
import numpy as np
ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(ROOT/'src'))
from CelerisLab.config import LBMConfig
import spacetime_gpu as gpu
import run_spacetime_gpu_n64 as trajectory
ref = gpu.ref
RECORDS = []
REPORT = None
OUTPUT = None
GPU_ADMITTED = False
def record(value):
RECORDS.append(value)
if OUTPUT is not None:
ref.write_receipt(OUTPUT,REPORT)
def config(n, shifted=False):
return LBMConfig(nx=n,ny=n,viscosity=.0192,velocity=0.,ddf_shifting=shifted)
def initial(plan, tagged=True):
arrays = []
for n, mask in ((plan.n,plan.maps.coarse_active),(plan.nf,plan.maps.fine_active)):
a = np.broadcast_to(ref.cpu.W[:,None,None], (9,n,n)).copy()
if tagged:
a *= 1+1e-3*np.random.default_rng(n).standard_normal(a.shape)
a[:,~mask] = np.nan
arrays.append(a)
return ref.cpu.State(*arrays)
class Host(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.plan = ref.Plan(12,(3,3,9,8))
def test_pins_and_supported_config(self):
gpu.check_sources()
for n in (12,16,64):
c,f = gpu.validate_config(config(n))
self.assertAlmostEqual(c.omega,1/.5576)
self.assertAlmostEqual(f.omega,1/.6152)
for change in (dict(ddf_shifting=1),dict(collision='TRT'),dict(nx=128,ny=128),
dict(ny=16),dict(les_enabled=True),dict(velocity=.03)):
with self.subTest(change=change), self.assertRaises(ValueError):
gpu.validate_config(replace(config(12),**change))
def test_shifted_codec_and_uploaded_physical_validation(self):
for n in (12,64):
self.assertTrue(gpu.validate_config(config(n,True))[1].ddf_shifting)
physical=gpu._WEIGHTS+2.**-35
stored=gpu._encode_physical(physical,True)
np.testing.assert_array_equal(stored,np.full((9,1,1),2.**-35,np.float32))
np.testing.assert_array_equal(gpu._decode_stored(stored,True),physical)
run=gpu.SpaceTimeGPU.__new__(gpu.SpaceTimeGPU)
run.status='new'; run.configs=gpu.validate_config(config(12,True)); run.plan=self.plan
run.ctx=SimpleNamespace(activate=nullcontext); run.cuda=Mock()
run.coarse=['coarse']; run.fine=['fine']; run.health_counts=np.zeros(6,np.uint64)
state=initial(self.plan,False)
run.initialize(state)
for call in run.cuda.memcpy_htod.call_args_list:
self.assertTrue((call.args[1][np.isfinite(call.args[1])]==0).all())
state.coarse_f[:]-=.01; state.fine_f[:]-=.01
run.initialize(state) # Valid negative stored values.
with self.assertRaisesRegex(ValueError,'exact shapes'):
run.initialize(ref.cpu.State(state.coarse_f[:1],state.fine_f))
for invalid in (-.01,np.nan):
bad=ref.cpu.State(state.coarse_f.copy(),state.fine_f.copy())
bad.coarse_f[0,0,0]=invalid
with self.assertRaisesRegex(ValueError,'positive finite'):
run.initialize(bad)
def test_shifted_transfer_translation_and_raw_ledger(self):
p=self.plan; t=gpu.pack_plan(p)
rng=np.random.default_rng(19)
for g in p.groups:
values=rng.uniform(.02,.45,len(g.samples)); weight=gpu._WEIGHTS[g.q,0,0]
np.testing.assert_allclose(g.weights@(values-weight),g.weights@values-weight,rtol=0,atol=2e-14)
# Uniform nonzero stored packets expose register omissions; exact rest would not.
arrays=[np.full((9,n,n),-.01,np.float32) for n in (p.n,p.nf)]
for a,m in zip(arrays,(p.maps.coarse_active,p.maps.fine_active)):
a[:,~m]=np.nan
state=ref.cpu.State(*arrays); g0=np.full((9,p.n,p.n),-.01,np.float32)
pre=np.repeat(g0.ravel()[t['origins'],None].astype(float),4,axis=1)
h=pre.astype(np.float32); registers=np.zeros(len(p.receivers))
args=(p,t,state,state,g0,g0,[(state.fine_f,state.fine_f)]*2,pre,h,registers)
ledger=gpu.rounding_ledger(*args,shifted=True)
self.assertEqual(max(map(abs,ledger['unexplained'])),0.)
self.assertGreater(ledger['unexplained_bound'],0.)
self.assertGreater(ledger['minimum_stored'],0.)
expected_mass=p.n*p.n*(1.+9*float(np.float32(-.01)))
self.assertEqual(ledger['unexplained_bound'],64*np.finfo(float).eps*expected_mass)
decoded=ref.cpu.State(*(gpu._decode_stored(a,True) for a in arrays))
with self.assertRaises(AssertionError):
gpu.rounding_ledger(p,t,decoded,*args[3:],shifted=True)
registers[0]+=.25*float(np.float32(-.01))
with self.assertRaisesRegex(AssertionError,'register mismatch'):
gpu.rounding_ledger(*args,shifted=True)
def test_packed_weights_and_packet_ownership(self):
for plan in (self.plan,ref.Plan(16,(4,5,11,12))):
t = gpu.pack_plan(plan)
self.assertEqual(t['sources'].shape,(len(plan.receivers),4))
self.assertEqual(np.count_nonzero(t['times']==-1),24)
self.assertEqual(gpu.table_hash(t),gpu.table_hash(gpu.pack_plan(plan)))
for i,g in enumerate(plan.groups):
np.testing.assert_array_equal(t['weights'][i,:,:len(g.samples)],g.weights)
np.testing.assert_allclose(t['weights'][i].sum(axis=1),1,rtol=0,atol=2e-14)
np.testing.assert_array_equal(t['samples'][i,:len(g.samples)],g.q*plan.n**2+g.samples)
for m in (0,1):
self.assertEqual(len(t[f'incoming{m}']),len(plan.outgoing[m]))
def test_duplicate_writer_is_rejected(self):
plan = self.plan
groups = list(plan.groups)
a,b = next((i,k) for i,g in enumerate(groups) for k,e in enumerate(g.events) if e is not None)
event = groups[a].events[b]
other = next(i for i,g in enumerate(groups) if i!=a and g.q==groups[a].q and event[0] in [e[0] for e in g.events if e])
events = list(groups[other].events)
k = next(k for k,e in enumerate(events) if e is not None and e[0]==event[0])
events[k] = event; groups[other] = replace(groups[other],events=tuple(events))
fake = SimpleNamespace(n=plan.n,nf=plan.nf,groups=groups,receivers=plan.receivers,outgoing=plan.outgoing)
with self.assertRaisesRegex(ValueError,'duplicate incoming'):
gpu.pack_plan(fake)
def test_rest_ledger_and_missing_packet_discrimination(self):
p,t = self.plan,gpu.pack_plan(self.plan)
state = initial(p,False)
state = ref.cpu.State(state.coarse_f.astype(np.float32),state.fine_f.astype(np.float32))
g0 = np.broadcast_to(ref.cpu.W.astype(np.float32)[:,None,None],(9,p.n,p.n)).copy()
pre = np.repeat(g0.ravel()[t['origins'],None].astype(float),4,axis=1)
h = pre.astype(np.float32); registers = np.zeros(len(p.receivers))
phases = [(state.fine_f,state.fine_f)]*2
args = (p,t,state,state,g0,g0,phases,pre,h)
ledger = gpu.rounding_ledger(*args,registers)
self.assertEqual(max(map(abs,ledger['unexplained'])),0)
np.testing.assert_array_equal(ledger['physical_collision'],np.array(ledger['coarse_collision'])
+ledger['fine0_collision']+ledger['fine1_collision'])
np.testing.assert_array_equal(ledger['child_rounding'],np.array(ledger['child_cast'])+ledger['child_interpolation'])
np.testing.assert_array_equal(ledger['collision_cell_mass'],np.zeros((3,4)))
registers[0] = .01
with self.assertRaisesRegex(AssertionError,'register mismatch'):
gpu.rounding_ledger(*args,registers)
def test_second_substep_characteristic_donor(self):
p = ref.Plan(16,(4,4,12,12)); t = gpu.pack_plan(p)
row = t['incoming1'][t['incoming1'][:,0]==5*32*32+10*32+7][0]
self.assertEqual(t['origins'][row[1]//4],5*16*16+4*16+3)
def test_failed_free_remains_retryable(self):
run=gpu.SpaceTimeGPU.__new__(gpu.SpaceTimeGPU)
run.ctx=SimpleNamespace(activate=nullcontext); run.stream=Mock()
first,second=Mock(),Mock(); first.free.side_effect=[RuntimeError('free failed'),None]
run.allocations=[first,second]; run.modules=[]; run.status='ready'
with self.assertRaisesRegex(RuntimeError,'free failed'):
run.close()
self.assertEqual(run.allocations,[first]); self.assertEqual(run.status,'cleanup_failed')
run.close()
self.assertEqual(run.status,'closed'); self.assertEqual(run.allocations,[])
second.free.assert_called_once()
def test_unmeasured_schedule_and_health_failure(self):
run=gpu.SpaceTimeGPU.__new__(gpu.SpaceTimeGPU)
run.status='ready'; run.step_count=0; run.identity=False; run.mode='spacetime'
run.ctx=SimpleNamespace(activate=nullcontext); run.cuda=Mock(); run.stream=Mock()
run.plan=self.plan; run.bounds=(3,3,9,8)
run.configs=gpu.validate_config(config(12)); run.coarse=list('abcd'); run.fine=list('efg')
run.tables=gpu.pack_plan(self.plan); run.device_tables={key:key for key in run.tables}
run.active='active'; run.pre='pre'; run.h='h'; run.registers='registers'; run.counts='counts'
run.health_counts=np.zeros(6,np.uint64)
run.launch=Mock(); run.read=Mock(return_value=np.zeros(6,np.uint32)); run.download=Mock()
names=['CoarseStart','Pull','Collide','Interpolate','Collide','Register','Inject','Pull',
'Collide','Register','Inject','Pull','ApplyRegisters']
for _ in range(2):
result=run.step(audit=False)
self.assertEqual(result['boundary_budget_status'],'not_measured')
self.assertIsNone(result['unexplained'])
self.assertEqual([call.args[1] for call in run.launch.call_args_list],names*2)
self.assertEqual(run.read.call_count,2); run.download.assert_not_called()
self.assertEqual(run.step_count,2); self.assertEqual(run.coarse,list('abcd'))
run.read.return_value=np.array([0,0,0,0,1,0],np.uint32)
with self.assertRaisesRegex(AssertionError,'health failed'):
run.step(audit=False)
self.assertEqual(run.status,'failed'); self.assertEqual(run.step_count,2)
self.assertEqual(run.health_counts.tolist(),[0,0,0,0,1,0])
def test_warmup_resets_arrivals_and_counters(self):
audits=[]
class Fake:
def initialize(self,state):
self.state=ref.cpu.State(state.coarse_f.copy(),state.fine_f.copy())
self.step_count=0; self.health_counts=np.zeros(6,np.uint64)
def download(self):
return ref.cpu.State(self.state.coarse_f.copy(),self.state.fine_f.copy(),tick=2*self.step_count)
def step(self,*,audit):
audits.append(audit)
self.step_count+=1; self.state.coarse_f[:]+=1e-4
fake=Fake(); check=Mock(); state=initial(self.plan)
actual,warm=trajectory.warm_and_reset(fake,state,check)
np.testing.assert_array_equal(actual.coarse_f,state.coarse_f)
self.assertEqual(check.call_count,32); self.assertEqual(fake.step_count,0)
self.assertTrue(warm['reset_verified']); self.assertEqual(warm['intervals'],32)
trajectory.warm_and_reset(fake,state,check,audit=True)
self.assertEqual(audits,[False]*32+[True]*32)
def test_accumulated_drift_ledger_and_exact_samples(self):
totals=trajectory.LedgerTotals()
for sign in (1.,-1.):
ledger={name:np.zeros(9) for name in totals.names}
ledger['coarse_collision'][0]=sign*2.**-20
ledger['physical_collision']=ledger['coarse_collision'].copy()
ledger['child_cast'][5]=sign*2.**-30
ledger['child_interpolation'][5]=sign*2.**-50
ledger['child_rounding']=ledger['child_cast']+ledger['child_interpolation']
ledger['receiver_rounding'][5]=-sign*2.**-32
ledger['net_change']=sum(ledger[k] for k in ('physical_collision','child_rounding','receiver_rounding'))
mass=ledger['coarse_collision'].sum()
ledger.update(unexplained_bound=1e-12,collision_cell_mass=[[mass,max(mass,0.),min(mass,0.),abs(mass)],[0.]*4,[0.]*4])
totals.add(ledger)
result=totals.snapshot(np.zeros(9))
self.assertEqual(result['intervals'],2)
self.assertEqual(result['components']['physical_collision']['directional'],[0.]*9)
self.assertGreater(result['components']['physical_collision']['absolute_step_mass_momentum_sum'][0],0)
self.assertEqual(result['components']['physical_collision']['max_absolute_interval_step'][0],1)
self.assertEqual(totals.snapshot()['endpoint_status'],'not_measured')
self.assertFalse(totals.snapshot(np.ones(9)*1e-8)['closure_pass'])
self.assertEqual(trajectory.differing_fields(dict(a=1.,b=[0.,2.]),dict(a=1.,b=[0.,2.])),[])
self.assertEqual(trajectory.differing_fields(dict(a=0.),dict(a=-0.)),['a'])
self.assertEqual(trajectory.differing_fields(dict(a=1.),dict(a=np.nextafter(1.,2.))),['a'])
def test_trajectory_screens_differences_and_json(self):
rows=[dict(tstar=t,velocity_l2=np.float64(.01),pressure_l2=.03,energy_relative=.01,
mass_relative=1e-7,momentum_change=[1e-6,-2e-6],minimum_population=.02,
minimum_boundary_pre=.02,minimum_boundary_stored=.02) for t in trajectory.TIMES]
cpu=dict(runs=[dict(worst=dict(velocity_l2=.02,pressure_l2=.04,energy_relative=.02)),
dict(rows=[dict(tstar=t,velocity_l2=.009,momentum_change=[0.,0.]) for t in trajectory.TIMES])])
uniform=dict(rows=[dict(tstar=.12,pressure_l2=.012253173806349701)])
result=trajectory.assess(rows,cpu,uniform,[0]*6)
self.assertTrue(result['campaign_pass']); self.assertTrue(result['experimental_pass'])
self.assertFalse(result['historical_pressure_gate']['passed'])
self.assertFalse(result['historical_pressure_gate']['matched_storage'])
self.assertAlmostEqual(result['gpu_minus_cpu']['rows'][0]['fields']['velocity_l2']['signed'],.001)
with tempfile.TemporaryDirectory() as directory:
path=Path(directory)/'receipt.json'
ref.write_receipt(path,dict(completed=True,**result))
self.assertTrue(json.loads(path.read_text())['completed'])
rows[0]['velocity_l2']=.06
failed=trajectory.assess(rows,cpu,uniform,[0,1,0,0,0,0])
ref.write_receipt(path,dict(completed=True,**failed))
saved=json.loads(path.read_text())
self.assertFalse(saved['campaign_pass']); self.assertFalse(saved['experimental_pass'])
for invalid in (float('nan'),-.01):
rows[0]['velocity_l2']=.01; rows[-1]['velocity_l2']=invalid
failed=trajectory.assess(rows,cpu,uniform,[0]*6)
self.assertFalse(failed['campaign_pass']); self.assertFalse(failed['experimental_pass'])
path.write_text('changed reference')
with self.assertRaisesRegex(ValueError,'pinned reference'):
trajectory.load_references(path,path)
def test_shifted_descriptive_comparison_and_runner_pin(self):
expected=dict(mass=4096.,momentum=[0.,-1e-4])
actual=dict(mass=4096.+2.**-20,momentum=[-1e-6,-2e-4])
difference=trajectory.sample_difference(actual,expected)
self.assertEqual(difference['mass']['signed'],2.**-20)
self.assertEqual(difference['momentum']['absolute'],[1e-6,1e-4])
self.assertEqual(trajectory.sha(Path(gpu.__file__).with_suffix('.cu')),trajectory.CUDA_SHA)
self.assertEqual(trajectory.LIMITS,dict(velocity_l2=.05,pressure_l2=.15,energy_relative=.05,mass_relative=1e-5))
self.assertEqual(trajectory.FACTORS,dict(velocity_l2=.8,pressure_l2=1.,energy_relative=1.))
def test_n64_table_guard_and_balanced_mutation(self):
guard=trajectory.table_guard
plan=ref.Plan(64,(8,8,56,48)); tables=gpu.pack_plan(plan)
self.assertEqual(guard.validate_tables(plan,tables)['status'],'passed')
face=next(i for i,g in enumerate(plan.groups) if g.kind=='face')
changed={key:value.copy() for key,value in tables.items()}
changed['weights'][face,0,0]=np.nextafter(changed['weights'][face,0,0],np.inf)
self.assertEqual(guard.validate_tables(plan,changed)['status'],'passed')
for mutation in ('coordinate','order','dtype','shape','layout','padding','negative_zero','nan','inf','balanced'):
changed={key:value.copy() for key,value in tables.items()}
w=changed['weights']
if mutation=='coordinate': changed['incoming1'][0,0]^=1
elif mutation=='order': changed['origins']=changed['origins'][::-1].copy()
elif mutation=='dtype': changed['weights']=w.astype(np.float32)
elif mutation=='shape': changed['weights']=w.reshape(532,10,4)
elif mutation=='layout': changed['weights']=np.asfortranarray(w)
elif mutation=='padding': w[face,0,9]=1e-30
elif mutation=='negative_zero': w[face,0,9]=-0.
elif mutation=='nan': w[face,0,0]=np.nan
elif mutation=='inf': w[face,0,0]=np.inf
else:
w[face,:2,:2] += 2.**-30*np.array([[1.,-1.],[-1.,1.]])
np.testing.assert_array_equal(w[face].sum(axis=1),tables['weights'][face].sum(axis=1))
np.testing.assert_array_equal(w[face].mean(axis=0),tables['weights'][face].mean(axis=0))
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
guard.validate_tables(plan,changed)
changed={key:value.copy() for key,value in tables.items()}
changed['weights'][-1,0,0]+=1.
with self.assertRaises(ValueError):
guard.validate_tables(SimpleNamespace(n=64,groups=plan.groups[:-1]),changed)
groups=list(plan.groups); groups[0]=replace(groups[0],samples=groups[0].samples[:-1])
with self.assertRaisesRegex(ValueError,'coverage/size'):
guard.validate_tables(SimpleNamespace(n=64,groups=groups),tables)
class Device(unittest.TestCase):
shifted = False
def config(self,n):
return config(n,self.shifted)
@classmethod
def setUpClass(cls):
if not GPU_ADMITTED:
raise unittest.SkipTest('CUDA compilation/execution requires explicit tiny-job admission')
from CelerisLab.cuda.context import CudaContext
cls.context = CudaContext(0)
@classmethod
def tearDownClass(cls):
cls.context.close()
def test_tiny_equivalence_budgets_and_phase(self):
for mode,identity in (('spacetime',False),('spacetime',True),('frozen',False),('frozen',True)):
with self.subTest(mode=mode,identity=identity), gpu.SpaceTimeGPU(
self.config(12),(3,3,9,8),context=self.context,mode=mode,identity=identity) as run:
spec = ref.Spec(n_coarse=12,patch_lo=1,halo=2,tau_coarse=.5576,tau_fine=.6152,
collision='identity' if identity else 'srt')
for tagged,steps in ((False,2),(True,8)):
expected = initial(run.plan,tagged); run.initialize(expected)
before = run.download()
if self.shifted:
expected = before
for k in range(1,steps+1):
expected,_ = ref.advance(spec,run.plan,expected,mode)
ledger = run.step(); actual = run.download()
self.assertEqual(actual.tick,2*k); self.assertEqual(run.step_count,k)
for a,b in zip((actual.coarse_f,actual.fine_f),(expected.coarse_f,expected.fine_f)):
np.testing.assert_array_equal(np.isnan(a),np.isnan(b))
np.testing.assert_allclose(a,b,rtol=0,atol=3e-6,equal_nan=True)
total = lambda s: ref.cpu.MOMENTS.T @ ref.directional(
ref.cpu.State(s.coarse_f.astype(float),s.fine_f.astype(float)),run.plan.maps)
np.testing.assert_allclose(total(actual),total(before),rtol=0,atol=2e-4)
if tagged and not identity and k==1:
for a,b,tau,mask in zip((actual.coarse_f,actual.fine_f),
(expected.coarse_f,expected.fine_f),(.5576,.6152),
(run.plan.maps.coarse_active,run.plan.maps.fine_active)):
wrong = ref.cpu._collide(spec,b,np.isfinite(b),tau,mask,np.zeros(mask.shape,int))
self.assertGreater(float(np.nanmax(abs(a-wrong))),3e-6)
if k in (1,2,8):
record(dict(case='tiny',shifted=self.shifted,mode=mode,identity=identity,tagged=tagged,
step=k,device_bytes=run.device_bytes,ledger=ledger))
def test_omitted_packets_and_ghosts_fail(self):
for mutation in (1,2,3,4):
with self.subTest(mutation=mutation), gpu.SpaceTimeGPU(
self.config(12),(3,3,9,8),context=self.context,identity=self.shifted) as run:
state=initial(run.plan,not self.shifted)
if self.shifted:
state.coarse_f[:]+=1e-4; state.fine_f[:]+=1e-4
for a,mask in zip((state.coarse_f,state.fine_f),(run.plan.maps.coarse_active,run.plan.maps.fine_active)):
self.assertTrue((gpu._encode_physical(a,True)[:,mask]>0).all())
run.initialize(state)
with self.assertRaises(AssertionError):
run.step(mutation=mutation)
self.assertEqual(run.step_count,0)
self.assertEqual(run.status,'failed')
with self.assertRaises(RuntimeError):
run.download()
def test_frozen_donor_trap(self):
with gpu.SpaceTimeGPU(self.config(16),(4,4,12,12),context=self.context,mode='frozen',identity=True) as run:
state = initial(run.plan,False)
state.coarse_f[5,4,3]=1.; state.coarse_f[5,5,3]=2.
run.initialize(state); donor=run.download().coarse_f[5,4,3]; run.step()
self.assertEqual(run.download().fine_f[5,11,8],donor)
# Negative stored values can be physical in shifted mode; never clip either case.
with self.context.activate():
for physical_negative in ((False,True) if self.shifted else (True,)):
data=np.full((9,16,16),-.01,np.float32)
if self.shifted and physical_negative:
data=np.broadcast_to(-gpu._WEIGHTS-.01,(9,16,16)).astype(np.float32).copy()
run.cuda.memcpy_htod(run.coarse[1],data); run.cuda.memcpy_htod(run.coarse[3],data)
run.cuda.memset_d32_async(run.counts,0,6,run.stream)
t=run.device_tables; ng=len(run.plan.groups)
run.launch(0,'Interpolate',run.coarse[1],run.coarse[3],t['samples'],t['origins'],
t['sizes'],t['weights'],run.pre,run.h,np.uint32(ng),np.uint32(0),run.counts,count=ng)
counts=run.read(run.counts,(6,),np.uint32)
self.assertEqual(counts[1],4*ng*physical_negative)
self.assertEqual(counts[3],4*ng*physical_negative)
self.assertTrue((run.read(run.h,(ng,4))<0).all())
def test_n64_one_interval_only(self):
sys.path.insert(0,str(Path(__file__).parents[1]/'taylor_green'))
import oracle as tg
with gpu.SpaceTimeGPU(self.config(64),(8,8,56,48),context=self.context) as run:
arrays = [tg.initial((64,64),0,64,.0192),tg.initial((128,128),1,64,.0192)]
for a,mask in zip(arrays,(run.plan.maps.coarse_active,run.plan.maps.fine_active)):
a[:,~mask]=np.nan
state = ref.cpu.State(*arrays); run.initialize(state)
spec = ref.Spec(n_coarse=64,patch_lo=8,halo=16,tau_coarse=.5576,tau_fine=.6152,collision='srt')
expected,_ = ref.advance(spec,run.plan,run.download() if self.shifted else state)
started=time.monotonic(); ledger=run.step(); seconds=time.monotonic()-started
actual=run.download()
errors=[]
for a,b in zip((actual.coarse_f,actual.fine_f),(expected.coarse_f,expected.fine_f)):
np.testing.assert_array_equal(np.isnan(a),np.isnan(b))
np.testing.assert_allclose(a,b,rtol=0,atol=3e-6,equal_nan=True)
errors.append(float(np.nanmax(abs(a-b))))
record(dict(case='N64_one_interval',shifted=self.shifted,errors=errors,ledger=ledger,seconds=seconds,
device_bytes=run.device_bytes,table_sha256=gpu.table_hash(run.tables),
build_metadata=run.build_metadata,device=self.context.name))
def test_unmeasured_path_and_warm_reset(self):
with gpu.SpaceTimeGPU(self.config(12),(3,3,9,8),context=self.context) as run:
state=initial(run.plan)
run.initialize(state)
raw_initial=run._download_stored()
for _ in range(2):
run.step()
expected=run._download_stored()
run.initialize(state)
for _ in range(2):
result=run.step(audit=False)
self.assertEqual(result['boundary_budget_status'],'not_measured')
actual=run._download_stored()
for a,b in zip((actual.coarse_f,actual.fine_f),(expected.coarse_f,expected.fine_f)):
np.testing.assert_array_equal(a.view(np.uint32),b.view(np.uint32))
reset,warm=trajectory.warm_and_reset(run,state,lambda:None)
raw_reset=run._download_stored()
for a,b in zip((raw_reset.coarse_f,raw_reset.fine_f),(raw_initial.coarse_f,raw_initial.fine_f)):
np.testing.assert_array_equal(a.view(np.uint32),b.view(np.uint32))
record(dict(case='unmeasured_bitwise_and_reset',shifted=self.shifted,warmup=warm,health_counts=run.health_counts.tolist()))
class ShiftedDevice(Device):
shifted = True
def main():
global REPORT,OUTPUT,GPU_ADMITTED
parser=argparse.ArgumentParser(description=__doc__)
mode=parser.add_mutually_exclusive_group(required=True)
mode.add_argument('--host-only',action='store_true')
mode.add_argument('--gpu-admitted',action='store_true')
parser.add_argument('--output',type=Path)
args=parser.parse_args()
GPU_ADMITTED=args.gpu_admitted
if args.gpu_admitted and (not os.environ.get('CUDA_VISIBLE_DEVICES') or args.output is None):
parser.error('GPU job requires explicit CUDA_VISIBLE_DEVICES and --output')
if args.output is not None and args.output.exists():
parser.error('output exists; retain evidence and choose a new path')
started=time.monotonic()
report=dict(scope='tiny GPU equivalence and rounding audit; no long trajectory',completed=False,
requested_mode='admitted_tiny_gpu' if args.gpu_admitted else 'host_only',
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:hashlib.sha256(p.read_bytes()).hexdigest() for p in
(Path(__file__),Path(gpu.__file__),Path(gpu.__file__).with_suffix('.cu'))},
reference_hashes=gpu.PINS,cuda_visible_devices=os.environ.get('CUDA_VISIBLE_DEVICES'),records=RECORDS)
REPORT,OUTPUT=report,args.output
if args.output:
ref.write_receipt(args.output,report)
suite=unittest.TestLoader().loadTestsFromTestCase(Host)
if args.gpu_admitted:
suite.addTests(unittest.TestLoader().loadTestsFromTestCase(Device))
suite.addTests(unittest.TestLoader().loadTestsFromTestCase(ShiftedDevice))
result=unittest.TextTestRunner(verbosity=2).run(suite)
report.update(completed=True,passed=result.wasSuccessful(),tests=result.testsRun,
failures=[dict(test=str(test),error=error) for test,error in result.failures+result.errors],
seconds=time.monotonic()-started)
if args.output:
ref.write_receipt(args.output,report)
return 0 if result.wasSuccessful() else 1
if __name__=='__main__':
raise SystemExit(main())
@@ -0,0 +1,205 @@
"""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())
@@ -0,0 +1,155 @@
"""Tiny CPU algebra/event tests; no full trajectory or GPU execution."""
from dataclasses import replace
import json
from pathlib import Path
import tempfile
import unittest
import numpy as np
import spacetime_reference as st
class SpaceTime(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.plan = st.Plan(12, (3, 3, 9, 8))
cls.spec = st.Spec(n_coarse=12, patch_lo=1, halo=2,
tau_coarse=.5576, tau_fine=.6152, collision="srt")
def state(self, tagged=False):
arrays = []
for n, mask in ((12,self.plan.maps.coarse_active),(24,self.plan.maps.fine_active)):
q, y, x = np.indices((9,n,n))
a = np.broadcast_to(st.cpu.W[:,None,None], (9,n,n)).copy()
if tagged:
a *= 1+1e-3*np.random.default_rng(n).normal(size=a.shape)
a[:,~mask] = np.nan
arrays.append(a)
return st.cpu.State(*arrays)
def test_events_and_corner_leftovers(self):
groups = [g for g in self.plan.groups if None in g.events]
self.assertEqual(len(groups), 8)
self.assertTrue(all(g.events.count(None)==3 for g in groups))
for m in (0,1):
events = list(map(tuple,self.plan.outgoing[m]))
self.assertEqual(len(events),len(set(events)))
st.Plan(16,(4,5,11,12))
def test_interpolation(self):
for g in self.plan.groups:
samples, a, e, coeff = st.interpolation(g.kind,g.points)
np.testing.assert_allclose(g.weights@np.ones(len(samples)),1,rtol=0,atol=2e-14)
np.testing.assert_allclose(g.weights.mean(axis=0), np.eye(len(samples))[1 if g.kind=='face' else 0],rtol=0,atol=2e-14)
if g.kind=='face':
beta=np.array([1.,.03,.02,0.])
else:
cx,cy=st.cpu.CX[g.q],st.cpu.CY[g.q]
beta=np.array([1.,.03,.02,-cx*.03-cy*.02])
steady=a@np.array([1.,.03,.02,0.])
self.assertLess(abs((coeff@steady)[3]),2e-14)
np.testing.assert_allclose(g.weights@(a@beta),e@beta,rtol=0,atol=2e-14)
def test_rest(self):
initial=self.state()
final,d=st.advance(self.spec,self.plan,initial)
for a,b in zip((initial.coarse_f,initial.fine_f),(final.coarse_f,final.fine_f)):
np.testing.assert_allclose(a,b,rtol=0,atol=2e-14,equal_nan=True)
def test_frozen_recovers_ivp(self):
for collision in ('identity','srt'):
spec=replace(self.spec,collision=collision)
a=b=self.state(True)
for _ in range(2):
a,_=st.advance(spec,self.plan,a,'frozen')
b,_=st.cpu.advance_coarse_once(spec,self.plan.maps,b)
for aa,bb in zip((a.coarse_f,a.fine_f),(b.coarse_f,b.fine_f)):
np.testing.assert_allclose(aa,bb,rtol=0,atol=2e-14,equal_nan=True)
def test_second_substep_uses_characteristic_origin(self):
# Review counterexample in (x,y): ghost (7,10), m=1 traces back
# to (6,9), parent (3,4), not its current parent (3,5).
plan=st.Plan(16,(4,4,12,12))
owners=[g for g in plan.groups if g.q==5 and (1,10*32+7) in g.events]
self.assertEqual(len(owners),1)
self.assertEqual(owners[0].source,4*16+3)
spec=st.Spec(n_coarse=16,patch_lo=2,halo=4,collision='identity')
arrays=[]
for n,mask in ((16,plan.maps.coarse_active),(32,plan.maps.fine_active)):
a=np.broadcast_to(st.cpu.W[:,None,None],(9,n,n)).copy()
a[:,~mask]=np.nan
arrays.append(a)
arrays[0][5,4,3]=1.; arrays[0][5,5,3]=2.
final,_=st.advance(spec,plan,st.cpu.State(*arrays),'frozen')
self.assertEqual(final.fine_f[5,11,8],1.)
self.assertNotEqual(final.fine_f[5,11,8],arrays[0][5,5,3])
def test_corner_affine_tradeoff_is_explicit(self):
group=next(g for g in self.plan.groups if g.kind=='corner')
_,a,e,coeff=st.interpolation(group.kind,group.points)
linear_x=np.array([0.,1.,0.,0.])
# The two paper constraints are consistent. Arbitrary affine exactness
# is the incompatible additional demand.
self.assertEqual(abs(float((e@linear_x).mean())),.125)
fitted=coeff@(a@linear_x)
self.assertLess(abs(fitted[3]),2e-14)
self.assertLess(abs(float((e@fitted).mean())),2e-14)
self.assertGreater(np.max(np.abs(e@fitted-e@linear_x)),.01)
def test_directional_budgets(self):
for collision in ('identity','srt'):
spec=replace(self.spec,collision=collision)
a=self.state(True)
for _ in range(2):
a,d=st.advance(spec,self.plan,a)
np.testing.assert_allclose(d['directional_residual'],0,rtol=0,atol=2e-12)
self.assertLess(d['group_error'],2e-14)
def test_missing_packets_fail(self):
for mutation in ('omit_leftover','omit_outgoing_0','omit_outgoing_1'):
_,d=st.advance(self.spec,self.plan,self.state(True),mutation=mutation)
self.assertGreater(np.max(np.abs(d['directional_residual'])),1e-5)
def test_negative_interpolation_is_reported_not_clipped(self):
initial=self.state()
group=self.plan.groups[0]
initial.coarse_f.reshape(9,-1)[group.q,group.source]=-.01
spec=replace(self.spec,collision='identity')
_,d=st.advance(spec,self.plan,initial,'frozen')
self.assertGreaterEqual(d['negative_boundary_count'],4)
self.assertEqual(d['min_boundary'],-.01)
self.assertEqual(d['nonfinite_boundary_count'],0)
class Receipts(unittest.TestCase):
def test_completed_and_failed_screen_receipts(self):
for passed in (True, False):
with self.subTest(passed=passed), tempfile.TemporaryDirectory() as directory:
# Same NumPy comparison type produced by the final metric screen.
checks = {'velocity_l2': np.float64(.5 if passed else 1.) <= .8,
'mass_momentum': True}
self.assertIsInstance(checks['velocity_l2'], np.bool_)
result = dict(completed=True, checks=checks,
experimental_pass=all(checks.values()),
runs=[dict(rows=[dict(step=np.int64(1024),
mass_relative=np.float32(0.))])])
path = Path(directory)/'receipt.json'
st.write_receipt(path, result)
saved = json.loads(path.read_text())
self.assertIs(saved['completed'], True)
self.assertIs(saved['experimental_pass'], passed)
self.assertIs(saved['checks']['velocity_l2'], passed)
self.assertEqual(saved['runs'][0]['rows'][0],
dict(step=1024, mass_relative=0.))
def test_receipts_still_reject_nonfinite_and_unknown_values(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory)/'receipt.json'
with self.assertRaises(ValueError):
st.write_receipt(path, {'invalid': np.float32('nan')})
with self.assertRaises(TypeError):
st.write_receipt(path, {'invalid': object()})
self.assertFalse(path.exists())
if __name__=='__main__':
unittest.main(verbosity=2)
@@ -0,0 +1,287 @@
"""Host-only ownership/geometry tests, never a GPU or accuracy admission."""
import copy
import json
import sys
import unittest
import numpy as np
from CelerisLab.config import LBMConfig
from CelerisLab.lbm.blocks import BlockLayout2D
from CelerisLab.lbm.refinement import FixedPatch2D, FixedPatchSet2D
from spacetime_union_plan import UnionPlan, UnresolvedTopology, owned_slots, validate_ownership
import spacetime_reference as ref
import spacetime_gpu as frozen # pack_plan only: no CUDA import/owner construction
from spacetime_table_guard import exact_weights
MATRIX=[]
ENUMERATION={}
CASES={
'rectangle': ([(4,4,12,12)],True),
'two_separated': ([(3,3,6,6),(9,9,12,12)],True),
'one_cell_face_gap': ([(3,3,6,10),(7,3,10,10)],True),
'one_cell_diagonal_gap': ([(3,3,6,6),(7,7,10,10)],True),
'staggered_gap': ([(3,3,6,8),(7,5,11,10)],True),
'thin_rectangle': ([(4,4,5,11)],True),
'concave_L': ([(4,4,5,7),(4,5,7,7)],True),
'corner_contact': ([(3,3,6,6),(6,6,9,9)],True),
'cross': ([(3,6,13,9),(6,3,9,13)],True),
'narrow_notch': ([(3,3,6,11),(7,3,10,11),(3,9,10,11)],True),
'hole': ([(3,3,13,5),(3,11,13,13),(3,5,5,11),(11,5,13,11)],True),
}
def pair(rects,n=16,ny=None):
cfg=LBMConfig(nx=n,ny=n if ny is None else ny,viscosity=.0192,velocity=0.)
return FixedPatchSet2D.from_config(cfg,rects)
def assert_same(test,a,b):
# Includes event-child order, zero padding, integer dtype/shape and FP64 bits.
aa,bb=frozen.pack_plan(a),frozen.pack_plan(b)
test.assertEqual(aa.keys(),bb.keys())
for key in aa:
test.assertEqual((aa[key].dtype,aa[key].shape,aa[key].tobytes()),
(bb[key].dtype,bb[key].shape,bb[key].tobytes()),key)
for x,y in zip(a.groups,b.groups):
test.assertEqual((x.q,x.source,x.receiver,x.kind,x.points,x.events),
(y.q,y.source,y.receiver,y.kind,y.points,y.events))
# Frozen pack_plan has no transit column and is NOT a union runtime format.
np.testing.assert_array_equal(getattr(a,'transit',np.empty((0,3),int)),
getattr(b,'transit',np.empty((0,3),int)))
class UnionContract(unittest.TestCase):
def test_single_rectangle_exact_707afe6(self):
frozen.check_sources()
for n,b in ((12,(3,3,9,8)),(16,(4,5,11,12)),(16,(4,4,5,7)),(64,(8,8,56,48))):
with self.subTest(n=n,bounds=b):
assert_same(self,UnionPlan(pair([b],n)),ref.Plan(n,b))
self.assertNotIn('pycuda.driver',sys.modules)
def test_union_decomposition_and_permutation_invariance(self):
baseline=UnionPlan(pair([(4,4,12,12)]))
decompositions=[[(4,4,8,12),(8,4,12,12)],[(4,4,10,12),(6,4,12,12)],
[(4,4,12,8),(4,8,12,12)],[(4,4,12,12),(4,4,12,12)]]
for rectangles in decompositions:
for order in (rectangles,list(reversed(rectangles))):
assert_same(self,baseline,UnionPlan(pair(order)))
two=CASES['two_separated'][0]
assert_same(self,UnionPlan(pair(two)),UnionPlan(pair(list(reversed(two)))))
lshape=CASES['concave_L'][0]
for decomposition in (list(reversed(lshape)),[(4,4,5,5),(4,5,7,7)],lshape+lshape):
assert_same(self,UnionPlan(pair(lshape)),UnionPlan(pair(decomposition)))
def test_supported_unresolved_matrix(self):
for name,(rectangles,supported) in CASES.items():
geometry=pair(rectangles) # Existing production accepts EVERY fixture.
row=dict(case=name,rectangles=rectangles,production_geometry_accepted=True,
runtime_supported=False,interpolation_accuracy='unqualified_for_new_unions')
if supported:
plan=UnionPlan(geometry); row.update(plan.proof)
self.assertEqual(plan.proof['receiver_quarter_packets'],4*len(plan.receivers))
else:
with self.assertRaises(UnresolvedTopology) as caught: UnionPlan(geometry)
row.update(topology='unresolved',code=caught.exception.code,witness=caught.exception.detail)
MATRIX.append(row)
def test_L_witness_and_reflections(self):
rectangles=CASES['concave_L'][0]
plan=UnionPlan(pair(rectangles))
self.assertIn((5,8*32+9,9*32+10),map(tuple,plan.transit))
directions=set()
for reflect in (False,True):
transformed=[(16-x1,y0,16-x0,y1) if reflect else (x0,y0,x1,y1) for x0,y0,x1,y1 in rectangles]
for _ in range(4):
plan=UnionPlan(pair(transformed)); self.assertEqual(len(plan.transit),2)
for q,source,ghost in plan.transit:
directions.add(q); p=np.array(divmod(source,32)); c=np.array((ref.cpu.CY[q],ref.cpu.CX[q]))
self.assertEqual(ghost,(p+c)[0]*32+(p+c)[1])
self.assertEqual([bool(plan.maps.fine_active[tuple(p+j*c)]) for j in range(3)],[True,False,True])
transformed=[(16-y1,x0,16-y0,x1) for x0,y0,x1,y1 in transformed]
self.assertEqual(directions,{5,6,7,8})
def test_Fig19_two_hop_transport_and_quarter_budget(self):
# Guzik §3.5.4 Fig.19: https://www.osti.gov/servlets/purl/1226215 .
# Isolated tagged transport only: no collision law or union evolution.
for name in ('concave_L','corner_contact','hole'):
p=UnionPlan(pair(CASES[name][0])); q,source,ghost=p.transit.T
stride=ref.cpu.CY[q]*p.nf+ref.cpu.CX[q]; target=ghost+stride
tags=(np.arange(len(q),dtype=np.float32)+1)/256
active=p.maps.fine_active
def run(shifted,mutation=None):
scratch=np.full((9,p.nf,p.nf),np.nan,np.float32)
scratch.reshape(9,-1)[q,source]=tags
def moment(field):
value=field.astype(float)+(ref.cpu.W[:,None,None] if shifted else 0)
return .25*np.where(np.isfinite(field)&active,value,0).sum(axis=(1,2))
def pull(field):
moved=np.stack([np.roll(field[j],(ref.cpu.CY[j],ref.cpu.CX[j]),axis=(0,1)) for j in range(9)])
moved[:,~active]=np.nan
return moved
initial=moment(scratch)
carry=[(int(j),int(h),v.copy()) for j,h,v in zip(q,ghost,scratch.reshape(9,-1)[q,source])]
if mutation=='drop': carry.pop()
if mutation=='duplicate': carry.append(carry[0])
first=pull(scratch) # m0 ghost arrivals are outside the ACTIVE mask.
inventory=np.zeros(9)
for j,h,value in carry: inventory[j]+=.25*(float(value)+(ref.cpu.W[j] if shifted else 0))
np.testing.assert_allclose(moment(first)+inventory,initial,rtol=0,atol=2e-14)
if mutation=='early':
for j,h,value in carry: scratch.reshape(9,-1)[j,h]=value
scratch[:]=np.nan # m1 collision scratch reuse destroys an early injection.
while carry:
j,h,value=carry.pop()
if mutation!='early': scratch.reshape(9,-1)[j,h]=value
final=pull(scratch)
self.assertEqual(carry,[]) # Consumed route inventory, not subtraction of identical sums.
np.testing.assert_allclose(moment(final),initial,rtol=0,atol=2e-14)
arrivals=final.reshape(9,-1)[q,target]
np.testing.assert_array_equal(arrivals.view(np.uint32),tags.view(np.uint32))
for shifted in (False,True):
run(shifted)
for mutation in ('drop','duplicate','early'):
with self.assertRaises(AssertionError): run(shifted,mutation)
for direction,slot,_ in p.transit:
self.assertNotIn((direction,slot),{tuple(row[:2]) for row in p.outgoing[0]})
for change in (lambda b:setattr(b,'transit',b.transit[1:]),
lambda b:setattr(b,'transit',np.vstack((b.transit,b.transit[0]))),
lambda b:b.transit.__setitem__((0,2),b.transit[0,2]+1),
lambda b:b.transit.__setitem__((0,0),1)):
bad=copy.deepcopy(p); change(bad)
with self.assertRaises(ValueError): validate_ownership(bad)
bad=copy.deepcopy(p); direction,slot,_=p.transit[0]
y,x=divmod(slot,p.nf); cy,cx=ref.cpu.CY[direction],ref.cpu.CX[direction]
covered_target=((y+2*cy)//2)*p.n+(x+2*cx)//2
bad.outgoing[0]=np.vstack((bad.outgoing[0],[direction,slot,covered_target]))
with self.assertRaises(ValueError): validate_ownership(bad)
def test_Fig13_first_entry_and_wrong_crossing(self):
p=UnionPlan(pair(CASES['concave_L'][0]))
g=next(g for g in p.groups if g.source==4*16+5 and g.q==8)
self.assertEqual(g.kind,'corner')
self.assertEqual(sorted(point[-1] for point in g.points),[-.25,-.25,-.25,.25])
# For a concave T first entry is min, not the frozen convex max.
damaged=copy.deepcopy(p); wrong=next(x for x in damaged.groups if x.source==g.source and x.q==g.q)
origin=np.array((4,5)); c=np.array((ref.cpu.CY[g.q],ref.cpu.CX[g.q])); points=[]
for child in ((0,0),(0,1),(1,0),(1,1)):
position=origin+(np.array(child)+.5)/2
time=max((origin+ (c>0)-position)/c); offset=position+c*time-(origin+.5+c*.5)
points.append((float(offset[1]),float(offset[0]),float(time-.5)))
wrong.points=tuple(points)
with self.assertRaises(ValueError): validate_ownership(damaged)
def test_all_511_small_masks(self):
counts=dict(masks=0,with_transit=0,transit_paths=0)
for bits in range(1,512):
rectangles=[(4+x,4+y,5+x,5+y) for y in range(3) for x in range(3) if bits & (1<<(3*y+x))]
p=UnionPlan(pair(rectangles)); counts['masks']+=1
counts['with_transit']+=bool(len(p.transit)); counts['transit_paths']+=len(p.transit)
# Independent first-entry check: fine-cell membership before/after
# each quarter crossing, without the plan's plane min/max helper.
for g in p.groups:
c=np.array((ref.cpu.CY[g.q],ref.cpu.CX[g.q])); origin=np.array(divmod(g.source,p.n))
for k,child in enumerate(((0,0),(0,1),(1,0),(1,1))):
start=2*origin+child; hits=[]
for t4 in (1,3):
before=(4*start+2+c*(2*t4-1))//4
after=(4*start+2+c*(2*t4+1))//4
if not p.maps.fine_active[tuple(before)] and p.maps.fine_active[tuple(after)]: hits.append(t4/4)
if hits: self.assertEqual(g.points[k][-1]+.5,hits[0])
self.assertEqual(counts,dict(masks=511,with_transit=384,transit_paths=2048)); ENUMERATION.update(counts)
def test_ordered_provenance_and_packet_mutations_fail(self):
baseline=UnionPlan(pair(CASES['two_separated'][0]))
def reject(change):
bad=copy.deepcopy(baseline); change(bad)
with self.assertRaises((ValueError,KeyError)): validate_ownership(bad)
reject(lambda p:setattr(p.groups[0],'source',p.groups[0].source+1))
reject(lambda p:setattr(p.groups[0],'samples',p.groups[0].samples+1))
reject(lambda p:setattr(p.groups[0],'events',tuple(reversed(p.groups[0].events))))
reject(lambda p:setattr(p.groups[0],'receiver',p.groups[0].receiver+1))
reject(lambda p:p.outgoing.__setitem__(0,p.outgoing[0][1:]))
reject(lambda p:p.outgoing.__setitem__(1,np.vstack((p.outgoing[1],p.outgoing[1][0]))))
reject(lambda p:setattr(p,'receivers',p.receivers[1:]))
def balanced(p):
p.groups[0].weights[:2,:2]+=1e-5*np.array([[1,-1],[-1,1]])
reject(balanced) # Preserves row/column sums, corrupts the actual operator.
def test_coefficients_and_directional_transfer_budget(self):
rng=np.random.default_rng(31)
for name in CASES:
p=UnionPlan(pair(CASES[name][0])); total=np.zeros(9)
receiver={tuple(row):[] for row in p.receivers}; first_leftover=None
for g in p.groups:
samples,a,e,coeff=ref.interpolation(g.kind,g.points)
np.testing.assert_allclose(g.weights,exact_weights(g.kind,g.points),rtol=0,atol=2e-14)
np.testing.assert_allclose(g.weights@np.ones(len(samples)),1,rtol=0,atol=2e-14)
if g.kind=='face': beta=np.array([1.,.03,.02,0.])
else:
beta=np.array([1.,.03,.02,-ref.cpu.CX[g.q]*.03-ref.cpu.CY[g.q]*.02])
self.assertLess(abs((coeff@(a@np.array([1.,.03,.02,0.])))[3]),2e-14)
np.testing.assert_allclose(g.weights@(a@beta),e@beta,rtol=0,atol=2e-14)
values=rng.normal(size=len(samples)); h=g.weights@values
origin=values[0 if g.kind=='corner' else 1]
# Fine incoming plus coarse leftovers must exactly split one
# coarse directional source budget, regardless of sample values.
incoming=sum(h[k] for k,event in enumerate(g.events) if event is not None)
leftover=sum(h[k] for k,event in enumerate(g.events) if event is None)
self.assertLess(abs(.25*(incoming+leftover)-origin),2e-14)
total[g.q]+=.25*incoming-origin
for k,event in enumerate(g.events):
if event is None:
receiver[(g.q,g.receiver)].append(h[k])
if first_leftover is None: first_leftover=(g.q,h[k])
# Tagged collision-free boundary-flux ledger, not a solver trajectory:
# every outgoing fine debit is credited to its actual receiver.
for rows in p.outgoing:
for q,slot,target in rows:
value=float(rng.normal()); total[q]-=.25*value
receiver[(q,target)].append(value)
for (q,target),values in receiver.items():
self.assertEqual(len(values),4); total[q]+=.25*sum(values)
np.testing.assert_allclose(total,0,rtol=0,atol=2e-14)
damaged=total.copy(); q,value=first_leftover; damaged[q]-=.25*value
self.assertGreater(float(abs(damaged).max()),1e-5)
self.assertEqual(validate_ownership(p)['topology'],'supported')
def test_existing_block_addressing_and_sparse_holes(self):
geometry=pair(CASES['two_separated'][0]); p=UnionPlan(geometry)
layouts=(BlockLayout2D.build(16,16,4,4),geometry.fine_layout((4,4)))
for fine,layout in enumerate(layouts):
cells=layout.shape[0]*layout.shape[1]
dense=np.arange(9*cells,dtype=np.float64).reshape(9,*layout.shape)
blocked=layout.to_blocked(dense).ravel()
refs=[]
if not fine:
refs=[(g.q,int(k)) for g in p.groups for k in [g.source,g.receiver,*g.samples]]
else:
refs=[(g.q,slot) for g in p.groups for event in g.events if event is not None for _,slot in [event]]
refs += [(int(q),int(slot)) for rows in p.outgoing for q,slot,_ in rows]
q,k=np.array(refs).T
np.testing.assert_array_equal(blocked[owned_slots(layout,q,k)],dense.reshape(9,-1)[q,k])
holes=np.flatnonzero(~layouts[1].owned)
self.assertGreater(len(holes),0)
with self.assertRaises(UnresolvedTopology): owned_slots(layouts[1],np.array([1]),holes[:1])
def test_deferred_layouts_and_phase_are_explicit(self):
with self.assertRaises(UnresolvedTopology) as caught: UnionPlan(pair([(3,3,8,8)],16,12))
self.assertEqual(caught.exception.code,'requires_level_wide_square_pair')
cfg=LBMConfig(nx=16,ny=16,viscosity=.0192,velocity=0.)
scalar=FixedPatch2D.from_config(cfg,(4,4,8,8))
with self.assertRaises(UnresolvedTopology): UnionPlan(scalar)
self.assertEqual(UnionPlan.phase,'arrival_f'); self.assertFalse(UnionPlan.runtime_supported)
# Restriction and collision do not commute, even at omega=1. This is
# why postcollision G storage cannot be relabeled/inverted into this F contract.
w=ref.cpu.W; cx,cy=ref.cpu.CX,ref.cpu.CY
def equilibrium(u):
cu=cx*u;return w*(1+3*cu+4.5*cu*cu-1.5*u*u)
children=np.array([equilibrium(u) for u in (.03,-.03,.03,-.03)])
self.assertAlmostEqual(float(abs(w-children.mean(axis=0)).max()),.0006,places=14)
if __name__=='__main__':
result=unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(UnionContract))
print(json.dumps(dict(passed=result.wasSuccessful(),tests=result.testsRun,
scope='host-only topology/provenance/coefficient checks; no runtime or accuracy admission',matrix=MATRIX,
enumeration=ENUMERATION),indent=2))
sys.exit(not result.wasSuccessful())