remove hub_from_spoke

Author: bknueven

Diff for: mpisppy/cylinders/hub.py

@@ -7,13 +7,10 @@
 # full copyright and license information.
 ###############################################################################
 
-import numpy as np
 import abc
 import logging
 import mpisppy.log
-from mpisppy.opt.aph import APH
 
-from mpisppy import MPI
 from mpisppy.cylinders.spcommunicator import RecvArray, SendArray, SPCommunicator
 from math import inf
 
@@ -83,7 +80,6 @@ def current_iteration(self):
     def main(self):
         pass
 
-
     def register_extension_recv_field(self, field: Field, strata_rank: int, buf_len: int = -1) -> RecvArray:
         """
         Register an extensions interest in the given field from the given spoke. The hub
@@ -125,15 +121,14 @@ def sync_extension_fields(self):
         for key in self.extension_recv:
             ext_buf = self.receive_buffers[key]
             (field, srank) = self._split_key(key)
-            ext_buf._is_new = self.hub_from_spoke(ext_buf, srank, field)
+            ext_buf._is_new = self.get_receive_buffer(ext_buf, field, srank)
         ## End for
         return
 
     def clear_latest_chars(self):
         self.latest_ib_char = None
         self.latest_ob_char = None
 
-
     def compute_gaps(self):
         """ Compute the current absolute and relative gaps,
             using the current self.BestInnerBound and self.BestOuterBound
@@ -157,7 +152,6 @@ def compute_gaps(self):
             rel_gap = float("inf")
         return abs_gap, rel_gap
 
-
     def get_update_string(self):
         if self.latest_ib_char is None and \
                 self.latest_ob_char is None:
@@ -236,7 +230,7 @@ def receive_innerbounds(self):
         """
         logging.debug("Hub is trying to receive from InnerBounds")
         for idx, cls, recv_buf in self.receive_field_spcomms[Field.OBJECTIVE_INNER_BOUND]:
-            is_new = self.hub_from_spoke(recv_buf, idx, Field.OBJECTIVE_INNER_BOUND)
+            is_new = self.get_receive_buffer(recv_buf, Field.OBJECTIVE_INNER_BOUND, idx)
             if is_new:
                 bound = recv_buf[0]
                 logging.debug("!! new InnerBound to opt {}".format(bound))
@@ -250,7 +244,7 @@ def receive_outerbounds(self):
         """
         logging.debug("Hub is trying to receive from OuterBounds")
         for idx, cls, recv_buf in self.receive_field_spcomms[Field.OBJECTIVE_OUTER_BOUND]:
-            is_new = self.hub_from_spoke(recv_buf, idx, Field.OBJECTIVE_OUTER_BOUND)
+            is_new = self.get_receive_buffer(recv_buf, Field.OBJECTIVE_OUTER_BOUND, idx)
             if is_new:
                 bound = recv_buf[0]
                 logging.debug("!! new OuterBound to opt {}".format(bound))
@@ -264,7 +258,7 @@ def OuterBoundUpdate(self, new_bound, cls=None, idx=None, char='*'):
                 self.latest_ob_char = char
                 self.last_ob_idx = 0
             else:
-                self.latest_ib_char = cls.converger_spoke_char
+                self.latest_ob_char = cls.converger_spoke_char
                 self.last_ob_idx = idx
             return new_bound
         else:
@@ -326,7 +320,6 @@ def register_receive_fields(self):
 
         return
 
-
     def register_send_fields(self):
         super().register_send_fields()
 
@@ -336,63 +329,6 @@ def register_send_fields(self):
 
         return
 
-    def hub_from_spoke(self,
-                       buf: RecvArray,
-                       spoke_num: int,
-                       field: Field,
-                       ):
-        """ spoke_num is the rank in the strata_comm, so it is 1-based not 0-based
-
-            Returns:
-                is_new (bool): Indicates whether the "gotten" values are new,
-                    based on the write_id.
-        """
-        buf._is_new = self._hub_from_spoke(buf.array(), spoke_num, field, buf.id())
-        if buf.is_new():
-            buf._pull_id()
-        return buf.is_new()
-
-    def _hub_from_spoke(self,
-                        values: np.typing.NDArray,
-                        spoke_num: int,
-                        field: Field,
-                        last_write_id: int,
-                        ):
-        """ spoke_num is the rank in the strata_comm, so it is 1-based not 0-based
-
-            Returns:
-                is_new (bool): Indicates whether the "gotten" values are new,
-                    based on the write_id.
-        """
-        # so the window in each rank gets read at approximately the same time,
-        # and so has the same write_id
-        if not isinstance(self.opt, APH):
-            self.cylinder_comm.Barrier()
-        ## End if
-        self.window.get(values, spoke_num, field)
-
-        if isinstance(self.opt, APH):
-            # # reverting part of changes from Ben getting rid of spoke sleep DLW jan 2023
-            if values[-1] > last_write_id:
-                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
-            ## End if
-            if new_id > last_write_id or new_id < 0:
-                return True
-            ## End if
-        ## End if
-
-        return False
-
-
     def send_terminate(self):
         """ Send an array of zeros with a -1 appended to the
             end to indicate termination. This function puts to the local
@@ -614,6 +550,15 @@ def main(self):
         logger.critical("aph debug main in hub.py")
         self.opt.APH_main(spcomm=self, finalize=False)
 
+    # overwrite the default behavior of this method for APH
+    def get_receive_buffer(self,
+                           buf: RecvArray,
+                           field: Field,
+                           origin: int = -1,
+                           synchronize: bool = False,
+                          ):
+        return super().get_receive_buffer(buf, field, origin, synchronize)
+
     def finalize(self):
         """ does PH.post_loops, returns Eobj """
         # NOTE: APH_main does NOT pass in extensions

Diff for: mpisppy/cylinders/spcommunicator.py

@@ -23,6 +23,7 @@
 import abc
 import time
 
+from mpisppy import MPI
 from mpisppy.cylinders.spwindow import Field, FieldLengths, SPWindow
 
 def communicator_array(size):
@@ -309,3 +310,62 @@ def put_send_buffer(self, buf: SendArray, field: Field):
         buf._next_write_id()
         self.window.put(buf.array(), field)
         return
+
+    def get_receive_buffer(self,
+                           buf: RecvArray,
+                           field: Field,
+                           origin: int = -1,
+                           synchronize: bool = True,
+                          ):
+        """ Gets the specified values from another cylinder and copies them into
+        the specified locally-owned buffer. Updates the write_id in the locally-
+        owned buffer, if appropriate.
+
+        Args:
+            buf (RecvArray) : Buffer to put the data in
+            field (Field) : The source field
+            origin (:obj:`int`, optional) : The rank on strata_comm to get the data.
+                If not provided (or -1), will attempt to infer a unique origin. If
+                no unique origin is found, will raise an error. Default: -1.
+            synchronize (:obj:`bool`, optional) : If True, will only report
+                updated data if the write_ids are the same across the cylinder_comm
+                are identical. Default: True.
+
+        Returns:
+            is_new (bool): Indicates whether the "gotten" values are new,
+                based on the write_id.
+        """
+        if not synchronize:
+            self.cylinder_comm.Barrier()
+
+        if origin == -1:
+            origin = self.fields_to_ranks[field][0]
+            if len(self.fields_to_ranks[field]) > 1:
+                raise RuntimeError(f"Non-unique origin for {field=}. Possible "
+                                   f"origins are {self.fields_to_ranks[field]=}.")
+
+        last_id = buf.id()
+
+        self.window.get(buf.array(), origin, field)
+
+        new_id = int(buf.array()[-1])
+        if synchronize:
+            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:
+                buf._is_new = False
+                return False
+
+        else:
+            if new_id <= last_id:
+                buf._is_new = False
+                return False
+
+        # in either case, now we have new data
+        buf._is_new = True
+        buf._pull_id()
+
+        return True

Diff for: mpisppy/cylinders/spoke.py

@@ -14,7 +14,7 @@
 import math
 
 from mpisppy import MPI
-from mpisppy.cylinders.spcommunicator import RecvArray, SendArray, SPCommunicator
+from mpisppy.cylinders.spcommunicator import RecvArray, SPCommunicator
 from mpisppy.cylinders.spwindow import Field