aph hub fixes (#294)

* adding some code to make it easier to avoid spurious warnings

* REVERT part of the 'trying to get rid of spoke sleep' commit, which is an allreduce in hub.py (search for the word 'revert')

* revert part of intra-hub sync for aph

* aircond cylinders has the trace files hooked up

* fix indentation error

* adding a slurm script to demonstrate traces

* code for traces was already factored up

* missed the update

* Update hub.py

aph cannot have the synchronization barriers

* Update hub.py

typo from last commit

Author: DLWoodruff

Diff for: examples/aircond/aircond_cylinders.py

@@ -223,6 +223,7 @@ def main():
     scenario_denouement = refmodule.scenario_denouement
 
     if cfg.EF_directly:
+        """
         ama_options = {"EF-mstage": not proper_bundles,
                        "EF-2stage": proper_bundles,
                        "EF_solver_name": cfg.solver_name,
@@ -231,8 +232,9 @@ def main():
                        "_mpisppy_probability": 1/ScenCount,  # is this needed?
                        "tee_ef_solves":False,
                        }
+        """
         ama = amalgamator.from_module(refmodule,
-                                      ama_options,use_command_line=False)
+                                      cfg, use_command_line=False)
         ama.run()
         print(f"EF inner bound=", ama.best_inner_bound)
         print(f"EF outer bound=", ama.best_outer_bound)
@@ -306,13 +308,17 @@ def main():
         soptions = option_string_to_dict(cfg.solver_options)
         hub_dict["opt_kwargs"]["options"]["iter0_solver_options"].update(soptions)
         hub_dict["opt_kwargs"]["options"]["iterk_solver_options"].update(soptions)
+        for sd in list_of_spoke_dict:
+            sd["opt_kwargs"]["options"]["iter0_solver_options"].update(soptions)
+            sd["opt_kwargs"]["options"]["iterk_solver_options"].update(soptions)
+
         if with_xhatspecific:
             xhatspecific_spoke["opt_kwargs"]["options"]["xhat_looper_options"]["xhat_solver_options"].update(soptions)
         if with_xhatshuffle:
             xhatshuffle_spoke["opt_kwargs"]["options"]["xhat_looper_options"]["xhat_solver_options"].update(soptions)
-        for sd in list_of_spoke_dict:
-            sd["opt_kwargs"]["options"]["iter0_solver_options"].update(soptions)
-            sd["opt_kwargs"]["options"]["iterk_solver_options"].update(soptions)
+    # special code to get a trace for xhatshuffle
+    if with_xhatshuffle and cfg.trace_prefix is not None:
+        xhatshuffle_spoke["opt_kwargs"]["options"]['shuffle_running_trace_prefix']  = cfg.trace_prefix
 
     wheel = WheelSpinner(hub_dict, list_of_spoke_dict)
     wheel.spin()

Diff for: examples/aircond/trace.slurm

@@ -0,0 +1,23 @@
+#!/bin/bash -l
+
+# Outputs a trace file (but throws an error if the file exists)
+
+#!/bin/bash -l
+#SBATCH --job-name=aircond_demo_slurm
+#SBATCH --output=demo_slurm.out
+#SBATCH --ntasks=9
+#SBATCH --cpus-per-task=2
+#SBATCH --time=0-0:30:00
+#SBATCH --nodelist=c[3]
+
+conda activate mpisppy
+SOLVERNAME="gurobi_persistent"
+export GRB_LICENSE_FILE=/home/dlwoodruff/software/gurobi950/licenses/c3/gurobi.lic
+
+# TBD: aircond uses start-seed (but seed is allowed as an arg that is ignored).
+
+# xhat output file name is hardwired
+mpiexec -np $SLURM_NTASKS python -m mpi4py aircond_cylinders.py --bundles-per-rank=0 --max-iterations=100 --default-rho=1 --solver-name=${SOLVERNAME} --branching-factors "20 5 4" --Capacity 200 --QuadShortCoeff 0.3  --BeginInventory 50 --rel-gap 0.001 --mu-dev 0 --sigma-dev 40 --max-solver-threads 2 --start-seed 0 --start-ups --lagrangian --xhatshuffle --trace-prefix ${SLURM_JOB_NAME}_
+
+# --no-lagrangian --with-lagranger --lagranger-rho-rescale-factors-json lagranger_factors.json
+# --with-lagrangian --no-lagranger

Diff for: mpisppy/cylinders/hub.py

@@ -5,6 +5,7 @@
 import logging
 import time
 import mpisppy.log
+from mpisppy.opt.aph import APH
 
 from mpisppy import MPI
 from mpisppy.cylinders.spcommunicator import SPCommunicator
@@ -347,7 +348,8 @@ def hub_to_spoke(self, values, spoke_strata_rank):
                 f"into local buffer of length {expected_length}"
             )
         # this is so the spoke ranks all get the same write_id at approximately the same time
-        self.cylinder_comm.Barrier()
+        if not isinstance(self.opt, APH):
+            self.cylinder_comm.Barrier()
         self.local_write_ids[spoke_strata_rank - 1] += 1
         values[-1] = self.local_write_ids[spoke_strata_rank - 1]
         window = self.windows[spoke_strata_rank - 1]
@@ -370,25 +372,31 @@ def hub_from_spoke(self, values, spoke_num):
             )
         # so the window in each rank gets read at approximately the same time,
         # and so has the same write_id
-        self.cylinder_comm.Barrier()
+        if not isinstance(self.opt, APH):
+            self.cylinder_comm.Barrier()
         window = self.windows[spoke_num - 1]
         window.Lock(spoke_num)
         window.Get((values, len(values), MPI.DOUBLE), spoke_num)
         window.Unlock(spoke_num)
 
-        new_id = int(values[-1])
-        local_val = np.array((new_id,), 'i')
-        sum_ids = np.zeros(1, 'i')
-        self.cylinder_comm.Allreduce((local_val, MPI.INT),
-                                     (sum_ids, MPI.INT),
-                                     op=MPI.SUM)
-
-        if new_id != sum_ids[0] / self.cylinder_comm.size:
-            return False
-
-        if (new_id > self.remote_write_ids[spoke_num - 1]) or (new_id < 0):
-            self.remote_write_ids[spoke_num - 1] = new_id
-            return True
+        if isinstance(self.opt, APH):
+            # reverting part of changes from Ben getting rid of spoke sleep DLW jan 2023
+            if values[-1] > self.remote_write_ids[spoke_num - 1]:
+                self.remote_write_ids[spoke_num - 1] = values[-1]
+                return True
+        else:
+            new_id = int(values[-1])
+            local_val = np.array((new_id,), 'i')
+            sum_ids = np.zeros(1, 'i')
+            self.cylinder_comm.Allreduce((local_val, MPI.INT),
+                                         (sum_ids, MPI.INT),
+                                         op=MPI.SUM)
+            if new_id != sum_ids[0] / self.cylinder_comm.size:
+                return False
+
+            if (new_id > self.remote_write_ids[spoke_num - 1]) or (new_id < 0):
+                self.remote_write_ids[spoke_num - 1] = new_id
+                return True
         return False
 
     def send_terminate(self):

Diff for: mpisppy/cylinders/xhatshufflelooper_bounder.py

@@ -32,23 +32,6 @@ def xhatbase_prep(self):
         self.verbose = self.opt.options["verbose"] # typing aid  
         self.solver_options = self.opt.options["xhat_looper_options"]["xhat_solver_options"]
 
-        # Start code to support running trace. TBD: factor this up?
-        if self.cylinder_rank == 0 and \
-                'suffle_running_trace_prefix' in self.opt.options and \
-                self.opt.options['shuffle_running_trace_prefix'] is not None:
-            running_trace_prefix =\
-                            self.opt.options['shuffle_running_trace_prefix']
-
-            filen = running_trace_prefix+self.__class__.__name__+'.csv'
-            if os.path.exists(filen):
-                raise RuntimeError(f"running trace file {filen} already exists!")
-            with open(filen, 'w') as f:
-                f.write("time,scen,value\n")
-            self.running_trace_filen = filen
-        else:
-            self.running_trace_filen = None
-        # end code to support running trace
-        
         if not isinstance(self.opt, Xhat_Eval):
             raise RuntimeError("XhatShuffleInnerBound must be used with Xhat_Eval.")
 
@@ -93,9 +76,6 @@ def _vb(msg):
             if self.verbose and self.opt.cylinder_rank == 0:
                 print ("(rank0) " + msg)
 
-        if self.running_trace_filen is not None:
-            with open(self.running_trace_filen, "a") as f:
-                f.write(f"{time.time()},{snamedict},{obj}\n")
         if obj is None:
             _vb(f"    Infeasible {snamedict}")
             return False