163 lines
9.5 KiB
Python
163 lines
9.5 KiB
Python
"""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()
|