Merge pull request #610 from DARMA-tasking/609-add-latest-changes-to-main

#609: add latest changes to master

Author: lifflander

config/challenging-toy-fewer-tasks.yaml

@@ -12,6 +12,7 @@ work_model:
   parameters:
     beta: 0.0
     gamma: 0.0
+    delta: 0.0
     upper_bounds:
       max_memory_usage: 8.0e+9
 

config/challenging-toy-hundreds-tasks.yaml

@@ -12,6 +12,7 @@ work_model:
   parameters:
     beta: 0.0
     gamma: 0.0
+    delta: 0.0
     upper_bounds:
       max_memory_usage: 8000000000.0
 
@@ -20,12 +21,12 @@ algorithm:
   name: InformAndTransfer
   phase_id: 0
   parameters:
-    n_iterations: 8
-    n_rounds: 4
+    n_iterations: 4
+    n_rounds: 3
     fanout: 4
     order_strategy: arbitrary
     transfer_strategy: Clustering
-    max_subclusters: 10
+    max_subclusters: 0
     cluster_swap_rtol: 0.05
     criterion: Tempered
     max_objects_per_transfer: 500

config/conf.yaml

@@ -11,6 +11,7 @@ work_model:
   parameters:
     beta: 0.0
     gamma: 0.0
+    delta: 0.0
 
 # Specify algorithm
 algorithm:

config/synthetic-blocks.yaml

@@ -12,6 +12,7 @@ work_model:
   parameters:
     beta: 0.0
     gamma: 0.0
+    delta: 0.1
     upper_bounds:
       max_memory_usage: 45.0
 
@@ -44,7 +45,7 @@ visualization:
   y_ranks: 2
   z_ranks: 1
   object_jitter: 0.5
-  rank_qoi: load
+  rank_qoi: homing
   object_qoi: shared_id
   save_meshes: true
   force_continuous_object_qoi: true

docs/pages/configuration.rst

@@ -76,6 +76,7 @@ Example configuration
       parameters:
         beta: 0.
         gamma: 0.
+        delta: 0.
 
     # Specify balancing algorithm
     algorithm:

docs/pages/testing.rst

@@ -70,6 +70,7 @@ Synthetic Blocks Test Configuration
     parameters:
       beta: 0.
       gamma: 0.
+      delta: 0.
 
   # Specify balancing algorithm
   algorithm:

src/lbaf/Applications/LBAF_app.py

@@ -390,6 +390,11 @@ def __print_statistics(self, phase: Phase, phase_name: str, work_model: WorkMode
             f"{phase_name} node maximum memory usage",
             self.__logger)
         if r_shared_mem_stats.get_maximum():
+            lbstats.print_function_statistics(
+                phase.get_ranks(),
+                lambda x: x.get_homing(),
+                f"{phase_name} homing cost",
+                self.__logger)
             lbstats.print_function_statistics(
                 phase.get_ranks(),
                 lambda x: x.get_homed_blocks_ratio(),

src/lbaf/Execution/lbsAlgorithmBase.py

@@ -99,6 +99,8 @@ def __init__(self, work_model: WorkModelBase, parameters: dict, logger: Logger):
         self.__statistics = {
             ("ranks", lambda x: x.get_load()): {
                 "maximum load": "maximum"},
+            ("ranks", lambda x: self._work_model.compute(x)): {
+                "maximum work": "maximum"},
             ("ranks", lambda x: self._work_model.compute(x)): {
                 "total work": "sum"}}
 

src/lbaf/Execution/lbsBruteForceAlgorithm.py

@@ -76,12 +76,13 @@ def execute(self, p_id: int, phases: list, statistics: dict):
         n_ranks = len(phase_ranks)
         affine_combination = isinstance(
             self._work_model, AffineCombinationWorkModel)
-        alpha, beta, gamma = [
+        alpha, beta, gamma, delta = [
             self._work_model.get_alpha() if affine_combination else 1.0,
             self._work_model.get_beta() if affine_combination else 0.0,
-            self._work_model.get_gamma() if affine_combination else 0.0]
+            self._work_model.get_gamma() if affine_combination else 0.0,
+            self._work_model.get_delta() if affine_combination else 0.0]
         _n_a, _w_min_max, a_min_max = compute_min_max_arrangements_work(
-            objects, alpha, beta, gamma, n_ranks,
+            objects, alpha, beta, gamma, delta, n_ranks,
             logger=self._logger)
 
         # Skip object transfers when requested

src/lbaf/Execution/lbsClusteringTransferStrategy.py

@@ -76,10 +76,10 @@ def __init__(self, criterion, parameters: dict, lgr: Logger):
         self._logger.info(
             f"Percentage of maximum load required for subclustering: {self.__subclustering_threshold}")
 
-        # Initialize fraction of local imbalance that must be resolved by subcluster
+        # Initialize fraction of local imbalance to be resolved by subcluster
         self.__subclustering_minimum_improvement = parameters.get("subclustering_minimum_improvement", 0.0)
         self._logger.info(
-            "Fraction of local imbalance that must be resolved by subcluster: "
+            "Local imbalance fraction to be resolved by subcluster: "
             f"{self.__subclustering_minimum_improvement}")
 
         # Initialize cluster swap relative threshold
@@ -201,7 +201,7 @@ def __swap_clusters(self, phase: Phase, r_src: Rank, clusters_src:dict, targets:
                         self._n_rejects += len(o_src) + len(o_try)
 
         # Return number of swaps performed from rank
-        n_rank_swaps = 0
+        return n_rank_swaps
 
     def __transfer_subclusters(self, phase: Phase, r_src: Rank, targets: set, ave_load: float, max_load: float) -> None:
         """Perform feasible subcluster transfers from given rank to possible targets."""
@@ -257,16 +257,13 @@ def execute(self, known_peers, phase: Phase, ave_load: float, max_load: float):
         rank_targets = self._get_ranks_to_traverse(phase.get_ranks(), known_peers)
 
         # Iterate over ranks
+        n_ranks = len(phase.get_ranks())
         for r_src, targets in rank_targets.items():
             # Cluster migratable objects on source rank
             clusters_src = self.__build_rank_clusters(r_src, True)
             self._logger.debug(
                 f"Constructed {len(clusters_src)} migratable clusters on source rank {r_src.get_id()}")
 
-            # Skip subclustering for this rank when it must be done later
-            if self.__separate_subclustering:
-                continue
-
             # Perform feasible cluster swaps from given rank to possible targets
             if (n_rank_swaps := self.__swap_clusters(phase, r_src, clusters_src, targets)):
                 # Report on swaps when some occurred
@@ -280,30 +277,36 @@ def execute(self, known_peers, phase: Phase, ave_load: float, max_load: float):
                     continue
 
             # Perform feasible subcluster swaps from given rank to possible targets
-            if self.__max_subclusters > 0:
-                self.__transfer_subclusters(phase, r_src, targets, ave_load, max_load)
+            if not self.__separate_subclustering:
+                if self.__max_subclusters > 0:
+                    self.__transfer_subclusters(phase, r_src, targets, ave_load, max_load)
+                else:
+                    self.__n_sub_skipped += 1
 
             # Report on new load and exit from rank
             self._logger.debug(
                 f"Rank {r_src.get_id()} load: {r_src.get_load()} after {self._n_transfers} object transfers")
 
         # Perform subclustering when it was not previously done
-        if self.__max_subclusters > 0 and self.__separate_subclustering:
-            # In non-deterministic case skip subclustering when swaps passed
-            if self.__n_swaps and not self._deterministic_transfer:
-                self.__n_sub_skipped += len(rank_targets)
+        if self.__separate_subclustering:
+            if self.__max_subclusters > 0:
+                # In non-deterministic case skip subclustering when swaps passed
+                if self.__n_swaps and not self._deterministic_transfer:
+                    self.__n_sub_skipped = n_ranks
+                else:
+                    # Iterate over ranks
+                    for r_src, targets in rank_targets.items():
+                        # Perform feasible subcluster swaps from given rank to possible targets
+                        self.__transfer_subclusters(phase, r_src, targets, ave_load, max_load)
+
+                        # Report on new load and exit from rank
+                        self._logger.debug(
+                            f"Rank {r_src.get_id()} load: {r_src.get_load()} after {self._n_transfers} object transfers")
             else:
-                # Iterate over ranks
-                for r_src, targets in rank_targets.items():
-                    # Perform feasible subcluster swaps from given rank to possible targets
-                    self.__transfer_subclusters(phase, r_src, targets, ave_load, max_load)
-
-                    # Report on new load and exit from rank
-                    self._logger.debug(
-                        f"Rank {r_src.get_id()} load: {r_src.get_load()} after {self._n_transfers} object transfers")
+                # Subclustering is skipped altogether for all ranks
+                self.__n_sub_skipped = n_ranks
 
         # Report on global transfer statistics
-        n_ranks = len(phase.get_ranks())
         self._logger.info(
             f"Swapped {self.__n_swaps} cluster pairs amongst {self.__n_swap_tries} tries "
             f"({100 * self.__n_swaps / self.__n_swap_tries:.2f}%)")