cleaning up windows; errata

Author: bknueven

Diff for: mpisppy/cylinders/cross_scen_spoke.py

@@ -48,6 +48,7 @@ def register_send_fields(self) -> None:
 
     def register_receive_fields(self):
         super().register_receive_fields()
+        # TODO: look up rank
         self.all_nonants = self.register_recv_field(Field.NONANT, 0)
         self.all_etas = self.register_recv_field(Field.CROSS_SCENARIO_COST, 0)
 

Diff for: mpisppy/cylinders/reduced_costs_spoke.py

@@ -15,7 +15,7 @@
 
 class ReducedCostsSpoke(LagrangianOuterBound):
 
-    send_fields = (*LagrangianOuterBound.send_fields, Field.EXPECTED_REDUCED_COST, Field.SCENARIO_REDUCED_COST ,)
+    send_fields = (*LagrangianOuterBound.send_fields, Field.EXPECTED_REDUCED_COST, Field.SCENARIO_REDUCED_COST,)
     receive_fields = (*LagrangianOuterBound.receive_fields,)
 
     converger_spoke_char = 'R'

Diff for: mpisppy/cylinders/spcommunicator.py

@@ -83,8 +83,8 @@ def __setitem__(self, key, value):
 
     def _next_write_id(self) -> int:
         """
-        Updates the internal id field to the next write id, uses that id in the field data,
-        and returns that id
+        Updates the internal id field to the next write id, sets that id in the
+        field data array, and returns that id
         """
         self._id += 1
         self._array[-1] = self._id
@@ -220,13 +220,6 @@ def _validate_recv_field(self, field: Field, origin: int, length: int):
 
     def register_recv_field(self, field: Field, origin: int, length: int = -1) -> RecvArray:
         # print(f"{self.__class__.__name__}.register_recv_field, {field=}, {origin=}")
-        # TODO: better handle this case...
-        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]=}.")
-
         key = self._make_key(field, origin)
         if length == -1:
             length = self._field_lengths[field]

Diff for: mpisppy/cylinders/spwindow.py

@@ -13,6 +13,7 @@
 import numpy.typing as nptyping
 
 import enum
+import weakref
 
 import pyomo.environ as pyo
 
@@ -110,6 +111,8 @@ def __init__(self, my_fields: dict, strata_comm: MPI.Comm, field_order=None):
 
         self.buffer_length = total_buffer_length
         self.window = MPI.Win.Allocate(window_size_bytes, MPI.DOUBLE.size, comm=strata_comm)
+        # ensure the memory allocated for the window is freed
+        self._window_finalizer = weakref.finalize(self, self.window.free)
         self.buff = np.ndarray(dtype="d", shape=(total_buffer_length,), buffer=self.window.tomemory())
         self.buff[:] = np.nan
 
@@ -121,22 +124,6 @@ def __init__(self, my_fields: dict, strata_comm: MPI.Comm, field_order=None):
 
         self.strata_buffer_layouts = strata_comm.allgather(self.buffer_layout)
 
-        self.window_constructed = True
-
-        return
-
-    def free(self):
-
-        if self.window_constructed:
-            self.window.Free()
-            self.buff = None
-            self.buffer_layout = None
-            self.buffer_length = 0
-            self.window = None
-            self.strata_buffer_layouts = None
-            self.window_constructed = False
-        ## End if
-
         return
 
     #### Functions ####

Diff for: mpisppy/extensions/reduced_costs_fixer.py

@@ -109,13 +109,15 @@ def register_receive_fields(self):
             Field.OBJECTIVE_OUTER_BOUND,
             self.reduced_costs_spoke_index,
         )
-        ## End if
 
         return
 
     def sync_with_spokes(self, pre_iter0 = False):
-        # TODO: if we calculate the new bounds in the spoke we don't need to check if the buffers
-        #       have the same ID
+        # TODO: If we calculate the new bounds in the spoke we don't need to
+        #       check if the buffers have the same ID.
+        # NOTE: If we do this, then the heuristic reduced cost fixing might fix
+        #       different variables in different subproblems. But this might be
+        #       fine.
         self.opt.spcomm.get_receive_buffer(
             self.reduced_cost_buf,
             Field.EXPECTED_REDUCED_COST,