promote Spoke.update_receive_buffers to SPCommuicator

Author: bknueven

mpisppy/cylinders/hub.py

@@ -53,13 +53,6 @@ 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
@@ -167,37 +160,32 @@ def determine_termination(self):
         return abs_gap_satisfied or rel_gap_satisfied or max_stalled_satisfied
 
     def hub_finalize(self):
-        self.receive_outerbounds()
-        self.receive_innerbounds()
+        self.update_receive_buffers()
+        self.update_outerbounds()
+        self.update_innerbounds()
 
         if self.global_rank == 0:
             self.print_init = True
             global_toc("Statistics at termination", True)
             self.screen_trace()
 
-    def receive_innerbounds(self):
-        """ Get inner bounds from inner bound spokes
-            NOTE: Does not check if there _are_ innerbound spokes
-            (but should be harmless to call if there are none)
+    def update_innerbounds(self):
+        """ Update the inner bounds after receiving them from the spokes
         """
-        logging.debug("Hub is trying to receive from InnerBounds")
+        logging.debug("Hub is trying to update from InnerBounds")
         for idx, cls, recv_buf in self.receive_field_spcomms[Field.OBJECTIVE_INNER_BOUND]:
-            is_new = self.get_receive_buffer(recv_buf, Field.OBJECTIVE_INNER_BOUND, idx)
-            if is_new:
+            if recv_buf.is_new():
                 bound = recv_buf[0]
                 logging.debug("!! new InnerBound to opt {}".format(bound))
                 self.BestInnerBound = self.InnerBoundUpdate(bound, cls, idx)
         logging.debug("ph back from InnerBounds")
 
-    def receive_outerbounds(self):
-        """ Get outer bounds from outer bound spokes
-            NOTE: Does not check if there _are_ outerbound spokes
-            (but should be harmless to call if there are none)
+    def update_outerbounds(self):
+        """ Update the outer bounds after receiving them from the spokes
         """
-        logging.debug("Hub is trying to receive from OuterBounds")
+        logging.debug("Hub is trying to update from OuterBounds")
         for idx, cls, recv_buf in self.receive_field_spcomms[Field.OBJECTIVE_OUTER_BOUND]:
-            is_new = self.get_receive_buffer(recv_buf, Field.OBJECTIVE_OUTER_BOUND, idx)
-            if is_new:
+            if recv_buf.is_new():
                 bound = recv_buf[0]
                 logging.debug("!! new OuterBound to opt {}".format(bound))
                 self.BestOuterBound = self.OuterBoundUpdate(bound, cls, idx)
@@ -320,17 +308,19 @@ def sync(self):
         self.send_ws()
         self.send_nonants()
         self.send_boundsout()
-        self.receive_outerbounds()
-        self.receive_innerbounds()
+        self.update_receive_buffers()
+        self.update_outerbounds()
+        self.update_innerbounds()
         if self.opt.extensions is not None:
             self.opt.extobject.sync_with_spokes()
 
     def sync_with_spokes(self):
         self.sync()
 
     def sync_bounds(self):
-        self.receive_outerbounds()
-        self.receive_innerbounds()
+        self.update_receive_buffers()
+        self.update_outerbounds()
+        self.update_innerbounds()
         self.send_boundsout()
 
     def sync_extensions(self):
@@ -431,8 +421,9 @@ def sync(self, send_nonants=True):
         """
         if send_nonants:
             self.send_nonants()
-        self.receive_outerbounds()
-        self.receive_innerbounds()
+        self.update_receive_buffers()
+        self.update_innerbounds()
+        self.update_outerbounds()
         # in case LShaped ever gets extensions
         if getattr(self.opt, "extensions", None) is not None:
             self.opt.extobject.sync_with_spokes()

mpisppy/cylinders/spcommunicator.py

@@ -240,16 +240,6 @@ 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
@@ -381,3 +371,10 @@ def get_receive_buffer(self,
         else:
             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

mpisppy/cylinders/spoke.py

@@ -38,17 +38,11 @@ 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.
+        with the Hub, and to receive new data
+        from other cylinders.
         """
         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