update RC rho for PR #476

Author: bknueven

Diff for: mpisppy/extensions/reduced_costs_fixer.py

@@ -85,7 +85,7 @@ def iter0_post_solver_creation(self):
             if self.opt.cylinder_rank == 0 and self.verbose:
                 print("Fixing based on reduced costs prior to iteration 0!")
             if self.reduced_cost_buf.id() == 0:
-                while not self.opt.spcomm.hub_from_spoke(self.outer_bound_buf, self.reduced_costs_spoke_index, Field.EXPECTED_REDUCED_COST):
+                while not self.opt.spcomm.get_receive_buffer(self.outer_bound_buf, Field.EXPECTED_REDUCED_COST, self.reduced_costs_spoke_index):
                     continue
             self.sync_with_spokes(pre_iter0 = True)
         self.fix_fraction_target = self._fix_fraction_target_iter0

Diff for: mpisppy/extensions/reduced_costs_rho.py

@@ -10,7 +10,8 @@
 import numpy as np
 from mpisppy import global_toc
 from mpisppy.extensions.sensi_rho import _SensiRhoBase
-from mpisppy.cylinders.reduced_costs_spoke import ReducedCostsSpoke 
+
+from mpisppy.cylinders.spwindow import Field
 
 class ReducedCostsRho(_SensiRhoBase):
     """
@@ -41,24 +42,20 @@ def __init__(self, ph, comm=None):
         self._last_serial_number = -1
         self.reduced_costs_spoke_index = None
 
-    def initialize_spoke_indices(self):
-        for (i, spoke) in enumerate(self.opt.spcomm.spokes):
-            if spoke["spoke_class"] == ReducedCostsSpoke:
-                self.reduced_costs_spoke_index = i + 1
-        if self.reduced_costs_spoke_index is None:
-            raise RuntimeError("ReducedCostsRho requires a ReducedCostsSpoke for calculations")
-
-    def _get_serial_number(self):
-        return int(round(self.opt.spcomm.outerbound_receive_buffers[self.reduced_costs_spoke_index][-1]))
+    def register_receive_fields(self):
+        spcomm = self.opt.spcomm
+        reduced_cost_ranks = spcomm.fields_to_ranks[Field.SCENARIO_REDUCED_COST]
+        assert len(reduced_cost_ranks) == 1
+        self.reduced_costs_spoke_index = reduced_cost_ranks[0]
 
-    def _get_reduced_costs_from_spoke(self):
-        return self.opt.spcomm.outerbound_receive_buffers[self.reduced_costs_spoke_index][1+self.nonant_length:1+self.nonant_length+len(self._scenario_rc_buffer)]
+        self.scenario_reduced_cost_buf = spcomm.register_extension_recv_field(
+            Field.SCENARIO_REDUCED_COST,
+            self.reduced_costs_spoke_index,
+        )
 
     def sync_with_spokes(self):
-        serial_number = self._get_serial_number()
-        if serial_number > self._last_serial_number:
-            self._last_serial_number = serial_number
-            self._scenario_rc_buffer[:] = self._get_reduced_costs_from_spoke()
+        if self.scenario_reduced_cost_buf.is_new():
+            self._scenario_rc_buffer[:] = self.scenario_reduced_cost_buf.value_array()
             # print(f"In ReducedCostsRho; {self._scenario_rc_buffer=}")
         else:
             if self.opt.cylinder_rank == 0 and self.verbose:
@@ -89,8 +86,8 @@ def post_iter0_after_sync(self):
         global_toc("Using reduced cost rho setter")
         self.update_caches()
         # wait until the spoke has data
-        if self._get_serial_number() == 0:
-            while not self.ph.spcomm.hub_from_spoke(self.opt.spcomm.outerbound_receive_buffers[self.reduced_costs_spoke_index], self.reduced_costs_spoke_index):
+        if self.scenario_reduced_cost_buf.id() == 0:
+            while not self.ph.spcomm.get_receive_buffer(self.scenario_reduced_cost_buf, Field.SCENARIO_REDUCED_COST, self.reduced_costs_spoke_index):
                 continue
             self.sync_with_spokes()
         self.compute_and_update_rho()