Frank14f/CelerisLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add bounded projected pair CUDA qualification gate

Browse Source
This commit is contained in:
Codex
2026-10-18 15:04:51 -07:00
parent a135252ae7
commit 1aa8110ec8
2 changed files with 587 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_projected_pair_gate.py
+162
View File
@@ -0,0 +1,162 @@
"""Host-only checks for the opt-in CUDA gate; never imports or compiles CUDA."""
from copy import deepcopy
import importlib.util
from pathlib import Path
import unittest
from unittest.mock import patch
import numpy as np
from CelerisLab.config import LBMConfig
from CelerisLab.lbm.blocks import BlockLayout2D
from CelerisLab.lbm.hierarchy import build_hierarchy
from CelerisLab.lbm.refinement import ACTIVE, OUTER_GHOST, FixedPatch2D, FixedPatchSet2D
from CelerisLab.lbm.refinement_tables import projected_explosion_tables
from tests.unit.test_hierarchy import CFG, FOUR
from tests.unit.test_projected_explosion import decode, reconstruct, totals
PATH = Path(__file__).resolve().parents[1] / 'validation/multigrid/check_projected_pairs.py'
spec = importlib.util.spec_from_file_location('projected_pair_gate', PATH)
gate = importlib.util.module_from_spec(spec)
spec.loader.exec_module(gate)
class ProjectedGateOracle(unittest.TestCase):
def check_pair(self, pair, layout, kind):
ids, offsets = projected_explosion_tables(pair, layout, kind)
neighbors = decode(layout, ids)
coarse = np.random.default_rng(731).normal(size=(9, *pair.coarse_shape))
legal = np.zeros(pair.coarse_shape, bool)
legal.ravel()[pair.ghost_coarse] = True
coarse[:, ~legal] = np.nan
expected, _ = reconstruct(coarse, pair, neighbors, offsets)
# The coordinate oracle must not accidentally share production tables.
with patch('CelerisLab.lbm.refinement_tables.projected_explosion_tables',
side_effect=AssertionError('oracle used production stencil')):
actual = gate.independent_explosion(coarse, pair, layout, kind)
np.testing.assert_allclose(actual, expected, rtol=0, atol=2e-15)
center = coarse.reshape(9, -1)[:, pair.ghost_coarse]
self.assertTrue(np.isfinite(actual).all())
self.assertGreater(np.max(abs(actual[0] - center[0])), .01)
self.assertGreater(np.max(abs(actual[1:] - center[1:])), .01)
return coarse, neighbors, offsets, actual
def test_scalar_union_nested_sparse_and_unequal_strides(self):
def fine_layout(pair, block):
return (pair.fine_layout(block) if isinstance(pair, FixedPatchSet2D) else
BlockLayout2D.build(*pair.fine_shape, *block, periodic=(False,False)))
small = LBMConfig(nx=16, ny=16, viscosity=.02)
fixtures = [(small, [(4,4,12,12), (8,8,16,16), (8,8,16,16)]),
(CFG, FOUR), (small, [[(4,4,5,7), (4,5,7,7)]])]
for cfg, refs in fixtures:
hierarchy = build_hierarchy(cfg, refs)
for index, pair in enumerate(hierarchy.pairs):
block, child_block = ((4,4), (4,8)) if index % 2 == 0 else ((4,8), (4,4))
layout = (BlockLayout2D.build(cfg.ny, cfg.nx, *block) if index == 0
else fine_layout(hierarchy.pairs[index-1], block))
with self.subTest(refs=refs, pair=index):
self.assertNotEqual(np.prod(layout.local_shape),
np.prod(fine_layout(pair, child_block).local_shape))
if index and isinstance(pair, FixedPatchSet2D):
self.assertTrue((~layout.owned).any())
self.check_pair(pair, layout, hierarchy.kinds[index])
def test_axis_periodicity_and_mean_preserving_mutants(self):
pair = FixedPatch2D.from_config(LBMConfig(nx=16,ny=16,viscosity=.1), (2,2,14,14))
kind = np.where(pair.covered, OUTER_GHOST, ACTIVE)
outputs = []
for periodic in ((False,False), (True,False), (False,True), (True,True)):
layout = BlockLayout2D.build(16,16,4,4,periodic=periodic)
coarse, neighbors, offsets, actual = self.check_pair(pair, layout, kind)
outputs.append(actual)
donors, first, inverse = np.unique(pair.ghost_coarse, return_index=True, return_inverse=True)
wrong = neighbors[first[np.roll(np.arange(len(donors)), 1)]][inverse]
for bad_neighbors, bad_offsets in ((neighbors, -offsets), (wrong, offsets)):
mutant, _ = reconstruct(coarse, pair, bad_neighbors, bad_offsets)
self.assertGreater(np.max(abs(mutant - actual)), .01)
for donor in donors:
rows = pair.ghost_coarse == donor
self.assertEqual(rows.sum(), 4)
np.testing.assert_allclose(mutant[:,rows].mean(axis=1),
coarse.reshape(9,-1)[:,donor], rtol=0, atol=1e-15)
bad_kind = kind.copy(); bad_kind.ravel()[pair.ghost_coarse[0]] = OUTER_GHOST
with self.assertRaisesRegex(AssertionError, 'illegal center donor'):
gate.independent_explosion(coarse, pair, layout, bad_kind)
for i, j in ((0,1), (0,2), (1,3), (2,3)):
self.assertGreater(np.max(abs(outputs[i] - outputs[j])), .01)
def test_concave_replay_preserves_existing_fp64_budget(self):
cfg, refs, blocks = gate.cases()['concave']
hierarchy = build_hierarchy(cfg, refs); pair = hierarchy.pairs[0]
layout = BlockLayout2D.build(cfg.ny, cfg.nx, *blocks[0])
rng = np.random.default_rng(17)
values = tuple(gate.W[:,None,None] * (1 + rng.normal(0,.002,(9,*shape)))
for shape in (pair.coarse_shape, pair.fine_shape))
start = totals(*values, pair)
values[0][:,pair.covered] = np.nan
for _ in range(8):
values = gate.replay_two_hops(values, pair, layout, hierarchy.kinds[0])
np.testing.assert_allclose(totals(*values, pair), start, rtol=0, atol=4e-12)
self.assertEqual(gate.REPLAY_ATOL, 3e-6) # CUDA replay bound is separate.
class ProjectedGateReceipts(unittest.TestCase):
def test_digest_exact_finite_bits_and_nan_masks(self):
values = np.array([0., -0., np.nan, 1.], dtype=np.float32)
self.assertEqual(gate.digest(values), gate.digest(values.copy()))
payload = values.copy(); payload.view(np.uint32)[2] = 0x7fc00001
self.assertEqual(gate.digest(values), gate.digest(payload))
variants = [np.array([-0., -0., np.nan, 1.], np.float32),
np.array([0., np.nan, -0., 1.], np.float32),
np.array([0., -0., np.nan, np.nextafter(np.float32(1), np.float32(2))], np.float32),
values.reshape(2,2), values.astype(np.float64)]
for changed in variants:
self.assertNotEqual(gate.digest(values), gate.digest(changed))
with self.assertRaisesRegex(AssertionError, 'infinite'):
gate.digest(np.array([np.inf]))
def test_snapshot_negative_controls_and_exact_dev_options(self):
snapshot = [dict(step=0, levels=[dict(raw='raw',public='public',active=4)],
mass_momentum=[1.,0.,0.],observed_level_hops=[0,0])]
gate.compare_snapshots(snapshot, deepcopy(snapshot))
diagnostic = deepcopy(snapshot); diagnostic[0]['mass_momentum'][0] = 2.
gate.compare_snapshots(snapshot, diagnostic) # Diagnostics are not bitwise gate fields.
mutants = [[]]
for key, value in (('raw','changed'), ('public','changed'), ('active',3)):
changed = deepcopy(snapshot); changed[0]['levels'][0][key] = value
mutants.append(changed)
changed = deepcopy(snapshot); changed[0]['step'] = 1; mutants.append(changed)
changed = deepcopy(snapshot); changed[0]['observed_level_hops'] = [0,1]; mutants.append(changed)
for changed in mutants:
with self.assertRaises(AssertionError):
gate.compare_snapshots(snapshot, changed)
self.assertEqual(gate.default_options('omitted'), {})
self.assertEqual(gate.default_options('constant'), {'explosion':'constant'})
self.assertEqual(gate.default_options('projected_linear'), {'explosion':'projected_linear'})
def test_baseline_receipt_rejects_provenance_and_contract_mutants(self):
keys = [f'{name}:shifted={shifted}' for name in gate.cases() for shifted in (False,True)]
good = dict(completed=True,dry_run=False,mode='baseline',production_revision=gate.BASE,
source=dict(revision=gate.BASE,dirty=False),driver_sha256='driver',
stops=list(gate.STOPS),operator_tolerance=[6e-8,2e-7],
rest_tolerance=[3e-7,1e-6],replay_tolerance=[3e-6,0],default_snapshots=dict.fromkeys(keys, []),
actual_build_sidecars=[{'ptx_sha256':'captured'}])
self.assertIs(gate.baseline_receipt(good, 'driver', keys), good)
mutants = dict(completed=False,dry_run=True,mode='candidate',production_revision='wrong',
driver_sha256='wrong',stops=[0,1,2],operator_tolerance=[1e-6,2e-7],
rest_tolerance=[3e-7,1e-5],replay_tolerance=[3e-6,1e-7],actual_build_sidecars=[],default_snapshots={})
for field, value in mutants.items():
bad = deepcopy(good); bad[field] = value
with self.subTest(field=field), self.assertRaises(ValueError):
gate.baseline_receipt(bad, 'driver', keys)
for changes in (dict(revision='wrong'), dict(dirty=True), dict(dirty=None)):
bad = deepcopy(good); bad['source'].update(changes)
with self.assertRaises(ValueError):
gate.baseline_receipt(bad, 'driver', keys)
bad = deepcopy(good); bad['default_snapshots']['unexpected'] = []
with self.assertRaises(ValueError):
gate.baseline_receipt(bad, 'driver', keys)
if __name__ == '__main__':
unittest.main()
tests/validation/multigrid/check_projected_pairs.py
+425
View File
@@ -0,0 +1,425 @@
"""Bounded exact-source CUDA plumbing gate; never a flow-accuracy certificate.
Run this driver externally against exact dev with --mode baseline first, then
against the experimental source with --mode candidate --baseline <receipt>.
Only scalar receipts/hashes and actual compiler sidecars are retained. Whole-step
mass/momentum are diagnostics: no verified FP32 conservation bound is claimed.
"""
import argparse
from contextlib import contextmanager
from dataclasses import replace
import hashlib
import json
import os
import shutil
from pathlib import Path
import subprocess
import sys
import numpy as np
BASE = 'ed9d8c937d446b854d6963bd92d718bb4fb98b96'
PRODUCTION = 'a135252ae78b47b57d7e77d19001700b3bedac8b'
STOPS = (0, 1, 2, 8)
OPERATOR_ATOL, OPERATOR_RTOL = 6e-8, 2e-7
REST_ATOL, REST_RTOL = 3e-7, 1e-6
REPLAY_ATOL = 3e-6
E = np.array([(0,0),(1,0),(-1,0),(0,1),(0,-1),(1,1),(-1,-1),(1,-1),(-1,1)])
W = np.array([4/9]+[1/9]*4+[1/36]*4)
def digest(array):
"""All finite bits (including signed zero) and exact NaN mask; no NaN payloads."""
array = np.asarray(array)
if np.isinf(array).any():
raise AssertionError('infinite population')
finite = np.isfinite(array)
h = hashlib.sha256()
h.update(str((array.shape, array.dtype.str)).encode())
h.update(np.ascontiguousarray(finite).tobytes())
h.update(np.ascontiguousarray(array[finite]).tobytes())
return h.hexdigest()
def independent_explosion(coarse, pair, layout, kind):
"""Coordinate oracle; never reads the production neighbor/offset tables."""
ny, nx = pair.coarse_shape
legal = np.zeros((ny,nx), bool)
legal.ravel()[pair.ghost_coarse] = True
legal &= layout.owned & ~pair.covered & (kind == 0)
if not legal.ravel()[pair.ghost_coarse].all():
raise AssertionError('illegal center donor')
out = np.empty((9,len(pair.ghost_fine)))
flat = np.asarray(coarse, np.float64).reshape(9,-1)
for row, (target, donor) in enumerate(zip(pair.ghost_fine,pair.ghost_coarse)):
y,x = divmod(int(donor),nx)
gradient = np.zeros((9,2))
for axis, (dx,dy) in enumerate(((1,0),(0,1))):
ends = [(x-dx,y-dy),(x+dx,y+dy)]
if not layout.periodic[axis] and any(not (0<=xx<nx and 0<=yy<ny) for xx,yy in ends):
continue
ends = [(xx%nx,yy%ny) for xx,yy in ends]
if all(legal[yy,xx] for xx,yy in ends):
(xm,ym),(xp,yp) = ends
gradient[:,axis] = (flat[:,yp*nx+xp]-flat[:,ym*nx+xm])/2
for q in range(1,9):
gradient[q] -= E[q]*np.dot(E[q],gradient[q])/np.dot(E[q],E[q])
fy,fx = divmod(int(target),pair.fine_shape[1])
offset = np.array([pair.fine_origin[0]+(fx+.5)/2-(x+.5),
pair.fine_origin[1]+(fy+.5)/2-(y+.5)])
out[:,row] = flat[:,int(donor)] + gradient@offset
return out
def default_options(selector):
# Exact dev has no explosion keyword. Baseline and omitted paths pass none.
return {} if selector == 'omitted' else {'explosion':selector}
def snapshot(run, raw):
from CelerisLab.lbm.refinement import ACTIVE, OUTER_GHOST
stored, public = raw(run), run.download()
hashes, totals = [], np.zeros(3)
for level,(values,visible,kind) in enumerate(zip(stored,public,run.hierarchy.kinds)):
active = kind == ACTIVE
if not np.isfinite(visible[:,active]).all() or not np.isnan(visible[:,~active]).all():
raise AssertionError('public ACTIVE/invalid mask mismatch')
if not np.isnan(values[:,kind==OUTER_GHOST]).all():
raise AssertionError('OUTER owner core is not poisoned')
population = visible[:,active].sum(axis=1,dtype=np.float64)*(.25**level)
totals += np.r_[population.sum(), E.T@population]
hashes.append(dict(raw=digest(values), public=digest(visible), active=int(active.sum())))
if not np.isfinite(totals).all():
raise AssertionError('nonfinite mass/momentum diagnostic')
return dict(step=run.step_count, levels=hashes, mass_momentum=totals.tolist())
def initialize(run, rest=False):
arrays, origin, spacing = [], np.zeros(2), 1.
for index,level in enumerate(run._levels):
if index:
origin += spacing*np.asarray(run.hierarchy.pairs[index-1].fine_origin)
spacing /= 2
y,x = np.indices(level.shape[1:],dtype=np.float64)
x,y = origin[0]+spacing*(x+.5),origin[1]+spacing*(y+.5)
q = np.arange(9)[:,None,None]
values = np.broadcast_to(W[:,None,None],level.shape).copy()
if not rest:
# q-distinct smooth non-equilibrium G; same physical frame on all levels.
values += 1e-4*np.sin(.17*x+.23*y+q)
arrays.append(values)
run.initialize(*arrays)
def operator_case(run, index):
import pycuda.driver as cuda
parent,child = run._levels[index:index+2]
geometry,pair = run.hierarchy.pairs[index],run._pairs[index]
c = np.random.default_rng(19+index).normal(0,.001,(9,*geometry.coarse_shape))
if not run._ddf_shifting:
c += W[:,None,None]
c = c.astype(np.float32)
legal = np.zeros(geometry.coarse_shape,bool)
legal.ravel()[geometry.ghost_coarse] = True
c[:,~legal] = np.nan
packed = parent.layout.to_blocked(c)
output = np.full(child.layout.blocked_shape,np.nan,np.float32)
cuda.memcpy_htod(parent.current,packed)
cuda.memcpy_htod(child.current,output)
calls, hop = [],run._hop
try:
run._hop = lambda *a,**kw: calls.append(a)
run._advance(index+1,parent.current,block=run._block,stream=run._stream)
run._stream.synchronize()
finally:
run._hop = hop
if calls != [(index+1,child.current,child.spare),(index+1,child.spare,child.current)]:
raise AssertionError('explosion changed two-hop launch contract')
cuda.memcpy_dtoh(output,child.current)
target = child.layout.owner.ravel()[geometry.ghost_fine]
slots = np.array([child.layout.slot(target,q) for q in range(9)])
actual = output.ravel()[slots]
expected = independent_explosion(c,geometry,parent.layout,run.hierarchy.kinds[index])
constant = c.reshape(9,-1)[:,geometry.ghost_coarse]
if not np.isfinite(actual).all() or np.max(abs(expected-constant)) <= 1e-5:
raise AssertionError('operator must exercise finite nonzero projected slopes')
np.testing.assert_allclose(actual,expected,atol=OPERATOR_ATOL,rtol=OPERATOR_RTOL)
untouched = np.ones(output.size,bool); untouched[slots.ravel()] = False
if not np.isnan(output.ravel()[untouched]).all():
raise AssertionError('explosion wrote outside its targets')
return dict(pair=index,rows=len(target),parent_stride=pair.parent_cells,
child_stride=int(np.prod(child.layout.local_shape)),
max_error_by_q=np.max(abs(actual-expected),axis=1).tolist(),
nonzero_projection=True,poisoned_nondonors=True,untouched_preserved=True)
def replay_two_hops(values, pair, layout, kind):
"""Existing FP64 two-hop algebra, using the independent coordinate oracle."""
from tests.unit.test_projected_explosion import collide
coarse,fine = (a.copy() for a in values)
fine.reshape(9,-1)[:,pair.ghost_fine] = independent_explosion(coarse,pair,layout,kind)
for _ in range(2):
fine = np.array([np.roll(fine[q],(ey,ex),(0,1)) for q,(ex,ey) in enumerate(E)])
fine[:,pair.fine_kind==2] = np.nan
active = pair.fine_kind==0
fine[:,active] = collide(fine[:,active],pair.tau_fine)
out = np.array([np.roll(coarse[q],(ey,ex),(0,1)) for q,(ex,ey) in enumerate(E)])
out.reshape(-1)[pair.coalesce_coarse] = fine.reshape(-1)[pair.coalesce_fine].mean(axis=1)
out[:,~pair.covered] = collide(out[:,~pair.covered],pair.tau_coarse)
out[:,pair.covered] = np.nan
return out,fine
def run_case(ctx, case, shifted, selector, raw, projected=False, replay=False):
from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
cfg,refs,blocks = case
cfg = replace(cfg,velocity=0,ddf_shifting=shifted)
with FixedPatchStepper(cfg,context=ctx,refinements=refs,level_blocks=blocks,
**default_options(selector)) as run:
operator = [operator_case(run,i) for i in range(len(run._pairs))] if projected else []
counts = [0]*len(run._levels)
for index,level in enumerate(run._levels):
def counted(*a, index=index, kernel=level.kernel, **kw):
counts[index] += 1
return kernel(*a,**kw)
level.kernel = counted
def check_counts(stop):
if counts != [stop*2**index for index in range(len(counts))]:
raise AssertionError('actual level hop counts differ from acoustic schedule')
rest = []
if projected:
initialize(run,rest=True)
for stop in STOPS:
run.step(stop-run.step_count)
check_counts(stop)
snapshot(run,raw) # actual public and raw invalid masks at every stop
maximum = 0.
for values,kind in zip(run.download(),run.hierarchy.kinds):
active = values[:,kind==0]
np.testing.assert_allclose(active,np.broadcast_to(W[:,None],active.shape),
atol=REST_ATOL,rtol=REST_RTOL)
maximum = max(maximum,float(np.max(abs(active-W[:,None]))))
rest.append(dict(step=stop,max_absolute_error=maximum))
counts[:] = [0]*len(counts)
initialize(run)
reference = tuple(a.astype(np.float64) for a in run.download()) if replay else None
snapshots,replay_errors,previous = [],[],0
for stop in STOPS:
run.step(stop-run.step_count)
if replay:
for _ in range(stop-previous):
reference = replay_two_hops(reference,run.hierarchy.pairs[0],
run._levels[0].layout,run.hierarchy.kinds[0])
errors = []
for actual,expected,kind in zip(run.download(),reference,run.hierarchy.kinds):
np.testing.assert_allclose(actual[:,kind==0],expected[:,kind==0],
atol=REPLAY_ATOL,rtol=0)
errors.append(np.max(abs(actual[:,kind==0]-expected[:,kind==0]),axis=1).tolist())
replay_errors.append(dict(step=stop,max_error_by_level_q=errors))
check_counts(stop)
snapshots.append(dict(snapshot(run,raw),observed_level_hops=counts.copy())); previous = stop
change = (np.array(snapshots[-1]['mass_momentum'])-
np.array(snapshots[0]['mass_momentum'])).tolist()
return dict(operator=operator,snapshots=snapshots,rest_fixed_point=rest,
replay_max_errors=replay_errors,mass_momentum_change_diagnostic=change,
conservation_status='diagnostic_only_no_FP32_bound')
def compare_snapshots(actual,expected):
if len(actual) != len(expected):
raise AssertionError('snapshot count differs')
for got,want in zip(actual,expected):
if (got['step'] != want['step'] or got['levels'] != want['levels']
or got.get('observed_level_hops') != want.get('observed_level_hops')):
raise AssertionError('owner bits, public values or masks differ')
def cases():
from CelerisLab.config import LBMConfig
from tests.unit.test_hierarchy import CFG, FOUR
scalar = (LBMConfig(nx=16,ny=16,viscosity=.02),
[(4,4,12,12),(8,8,16,16),(8,8,16,16)],
[(4,4),(4,8),(6,4),(4,6)])
four = (CFG,FOUR,[(6,8),(8,6),(12,8),(16,12)])
concave = (LBMConfig(nx=30,ny=30,viscosity=.1),
[[(4,4,5,7),(4,5,7,7)]],[(6,5),(4,6)])
return dict(scalar=scalar,four=four,concave=concave)
def baseline_receipt(receipt, driver_hash, keys):
if (receipt.get('completed') is not True or receipt.get('dry_run') is not False
or receipt.get('mode') != 'baseline' or receipt.get('production_revision') != BASE
or receipt.get('source',{}).get('revision') != BASE
or receipt.get('source',{}).get('dirty') is not False
or receipt.get('driver_sha256') != driver_hash
or receipt.get('stops') != list(STOPS)
or receipt.get('operator_tolerance') != [OPERATOR_ATOL,OPERATOR_RTOL]
or receipt.get('rest_tolerance') != [REST_ATOL,REST_RTOL]
or receipt.get('replay_tolerance') != [REPLAY_ATOL,0]
or set(receipt.get('default_snapshots',{})) != set(keys)
or not receipt.get('actual_build_sidecars')):
raise ValueError('baseline receipt does not match the frozen exact-dev gate')
return receipt
def remap_case(ctx, shifted, raw):
from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
cfg,refs,blocks = cases()['four']
with FixedPatchStepper(replace(cfg,velocity=0,ddf_shifting=shifted),context=ctx,
refinements=refs[:2],level_blocks=blocks[:3],explosion='projected_linear') as run:
initialize(run); run.step(1)
before = snapshot(run,raw)
with run.remap_child([(22,22,42,42),(72,68,76,76)]) as clone:
if clone._explosion != 'projected_linear' or any(p.neighbors is None for p in clone._pairs):
raise AssertionError('remap lost projected selection')
if clone.step_count != 1: raise AssertionError('remap lost synchronized count')
initial = snapshot(clone,raw)
clone.step(1)
after = snapshot(clone,raw)
if clone.step_count != 2: raise AssertionError('remap did not continue count')
compare_snapshots([snapshot(run,raw)],[before])
return dict(shifted=shifted,source_unchanged=True,initial=initial,continued=after)
@contextmanager
def capture_builds(report, save, baseline=None):
from CelerisLab.cuda import compiler_v2 as compiler
original = compiler.load_module
def load(path):
path = Path(path)
metadata = json.loads(Path(str(path)+'.json').read_text())
if metadata['ptx_sha256'] != hashlib.sha256(path.read_bytes()).hexdigest():
raise AssertionError('actual PTX and build sidecar disagree')
if baseline and (metadata['arch'],metadata['flags']) not in [
(m['arch'],m['flags']) for m in baseline['actual_build_sidecars']]:
raise AssertionError('baseline build architecture or flags differ')
report['actual_build_sidecars'].append(metadata)
save() # capture AMR metadata before production cleanup removes scratch
return original(str(path))
compiler.load_module = load
try:
yield
finally:
compiler.load_module = original
def preflight(repo, revision, mode):
git = lambda *a: subprocess.check_output(['git','-C',str(repo),*a],text=True).strip()
if git('rev-parse','HEAD') != revision or git('status','--porcelain'):
raise ValueError('requires the exact expected clean checkout')
source = BASE if mode == 'baseline' else PRODUCTION
if mode == 'baseline' and revision != BASE:
raise ValueError('baseline must be exact ed9d8c9')
if subprocess.call(['git','-C',str(repo),'diff','--quiet',source,'--','src']):
raise ValueError('production source differs from the frozen source revision')
return source
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--repo',type=Path,required=True)
parser.add_argument('--expected-revision',required=True)
parser.add_argument('--mode',choices=('baseline','candidate'),required=True)
parser.add_argument('--baseline',type=Path)
parser.add_argument('--output',type=Path,required=True)
parser.add_argument('--run-gpu',action='store_true')
parser.add_argument('--dry-run',action='store_true')
args = parser.parse_args()
if args.output.exists(): parser.error('output already exists')
selected = os.environ.get('CUDA_VISIBLE_DEVICES','')
if not args.dry_run and (not args.run_gpu or not selected or ',' in selected or selected=='-1'):
parser.error('--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required')
repo = args.repo.resolve()
source = preflight(repo,args.expected_revision,args.mode)
baseline = None
if args.mode == 'candidate':
if args.baseline is None: parser.error('candidate requires --baseline receipt')
baseline = json.loads(args.baseline.read_text())
sys.path[:0] = [str(repo/'src'),str(repo),str(repo/'tests/validation/multigrid'),
str(repo/'tests/validation/taylor_green')]
import CelerisLab
if Path(CelerisLab.__file__).resolve() != repo/'src/CelerisLab/__init__.py':
raise RuntimeError('wrong imported checkout')
from run_case import provenance
driver_hash = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
keys = [f'{name}:shifted={shifted}' for name in cases() for shifted in (False,True)]
if baseline is not None: baseline_receipt(baseline,driver_hash,keys)
report = dict(mode=args.mode,production_revision=source,source=provenance(repo),
driver_sha256=driver_hash,
device_selection=selected,completed=False,dry_run=args.dry_run,
operator_tolerance=[OPERATOR_ATOL,OPERATOR_RTOL],
rest_tolerance=[REST_ATOL,REST_RTOL],replay_tolerance=[REPLAY_ATOL,0],stops=STOPS,
conservation_status='diagnostic_only_no_FP32_bound',
actual_build_sidecars=[],default_snapshots={},default_equivalence=[],
projected_operator=[],union_multilevel_smoke=[],changed_remap=[])
args.output.parent.mkdir(parents=True,exist_ok=True)
def save():
text = json.dumps(report,indent=2,allow_nan=False)+'\n'
if len(text.encode()) > 8*1024*1024: raise RuntimeError('receipt exceeds 8 MiB cap')
temporary = args.output.with_suffix(args.output.suffix+'.tmp')
temporary.write_text(text); os.replace(temporary,args.output)
save()
if args.dry_run:
print('Source preflight only; no CUDA imports or execution.'); return
try:
from CelerisLab.cuda.context import CudaContext
from test_refinement_blocked_base import raw
with CudaContext(0) as ctx, capture_builds(report,save,baseline):
import pycuda.driver as cuda
report['device'] = dict(name=ctx.device.name(),pci_bus_id=ctx.device.pci_bus_id(),
capability=list(ctx.device.compute_capability()),
driver=cuda.get_driver_version(),runtime=list(cuda.get_version()))
nvcc = shutil.which('nvcc') or '/usr/local/cuda/bin/nvcc'
report['nvcc_version'] = subprocess.check_output([nvcc,'--version'],text=True).strip()
if baseline and (report['device'] != baseline.get('device')
or report['nvcc_version'] != baseline.get('nvcc_version')
or selected != baseline.get('device_selection')):
raise ValueError('baseline GPU/device selection or toolchain differs')
for name,case in cases().items():
for shifted in (False,True):
key = f'{name}:shifted={shifted}'
report['active_case'] = f'default:{key}'
result = run_case(ctx,case,shifted,'omitted',raw)
report['default_snapshots'][key] = result['snapshots']
if baseline:
compare_snapshots(result['snapshots'],baseline['default_snapshots'][key])
explicit = run_case(ctx,case,shifted,'constant',raw)
compare_snapshots(explicit['snapshots'],result['snapshots'])
report['default_equivalence'].append(dict(case=key,passed=True))
save()
if baseline:
for name,case in cases().items():
for shifted in (False,True):
report['active_case'] = f'projected:{name}:shifted={shifted}'
result = run_case(ctx,case,shifted,'projected_linear',raw,True,name=='concave')
cfg,refs,blocks = case
if name == 'four':
variant_refs = [list(reversed(rects)) for rects in refs]
elif name == 'concave':
variant_refs = [[(4,5,7,7),(4,4,5,5)]] # same union, disjoint decomposition
else:
variant_refs = refs
variant = (cfg,variant_refs,[(2,2)]*len(blocks))
alternate = run_case(ctx,variant,shifted,'projected_linear',raw,True)
compare_snapshots(alternate['snapshots'],result['snapshots'])
report['projected_operator'].append(dict(case=name,shifted=shifted,rows=result['operator']))
report['union_multilevel_smoke'].append(dict(case=name,shifted=shifted,result=result,
alternate_bits_identical=True))
save()
for shifted in (False,True):
report['active_case'] = f'changed-remap:shifted={shifted}'
report['changed_remap'].append(remap_case(ctx,shifted,raw)); save()
expected_builds = {(s['arch'],tuple(s['flags'])) for s in baseline['actual_build_sidecars']}
actual_builds = {(s['arch'],tuple(s['flags'])) for s in report['actual_build_sidecars']}
if actual_builds != expected_builds:
raise AssertionError('baseline build architecture or compiler flags differ')
report['completed'] = True
save()
except Exception as error:
report['error'] = f'{type(error).__name__}: {error}'
save(); raise
print(json.dumps(dict(completed=True,mode=args.mode,receipt=str(args.output))))
if __name__ == '__main__':
main()