Merge pull request #496 from bknueven/communicator_refactor

Communicator refactor

Author: bknueven

examples/run_all.py

@@ -280,6 +280,7 @@ def do_one_mmw(dirname, runefstring, npyfile, mmwargstring):
        "--integer-relax-then-enforce "
        "--integer-relax-then-enforce-ratio=0.95 "
        "--lagrangian "
+       "--reduced-costs-rho "
        "--max-iterations=100 --default-rho=1e-6 "
        "--reduced-costs --rc-fixer --xhatshuffle "
        "--linearize-proximal-terms "

examples/sslp/sslp_cylinders.py

@@ -102,6 +102,7 @@ def main():
     if cfg.reduced_costs_rho:
         vanilla.add_reduced_costs_rho(hub_dict, cfg)
 
+
     if cfg.use_primal_dual_rho_updater:
         vanilla.extension_adder(hub_dict, PrimalDualRho)
         hub_dict['opt_kwargs']['options']['primal_dual_rho_options'] = {

mpisppy/cylinders/cross_scen_spoke.py

@@ -11,14 +11,15 @@
 from mpisppy import MPI
 from mpisppy.utils.lshaped_cuts import LShapedCutGenerator
 from mpisppy.cylinders.spwindow import Field
+from mpisppy.cylinders.spoke import Spoke
 
 import numpy as np
 import pyomo.environ as pyo
-import mpisppy.cylinders.spoke as spoke
 
-class CrossScenarioCutSpoke(spoke.Spoke):
-    def __init__(self, spbase_object, fullcomm, strata_comm, cylinder_comm, options=None):
-        super().__init__(spbase_object, fullcomm, strata_comm, cylinder_comm, options=options)
+class CrossScenarioCutSpoke(Spoke):
+
+    send_fields = (*Spoke.send_fields, Field.CROSS_SCENARIO_CUT)
+    receive_fields = (*Spoke.receive_fields, Field.NONANT, Field.CROSS_SCENARIO_COST)
 
     def register_send_fields(self) -> None:
 
@@ -37,18 +38,28 @@ def register_send_fields(self) -> None:
         (self.nonant_per_scen, remainder) = divmod(vbuflen, local_scen_count)
         assert(remainder == 0)
 
-        ## the _locals will also have the kill signal
         self.all_nonant_len = vbuflen
         self.all_eta_len = nscen*local_scen_count
 
-        self.all_nonants = self.register_recv_field(Field.NONANT, 0, vbuflen)
-        self.all_etas = self.register_recv_field(Field.CROSS_SCENARIO_COST, 0, nscen * nscen)
 
-        self.all_coefs = self.register_send_field(Field.CROSS_SCENARIO_CUT,
-                                                  nscen*(self.nonant_per_scen + 1 + 1))
+        self.all_coefs = self.send_buffers[Field.CROSS_SCENARIO_CUT]
 
         return
 
+    def register_receive_fields(self):
+        super().register_receive_fields()
+
+        nonant_ranks = self.opt.spcomm.fields_to_ranks[Field.NONANT]
+        cs_cost_ranks = self.opt.spcomm.fields_to_ranks[Field.CROSS_SCENARIO_COST]
+
+        assert len(nonant_ranks) == 1
+        assert len(cs_cost_ranks) == 1
+        assert nonant_ranks[0] == cs_cost_ranks[0]
+        source_rank = nonant_ranks[0]
+
+        self.all_nonants = self.register_recv_field(Field.NONANT, source_rank)
+        self.all_etas = self.register_recv_field(Field.CROSS_SCENARIO_COST, source_rank)
+
     def prep_cs_cuts(self):
         # create a map scenario -> index, this index is used for various lists containing scenario dependent info.
         self.scenario_to_index = { scen : indx for indx, scen in enumerate(self.opt.all_scenario_names) }
@@ -135,7 +146,7 @@ def make_eta_lb_cut(self):
             ## this cut  -- [ LB, -1, *0s ], i.e., -1*\eta + LB <= 0
             all_coefs[row_len*idx] = self._eta_lb_array[idx]
             all_coefs[row_len*idx+1] = -1
-        self.spoke_to_hub(all_coefs, Field.CROSS_SCENARIO_CUT)
+        self.put_send_buffer(all_coefs, Field.CROSS_SCENARIO_CUT)
 
     def make_cut(self):
 
@@ -293,7 +304,7 @@ def make_cut(self):
                 all_coefs[row_len*idx:row_len*(idx+1)] = coef_dict[k]
             elif feas_cuts:
                 all_coefs[row_len*idx:row_len*(idx+1)] = feas_cuts.pop()
-        self.spoke_to_hub(all_coefs, Field.CROSS_SCENARIO_CUT)
+        self.put_send_buffer(all_coefs, Field.CROSS_SCENARIO_CUT)
 
     def main(self):
         # call main cut generation routine
@@ -303,7 +314,6 @@ def main(self):
 
         # main loop
         while not (self.got_kill_signal()):
-            # if self._new_locals:
             if self.all_nonants.is_new() and self.all_etas.is_new():
                 self.make_cut()
             ## End if

mpisppy/cylinders/fwph_spoke.py

@@ -25,7 +25,7 @@ def sync(self):
         if not hasattr(self.opt, '_local_bound'):
             return
         # Tell the hub about the most recent bound
-        self.bound = self.opt._local_bound
+        self.send_bound(self.opt._local_bound)
 
     def finalize(self):
         # The FWPH spoke can call "finalize" before it
@@ -34,6 +34,6 @@ def finalize(self):
         # if we terminated early
         if not hasattr(self.opt, '_local_bound'):
             return
-        self.bound = self.opt._local_bound
+        self.send_bound(self.opt._local_bound)
         self.final_bound = self.opt._local_bound
         return self.final_bound