ClaudeandClaude Opus 5.5
0d62c3cfb8
Simulation, PeriodicBlockStepper and FixedPatchStepper levels each repeated generate_config -> compile_kernel -> load_module -> cleanup and the d_params pack/upload. They now call compiler.build_module(cfg, entry, arch=..., n_objects=..., block_root_bc=...) and compiler.set_params. Simulation no longer keeps its build directory alive until close (nothing read it after loading). Removes the unused compile_kernel_v2 alias. Host fakes and lifecycle tests follow the new seam; the close-retry case for build-directory cleanup goes with the removed cleanup step. No kernel, numerical or launch change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
"""Simulation cleanup orchestration, using a CPU owner model (no GPU calls)."""
|
|
from contextlib import contextmanager
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("pycuda.driver")
|
|
from CelerisLab import simulation as module
|
|
from CelerisLab.config import LBMConfig, BodyConfig
|
|
|
|
|
|
class Owner:
|
|
def __init__(self, *args):
|
|
self.is_active = False
|
|
self.is_closed = False
|
|
self.sm_arch = "sm_70"
|
|
self.detach_failure = False
|
|
|
|
@contextmanager
|
|
def activate(self):
|
|
assert not self.is_closed
|
|
self.is_active = True
|
|
try:
|
|
yield self
|
|
finally:
|
|
self.is_active = False
|
|
|
|
def close(self):
|
|
if self.detach_failure:
|
|
self.detach_failure = False
|
|
raise RuntimeError("detach failed")
|
|
self.is_closed = True
|
|
|
|
|
|
def make_simulation(monkeypatch):
|
|
sim = module.Simulation.__new__(module.Simulation)
|
|
sim.ctx = Owner()
|
|
sim._closed = False
|
|
for name in ("stepper", "bodies", "field", "_module", "_stream"):
|
|
setattr(sim, name, object())
|
|
monkeypatch.setattr(module, "cuda", SimpleNamespace(
|
|
Context=SimpleNamespace(synchronize=lambda: None)))
|
|
return sim
|
|
|
|
|
|
def test_close_inside_activation_preserves_resources(monkeypatch):
|
|
sim = make_simulation(monkeypatch)
|
|
original_field = sim.field
|
|
with sim.ctx.activate():
|
|
with pytest.raises(RuntimeError, match="active CUDA"):
|
|
sim.close()
|
|
assert sim.field is original_field
|
|
assert not sim._closed
|
|
sim.close()
|
|
assert sim._closed and sim.ctx.is_closed
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["synchronize", "detach"])
|
|
def test_close_failure_is_retryable(monkeypatch, failure):
|
|
sim = make_simulation(monkeypatch)
|
|
if failure == "detach":
|
|
sim.ctx.detach_failure = True
|
|
else:
|
|
calls = []
|
|
def fail_once(*args):
|
|
calls.append(args)
|
|
if len(calls) == 1:
|
|
raise RuntimeError(failure + " failed")
|
|
monkeypatch.setattr(module.cuda.Context, "synchronize", fail_once)
|
|
with pytest.raises(RuntimeError, match="failed"):
|
|
sim.close()
|
|
assert not sim._closed
|
|
sim.close()
|
|
assert sim._closed and sim.ctx.is_closed
|
|
sim.close()
|
|
|
|
|
|
def test_constructor_cleanup_preserves_original_failure(monkeypatch):
|
|
original = RuntimeError("compile setup failed")
|
|
def fail_build(*args, **kwargs):
|
|
raise original
|
|
def fail_close(self):
|
|
raise RuntimeError("cleanup failed")
|
|
monkeypatch.setattr(module, "load_lbm_config", lambda path: LBMConfig())
|
|
monkeypatch.setattr(module, "load_body_config", lambda path: BodyConfig())
|
|
monkeypatch.setattr(module, "CudaContext", Owner)
|
|
monkeypatch.setattr(module.compiler, "generate_config", fail_build)
|
|
monkeypatch.setattr(module.Simulation, "close", fail_close)
|
|
with pytest.raises(RuntimeError) as exc:
|
|
module.Simulation()
|
|
assert exc.value is original
|
|
|
|
|
|
@pytest.mark.parametrize("name,value,error", [
|
|
("steps", -1, ValueError), ("steps", 1.5, TypeError),
|
|
("checkpoint_interval", -1, ValueError),
|
|
("checkpoint_interval", 1.5, TypeError),
|
|
])
|
|
def test_invalid_step_request_has_no_state_side_effects(monkeypatch, name, value, error):
|
|
sim = make_simulation(monkeypatch)
|
|
sim._initialized = True
|
|
calls = []
|
|
sim._assert_runtime_contracts = lambda: calls.append("accessed runtime")
|
|
kwargs = {"steps": 1, name: value}
|
|
with pytest.raises(error):
|
|
sim.run(**kwargs)
|
|
assert calls == [] # no pending edits discarded, obs cleared or GPU work queued
|
|
sim.close()
|