ClaudeandClaude Opus 5.5
7ecb179a41
Schedule test drives _pairs[i].explode/coalesce and patches _advance(index, buffer) for the stale L1->L2 feed mutant; remap rejection test patches _boundary_hosts entries and hierarchy.pairs instead of the read-only views; the GPU launch-failure test patches _pairs[0].explode. Assertions unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
316 lines
17 KiB
Python
316 lines
17 KiB
Python
"""CPU-only union geometry and production host-control-flow tests.
|
|
|
|
The AST loader omits CUDA imports and supplies explicit mocks; it executes
|
|
unchanged production definitions, never CUDA kernels or a numerical emulator.
|
|
"""
|
|
import ast
|
|
from contextlib import nullcontext
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
import types
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
import numpy as np
|
|
from CelerisLab.config import LBMConfig
|
|
from CelerisLab.lbm import refinement as r
|
|
|
|
|
|
def host_module():
|
|
path=Path(r.__file__).with_name('refinement_stepper.py')
|
|
tree=ast.parse(path.read_text())
|
|
tree.body=[node for node in tree.body if not (
|
|
isinstance(node,ast.Import) and any(a.name=='pycuda.driver' for a in node.names)
|
|
or isinstance(node,ast.ImportFrom) and node.module in (
|
|
'cuda','block_stepper','curved_links'))]
|
|
module=types.ModuleType('CelerisLab.lbm._test_host_refinement')
|
|
module.__package__='CelerisLab.lbm'
|
|
from CelerisLab.body.jets import surface_link_velocity
|
|
module.__dict__.update(cuda=Mock(),compiler=Mock(),MAX_GRID_DIM=65535,
|
|
surface_link_velocity=surface_link_velocity)
|
|
exec(compile(tree,str(path),'exec'),module.__dict__)
|
|
return module
|
|
|
|
|
|
host=host_module()
|
|
CFG = LBMConfig(nx=48, ny=32, viscosity=.1)
|
|
PARENTS = ((4,4,20,28), (24,4,44,28)) # exact gap=4 coarse cells
|
|
CHILDREN = ((16,16,32,40), (56,16,80,40)) # parent-allocated L1 cells
|
|
|
|
|
|
class Gate(unittest.TestCase):
|
|
def setUp(self):
|
|
self.parent = r.FixedPatchSet2D.from_config(CFG, PARENTS)
|
|
|
|
def test_one_element_equivalence(self):
|
|
cfg = LBMConfig(nx=16,ny=16,viscosity=.1)
|
|
local = r.FixedPatch2D.from_config(cfg,(4,4,12,12))
|
|
global_parent = r.FixedPatchSet2D.from_config(cfg,[(4,4,12,12)])
|
|
a, ak = r.nest_patch(cfg,local,(8,8,16,16))
|
|
b, bk = r.nest_patch_set(cfg,global_parent,[(12,12,20,20)])
|
|
local_set, local_kind = r.nest_patch_set(cfg,local,[(8,8,16,16)])
|
|
np.testing.assert_array_equal(ak,local_kind)
|
|
np.testing.assert_array_equal(a.fine_kind,local_set.patches[0].fine_kind)
|
|
np.testing.assert_array_equal(a.ghost_coarse,local_set.ghost_coarse)
|
|
np.testing.assert_array_equal(a.coalesce_coarse,local_set.coalesce_coarse)
|
|
np.testing.assert_array_equal(ak, bk[4:28,4:28])
|
|
# L2 node-index offsets in global L2 grid: local parent origin=2 L0.
|
|
ay,ax=np.nonzero(a.fine_kind == r.ACTIVE)
|
|
by,bx=np.nonzero(b.fine_kind == r.ACTIVE)
|
|
np.testing.assert_array_equal(np.stack([ay+20,ax+20],-1),np.stack([by,bx],-1))
|
|
np.testing.assert_array_equal(a.fine_kind,b.fine_kind[20:44,20:44])
|
|
# Compare complete per-region maps after undoing set addressing.
|
|
bp=b.patches[0]
|
|
np.testing.assert_array_equal(a.coalesce_fine,bp.coalesce_fine)
|
|
aq,ac=np.divmod(a.coalesce_coarse,24*24)
|
|
bq,bc=np.divmod(bp.coalesce_coarse,32*32)
|
|
np.testing.assert_array_equal(aq,bq)
|
|
np.testing.assert_array_equal(np.stack(np.divmod(ac,24),-1)+4,np.stack(np.divmod(bc,32),-1))
|
|
ay,ax=np.divmod(a.ghost_coarse,24)
|
|
by,bx=np.divmod(bp.ghost_coarse,32)
|
|
np.testing.assert_array_equal(np.stack([ay+4,ax+4],-1),np.stack([by,bx],-1))
|
|
|
|
def test_twin_tile_chains_and_order(self):
|
|
children=((16,16,32,40),(56,16,72,40))
|
|
child,kind=r.nest_patch_set(CFG,self.parent,children)
|
|
other,ok=r.nest_patch_set(CFG,r.FixedPatchSet2D.from_config(CFG,PARENTS[::-1]),children[::-1])
|
|
for name in ('fine_kind','ghost_fine','ghost_coarse','coalesce_coarse','coalesce_fine'):
|
|
np.testing.assert_array_equal(getattr(child,name),getattr(other,name))
|
|
np.testing.assert_array_equal(kind,ok)
|
|
for p in child.patches:
|
|
self.assertEqual(p.fine_shape,child.patches[0].fine_shape)
|
|
np.testing.assert_array_equal(child.patches[0].fine_kind,child.patches[1].fine_kind)
|
|
self.assertEqual(child.tau_coarse,self.parent.tau_fine)
|
|
self.assertAlmostEqual(child.tau_fine,1.7)
|
|
|
|
def test_asymmetric_no_child_region(self):
|
|
child,kind=r.nest_patch_set(CFG,self.parent,[CHILDREN[0]])
|
|
np.testing.assert_array_equal(kind[:,48:],self.parent.fine_kind[:,48:])
|
|
self.assertEqual(int(child.covered.sum()),16*24)
|
|
|
|
def test_two_children_one_parent(self):
|
|
child,kind=r.nest_patch_set(CFG,self.parent,[(16,16,32,24),(16,28,32,40)])
|
|
self.assertEqual(len(child.patches),2)
|
|
self.assertEqual(int(child.covered.sum()),16*20)
|
|
self.assertEqual(int((kind==r.ACTIVE).sum()),int((self.parent.fine_kind==r.ACTIVE).sum())-16*20)
|
|
|
|
def test_holes_donors_receivers_footprint_and_invalid_pull(self):
|
|
child,_=r.nest_patch_set(CFG,self.parent,CHILDREN)
|
|
cases=[('inherited ACTIVE', lambda: r.nest_patch_set(CFG,self.parent,[(40,16,48,40)])),
|
|
('donors/receivers', lambda: r.nest_patch_set(CFG,self.parent,[(8,16,32,40)]))]
|
|
for text,call in cases:
|
|
with self.subTest(text=text),self.assertRaisesRegex(ValueError,text):call()
|
|
# Synthetic upstream dependency inside child coverage isolates the final guard.
|
|
upstream=self.parent.coalesce_fine.copy()
|
|
upstream[0,:]=20*self.parent.fine_shape[1]+20 # q=0, deep inherited ACTIVE
|
|
with self.assertRaisesRegex(ValueError,'restriction footprint'):
|
|
r._nested_kind(replace(self.parent,coalesce_fine=upstream),child)
|
|
hole=int(np.flatnonzero(~self.parent.fine_layout((4,4)).owned.ravel())[0])
|
|
donors=child.ghost_coarse.copy();donors[0]=hole
|
|
with self.assertRaisesRegex(ValueError,'donors/receivers'):
|
|
r._nested_kind(self.parent,replace(child,ghost_coarse=donors))
|
|
receivers=child.coalesce_coarse.copy();receivers[0]=hole
|
|
with self.assertRaisesRegex(ValueError,'donors/receivers'):
|
|
r._nested_kind(self.parent,replace(child,coalesce_coarse=receivers))
|
|
# Remove all supplying rows; covered-to-uncovered pulls must be rejected.
|
|
with self.assertRaisesRegex(ValueError,'surviving parent ACTIVE pull'):
|
|
r._nested_kind(self.parent,replace(child,coalesce_coarse=child.coalesce_coarse[:0],coalesce_fine=child.coalesce_fine[:0]))
|
|
|
|
def test_existing_l2_body_jet_preparation_on_union(self):
|
|
from CelerisLab.body.geometry.circle import CircleGeometry
|
|
from CelerisLab.body.objects import SimObject
|
|
from CelerisLab.body.jets import Jet, Tube, Parallel, jet_link_shapes
|
|
child,_=r.nest_patch_set(CFG,self.parent,CHILDREN)
|
|
centers=((12.,14.),(34.,14.))
|
|
objects=[SimObject(99,geometry=CircleGeometry(*c,.8),center=c,radius=.8) for c in centers]
|
|
jets=[Jet(Tube((c[0]-1.5,c[1]),0.,3.,.3),Parallel(0.),amplitude=.002) for c in centers]
|
|
middle=replace(CFG,nx=self.parent.fine_shape[1],ny=self.parent.fine_shape[0],viscosity=.2)
|
|
boundary=r.prepare_fine_boundary(middle,child,objects,jets=jets,
|
|
parent_origin=self.parent.fine_origin,parent_spacing=.5)
|
|
ids=boundary.curved_host[3]
|
|
np.testing.assert_array_equal(np.unique(ids),[0,1])
|
|
self.assertEqual(int((ids==0).sum()),int((ids==1).sum()))
|
|
np.testing.assert_allclose(boundary.curved_hits[ids==1]-boundary.curved_hits[ids==0],
|
|
np.broadcast_to([22.,0.],boundary.curved_hits[ids==0].shape),atol=1e-14,rtol=0)
|
|
_,jid=jet_link_shapes(jets,boundary.curved_hits)
|
|
for i in (0,1):
|
|
self.assertTrue((ids[jid==i]==i).all())
|
|
self.assertTrue((jid==i).any())
|
|
|
|
def test_sparse_parent_tables_preserve_inherited_kinds(self):
|
|
child,kind=r.nest_patch_set(CFG,self.parent,CHILDREN)
|
|
layout=self.parent.fine_layout((4,4));order=np.argsort(child.coalesce_coarse)
|
|
donors,targets,bkind=host.blocked_base_tables(child,layout,order,composed_kind=kind)
|
|
self.assertTrue((donors < layout.hole).all())
|
|
self.assertTrue((targets < np.prod(layout.blocked_shape)).all())
|
|
np.testing.assert_array_equal(bkind[layout.owner[layout.owned]],kind[layout.owned])
|
|
inherited=self.parent.fine_kind != r.ACTIVE
|
|
np.testing.assert_array_equal(kind[inherited],self.parent.fine_kind[inherited])
|
|
self.assertTrue((kind[child.covered] == r.OUTER_GHOST).all())
|
|
l2=child.fine_layout((4,4))
|
|
ghosts,children,ckind=host.blocked_fine_tables(child,l2,order)
|
|
self.assertTrue((ghosts<l2.hole).all())
|
|
self.assertTrue((children<np.prod(l2.blocked_shape)).all())
|
|
for p,lay in ((self.parent,layout),(child,l2)):
|
|
self.assertTrue(lay.owned.ravel()[p.ghost_fine].all())
|
|
self.assertTrue(lay.owned.ravel()[p.coalesce_fine % np.uint64(np.prod(p.fine_shape))].all())
|
|
hole=int(np.flatnonzero(~layout.owned.ravel())[0])
|
|
bad=child.ghost_coarse.copy();bad[0]=hole
|
|
with self.assertRaisesRegex(ValueError,'owned uncovered'):
|
|
host.blocked_base_tables(replace(child,ghost_coarse=bad),layout,order,composed_kind=kind)
|
|
bad=child.coalesce_coarse.copy();bad[0]=hole
|
|
with self.assertRaisesRegex(ValueError,'owned uncovered'):
|
|
host.blocked_base_tables(replace(child,coalesce_coarse=bad),layout,order,composed_kind=kind)
|
|
|
|
|
|
def bodies_and_jets(speed=.002):
|
|
from CelerisLab.body.geometry.circle import CircleGeometry
|
|
from CelerisLab.body.objects import SimObject
|
|
from CelerisLab.body.jets import Jet, Tube, Parallel
|
|
centers=((12.,14.),(34.,14.))
|
|
bodies=[SimObject(99,geometry=CircleGeometry(*c,.8),center=c,radius=.8) for c in centers]
|
|
jets=[Jet(Tube((c[0]-1.5,c[1]),0.,3.,.3),Parallel(0.),amplitude=speed) for c in centers]
|
|
return bodies,jets
|
|
|
|
|
|
KW=dict(base_block=(4,4),fine_block=(4,4),child_block=(4,4),
|
|
child_rectangles_in_parent=CHILDREN)
|
|
|
|
|
|
def mock_backend(*, children=CHILDREN, bodies=False, fail_compile=False, fail_upload=False):
|
|
module=host_module()
|
|
context=Mock(activate=nullcontext,sm_arch='sm_70')
|
|
allocations=[]
|
|
def allocate(size):
|
|
pointer=Mock(name=f'allocation_{len(allocations)}')
|
|
allocations.append(pointer)
|
|
return pointer
|
|
module.cuda.mem_alloc.side_effect=allocate
|
|
def load(ptx):
|
|
gpu=Mock()
|
|
gpu.get_function.side_effect=lambda name: Mock(name=name)
|
|
gpu.get_global.side_effect=lambda name: (Mock(),20 if name=='d_params' else 16)
|
|
return gpu
|
|
module.compiler.load_module.side_effect=load
|
|
if fail_compile:
|
|
module.compiler.compile_kernel.side_effect=['first','second',RuntimeError('L2 compile failed')]
|
|
if fail_upload:
|
|
def upload(*args):
|
|
if len(allocations)>20:
|
|
raise RuntimeError('table upload failed')
|
|
module.cuda.memcpy_htod.side_effect=upload
|
|
objects,jets=bodies_and_jets() if bodies else ((),())
|
|
try:
|
|
run=module.FixedPatchStepper(CFG,PARENTS,context=context,
|
|
child_objects=objects,child_jets=jets,
|
|
**dict(KW,child_rectangles_in_parent=children))
|
|
except RuntimeError:
|
|
for allocation in allocations:
|
|
allocation.free.assert_called_once_with()
|
|
raise
|
|
return run,module,allocations
|
|
|
|
|
|
class ConstructorAndSchedule(unittest.TestCase):
|
|
def test_invalid_inputs_fail_before_context_or_allocations(self):
|
|
context=Mock()
|
|
cases=[dict(child=(8,8,16,16)),dict(child_rectangles_in_parent=[]),
|
|
dict(child_rectangles_in_parent=[(16,16,32)]),
|
|
dict(child_rectangles_in_parent=[(16.5,16,32,40)]),
|
|
dict(child_rectangles_in_parent=[(16,16,32,24),(16,27,32,40)]),
|
|
dict(child_rectangles_in_parent=[(40,16,48,40)]),
|
|
dict(child_rectangles_in_parent=[(8,16,32,40)]),
|
|
dict(child_block=None),dict(fine_block=None),dict(base_block=None),
|
|
dict(child_block=(5,4)),dict(fine_block=(5,4)),
|
|
dict(child_block=(4,)),dict(child_rectangles_in_parent=(16,16,32,40))]
|
|
for changes in cases:
|
|
with self.subTest(changes=changes),self.assertRaises((ValueError,TypeError)):
|
|
host.FixedPatchStepper(CFG,PARENTS,context=context,**dict(KW,**changes))
|
|
with self.assertRaisesRegex(ValueError,'closer'):
|
|
host.FixedPatchStepper(CFG,[(4,4,20,28),(23,4,44,28)],context=context,**KW)
|
|
objects,jets=bodies_and_jets()
|
|
with self.assertRaisesRegex(ValueError,'only on L2'):
|
|
host.FixedPatchStepper(CFG,PARENTS,context=context,fine_objects=objects,**KW)
|
|
from CelerisLab.body.jets import Jet,Tube,Parallel
|
|
spanning=Jet(Tube((10.5,14.),0.,25.,.3),Parallel(0.),amplitude=0.)
|
|
with self.assertRaisesRegex(ValueError,'one jet'):
|
|
host.FixedPatchStepper(CFG,PARENTS,context=context,child_objects=objects,
|
|
child_jets=[spanning],**KW)
|
|
context.activate.assert_not_called()
|
|
host.cuda.mem_alloc.assert_not_called()
|
|
|
|
def test_aggregate_schedule_independent_of_region_count(self):
|
|
for children,bodies in ((CHILDREN,True),(CHILDREN[:1],False),
|
|
(((16,16,32,24),(16,28,32,40)),False)):
|
|
run,_,allocations=mock_backend(children=children,bodies=bodies)
|
|
self.assertEqual(len(run._levels),3)
|
|
c,f,g=run._levels
|
|
self.assertEqual(len({id(level.current) for level in run._levels} |
|
|
{id(level.spare) for level in run._levels}),6)
|
|
events=[]
|
|
initial_c=c.current
|
|
def record(name):
|
|
return lambda source,*a,**kw: events.append((name,source))
|
|
run._pairs[0].explode.side_effect=record('explode01')
|
|
run._pairs[1].explode.side_effect=record('explode12')
|
|
for i,level in enumerate(run._levels):
|
|
level.kernel.side_effect=record(f'step{i}')
|
|
if bodies:
|
|
g.boundary.kernel.side_effect=record('bfl2')
|
|
run._pairs[1].coalesce.side_effect=lambda src,dst,*a,**kw: events.append(('restrict21',dst))
|
|
run._pairs[0].coalesce.side_effect=lambda src,dst,*a,**kw: events.append(('restrict10',dst))
|
|
run._status='ready'
|
|
expected=[('explode01',initial_c)]
|
|
for middle in (f.current,f.spare):
|
|
expected.append(('explode12',middle))
|
|
for leaf in (g.current,g.spare):
|
|
expected.append(('halo2',leaf))
|
|
if bodies:
|
|
expected.append(('bfl2',leaf))
|
|
expected.append(('step2',leaf))
|
|
expected.extend([('halo1',middle),('restrict21',middle),('step1',middle)])
|
|
expected.extend([('halo0',initial_c),('restrict10',initial_c),('step0',initial_c)])
|
|
def gather(level,buffer,**kw):
|
|
events.append((f'halo{next(i for i,l in enumerate(run._levels) if l is level)}',buffer))
|
|
with patch.object(run,'_gather_halo',side_effect=gather):
|
|
run.step()
|
|
self.assertEqual(events,expected)
|
|
self.assertEqual([l.kernel.call_count for l in run._levels],[1,2,4])
|
|
# A stale-feed implementation must fail the source-identity guard.
|
|
feeds=[]
|
|
advance=run._advance
|
|
def checked_advance(index, buffer, **kw):
|
|
if index == 2: # L1 -> L2 feed: must be the L1 hop source
|
|
self.assertIs(buffer,(f.current,f.spare)[len(feeds)])
|
|
feeds.append(buffer)
|
|
advance(index,buffer,**kw)
|
|
with patch.object(run,'_advance',side_effect=lambda index,buffer,**kw:
|
|
checked_advance(index,f.current if index == 2 else buffer,**kw)):
|
|
with self.assertRaises(AssertionError):
|
|
run.step()
|
|
# A per-region shared-parent advance must fail the 1:2:4 guard.
|
|
run._status='ready'
|
|
counts=[0,0,0]
|
|
def count(index, multiple=1):
|
|
def step(*a,**kw):
|
|
counts[index]+=multiple
|
|
return step
|
|
c.kernel.side_effect=count(0)
|
|
f.kernel.side_effect=count(1,2) # mutant: twice per shared L1 hop
|
|
g.kernel.side_effect=count(2)
|
|
run.step()
|
|
with self.assertRaises(AssertionError):
|
|
self.assertEqual(counts,[1,2,4])
|
|
run.close();run.close()
|
|
for allocation in allocations:
|
|
allocation.free.assert_called_once_with()
|
|
|
|
def test_partial_construction_frees_every_allocation(self):
|
|
for options in (dict(fail_compile=True),dict(fail_upload=True)):
|
|
with self.subTest(options=options),self.assertRaisesRegex(RuntimeError,'failed'):
|
|
mock_backend(**options)
|
|
|
|
|
|
if __name__=='__main__':
|
|
unittest.main()
|