146 lines
6.8 KiB
Python
146 lines
6.8 KiB
Python
"""Explicit tiny CUDA explosion parity and constructor/one-interval smoke check.
|
|
|
|
No installs or device selection defaults. Run only on the authorized GPU host.
|
|
This is operator/runtime evidence, not Taylor-Green accuracy qualification.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
import numpy as np
|
|
|
|
|
|
def check_case(ctx, shifted, projected):
|
|
import pycuda.driver as cuda
|
|
from CelerisLab.config import LBMConfig
|
|
from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
|
|
from CelerisLab.lbm.refinement_tables import projected_explosion_tables
|
|
from CelerisLab.lbm.refinement import ACTIVE
|
|
from test_projected_explosion import reconstruct, W
|
|
|
|
cfg = LBMConfig(nx=16, ny=16, viscosity=.1, velocity=0,
|
|
ddf_shifting=shifted)
|
|
cfg.validate()
|
|
options = {"explosion": "projected_linear"} if projected else {}
|
|
# Omitting explosion in the constant case tests the production default.
|
|
with FixedPatchStepper(cfg, context=ctx, refinements=[[(4, 4, 12, 12)]],
|
|
level_blocks=[(4, 4), (8, 4)], **options) as run:
|
|
parent, child = run._levels
|
|
pair = run._pairs[0]
|
|
patch = run.hierarchy.pairs[0]
|
|
coarse_cells = int(np.prod(parent.layout.local_shape))
|
|
fine_cells = int(np.prod(child.layout.local_shape))
|
|
assert coarse_cells != fine_cells
|
|
assert pair.parent_cells == coarse_cells
|
|
assert len(pair.interpolation) == (2 if projected else 0)
|
|
donor = patch.ghost_coarse.astype(np.intp)
|
|
# Random variation must exercise the gradient, not just a constant field.
|
|
c = np.random.default_rng(11).normal(0, .001, (9, 16, 16))
|
|
if not shifted:
|
|
c += W[:, None, None]
|
|
c = c.astype(np.float32)
|
|
legal = np.zeros(256, bool)
|
|
legal[donor] = True
|
|
c.reshape(9, -1)[:, ~legal] = np.nan
|
|
coarse = np.full(np.prod(parent.layout.blocked_shape), np.nan, np.float32)
|
|
for q in range(9):
|
|
coarse[parent.layout.slot(parent.layout.owner.ravel()[legal], q)] = c.reshape(9, -1)[q, legal]
|
|
fine = np.full(np.prod(child.layout.blocked_shape), np.nan, np.float32)
|
|
cuda.memcpy_htod(parent.current, coarse)
|
|
cuda.memcpy_htod(child.current, fine)
|
|
calls = []
|
|
real_hop = run._hop
|
|
try:
|
|
# Execute the production explosion launch exactly; isolate it from
|
|
# subsequent streaming/collision without replacing the kernel/tables.
|
|
run._hop = lambda *args, **kwargs: calls.append(args)
|
|
run._advance(1, parent.current, block=run._block, stream=run._stream)
|
|
run._stream.synchronize()
|
|
finally:
|
|
run._hop = real_hop
|
|
assert calls == [(1, child.current, child.spare), (1, child.spare, child.current)]
|
|
cuda.memcpy_dtoh(fine, child.current)
|
|
targets = child.layout.owner.ravel()[patch.ghost_fine]
|
|
slots = np.array([child.layout.slot(targets, q) for q in range(9)])
|
|
actual = fine[slots]
|
|
neighbors, offsets = projected_explosion_tables(patch, parent.layout)
|
|
inverse = np.full(parent.layout.nblocks * coarse_cells, -1, dtype=int)
|
|
inverse[parent.layout.owner.ravel()] = np.arange(256)
|
|
constant = c.reshape(9, -1)[:, donor].astype(float)
|
|
expected = reconstruct(c.astype(float), patch, inverse[neighbors], offsets)[0] if projected else constant
|
|
assert np.isfinite(actual).all()
|
|
if projected:
|
|
assert np.max(np.abs(expected - constant)) > 1e-5
|
|
np.testing.assert_allclose(actual, expected, atol=6e-8, rtol=2e-7)
|
|
untouched = np.ones(fine.size, bool)
|
|
untouched[slots.ravel()] = False
|
|
assert np.isnan(fine[untouched]).all(), "explosion wrote outside target slots"
|
|
error = float(np.max(np.abs(actual - expected)))
|
|
|
|
# Restore ordinary execution and exercise real constructor wiring,
|
|
# stream-only ghost hops, restriction, halo copies and collision once.
|
|
initial = [np.broadcast_to(W[:, None, None], (9, *k.shape)).copy()
|
|
for k in run.hierarchy.kinds]
|
|
run.initialize(*initial)
|
|
run.step(1)
|
|
for values, kind in zip(run.download(), run.hierarchy.kinds):
|
|
active = values[:, kind == ACTIVE]
|
|
assert np.isfinite(active).all()
|
|
np.testing.assert_allclose(active, np.broadcast_to(W[:, None], active.shape),
|
|
atol=3e-7, rtol=1e-6)
|
|
assert run.step_count == 1
|
|
return dict(shifted=shifted, explosion="projected_linear" if projected else "constant_default",
|
|
parent_stride=coarse_cells, child_stride=fine_cells,
|
|
compared_slots=int(actual.size), max_abs_error=error,
|
|
poisoned_nondonors=True, untouched_slots_preserved=True, equilibrium_step=True)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--repo", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
parser.add_argument("--run-gpu", action="store_true", help="Explicitly opt into GPU execution")
|
|
args = parser.parse_args()
|
|
selected = os.environ.get("CUDA_VISIBLE_DEVICES", "")
|
|
if not args.run_gpu or not selected or "," in selected or selected == "-1":
|
|
parser.error("--run-gpu and exactly one explicit physical CUDA_VISIBLE_DEVICES are required")
|
|
if args.output.exists():
|
|
parser.error("output already exists; choose a new receipt path")
|
|
root = args.repo.resolve()
|
|
sys.path[:0] = [str(root / "src"), str(root / "tests/unit"),
|
|
str(root / "tests/validation/taylor_green")]
|
|
import CelerisLab
|
|
if Path(CelerisLab.__file__).resolve() != root / "src/CelerisLab/__init__.py":
|
|
raise RuntimeError("Wrong imported source checkout")
|
|
from run_case import provenance
|
|
report = dict(source=provenance(root),
|
|
numpy_oracle_sha256=hashlib.sha256((root / "tests/unit/test_projected_explosion.py").read_bytes()).hexdigest(),
|
|
cuda_visible_devices=selected,
|
|
logical_device=0, completed=False, cases=[])
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
def save():
|
|
args.output.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n")
|
|
save()
|
|
try:
|
|
from CelerisLab.cuda.context import CudaContext
|
|
with CudaContext(0) as ctx:
|
|
report["device"] = ctx.device.name()
|
|
for shifted in (False, True):
|
|
for projected in (False, True):
|
|
report["cases"].append(check_case(ctx, shifted, projected))
|
|
save()
|
|
report["completed"] = True
|
|
save()
|
|
except Exception as exc:
|
|
report["error"] = f"{type(exc).__name__}: {exc}"
|
|
save()
|
|
raise
|
|
print(json.dumps(report, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|