merge queue: checking #2376 + #2363 + #2333 + #2375 + #2372 together on main (8c054a1)

.github/workflows/main_workflow.yml

@@ -170,6 +170,27 @@ jobs:
     with:
       fail_fast: ${{ inputs.is_merge_queue }}
 
+  # On the Mergify merge queue, dockerbuild/doc/phystest failing means the
+  # merge gate ("all") will fail anyway, so cancel the whole run early
+  # instead of waiting on the remaining CI jobs (asan, ubsan, tidy, coverage,
+  # pylib) to finish.
+  cancel_on_merge_queue_phystest_failure:
+    name: Cancel run (merge queue phystest failure)
+    needs:
+      [build_push_docker, make_documentation, shamrock_linux_acpp_phystests]
+    if: ${{ always() && inputs.is_merge_queue && contains(needs.*.result, 'failure') }}
+    runs-on: ubuntu-latest
+    permissions:
+      actions: write
+    steps:
+      - name: Cancel this workflow run
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GH_REPO: ${{ github.repository }}
+        run: |
+          echo "Mergify merge queue: dockerbuild/doc/phystest failed, cancelling run ${{ github.run_id }}."
+          gh run cancel "${{ github.run_id }}"
+
   shamrock_linux_acpp_pylib:
     needs: [src_check]
     name: Tests

src/shammodels/sph/include/shammodels/sph/Solver.hpp

@@ -62,6 +62,85 @@ namespace shammodels::sph {
         i32 iter_count;
     };
 
+    /**
+     * @brief Tracks the wall-clock budget for Solver::evolve_until(): decides when the next
+     * walltime check is due, and whether the limit has been exceeded.
+     */
+    struct WalltimeLimiter {
+        bool active;
+        f64 max_walltime;
+        f64 start_wall_time;
+        i32 next_check_iter;
+
+        inline WalltimeLimiter(bool active, f64 max_walltime)
+            : active(active), max_walltime(max_walltime) {
+            start_wall_time = active ? synced_wtime() : 0;
+            next_check_iter = active ? 1 : std::numeric_limits<i32>::max();
+        }
+
+        inline f64 synced_wtime() {
+            if (active) {
+                return shamalgs::collective::allreduce_max(shambase::details::get_wtime());
+            }
+            return 0;
+        }
+
+        /// True if the next walltime check is due at this iteration count
+        inline bool due(i32 iter_count) const { return active && iter_count >= next_check_iter; }
+
+        /// Must only be called when due(iter_count) is true. Returns true if the walltime
+        /// limit has been reached, otherwise updates the next check iteration estimate.
+        inline bool exceeded(i32 iter_count) {
+            f64 global_walltime = synced_wtime();
+
+            // if the global walltime is greater than the max walltime
+            if (global_walltime >= max_walltime) {
+                if (shamcomm::world_rank() == 0) {
+                    logger::info_ln(
+                        "SPH",
+                        sham::format(
+                            "stopping evolve until because of "
+                            "max_walltime = {:.2f}s > {:.2f}s",
+                            global_walltime,
+                            max_walltime));
+                }
+                return true;
+            }
+
+            f64 sec_per_iter = (global_walltime - start_wall_time) / static_cast<f64>(iter_count);
+
+            auto get_remaining_iters = [&](f64 delta_walltime, f64 factor) -> i32 {
+                if (sec_per_iter > 0) {
+                    f64 tmp = factor * delta_walltime / sec_per_iter;
+                    if (tmp > std::numeric_limits<i32>::max()) {
+                        return std::numeric_limits<i32>::max();
+                    }
+                    return static_cast<i32>(tmp);
+                }
+                return 1000; // default to 1000 iterations if sec_per_iter is 0
+            };
+
+            i32 iters_to_limit      = get_remaining_iters(max_walltime - global_walltime, 0.25);
+            i32 iters_to_next_check = iters_to_limit;
+
+            next_check_iter = iter_count + std::max(1, iters_to_next_check);
+
+            if (shamcomm::world_rank() == 0) {
+                logger::info_ln(
+                    "SPH",
+                    sham::format(
+                        "next walltime check in {:.2f}s (niter = {}) global walltime = "
+                        "{:.2f}s (max_walltime = {:.2f}s)",
+                        iters_to_next_check * sec_per_iter,
+                        iters_to_next_check,
+                        global_walltime,
+                        max_walltime));
+            }
+
+            return false;
+        }
+    };
+
     /**
      * @brief The shamrock SPH model
      *
@@ -320,13 +399,6 @@ namespace shammodels::sph {
                         max_walltime));
             }
 
-            auto synced_wtime = [&]() -> f64 {
-                if (walltime_limit_active) {
-                    return shamalgs::collective::allreduce_max(shambase::details::get_wtime());
-                }
-                return 0;
-            };
-
             auto step = [&]() {
                 Tscal dt = get_dt_sph();
                 Tscal t  = get_time();
@@ -342,10 +414,7 @@ namespace shammodels::sph {
                 evolve_once();
             };
 
-            f64 start_wall_time = (walltime_limit_active) ? synced_wtime() : 0;
-
-            i32 next_walltime_check_iter
-                = walltime_limit_active ? 1 : std::numeric_limits<i32>::max();
+            WalltimeLimiter walltime_limiter(walltime_limit_active, max_walltime);
 
             i32 iter_count = 0;
 
@@ -368,58 +437,16 @@ namespace shammodels::sph {
                 }
 
                 // if walltime limit is active and the next walltime check is due
-                if (walltime_limit_active && iter_count >= next_walltime_check_iter) {
-                    f64 global_walltime = synced_wtime();
-
-                    // if the global walltime is greater than the max walltime
-                    if (global_walltime >= max_walltime) {
-                        if (shamcomm::world_rank() == 0) {
-                            logger::info_ln(
-                                "SPH",
-                                sham::format(
-                                    "stopping evolve until because of "
-                                    "max_walltime = {:.2f}s > {:.2f}s",
-                                    global_walltime,
-                                    max_walltime));
-                        }
+                if (walltime_limiter.due(iter_count)) {
+                    // must be inside the .due if since there is a MPI reduction in exceeded
+                    if (walltime_limiter.exceeded(iter_count)) {
                         return {
                             .reach_target_time  = false,
                             .reach_niter_max    = false,
                             .reach_max_walltime = true,
                             .iter_count         = iter_count,
                         };
                     }
-
-                    f64 sec_per_iter
-                        = (global_walltime - start_wall_time) / static_cast<f64>(iter_count);
-
-                    auto get_remaining_iters = [&](f64 delta_walltime, f64 factor) -> i32 {
-                        if (sec_per_iter > 0) {
-                            f64 tmp = factor * delta_walltime / sec_per_iter;
-                            if (tmp > std::numeric_limits<i32>::max()) {
-                                return std::numeric_limits<i32>::max();
-                            }
-                            return static_cast<i32>(tmp);
-                        }
-                        return 1000; // default to 1000 iterations if sec_per_iter is 0
-                    };
-
-                    i32 iters_to_limit = get_remaining_iters(max_walltime - global_walltime, 0.25);
-                    i32 iters_to_next_check = iters_to_limit;
-
-                    next_walltime_check_iter = iter_count + std::max(1, iters_to_next_check);
-
-                    if (shamcomm::world_rank() == 0) {
-                        logger::info_ln(
-                            "SPH",
-                            sham::format(
-                                "next walltime check in {:.2f}s (niter = {}) global walltime = "
-                                "{:.2f}s (max_walltime = {:.2f}s)",
-                                iters_to_next_check * sec_per_iter,
-                                iters_to_next_check,
-                                global_walltime,
-                                max_walltime));
-                    }
                 }
             }
 

src/shammodels/sph/include/shammodels/sph/modules/SinkParticlesUpdate.hpp

@@ -41,7 +41,6 @@ namespace shammodels::sph::modules {
         SinkParticlesUpdate(ShamrockCtx &context, Config &solver_config, Storage &storage)
             : context(context), solver_config(solver_config), storage(storage) {}
 
-        void predictor_step(Tscal dt);
         void compute_sph_forces();
         void corrector_step(Tscal dt);
 

src/shammodels/sph/src/Solver.cpp

@@ -111,6 +111,8 @@
 #include "shamsolvergraph/SolverGraph.hpp"
 #include "shamsolvergraph/edge/IDataEdge.hpp"
 #include "shamsolvergraph/edge/IDataEdgeSerializable.hpp"
+#include "shamsolvergraph/node/ForwardEulerHost.hpp"
+#include "shamsolvergraph/node/ForwardEulerHost2Deriv.hpp"
 #include "shamsolvergraph/node/NodeFreeAlloc.hpp"
 #include "shamsolvergraph/node/NodeMapEdge.hpp"
 #include "shamsolvergraph/node/NodeSetEdge.hpp"
@@ -845,6 +847,62 @@ void shammodels::sph::Solver<Tvec, Kern>::init_solver_graph() {
             .set_edges(solver_graph.get_edge_ptr<IDataEdge<bool>>("has_sinks"));
     }
 
+    ////////////////////////////////////////////////////////////////////////////////////////
+    // sink predictor step (leapfrog kick-drift of the sink particles themselves)
+    ////////////////////////////////////////////////////////////////////////////////////////
+    {
+        solver_graph.register_edge(
+            "sink_predictor_dt_half", IDataEdge<Tscal>("dt_half", "\\frac{dt}{2}"));
+
+        auto sink_predictor_dt_to_half_dt = solver_graph.register_node(
+            "sink_predictor_dt_to_half_dt",
+            NodeMapEdge<IDataEdge<Tscal>, IDataEdge<Tscal>>{
+                [](const IDataEdge<Tscal> &dt, IDataEdge<Tscal> &half_dt) {
+                    half_dt.data = dt.data / 2;
+                }});
+        shambase::get_check_ref(sink_predictor_dt_to_half_dt)
+            .set_edges(
+                sync_data.get_edge_ptr<IDataEdge<Tscal>>("dt"),
+                solver_graph.get_edge_ptr<IDataEdge<Tscal>>("sink_predictor_dt_half"));
+
+        auto sink_predictor_vel_update = solver_graph.register_node(
+            "sink_predictor_vel_update", ForwardEulerHost2Deriv<Tvec, Tscal>{});
+        shambase::get_check_ref(sink_predictor_vel_update)
+            .set_edges(
+                solver_graph.get_edge_ptr<IDataEdge<Tscal>>("sink_predictor_dt_half"),
+                sync_data.get_edge_ptr<IDataEdgeSerializable<std::vector<Tvec>>>("sink_acc_sph"),
+                sync_data.get_edge_ptr<IDataEdgeSerializable<std::vector<Tvec>>>("sink_acc_ext"),
+                sync_data.get_edge_ptr<IDataEdgeSerializable<std::vector<Tvec>>>("sink_vel"));
+
+        auto sink_predictor_pos_update = solver_graph.register_node(
+            "sink_predictor_pos_update", ForwardEulerHost<Tvec, Tscal>{});
+        shambase::get_check_ref(sink_predictor_pos_update)
+            .set_edges(
+                sync_data.get_edge_ptr<IDataEdge<Tscal>>("dt"),
+                sync_data.get_edge_ptr<IDataEdgeSerializable<std::vector<Tvec>>>("sink_vel"),
+                sync_data.get_edge_ptr<IDataEdgeSerializable<std::vector<Tvec>>>("sink_pos"));
+
+        auto sink_predictor_body = solver_graph.register_node(
+            "sink_predictor_body",
+            OperationSequence(
+                "sink predictor body",
+                {
+                    // recompute the sink self-gravity at the current (pre-predictor) sink
+                    // positions before using it to kick the sink velocities
+                    solver_graph.get_node_ptr_base("sink ext force"),
+                    sink_predictor_dt_to_half_dt,
+                    sink_predictor_vel_update,
+                    sink_predictor_pos_update,
+                }));
+
+        // register the actual node that will be used, gated on the same "has_sinks" edge
+        // maintained by the "sink accretion" section above
+        auto sink_predictor = solver_graph.register_node(
+            "sink predictor", OperationIf("sink predictor", sink_predictor_body));
+        shambase::get_check_ref(sink_predictor)
+            .set_edges(solver_graph.get_edge_ptr<IDataEdge<bool>>("has_sinks"));
+    }
+
     ////////////////////////////////////////////////////////////////////////////////////////
     // external force (point mass) accretion
     ////////////////////////////////////////////////////////////////////////////////////////
@@ -2233,7 +2291,7 @@ shammodels::sph::TimestepLog shammodels::sph::Solver<Tvec, Kern>::evolve_once()
 
     modules::SinkParticlesUpdate<Tvec, Kern> sink_update(context, solver_config, storage);
 
-    sink_update.predictor_step(dt);
+    storage.solver_graph.get_node_ref_base("sink predictor").evaluate();
 
     {
         // beginning of SolverGraph migration

src/shammodels/sph/src/modules/SinkParticlesUpdate.cpp

@@ -18,49 +18,8 @@
 #include "shammodels/sph/modules/SinkParticlesUpdate.hpp"
 #include "shammath/sphkernels.hpp"
 #include "shammodels/sph/sink_edges_helper.hpp"
-#include "shamsolvergraph/edge/IDataEdge.hpp"
-#include "shamsolvergraph/edge/IDataEdgeSerializable.hpp"
-#include "shamsolvergraph/node/ForwardEulerHost.hpp"
-#include "shamsolvergraph/node/ForwardEulerHost2Deriv.hpp"
 #include <vector>
 
-template<class Tvec, template<class> class SPHKernel>
-void shammodels::sph::modules::SinkParticlesUpdate<Tvec, SPHKernel>::predictor_step(Tscal dt) {
-
-    StackEntry stack_loc{};
-
-    auto &sync = scheduler().synchronized_data;
-    auto &pos  = get_sink_pos<Tvec>(sync);
-    if (pos.empty()) {
-        return;
-    }
-
-    storage.solver_graph.get_node_ref_base("sink ext force").evaluate();
-
-    using VecEdge = shamrock::solvergraph::IDataEdgeSerializable<std::vector<Tvec>>;
-
-    auto dt_half_edge  = shamrock::solvergraph::IDataEdge<Tscal>::make_shared("dt_half", "dt/2");
-    dt_half_edge->data = dt / 2;
-
-    shamrock::solvergraph::ForwardEulerHost2Deriv<Tvec, Tscal> vel_update{};
-    vel_update.set_edges(
-        dt_half_edge,
-        sync.template get_edge_ptr<VecEdge>("sink_acc_sph"),
-        sync.template get_edge_ptr<VecEdge>("sink_acc_ext"),
-        sync.template get_edge_ptr<VecEdge>("sink_vel"));
-    vel_update.evaluate();
-
-    auto dt_edge  = shamrock::solvergraph::IDataEdge<Tscal>::make_shared("dt", "dt");
-    dt_edge->data = dt;
-
-    shamrock::solvergraph::ForwardEulerHost<Tvec, Tscal> pos_update{};
-    pos_update.set_edges(
-        dt_edge,
-        sync.template get_edge_ptr<VecEdge>("sink_vel"),
-        sync.template get_edge_ptr<VecEdge>("sink_pos"));
-    pos_update.evaluate();
-}
-
 template<class Tvec, template<class> class SPHKernel>
 void shammodels::sph::modules::SinkParticlesUpdate<Tvec, SPHKernel>::corrector_step(Tscal dt) {
 

src/shamrock/include/shamrock/solvergraph/ExchangeGhostLayerDebugDotGraph.hpp

@@ -24,6 +24,13 @@
 #include "shamrock/solvergraph/ScalarsEdge.hpp"
 #include "shamsolvergraph/node/INode.hpp"
 
+#define NODE_EDGES(X_RO, X_RW)                                                                     \
+    /* ------------------- inputs ------------------- */                                           \
+    X_RO(shamrock::solvergraph::ScalarsEdge<u64>, object_counts)                                   \
+                                                                                                   \
+    /* ------------------- outputs ------------------- */                                          \
+    X_RW(shamrock::solvergraph::PatchDataLayerDDShared, ghost_layer)
+
 namespace shamrock::solvergraph {
 
     class ExchangeGhostLayerDebugDotGraph : public shamrock::solvergraph::INode {
@@ -33,24 +40,7 @@ namespace shamrock::solvergraph {
         public:
         ExchangeGhostLayerDebugDotGraph() {}
 
-        struct Edges {
-            const shamrock::solvergraph::ScalarsEdge<u64> &object_counts;
-            shamrock::solvergraph::PatchDataLayerDDShared &ghost_layer;
-        };
-
-        inline void set_edges(
-            std::shared_ptr<shamrock::solvergraph::ScalarsEdge<u64>> object_counts,
-            std::shared_ptr<shamrock::solvergraph::PatchDataLayerDDShared> ghost_layer) {
-            __internal_set_ro_edges({object_counts});
-            __internal_set_rw_edges({ghost_layer});
-        }
-
-        inline Edges get_edges() {
-            return Edges{
-                .object_counts = get_ro_edge<shamrock::solvergraph::ScalarsEdge<u64>>(0),
-                .ghost_layer   = get_rw_edge<shamrock::solvergraph::PatchDataLayerDDShared>(0),
-            };
-        }
+        EXPAND_NODE_EDGES(NODE_EDGES)
 
         void _impl_evaluate_internal() {
             auto edges        = get_edges();
@@ -155,3 +145,5 @@ namespace shamrock::solvergraph {
         inline virtual std::string _impl_get_tex() const { return ""; };
     };
 } // namespace shamrock::solvergraph
+
+#undef NODE_EDGES