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

@@ -270,7 +270,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:
@@ -289,7 +289,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)
@@ -311,7 +311,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,46 @@ def _pull_id(self) -> int:
         return self._id
 
 
+class _CircularBuffer:
+
+    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 __init__(self, data: SendArray, field_length: int, buffer_size: int):
+        super().__init__(data, field_length, buffer_size)
+
+    def next_value_array(self):
+        return self._get_value_array(self.data.id())
+
+
+class RecvCircularBuffer(_CircularBuffer):
+
+    def __init__(self, data: RecvArray, field_length: int, buffer_size: int):
+        super().__init__(data, field_length, buffer_size)
+        self._read_id = 0
+
+    def next_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()
+        ci = 0
+        for s in self.opt.local_scenarios.values():
+            solution_cache = s._mpisppy_data.latest_solution_cache._dict
+            for ndn_varid in s._mpisppy_data.varid_to_nonant_index:
+                recent_xhat_buf[ci] = solution_cache[ndn_varid][1]
+                ci += 1
+            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

mpisppy/cylinders/spwindow.py

@@ -30,29 +30,36 @@ class Field(enum.IntEnum):
     CROSS_SCENARIO_COST=400
     NONANT_LOWER_BOUNDS=500
     NONANT_UPPER_BOUNDS=501
+    BEST_XHAT=600 # buffer having the best xhat and its total cost per scenario
+    RECENT_XHATS=601 # buffer having some recent xhats and their total cost per scenario
     WHOLE=1_000_000
 
 
-_field_length_components = pyo.ConcreteModel()
-_field_length_components.local_nonant_length = pyo.Param(mutable=True)
-_field_length_components.local_scenario_length = pyo.Param(mutable=True)
-_field_length_components.total_number_nonants = pyo.Param(mutable=True)
-_field_length_components.total_number_scenarios = pyo.Param(mutable=True)
+field_length_components = pyo.ConcreteModel()
+field_length_components._local_nonant_length = pyo.Param(mutable=True)
+field_length_components._local_scenario_length = pyo.Param(mutable=True)
+field_length_components._total_number_nonants = pyo.Param(mutable=True)
+field_length_components._total_number_scenarios = pyo.Param(mutable=True)
+
+# these could be modified by the user...
+field_length_components.total_number_recent_xhats = pyo.Param(mutable=True, initialize=10, within=pyo.NonNegativeIntegers)
 
 _field_lengths = {
         Field.SHUTDOWN : 1,
-        Field.NONANT : _field_length_components.local_nonant_length,
-        Field.DUALS : _field_length_components.local_nonant_length,
-        Field.RELAXED_NONANT : _field_length_components.local_nonant_length,
+        Field.NONANT : field_length_components._local_nonant_length,
+        Field.DUALS : field_length_components._local_nonant_length,
+        Field.RELAXED_NONANT : field_length_components._local_nonant_length,
         Field.BEST_OBJECTIVE_BOUNDS : 2,
         Field.OBJECTIVE_INNER_BOUND : 1,
         Field.OBJECTIVE_OUTER_BOUND : 1,
-        Field.EXPECTED_REDUCED_COST : _field_length_components.total_number_nonants,
-        Field.SCENARIO_REDUCED_COST : _field_length_components.local_nonant_length,
-        Field.CROSS_SCENARIO_CUT : _field_length_components.total_number_scenarios * (_field_length_components.total_number_nonants + 1 + 1),
-        Field.CROSS_SCENARIO_COST : _field_length_components.total_number_scenarios * _field_length_components.total_number_scenarios,
-        Field.NONANT_LOWER_BOUNDS : _field_length_components.total_number_nonants,
-        Field.NONANT_UPPER_BOUNDS : _field_length_components.total_number_nonants,
+        Field.EXPECTED_REDUCED_COST : field_length_components._total_number_nonants,
+        Field.SCENARIO_REDUCED_COST : field_length_components._local_nonant_length,
+        Field.CROSS_SCENARIO_CUT : field_length_components._total_number_scenarios * (field_length_components._total_number_nonants + 1 + 1),
+        Field.CROSS_SCENARIO_COST : field_length_components._total_number_scenarios * field_length_components._total_number_scenarios,
+        Field.NONANT_LOWER_BOUNDS : field_length_components._total_number_nonants,
+        Field.NONANT_UPPER_BOUNDS : field_length_components._total_number_nonants,
+        Field.BEST_XHAT : field_length_components._local_nonant_length + field_length_components._local_scenario_length,
+        Field.RECENT_XHATS : field_length_components.total_number_recent_xhats * (field_length_components._local_nonant_length + field_length_components._local_scenario_length),
 }
 
 
@@ -65,16 +72,19 @@ def __init__(self, opt):
                )
         )
 
-        _field_length_components.local_nonant_length.value = number_nonants
-        _field_length_components.local_scenario_length.value = len(opt.local_scenarios)
-        _field_length_components.total_number_nonants.value = opt.nonant_length
-        _field_length_components.total_number_scenarios.value = len(opt.local_scenarios)
+        field_length_components._local_nonant_length.value = number_nonants
+        field_length_components._local_scenario_length.value = len(opt.local_scenarios)
+        field_length_components._total_number_nonants.value = opt.nonant_length
+        field_length_components._total_number_scenarios.value = len(opt.local_scenarios)
 
         self._field_lengths = {k : pyo.value(v) for k, v in _field_lengths.items()}
 
-        # reset the _field_length_components
-        for p in _field_length_components.component_data_objects():
-            p.clear()
+        # reset the field_length_components
+        for p in field_length_components.component_data_objects():
+            # leave user-set parameter alone, just clear the
+            # "private" parameters
+            if p.name[0] == "_":
+                p.clear()
 
     def __getitem__(self, field: Field):
         return self._field_lengths[field]

mpisppy/extensions/xhatbase.py

@@ -202,6 +202,10 @@ def _try_one(self, snamedict, solver_options=None, verbose=False,
                 self.opt.local_scenarios[sname].pprint()
             # get the global obj
             obj = self.opt.Eobjective(verbose=verbose)
+            # set the scenario objective value for communication
+            for k,s in self.opt.local_scenarios.items():
+                objfct = self.opt.saved_objectives[k]
+                s._mpisppy_data.inner_bound = pyo.value(objfct)
             self.opt.update_best_solution_if_improving(obj)
             if restore_nonants:
                 self.opt._restore_nonants()