body: stage motion-record uploads in an owned page-locked buffer
memcpy_htod_async needs page-locked host memory. MovingTables allocates one pinned FP64 buffer at construction (manager.py action pattern) and rewrites it only when the records change; every upload is followed by a stream sync in Simulation._run_moving before the next one, so no copy is in flight when it is reused. The advance test fixture now binds the real LBMStepper.step. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
25cc5cfd68
commit
653fd997a6
@@ -197,6 +197,8 @@ class MovingTables:
|
|||||||
self.geom = list(geom) # host copy of the uploaded records
|
self.geom = list(geom) # host copy of the uploaded records
|
||||||
self.band_gpu = cuda.to_device(np.asarray(band, np.int32))
|
self.band_gpu = cuda.to_device(np.asarray(band, np.int32))
|
||||||
self.geom_gpu = cuda.to_device(np.asarray(geom, np.float64))
|
self.geom_gpu = cuda.to_device(np.asarray(geom, np.float64))
|
||||||
|
# Owned page-locked staging for async record uploads (allocated once).
|
||||||
|
self._geom_pinned = cuda.pagelocked_empty(len(geom), np.float64)
|
||||||
self.diag_gpu = cuda.to_device(np.zeros(self.n_bodies * self.DIAG_FLOATS, np.float32))
|
self.diag_gpu = cuda.to_device(np.zeros(self.n_bodies * self.DIAG_FLOATS, np.float32))
|
||||||
self.slot_debug_gpu = None # optional (q, fallback, rx, ry) per slot for parity tests
|
self.slot_debug_gpu = None # optional (q, fallback, rx, ry) per slot for parity tests
|
||||||
|
|
||||||
@@ -224,13 +226,18 @@ class MovingTables:
|
|||||||
return [*c_old, *c_new, motion.radius, *u]
|
return [*c_old, *c_new, motion.radius, *u]
|
||||||
|
|
||||||
def upload_records(self, circles, stream) -> bool:
|
def upload_records(self, circles, stream) -> bool:
|
||||||
"""Upload the records when they differ from the device copy; True if advancing."""
|
"""Enqueue the records when they differ from the device copy; True if advancing.
|
||||||
import numpy as np
|
|
||||||
|
The pinned staging buffer is reused: callers synchronize ``stream`` after every
|
||||||
|
upload before the next one (``Simulation._run_moving``), so no copy is in flight
|
||||||
|
when it is rewritten. ``geom`` is the enqueued state; after a failed copy the
|
||||||
|
owner is failed and the cache is never trusted again.
|
||||||
|
"""
|
||||||
import pycuda.driver as cuda
|
import pycuda.driver as cuda
|
||||||
geom = [v for motion, _ in circles for v in self.record(motion)]
|
geom = [v for motion, _ in circles for v in self.record(motion)]
|
||||||
if geom != self.geom:
|
if geom != self.geom:
|
||||||
self._staged = np.asarray(geom, np.float64) # keep alive until the copy completes
|
self._geom_pinned[:] = geom
|
||||||
cuda.memcpy_htod_async(self.geom_gpu, self._staged, stream)
|
cuda.memcpy_htod_async(self.geom_gpu, self._geom_pinned, stream)
|
||||||
self.geom = geom
|
self.geom = geom
|
||||||
g = self.GEOM_DOUBLES
|
g = self.GEOM_DOUBLES
|
||||||
return any(geom[i:i + 2] != geom[i + 2:i + 4] for i in range(0, len(geom), g))
|
return any(geom[i:i + 2] != geom[i + 2:i + 4] for i in range(0, len(geom), g))
|
||||||
|
|||||||
@@ -272,6 +272,8 @@ def advancing_simulation(monkeypatch, **tables):
|
|||||||
sim.stepper = SimpleNamespace(
|
sim.stepper = SimpleNamespace(
|
||||||
moving=Tables(writes, **tables), blocked=None,
|
moving=Tables(writes, **tables), blocked=None,
|
||||||
_step=lambda n, advance=False, **k: writes.append(("step", n, advance)))
|
_step=lambda n, advance=False, **k: writes.append(("step", n, advance)))
|
||||||
|
from CelerisLab.lbm.stepper import LBMStepper
|
||||||
|
sim.stepper.step = LBMStepper.step.__get__(sim.stepper) # real public entry
|
||||||
sim._moving_tables = sim.stepper.moving
|
sim._moving_tables = sim.stepper.moving
|
||||||
return sim, writes, sim.bodies.objects[0].motion
|
return sim, writes, sim.bodies.objects[0].motion
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user