src/ck-ldb/TempAwareCommLB.C

   1 /** \file TempAwareCommLB.C
   2  *
   3  *  Written by Osman Sarood
   4  *  This load balancing strategy is meant for temperature control. It is mandotry to
   5  *  have access to frequency control to use it. It is a mixture of Refine and Comm LBs
   6  *  in addition to normalization being done according to frequencies.
   7  *  This load balancer can also cater for sensitivity to frequency that an applicaiton
   8  *  has to frequency.
   9  */
  10
  11 /*@{*/
  12 //#define NO_TEMP_LB
  13 //#define ORG_VERSION
  14 //#define MAX_MIN
  15 #define MAX_TEMP 49
  16 //#define tolerance 0.03
  17
  18 #include "TempAwareCommLB.h"
  19 #include "ckgraphTemp.h"
  20 #include <algorithm>
  21 #include <map>
  22
  23 #include <time.h>
  24
  25 #define THRESHOLD 0.02
  26 #define SWAP_MULTIPLIER 5
  27 #ifdef TEMP_LDB
  28
  29
  30 static int cpufreq_sysfs_write (
  31                      const char *setting,int proc
  32                      )
  33 {
  34 char path[100];
  35 sprintf(path,"/sys/devices/system/cpu/cpu%d/cpufreq/scaling_setspeed",proc);
  36                 FILE *fd = fopen (path, "w");
  37
  38                 if (!fd) {
  39                         printf("PROC#%d ooooooo666 FILE OPEN ERROR file=%s\n",CkMyPe(),path);
  40                         return -1;
  41                 }
  42 //                else CkPrintf("PROC#%d opened freq file=%s\n",proc,path);
  43
  44         fseek ( fd , 0 , SEEK_SET );
  45         int numw=fprintf (fd, setting);
  46         if (numw <= 0) {
  47
  48                 fclose (fd);
  49                 printf("FILE WRITING ERROR\n");
  50                 return 0;
  51         }
  52 //        else CkPrintf("Freq for Proc#%d set to %s numw=%d\n",proc,setting,numw);
  53         fclose(fd);
  54         return 1;
  55 }
  56
  57 static int cpufreq_sysfs_read (int proc)
  58 {
  59         FILE *fd;
  60         char path[100];
  61         int i=proc;
  62         sprintf(path,"/sys/devices/system/cpu/cpu%d/cpufreq/scaling_setspeed",i);
  63
  64         fd = fopen (path, "r");
  65
  66         if (!fd) {
  67                 printf("22 FILE OPEN ERROR file=%s\n",path);
  68                 return 0;
  69         }
  70         char val[10];
  71         fgets(val,10,fd);
  72         int ff=atoi(val);
  73         fclose (fd);
  74
  75         return ff;
  76 }
  77
  78 void printCurrentTemperature(void *LB, double curWallTime)
  79 {
  80   TempAwareCommLB *taalb = static_cast<TempAwareCommLB *>(LB);
  81   int pe = CkMyPe();
  82   float temp = taalb->getTemp(pe % taalb->physicalCoresPerNode);
  83   int freq = cpufreq_sysfs_read (pe % taalb->logicalCoresPerNode);
  84   fprintf(taalb->logFD, "%f, %d, %f, %d\n", curWallTime, pe, temp, freq);
  85 }
  86
  87 #endif
  88 inline void eraseObjFromParrObjs(std::vector<int> & parr_objs, int remove_objid);
  89 inline void printMapping(std::vector<Vertex> &vertices);
  90 inline void removeFromArray(int pe_id, std::vector<int> &array);
  91 inline int popFromProcHeap(std::vector<int> & parr_above_avg, ProcArray *parr);
  92 inline void handleTransfer(int randomly_obj_id, ProcInfo& p, int possible_pe, std::vector<int> *parr_objs, ObjGraph *ogr, ProcArray* parr);
  93 inline void updateLoadInfo(int p_index, int possible_pe, double upper_threshold_temp, double lower_threshold_temp,
  94                            std::vector<int> &parr_above_avg, std::vector<int> &parr_below_avg,
  95                            std::vector<bool> &proc_load_info, ProcArray *parr);
  96 inline void getPossiblePes(std::vector<int>& possible_pes, int randomly_obj_id,
  97     ObjGraph *ogr, ProcArray* parr);
  98
  99 double upper_threshold_temp;
 100 double lower_threshold_temp;
 101
 102 extern int quietModeRequested;
 103
 104 CreateLBFunc_Def(TempAwareCommLB, "always assign the heaviest obj onto lightest loaded processor.")
 105
 106 TempAwareCommLB::TempAwareCommLB(const CkLBOptions &opt): CBase_TempAwareCommLB(opt)
 107 {
 108 #ifdef TEMP_LDB
 109   lbname = "TempAwareCommLB";
 110   if (CkMyPe()==0 && !quietModeRequested)
 111     CkPrintf("CharmLB> TempAwareCommLB created.\n");
 112
 113         starting=CmiWallTimer();
 114         migFile=fopen("migInfo","w");
 115   numAvailFreqs = 11;
 116         freqs=new int[numAvailFreqs];
 117         freqsEffect=new int[numAvailFreqs];
 118 // for tarekc cluster
 119   freqs[0] = 2395000;
 120   freqs[1] = 2394000;
 121   freqs[2] = 2261000;
 122   freqs[3] = 2128000;
 123   freqs[4] = 1995000;
 124   freqs[5] = 1862000;
 125   freqs[6] = 1729000;
 126   freqs[7] = 1596000;
 127   freqs[8] = 1463000;
 128   freqs[9] = 1330000;
 129   freqs[10] = 1197000;
 130
 131   freqsEffect[0] = 1979886;
 132   freqsEffect[1] = 1943017;
 133   freqsEffect[2] = 1910989;
 134   freqsEffect[3] = 1876619;
 135   freqsEffect[4] = 1824126;
 136   freqsEffect[5] = 1763990;
 137   freqsEffect[6] = 1666773;
 138   freqsEffect[7] = 1560224;
 139   freqsEffect[8] = 1443154;
 140   freqsEffect[9] = 1317009;
 141   freqsEffect[10] = 1200000;
 142
 143   procFreqPtr = new int[CkNumPes()];
 144
 145   for(int i=0;i<CkNumPes();i++)
 146   {
 147         char newfreq[10];
 148         sprintf(newfreq,"%d",freqs[0]);
 149   cpufreq_sysfs_write(newfreq,i%physicalCoresPerNode);
 150   procFreqPtr[i]=0;
 151   }
 152
 153   procFreq=NULL;
 154   procTemp=NULL;
 155   procFreqNew=NULL;
 156   procFreqNewEffect = NULL;
 157   avgChipTemp=NULL;
 158
 159   char logFile[100];
 160   snprintf(logFile, sizeof(logFile), "temp_freq.log.%d", CkMyPe());
 161   if ((logFD = fopen(logFile, "a"))) {
 162     fprintf(logFD, "Time, PE, Temperature, Frequency\n");
 163   } else {
 164     CkAbort("Couldn't open temperature/frequency log file");
 165   }
 166
 167
 168   CcdCallOnConditionKeep(CcdPERIODIC_1second, &printCurrentTemperature, this);
 169 #else
 170 #endif
 171 }
 172
 173 bool TempAwareCommLB::QueryBalanceNow(int _step)
 174 {
 175   //  CkPrintf("[%d] Balancing on step %d\n",CkMyPe(),_step);
 176   return true;
 177 }
 178
 179 class TempAwareCommLB::ProcLoadGreater {
 180   public:
 181     ProcLoadGreater(ProcArray *parr) : parr(parr) {
 182     }
 183     bool operator()(int lhs, int rhs) {
 184       return (parr->procs[lhs].getTotalLoad() < parr->procs[rhs].getTotalLoad());
 185     }
 186
 187   private:
 188     ProcArray *parr;
 189 };
 190
 191 class TempAwareCommLB::ObjLoadGreater {
 192   public:
 193     bool operator()(Vertex v1, Vertex v2) {
 194       return (v1.getVertexLoad() > v2.getVertexLoad());
 195     }
 196 };
 197
 198 class TempAwareCommLB::PeCommInfo {
 199   public:
 200     PeCommInfo() : num_msg(0), num_bytes(0) {
 201     }
 202
 203     PeCommInfo(int pe_id) : pe_id(pe_id), num_msg(0) , num_bytes(0) {
 204     }
 205     int pe_id;
 206     int num_msg;
 207     int num_bytes;
 208     // TODO: Should probably have a communication cost
 209 };
 210
 211 // Consists of communication information of an object with is maintained
 212 // as a list of PeCommInfo containing the processor id and the bytes transferred
 213 class TempAwareCommLB::ObjPeCommInfo {
 214   public:
 215     ObjPeCommInfo() {
 216     }
 217
 218     int obj_id;
 219     std::vector<TempAwareCommLB::PeCommInfo> pcomm;
 220 };
 221
 222 class TempAwareCommLB::ProcCommGreater {
 223   public:
 224     bool operator()(TempAwareCommLB::PeCommInfo p1, TempAwareCommLB::PeCommInfo p2) {
 225       // TODO(Harshitha): Should probably consider total communication cost
 226       return (p1.num_bytes > p2.num_bytes);
 227     }
 228 };
 229
 230 void PrintProcLoadTemp(ProcArray *parr) {
 231   int vert;
 232   double pe_load;
 233   for (vert = 0; vert < parr->procs.size(); vert++) {
 234     pe_load = parr->procs[vert].getTotalLoad();
 235     if (pe_load > upper_threshold_temp) {
 236       CkPrintf("Above load : %d load : %E overhead : %E\n",
 237         parr->procs[vert].getProcId(), parr->procs[vert].getTotalLoad(),
 238         parr->procs[vert].overhead());
 239     } else if (pe_load < lower_threshold_temp) {
 240       CkPrintf("Below load : %d load : %E overhead : %E\n",
 241         parr->procs[vert].getProcId(), parr->procs[vert].getTotalLoad(),
 242         parr->procs[vert].overhead());
 243     } else {
 244       CkPrintf("Within avg load : %d load : %E overhead : %E\n",
 245         parr->procs[vert].getProcId(), parr->procs[vert].getTotalLoad(),
 246         parr->procs[vert].overhead());
 247     }
 248   }
 249 }
 250
 251 void PrintProcObjTemp(ProcArray *parr, std::vector<int>* parr_objs) {
 252   int i, j;
 253   CkPrintf("---------------------\n");
 254   for (i = 0; i < parr->procs.size(); i++) {
 255     CkPrintf("[%d] contains ", i);
 256     for (j = 0; j < parr_objs[i].size(); j++) {
 257       CkPrintf(" %d, ", parr_objs[i][j]);
 258     }
 259     CkPrintf("\n");
 260   }
 261   CkPrintf("---------------------\n");
 262 }
 263
 264 void TempAwareCommLB::populateEffectiveFreq(int numProcs)
 265 {
 266 #ifdef TEMP_LDB
 267   for(int i=0;i<numProcs;i++)
 268   {
 269     for(int j=0;j<numAvailFreqs;j++)
 270     {
 271       if(freqs[j] == procFreqNew[i]) // same freq . copy effective freq
 272       {
 273         procFreqNewEffect[i] = freqsEffect[j];
 274 //        CkPrintf("** Proc%d j:%d NEWFreq:%d\n",i,j,procFreqNewEffect[i]);
 275       }
 276       if(freqs[j] == procFreq[i])
 277       {
 278         procFreqEffect[i] = freqsEffect[j];
 279 //        CkPrintf("-- Proc%d j:%d OLDFreq:%d procFreq:%d \n",i,j,procFreqEffect[i],procFreq[i]);
 280       }
 281     }
 282   }
 283 #endif
 284 }
 285
 286
 287 void TempAwareCommLB::initStructs(LDStats* stats)
 288 {
 289 #ifdef TEMP_LDB
 290   numProcs=stats->nprocs();
 291   numChips=numProcs/logicalCoresPerChip;
 292   avgChipTemp=new float[numChips];
 293   if(procFreq!=NULL) delete [] procFreq;
 294   if(procFreqEffect!=NULL) delete [] procFreqEffect;
 295   if(procTemp!=NULL) delete [] procTemp;
 296   if(procFreqNew!=NULL) delete [] procFreqNew;
 297   if(procFreqNewEffect!=NULL) delete [] procFreqNewEffect;
 298   if(avgChipTemp!=NULL) delete [] avgChipTemp;
 299
 300   procFreq = new int[numProcs];
 301   procFreqEffect = new int[numProcs];
 302   procTemp = new float[numProcs];
 303   procFreqNew = new int[numProcs];
 304   procFreqNewEffect = new int[numProcs];
 305   avgChipTemp = new float[numChips];
 306
 307   for(int i=0;i<numChips;i++) avgChipTemp[i]=0;
 308
 309   for(int i=0;i<numProcs;i++)
 310   {
 311         procFreq[i] = stats->procs[i].pe_speed;
 312         procTemp[i] = stats->procs[i].pe_temp;
 313         avgChipTemp[i/logicalCoresPerChip] += procTemp[i];
 314   }
 315   for(int i=0;i<numChips;i++)
 316   {
 317         avgChipTemp[i]/=logicalCoresPerChip;
 318 //CkPrintf("---- CHIP#%d has temp=%f ----------\n",i,avgChipTemp[i]);
 319   }
 320 #endif
 321 }
 322
 323 void TempAwareCommLB::tempControl()
 324 {
 325 #ifdef TEMP_LDB
 326         CkPrintf("----------- TempAwareCommLB::tempControl ---------\n");
 327   for(int i=0;i<numChips;i++)
 328   {
 329   int over=0,under=0;
 330         if(avgChipTemp[i] > MAX_TEMP)
 331         {
 332     over=1;
 333                 if(procFreqPtr[i*logicalCoresPerChip]==numAvailFreqs-1)
 334                 {
 335                         for(int j=i*logicalCoresPerChip;j<i*logicalCoresPerChip+logicalCoresPerChip;j++) procFreqNew[j] = freqs[procFreqPtr[j]];
 336                         CkPrintf("CHIP#%d RUNNING HOT EVEN WITH MIN FREQUENCY!!\n",i);
 337                 }
 338                 else
 339                 {
 340                         for(int j=i*logicalCoresPerChip;j<i*logicalCoresPerChip+logicalCoresPerChip;j++)
 341                         {
 342                                 if(procFreqPtr[j]<numAvailFreqs-1) procFreqPtr[j]++;
 343 #ifdef MAX_MIN
 344 /// PLEASE COMMENT OUT .. TESTING ONLY
 345 if(i==0) {procFreqPtr[j] = numAvailFreqs-1;/*CkPrintf("C for i:%d\n",j);*/}
 346 //if(i<numChips-1) procFreqPtr[j]=0;
 347 else  procFreqPtr[j]=0;
 348 /////////////////////////
 349 #endif
 350                                 procFreqNew[j] = freqs[procFreqPtr[j]];
 351                         }
 352 #ifndef ORG_VERSION
 353                         CkPrintf("!!!!! Chip#%d running HOT shifting from %d to %d temp=%f\n",i,procFreq[i*logicalCoresPerChip],procFreqNew[i*logicalCoresPerChip],avgChipTemp[i]);
 354 #endif
 355                 }
 356         }
 357         else
 358 //  if(avgChipTemp[i] < MAX_TEMP-1)
 359         {
 360     under=1;
 361                 if(procFreqPtr[i*logicalCoresPerChip]>0)
 362                 {
 363                         for(int j=i*logicalCoresPerChip;j<i*logicalCoresPerChip+logicalCoresPerChip;j++)
 364                         {
 365                                 if(procFreqPtr[j]>0)
 366                                         procFreqPtr[j]--;
 367 #ifdef MAX_MIN
 368 /// PLEASE COMMENT OUT .. TESTING ONLY
 369 if(i==0) procFreqPtr[j] = numAvailFreqs-1;
 370 //if(i<numChips-1) procFreqPtr[j]=0;
 371 else  procFreqPtr[j]=0;
 372 /////////////////////////
 373 #endif
 374                                 procFreqNew[j] = freqs[procFreqPtr[j]];
 375                         }
 376 #ifndef ORG_VERSION
 377                         CkPrintf("!!!!! Chip#%d running COLD shifting from %d to %d temp=%f\n",i,procFreq[i*logicalCoresPerChip],procFreqNew[i*logicalCoresPerChip],avgChipTemp[i]);
 378 #endif
 379                 }
 380                 else
 381                 {
 382                         for(int j=i*logicalCoresPerChip;j<i*logicalCoresPerChip+logicalCoresPerChip;j++) procFreqNew[j] = freqs[procFreqPtr[j]];
 383                 }
 384         }
 385 #ifdef ORG_VERSION
 386         for(int j=i*logicalCoresPerChip;j<i*logicalCoresPerChip+logicalCoresPerChip;j++) procFreqNew[j] = freqs[0];
 387 #endif
 388   }
 389         for(int x=0;x<numProcs;x++)
 390   {
 391 //CkPrintf("--------- Proc#%d %d numProcs=%d\n",x,procFreqNew[x],numProcs);
 392                 if(procFreq[x]!=procFreqNew[x]) thisProxy[x].changeFreq(procFreqNew[x]);
 393         }
 394 #endif
 395 }
 396
 397 void TempAwareCommLB::work(LDStats* stats) {
 398   /** ========================== INITIALIZATION ============================= */
 399 #ifdef TEMP_LDB
 400 //////////////////////////////////////////////////////
 401 // initialize structures for TempLBs
 402         initStructs(stats);
 403         tempControl();
 404         populateEffectiveFreq(stats->nprocs());
 405 //////////////////////////////////////////////////////
 406         CkPrintf(" ================== in TempAwareCommLB::work() ===========\n");
 407   ProcArrayTemp *parr = new ProcArrayTemp(stats,procFreq,procFreqNew);       // Processor Array
 408         parr->convertToInsts(stats);
 409   ObjGraphTemp *ogr = new ObjGraphTemp(stats,procFreq,procFreqNew);          // Object Graph
 410         ogr->convertToInsts(stats);
 411   double avgload = parr->getAverageLoad();      // Average load of processors
 412
 413   // Sets to false if it is overloaded, else to true
 414   vector<bool> proc_load_info(parr->procs.size(), false);
 415
 416   // Create an array of vectors for each processor mapping to the objects in
 417   // that processor
 418   std::vector<int>* parr_objs = new std::vector<int>[parr->procs.size()];
 419
 420   upper_threshold_temp = avgload + (avgload * THRESHOLD);
 421   //lower_threshold = avgload - (avgload * THRESHOLD * THRESHOLD);
 422   lower_threshold_temp = avgload;
 423
 424   int less_loaded_counter = 0;
 425
 426   srand(time(NULL));
 427   /** ============================= STRATEGY ================================ */
 428
 429   if (_lb_args.debug()>1)
 430     CkPrintf("[%d] In TempAwareCommLB strategy\n",CkMyPe());
 431
 432   CkPrintf("Average load %E\n", avgload);
 433
 434   int vert, i, j;
 435   int curr_pe;
 436
 437   // Iterate over all the chares and construct the peid, vector<chareid> array
 438   for(vert = 0; vert < ogr->vertices.size(); vert++) {
 439     curr_pe = ogr->vertices[vert].getCurrentPe();
 440     parr_objs[curr_pe].push_back(vert);
 441     ogr->vertices[vert].setNewPe(curr_pe);
 442   }
 443
 444   std::vector<int> parr_above_avg;
 445   std::vector<int> parr_below_avg;
 446
 447   double pe_load;
 448
 449   // Insert into parr_above_avg if the processor fits under the criteria of
 450   // overloaded processor.
 451   // Insert the processor id into parr_below_avg if the processor is underloaded
 452   for (vert = 0; vert < parr->procs.size(); vert++) {
 453     pe_load = parr->procs[vert].getTotalLoad();
 454     if (pe_load > upper_threshold_temp) {
 455       // Pushing ProcInfo into this list
 456       parr_above_avg.push_back(vert);
 457     } else if (pe_load < lower_threshold_temp) {
 458       parr_below_avg.push_back(parr->procs[vert].getProcId());
 459       proc_load_info[parr->procs[vert].getProcId()] = true;
 460       less_loaded_counter++;
 461     }
 462   }
 463
 464   std::make_heap(parr_above_avg.begin(), parr_above_avg.end(),
 465       TempAwareCommLB::ProcLoadGreater(parr));
 466
 467   int random;
 468   int randomly_obj_id;
 469   bool obj_allocated;
 470   int num_tries;
 471   // Allow as many swaps as there are chares
 472   int total_swaps = ogr->vertices.size() * SWAP_MULTIPLIER;
 473   int possible_pe;
 474   double obj_load;
 475
 476   // Keep on loadbalancing until the number of above avg processors is 0
 477   while (parr_above_avg.size() != 0 && total_swaps > 0 && parr_below_avg.size() != 0) {
 478     // CkPrintf("Above avg : %d Below avg : %d Total swaps: %d\n", parr_above_avg.size(),
 479     //    parr_below_avg.size(), total_swaps);
 480     obj_allocated = false;
 481     num_tries = 0;
 482
 483     // Pop the heaviest processor
 484     int p_index = popFromProcHeap(parr_above_avg, parr);
 485     ProcInfo& p = parr->procs[p_index];
 486
 487     while (!obj_allocated && num_tries < parr_objs[p.getProcId()].size()) {
 488
 489       // It might so happen that due to overhead load, it might not have any
 490       // more objects in its list
 491       if (parr_objs[p.getProcId()].size() == 0) {
 492         // CkPrintf("No obj left to be allocated\n");
 493         obj_allocated = true;
 494         break;
 495       }
 496
 497       int randd = rand();
 498       random = randd % parr_objs[p.getProcId()].size();
 499       randomly_obj_id = parr_objs[p.getProcId()][random];
 500 //need to update the load below .. account for freqs
 501       obj_load = ogr->vertices[randomly_obj_id].getVertexLoad();
 502
 503       // CkPrintf("Heavy %d: Parr obj size : %d random : %d random obj id : %d\n", p_index,
 504       //     parr_objs[p.getProcId()].size(), randd, randomly_obj_id);
 505       std::vector<int> possible_pes;
 506       getPossiblePes(possible_pes, randomly_obj_id, ogr, parr);
 507       for (i = 0; i < possible_pes.size(); i++) {
 508
 509         // If the heaviest communicating processor is there in the list, then
 510         // assign it to that.
 511         possible_pe = possible_pes[i];
 512
 513         if ((parr->procs[possible_pe].getTotalLoad() + obj_load) < upper_threshold_temp) {
 514          // CkPrintf("**  Transfered %d(Load %lf) from %d:%d(Load %lf) to %d:%d(Load %lf)\n",
 515          //     randomly_obj_id, obj_load, CkNodeOf(p.getProcId()), p.getProcId(), p.getTotalLoad(),
 516          //     CkNodeOf(possible_pe), possible_pe,
 517          //     parr->procs[possible_pe].getTotalLoad());
 518
 519           handleTransfer(randomly_obj_id, p, possible_pe, parr_objs, ogr, parr);
 520           obj_allocated = true;
 521           total_swaps--;
 522           updateLoadInfo(p_index, possible_pe, upper_threshold_temp, lower_threshold_temp,
 523               parr_above_avg, parr_below_avg, proc_load_info, parr);
 524
 525           break;
 526         }
 527       }
 528
 529       // Since there is no processor in the least loaded list with which this
 530       // chare communicates, pick a random least loaded processor.
 531       if (!obj_allocated) {
 532         //CkPrintf(":( Could not transfer to the nearest communicating ones\n");
 533         for (int x = 0; x < parr_below_avg.size(); x++) {
 534           int random_pe = parr_below_avg[x];
 535           if ((parr->procs[random_pe].getTotalLoad() + obj_load) < upper_threshold_temp) {
 536             obj_allocated = true;
 537             total_swaps--;
 538             handleTransfer(randomly_obj_id, p, random_pe, parr_objs, ogr, parr);
 539             updateLoadInfo(p_index, random_pe, upper_threshold_temp, lower_threshold_temp,
 540                            parr_above_avg, parr_below_avg, proc_load_info, parr);
 541             break;
 542           }
 543           num_tries++;
 544         }
 545       }
 546     }
 547
 548     if (!obj_allocated) {
 549       //CkPrintf("!!!! Could not handle the heavy proc %d so giving up\n", p_index);
 550       // parr_above_avg.push_back(p_index);
 551       // std::push_heap(parr_above_avg.begin(), parr_above_avg.end(),
 552       //     TempAwareCommLB::ProcLoadGreater(parr));
 553     }
 554   }
 555
 556   //CkPrintf("CommAwareRefine> After lb max load: %lf avg load: %lf\n", max_load, avg_load/parr->procs.size());
 557
 558   /** ============================== CLEANUP ================================ */
 559   ogr->convertDecisions(stats);         // Send decisions back to LDStats
 560   delete parr;
 561   delete ogr;
 562   delete[] parr_objs;
 563 #else
 564         CmiAbort("TempLBs are not supported without the TEMP_LDB flag\n");
 565 #endif
 566 }
 567
 568 inline void eraseObjFromParrObjs(std::vector<int> & parr_objs, int remove_objid) {
 569   for (int i = 0; i < parr_objs.size(); i++) {
 570     if (parr_objs[i] == remove_objid) {
 571       parr_objs.erase(parr_objs.begin() + i);
 572       return;
 573     }
 574   }
 575 }
 576
 577 inline void printMapping(std::vector<Vertex> &vertices) {
 578   for (int i = 0; i < vertices.size(); i++) {
 579     CkPrintf("%d: old map : %d new map : %d\n", i, vertices[i].getCurrentPe(),
 580         vertices[i].getNewPe());
 581   }
 582 }
 583
 584 inline void removeFromArray(int pe_id, std::vector<int> &array) {
 585   for (int i = 0; i < array.size(); i++) {
 586     if (array[i] == pe_id) {
 587       array.erase(array.begin() + i);
 588     }
 589   }
 590 }
 591
 592 inline int popFromProcHeap(std::vector<int> & parr_above_avg, ProcArray *parr) {
 593   int p_index = parr_above_avg.front();
 594   std::pop_heap(parr_above_avg.begin(), parr_above_avg.end(),
 595       TempAwareCommLB::ProcLoadGreater(parr));
 596   parr_above_avg.pop_back();
 597   return p_index;
 598 }
 599
 600
 601 inline void handleTransfer(int randomly_obj_id, ProcInfo& p, int possible_pe, std::vector<int>* parr_objs, ObjGraph *ogr, ProcArray* parr) {
 602   ogr->vertices[randomly_obj_id].setNewPe(possible_pe);
 603   parr_objs[possible_pe].push_back(randomly_obj_id);
 604   ProcInfo &possible_pe_procinfo = parr->procs[possible_pe];
 605
 606   p.totalLoad() -= ogr->vertices[randomly_obj_id].getVertexLoad();
 607   possible_pe_procinfo.totalLoad() += ogr->vertices[randomly_obj_id].getVertexLoad();
 608   eraseObjFromParrObjs(parr_objs[p.getProcId()], randomly_obj_id);
 609   //CkPrintf("After transfered %d from %d : Load %E to %d : Load %E\n", randomly_obj_id, p.getProcId(), p.getTotalLoad(),
 610   //    possible_pe, possible_pe_procinfo.getTotalLoad());
 611 }
 612
 613 inline void updateLoadInfo(int p_index, int possible_pe, double upper_threshold_temp, double lower_threshold_temp,
 614                            std::vector<int>& parr_above_avg, std::vector<int>& parr_below_avg,
 615                            std::vector<bool> &proc_load_info, ProcArray *parr) {
 616
 617   ProcInfo& p = parr->procs[p_index];
 618   ProcInfo& possible_pe_procinfo = parr->procs[possible_pe];
 619
 620   // If the updated load is still greater than the average by the
 621   // threshold value, then push it back to the max heap
 622   if (p.getTotalLoad() > upper_threshold_temp) {
 623     parr_above_avg.push_back(p_index);
 624     std::push_heap(parr_above_avg.begin(), parr_above_avg.end(),
 625         TempAwareCommLB::ProcLoadGreater(parr));
 626     //CkPrintf("\t Pushing pe : %d to max heap\n", p.getProcId());
 627   } else if (p.getTotalLoad() < lower_threshold_temp) {
 628     parr_below_avg.push_back(p_index);
 629     proc_load_info[p_index] = true;
 630     //CkPrintf("\t Adding pe : %d to less loaded\n", p.getProcId());
 631   }
 632
 633   // If the newly assigned processor's load is greater than the average
 634   // by the threshold value, then push it into the max heap.
 635   if (possible_pe_procinfo.getTotalLoad() > upper_threshold_temp) {
 636     // TODO: It should be the index in procarray :(
 637     parr_above_avg.push_back(possible_pe);
 638     std::push_heap(parr_above_avg.begin(), parr_above_avg.end(),
 639         TempAwareCommLB::ProcLoadGreater(parr));
 640     removeFromArray(possible_pe, parr_below_avg);
 641     proc_load_info[possible_pe] = false;
 642     //CkPrintf("\t Pusing pe : %d to max heap\n", possible_pe);
 643   } else if (possible_pe_procinfo.getTotalLoad() < lower_threshold_temp) {
 644   } else {
 645     removeFromArray(possible_pe, parr_below_avg);
 646     proc_load_info[possible_pe] = false;
 647     //CkPrintf("\t Removing from lower list pe : %d\n", possible_pe);
 648   }
 649
 650 }
 651
 652 inline void getPossiblePes(std::vector<int>& possible_pes, int vert,
 653     ObjGraph *ogr, ProcArray* parr) {
 654   std::map<int, int> tmp_map_pid_index;
 655   int counter = 0;
 656   int index;
 657   int i, j, nbrid;
 658   TempAwareCommLB::ObjPeCommInfo objpcomm;
 659  // CkPrintf("%d sends msgs to %d and recv msgs from %d\n", vert,
 660  //   ogr->vertices[vert].sendToList.size(),
 661  //   ogr->vertices[vert].recvFromList.size());
 662
 663   for (i = 0; i < ogr->vertices[vert].sendToList.size(); i++) {
 664     nbrid = ogr->vertices[vert].sendToList[i].getNeighborId();
 665     j = ogr->vertices[nbrid].getNewPe(); // Fix me!! New PE
 666     // TODO: Should it index with vertexId?
 667     if (tmp_map_pid_index.count(j) == 0) {
 668       tmp_map_pid_index[j] = counter;
 669       TempAwareCommLB::PeCommInfo pecomminf(j);
 670       // TODO: Shouldn't it use vertexId instead of vert?
 671       objpcomm.pcomm.push_back(pecomminf);
 672       counter++;
 673     }
 674     index = tmp_map_pid_index[j];
 675
 676     objpcomm.pcomm[index].num_msg +=
 677       ogr->vertices[vert].sendToList[i].getNumMsgs();
 678     objpcomm.pcomm[index].num_bytes +=
 679       ogr->vertices[vert].sendToList[i].getNumBytes();
 680   }
 681
 682   for (i = 0; i < ogr->vertices[vert].recvFromList.size(); i++) {
 683     nbrid = ogr->vertices[vert].recvFromList[i].getNeighborId();
 684     j = ogr->vertices[nbrid].getNewPe();
 685
 686     if (tmp_map_pid_index.count(j) == 0) {
 687       tmp_map_pid_index[j] = counter;
 688       TempAwareCommLB::PeCommInfo pecomminf(j);
 689       // TODO: Shouldn't it use vertexId instead of vert?
 690       objpcomm.pcomm.push_back(pecomminf);
 691       counter++;
 692     }
 693     index = tmp_map_pid_index[j];
 694
 695     objpcomm.pcomm[index].num_msg +=
 696       ogr->vertices[vert].sendToList[i].getNumMsgs();
 697     objpcomm.pcomm[index].num_bytes +=
 698       ogr->vertices[vert].sendToList[i].getNumBytes();
 699   }
 700
 701   // Sort the pe communication vector for this chare
 702   std::sort(objpcomm.pcomm.begin(), objpcomm.pcomm.end(),
 703       TempAwareCommLB::ProcCommGreater());
 704 /*
 705   int pe_id;
 706   int node_id;
 707   int node_size;
 708   int node_first;
 709   //CkPrintf("%d talks to %d pes and possible pes are :\n", vert,
 710   //    objpcomm.pcomm.size());
 711   for (i = 0; i < objpcomm.pcomm.size(); i++) {
 712     pe_id = objpcomm.pcomm[i].pe_id;
 713     node_id = CkNodeOf(pe_id);
 714     node_size = CkNodeSize(node_id);
 715     node_first = CkNodeFirst(node_id);
 716    // CkPrintf("smp details pe_id %d, node_id %d, node_size %d, node_first %d\n",
 717    //   pe_id, node_id, node_size, node_first);
 718     for (j = 0; j < node_size; j++) {
 719       possible_pes.push_back(node_first + j);
 720       //CkPrintf("\t %d:%d (comm: %d)\n",node_id, node_first+j, objpcomm.pcomm[i].num_bytes);
 721     }
 722   }
 723 */
 724 }
 725
 726
 727 #include "TempAwareCommLB.def.h"
 728
 729 /*@}*/
 730