"""CPU reduction and real host scheduling with CUDA doubles, not GPU physics.""" from contextlib import nullcontext from dataclasses import replace from unittest.mock import Mock import numpy as np import unittest from CelerisLab.body.geometry.circle import CircleGeometry from CelerisLab.body.objects import SimObject, SurfaceActuation from CelerisLab.config import LBMConfig from CelerisLab.lbm.blocks import BlockLayout2D from CelerisLab.lbm.refinement import FixedPatch2D, prepare_fine_boundary from CelerisLab.lbm.refinement_tables import _prepare_blocked_boundary from CelerisLab.lbm.raw_exchange import summarize_exchange from tests.unit.test_refinement_three_level_multipatch import host_module def body(center=(16., 16.), radius=.5, speed=0.): return SimObject(91, geometry=CircleGeometry(*center, radius), center=center, radius=radius, surface_actuation=SurfaceActuation(.7, .85, speed)) def backend(observe=True, obj=None): host = host_module() calls = [] module = Mock() module.get_global.return_value = (Mock(), 20) module.get_function.side_effect = lambda name: Mock(side_effect=lambda *a, **k: calls.append((name, a, k))) host.compiler.load_module.return_value = module host.cuda.mem_alloc.side_effect = lambda n: Mock(size=n) host.cuda.memcpy_dtoh.side_effect = lambda array, pointer: array.fill(0) ctx = Mock(activate=nullcontext, sm_arch='sm_70') run = host.FixedPatchStepper(LBMConfig(nx=48, ny=48, viscosity=.02), context=ctx, refinements=[[(6, 6, 26, 26)], [(20, 20, 44, 44)], [(56, 56, 72, 72)]], level_blocks=[(4, 4)]*4, level_objects=[(), (), (), (body() if obj is None else obj,)], observe_raw_exchange=observe) arrays = [np.full(l.shape, 1/9, np.float32) for l in run._levels] run.initialize(*arrays) return run, host, calls, arrays class RawExchange(unittest.TestCase): def _check_weights_split_and_window(self, level): h = 2. ** -level rows = np.array([[2, -3, 5, 7], [-1, 4, -2, -6], [99, 99, 99, 99]], np.float32) kw = dict(body_id=0, level=level, h=h, time_end=3, reference_center=(7, 8)) last = summarize_exchange(rows, [0, 0, 1], [False, True, False], window='last_substep', **kw) integral = rows.astype(float)*[h*h, h*h, h**3, h*h]/h full = summarize_exchange(integral, [0, 0, 1], [False, True, False], window='last_base_step', **kw) np.testing.assert_array_equal(last['total']['force'], np.array([1, 1])*h) self.assertEqual(last['total']['torque'], 3 * h * h) self.assertEqual(last['total']['mass_into_fluid'], h * h) for name in ('force', 'torque', 'mass_rate'): np.testing.assert_array_equal(last['total'][name], full['total'][name]) for name in ('force', 'impulse', 'torque', 'angular_impulse', 'mass_into_fluid'): np.testing.assert_array_equal(last['total'][name], np.asarray(last['wall'][name])+last['aperture'][name]) self.assertEqual(last['time_start'], 3 - h) self.assertEqual(full['time_start'], 2) self.assertEqual(last['sample_count'], 1) self.assertEqual(full['sample_count'], 2 ** level) self.assertEqual(full['reference_center'], (7, 8)) def test_final_substep_is_not_interval_mean(self): kw = dict(body_id=0, level=1, h=.5, time_end=1, reference_center=(0, 0)) last = summarize_exchange([[4, 0, 0, 0]], [0], [False], window='last_substep', **kw) full = summarize_exchange([[1.5, 0, 0, 0]], [0], [False], window='last_base_step', **kw) self.assertEqual(last['total']['force'][0], 2) self.assertEqual(full['total']['force'][0], 1.5) def test_host_metadata_zero_jet_and_rows(self): cfg = LBMConfig(nx=32, ny=32, viscosity=.1) pair = FixedPatch2D.from_config(cfg, (4, 4, 28, 28)) boundary = prepare_fine_boundary(cfg, pair, [body(radius=2.15)]) local = replace(cfg, nx=pair.fine_shape[1], ny=pair.fine_shape[0], viscosity=.2) local.validate() layout = BlockLayout2D.build(*pair.fine_shape, 8, 8, periodic=(False, False)) b = _prepare_blocked_boundary(boundary, layout, local, ()) for key, idx in [('body_id', 3), ('rx', 4), ('ry', 5)]: np.testing.assert_array_equal(b.tables[key], boundary.curved_host[idx]) self.assertTrue(b.tables['aperture'].any()) self.assertTrue((~b.tables['aperture']).any()) self.assertFalse(b.tables['surface_velocity'].any()) # Classification does not disappear when prescribed velocity is zero. self.assertEqual(len(b.tables['cell']), len(boundary.curved_host[0])) def test_four_level_schedule_publication_and_fixed_storage(self): run, host, calls, arrays = backend() b = run._level_boundaries[3] rows = len(b.tables['cell']) self.assertEqual(b.last_raw.size, rows * 16) self.assertEqual(b.interval_integral.size, rows * 32) with self.assertRaisesRegex(RuntimeError, 'completed'): run.read_raw_exchange(0, level=3) run.step() names = [name for name, _, _ in calls] self.assertEqual(names.count('BlockAccumulateExchange'), 8) for i, (name, args, kw) in enumerate(calls): if name == 'BlockAccumulateExchange': self.assertEqual(names[i - 1], 'BlockCurvedBoundary') self.assertEqual(names[i + 1], 'BlockStep') self.assertIs(args[0], b.last_raw) self.assertIs(args[1], b.interval_integral) self.assertEqual(args[3:], (0.125, 0.125)) self.assertIs(kw['stream'], run._stream) self.assertEqual(host.cuda.memset_d8_async.call_count, 1) value = run.read_raw_exchange(0, level=3) self.assertEqual(value['sample_count'], 8) self.assertEqual(value['time_start'], 0) run.step(0) self.assertEqual(run.read_raw_exchange(0, level=3)['time_end'], 1) self.assertEqual(run.read_raw_exchange(0, level=3, window='last_substep')['time_start'], 0.875) run.step(2) self.assertEqual(run.read_raw_exchange(0, level=3)['time_start'], 2) run.initialize(*arrays) with self.assertRaisesRegex(RuntimeError, 'completed'): run.read_raw_exchange(0, level=3) run.close() def test_torque_reference_uses_geometry_center_not_optional_object_center(self): for center in (None, (3., 4.)): obj = SimObject(91, geometry=CircleGeometry(16., 16., .5), center=center, radius=.5) run, _, _, _ = backend(obj=obj) run.step() result = run.read_raw_exchange(0, level=3) self.assertEqual(result['reference_center'], (16.0, 16.0)) self.assertNotEqual(result['reference_center'], tuple(obj.center)) run.close() def test_disabled_has_nulls_and_no_observation_allocation_or_launch(self): run, host, calls, _ = backend(False) b = run._level_boundaries[3] self.assertIs(b.last_raw, b.interval_integral) self.assertIs(b.interval_integral, None) run.step() for name, args, _ in calls: if name == 'BlockCurvedBoundary': self.assertEqual(tuple(args[8:]), (0, 0, 0)) self.assertTrue(all((n != 'BlockAccumulateExchange' for n, _, _ in calls))) host.cuda.memset_d8_async.assert_not_called() with self.assertRaisesRegex(RuntimeError, 'disabled'): run.read_raw_exchange(0, level=3) run.close() def test_failed_sync_never_publishes_partial_interval(self): run, host, calls, _ = backend() run.step() run._stream.synchronize.side_effect = RuntimeError('device failure') with self.assertRaisesRegex(RuntimeError, 'device failure'): run.step() self.assertFalse(run._exchange_valid) with self.assertRaisesRegex(RuntimeError, 'failed'): run.read_raw_exchange(0, level=3) run._stream.synchronize.side_effect = None run.close() def test_dense_observation_rejected_before_device_access(self): host = host_module() ctx = Mock() with self.assertRaisesRegex(ValueError, 'blocked body'): host.FixedPatchStepper(LBMConfig(nx=32, ny=32, viscosity=.1), (4, 4, 28, 28), context=ctx, fine_objects=[body(radius=2.15)], observe_raw_exchange=True) ctx.activate.assert_not_called() def test_all_depth_weights(self): for level in (1, 2, 3, 7): self._check_weights_split_and_window(level) if __name__ == "__main__": unittest.main()