153 lines
7.2 KiB
Python
153 lines
7.2 KiB
Python
"""CPU-only regressions for configuration and isolated compilation inputs."""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from threading import Barrier
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from CelerisLab.config import LBMConfig, load_lbm_config
|
|
from CelerisLab.cuda import compiler
|
|
|
|
|
|
class ConfigContracts(unittest.TestCase):
|
|
def test_explicit_missing_config_never_falls_back(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
with self.assertRaises(FileNotFoundError):
|
|
load_lbm_config(str(Path(d) / "missing.json"))
|
|
with self.assertRaises(FileNotFoundError):
|
|
load_lbm_config("")
|
|
|
|
def test_float_literals_preserve_selected_physics(self):
|
|
cfg = LBMConfig(rho=1.04, omega_min=.01234, outlet_blend_alpha=.712345)
|
|
macros = cfg.to_macros()
|
|
self.assertEqual(float(macros["RHO"]), cfg.rho)
|
|
self.assertEqual(float(macros["OMEGA_COLLISION_MIN"].rstrip("f")), cfg.omega_min)
|
|
self.assertEqual(float(macros["OUTLET_BLEND_ALPHA"].rstrip("f")), cfg.outlet_blend_alpha)
|
|
cfg.rho = .04
|
|
self.assertEqual(float(cfg.to_macros()["RHO"]), .04)
|
|
|
|
def test_invalid_inputs_raise_even_without_assertions(self):
|
|
for kwargs in ({"nx": 0}, {"threads_per_block": 0}, {"threads_per_block": 2048},
|
|
{"rho": -1}, {"rho": float("nan")}, {"viscosity": float("inf")},
|
|
{"inlet_profile": "missing"}, {"outlet_mode": "missing"}):
|
|
with self.subTest(kwargs=kwargs), self.assertRaises(ValueError):
|
|
LBMConfig(**kwargs).validate()
|
|
|
|
def test_d3_schema_remains_distinct_from_runtime_capability(self):
|
|
cfg = LBMConfig(lattice_model="D3Q19", nz=8)
|
|
cfg.validate()
|
|
self.assertEqual((cfg.dim, cfg.nq), (3, 19))
|
|
with self.assertRaisesRegex(NotImplementedError, "dimension-complete"):
|
|
cfg.validate_runtime()
|
|
|
|
def test_runtime_grid_requires_interior_and_ghost_cells(self):
|
|
for kwargs in ({"nx": 1}, {"nx": 2}, {"ny": 1}, {"ny": 2}, {"ny": 3}):
|
|
with self.subTest(kwargs=kwargs):
|
|
cfg = LBMConfig(**kwargs)
|
|
cfg.validate()
|
|
with self.assertRaisesRegex(ValueError, "wall-adjacent rows"):
|
|
cfg.validate_runtime()
|
|
LBMConfig(nx=3, ny=4).validate_runtime()
|
|
|
|
def test_allocation_signature_tracks_layout_not_collision(self):
|
|
cfg = LBMConfig()
|
|
signature = cfg.allocation_signature()
|
|
cfg.collision = "MRT"
|
|
cfg.threads_per_block = 128
|
|
self.assertEqual(signature, cfg.allocation_signature())
|
|
for name, value in (("nx", 1024), ("store_precision", "FP16S"),
|
|
("streaming", "esopull"), ("ddf_shifting", True)):
|
|
changed = LBMConfig(**{name: value})
|
|
self.assertNotEqual(signature, changed.allocation_signature())
|
|
|
|
def test_cpu_imports_do_not_require_pycuda(self):
|
|
# Refuse the dependency even on development machines with it installed.
|
|
code = '''
|
|
import sys
|
|
sys.modules["pycuda"] = None
|
|
from CelerisLab.config import LBMConfig
|
|
from CelerisLab.lbm.equilibrium import compute_feq_d2q9
|
|
from CelerisLab.body.geometry.circle import CircleGeometry
|
|
from CelerisLab.cuda import compiler
|
|
assert "pycuda.driver" not in sys.modules
|
|
assert abs(sum(compute_feq_d2q9(1.,0.,0.)) - 1.) < 1e-6
|
|
'''
|
|
subprocess.run([sys.executable, "-c", code], check=True)
|
|
|
|
|
|
class BuildContracts(unittest.TestCase):
|
|
@staticmethod
|
|
def fake_compile(command, check):
|
|
source = Path(command[-3])
|
|
grid = (source.parent / "config/config_grid.h").read_text()
|
|
objects = (source.parent / "config/config_objects.h").read_text()
|
|
data = {"nx": int(re.search(r"#define NX\s+(\d+)", grid)[1]),
|
|
"objects": int(re.search(r"#define N_OBJS\s+(\d+)", objects)[1])}
|
|
Path(command[-1]).write_text(json.dumps(data))
|
|
|
|
def test_parallel_builds_never_share_headers_or_outputs(self):
|
|
sources = list(Path(compiler.kernel_path("config")).glob("*.h"))
|
|
before = {p: hashlib.sha256(p.read_bytes()).hexdigest() for p in sources}
|
|
barrier = Barrier(2)
|
|
with tempfile.TemporaryDirectory() as root:
|
|
def work(nx, count):
|
|
build = compiler.generate_config(LBMConfig(nx=nx), count, build_root=root)
|
|
barrier.wait()
|
|
# Compatibility form is context-local even without explicit path.
|
|
output = compiler.compile_kernel()
|
|
data = json.loads(Path(output).read_text())
|
|
self.assertEqual(data, {"nx": nx, "objects": count})
|
|
metadata = json.loads(Path(output + ".json").read_text())
|
|
self.assertEqual(metadata["config"]["nx"], nx)
|
|
compiler.cleanup_build_dir(build)
|
|
return output
|
|
with patch.object(compiler.subprocess, "run", self.fake_compile):
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
a, b = pool.submit(work, 64, 1), pool.submit(work, 256, 3)
|
|
self.assertNotEqual(a.result(), b.result())
|
|
self.assertEqual(before, {p: hashlib.sha256(p.read_bytes()).hexdigest() for p in sources})
|
|
|
|
def test_explicit_build_survives_interleaved_generation(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
a = compiler.generate_config(LBMConfig(nx=64), 1, build_root=root)
|
|
b = compiler.generate_config(LBMConfig(nx=128), 2, build_root=root)
|
|
with patch.object(compiler.subprocess, "run", self.fake_compile):
|
|
output = compiler.compile_kernel(build_dir=a)
|
|
self.assertEqual(json.loads(Path(output).read_text())["nx"], 64)
|
|
self.assertTrue(Path(b, "config/config_grid.h").exists())
|
|
with self.assertRaises(ValueError):
|
|
compiler.compile_kernel(build_dir=a, output="../unsafe.ptx")
|
|
with self.assertRaises(ValueError):
|
|
compiler.cleanup_build_dir(compiler.kernel_path(""))
|
|
compiler.cleanup_build_dir(a)
|
|
compiler.cleanup_build_dir(a)
|
|
compiler.cleanup_build_dir(b)
|
|
with self.assertRaises(RuntimeError):
|
|
compiler.compile_kernel()
|
|
|
|
def test_failed_compile_cannot_implicitly_load_previous_artifact(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
build = compiler.generate_config(LBMConfig(nx=64), build_root=root)
|
|
with patch.object(compiler.subprocess, "run", self.fake_compile):
|
|
previous = compiler.compile_kernel(build_dir=build)
|
|
old_bytes = Path(previous).read_bytes()
|
|
failure = subprocess.CalledProcessError(1, "nvcc")
|
|
with patch.object(compiler.subprocess, "run", side_effect=failure):
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
compiler.compile_kernel(build_dir=build)
|
|
self.assertEqual(Path(previous).read_bytes(), old_bytes)
|
|
with self.assertRaisesRegex(RuntimeError, "No compiled PTX"):
|
|
compiler.load_module()
|
|
compiler.cleanup_build_dir(build)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|