FWPH rewrite

mpisppy/cylinders/fwph_cylinder.py

@@ -0,0 +1,69 @@
+###############################################################################
+# mpi-sppy: MPI-based Stochastic Programming in PYthon
+#
+# Copyright (c) 2024, Lawrence Livermore National Security, LLC, Alliance for
+# Sustainable Energy, LLC, The Regents of the University of California, et al.
+# All rights reserved. Please see the files COPYRIGHT.md and LICENSE.md for
+# full copyright and license information.
+###############################################################################
+from mpisppy.cylinders.spwindow import Field
+from mpisppy.cylinders.spcommunicator import SPCommunicator, RecvCircularBuffer
+
+class FWPH_Cylinder(SPCommunicator):
+
+    send_fields = ()
+    receive_fields = (Field.BEST_XHAT, Field.RECENT_XHATS, )
+
+    def register_receive_fields(self):
+        super().register_receive_fields()
+        self._recent_xhat_recv_circular_buffers = [
+            (idx, cls, RecvCircularBuffer(
+                    recv_buf,
+                    self._field_lengths[Field.BEST_XHAT],
+                    self._field_lengths[Field.RECENT_XHATS] // self._field_lengths[Field.BEST_XHAT],
+                )
+            )
+            for (idx, cls, recv_buf) in self.receive_field_spcomms[Field.RECENT_XHATS] 
+        ]
+
+    def add_cylinder_columns(self):
+        self.add_best_xhat_column()
+        self.add_recent_xhat_columns()
+        # self.add_fwph_sdm_column()
+
+    def add_best_xhat_column(self):
+        for idx, cls, recv_buf in self.receive_field_spcomms[Field.BEST_XHAT]:
+            is_new = self.get_receive_buffer(recv_buf, Field.BEST_XHAT, idx)
+            if is_new:
+                self._add_QP_columns_from_buf(recv_buf.value_array())
+
+    def add_recent_xhat_columns(self):
+        for idx, cls, recv_buf_circular in self._recent_xhat_recv_circular_buffers:
+            is_new = self.get_receive_buffer(recv_buf_circular.data, Field.RECENT_XHATS, idx)
+            if is_new:
+                for value_array in recv_buf_circular.most_recent_value_arrays():
+                    self._add_QP_columns_from_buf(value_array)
+
+    def _add_QP_columns_from_buf(self, xhat_array):
+        self.opt._save_nonants()
+        inner_bound_cache = self._cache_inner_bounds()
+        # NOTE: this does not work with "loose" bundles
+        # print(f"{self.cylinder_rank=} got {xhat_array=}")
+        ci = 0
+        for k,s in self.opt.local_scenarios.items():
+            for ndn_var in s._mpisppy_data.nonant_indices.values():
+                ndn_var._value = xhat_array[ci]
+                ci += 1
+            s._mpisppy_data.inner_bound = xhat_array[ci]
+            ci += 1
+            self.opt._add_QP_column(k, disable_W=True)
+
+        self._restore_inner_bounds(inner_bound_cache)
+        self.opt._restore_nonants(update_persistent=False)
+
+    def _cache_inner_bounds(self):
+        return {s : s._mpisppy_data.inner_bound for s in self.opt.local_scenarios.values()}
+
+    def _restore_inner_bounds(self, inner_bound_cache):
+        for s in self.opt.local_scenarios.values():
+            s._mpisppy_data.inner_bound = inner_bound_cache[s]

mpisppy/cylinders/fwph_spoke.py

@@ -6,9 +6,13 @@
 # All rights reserved. Please see the files COPYRIGHT.md and LICENSE.md for
 # full copyright and license information.
 ###############################################################################
-import mpisppy.cylinders.spoke
+from mpisppy.cylinders.spoke import OuterBoundSpoke
+from mpisppy.cylinders.fwph_cylinder import FWPH_Cylinder
 
-class FrankWolfeOuterBound(mpisppy.cylinders.spoke.OuterBoundSpoke):
+class FrankWolfeOuterBound(OuterBoundSpoke, FWPH_Cylinder):
+
+    send_fields = (*OuterBoundSpoke.send_fields, *FWPH_Cylinder.send_fields)
+    receive_fields = (*OuterBoundSpoke.receive_fields, *FWPH_Cylinder.receive_fields)
 
     converger_spoke_char = 'F'
 

mpisppy/cylinders/hub.py

@@ -16,6 +16,7 @@
 from mpisppy import global_toc
 
 from mpisppy.cylinders.spwindow import Field
+from mpisppy.cylinders.fwph_cylinder import FWPH_Cylinder
 
 # Could also pass, e.g., sys.stdout instead of a filename
 mpisppy.log.setup_logger(__name__,
@@ -255,7 +256,7 @@ def sync_nonants(self):
     def sync_Ws(self):
         self.send_ws()
 
-    def is_converged(self):
+    def is_converged(self, screen_trace=True):
         if self.opt.best_bound_obj_val is not None:
             self.BestOuterBound = self.OuterBoundUpdate(self.opt.best_bound_obj_val)
         if self.opt.best_solution_obj_val is not None:
@@ -269,7 +270,7 @@ def is_converged(self):
                 )
 
             ## you still want to output status, even without inner bounders configured
-            if self.global_rank == 0:
+            if self.global_rank == 0 and screen_trace:
                 self.screen_trace()
 
             return False
@@ -281,7 +282,7 @@ def is_converged(self):
                     "will be made on the Best Bound")
 
         ## log some output
-        if self.global_rank == 0:
+        if self.global_rank == 0 and screen_trace:
             self.screen_trace()
 
         return self.determine_termination()
@@ -440,7 +441,10 @@ def finalize(self):
         Eobj = self.opt.post_loops()
         return Eobj
 
-class FWPHHub(PHHub):
+class FWPHHub(PHHub, FWPH_Cylinder):
+
+    send_fields = (*PHHub.send_fields, *FWPH_Cylinder.send_fields)
+    receive_fields = (*PHHub.receive_fields, *FWPH_Cylinder.receive_fields)
 
     _hub_algo_best_bound_provider = True
 

mpisppy/cylinders/spoke.py

@@ -29,7 +29,7 @@ def got_kill_signal(self):
         self.get_receive_buffer(shutdown_buf, Field.SHUTDOWN, 0, synchronize=False)
         return self.allreduce_or(shutdown_buf[0] == 1.0)
 
-    def is_converged(self):
+    def is_converged(self, screen_trace=False):
         """ Alias for got_kill_signal; useful for algorithms working as both
             hub and spoke
         """