adding fields for best xhat and recent xhats

.github/workflows/test_pr_and_main.yml

@@ -86,6 +86,10 @@ jobs:
           cd examples
           python afew.py xpress_persistent
 
+      - name: Test ComponentMap
+        run: |
+          pytest mpisppy/tests/test_component_map_usage.py
+
       - name: Test docs
         run: |
           cd ./doc/src/

examples/farmer/agnostic/farmer_ampl_agnostic.py

@@ -326,7 +326,7 @@ def solve_one(Ag, s, solve_keyword_args, gripe, tee, need_solution=True):
         s._mpisppy_data.nonant_indices[ndn_i]._value = gxvar.value()
 
     # the next line ignores bundling
-    s._mpisppy_data._obj_from_agnostic = objval
+    s._mpisppy_data.inner_bound = objval
 
     # TBD: deal with other aspects of bundling (see solve_one in spopt.py)
 

examples/farmer/agnostic/farmer_gurobipy_agnostic.py

@@ -285,7 +285,7 @@ def solve_one(Ag, s, solve_keyword_args, gripe, tee, need_solution=True):
         s._mpisppy_data.nonant_indices[ndn_i]._value = grb_var.X
 
     # Store the objective function value in the host scenario
-    s._mpisppy_data._obj_from_agnostic = objval
+    s._mpisppy_data.inner_bound = objval
 
     # Additional checks and operations for bundling if needed (depending on the problem)
     # ...

examples/farmer/agnostic/farmer_pyomo_agnostic.py

@@ -241,7 +241,7 @@ def solve_one(Ag, s, solve_keyword_args, gripe, tee=False, need_solution=True):
             s._mpisppy_data.nonant_indices[ndn_i]._value = gxvar._value
 
         # the next line ignore bundling
-        s._mpisppy_data._obj_from_agnostic = pyo.value(gs.Total_Cost_Objective)
+        s._mpisppy_data.inner_bound = pyo.value(gs.Total_Cost_Objective)
 
     # TBD: deal with other aspects of bundling (see solve_one in spopt.py)
 

mpisppy/agnostic/ampl_guest.py

@@ -277,7 +277,7 @@ def solve_one(self, Ag, s, solve_keyword_args, gripe, tee=False, need_solution=T
             if gripe:
                 print (f"Solve failed for scenario {s.name} on rank {global_rank}")
                 print(f"{gs.solve_result =}")
-            s._mpisppy_data._obj_from_agnostic = None
+            s._mpisppy_data.inner_bound = None
             return
 
         else:
@@ -296,7 +296,7 @@ def solve_one(self, Ag, s, solve_keyword_args, gripe, tee=False, need_solution=T
         if gd["sense"] == pyo.minimize:
             s._mpisppy_data.outer_bound = objval - mipgap
         else:
-            s._mpisppy_data.inner_bound = objval + mipgap
+            s._mpisppy_data.outer_bound = objval + mipgap
 
         # copy the nonant x values from gs to s so mpisppy can use them in s
         # in general, we need more checks (see the pyomo agnostic guest example)
@@ -318,7 +318,7 @@ def solve_one(self, Ag, s, solve_keyword_args, gripe, tee=False, need_solution=T
 
             s._mpisppy_data.nonant_indices[ndn_i]._value = gxvar.value()
 
-        s._mpisppy_data._obj_from_agnostic = objval
+        s._mpisppy_data.inner_bound = objval
 
 
     # local helper

mpisppy/agnostic/examples/farmer_gurobipy_model.py

@@ -200,7 +200,7 @@ def solve_one(Ag, s, solve_keyword_args, gripe, tee, need_solution=True):
         s._mpisppy_data.nonant_indices[ndn_i]._value = grb_var.X
 
     # Store the objective function value in the host scenario
-    s._mpisppy_data._obj_from_agnostic = objval
+    s._mpisppy_data.inner_bound = objval
 
     # Additional checks and operations for bundling if needed (depending on the problem)
     # ...

mpisppy/agnostic/gams_guest.py

@@ -204,7 +204,7 @@ def solve_one(self, Ag, s, solve_keyword_args, gripe, tee, need_solution=True):
             if gripe:
                 print (f"Solve failed for scenario {s.name} on rank {global_rank}")
                 print(f"{gs.model_status =}")
-            s._mpisppy_data._obj_from_agnostic = None
+            s._mpisppy_data.inner_bound = None
             return
 
         if solver_exception is not None and need_solution:
@@ -248,7 +248,7 @@ def solve_one(self, Ag, s, solve_keyword_args, gripe, tee, need_solution=True):
             s._mpisppy_data.outer_bound = objval
 
         # the next line ignores bundling
-        s._mpisppy_data._obj_from_agnostic = objval
+        s._mpisppy_data.inner_bound = objval
 
         # TBD: deal with other aspects of bundling (see solve_one in spopt.py)
         #print(f"For {s.name} in {global_rank=}: {objval=}")

mpisppy/agnostic/pyomo_guest.py

@@ -262,7 +262,7 @@ def solve_one(self, Ag, s, solve_keyword_args, gripe, tee=False, need_solution=T
                 s._mpisppy_data.nonant_indices[ndn_i]._value = gxvar._value
 
             # the next line ignore bundles (other than proper bundles)
-            s._mpisppy_data._obj_from_agnostic = pyo.value(sputils.get_objs(gs)[0])
+            s._mpisppy_data.inner_bound = pyo.value(sputils.get_objs(gs)[0])
 
 
     # local helper

mpisppy/cylinders/spcommunicator.py

@@ -114,6 +114,65 @@ def _pull_id(self) -> int:
         return self._id
 
 
+class _CircularBuffer:
+    """
+    The circular buffer is meant for holding several versions of a Field
+    (defined by the `buffer_size`). The `data` object is an instance of
+    `FieldArray`.
+
+    To know where in the buffer we are, we use the FieldArray._id. The layout
+    looks like this for a `buffer_size` of 4:
+
+    |--0--|--1--|--2--|--3--|id|
+
+    The id % buffer_size tells us which data point is the most recent, such
+    that individual ids are not needed for each instance.
+    """
+
+    def __init__(self, data: FieldArray, field_length: int, buffer_size: int):
+        # last byte is the "write pointer"
+        assert len(data.value_array()) == field_length * buffer_size
+        self.data = data
+        self._field_length = field_length
+        self._buffer_size = buffer_size
+
+    def _get_value_array(self, read_write_index):
+        position = read_write_index % self._buffer_size
+        return self.data._array[(position*self._field_length):((position+1)*self._field_length)]
+
+
+class SendCircularBuffer(_CircularBuffer):
+
+    def next_value_array_reference(self):
+        # NOTE: The id gets incremented in the call
+        #       to `put_send_buffer`, which is necessarily
+        #       called *after* this method. Therefore
+        #       we start at 0 and go up, and when sent
+        #       will be the id of the next *open* position
+        return self._get_value_array(self.data.id())
+
+
+class RecvCircularBuffer(_CircularBuffer):
+    # The _read_id tells us where we last read from, and the
+    # (data.id % buffer_size) - 1
+    # has the last place written to. Therefore, we know which
+    # items in the buffer are new based on their difference.
+
+    def __init__(self, data: RecvArray, field_length: int, buffer_size: int):
+        super().__init__(data, field_length, buffer_size)
+        self._read_id = 0
+
+    def most_recent_value_arrays(self):
+        # if the writes have already "wrapped around" the buffer,
+        # we need to fast-forward the read index so we don't read
+        # the same data multiple times
+        while self.data.id() > self._read_id + self._buffer_size:
+            self._read_id += 1
+        while self._read_id < self.data.id():
+            yield self._get_value_array(self._read_id)
+            self._read_id += 1
+
+
 class SPCommunicator:
     """ Base class for communicator objects. Each communicator object should register
         as a class attribute what Field attributes it provides in its buffer

mpisppy/cylinders/spoke.py

@@ -12,14 +12,14 @@
 import os
 import math
 
-from mpisppy.cylinders.spcommunicator import SPCommunicator
+from mpisppy.cylinders.spcommunicator import SPCommunicator, SendCircularBuffer
 from mpisppy.cylinders.spwindow import Field
 
 
 class Spoke(SPCommunicator):
 
     send_fields = (*SPCommunicator.send_fields, )
-    receive_fields = (*SPCommunicator.receive_fields, Field.SHUTDOWN, )
+    receive_fields = (*SPCommunicator.receive_fields, Field.SHUTDOWN, Field.BEST_OBJECTIVE_BOUNDS, )
 
     def got_kill_signal(self):
         """ Spoke should call this method at least every iteration
@@ -50,9 +50,6 @@ class _BoundSpoke(Spoke):
     """ A base class for bound spokes
     """
 
-    send_fields = (*Spoke.send_fields, )
-    receive_fields = (*Spoke.receive_fields, Field.BEST_OBJECTIVE_BOUNDS)
-
     def __init__(self, spbase_object, fullcomm, strata_comm, cylinder_comm, options=None):
         super().__init__(spbase_object, fullcomm, strata_comm, cylinder_comm, options)
         if self.cylinder_rank == 0 and \
@@ -154,11 +151,75 @@ class InnerBoundSpoke(_BoundSpoke):
         Hub, and do not need information from the main PH OPT hub.
     """
 
-    send_fields = (*_BoundSpoke.send_fields, Field.OBJECTIVE_INNER_BOUND, )
+    send_fields = (*_BoundSpoke.send_fields, Field.OBJECTIVE_INNER_BOUND, Field.BEST_XHAT, Field.RECENT_XHATS, )
     receive_fields = (*_BoundSpoke.receive_fields, )
 
     converger_spoke_char = 'I'
 
+    def __init__(self, spbase_object, fullcomm, strata_comm, cylinder_comm, options=None):
+        super().__init__(spbase_object, fullcomm, strata_comm, cylinder_comm, options)
+        self.is_minimizing = self.opt.is_minimizing
+        self.best_inner_bound = math.inf if self.is_minimizing else -math.inf
+        self.solver_options = None # can be overwritten by derived classes
+
+    def register_send_fields(self):
+        super().register_send_fields()
+        self._recent_xhat_send_circular_buffer = SendCircularBuffer(
+            self.send_buffers[Field.RECENT_XHATS],
+            self._field_lengths[Field.BEST_XHAT],
+            self._field_lengths[Field.RECENT_XHATS] // self._field_lengths[Field.BEST_XHAT],
+        )
+
+    def update_if_improving(self, candidate_inner_bound, update_best_solution_cache=True):
+        if update_best_solution_cache:
+            update = self.opt.update_best_solution_if_improving(candidate_inner_bound)
+        else:
+            update = ( (candidate_inner_bound < self.best_inner_bound)
+                if self.is_minimizing else
+                (self.best_inner_bound < candidate_inner_bound)
+                )
+        self.send_latest_xhat()
+        if update:
+            self.best_inner_bound = candidate_inner_bound
+            # send to hub
+            self.send_bound(candidate_inner_bound)
+            self.send_best_xhat()
+            return True
+        return False
+
+    def send_best_xhat(self):
+        best_xhat_buf = self.send_buffers[Field.BEST_XHAT]
+        # NOTE: this does not work with "loose" bundles
+        ci = 0
+        for s in self.opt.local_scenarios.values():
+            solution_cache = s._mpisppy_data.best_solution_cache._dict
+            for ndn_varid in s._mpisppy_data.varid_to_nonant_index:
+                best_xhat_buf[ci] = solution_cache[ndn_varid][1]
+                ci += 1
+            best_xhat_buf[ci] = s._mpisppy_data.inner_bound
+            ci += 1
+        # print(f"{self.cylinder_rank=} sending {best_xhat_buf.value_array()=}")
+        self.put_send_buffer(best_xhat_buf, Field.BEST_XHAT)
+
+    def send_latest_xhat(self):
+        recent_xhat_buf = self._recent_xhat_send_circular_buffer.next_value_array_reference()
+        ci = 0
+        for s in self.opt.local_scenarios.values():
+            solution_cache = s._mpisppy_data.latest_nonant_solution_cache
+            len_nonants = len(s._mpisppy_data.nonant_indices)
+            recent_xhat_buf[ci:ci+len_nonants] = solution_cache[:]
+            ci += len_nonants
+            recent_xhat_buf[ci] = s._mpisppy_data.inner_bound
+            ci += 1
+        # print(f"{self.cylinder_rank=} sending {recent_xhat_buf=}")
+        self.put_send_buffer(self._recent_xhat_send_circular_buffer.data, Field.RECENT_XHATS)
+
+    def finalize(self):
+        if self.opt.load_best_solution():
+            self.final_bound = self.bound
+            return self.final_bound
+        return None
+
     def bound_type(self) -> Field:
         return Field.OBJECTIVE_INNER_BOUND
 
@@ -236,7 +297,7 @@ def update_nonants(self) -> bool:
         return self._update_nonant_len_buffer()
 
 
-class InnerBoundNonantSpoke(_BoundNonantSpoke):
+class InnerBoundNonantSpoke(_BoundNonantSpoke, InnerBoundSpoke):
     """ For Spokes that provide an inner (incumbent)
         bound through self.send_bound to the Hub,
         and receive the nonants from
@@ -246,42 +307,11 @@ class InnerBoundNonantSpoke(_BoundNonantSpoke):
         and restoring results
     """
 
-    send_fields = (*_BoundNonantSpoke.send_fields, Field.OBJECTIVE_INNER_BOUND, )
-    receive_fields = (*_BoundNonantSpoke.receive_fields, Field.NONANT)
+    send_fields = (*InnerBoundSpoke.send_fields, )
+    receive_fields = (*InnerBoundSpoke.receive_fields, Field.NONANT)
 
     converger_spoke_char = 'I'
 
-    def __init__(self, spbase_object, fullcomm, strata_comm, cylinder_comm, options=None):
-        super().__init__(spbase_object, fullcomm, strata_comm, cylinder_comm, options)
-        self.is_minimizing = self.opt.is_minimizing
-        self.best_inner_bound = math.inf if self.is_minimizing else -math.inf
-        self.solver_options = None # can be overwritten by derived classes
-
-    def update_if_improving(self, candidate_inner_bound, update_best_solution_cache=True):
-        if update_best_solution_cache:
-            update = self.opt.update_best_solution_if_improving(candidate_inner_bound)
-        else:
-            update = ( (candidate_inner_bound < self.best_inner_bound)
-                if self.is_minimizing else
-                (self.best_inner_bound < candidate_inner_bound)
-                )
-        if update:
-            self.best_inner_bound = candidate_inner_bound
-            # send to hub
-            self.send_bound(candidate_inner_bound)
-            return True
-        return False
-
-    def finalize(self):
-        if self.opt.load_best_solution():
-            self.final_bound = self.bound
-            return self.final_bound
-        return None
-
-    def bound_type(self) -> Field:
-        return Field.OBJECTIVE_INNER_BOUND
-
-
 
 class OuterBoundNonantSpoke(_BoundNonantSpoke):
     """ For Spokes that provide an outer