Revert "promote Spoke.update_receive_buffers to SPCommuicator"

This reverts commit 0e680a9.

Author: bknueven

mpisppy/cylinders/hub.py

@@ -46,6 +46,13 @@ def __init__(self, spbase_object, fullcomm, strata_comm, cylinder_comm, communic
     def setup_hub(self):
         pass
 
+    @abc.abstractmethod
+    def sync(self):
+        """ To be called within the whichever optimization algorithm
+            is being run on the hub (e.g. PH)
+        """
+        pass
+
     @abc.abstractmethod
     def is_converged(self):
         """ The hub has the ability to halt the optimization algorithm on the
@@ -153,9 +160,8 @@ def determine_termination(self):
         return abs_gap_satisfied or rel_gap_satisfied or max_stalled_satisfied
 
     def hub_finalize(self):
-        self.update_receive_buffers()
-        self.update_outerbounds()
-        self.update_innerbounds()
+        self.receive_outerbounds()
+        self.receive_innerbounds()
 
         if self.global_rank == 0:
             self.print_init = True
@@ -241,9 +247,8 @@ def sync(self):
         self.send_ws()
         self.send_nonants()
         self.send_boundsout()
-        self.update_receive_buffers()
-        self.update_outerbounds()
-        self.update_innerbounds()
+        self.receive_outerbounds()
+        self.receive_innerbounds()
         self.update_nonant_bounds()
         if self.opt.extensions is not None:
             self.opt.extobject.sync_with_spokes()
@@ -252,9 +257,8 @@ def sync_with_spokes(self):
         self.sync()
 
     def sync_bounds(self):
-        self.update_receive_buffers()
-        self.update_outerbounds()
-        self.update_innerbounds()
+        self.receive_outerbounds()
+        self.receive_innerbounds()
         self.update_nonant_bounds()
         self.send_boundsout()
 
@@ -356,9 +360,8 @@ def sync(self, send_nonants=True):
         """
         if send_nonants:
             self.send_nonants()
-        self.update_receive_buffers()
-        self.update_innerbounds()
-        self.update_outerbounds()
+        self.receive_outerbounds()
+        self.receive_innerbounds()
         self.update_nonant_bounds()
         # in case LShaped ever gets extensions
         if getattr(self.opt, "extensions", None) is not None:

mpisppy/cylinders/spcommunicator.py

@@ -252,6 +252,16 @@ def main(self):
         """
         pass
 
+    def sync(self):
+        """ Every hub/spoke may have a sync function
+        """
+        pass
+
+    def is_converged(self):
+        """ Every hub/spoke may have a is_converged function
+        """
+        return False
+
     def finalize(self):
         """ Every hub/spoke may have a finalize function,
             which does some final calculations/flushing to
@@ -384,13 +394,6 @@ def get_receive_buffer(self,
             buf._is_new = False
             return False
 
-    def update_receive_buffers(self):
-        for (key, recv_buf) in self.receive_buffers.items():
-            field, rank = self._split_key(key)
-            self.get_receive_buffer(recv_buf, field, rank)
-        ## End for
-        return
-
     def update_nonant_bounds(self):
         """ update the bounds on the nonanticipative variables based on
         Field.NONANT_LOWER_BOUNDS and Field.NONANT_UPPER_BOUNDS. The lower and
@@ -426,23 +429,25 @@ def update_nonant_bounds(self):
         if bounds_modified > 0:
             global_toc(f"{self.__class__.__name__}: tightened {int(bounds_modified)} variable bounds", self.cylinder_rank == 0)
 
-    def update_innerbounds(self):
-        """ Update the inner bounds after receiving them from the spokes
+    def receive_innerbounds(self):
+        """ Get inner bounds from inner bound providers
         """
-        logger.debug(f"{self.__class__.__name__} is trying to update from InnerBounds")
+        logger.debug(f"{self.__class__.__name__} is trying to receive from InnerBounds")
         for idx, cls, recv_buf in self.receive_field_spcomms[Field.OBJECTIVE_INNER_BOUND]:
-            if recv_buf.is_new():
+            is_new = self.get_receive_buffer(recv_buf, Field.OBJECTIVE_INNER_BOUND, idx)
+            if is_new:
                 bound = recv_buf[0]
                 logger.debug("!! new InnerBound to opt {}".format(bound))
                 self.BestInnerBound = self.InnerBoundUpdate(bound, cls, idx)
         logger.debug(f"{self.__class__.__name__} back from InnerBounds")
 
-    def update_outerbounds(self):
-        """ Update the outer bounds after receiving them from the spokes
+    def receive_outerbounds(self):
+        """ Get outer bounds from outer bound providers
         """
-        logger.debug(f"{self.__class__.__name__} is trying to update from OuterBounds")
+        logger.debug(f"{self.__class__.__name__} is trying to receive from OuterBounds")
         for idx, cls, recv_buf in self.receive_field_spcomms[Field.OBJECTIVE_OUTER_BOUND]:
-            if recv_buf.is_new():
+            is_new = self.get_receive_buffer(recv_buf, Field.OBJECTIVE_OUTER_BOUND, idx) 
+            if is_new:
                 bound = recv_buf[0]
                 logger.debug("!! new OuterBound to opt {}".format(bound))
                 self.BestOuterBound = self.OuterBoundUpdate(bound, cls, idx)
@@ -485,4 +490,3 @@ def initialize_bound_values(self):
             self.BestOuterBound = inf
             self._inner_bound_update = lambda new, old : (new > old)
             self._outer_bound_update = lambda new, old : (new < old)
-

mpisppy/cylinders/spoke.py

@@ -35,11 +35,17 @@ def main(self):
         The main call for the Spoke. Derived classe
         should call the got_kill_signal method
         regularly to ensure all ranks terminate
-        with the Hub, and to receive new data
-        from other cylinders.
+        with the Hub.
         """
         pass
 
+    def update_receive_buffers(self):
+        for (key, recv_buf) in self.receive_buffers.items():
+            field, rank = self._split_key(key)
+            self.get_receive_buffer(recv_buf, field, rank)
+        ## End for
+        return
+
 
 class _BoundSpoke(Spoke):
     """ A base class for bound spokes