Merge branch 'main' into fwph_hub

Author: bknueven

Diff for: .github/workflows/test_pr_and_main.yml

@@ -407,7 +407,7 @@ jobs:
         run: |
           cd mpisppy/tests
           python test_gradient_rho.py
-          python test_w_writer.py
+          python test_xbar_w_reader_writer.py
 
   test-headers:
     name: header test

Diff for: examples/farmer/farmer_cylinders.py

@@ -8,6 +8,10 @@
 ###############################################################################
 # general example driver for farmer with cylinders
 
+## Uncomment the below to suppress warnings when running this file. Helpful for debugging.
+# import warnings
+# warnings.filterwarnings("ignore")
+
 import farmer
 
 # Make it all go
@@ -112,6 +116,8 @@ def main():
                                   ph_converger=ph_converger,
                                   rho_setter = rho_setter)
 
+    hub_dict['opt_kwargs']['options']['cfg'] = cfg                    
+
     if cfg.primal_dual_converger:
         hub_dict['opt_kwargs']['options']\
             ['primal_dual_converger_options'] = {

Diff for: examples/generic_cylinders.bash

@@ -4,6 +4,12 @@
 SOLVER="cplex"
 SPB=1
 
+echo "^^^ hub only with w-writer (smoke) ^^^"
+python -m mpi4py ../mpisppy/generic_cylinders.py --module-name farmer/farmer --num-scens 3 --solver-name ${SOLVER} --max-iterations 10 --max-solver-threads 4 --default-rho 1 --W-writer --W-fname w_values.csv 
+
+echo "^^^ hub only with w-reader (smoke) ^^^"
+python -m mpi4py ../mpisppy/generic_cylinders.py --module-name farmer/farmer --num-scens 3 --solver-name ${SOLVER} --max-iterations 10 --max-solver-threads 4 --default-rho 1 --W-reader --init-W-fname w_values.csv
+
 echo "^^^ Multi-stage AirCond ^^^"
 mpiexec -np 3 python -m mpi4py ../mpisppy/generic_cylinders.py --module-name mpisppy.tests.examples.aircond --branching-factors "3 3 3" --solver-name ${SOLVER} --max-iterations 10 --max-solver-threads 4 --default-rho 1 --lagrangian --xhatxbar --rel-gap 0.01 --solution-base-name aircond_nonants
 # --xhatshuffle --stag2EFsolvern

Diff for: mpi_one_sided_test.py

@@ -25,10 +25,10 @@ def main():
     rank = mpi.COMM_WORLD.Get_rank()
 
     array_size = 10
-    win = mpi.Win.Allocate(mpi.DOUBLE.size*array_size, mpi.DOUBLE.size, 
+    win = mpi.Win.Allocate(mpi.DOUBLE.size*array_size, mpi.DOUBLE.size,
                            comm=mpi.COMM_WORLD)
-    buff = np.ndarray(buffer=win.tomemory(), dtype='d', shape=(array_size,)) 
-     
+    buff = np.ndarray(buffer=win.tomemory(), dtype='d', shape=(array_size,))
+
     if (rank == 0):
         buff[:] = 3. * np.ones(array_size, dtype='d')
         time.sleep(3)

Diff for: mpisppy/cylinders/cross_scen_spoke.py

@@ -6,9 +6,11 @@
 # All rights reserved. Please see the files COPYRIGHT.md and LICENSE.md for
 # full copyright and license information.
 ###############################################################################
+
 from pyomo.repn.standard_repn import generate_standard_repn
 from mpisppy import MPI
 from mpisppy.utils.lshaped_cuts import LShapedCutGenerator
+from mpisppy.cylinders.spwindow import Field
 
 import numpy as np
 import pyomo.environ as pyo
@@ -18,7 +20,10 @@ 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)
 
-    def make_windows(self):
+    def register_send_fields(self) -> None:
+
+        super().register_send_fields()
+
         nscen = len(self.opt.all_scenario_names)
         if nscen == 0:
             raise RuntimeError(f"(rank: {self.cylinder_rank}), no local_scenarios")
@@ -29,24 +34,20 @@ def make_windows(self):
         for s in self.opt.local_scenarios.values():
             vbuflen += len(s._mpisppy_data.nonant_indices)
         local_scen_count = len(self.opt.local_scenario_names)
-        self.nonant_per_scen = int(vbuflen / local_scen_count)
+        (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._locals = np.zeros(nscen*local_scen_count + vbuflen + 1)
-        self._coefs = np.zeros(nscen*(nscen + self.nonant_per_scen) + 1 + 1)
-        self._new_locals = False
 
-        # local, remote
-        # send, receive
-        self._make_windows(nscen*(self.nonant_per_scen + 1 + 1), nscen*local_scen_count + vbuflen)
+        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)
 
-    def _got_kill_signal(self):
-        ''' returns True if a kill signal was received,
-            and refreshes the array and _locals'''
-        self._new_locals = self.spoke_from_hub(self._locals)
-        return self.remote_write_id == -1 
+        self.all_coefs = self.register_send_field(Field.CROSS_SCENARIO_CUT,
+                                                  nscen*(self.nonant_per_scen + 1 + 1))
+
+        return
 
     def prep_cs_cuts(self):
         # create a map scenario -> index, this index is used for various lists containing scenario dependent info.
@@ -62,7 +63,7 @@ def prep_cs_cuts(self):
         # add copies of the nonanticipatory variables to the root problem
         # NOTE: the LShaped code expects the nonant vars to be in a particular
         #       order and with a particular *name*.
-        #       We're also creating an index for reference against later 
+        #       We're also creating an index for reference against later
         nonant_vid_to_copy_map = dict()
         root_vars = list()
         for v in non_ants:
@@ -84,7 +85,7 @@ def prep_cs_cuts(self):
         self.opt.root.eta = pyo.Var(self.opt.all_scenario_names)
 
         self.opt.root.bender = LShapedCutGenerator()
-        self.opt.root.bender.set_input(root_vars=self.opt.root_vars, 
+        self.opt.root.bender.set_input(root_vars=self.opt.root_vars,
                                             tol=1e-4, comm=self.cylinder_comm)
         self.opt.root.bender.set_ls(self.opt)
 
@@ -127,36 +128,38 @@ def make_eta_lb_cut(self):
         ## we'll be storing a matrix as an array
         ## row_len is the length of each row
         row_len = 1+1+len(self.root_nonants)
-        all_coefs = np.zeros( self.nscen*row_len+1, dtype='d')
+        all_coefs = self.all_coefs
+        all_coefs.value_array().fill(0.0)
         for idx, k in enumerate(self.opt.all_scenario_names):
             ## cut_array -- [ constant, eta_coef, *nonant_coefs ]
             ## 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)
+        self.spoke_to_hub(all_coefs, Field.CROSS_SCENARIO_CUT)
 
     def make_cut(self):
 
         ## cache opt
         opt = self.opt
 
         ## unpack these the way they were packed:
-        all_nonants_and_etas = self._locals
+        all_nonants = self.all_nonants
         nonants = dict()
         etas = dict()
         ci = 0
         for k, s in opt.local_scenarios.items():
             for ndn, i in s._mpisppy_data.nonant_indices:
-                nonants[k, ndn, i] = all_nonants_and_etas[ci]
+                nonants[k, ndn, i] = all_nonants[ci]
                 ci += 1
 
         # get all the etas
+        all_etas = self.all_etas
+        ci = 0
         for k, s in opt.local_scenarios.items():
             for sn in opt.all_scenario_names:
-                etas[k, sn] = all_nonants_and_etas[ci]
+                etas[k, sn] = all_etas[ci]
                 ci += 1
 
-        ## self.nscen == len(opt.all_scenario_names)
         # compute local min etas
         min_eta_vals = np.fromiter(( min(etas[k,sn] for k in opt.local_scenarios) \
                                        for sn in opt.all_scenario_names ),
@@ -215,7 +218,7 @@ def make_cut(self):
 
         # if we are the winner, grab the xhat and bcast it to the other ranks
         if self.cylinder_comm.Get_rank() == global_rank[0]:
-            farthest_xhat = np.fromiter( (nonants[local_winner, nname, ix] 
+            farthest_xhat = np.fromiter( (nonants[local_winner, nname, ix]
                                             for nname, ix in root_nonants),
                                          dtype='d', count=len(root_nonants) )
         else:
@@ -283,13 +286,14 @@ def make_cut(self):
         ## we'll be storing a matrix as an array
         ## row_len is the length of each row
         row_len = 1+1+len(root_nonants)
-        all_coefs = np.zeros( self.nscen*row_len +1, dtype='d')
+        # all_coefs = np.zeros( self.nscen*row_len +1, dtype='d')
+        all_coefs = self.all_coefs
         for idx, k in enumerate(opt.all_scenario_names):
             if k in coef_dict:
                 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)
+        self.spoke_to_hub(all_coefs, Field.CROSS_SCENARIO_CUT)
 
     def main(self):
         # call main cut generation routine
@@ -299,5 +303,10 @@ def main(self):
 
         # main loop
         while not (self.got_kill_signal()):
-            if self._new_locals:
+            # if self._new_locals:
+            if self.all_nonants.is_new() and self.all_etas.is_new():
                 self.make_cut()
+            ## End if
+        ## End while
+
+        return