perf: add nexus flush() to bound memory on long runs

include/core/nexus/HY_HydroNexus.hpp

@@ -39,6 +39,28 @@ class HY_HydroNexus
 
     /** get the units that the flows are described in */
     virtual std::string get_flow_units()=0;
+
+    /**
+     * @brief Release the internal per-timestep state of the nexus.
+     *
+     * A nexus accumulates flow state for each time step it sees. On long runs
+     * that growth is unbounded, so callers should release state once a time
+     * step's data has been consumed. The accumulator state is always released.
+     *
+     * Separately, the nexus records which time steps are "completed" (fully
+     * drained -- 100% of the flow requested). That record is a correctness
+     * guard: add/get on a completed time step throws rather than re-processing
+     * water that is already allocated. It only ever grows.
+     *
+     * @param clear_completed also drop the completed-time-step record.
+     *        - false: keep the guard, so a stray add/get on an already-drained
+     *          time step still throws (but the record keeps growing).
+     *        - true: bound that record too, at the cost of the guard -- a later
+     *          add/get on a dropped time step is silently re-accumulated, not
+     *          rejected. Safe only for a caller that advances strictly forward
+     *          and never revisits a flushed time step (e.g. SurfaceLayer).
+     */
+    virtual void flush(bool clear_completed = false) = 0;
 
     const Catchments& get_receiving_catchments() {
         return receiving_catchments;

include/core/nexus/HY_PointHydroNexus.hpp

@@ -28,7 +28,8 @@ class HY_PointHydroNexus : public HY_HydroNexus
         /** get the units that flows are represented in. */
         std::string get_flow_units() override;
 
-        void set_mintime(time_step_t);
+        /** @brief Release accumulated per-timestep nexus state (see HY_HydroNexus::flush). */
+        void flush(bool clear_completed = false) override;
 
     protected:
     using flows = std::pair<std::string, double>;
@@ -40,7 +41,6 @@ class HY_PointHydroNexus : public HY_HydroNexus
     std::unordered_map<time_step_t, double> summed_flows;
     std::unordered_map<time_step_t, double> total_requests;
 
-    time_step_t min_timestep{0};
     std::unordered_set<time_step_t> completed;
 
 };

include/core/nexus/HY_PointHydroNexusRemote.hpp

@@ -70,6 +70,9 @@ class HY_PointHydroNexusRemote : public HY_PointHydroNexus
         /** return the communicator type for this nexus */
 		communication_type get_communicator_type() { return type; }
 
+        /** @brief Release accumulated nexus state, draining in-flight MPI first. */
+        void flush(bool clear_completed = false) override;
+
     private:
         void post_receives();
         void process_communications();

src/core/SurfaceLayer.cpp

@@ -73,7 +73,11 @@ void ngen::SurfaceLayer::update_models(boost::span<double> catchment_outflows,
         //nexus_outputs_mgr->receive_data_entry(form_id, id, current_time_index, current_timestamp, contribution_at_t);
         nexus_outputs_mgr->receive_data_entry(id, current_time_marker, contribution_at_t);
 
-        //std::cout<<"\tNexus "<<id<<" has "<<contribution_at_t<<" m^3/s"<<std::endl;
+        // Release this nexus's accumulated per-timestep state to bound memory on
+        // long runs. clear_completed=true is safe here: this nexus's contribution
+        // for the current time step has just been collected and committed, and
+        // nothing re-reads it afterward for this step.
+        nexus->flush(true);
     } //done nexuses
     nexus_outputs_mgr->commit_writes();
 }

src/core/nexus/HY_PointHydroNexus.cpp

@@ -24,11 +24,6 @@ struct completed_time_step : public boost::exception, public std::exception
   const char *what() const noexcept override { return "Can not operate on a completed time step"; }
 };
 
-struct invalid_time_step : public boost::exception, public std::exception
-{
-  const char *what() const noexcept override { return "Time step before minimum time step requested"; }
-};
-
 HY_PointHydroNexus::HY_PointHydroNexus(std::string nexus_id, Catchments receiving_catchments) : HY_HydroNexus( nexus_id, receiving_catchments), upstream_flows()
 {
 
@@ -47,7 +42,6 @@ HY_PointHydroNexus::~HY_PointHydroNexus()
 double HY_PointHydroNexus::get_downstream_flow(std::string catchment_id, time_step_t t, double percent_flow)
 {
 
-    if ( t < min_timestep ) BOOST_THROW_EXCEPTION(invalid_time_step());
     if ( completed.find(t) != completed.end() ) BOOST_THROW_EXCEPTION(completed_time_step());
 
     auto s1 = upstream_flows.find(t);
@@ -144,7 +138,6 @@ double HY_PointHydroNexus::get_downstream_flow(std::string catchment_id, time_st
 
 void HY_PointHydroNexus::add_upstream_flow(double val, std::string catchment_id, time_step_t t)
 {
-     if ( t < min_timestep ) BOOST_THROW_EXCEPTION(invalid_time_step());
     if ( completed.find(t) != completed.end() ) BOOST_THROW_EXCEPTION(completed_time_step());
 
     auto s1 = upstream_flows.find(t);
@@ -226,47 +219,14 @@ std::string HY_PointHydroNexus::get_flow_units()
     return std::string("m3/s");
 }
 
-void HY_PointHydroNexus::set_mintime(time_step_t t)
+void HY_PointHydroNexus::flush(bool clear_completed)
 {
-    min_timestep = t;
-
-    // remove expired time steps from completed
-    for( auto& t: completed)
-    {
-        if ( t < min_timestep )
-        {
-            completed.erase(t);
-        }
+    // Release the memory held for all accumulated time steps.
+    upstream_flows.clear();
+    downstream_requests.clear();
+    summed_flows.clear();
+    total_requests.clear();
+    if (clear_completed) {
+        completed.clear();
     }
-
-    // C++ 2014 would allow this do be done with a single lambda
-    auto l1 = [](int min_v, std::unordered_map<long,flow_vector>& v)
-    {
-        for( auto& t: v)
-        {
-            if ( t.first < min_v )
-            {
-                v.erase(t.first);
-            }
-        }
-    };
-
-    // C++ 2014 would allow this do be done with a single lambda
-    auto l2 = [](int min_v, std::unordered_map<long,double>& v)
-    {
-        for( auto& t: v)
-        {
-            if ( t.first < min_v )
-            {
-                v.erase(t.first);
-            }
-        }
-    };
-
-    // remove expired time steps from all maps
-    l1(min_timestep,downstream_requests);
-    l1(min_timestep,upstream_flows);
-    l2(min_timestep,summed_flows);
-    l2(min_timestep,total_requests);
-
 }

src/core/nexus/HY_PointHydroNexusRemote.cpp

@@ -311,6 +311,27 @@ void HY_PointHydroNexusRemote::process_communications()
     }
 }
 
+void HY_PointHydroNexusRemote::flush(bool clear_completed)
+{
+    // Release the unbounded per-timestep accumulator state (the actual memory
+    // growth on long runs). This is intentionally NON-BLOCKING:
+    //
+    //  - process_communications() opportunistically reaps already-completed
+    //    sends/receives but never waits. In-flight sends are left in
+    //    stored_sends to complete naturally; they are already bounded by the
+    //    spinlock in add_upstream_flow and carry their own payload buffers, so
+    //    clearing the accumulators below cannot corrupt them.
+    //  - No new receives are posted: get_downstream_flow has already drained
+    //    this nexus's receives for the step, and posting one here would consume
+    //    a future timestep's message that the clear would then discard.
+    //
+    // Blocking here (e.g. waiting for sends to be received) would turn a
+    // per-timestep flush into an inter-rank synchronization point and can stall
+    // forward progress under load imbalance, so it is deliberately avoided.
+    process_communications();
+    HY_PointHydroNexus::flush(clear_completed);
+}
+
 long HY_PointHydroNexusRemote::get_time_step()
 {
    return time_step;

test/core/nexus/NexusRemoteTests.cpp

@@ -6,6 +6,8 @@
 
 #include <vector>
 #include <memory>
+#include <cmath>
+#include <string>
 
 #include <unistd.h>
 
@@ -123,6 +125,58 @@ TEST_F(Nexus_Remote_Test, TestInit0)
 
 }
 
+// Exercise the production pattern: exchange a flow each timestep, then call
+// flush(true) on every rank (as SurfaceLayer does). flush() must release state
+// without deadlocking, including on the final timestep where no further send
+// follows. Run with `mpirun -np 2`.
+TEST_F(Nexus_Remote_Test, TestFlushPerTimestepNoDeadlock)
+{
+    if ( mpi_num_procs < 2 ) {
+        GTEST_SKIP();
+    }
+
+    HY_PointHydroNexusRemote::catcment_location_map_t loc_map;
+    std::shared_ptr<HY_PointHydroNexusRemote> nexus;
+    std::vector<std::string> upstream_catchments = {"cat-26"};
+    std::vector<std::string> downstream_catchments = {"cat-27"};
+
+    // Only ranks 0 (sender) and 1 (receiver) take part in the exchange, but
+    // every rank MUST still reach the collective MPI_Barrier below. Returning
+    // early on the other ranks (e.g. GTEST_SKIP) would leave them out of the
+    // MPI_COMM_WORLD barrier, desynchronizing the world-collective sequence for
+    // the whole process set and deadlocking a later test.
+    if ( mpi_rank == 0 ) {
+        loc_map["cat-27"] = 1;
+        nexus = std::make_shared<HY_PointHydroNexusRemote>("nex-26", downstream_catchments, upstream_catchments, loc_map);
+    } else if ( mpi_rank == 1 ) {
+        loc_map["cat-26"] = 0;
+        nexus = std::make_shared<HY_PointHydroNexusRemote>("nex-26", downstream_catchments, upstream_catchments, loc_map);
+    }
+
+    if ( nexus )
+    {
+        long ts = 0;
+        for ( auto discharge : stored_discharge )
+        {
+            if ( mpi_rank == 0 ) {
+                nexus->add_upstream_flow(discharge, "cat-26", ts);
+            } else if ( mpi_rank == 1 ) {
+                double received = nexus->get_downstream_flow("cat-27", ts, 100);
+                // EXPECT (not ASSERT): a failure must not early-return past the
+                // collective barrier below and desynchronize the other ranks.
+                EXPECT_EQ(discharge, received);
+            }
+
+            // Production SurfaceLayer pattern: flush each nexus every timestep.
+            nexus->flush(true);
+            ++ts;
+        }
+    }
+
+    MPI_Barrier(MPI_COMM_WORLD);
+    ASSERT_TRUE(true);
+}
+
 //Test sending data with MPI from an two upstream remote nexus
 //to a downstream remote nexus.
 TEST_F(Nexus_Remote_Test, Test2RemoteSenders)
@@ -1109,12 +1163,90 @@ TEST_F(Nexus_Remote_Test, TestRemoteNexusDeadlockFree)
 
     senders.clear();
     receivers.clear();
-    
+
     MPI_Barrier(MPI_COMM_WORLD);
     std::cerr << "Rank " << mpi_rank << ": Test PASSED - no deadlock with remote nexus\n";
 }
 
 
+// Multi-timestep pipeline with timing skew, flushing every nexus every step.
+//
+// Chain of links r -> r+1 across ranks 0..3. Rank 1 deliberately lags each
+// timestep, so its upstream sender (rank 0) races ahead and future-timestep
+// messages pile up in MPI before rank 1 consumes/flushes the current step.
+// The flow on link L at time ts is (L+1)*1000 + ts; each receiver verifies it
+// gets exactly what was sent, every step through the last, with flush(true) on
+// every nexus every step. This exercises that flush() neither loses buffered
+// in-flight data nor deadlocks under skew. Run with `mpirun -np 4`.
+TEST_F(Nexus_Remote_Test, TestFlushUnderTimingSkew)
+{
+    if ( mpi_num_procs < 4 ) {
+        GTEST_SKIP();
+    }
+
+    const int N = 4;          // ranks participating in the chain
+    const long NUM_TS = 5;
+    auto flow_for = [](int link, long ts) { return (link + 1) * 1000.0 + ts; };
+
+    std::shared_ptr<HY_PointHydroNexusRemote> sender;   // link mpi_rank   -> mpi_rank+1
+    std::shared_ptr<HY_PointHydroNexusRemote> receiver; // link mpi_rank-1 -> mpi_rank
+
+    // This rank is the sender side of link mpi_rank -> mpi_rank+1
+    if ( mpi_rank < N - 1 ) {
+        int link = mpi_rank;
+        std::string up   = "cat-up-"   + std::to_string(link);
+        std::string down = "cat-down-" + std::to_string(link);
+        HY_PointHydroNexusRemote::catcment_location_map_t loc_map;
+        loc_map[down] = mpi_rank + 1; // downstream catchment lives on the next rank
+        sender = std::make_shared<HY_PointHydroNexusRemote>(
+            "nex-" + std::to_string(link),
+            std::vector<std::string>{down}, std::vector<std::string>{up}, loc_map);
+    }
+
+    // This rank is the receiver side of link mpi_rank-1 -> mpi_rank
+    if ( mpi_rank > 0 && mpi_rank < N ) {
+        int link = mpi_rank - 1;
+        std::string up   = "cat-up-"   + std::to_string(link);
+        std::string down = "cat-down-" + std::to_string(link);
+        HY_PointHydroNexusRemote::catcment_location_map_t loc_map;
+        loc_map[up] = mpi_rank - 1; // upstream catchment lives on the previous rank
+        receiver = std::make_shared<HY_PointHydroNexusRemote>(
+            "nex-" + std::to_string(link),
+            std::vector<std::string>{down}, std::vector<std::string>{up}, loc_map);
+    }
+
+    MPI_Barrier(MPI_COMM_WORLD);
+
+    for ( long ts = 0; ts < NUM_TS; ++ts )
+    {
+        // Skew: rank 1 lags so rank 0 races ahead and buffers future messages.
+        if ( mpi_rank == 1 ) {
+            volatile double dummy = 0.0;
+            for ( int i = 0; i < 20000000; ++i ) {
+                dummy += std::sin(i * 0.0001) * std::cos(i * 0.0002);
+            }
+        }
+
+        if ( sender ) {
+            sender->add_upstream_flow(flow_for(mpi_rank, ts), "cat-up-" + std::to_string(mpi_rank), ts);
+        }
+        if ( receiver ) {
+            int link = mpi_rank - 1;
+            double recv = receiver->get_downstream_flow("cat-down-" + std::to_string(link), ts, 100.0);
+            EXPECT_DOUBLE_EQ(recv, flow_for(link, ts))
+                << "rank " << mpi_rank << " link " << link << " timestep " << ts;
+        }
+
+        // Production SurfaceLayer pattern: flush every nexus every timestep.
+        if ( sender )   sender->flush(true);
+        if ( receiver ) receiver->flush(true);
+    }
+
+    MPI_Barrier(MPI_COMM_WORLD);
+    ASSERT_TRUE(true);
+}
+
+
 //#endif  // NGEN_MPI_TESTS_ACTIVE
 
 //#endif  // NGEN_MPI_TESTS_ACTIVE