From 8b6133cc8b29cf37f45e4b604f0285bd443e170e Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 06:33:52 +0200 Subject: [PATCH 1/5] feat(amr): execute prepared local Reflux kernels --- .../time/amr/levels/amr_patch_range.hpp | 92 ++++++++ .../time/amr/levels/amr_subcycling.hpp | 217 ++++++++++++++++-- .../pops/runtime/amr/amr_program_reflux.hpp | 5 +- .../runtime/program/amr_program_context.hpp | 2 +- 4 files changed, 300 insertions(+), 16 deletions(-) diff --git a/include/pops/numerics/time/amr/levels/amr_patch_range.hpp b/include/pops/numerics/time/amr/levels/amr_patch_range.hpp index 02a52a46b..49d4d497a 100644 --- a/include/pops/numerics/time/amr/levels/amr_patch_range.hpp +++ b/include/pops/numerics/time/amr/levels/amr_patch_range.hpp @@ -593,6 +593,29 @@ struct RefluxStripConstView { int components = 0; }; +/// Four already-computed, signed coarse-cell corrections around one fine-patch footprint. +/// +/// A native Reflux component may fill these contiguous face buffers, but it never receives the +/// sparse global register, coverage mask, periodicity or MPI communicator. PoPS alone maps the +/// values onto canonical uncovered parent cells. +struct RefluxFaceCorrectionView { + int I0 = 0, I1 = -1, J0 = 0, J1 = -1; + Real* x_low = nullptr; + Real* x_high = nullptr; + Real* y_low = nullptr; + Real* y_high = nullptr; + int components = 0; +}; + +struct RefluxFaceCorrectionConstView { + int I0 = 0, I1 = -1, J0 = 0, J1 = -1; + const Real* x_low = nullptr; + const Real* x_high = nullptr; + const Real* y_low = nullptr; + const Real* y_high = nullptr; + int components = 0; +}; + template inline RefluxStripView reflux_strip_view(Strip& strip, int components) { return {strip.I0, strip.I1, strip.J0, strip.J1, strip.cL.data(), @@ -821,6 +844,64 @@ struct RouteRefluxStripKernel { } }; +/// Deposit one external/local Reflux result into PoPS' sparse correction authority. The provider +/// has already applied side*(fine-coarse)/spacing; this kernel owns only topology canonicalisation, +/// coverage exclusion and deterministic face order. +struct RoutePreparedRefluxCorrectionKernel { + RefluxFaceCorrectionConstView faces; + FluxRegisterView correction; + CoverageMaskView coverage; + Box2D coarse_domain; + Periodicity periodicity; + + POPS_HD static int wrap_index(int value, int lo, int extent) { + const std::int64_t relative = static_cast(value) - lo; + std::int64_t quotient = relative / extent; + if (relative % extent < 0) + --quotient; + return static_cast(static_cast(lo) + relative - quotient * extent); + } + + POPS_HD bool canonicalize(int& I, int& J) const { + if (I < coarse_domain.lo[0] || I > coarse_domain.hi[0]) { + if (!periodicity.x) + return false; + I = wrap_index(I, coarse_domain.lo[0], coarse_domain.nx()); + } + if (J < coarse_domain.lo[1] || J > coarse_domain.hi[1]) { + if (!periodicity.y) + return false; + J = wrap_index(J, coarse_domain.lo[1], coarse_domain.ny()); + } + return true; + } + + POPS_HD void add_if_uncovered(int I, int J, int component, Real amount) const { + if (!canonicalize(I, J) || coverage.covered(I, J)) + return; + correction.add(I, J, component, amount); + } + + POPS_HD void operator()(int, int) const { + for (int J = faces.J0; J <= faces.J1; ++J) + for (int component = 0; component < faces.components; ++component) { + const std::size_t index = + static_cast(J - faces.J0) * static_cast(faces.components) + + static_cast(component); + add_if_uncovered(faces.I0 - 1, J, component, faces.x_low[index]); + add_if_uncovered(faces.I1 + 1, J, component, faces.x_high[index]); + } + for (int I = faces.I0; I <= faces.I1; ++I) + for (int component = 0; component < faces.components; ++component) { + const std::size_t index = + static_cast(I - faces.I0) * static_cast(faces.components) + + static_cast(component); + add_if_uncovered(I, faces.J0 - 1, component, faces.y_low[index]); + add_if_uncovered(I, faces.J1 + 1, component, faces.y_high[index]); + } + } +}; + } // namespace detail inline void sample_coarse_x_strip(const ConstArray4& left, const ConstArray4& right, @@ -1087,6 +1168,17 @@ struct CoarseFineInterface { Real(1) / dx, Real(1) / dy, Real(1)}); } + void route_prepared_reflux_correction_(const RefluxFaceCorrectionConstView& faces, + FluxRegister& ref, int nc) const { + if (nc <= 0 || ref.nc != nc || faces.components != nc || faces.I1 < faces.I0 || + faces.J1 < faces.J0 || faces.x_low == nullptr || faces.x_high == nullptr || + faces.y_low == nullptr || faces.y_high == nullptr) + throw std::invalid_argument("prepared Reflux correction view is incomplete"); + for_each_cell(Box2D{{0, 0}, {0, 0}}, + detail::RoutePreparedRefluxCorrectionKernel{faces, ref.view(), cmask.view(), + coarse_region, periodicity}); + } + template static void validate_route_inputs_(const Reg& coarse, const Reg& fine, Real dx, Real dy, Real coarse_scale, const FluxRegister& ref, int nc, diff --git a/include/pops/numerics/time/amr/levels/amr_subcycling.hpp b/include/pops/numerics/time/amr/levels/amr_subcycling.hpp index 12292338a..ac48aaece 100644 --- a/include/pops/numerics/time/amr/levels/amr_subcycling.hpp +++ b/include/pops/numerics/time/amr/levels/amr_subcycling.hpp @@ -7,6 +7,7 @@ #include #include +#include #include #include #include @@ -796,6 +797,90 @@ inline void clear_reflux_storage_on_device(RefluxStorage& values) { } // namespace detail +/// Persistent, patch-local output storage for one external Reflux invocation. It is allocated with +/// the topology plan, poisoned before every callback and consumed by PoPS only after all entries are +/// finite. Non-owning ABI views never outlive this workspace. +struct PreparedAmrRefluxFaceWorkspace { + int I0 = 0, I1 = -1, J0 = 0, J1 = -1; + int components = 0; + RefluxStorage x_low; + RefluxStorage x_high; + RefluxStorage y_low; + RefluxStorage y_high; + std::string patch_identity; + std::array interface_identities; + + static PreparedAmrRefluxFaceWorkspace prepare(const Box2D& footprint, int ncomp, + std::string transition_identity, + std::size_t global_child) { + if (footprint.empty() || ncomp <= 0 || transition_identity.empty()) + throw std::invalid_argument("prepared external Reflux workspace is incomplete"); + const auto checked_size = [ncomp](std::int64_t extent) { + const std::size_t components = static_cast(ncomp); + if (extent <= 0 || + static_cast(extent) > std::numeric_limits::max() / components) + throw std::overflow_error("prepared external Reflux face size overflow"); + return static_cast(extent) * components; + }; + PreparedAmrRefluxFaceWorkspace result; + result.I0 = footprint.lo[0]; + result.I1 = footprint.hi[0]; + result.J0 = footprint.lo[1]; + result.J1 = footprint.hi[1]; + result.components = ncomp; + result.x_low.resize(checked_size(footprint.ny())); + result.x_high.resize(result.x_low.size()); + result.y_low.resize(checked_size(footprint.nx())); + result.y_high.resize(result.y_low.size()); + result.patch_identity = transition_identity + "/patch=" + std::to_string(global_child); + result.interface_identities = { + result.patch_identity + "/x-low", result.patch_identity + "/x-high", + result.patch_identity + "/y-low", result.patch_identity + "/y-high"}; + return result; + } + + void poison() { + const Real sentinel = std::numeric_limits::quiet_NaN(); + for (auto* values : {&x_low, &x_high, &y_low, &y_high}) + std::fill(values->begin(), values->end(), sentinel); + } + + [[nodiscard]] bool all_finite() const { + for (const auto* values : {&x_low, &x_high, &y_low, &y_high}) + if (std::any_of(values->begin(), values->end(), + [](Real value) { return !std::isfinite(static_cast(value)); })) + return false; + return true; + } + + [[nodiscard]] RefluxFaceCorrectionView view() { + return {I0, I1, J0, J1, x_low.data(), x_high.data(), y_low.data(), y_high.data(), components}; + } + + [[nodiscard]] RefluxFaceCorrectionConstView view() const { + return {I0, I1, J0, J1, x_low.data(), x_high.data(), y_low.data(), y_high.data(), components}; + } +}; + +/// Complete local/noncollective invocation data. Flux strips are already integrated in time and +/// averaged onto coarse faces; the callback may only fill `correction`. +struct PreparedAmrRefluxLocalRequest { + const std::string* transition_identity = nullptr; + const std::string* patch_identity = nullptr; + std::array interface_identities{}; + int parent_level = -1; + int child_level = -1; + std::size_t global_child = 0; + RefluxStripConstView coarse; + RefluxStripConstView fine; + RefluxFaceCorrectionView correction; + amr::ClockStamp logical_time; + Real dx = Real(0); + Real dy = Real(0); +}; + +using PreparedAmrRefluxLocalKernel = std::function; + /// Prepared spatial reflux storage for one exact Program-owned parent/child transition. It owns /// only the interface topology and collective correction register; ProgramGraph supplies the /// already time-integrated coarse/fine flux strips. @@ -812,19 +897,42 @@ class PreparedAmrProgramRefluxTransition { const Box2D& parent_domain, Periodicity periodicity, const CommunicatorView& communicator) { + return prepare_with_local_kernel(parent, child, parent_domain, periodicity, 0, + "pops://runtime/amr/program-reflux/parent=0/child=1", {}, + communicator); + } + + static PreparedAmrProgramRefluxTransition prepare_with_local_kernel( + const AmrLevelMP& parent, const AmrLevelMP& child, const Box2D& parent_domain, + Periodicity periodicity, int parent_level, std::string transition_identity, + PreparedAmrRefluxLocalKernel local_kernel, const CommunicatorView& communicator) { if (parent.U.ncomp() != child.U.ncomp()) throw std::invalid_argument("prepared AMR Program reflux transition component mismatch"); + if (parent_level < 0 || transition_identity.empty()) + throw std::invalid_argument("prepared AMR Program reflux transition identity is incomplete"); validate_ratio_aligned_disjoint_fine_layout(child.U.box_array(), &parent_domain); CoarseFineInterface interface(parent_domain, child.U.box_array(), periodicity); std::vector correction_regions = interface.reflux_register_regions(child.U.box_array()); - return PreparedAmrProgramRefluxTransition(parent, child, communicator, std::move(interface), - std::move(correction_regions)); + std::vector local_workspaces( + static_cast(child.U.box_array().size())); + if (local_kernel) + for (int global_child = 0; global_child < child.U.box_array().size(); ++global_child) + if (child.U.dmap()[global_child] == communicator.rank()) + local_workspaces[static_cast(global_child)] = + PreparedAmrRefluxFaceWorkspace::prepare( + PatchRange(child.U.box_array()[global_child]).box(), parent.U.ncomp(), + transition_identity, static_cast(global_child)); + return PreparedAmrProgramRefluxTransition(parent, child, communicator, parent_level, + std::move(transition_identity), + std::move(local_kernel), std::move(local_workspaces), + std::move(interface), std::move(correction_regions)); } template void synchronize_integrated(MultiFab& parent_state, Real dx, Real dy, const CoarseStripRange& coarse_role, const FineStripRange& fine_role, - const CommunicatorView& communicator) { + const CommunicatorView& communicator, + const amr::ClockStamp* logical_time = nullptr) { validate_communicator_(communicator); using CoarseStrip = typename CoarseStripRange::value_type; using FineStrip = typename FineStripRange::value_type; @@ -836,6 +944,11 @@ class PreparedAmrProgramRefluxTransition { // enter the correction Allreduce while its peer unwinds. std::exception_ptr local_failure; try { + if (local_kernel_ && + (logical_time == nullptr || logical_time->level != parent_level_ || + logical_time->macro_step < 0 || !std::isfinite(logical_time->physical_time))) + throw std::invalid_argument( + "prepared external Reflux requires the exact parent logical time"); validate_parent_state_(parent_state); if (coarse_role.size() != child_global_size_ || fine_role.size() != child_global_size_) throw std::runtime_error( @@ -863,15 +976,69 @@ class PreparedAmrProgramRefluxTransition { } catch (...) { local_failure = std::current_exception(); } - const std::uint64_t rejected = - all_reduce_max(local_failure ? std::uint64_t(1) : std::uint64_t(0), communicator); - if (rejected != 0) { + // Presence, rank-local preflight failure and the later execution branch are decided by one + // collective bitmask. A rank can therefore never enter the builtin gather while a peer invokes + // an external callback. + constexpr char kExternalSelected = char{1}; + constexpr char kBuiltinSelected = char{2}; + constexpr char kPreflightFailed = char{4}; + char preflight_consensus = local_kernel_ ? kExternalSelected : kBuiltinSelected; + if (local_failure) + preflight_consensus |= kPreflightFailed; + all_reduce_or_inplace(&preflight_consensus, std::size_t{1}, communicator); + const bool provider_mismatch = (preflight_consensus & kExternalSelected) != 0 && + (preflight_consensus & kBuiltinSelected) != 0; + if ((preflight_consensus & kPreflightFailed) != 0 || provider_mismatch) { if (local_failure) std::rethrow_exception(local_failure); - throw std::runtime_error("AMR Program reflux preflight failed on another communicator rank"); + throw std::runtime_error(provider_mismatch + ? "prepared Reflux provider differs between communicator ranks" + : "AMR Program reflux preflight failed on another " + "communicator rank"); } + const bool use_external = (preflight_consensus & kExternalSelected) != 0; - try { + if (use_external) { + std::exception_ptr local_failure; + try { + correction_.clear_on_device(); + device_fence(); + for (std::size_t global_child = 0; global_child < child_global_size_; ++global_child) { + const CoarseStrip& coarse = coarse_role[global_child]; + const FineStrip& fine = fine_role[global_child]; + if (!coarse_role_present_(coarse)) + continue; + PreparedAmrRefluxFaceWorkspace& workspace = local_workspaces_[global_child]; + workspace.poison(); + std::array interface_identities; + for (std::size_t face = 0; face < interface_identities.size(); ++face) + interface_identities[face] = &workspace.interface_identities[face]; + local_kernel_(PreparedAmrRefluxLocalRequest{ + &transition_identity_, &workspace.patch_identity, interface_identities, parent_level_, + parent_level_ + 1, global_child, reflux_strip_const_view(coarse, ncomp_), + reflux_strip_const_view(fine, ncomp_), workspace.view(), *logical_time, dx, dy}); + if (!workspace.all_finite()) + throw std::runtime_error( + "native Reflux component left a non-finite or unwritten correction"); + const PreparedAmrRefluxFaceWorkspace& completed = workspace; + interface_.route_prepared_reflux_correction_(completed.view(), correction_, ncomp_); + } + device_fence(); + } catch (...) { + local_failure = std::current_exception(); + try { + device_fence(); + } catch (...) { + } + } + const std::uint64_t rejected = + all_reduce_max(local_failure ? std::uint64_t(1) : std::uint64_t(0), communicator); + if (rejected != 0) { + if (local_failure) + std::rethrow_exception(local_failure); + throw std::runtime_error("native Reflux component failed on another communicator rank"); + } + } else { correction_.clear_on_device(); for (std::size_t global_child = 0; global_child < child_global_size_; ++global_child) { const CoarseStrip& coarse = coarse_role[global_child]; @@ -881,6 +1048,8 @@ class PreparedAmrProgramRefluxTransition { interface_.route_reflux_integrated_pair_prevalidated_(coarse, fine, dx, dy, correction_, ncomp_); } + } + try { correction_.gather(communicator); for (int local_parent = 0; local_parent < parent_state.local_size(); ++local_parent) for_each_cell(parent_state.box(local_parent), @@ -900,7 +1069,10 @@ class PreparedAmrProgramRefluxTransition { private: PreparedAmrProgramRefluxTransition(const AmrLevelMP& parent, const AmrLevelMP& child, - const CommunicatorView& communicator, + const CommunicatorView& communicator, int parent_level, + std::string transition_identity, + PreparedAmrRefluxLocalKernel local_kernel, + std::vector local_workspaces, CoarseFineInterface interface, std::vector correction_regions) : parent_boxes_(parent.U.box_array().boxes()), @@ -913,6 +1085,10 @@ class PreparedAmrProgramRefluxTransition { communicator_size_(communicator.size()), communicator_rank_(communicator.rank()), communicator_identity_(detail::parallel_copy_communicator_identity(communicator)), + parent_level_(parent_level), + transition_identity_(std::move(transition_identity)), + local_kernel_(std::move(local_kernel)), + local_workspaces_(std::move(local_workspaces)), interface_(std::move(interface)), correction_(std::move(correction_regions), ncomp_) { if (child_footprints_.size() != child_global_size_ || child_ranks_.size() != child_global_size_) @@ -922,6 +1098,10 @@ class PreparedAmrProgramRefluxTransition { if (owner < 0 || owner >= communicator_size_) throw std::invalid_argument( "prepared AMR Program reflux child owner lies outside the communicator"); + if (parent_level_ < 0 || transition_identity_.empty() || + local_workspaces_.size() != child_global_size_) + throw std::invalid_argument( + "prepared AMR Program reflux local-provider metadata is inconsistent"); } static std::vector make_child_footprints_(const BoxArray& child_boxes) { @@ -972,6 +1152,10 @@ class PreparedAmrProgramRefluxTransition { int communicator_size_ = 1; int communicator_rank_ = 0; std::int64_t communicator_identity_ = 0; + int parent_level_ = 0; + std::string transition_identity_; + PreparedAmrRefluxLocalKernel local_kernel_; + std::vector local_workspaces_; CoarseFineInterface interface_; FluxRegister correction_; }; @@ -988,16 +1172,23 @@ class PreparedAmrProgramRefluxPlan { static PreparedAmrProgramRefluxPlan prepare( const std::vector& levels, const Box2D& base_domain, Periodicity periodicity, std::uint64_t topology_generation, - const CommunicatorView& communicator = world_communicator_view()) { + const CommunicatorView& communicator = world_communicator_view(), + PreparedAmrRefluxLocalKernel local_kernel = {}, std::string block_identity = {}) { if (levels.empty() || base_domain.empty()) throw std::invalid_argument("prepared AMR Program reflux requires a non-empty hierarchy"); + if (local_kernel && block_identity.empty()) + throw std::invalid_argument("prepared external Reflux requires one qualified block identity"); std::vector transitions; transitions.reserve(levels.size() - 1); - for (std::size_t parent = 0; parent + 1 < levels.size(); ++parent) - transitions.push_back(PreparedAmrProgramRefluxTransition::prepare( + for (std::size_t parent = 0; parent + 1 < levels.size(); ++parent) { + const std::string transition_identity = + (block_identity.empty() ? "pops://runtime/amr/program-reflux" : block_identity) + + "/parent=" + std::to_string(parent) + "/child=" + std::to_string(parent + 1); + transitions.push_back(PreparedAmrProgramRefluxTransition::prepare_with_local_kernel( levels[parent], levels[parent + 1], amr_level_index_domain(base_domain, static_cast(parent)), periodicity, - communicator)); + static_cast(parent), transition_identity, local_kernel, communicator)); + } return PreparedAmrProgramRefluxPlan(static_cast(levels.size()), topology_generation, std::move(transitions)); } diff --git a/include/pops/runtime/amr/amr_program_reflux.hpp b/include/pops/runtime/amr/amr_program_reflux.hpp index 96ab1bd65..3f20266e4 100644 --- a/include/pops/runtime/amr/amr_program_reflux.hpp +++ b/include/pops/runtime/amr/amr_program_reflux.hpp @@ -525,14 +525,15 @@ inline void sample_fine_role_strip(const MultiFab& state, const MultiFab& Fx, co /// per (cell,direction) (ADC-636 ownership: each C/F face is owned by the rank holding the covering fine /// patch), so the gather is associativity-free -> distributed == replicated bit-for-bit. inline void route_reflux_program(AmrRuntime& eng, std::size_t b, int k, const EdgeFlux& coarse_role, - const EdgeFlux& fine_role) { + const EdgeFlux& fine_role, const amr::ClockStamp& logical_time) { MultiFab& Uc = eng.level_state(b, k - 1); // the PARENT (coarse) live state we correct const BoxArray child_ba = eng.level_state(b, k).box_array(); // GLOBAL level-k patches if (child_ba.size() == 0) return; const Geometry gc = eng.level_geom(k - 1); eng.prepared_reflux_transition(b, k).synchronize_integrated( - Uc, gc.dx(), gc.dy(), coarse_role.coarse, fine_role.fine, world_communicator_view()); + Uc, gc.dx(), gc.dy(), coarse_role.coarse, fine_role.fine, world_communicator_view(), + &logical_time); } } // namespace detail diff --git a/include/pops/runtime/program/amr_program_context.hpp b/include/pops/runtime/program/amr_program_context.hpp index fa53c479a..beaf3c441 100644 --- a/include/pops/runtime/program/amr_program_context.hpp +++ b/include/pops/runtime/program/amr_program_context.hpp @@ -1124,7 +1124,7 @@ class AmrProgramContext : public ProgramExecutionServices { throw std::runtime_error( "AMR conservative ledger contains only one side of a parent/child flux pair"); if (!coarse_role.empty()) - pops::detail::route_reflux_program(*eng_, sb, child, coarse_role, fine_role); + pops::detail::route_reflux_program(*eng_, sb, child, coarse_role, fine_role, sync_clock); } sync_report_.push_back({parent, child, b, SyncPhase::AverageDown, sync_clock}); eng_->average_down_level(sb, child); From cddd9dac58b3a7d8bfd544842735c30130fa2c2a Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 06:34:05 +0200 Subject: [PATCH 2/5] feat(components): install Reflux into AMR transitions --- include/pops/runtime/amr/amr_runtime.hpp | 72 +++++- .../amr/prepared_component_providers.hpp | 207 ++++++++++++++++++ include/pops/runtime/amr_system.hpp | 2 + python/bindings/core/init/init_amr.cpp | 21 ++ src/runtime/amr/amr_system.cpp | 16 ++ 5 files changed, 317 insertions(+), 1 deletion(-) diff --git a/include/pops/runtime/amr/amr_runtime.hpp b/include/pops/runtime/amr/amr_runtime.hpp index 0326b3e63..bfc751ae6 100644 --- a/include/pops/runtime/amr/amr_runtime.hpp +++ b/include/pops/runtime/amr/amr_runtime.hpp @@ -37,6 +37,7 @@ #include #include #include // n_ranks() / comm_active(): MPI message+reduction counts (Spec 5 criterion 43) +#include #include #include #include @@ -1526,6 +1527,65 @@ class AmrRuntime { external_clustering_ = std::move(provider); } + void install_external_reflux(std::shared_ptr provider) { + const CommunicatorView communicator = world_communicator_view(); + std::exception_ptr local_failure; + try { + if (external_reflux_configured_ || bootstrap_pending_) + throw std::runtime_error( + "AmrRuntime external Reflux must be configured exactly once before bootstrap"); + } catch (...) { + local_failure = std::current_exception(); + } + if (all_reduce_max(local_failure ? std::uint64_t{1} : std::uint64_t{0}, communicator) != 0) { + if (communicator.size() == 1 && local_failure) + std::rethrow_exception(local_failure); + throw std::runtime_error( + "AmrRuntime external Reflux configuration failed on another communicator rank"); + } + + struct OptionalRefluxSelection { + const runtime::amr::PreparedRefluxComponent* provider = nullptr; + explicit operator bool() const noexcept { return provider != nullptr; } + [[nodiscard]] std::string_view collective_contract() const noexcept { + return provider == nullptr ? std::string_view{} : provider->collective_contract(); + } + }; + require_prepared_provider_collective_consensus(OptionalRefluxSelection{provider.get()}); + external_reflux_configured_ = true; + if (!provider) + return; + + external_reflux_ = std::move(provider); + local_failure = nullptr; + try { + rematerialize_persistent_topology_resources_(topology_materialization_generation_); + } catch (...) { + local_failure = std::current_exception(); + } + if (all_reduce_max(local_failure ? std::uint64_t{1} : std::uint64_t{0}, communicator) == 0) + return; + + external_reflux_.reset(); + std::exception_ptr rollback_failure; + try { + rematerialize_persistent_topology_resources_(topology_materialization_generation_); + } catch (...) { + rollback_failure = std::current_exception(); + } + external_reflux_configured_ = false; + if (all_reduce_max(rollback_failure ? std::uint64_t{1} : std::uint64_t{0}, communicator) != 0) { + if (communicator.size() == 1 && rollback_failure) + std::rethrow_exception(rollback_failure); + throw std::runtime_error( + "AmrRuntime external Reflux rollback failed on another communicator rank"); + } + if (communicator.size() == 1 && local_failure) + std::rethrow_exception(local_failure); + throw std::runtime_error( + "AmrRuntime external Reflux preparation failed on another communicator rank"); + } + /// Inject the current Program evaluation coordinate used by external Tagger/boundary component /// calls. This is not an accepted clock: it is never read for cadence/restart, is absent from /// StepSnapshot, and is overwritten by AmrProgramContext at the exact tagger/regrid boundary. @@ -5757,6 +5817,8 @@ class AmrRuntime { cluster_{}; ///< ADC-616: Berger-Rigoutsos params; default {0.7,1,32} (bit-identical). std::shared_ptr external_tagger_; std::shared_ptr external_clustering_; + std::shared_ptr external_reflux_; + bool external_reflux_configured_ = false; std::shared_ptr clustering_provider_ = std::make_shared(ClusterParams{}); // Ephemeral Program evaluation metadata required by the prepared component ABI. These values @@ -5924,6 +5986,13 @@ class AmrRuntime { coarse_fine_spatial_candidate.reserve(blocks_.size()); average_down_candidate.reserve(blocks_.size()); program_reflux_candidate.reserve(blocks_.size()); + PreparedAmrRefluxLocalKernel external_reflux_kernel; + if (external_reflux_) { + const std::shared_ptr provider = external_reflux_; + external_reflux_kernel = [provider](const PreparedAmrRefluxLocalRequest& request) { + provider->apply(request); + }; + } for (std::size_t block_index = 0; block_index < blocks_.size(); ++block_index) { const AmrRuntimeBlock& block = blocks_[block_index]; const auto& authority = block_transfer_authorities_[block_index]; @@ -5953,7 +6022,8 @@ class AmrRuntime { average_down_candidate.push_back( PreparedAmrAverageDownPlan::prepare(*block.levels, generation)); program_reflux_candidate.push_back(PreparedAmrProgramRefluxPlan::prepare( - *block.levels, dom_, base_per_, generation, world_communicator_view())); + *block.levels, dom_, base_per_, generation, world_communicator_view(), + external_reflux_kernel, block.state_identity)); } auto tagging_candidate = make_tagging_execution_plan_(tagging_program_, generation); temporal_parent_workspaces_.swap(temporal_candidate); diff --git a/include/pops/runtime/amr/prepared_component_providers.hpp b/include/pops/runtime/amr/prepared_component_providers.hpp index 6a13b1580..9b9aa7cc8 100644 --- a/include/pops/runtime/amr/prepared_component_providers.hpp +++ b/include/pops/runtime/amr/prepared_component_providers.hpp @@ -3,7 +3,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -20,6 +22,7 @@ #include #include #include +#include #include #include #include @@ -186,6 +189,16 @@ struct PreparedClusteringSpec { std::shared_ptr execution; }; +struct PreparedRefluxSpec { + std::string provider_identity; + std::string component_id; + std::string manifest_identity; + std::string layout_identity; + std::string clock_identity; + std::uint32_t interface_version = 1; + std::shared_ptr execution; +}; + /// Prepared external Tagger. One invocation per local patch sees every graph input as a qualified /// borrowed SoA view and evaluates the exact resolved graph program. Only four Boolean candidate /// bitmaps are reduced across ranks; state arrays are never packed or globally reduced. @@ -600,6 +613,200 @@ class PreparedTaggerComponent final { void* state_ = nullptr; }; +/// Prepared adapter for the deliberately narrow Reflux ABI. One callback receives four contiguous +/// faces of one rank-local child patch. It has no communicator, topology mask, global register or +/// live state; the enclosing PreparedAmrProgramRefluxTransition validates and publishes its result. +class PreparedRefluxComponent final { + public: + PreparedRefluxComponent(PreparedRefluxSpec spec, + std::shared_ptr component) + : spec_(std::move(spec)), component_(std::move(component)) { + validate_(); + prepare_provider_contract_(); + local_execution_ = std::make_shared( + spec_.execution->without_collective_authority()); + state_owner_ = component_->prepare_fresh_state( + POPS_NATIVE_INTERFACE_REFLUX_V1, spec_.interface_version, local_execution_->view()); + state_ = state_owner_.get(); + } + + [[nodiscard]] const std::string& provider_identity() const noexcept { + return spec_.provider_identity; + } + [[nodiscard]] std::string_view collective_contract() const noexcept { + return collective_contract_; + } + + void apply(const PreparedAmrRefluxLocalRequest& request) const { + static_assert(sizeof(Real) == sizeof(double), + "Reflux ABI v1 requires the binary64 PoPS backend"); + if (request.transition_identity == nullptr || request.transition_identity->empty() || + request.patch_identity == nullptr || request.patch_identity->empty() || + request.parent_level < 0 || request.child_level != request.parent_level + 1 || + request.logical_time.level != request.parent_level || request.logical_time.macro_step < 0 || + request.coarse.components <= 0 || request.coarse.components != request.fine.components || + request.coarse.components != request.correction.components || + request.coarse.I0 != request.fine.I0 || request.coarse.I1 != request.fine.I1 || + request.coarse.J0 != request.fine.J0 || request.coarse.J1 != request.fine.J1 || + request.coarse.I0 != request.correction.I0 || request.coarse.I1 != request.correction.I1 || + request.coarse.J0 != request.correction.J0 || request.coarse.J1 != request.correction.J1 || + !std::isfinite(request.dx) || !std::isfinite(request.dy) || request.dx <= Real(0) || + request.dy <= Real(0)) + throw std::invalid_argument("prepared native Reflux invocation is incomplete"); + for (const std::string* identity : request.interface_identities) + if (identity == nullptr || identity->empty()) + throw std::invalid_argument("prepared native Reflux face identity is empty"); + + const auto make_const_view = [&](const Real* data, int axis) { + if (data == nullptr) + throw std::invalid_argument("prepared native Reflux input face is absent"); + const std::size_t tangent = + static_cast(axis == 0 ? request.coarse.J1 - request.coarse.J0 + 1 + : request.coarse.I1 - request.coarse.I0 + 1); + const std::size_t components = static_cast(request.coarse.components); + const std::size_t extent0 = axis == 0 ? 1u : tangent; + const std::size_t extent1 = axis == 0 ? tangent : 1u; + return PopsConstFieldViewV1{sizeof(PopsConstFieldViewV1), + data, + 2, + {extent0, extent1, 1}, + {static_cast(extent1 * components), + static_cast(components), 0}, + components, + 1, + POPS_FIELD_CENTERING_FACE_V1, + 1u << static_cast(axis), + {0, 0, 0}, + {0, 0, 0}, + POPS_SCALAR_FLOAT64_V1, + POPS_MEMORY_SPACE_HOST_V1, + spec_.layout_identity.c_str(), + request.patch_identity->c_str(), + POPS_FIELD_OWNERSHIP_RUNTIME_BORROWED_V1}; + }; + const auto make_output_view = [&](Real* data, int axis) { + if (data == nullptr) + throw std::invalid_argument("prepared native Reflux output face is absent"); + const std::size_t tangent = + static_cast(axis == 0 ? request.coarse.J1 - request.coarse.J0 + 1 + : request.coarse.I1 - request.coarse.I0 + 1); + const std::size_t components = static_cast(request.coarse.components); + const std::size_t extent0 = axis == 0 ? 1u : tangent; + const std::size_t extent1 = axis == 0 ? tangent : 1u; + return PopsFieldViewV1{sizeof(PopsFieldViewV1), + data, + 2, + {extent0, extent1, 1}, + {static_cast(extent1 * components), + static_cast(components), 0}, + components, + 1, + POPS_FIELD_CENTERING_CELL_V1, + 0, + {0, 0, 0}, + {0, 0, 0}, + POPS_SCALAR_FLOAT64_V1, + POPS_MEMORY_SPACE_HOST_V1, + spec_.layout_identity.c_str(), + request.patch_identity->c_str(), + POPS_FIELD_OWNERSHIP_RUNTIME_BORROWED_V1}; + }; + + const std::array coarse{request.coarse.cL, request.coarse.cR, request.coarse.cB, + request.coarse.cT}; + const std::array fine{request.fine.fL, request.fine.fR, request.fine.fB, + request.fine.fT}; + const std::array correction{request.correction.x_low, request.correction.x_high, + request.correction.y_low, request.correction.y_high}; + const std::array axes{0, 0, 1, 1}; + const std::array sides{ + POPS_REFLUX_FACE_LOW_V1, POPS_REFLUX_FACE_HIGH_V1, POPS_REFLUX_FACE_LOW_V1, + POPS_REFLUX_FACE_HIGH_V1}; + std::array faces; + for (std::size_t face = 0; face < faces.size(); ++face) + faces[face] = PopsRefluxFaceV1{ + sizeof(PopsRefluxFaceV1), + request.interface_identities[face]->c_str(), + axes[face], + sides[face], + static_cast(Real(1) / (axes[face] == 0 ? request.dx : request.dy)), + make_const_view(coarse[face], axes[face]), + make_const_view(fine[face], axes[face]), + make_output_view(correction[face], axes[face])}; + + const PopsLogicalTimeV1 logical_time{sizeof(PopsLogicalTimeV1), + spec_.clock_identity.c_str(), + request.logical_time.macro_step, + request.parent_level, + 0, + 0, + request.logical_time.phase.numerator, + request.logical_time.phase.denominator, + 0.0, + request.logical_time.physical_time}; + const PopsRefluxRequestV1 abi_request{sizeof(PopsRefluxRequestV1), + request.transition_identity->c_str(), + request.parent_level, + request.child_level, + faces.size(), + faces.data(), + logical_time, + local_execution_->view()}; + PopsComponentStatusV1 status = component::unwritten_component_status(); + const auto& api = component_->table(POPS_NATIVE_INTERFACE_REFLUX_V1, + spec_.interface_version); + const int code = component::apply_reflux_interface_batch(api, state_, abi_request, status); + if (code != 0) + throw std::runtime_error(status.reason == nullptr ? "native Reflux component failed" + : status.reason); + } + + private: + void prepare_provider_contract_() { + ExactContractBuilder contract; + contract.text("pops.runtime.external-amr-reflux-provider") + .scalar(std::uint32_t{1}) + .text(spec_.provider_identity) + .text(spec_.component_id) + .text(spec_.manifest_identity) + .text(spec_.layout_identity) + .text(spec_.clock_identity) + .scalar(spec_.interface_version) + .text(spec_.execution->identity()); + collective_contract_ = std::move(contract).release(); + } + + void validate_() const { + if (!component_ || !spec_.execution || spec_.provider_identity.empty() || + spec_.component_id.empty() || spec_.manifest_identity.empty() || + spec_.layout_identity.empty() || spec_.clock_identity.empty() || + spec_.interface_version != 1) + throw std::invalid_argument("prepared AMR Reflux specification is incomplete"); + if constexpr (!std::is_same_v) + throw std::invalid_argument( + "prepared external Reflux v1 is qualified only for a host execution backend"); + component::validate_execution_context(spec_.execution->view()); + if (spec_.execution->view().memory_space != POPS_MEMORY_SPACE_HOST_V1) + throw std::invalid_argument( + "prepared external Reflux v1 requires host-resident face storage"); + const auto& api = component_->api(); + if (api.component_id == nullptr || api.manifest_identity == nullptr || + spec_.component_id != api.component_id || spec_.manifest_identity != api.manifest_identity) + throw std::invalid_argument("prepared AMR Reflux changed native component identity"); + component::require_operation( + component_->table(POPS_NATIVE_INTERFACE_REFLUX_V1, spec_.interface_version) + .apply_interface_batch != nullptr, + "apply_interface_batch"); + } + + PreparedRefluxSpec spec_; + std::shared_ptr component_; + std::shared_ptr local_execution_; + component::LoadedComponent::PreparedState state_owner_; + void* state_ = nullptr; + std::string collective_contract_; +}; + /// External Clustering ABI contract: each result is `2 * dimension` signed integers laid out as /// `[lo_0, ..., lo_(d-1), hi_0, ..., hi_(d-1)]`, inclusive and relative to the supplied region. class PreparedClusteringComponent final : public pops::amr::ClusteringProvider { diff --git a/include/pops/runtime/amr_system.hpp b/include/pops/runtime/amr_system.hpp index 7600c0c01..af778c882 100644 --- a/include/pops/runtime/amr_system.hpp +++ b/include/pops/runtime/amr_system.hpp @@ -384,6 +384,8 @@ class AmrSystem { POPS_EXPORT void install_amr_clustering_component( runtime::amr::PreparedClusteringSpec spec, std::shared_ptr component); + POPS_EXPORT void install_amr_reflux_component( + runtime::amr::PreparedRefluxSpec spec, std::shared_ptr component); POPS_EXPORT void discard_amr_provider_components(); /// Materialize one exact shared NumericalFlux route on a frozen AMR level. This seam is called /// only after the lazy AmrRuntime has been built and before bind freezes composition. diff --git a/python/bindings/core/init/init_amr.cpp b/python/bindings/core/init/init_amr.cpp index c9a3b55be..ff7b5e8e7 100644 --- a/python/bindings/core/init/init_amr.cpp +++ b/python/bindings/core/init/init_amr.cpp @@ -182,6 +182,19 @@ pops::runtime::amr::PreparedClusteringSpec amr_clustering_spec_from_python( return spec; } +pops::runtime::amr::PreparedRefluxSpec amr_reflux_spec_from_python(const py::dict& row, + const py::dict& execution) { + pops::runtime::amr::PreparedRefluxSpec spec; + spec.provider_identity = py::cast(row["provider_identity"]); + spec.component_id = py::cast(row["component_id"]); + spec.manifest_identity = py::cast(row["component_manifest_identity"]); + spec.layout_identity = py::cast(row["layout_identity"]); + spec.clock_identity = py::cast(row["clock_identity"]); + spec.interface_version = py::cast(row["interface_version"]); + spec.execution = pops::python::detail::make_component_execution_context(execution); + return spec; +} + // Assembly seams: per-block composition, native block, and refinement tagging. void bind_amr_assembly(py::class_& cls) { cls.def(py::init()) @@ -272,6 +285,14 @@ void bind_amr_assembly(py::class_& cls) { amr_clustering_spec_from_python(binding, execution), std::move(component)); }, py::arg("component"), py::arg("binding"), py::arg("execution_context")) + .def( + "_install_amr_reflux_component", + [](AmrSystem& system, std::shared_ptr component, + const py::dict& binding, const py::dict& execution) { + system.install_amr_reflux_component(amr_reflux_spec_from_python(binding, execution), + std::move(component)); + }, + py::arg("component"), py::arg("binding"), py::arg("execution_context")) .def("_discard_amr_provider_components", &AmrSystem::discard_amr_provider_components, "Roll back one failed external AMR provider transaction.") .def( diff --git a/src/runtime/amr/amr_system.cpp b/src/runtime/amr/amr_system.cpp index ae23d0cd3..1487f7ec7 100644 --- a/src/runtime/amr/amr_system.cpp +++ b/src/runtime/amr/amr_system.cpp @@ -246,6 +246,7 @@ struct AmrSystem::Impl { std::map field_storage_routes_; std::shared_ptr amr_tagger_component_; std::shared_ptr amr_clustering_component_; + std::shared_ptr amr_reflux_component_; struct BootstrapArray { std::string centering; int ncomp = 0; @@ -890,6 +891,9 @@ struct AmrSystem::Impl { runtime->install_external_tagger(amr_tagger_component_); if (amr_clustering_component_) runtime->install_external_clustering(amr_clustering_component_); + // Reflux selection is a collective optional-provider contract: every rank enters this call, + // including ranks where no external provider was selected. + runtime->install_external_reflux(amr_reflux_component_); if (!boundary_plans_.empty()) runtime->install_boundary_storage_routes(field_storage_routes_); // Low-level facade compatibility has no authored AMRTransfer object. Resolve its exact @@ -1490,6 +1494,17 @@ POPS_EXPORT void AmrSystem::install_amr_clustering_component( std::move(spec), std::move(component)); } +POPS_EXPORT void AmrSystem::install_amr_reflux_component( + runtime::amr::PreparedRefluxSpec spec, std::shared_ptr component) { + Impl* P = p_.get(); + require_assembling_amr(P->bound_, "install_amr_reflux_component"); + if (P->built || P->amr_reflux_component_) + throw std::runtime_error( + "AmrSystem external Reflux requires one installation before runtime build"); + P->amr_reflux_component_ = std::make_shared( + std::move(spec), std::move(component)); +} + POPS_EXPORT void AmrSystem::discard_amr_provider_components() { Impl* P = p_.get(); require_assembling_amr(P->bound_, "discard_amr_provider_components"); @@ -1497,6 +1512,7 @@ POPS_EXPORT void AmrSystem::discard_amr_provider_components() { throw std::runtime_error("AmrSystem cannot discard AMR providers after runtime build"); P->amr_tagger_component_.reset(); P->amr_clustering_component_.reset(); + P->amr_reflux_component_.reset(); } POPS_EXPORT void AmrSystem::install_interface_flux_component( From 1f2add44df29311d64edb317f31198a434ecaba9 Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 06:34:16 +0200 Subject: [PATCH 3/5] test(architecture): fence prepared Reflux authority --- ...TION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md | 20 ++- ...prepared_reflux_runtime_execution_fence.py | 132 ++++++++++++++++++ 2 files changed, 145 insertions(+), 7 deletions(-) create mode 100644 tests/python/architecture/test_prepared_reflux_runtime_execution_fence.py diff --git a/docs/design/SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md b/docs/design/SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md index d279bfee2..935aac84b 100644 --- a/docs/design/SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md +++ b/docs/design/SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md @@ -1418,13 +1418,19 @@ taille/header de table et opérations requises avant de conserver le handle de b sont résolues une fois à l'installation ; aucun `dlsym`, nom de classe ou dispatch Python n'entre dans une boucle de cellules. -Le contrat `Reflux` v1 est volontairement livré avant son branchement dans -`PreparedAmrProgramRefluxTransition` : catalogue, manifest, loader et consumer typé peuvent qualifier -un conformer, mais le runtime AMR continue d'utiliser son kernel interne tant qu'un adaptateur préparé -ne peut pas fournir les vues locales sans dupliquer le ledger ni transférer l'autorité collective. Une -configuration AMR ne prétend donc pas encore avoir sélectionné un provider `Reflux` externe. Cette -première qualification est limitée à la cible 2D, `float64`, CPU déjà admise par le loader de -composants ; elle ne constitue pas une promesse GPU. +Le contrat `Reflux` v1 possède maintenant un adaptateur préparé interne vers +`PreparedAmrProgramRefluxTransition`. Pour chaque patch enfant local, l'adaptateur reçoit quatre +paires de flux déjà intégrés et écrit quatre corrections dans des buffers persistants empoisonnés +avant l'appel. PoPS vérifie que chaque valeur a été écrite et reste finie, atteint un consensus +d'échec entre rangs, puis applique seul périodicité, masque de couverture, réduction MPI et +publication transactionnelle. La présence et le contrat exact du provider sont également comparés +entre rangs avant toute exécution. + +Cette tranche ne publie pas encore la sélection `Reflux` dans la résolution normalisée des providers +AMR : le seam d'installation demeure interne et les configurations publiques continuent donc +d'utiliser le kernel builtin. La qualification initiale de l'adaptateur reste limitée à la cible 2D, +`float64`, CPU avec stockage hôte. Le chemin n'est pas encore prouvé par compilation native, exécution +MPI avec un composant externe, mesure de conservation ni backend GPU. Les champs sémantiques inconnus, capacités sans preuve, collisions d'identité et entry points manquants sont refusés. Un vieux manifest n'est pas « réparé » silencieusement. diff --git a/tests/python/architecture/test_prepared_reflux_runtime_execution_fence.py b/tests/python/architecture/test_prepared_reflux_runtime_execution_fence.py new file mode 100644 index 000000000..9616dbb72 --- /dev/null +++ b/tests/python/architecture/test_prepared_reflux_runtime_execution_fence.py @@ -0,0 +1,132 @@ +"""ADC-681: a prepared Reflux component executes without owning AMR authority.""" + +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[3] +PATCH_RANGE = ( + ROOT / "include" / "pops" / "numerics" / "time" / "amr" / "levels" + / "amr_patch_range.hpp" +) +SUBCYCLING = ( + ROOT / "include" / "pops" / "numerics" / "time" / "amr" / "levels" + / "amr_subcycling.hpp" +) +PROVIDERS = ( + ROOT / "include" / "pops" / "runtime" / "amr" + / "prepared_component_providers.hpp" +) +AMR_RUNTIME = ROOT / "include" / "pops" / "runtime" / "amr" / "amr_runtime.hpp" +PROGRAM_REFLUX = ( + ROOT / "include" / "pops" / "runtime" / "amr" / "amr_program_reflux.hpp" +) +PROGRAM_CONTEXT = ( + ROOT / "include" / "pops" / "runtime" / "program" / "amr_program_context.hpp" +) +AMR_SYSTEM = ROOT / "src" / "runtime" / "amr" / "amr_system.cpp" +AMR_BINDING = ROOT / "python" / "bindings" / "core" / "init" / "init_amr.cpp" +RUNTIME_AUTHORITIES = ROOT / "python" / "pops" / "runtime" / "_runtime_authorities.py" + + +def _between(text: str, begin: str, end: str) -> str: + return text.split(begin, 1)[1].split(end, 1)[0] + + +def test_transition_executes_local_kernel_before_pops_collective_publication() -> None: + source = SUBCYCLING.read_text() + transition = _between( + source, + "class PreparedAmrProgramRefluxTransition", + "class PreparedAmrProgramRefluxPlan", + ) + assert "PreparedAmrRefluxLocalKernel local_kernel_" in transition + assert "workspace.poison();" in transition + assert "local_kernel_(PreparedAmrRefluxLocalRequest{" in transition + assert "workspace.all_finite()" in transition + assert "all_reduce_or_inplace(&preflight_consensus" in transition + assert "prepared Reflux provider differs between communicator ranks" in transition + assert transition.index("local_kernel_(PreparedAmrRefluxLocalRequest{") < ( + transition.index("route_prepared_reflux_correction_") + ) + assert transition.index("all_reduce_max(local_failure") < transition.index( + "correction_.gather(communicator);" + ) + assert "route_reflux_integrated_pair_prevalidated_" in transition + assert "apply_reflux_interface_batch" not in transition + + +def test_component_adapter_is_host_local_noncollective_and_has_no_topology() -> None: + source = PROVIDERS.read_text() + adapter = _between( + source, + "class PreparedRefluxComponent final", + "/// External Clustering ABI contract", + ) + assert "without_collective_authority()" in adapter + assert "collective_contract() const noexcept" in adapter + assert "POPS_MEMORY_SPACE_HOST_V1" in adapter + assert "apply_reflux_interface_batch" in adapter + assert "POPS_NATIVE_INTERFACE_REFLUX_V1" in adapter + assert "FluxRegister" not in adapter + assert "CoverageMask" not in adapter + assert "all_reduce" not in adapter + + +def test_pops_maps_validated_faces_through_coverage_and_periodicity() -> None: + source = PATCH_RANGE.read_text() + kernel = _between( + source, + "struct RoutePreparedRefluxCorrectionKernel", + "} // namespace detail", + ) + assert "canonicalize" in kernel + assert "coverage.covered" in kernel + assert "correction.add" in kernel + assert "faces.x_low[index]" in kernel + assert "faces.x_high[index]" in kernel + assert "faces.y_low[index]" in kernel + assert "faces.y_high[index]" in kernel + + +def test_runtime_installation_reprepares_transitions_and_routes_logical_time() -> None: + runtime = AMR_RUNTIME.read_text() + install = _between( + runtime, + "void install_external_reflux(", + "/// Inject the current Program evaluation coordinate", + ) + assert "external_reflux_ = std::move(provider);" in install + assert "require_prepared_provider_collective_consensus" in install + assert "rematerialize_persistent_topology_resources_" in install + rematerialize = _between( + runtime, + "void rematerialize_persistent_topology_resources_(", + "void record_topology_replacement_()", + ) + assert "provider->apply(request);" in rematerialize + assert "external_reflux_kernel, block.state_identity" in rematerialize + + route = PROGRAM_REFLUX.read_text() + assert "const amr::ClockStamp& logical_time" in route + assert "&logical_time" in route + context = PROGRAM_CONTEXT.read_text() + assert ( + "route_reflux_program(*eng_, sb, child, coarse_role, fine_role, sync_clock)" + in context + ) + + +def test_internal_install_seam_exists_without_claiming_public_resolution() -> None: + system = AMR_SYSTEM.read_text() + binding = AMR_BINDING.read_text() + authorities = RUNTIME_AUTHORITIES.read_text() + assert "install_amr_reflux_component(" in system + assert "runtime->install_external_reflux(amr_reflux_component_);" in system + assert "if (amr_reflux_component_)" not in _between( + system, + "runtime->install_external_tagger(amr_tagger_component_);", + "if (!boundary_plans_.empty())", + ) + assert '"_install_amr_reflux_component"' in binding + assert '"_install_amr_reflux_component"' not in authorities + assert 'tuple(providers) != ("clustering", "tagger")' in authorities From df0449582dd874b68d5b06f87185d2212145343b Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 06:51:43 +0200 Subject: [PATCH 4/5] fix(amr): include logical clock in reflux contract --- include/pops/numerics/time/amr/levels/amr_subcycling.hpp | 1 + 1 file changed, 1 insertion(+) diff --git a/include/pops/numerics/time/amr/levels/amr_subcycling.hpp b/include/pops/numerics/time/amr/levels/amr_subcycling.hpp index ac48aaece..fc16d48b2 100644 --- a/include/pops/numerics/time/amr/levels/amr_subcycling.hpp +++ b/include/pops/numerics/time/amr/levels/amr_subcycling.hpp @@ -3,6 +3,7 @@ #include #include // coarsen, parallel_copy #include +#include #include #include From e5ef1033e8568005fb6c3da4e9599999f4ea7ad0 Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 07:02:26 +0200 Subject: [PATCH 5/5] test(amr): execute prepared Reflux kernel --- .../amr/test_program_reflux_ledger.cpp | 104 ++++++++++++++++++ 1 file changed, 104 insertions(+) diff --git a/tests/cpp/integration/amr/test_program_reflux_ledger.cpp b/tests/cpp/integration/amr/test_program_reflux_ledger.cpp index b09dc2b0a..4791ff1a6 100644 --- a/tests/cpp/integration/amr/test_program_reflux_ledger.cpp +++ b/tests/cpp/integration/amr/test_program_reflux_ledger.cpp @@ -115,6 +115,110 @@ TEST(test_program_reflux_ledger, prepared_transition_does_not_skip_right_or_top_ << "every rejected preflight leaves the parent untouched"; } +TEST(test_program_reflux_ledger, prepared_local_kernel_routes_exact_face_corrections) { + const Box2D coarse_domain{{0, 0}, {7, 7}}; + const BoxArray coarse_boxes(std::vector{coarse_domain}); + const BoxArray fine_boxes(std::vector{Box2D{{4, 4}, {11, 11}}}); + const DistributionMapping coarse_mapping(coarse_boxes.size(), n_ranks()); + const DistributionMapping fine_mapping(fine_boxes.size(), n_ranks()); + AmrLevelMP coarse{MultiFab(coarse_boxes, coarse_mapping, 1, 0), nullptr, Real(0.5), Real(0.5)}; + AmrLevelMP fine{MultiFab(fine_boxes, fine_mapping, 1, 0), nullptr, Real(0.25), Real(0.25)}; + coarse.U.set_val(Real(0)); + fine.U.set_val(Real(0)); + + int calls = 0; + const auto kernel = [&calls](const PreparedAmrRefluxLocalRequest& request) { + ++calls; + EXPECT_EQ(*request.transition_identity, "pops://test/reflux"); + EXPECT_EQ(*request.patch_identity, "pops://test/reflux/patch=0"); + EXPECT_EQ(request.parent_level, 0); + EXPECT_EQ(request.child_level, 1); + EXPECT_EQ(request.global_child, 0u); + EXPECT_EQ(request.logical_time, (amr::ClockStamp{0, 3, amr::Rational(0, 1), 0.3})); + const auto x_size = + static_cast(request.correction.J1 - request.correction.J0 + 1) * + static_cast(request.correction.components); + const auto y_size = + static_cast(request.correction.I1 - request.correction.I0 + 1) * + static_cast(request.correction.components); + std::fill_n(request.correction.x_low, x_size, Real(3)); + std::fill_n(request.correction.x_high, x_size, Real(4)); + std::fill_n(request.correction.y_low, y_size, Real(5)); + std::fill_n(request.correction.y_high, y_size, Real(6)); + }; + auto transition = PreparedAmrProgramRefluxTransition::prepare_with_local_kernel( + coarse, fine, coarse_domain, Periodicity{false, false}, 0, "pops://test/reflux", kernel, + world_communicator_view()); + + EdgeStrip coarse_role = make_strip(2, 5, 2, 5, 1); + EdgeStrip fine_role = make_strip(2, 5, 2, 5, 1); + coarse_role.cL.assign(4, Real(1)); + coarse_role.cR.assign(4, Real(1)); + coarse_role.cB.assign(4, Real(1)); + coarse_role.cT.assign(4, Real(1)); + fine_role.fL.assign(4, Real(2)); + fine_role.fR.assign(4, Real(2)); + fine_role.fB.assign(4, Real(2)); + fine_role.fT.assign(4, Real(2)); + const amr::ClockStamp logical_time{0, 3, amr::Rational(0, 1), 0.3}; + transition.synchronize_integrated( + coarse.U, coarse.dx, coarse.dy, std::vector{coarse_role}, + std::vector{fine_role}, world_communicator_view(), &logical_time); + + EXPECT_EQ(calls, 1); + ASSERT_EQ(coarse.U.local_size(), 1); + EXPECT_EQ(coarse.U.fab(0)(1, 3, 0), Real(3)); + EXPECT_EQ(coarse.U.fab(0)(6, 3, 0), Real(4)); + EXPECT_EQ(coarse.U.fab(0)(3, 1, 0), Real(5)); + EXPECT_EQ(coarse.U.fab(0)(3, 6, 0), Real(6)); + EXPECT_EQ(coarse.U.fab(0)(2, 2, 0), Real(0)) + << "covered parent cells remain outside the sparse correction"; +} + +TEST(test_program_reflux_ledger, prepared_local_kernel_rejects_unwritten_output_atomically) { + const Box2D coarse_domain{{0, 0}, {7, 7}}; + const BoxArray coarse_boxes(std::vector{coarse_domain}); + const BoxArray fine_boxes(std::vector{Box2D{{4, 4}, {11, 11}}}); + const DistributionMapping coarse_mapping(coarse_boxes.size(), n_ranks()); + const DistributionMapping fine_mapping(fine_boxes.size(), n_ranks()); + AmrLevelMP coarse{MultiFab(coarse_boxes, coarse_mapping, 1, 0), nullptr, Real(0.5), Real(0.5)}; + AmrLevelMP fine{MultiFab(fine_boxes, fine_mapping, 1, 0), nullptr, Real(0.25), Real(0.25)}; + coarse.U.set_val(Real(0)); + fine.U.set_val(Real(0)); + + const auto incomplete = [](const PreparedAmrRefluxLocalRequest& request) { + const auto x_size = + static_cast(request.correction.J1 - request.correction.J0 + 1) * + static_cast(request.correction.components); + std::fill_n(request.correction.x_low, x_size, Real(1)); + }; + auto transition = PreparedAmrProgramRefluxTransition::prepare_with_local_kernel( + coarse, fine, coarse_domain, Periodicity{false, false}, 0, "pops://test/reflux", incomplete, + world_communicator_view()); + + EdgeStrip coarse_role = make_strip(2, 5, 2, 5, 1); + EdgeStrip fine_role = make_strip(2, 5, 2, 5, 1); + coarse_role.cL.assign(4, Real(1)); + coarse_role.cR.assign(4, Real(1)); + coarse_role.cB.assign(4, Real(1)); + coarse_role.cT.assign(4, Real(1)); + fine_role.fL.assign(4, Real(2)); + fine_role.fR.assign(4, Real(2)); + fine_role.fB.assign(4, Real(2)); + fine_role.fT.assign(4, Real(2)); + const amr::ClockStamp logical_time{0, 3, amr::Rational(0, 1), 0.3}; + EXPECT_THROW(transition.synchronize_integrated( + coarse.U, coarse.dx, coarse.dy, std::vector{coarse_role}, + std::vector{fine_role}, world_communicator_view(), &logical_time), + std::runtime_error); + + ASSERT_EQ(coarse.U.local_size(), 1); + EXPECT_EQ(coarse.U.fab(0)(1, 3, 0), Real(0)); + EXPECT_EQ(coarse.U.fab(0)(6, 3, 0), Real(0)); + EXPECT_EQ(coarse.U.fab(0)(3, 1, 0), Real(0)); + EXPECT_EQ(coarse.U.fab(0)(3, 6, 0), Real(0)); +} + TEST(test_program_reflux_ledger, edge_flux_axpy_rejects_shifted_equal_width_footprints) { EdgeFlux destination; destination.fine.push_back(make_strip(2, 5, 2, 5, 1));