1 /**************************************************************************
2 ** Greg Koenig (koenig@uiuc.edu)
5 ** This is GridCommRefineLB.C
7 ** GridCommRefineLB is a load balancer for the Charm++ load balancing
8 ** framework. It is designed to work in a Grid computing environment
9 ** consisting of two or more clusters separated by wide-area communication
10 ** links. Communication between objects within a cluster is assumed to be
11 ** light weight (measured in microseconds) while communication between
12 ** objects on different clusters is assumed to be heavy weight (measured in
15 ** The load balancer examines all communications in a computation and
16 ** attempts to spread the objects that communicate with objects on remote
17 ** clusters evenly across the PEs in the local cluster. No objects are
18 ** ever migrated across cluster boundaries, they are simply distributed
19 ** as evenly as possible across the PEs in the cluster in which they were
20 ** originally placed. The idea is that by spreading objects that
21 ** communicate over the wide-area evenly, a relatively small number of
22 ** WAN objects will be mixed with a relatively large number of LAN
23 ** objects, allowing the message-driven characteristics of Charm++ the
24 ** greatest possibility of overlapping the high-cost WAN communication
25 ** with locally-driven work.
27 ** The load balancer secondarily balances on scaled processor load
28 ** (i.e., a processor that is 2x the speed of another processor in
29 ** the local cluster will get 2x the work) as well as the number of
32 ** This load balancer applies a "refinement" approach which attempts to
33 ** avoid disrupting the object-to-PE mapping by causing large numbers of
34 ** objects to migrate with each load balancing iĀ®nvocation. This may be
35 ** undesirable in some cases. (For example, if the vmi-linux "eager
36 ** protocol" is used, eager channels may be pinned between two PEs, and
37 ** migrating objects that communicate heavily with each other onto other
38 ** PEs could actually slow the computationif they no longer communicate
39 ** with each other over an eager channel.) To prevent this, the balancer
40 ** determines the average number of objects per PE that communicate with
41 ** objects on remote clusters, and then migrates objects away from PEs
42 ** that exceed this average plus some tolerance (e.g., 110% of average).
43 ** This means that only the objects on the most overloaded PEs will be
47 #include "GridCommRefineLB.decl.h"
49 #include "GridCommRefineLB.h"
52 CreateLBFunc_Def (GridCommRefineLB, "Grid communication load balancer (refines object mapping within each cluster)")
56 /**************************************************************************
59 GridCommRefineLB::GridCommRefineLB (const CkLBOptions &opt) : CBase_GridCommRefineLB (opt)
64 lbname = (char *) "GridCommRefineLB";
67 CkPrintf ("[%d] GridCommRefineLB created.\n", CkMyPe());
70 if (value = getenv ("CK_LDB_GRIDCOMMREFINELB_TOLERANCE")) {
71 CK_LDB_GridCommRefineLB_Tolerance = atof (value);
73 CK_LDB_GridCommRefineLB_Tolerance = CK_LDB_GRIDCOMMREFINELB_TOLERANCE;
81 /**************************************************************************
84 GridCommRefineLB::GridCommRefineLB (CkMigrateMessage *msg) : CBase_GridCommRefineLB (msg)
89 lbname = (char *) "GridCommRefineLB";
91 if (value = getenv ("CK_LDB_GRIDCOMMREFINELB_TOLERANCE")) {
92 CK_LDB_GridCommRefineLB_Tolerance = atof (value);
94 CK_LDB_GridCommRefineLB_Tolerance = CK_LDB_GRIDCOMMREFINELB_TOLERANCE;
102 /**************************************************************************
103 ** The Charm++ load balancing framework invokes this method to determine
104 ** whether load balancing can be performed at a specified time.
106 bool GridCommRefineLB::QueryBalanceNow (int step)
108 if (_lb_args.debug() > 2) {
109 CkPrintf ("[%d] GridCommRefineLB is balancing on step %d.\n", CkMyPe(), step);
117 /**************************************************************************
118 ** The vmi-linux machine layer incorporates the idea that PEs are located
119 ** within identifiable clusters. This information can be supplied by the
120 ** user or can be probed automatically by the machine layer. The exposed
121 ** API call CmiGetCluster() returns the integer cluster number for a
122 ** specified PE or -1 if the information is unknown.
124 ** For machine layers other than vmi-linux, simply return the constant 0.
125 ** GridCommRefineLB will assume a single-cluster computation and will
126 ** balance on the scaled processor load and number of LAN messages.
128 int GridCommRefineLB::Get_Cluster (int pe)
135 /**************************************************************************
136 ** Instantiate and initialize the PE_Data[] data structure.
138 ** While doing this...
139 ** - ensure that there is at least one available PE
140 ** - ensure that all PEs are mapped to a cluster
141 ** - determine the maximum cluster number (gives the number of clusters)
142 ** - determine the minimum speed PE (used to compute relative PE speeds)
144 void GridCommRefineLB::Initialize_PE_Data (CentralLB::LDStats *stats)
150 PE_Data = new PE_Data_T[Num_PEs];
153 for (i = 0; i < Num_PEs; i++) {
154 (&PE_Data[i])->available = stats->procs[i].available;
155 (&PE_Data[i])->cluster = Get_Cluster (i);
156 (&PE_Data[i])->num_objs = 0;
157 (&PE_Data[i])->num_lan_objs = 0;
158 (&PE_Data[i])->num_lan_msgs = 0;
159 (&PE_Data[i])->num_wan_objs = 0;
160 (&PE_Data[i])->num_wan_msgs = 0;
161 (&PE_Data[i])->relative_speed = 0.0;
162 (&PE_Data[i])->scaled_load = 0.0;
164 if (stats->procs[i].pe_speed < min_speed) {
165 min_speed = stats->procs[i].pe_speed;
169 // Compute the relative PE speeds.
170 // Also add background CPU time to each PE's scaled load.
171 for (i = 0; i < Num_PEs; i++) {
172 (&PE_Data[i])->relative_speed = (double) (stats->procs[i].pe_speed / min_speed);
173 (&PE_Data[i])->scaled_load += stats->procs[i].bg_walltime;
179 /**************************************************************************
182 int GridCommRefineLB::Available_PE_Count ()
184 int available_pe_count;
188 available_pe_count = 0;
189 for (i = 0; i < Num_PEs; i++) {
190 if ((&PE_Data[i])->available) {
191 available_pe_count += 1;
194 return (available_pe_count);
199 /**************************************************************************
202 int GridCommRefineLB::Compute_Number_Of_Clusters ()
209 for (i = 0; i < Num_PEs; i++) {
210 if ((&PE_Data[i])->cluster < 0) {
214 if ((&PE_Data[i])->cluster > max_cluster) {
215 max_cluster = (&PE_Data[i])->cluster;
218 return (max_cluster + 1);
223 /**************************************************************************
226 void GridCommRefineLB::Initialize_Object_Data (CentralLB::LDStats *stats)
231 Object_Data = new Object_Data_T[Num_Objects];
233 for (i = 0; i < Num_Objects; i++) {
234 (&Object_Data[i])->migratable = (&stats->objData[i])->migratable;
235 (&Object_Data[i])->cluster = Get_Cluster (stats->from_proc[i]);
236 (&Object_Data[i])->from_pe = stats->from_proc[i];
237 (&Object_Data[i])->to_pe = stats->from_proc[i];
238 (&Object_Data[i])->num_lan_msgs = 0;
239 (&Object_Data[i])->num_wan_msgs = 0;
240 (&Object_Data[i])->load = (&stats->objData[i])->wallTime;
242 //(&PE_Data[(&Object_Data[i])->from_pe])->num_objs += 1;
243 //(&PE_Data[(&Object_Data[i])->from_pe])->scaled_load += (&Object_Data[i])->load / (&PE_Data[(&Object_Data[i])->from_pe])->relative_speed;
249 /**************************************************************************
252 void GridCommRefineLB::Examine_InterObject_Messages (CentralLB::LDStats *stats)
256 LDCommData *com_data;
263 LDObjKey *recv_objects;
267 for (i = 0; i < stats->n_comm; i++) {
268 com_data = &(stats->commData[i]);
269 if ((!com_data->from_proc()) && (com_data->recv_type() == LD_OBJ_MSG)) {
270 send_object = stats->getHash (com_data->sender);
271 recv_object = stats->getHash (com_data->receiver.get_destObj());
273 if ((send_object < 0) || (send_object > Num_Objects) || (recv_object < 0) || (recv_object > Num_Objects)) {
277 send_pe = (&Object_Data[send_object])->from_pe;
278 recv_pe = (&Object_Data[recv_object])->from_pe;
280 send_cluster = Get_Cluster (send_pe);
281 recv_cluster = Get_Cluster (recv_pe);
283 if (send_cluster == recv_cluster) {
284 (&Object_Data[send_object])->num_lan_msgs += com_data->messages;
286 (&Object_Data[send_object])->num_wan_msgs += com_data->messages;
288 } else if (com_data->receiver.get_type() == LD_OBJLIST_MSG) {
289 send_object = stats->getHash (com_data->sender);
291 if ((send_object < 0) || (send_object > Num_Objects)) {
295 send_pe = (&Object_Data[send_object])->from_pe;
296 send_cluster = Get_Cluster (send_pe);
298 recv_objects = com_data->receiver.get_destObjs (num_objects); // (num_objects is passed by reference)
300 for (j = 0; j < num_objects; j++) {
301 recv_object = stats->getHash (recv_objects[j]);
303 if ((recv_object < 0) || (recv_object > Num_Objects)) {
307 recv_pe = (&Object_Data[recv_object])->from_pe;
308 recv_cluster = Get_Cluster (recv_pe);
310 if (send_cluster == recv_cluster) {
311 (&Object_Data[send_object])->num_lan_msgs += com_data->messages;
313 (&Object_Data[send_object])->num_wan_msgs += com_data->messages;
322 /**************************************************************************
325 void GridCommRefineLB::Place_Objects_On_PEs ()
330 for (i = 0; i < Num_Objects; i++) {
331 Assign_Object_To_PE (i, (&Object_Data[i])->from_pe);
337 /**************************************************************************
340 void GridCommRefineLB::Remap_Objects_To_PEs (int cluster)
350 // Compute average number of objects per PE for this cluster.
353 for (i = 0; i < Num_PEs; i++) {
354 if (cluster == (&PE_Data[i])->cluster) {
355 num_cluster_pes += 1;
356 num_wan_msgs += (&PE_Data[i])->num_wan_msgs;
359 avg_wan_msgs = num_wan_msgs / num_cluster_pes;
361 // Move objects away from PEs that exceed the average.
362 for (i = 0; i < Num_PEs; i++) {
363 if (cluster == (&PE_Data[i])->cluster) {
364 while ((&PE_Data[i])->num_wan_msgs > (avg_wan_msgs * CK_LDB_GridCommRefineLB_Tolerance)) {
365 target_object = Find_Maximum_WAN_Object (i);
366 target_pe = Find_Minimum_WAN_PE (cluster);
368 if ((target_object == -1) || (target_pe == -1)) {
372 Remove_Object_From_PE (target_object, i);
373 Assign_Object_To_PE (target_object, target_pe);
379 // Compute average number of objects per PE for this cluster.
382 for (j = 0; j < Num_PEs; j++) {
383 if (cluster == (&PE_Data[j])->cluster) {
384 num_cluster_pes += 1;
385 num_wan_objs += (&PE_Data[j])->num_wan_objs;
388 avg_wan_objs = num_wan_objs / num_cluster_pes;
390 // Move objects away from PEs that exceed the average.
391 for (j = 0; j < Num_PEs; j++) {
392 if (cluster == (&PE_Data[j])->cluster) {
393 while ((&PE_Data[j])->num_wan_objs > (avg_wan_objs * CK_LDB_GridCommRefineLB_Tolerance)) {
394 target_object = Find_Maximum_WAN_Object (j);
395 target_pe = Find_Minimum_WAN_PE (i);
397 if ((target_object == -1) || (target_pe == -1)) {
401 Remove_Object_From_PE (target_object, j);
402 Assign_Object_To_PE (target_object, target_pe);
411 /**************************************************************************
412 ** This method locates the maximum WAN object in terms of number of
413 ** messages that traverse a wide-area connection. The search is
414 ** constrained to objects on the specified PE.
416 ** The method returns -1 if no matching object is found.
418 int GridCommRefineLB::Find_Maximum_WAN_Object (int pe)
428 for (i = 0; i < Num_Objects; i++) {
429 if ((&Object_Data[i])->from_pe == pe) {
430 if ((&Object_Data[i])->migratable) {
431 if ((&Object_Data[i])->num_wan_msgs > max_wan_msgs) {
433 max_wan_msgs = (&Object_Data[i])->num_wan_msgs;
444 /**************************************************************************
445 ** This method locates the minimum WAN PE in terms of number of objects
446 ** that communicate with objects across a wide-area connection. The search
447 ** is constrained to PEs within the specified cluster.
449 ** In the event of a "tie" (i.e., the number of WAN objects on a candidate
450 ** PE is equal to the minimum number of WAN objects discovered so far) the
451 ** tie is broken by considering the scaled CPU loads on the PEs. The PE
452 ** with the smaller scaled load is the better candidate. In the event of
453 ** a secondary tie, the secondary tie is broken by considering the number
454 ** of LAN objects on the two PEs.
456 ** The method returns -1 if no matching PE is found.
458 int GridCommRefineLB::Find_Minimum_WAN_PE (int cluster)
466 min_wan_msgs = MAXINT;
468 for (i = 0; i < Num_PEs; i++) {
469 if (((&PE_Data[i])->available) && ((&PE_Data[i])->cluster == cluster)) {
470 if ((&PE_Data[i])->num_wan_msgs < min_wan_msgs) {
472 min_wan_msgs = (&PE_Data[i])->num_wan_msgs;
473 } else if (((&PE_Data[i])->num_wan_msgs == min_wan_msgs) &&
474 ((&PE_Data[i])->scaled_load < (&PE_Data[min_index])->scaled_load)) {
476 min_wan_msgs = (&PE_Data[i])->num_wan_msgs;
477 } else if (((&PE_Data[i])->num_wan_msgs == min_wan_msgs) &&
478 ((&PE_Data[i])->scaled_load == (&PE_Data[min_index])->scaled_load) &&
479 ((&PE_Data[i])->num_objs < (&PE_Data[min_index])->num_objs)) {
481 min_wan_msgs = (&PE_Data[i])->num_wan_msgs;
495 min_wan_objs = MAXINT;
497 for (i = 0; i < Num_PEs; i++) {
498 if (((&PE_Data[i])->available) && ((&PE_Data[i])->cluster == cluster)) {
499 if ((&PE_Data[i])->num_wan_objs < min_wan_objs) {
501 min_wan_objs = (&PE_Data[i])->num_wan_objs;
502 } else if (((&PE_Data[i])->num_wan_objs == min_wan_objs) &&
503 ((&PE_Data[i])->scaled_load < (&PE_Data[min_index])->scaled_load)) {
505 min_wan_objs = (&PE_Data[i])->num_wan_objs;
506 } else if (((&PE_Data[i])->num_wan_objs == min_wan_objs) &&
507 ((&PE_Data[i])->scaled_load == (&PE_Data[min_index])->scaled_load) &&
508 ((&PE_Data[i])->num_lan_objs < (&PE_Data[min_index])->num_lan_objs)) {
510 min_wan_objs = (&PE_Data[i])->num_wan_objs;
521 /**************************************************************************
522 ** This method removes target_object from target_pe. The data structure
523 ** entry for target_pe is updated appropriately with measurements from
526 void GridCommRefineLB::Remove_Object_From_PE (int target_object, int target_pe)
528 (&Object_Data[target_object])->to_pe = -1;
530 (&PE_Data[target_pe])->num_objs -= 1;
532 if ((&Object_Data[target_object])->num_lan_msgs > 0) {
533 (&PE_Data[target_pe])->num_lan_objs -= 1;
534 (&PE_Data[target_pe])->num_lan_msgs -= (&Object_Data[target_object])->num_lan_msgs;
537 if ((&Object_Data[target_object])->num_wan_msgs > 0) {
538 (&PE_Data[target_pe])->num_wan_objs -= 1;
539 (&PE_Data[target_pe])->num_wan_msgs -= (&Object_Data[target_object])->num_wan_msgs;
542 (&PE_Data[target_pe])->scaled_load -= (&Object_Data[target_object])->load / (&PE_Data[target_pe])->relative_speed;
547 /**************************************************************************
548 ** This method assigns target_object to target_pe. The data structure
549 ** entry for target_pe is updated appropriately with measurements from
552 void GridCommRefineLB::Assign_Object_To_PE (int target_object, int target_pe)
554 (&Object_Data[target_object])->to_pe = target_pe;
556 (&PE_Data[target_pe])->num_objs += 1;
558 if ((&Object_Data[target_object])->num_lan_msgs > 0) {
559 (&PE_Data[target_pe])->num_lan_objs += 1;
560 (&PE_Data[target_pe])->num_lan_msgs += (&Object_Data[target_object])->num_lan_msgs;
563 if ((&Object_Data[target_object])->num_wan_msgs > 0) {
564 (&PE_Data[target_pe])->num_wan_objs += 1;
565 (&PE_Data[target_pe])->num_wan_msgs += (&Object_Data[target_object])->num_wan_msgs;
568 (&PE_Data[target_pe])->scaled_load += (&Object_Data[target_object])->load / (&PE_Data[target_pe])->relative_speed;
573 /**************************************************************************
574 ** The Charm++ load balancing framework invokes this method to cause the
575 ** load balancer to migrate objects to "better" PEs.
577 void GridCommRefineLB::work (LDStats *stats)
582 // bool all_pes_mapped;
591 // LDCommData *com_data;
594 if (_lb_args.debug() > 0) {
595 CkPrintf ("[%d] GridCommRefineLB is working.\n", CkMyPe());
598 // Since this load balancer looks at communications data, it must call stats->makeCommHash().
599 stats->makeCommHash ();
601 // Initialize object variables for the number of PEs and number of objects.
602 Num_PEs = stats->nprocs();
603 Num_Objects = stats->n_objs;
605 if (_lb_args.debug() > 0) {
606 CkPrintf ("[%d] GridCommRefineLB is examining %d PEs and %d objects.\n", CkMyPe(), Num_PEs, Num_Objects);
609 // Initialize the PE_Data[] data structure.
610 Initialize_PE_Data (stats);
612 // If at least one available PE does not exist, return from load balancing.
613 if (Available_PE_Count() < 1) {
614 if (_lb_args.debug() > 0) {
615 CkPrintf ("[%d] GridCommRefineLB finds no available PEs -- no balancing done.\n", CkMyPe());
623 // Determine the number of clusters.
624 // If any PE is not mapped to a cluster, return from load balancing.
625 Num_Clusters = Compute_Number_Of_Clusters ();
626 if (Num_Clusters < 1) {
627 if (_lb_args.debug() > 0) {
628 CkPrintf ("[%d] GridCommRefineLB finds incomplete PE cluster map -- no balancing done.\n", CkMyPe());
636 if (_lb_args.debug() > 0) {
637 CkPrintf ("[%d] GridCommRefineLB finds %d clusters.\n", CkMyPe(), Num_Clusters);
640 // Initialize the Object_Data[] data structure.
641 Initialize_Object_Data (stats);
643 // Examine all object-to-object messages for intra-cluster and inter-cluster communications.
644 Examine_InterObject_Messages (stats);
646 // Place objects on the PE they are currently assigned to.
647 Place_Objects_On_PEs ();
649 // Remap objects to PEs in each cluster.
650 for (i = 0; i < Num_Clusters; i++) {
651 Remap_Objects_To_PEs (i);
654 // Make the assignment of objects to PEs in the load balancer framework.
655 for (i = 0; i < Num_Objects; i++) {
656 stats->to_proc[i] = (&Object_Data[i])->to_pe;
658 if (_lb_args.debug() > 2) {
659 CkPrintf ("[%d] GridCommRefineLB migrates object %d from PE %d to PE %d.\n", CkMyPe(), i, stats->from_proc[i], stats->to_proc[i]);
660 } else if (_lb_args.debug() > 1) {
661 if (stats->to_proc[i] != stats->from_proc[i]) {
662 CkPrintf ("[%d] GridCommRefineLB migrates object %d from PE %d to PE %d.\n", CkMyPe(), i, stats->from_proc[i], stats->to_proc[i]);
668 delete [] Object_Data;
672 #include "GridCommRefineLB.def.h"