#566: eliminated "distributions" that are no longer needed

Author: ppebay

config/conf.yaml

@@ -9,8 +9,8 @@ check_schema: false
 work_model:
   name: AffineCombination
   parameters:
-    alpha: 0.0
-    beta: 1.0
+    alpha: 1.0
+    beta: 0.0
     gamma: 0.0
 
 # Specify algorithm

src/lbaf/Applications/LBAF_app.py

@@ -587,6 +587,8 @@ def run(self, cfg=None, cfg_dir=None):
         # Execute runtime for specified phases
         offline_LB_compatible = self.__parameters.json_params.get(
             "offline_LB_compatible", False)
+        lb_iterations = self.__parameters.json_params.get(
+            "lb_iterations", False)
         rebalanced_phase = runtime.execute(
             self.__parameters.algorithm.get("phase_id", 0),
             1 if offline_LB_compatible else 0)
@@ -620,7 +622,7 @@ def run(self, cfg=None, cfg_dir=None):
             else:
                 # Add new phase when load balancing when offline mode not selected
                 self.__logger.info(f"Creating rebalanced phase {phase_id}")
-                self.__jsoTn_writer.write(
+                self.__json_writer.write(
                     {phase_id: rebalanced_phase})
 
         # Generate meshes and multimedia when requested

src/lbaf/Execution/lbsAlgorithmBase.py

@@ -154,40 +154,8 @@ def factory(
             logger.error(f"Could not create an algorithm with name {algorithm_name}")
             raise SystemExit(1) from e
 
-    def _update_distributions_and_statistics(self, distributions: dict, statistics: dict):
-        """Compute and update run distributions and statistics."""
-        # Create or update distributions of object quantities of interest
-        for object_qoi_name in tuple({"load", self.__object_qoi}):
-            if not object_qoi_name:
-                continue
-            try:
-                distributions.setdefault(f"object {object_qoi_name}", []).append(
-                    {o.get_id(): getattr(o, f"get_{object_qoi_name}")()
-                    for o in self._rebalanced_phase.get_objects()})
-            except AttributeError as err:
-                self.__print_QOI("obj")
-                self._logger.error(f"Invalid object_qoi name '{object_qoi_name}'")
-                raise SystemExit(1) from err
-
-        # Create or update distributions of rank quantities of interest
-        for rank_qoi_name in tuple({"objects", "load", self.__rank_qoi}):
-            if not rank_qoi_name or rank_qoi_name == "work":
-                continue
-            try:
-                distributions.setdefault(f"rank {rank_qoi_name}", []).append(
-                    [getattr(p, f"get_{rank_qoi_name}")()
-                    for p in self._rebalanced_phase.get_ranks()])
-            except AttributeError as err:
-                self.__print_QOI("rank")
-                self._logger.error(f"Invalid rank_qoi name '{rank_qoi_name}'")
-                raise SystemExit(1) from err
-        distributions.setdefault("rank work", []).append(
-            [self._work_model.compute(p) for p in self._rebalanced_phase.get_ranks()])
-
-        # Create or update distributions of edge quantities of interest
-        distributions.setdefault("sent", []).append(dict(
-            self._rebalanced_phase.get_edge_maxima().items()))
-
+    def _update_statistics(self, statistics: dict):
+        """Compute and update run statistics."""
         # Create or update statistics dictionary entries
         for (support, getter), stat_names in self.__statistics.items():
             for k, v in stat_names.items():
@@ -253,7 +221,7 @@ def _report_final_mapping(self, logger):
                             logger.debug(
                                 f"object {k.get_id()} on rank {k.get_rank_id()}: {v}")
 
-    def _initialize(self, p_id, phases, distributions, statistics):
+    def _initialize(self, p_id, phases, statistics):
         """Factor out pre-execution checks and initalizations."""
         # Ensure that a list with at least one phase was provided
         if not isinstance(phases, dict) or not all(
@@ -286,16 +254,15 @@ def _initialize(self, p_id, phases, distributions, statistics):
             f"across {self._rebalanced_phase.get_number_of_ranks()} ranks "
             f"into phase {self._rebalanced_phase.get_id()}")
 
-        # Initialize run distributions and statistics
-        self._update_distributions_and_statistics(distributions, statistics)
+        # Initialize run statistics
+        self._update_statistics(statistics)
 
     @abc.abstractmethod
-    def execute(self, p_id, phases, distributions, statistics, a_min_max):
+    def execute(self, p_id, phases, statistics, a_min_max):
         """Execute balancing algorithm on Phase instance.
 
         :param: p_id: index of phase to be rebalanced (all if equal to _)
         :param: phases: list of Phase instances
-        :param: distributions: dictionary of load-varying variables
         :param: statistics: dictionary of  statistics
         :param: a_min_max: possibly empty list of optimal arrangements.
         """

src/lbaf/Execution/lbsBruteForceAlgorithm.py

@@ -67,10 +67,10 @@ def __init__(self, work_model, parameters: dict, lgr: Logger, rank_qoi: str, obj
         self._logger.info(
             f"Instantiated {'with' if self.__skip_transfer else 'without'} transfer stage skipping")
 
-    def execute(self, p_id: int, phases: list, distributions: dict, statistics: dict, _):
+    def execute(self, p_id: int, phases: list, statistics: dict, _):
         """Execute brute force optimization algorithm on phase with index p_id."""
         # Perform pre-execution checks and initializations
-        self._initialize(p_id, phases, distributions, statistics)
+        self._initialize(p_id, phases, statistics)
         self._logger.info("Starting brute force optimization")
         initial_phase = phases[min(phases.keys())]
         phase_ranks = initial_phase.get_ranks()
@@ -113,8 +113,8 @@ def execute(self, p_id: int, phases: list, distributions: dict, statistics: dict
         # Report on object transfers
         self._logger.info(f"{n_transfers} transfers occurred")
 
-        # Update run distributions and statistics
-        self._update_distributions_and_statistics(distributions, statistics)
+        # Update run statistics
+        self._update_statistics(statistics)
 
         # Report final mapping in debug mode
         self._report_final_mapping(self._logger)

src/lbaf/Execution/lbsCentralizedPrefixOptimizerAlgorithm.py

@@ -63,18 +63,18 @@ def __init__(self, work_model, parameters: dict, lgr: Logger, qoi_name: str, obj
         self._phase = None
         self._max_shared_ids = None
 
-    def execute(self, p_id: int, phases: list, distributions: dict, statistics: dict, _):
+    def execute(self, p_id: int, phases: list, statistics: dict, _):
         """ Execute centralized prefix memory-constrained optimizer"""
 
         p_id = 0
 
         # Ensure that a list with at least one phase was provided
-        self._initialize(p_id, phases, distributions, statistics)
+        self._initialize(p_id, phases, statistics)
 
         self._phase = self._rebalanced_phase
 
-        # Initialize run distributions and statistics
-        self._update_distributions_and_statistics(distributions, statistics)
+        # Initialize run statistics
+        self._update_statistics(statistics)
 
         # Prepare input data for rank order enumerator
         self._logger.info("Starting optimizer")
@@ -192,8 +192,8 @@ def execute(self, p_id: int, phases: list, distributions: dict, statistics: dict
                     f"iteration {i + 1} rank work",
                     self._logger)
 
-                # Update run distributions and statistics
-                self._update_distributions_and_statistics(distributions, statistics)
+                # Update run statistics
+                self._update_statistics(statistics)
 
         # Report final mapping in debug mode
         self._report_final_mapping(self._logger)

src/lbaf/Execution/lbsInformAndTransferAlgorithm.py

@@ -218,10 +218,10 @@ def __execute_information_stage(self):
         self._logger.info(
             f"Average number of peers known to ranks: {n_k} ({100 * n_k / n_r:.2f}% of {n_r})")
 
-    def execute(self, p_id: int, phases: list, distributions: dict, statistics: dict, a_min_max):
+    def execute(self, p_id: int, phases: list, statistics: dict, a_min_max):
         """ Execute 2-phase information+transfer algorithm on Phase with index p_id."""
         # Perform pre-execution checks and initializations
-        self._initialize(p_id, phases, distributions, statistics)
+        self._initialize(p_id, phases, statistics)
         print_function_statistics(
             self._rebalanced_phase.get_ranks(),
             self._work_model.compute,
@@ -244,8 +244,6 @@ def execute(self, p_id: int, phases: list, distributions: dict, statistics: dict
             # Start with information stage
             self.__execute_information_stage()
 
-            print(f"statistics: {statistics}")
-
             # Execute transfer stage
             n_ignored, n_transfers, n_rejects = self.__transfer_strategy.execute(
                 self.__known_peers, self._rebalanced_phase, statistics["average load"], statistics["maximum load"][-1])
@@ -267,8 +265,8 @@ def execute(self, p_id: int, phases: list, distributions: dict, statistics: dict
                 f"iteration {i + 1} rank work",
                 self._logger)
 
-            # Update run distributions and statistics
-            self._update_distributions_and_statistics(distributions, statistics)
+            # Update run statistics
+            self._update_statistics(statistics)
 
             # Compute current arrangement
             arrangement = tuple(sorted(

src/lbaf/Execution/lbsPhaseStepperAlgorithm.py

@@ -60,7 +60,7 @@ def __init__(self, work_model, parameters: dict, lgr: Logger, rank_qoi: str, obj
         # Call superclass init
         super().__init__(work_model, parameters, lgr, rank_qoi, object_qoi)
 
-    def execute(self, _, phases: list, distributions: dict, statistics: dict, __):
+    def execute(self, _, phases: list, statistics: dict, __):
         """Steps through all phases."""
 
         # Ensure that a list with at least one phase was provided
@@ -81,9 +81,8 @@ def execute(self, _, phases: list, distributions: dict, statistics: dict, __):
                 f"phase {p_id} rank works",
                 self._logger)
 
-            # Update run distributions and statistics
-            self._update_distributions_and_statistics(
-                distributions, statistics)
+            # Update run statistics
+            self._update_statistics(statistics)
 
             # Report current mapping in debug mode
             self._report_final_mapping(self._logger)

src/lbaf/Execution/lbsRuntime.py

@@ -50,8 +50,15 @@
 class Runtime:
     """A class to handle the execution of the LBS."""
 
-    def __init__(self, phases: dict, work_model: dict, algorithm: dict, arrangements: list, logger: Logger,
-                rank_qoi: str, object_qoi: str):
+    def __init__(
+            self,
+            phases: dict,
+            work_model: dict,
+            algorithm: dict,
+            arrangements: list,
+            logger: Logger,
+            rank_qoi: str,
+            object_qoi: str):
         """Class constructor.
 
         :param phases: dictionary of Phase instances
@@ -96,9 +103,8 @@ def __init__(self, phases: dict, work_model: dict, algorithm: dict, arrangements
                 f"Could not instantiate an algorithm of type {self.__algorithm}")
             raise SystemExit(1)
 
-        # Initialize run distributions and statistics
+        # Initialize run statistics
         phase_0 = self.__phases[min(self.__phases.keys())]
-        self.__distributions = {}
         l_stats = compute_function_statistics(
             phase_0.get_ranks(),
             lambda x: x.get_load())
@@ -122,10 +128,6 @@ def get_work_model(self):
         """Return runtime work model."""
         return self.__work_model
 
-    def get_distributions(self):
-        """Return runtime distributions."""
-        return self.__distributions
-
     def get_statistics(self):
         """Return runtime statistics."""
         return self.__statistics
@@ -139,19 +141,18 @@ def execute(self, p_id: int, phase_increment: int=0):
         self.__algorithm.execute(
             p_id,
             self.__phases,
-            self.__distributions,
             self.__statistics,
             self.__a_min_max)
 
         # Retrieve possibly null rebalanced phase and return it
-        if (pp := self.__algorithm.get_rebalanced_phase()):
+        if (lbp := self.__algorithm.get_rebalanced_phase()):
             # Increment rebalanced phase ID as requested
-            pp.set_id((pp_id := pp.get_id() + phase_increment))
+            lbp.set_id((lbp_id := lbp.get_id() + phase_increment))
 
             # Share communications from original phase with new phase
             initial_communications = self.__algorithm.get_initial_communications()
-            pp.set_communications(initial_communications[p_id])
-            self.__logger.info(f"Created rebalanced phase {pp_id}")
+            lbp.set_communications(initial_communications[p_id])
+            self.__logger.info(f"Created rebalanced phase {lbp_id}")
 
         # Return rebalanced phase
-        return pp
+        return lbp