454e3bc642018fdfaf079302ba75e6e6f12584ce
[charm.git] / src / ck-ldb / LBDatabase.C
1 /**
2  * \addtogroup CkLdb
3 */
4 /*@{*/
5
6 #include "converse.h"
7
8 /*
9  * This C++ file contains the Charm stub functions
10  */
11
12 #include "LBDatabase.h"
13 #include "LBSimulation.h"
14 #include "topology.h"
15
16 #include "limits.h"
17
18 #include "NullLB.h"
19
20 struct AdaptiveData {
21   int iteration;
22   double max_load;
23   double avg_load;
24 };
25
26 struct AdaptiveLBDatabase {
27   std::vector<AdaptiveData> history_data;
28 } adaptive_lbdb;
29
30 struct AdaptiveLBStructure {
31   int lb_ideal_period;
32   int lb_calculated_period;
33   int lb_no_iterations;
34   int global_max_iter_no;
35   int global_recv_iter_counter;
36   bool in_progress;
37   double prev_load;
38   double lb_strategy_cost;
39   double lb_migration_cost;
40   bool lb_period_informed;
41   int lb_msg_send_no;
42   int lb_msg_recv_no;
43 } adaptive_struct;
44
45
46 CkReductionMsg* lbDataCollection(int nMsg, CkReductionMsg** msgs) {
47   double lb_data[4];
48   lb_data[0] = 0;
49   lb_data[1] = 0;
50   lb_data[2] = 0;
51   for (int i = 0; i < nMsg; i++) {
52     CkAssert(msgs[i]->getSize() == 4*sizeof(double));
53     double* m = (double *)msgs[i]->getData();
54     lb_data[0] += m[0];
55     lb_data[1] = ((m[1] > lb_data[1])? m[1] : lb_data[1]);
56     lb_data[2] += m[2];
57     if (i == 0) {
58       lb_data[3] = m[3];
59     }
60     if (m[3] != lb_data[3]) {
61       CkPrintf("Error!!! Reduction is intermingled between iteration %lf and\
62       %lf\n", lb_data[3], m[3]);
63     }
64   }
65   return CkReductionMsg::buildNew(4*sizeof(double), lb_data);
66 }
67
68 /*global*/ CkReduction::reducerType lbDataCollectionType;
69 /*initcall*/ void registerLBDataCollection(void) {
70   lbDataCollectionType = CkReduction::addReducer(lbDataCollection);
71 }
72
73 CkGroupID _lbdb;
74
75 CkpvDeclare(int, numLoadBalancers);  /**< num of lb created */
76 CkpvDeclare(int, hasNullLB);         /**< true if NullLB is created */
77 CkpvDeclare(int, lbdatabaseInited);  /**< true if lbdatabase is inited */
78
79 // command line options
80 CkLBArgs _lb_args;
81 int _lb_predict=0;
82 int _lb_predict_delay=10;
83 int _lb_predict_window=20;
84
85 // registry class stores all load balancers linked and created at runtime
86 class LBDBRegistry {
87 friend class LBDBInit;
88 friend class LBDatabase;
89 private:
90   // table for all available LBs linked in
91   struct LBDBEntry {
92     const char *name;
93     LBCreateFn  cfn;
94     LBAllocFn   afn;
95     const char *help;
96     int         shown;          // if 0, donot show in help page
97     LBDBEntry(): name(0), cfn(0), afn(0), help(0), shown(1) {}
98     LBDBEntry(int) {}
99     LBDBEntry(const char *n, LBCreateFn cf, LBAllocFn af, 
100               const char *h, int show=1):
101       name(n), cfn(cf), afn(af), help(h), shown(show) {};
102   };
103   CkVec<LBDBEntry> lbtables;            // a list of available LBs linked
104   CkVec<const char *>   compile_lbs;    // load balancers at compile time
105   CkVec<const char *>   runtime_lbs;    // load balancers at run time
106 public:
107   LBDBRegistry() {}
108   void displayLBs()
109   {
110     CmiPrintf("\nAvailable load balancers:\n");
111     for (int i=0; i<lbtables.length(); i++) {
112       LBDBEntry &entry = lbtables[i];
113       if (entry.shown) CmiPrintf("* %s: %s\n", entry.name, entry.help);
114     }
115     CmiPrintf("\n");
116   }
117   void addEntry(const char *name, LBCreateFn fn, LBAllocFn afn, const char *help, int shown) {
118     lbtables.push_back(LBDBEntry(name, fn, afn, help, shown));
119   }
120   void addCompiletimeBalancer(const char *name) {
121     compile_lbs.push_back(name); 
122   }
123   void addRuntimeBalancer(const char *name) {
124     runtime_lbs.push_back(name); 
125   }
126   LBCreateFn search(const char *name) {
127     char *ptr = strpbrk((char *)name, ":,");
128     int slen = ptr!=NULL?ptr-name:strlen(name);
129     for (int i=0; i<lbtables.length(); i++)
130       if (0==strncmp(name, lbtables[i].name, slen)) return lbtables[i].cfn;
131     return NULL;
132   }
133   LBAllocFn getLBAllocFn(const char *name) {
134     char *ptr = strpbrk((char *)name, ":,");
135     int slen = ptr-name;
136     for (int i=0; i<lbtables.length(); i++)
137       if (0==strncmp(name, lbtables[i].name, slen)) return lbtables[i].afn;
138     return NULL;
139   }
140 };
141
142 static LBDBRegistry lbRegistry;
143
144 void LBDefaultCreate(const char *lbname)
145 {
146   lbRegistry.addCompiletimeBalancer(lbname);
147 }
148
149 // default is to show the helper
150 void LBRegisterBalancer(const char *name, LBCreateFn fn, LBAllocFn afn, const char *help, int shown)
151 {
152   CkPrintf("Adding to registr %s \n", name);
153   lbRegistry.addEntry(name, fn, afn, help, shown);
154 }
155
156 LBAllocFn getLBAllocFn(char *lbname) {
157     return lbRegistry.getLBAllocFn(lbname);
158 }
159
160 LBCreateFn getLBCreateFn(const char *lbname) {
161     return lbRegistry.search(lbname);
162 }
163 // create a load balancer group using the strategy name
164 static void createLoadBalancer(const char *lbname)
165 {
166   CkPrintf("Creating load balancer '%s'\n", lbname);
167       lbRegistry.displayLBs();    // display help page
168     LBCreateFn fn = lbRegistry.search(lbname);
169     if (!fn) {    // invalid lb name
170       CmiPrintf("Abort: Unknown load balancer: '%s'!\n", lbname);
171       lbRegistry.displayLBs();    // display help page
172       CkAbort("Abort");
173     }
174     // invoke function to create load balancer 
175     fn();
176 }
177
178 // mainchare
179 LBDBInit::LBDBInit(CkArgMsg *m)
180 {
181 #if CMK_LBDB_ON
182   _lbdb = CProxy_LBDatabase::ckNew();
183
184   // runtime specified load balancer
185   if (lbRegistry.runtime_lbs.size() > 0) {
186     for (int i=0; i<lbRegistry.runtime_lbs.size(); i++) {
187       const char *balancer = lbRegistry.runtime_lbs[i];
188       createLoadBalancer(balancer);
189     }
190   }
191   else if (lbRegistry.compile_lbs.size() > 0) {
192     for (int i=0; i<lbRegistry.compile_lbs.size(); i++) {
193       const char* balancer = lbRegistry.compile_lbs[i];
194       createLoadBalancer(balancer);
195     }
196   }
197   else {
198     // NullLB is the default when none of above lb created
199     // note user may create his own load balancer in his code manually like
200     // in NAMD, but never mind NullLB can disable itself if there is 
201     // a non NULL LB.
202     createLoadBalancer("NullLB");
203   }
204
205   // simulation mode
206   if (LBSimulation::doSimulation) {
207     CmiPrintf("Charm++> Entering Load Balancer Simulation Mode ... \n");
208     CProxy_LBDatabase(_lbdb).ckLocalBranch()->StartLB();
209   }
210 #endif
211   delete m;
212 }
213
214 // called from init.C
215 void _loadbalancerInit()
216 {
217   CkpvInitialize(int, lbdatabaseInited);
218   CkpvAccess(lbdatabaseInited) = 0;
219   CkpvInitialize(int, numLoadBalancers);
220   CkpvAccess(numLoadBalancers) = 0;
221   CkpvInitialize(int, hasNullLB);
222   CkpvAccess(hasNullLB) = 0;
223
224   char **argv = CkGetArgv();
225   char *balancer = NULL;
226   CmiArgGroup("Charm++","Load Balancer");
227   while (CmiGetArgStringDesc(argv, "+balancer", &balancer, "Use this load balancer")) {
228     if (CkMyRank() == 0)                
229       lbRegistry.addRuntimeBalancer(balancer);   /* lbRegistry is a static */
230   }
231
232   // set up init value for LBPeriod time in seconds
233   // it can also be set by calling LDSetLBPeriod()
234   CmiGetArgDoubleDesc(argv,"+LBPeriod", &_lb_args.lbperiod(),"the minimum time period in seconds allowed for two consecutive automatic load balancing");
235   _lb_args.loop() = CmiGetArgFlagDesc(argv, "+LBLoop", "Use multiple load balancing strategies in loop");
236
237   // now called in cldb.c: CldModuleGeneralInit()
238   // registerLBTopos();
239   CmiGetArgStringDesc(argv, "+LBTopo", &_lbtopo, "define load balancing topology");
240   //Read the K parameter for RefineKLB
241   CmiGetArgIntDesc(argv, "+LBNumMoves", &_lb_args.percentMovesAllowed() , "Percentage of chares to be moved (used by RefineKLB) [0-100]");
242
243   /**************** FUTURE PREDICTOR ****************/
244   _lb_predict = CmiGetArgFlagDesc(argv, "+LBPredictor", "Turn on LB future predictor");
245   CmiGetArgIntDesc(argv, "+LBPredictorDelay", &_lb_predict_delay, "Number of balance steps before learning a model");
246   CmiGetArgIntDesc(argv, "+LBPredictorWindow", &_lb_predict_window, "Number of steps to use to learn a model");
247   if (_lb_predict_window < _lb_predict_delay) {
248     CmiPrintf("LB> [%d] Argument LBPredictorWindow (%d) less than LBPredictorDelay (%d) , fixing\n", CkMyPe(), _lb_predict_window, _lb_predict_delay);
249     _lb_predict_delay = _lb_predict_window;
250   }
251
252   /******************* SIMULATION *******************/
253   // get the step number at which to dump the LB database
254   CmiGetArgIntDesc(argv, "+LBVersion", &_lb_args.lbversion(), "LB database file version number");
255   CmiGetArgIntDesc(argv, "+LBCentPE", &_lb_args.central_pe(), "CentralLB processor");
256   int _lb_dump_activated = 0;
257   if (CmiGetArgIntDesc(argv, "+LBDump", &LBSimulation::dumpStep, "Dump the LB state from this step"))
258     _lb_dump_activated = 1;
259   if (_lb_dump_activated && LBSimulation::dumpStep < 0) {
260     CmiPrintf("LB> Argument LBDump (%d) negative, setting to 0\n",LBSimulation::dumpStep);
261     LBSimulation::dumpStep = 0;
262   }
263   CmiGetArgIntDesc(argv, "+LBDumpSteps", &LBSimulation::dumpStepSize, "Dump the LB state for this amount of steps");
264   if (LBSimulation::dumpStepSize <= 0) {
265     CmiPrintf("LB> Argument LBDumpSteps (%d) too small, setting to 1\n",LBSimulation::dumpStepSize);
266     LBSimulation::dumpStepSize = 1;
267   }
268   CmiGetArgStringDesc(argv, "+LBDumpFile", &LBSimulation::dumpFile, "Set the LB state file name");
269   // get the simulation flag and number. Now the flag can also be avoided by the presence of the number
270   LBSimulation::doSimulation = CmiGetArgIntDesc(argv, "+LBSim", &LBSimulation::simStep, "Read LB state from LBDumpFile since this step");
271   // check for stupid LBSim parameter
272   if (LBSimulation::doSimulation && LBSimulation::simStep < 0) {
273     CmiPrintf("LB> Argument LBSim (%d) invalid, should be >= 0\n");
274     CkExit();
275     return;
276   }
277   CmiGetArgIntDesc(argv, "+LBSimSteps", &LBSimulation::simStepSize, "Read LB state for this number of steps");
278   if (LBSimulation::simStepSize <= 0) {
279     CmiPrintf("LB> Argument LBSimSteps (%d) too small, setting to 1\n",LBSimulation::simStepSize);
280     LBSimulation::simStepSize = 1;
281   }
282
283
284   LBSimulation::simProcs = 0;
285   CmiGetArgIntDesc(argv, "+LBSimProcs", &LBSimulation::simProcs, "Number of target processors.");
286
287   LBSimulation::showDecisionsOnly = 
288     CmiGetArgFlagDesc(argv, "+LBShowDecisions",
289                       "Write to File: Load Balancing Object to Processor Map decisions during LB Simulation");
290
291   // force a global barrier after migration done
292   _lb_args.syncResume() = CmiGetArgFlagDesc(argv, "+LBSyncResume", 
293                   "LB performs a barrier after migration is finished");
294
295   // both +LBDebug and +LBDebug level should work
296   if (!CmiGetArgIntDesc(argv, "+LBDebug", &_lb_args.debug(), 
297                                           "Turn on LB debugging printouts"))
298     _lb_args.debug() = CmiGetArgFlagDesc(argv, "+LBDebug", 
299                                              "Turn on LB debugging printouts");
300
301   // getting the size of the team with +teamSize
302   if (!CmiGetArgIntDesc(argv, "+teamSize", &_lb_args.teamSize(), 
303                                           "Team size"))
304     _lb_args.teamSize() = 1;
305
306   // ask to print summary/quality of load balancer
307   _lb_args.printSummary() = CmiGetArgFlagDesc(argv, "+LBPrintSummary",
308                 "Print load balancing result summary");
309
310   // to ignore baclground load
311   _lb_args.ignoreBgLoad() = CmiGetArgFlagDesc(argv, "+LBNoBackground", 
312                       "Load balancer ignores the background load.");
313 #ifdef __BIGSIM__
314   _lb_args.ignoreBgLoad() = 1;
315 #endif
316   _lb_args.migObjOnly() = CmiGetArgFlagDesc(argv, "+LBObjOnly", 
317                       "Only load balancing migratable objects, ignoring all others.");
318   if (_lb_args.migObjOnly()) _lb_args.ignoreBgLoad() = 1;
319
320   // assume all CPUs are identical
321   _lb_args.testPeSpeed() = CmiGetArgFlagDesc(argv, "+LBTestPESpeed", 
322                       "Load balancer test all CPUs speed.");
323   _lb_args.samePeSpeed() = CmiGetArgFlagDesc(argv, "+LBSameCpus", 
324                       "Load balancer assumes all CPUs are of same speed.");
325   if (!_lb_args.testPeSpeed()) _lb_args.samePeSpeed() = 1;
326
327   _lb_args.useCpuTime() = CmiGetArgFlagDesc(argv, "+LBUseCpuTime", 
328                       "Load balancer uses CPU time instead of wallclock time.");
329
330   // turn instrumentation off at startup
331   _lb_args.statsOn() = !CmiGetArgFlagDesc(argv, "+LBOff",
332                         "Turn load balancer instrumentation off");
333
334   // turn instrumentation of communicatin off at startup
335   _lb_args.traceComm() = !CmiGetArgFlagDesc(argv, "+LBCommOff",
336                 "Turn load balancer instrumentation of communication off");
337
338   // set alpha and beeta
339   _lb_args.alpha() = PER_MESSAGE_SEND_OVERHEAD_DEFAULT;
340   _lb_args.beeta() = PER_BYTE_SEND_OVERHEAD_DEFAULT;
341   CmiGetArgDoubleDesc(argv,"+LBAlpha", &_lb_args.alpha(),
342                            "per message send overhead");
343   CmiGetArgDoubleDesc(argv,"+LBBeta", &_lb_args.beeta(),
344                            "per byte send overhead");
345
346   if (CkMyPe() == 0) {
347     if (_lb_args.debug()) {
348       CmiPrintf("CharmLB> Verbose level %d, load balancing period: %g seconds\n", _lb_args.debug(), _lb_args.lbperiod());
349     }
350     if (_lb_args.debug() > 1) {
351       CmiPrintf("CharmLB> Topology %s alpha: %es beta: %es.\n", _lbtopo, _lb_args.alpha(), _lb_args.beeta());
352     }
353     if (_lb_args.printSummary())
354       CmiPrintf("CharmLB> Load balancer print summary of load balancing result.\n");
355     if (_lb_args.ignoreBgLoad())
356       CmiPrintf("CharmLB> Load balancer ignores processor background load.\n");
357     if (_lb_args.samePeSpeed())
358       CmiPrintf("CharmLB> Load balancer assumes all CPUs are same.\n");
359     if (_lb_args.useCpuTime())
360       CmiPrintf("CharmLB> Load balancer uses CPU time instead of wallclock time.\n");
361     if (LBSimulation::doSimulation)
362       CmiPrintf("CharmLB> Load balancer running in simulation mode on file '%s' version %d.\n", LBSimulation::dumpFile, _lb_args.lbversion());
363     if (_lb_args.statsOn()==0)
364       CkPrintf("CharmLB> Load balancing instrumentation is off.\n");
365     if (_lb_args.traceComm()==0)
366       CkPrintf("CharmLB> Load balancing instrumentation for communication is off.\n");
367     if (_lb_args.migObjOnly())
368       CkPrintf("LB> Load balancing strategy ignores non-migratable objects.\n");
369   }
370 }
371
372 int LBDatabase::manualOn = 0;
373 char *LBDatabase::avail_vector = NULL;
374 CmiNodeLock avail_vector_lock;
375
376 static LBRealType * _expectedLoad = NULL;
377
378 void LBDatabase::initnodeFn()
379 {
380   int proc;
381   int num_proc = CkNumPes();
382   avail_vector= new char[num_proc];
383   for(proc = 0; proc < num_proc; proc++)
384       avail_vector[proc] = 1;
385   avail_vector_lock = CmiCreateLock();
386
387   _expectedLoad = new LBRealType[num_proc];
388   for (proc=0; proc<num_proc; proc++) _expectedLoad[proc]=0.0;
389 }
390
391 // called my constructor
392 void LBDatabase::init(void) 
393 {
394   //thisProxy = CProxy_LBDatabase(thisgroup);
395   myLDHandle = LDCreate();
396   mystep = 0;
397   nloadbalancers = 0;
398   new_ld_balancer = 0;
399
400   CkpvAccess(lbdatabaseInited) = 1;
401 #if CMK_LBDB_ON
402   if (manualOn) TurnManualLBOn();
403 #endif
404   
405   max_load_vec.resize(100, 0);
406   total_load_vec.resize(100, 0);
407   total_contrib_vec.resize(100, 0);
408   max_iteration = -1;
409
410   // If metabalancer enabled, initialize the variables
411   adaptive_struct.lb_ideal_period =  INT_MAX;
412   adaptive_struct.lb_calculated_period = INT_MAX;
413   adaptive_struct.lb_no_iterations = -1;
414   adaptive_struct.global_max_iter_no = 0;
415   adaptive_struct.global_recv_iter_counter = 0;
416   adaptive_struct.in_progress = false;
417   adaptive_struct.prev_load = 0.0;
418   adaptive_struct.lb_strategy_cost = 0.0;
419   adaptive_struct.lb_migration_cost = 0.0;
420   adaptive_struct.lb_msg_send_no = 0;
421   adaptive_struct.lb_msg_recv_no = 0;
422
423 }
424
425 LBDatabase::LastLBInfo::LastLBInfo()
426 {
427   expectedLoad = _expectedLoad;
428 }
429
430 void LBDatabase::get_avail_vector(char * bitmap) {
431     CmiAssert(bitmap && avail_vector);
432     const int num_proc = CkNumPes();
433     for(int proc = 0; proc < num_proc; proc++){
434       bitmap[proc] = avail_vector[proc];
435     }
436 }
437
438 // new_ld == -1(default) : calcualte a new ld
439 //           -2 : ignore new ld
440 //           >=0: given a new ld
441 void LBDatabase::set_avail_vector(char * bitmap, int new_ld){
442     int assigned = 0;
443     const int num_proc = CkNumPes();
444     if (new_ld == -2) assigned = 1;
445     else if (new_ld >= 0) {
446       CmiAssert(new_ld < num_proc);
447       new_ld_balancer = new_ld;
448       assigned = 1;
449     }
450     CmiAssert(bitmap && avail_vector);
451     for(int count = 0; count < num_proc; count++){
452         avail_vector[count] = bitmap[count];
453         if((bitmap[count] == 1) && !assigned){
454             new_ld_balancer = count;
455             assigned = 1;
456         }
457     }
458 }
459
460 // called in CreateFooLB() when multiple load balancers are created
461 // on PE0, BaseLB of each load balancer applies a ticket number
462 // and broadcast the ticket number to all processors
463 int LBDatabase::getLoadbalancerTicket()  { 
464   int seq = nloadbalancers;
465   nloadbalancers ++;
466   loadbalancers.resize(nloadbalancers); 
467   loadbalancers[seq] = NULL;
468   return seq; 
469 }
470
471 void LBDatabase::addLoadbalancer(BaseLB *lb, int seq) {
472 //  CmiPrintf("[%d] addLoadbalancer for seq %d\n", CkMyPe(), seq);
473   if (seq == -1) return;
474   if (CkMyPe() == 0) {
475     CmiAssert(seq < nloadbalancers);
476     if (loadbalancers[seq]) {
477       CmiPrintf("Duplicate load balancer created at %d\n", seq);
478       CmiAbort("LBDatabase");
479     }
480   }
481   else
482     nloadbalancers ++;
483   loadbalancers.resize(seq+1);
484   loadbalancers[seq] = lb;
485 }
486
487 // switch strategy in order
488 void LBDatabase::nextLoadbalancer(int seq) {
489   if (seq == -1) return;                // -1 means this is the only LB
490   int next = seq+1;
491   if (_lb_args.loop()) {
492     if (next == nloadbalancers) next = 0;
493   }
494   else {
495     if (next == nloadbalancers) next --;  // keep using the last one
496   }
497   if (seq != next) {
498     loadbalancers[seq]->turnOff();
499     CmiAssert(loadbalancers[next]);
500     loadbalancers[next]->turnOn();
501   }
502 }
503
504 // return the seq-th load balancer string name of
505 // it can be specified in either compile time or runtime
506 // runtime has higher priority
507 const char *LBDatabase::loadbalancer(int seq) {
508   if (lbRegistry.runtime_lbs.length()) {
509     CmiAssert(seq < lbRegistry.runtime_lbs.length());
510     return lbRegistry.runtime_lbs[seq];
511   }
512   else {
513     CmiAssert(seq < lbRegistry.compile_lbs.length());
514     return lbRegistry.compile_lbs[seq];
515   }
516 }
517
518 void LBDatabase::pup(PUP::er& p)
519
520         IrrGroup::pup(p); 
521         // the memory should be already allocated
522         int np;
523         if (!p.isUnpacking()) np = CkNumPes();
524         p|np;
525         CmiAssert(avail_vector);
526         // in case number of processors changes
527         if (p.isUnpacking() && np > CkNumPes()) {
528                 CmiLock(avail_vector_lock);
529                 delete [] avail_vector;
530                 avail_vector = new char[np];
531                 for (int i=0; i<np; i++) avail_vector[i] = 1;
532                 CmiUnlock(avail_vector_lock);
533         }
534         p(avail_vector, np);
535         p|mystep;
536         if(p.isUnpacking()) nloadbalancers = 0;
537 }
538
539
540 void LBDatabase::EstObjLoad(const LDObjHandle &_h, double cputime)
541 {
542 #if CMK_LBDB_ON
543   LBDB *const db = (LBDB*)(_h.omhandle.ldb.handle);
544   LBObj *const obj = db->LbObj(_h);
545
546   CmiAssert(obj != NULL);
547   obj->setTiming(cputime);
548 #endif
549 }
550
551 void LBDatabase::ResumeClients() {
552   // If metabalancer enabled, initialize the variables
553   adaptive_struct.lb_ideal_period =  INT_MAX;
554   adaptive_struct.lb_calculated_period = INT_MAX;
555   adaptive_struct.lb_no_iterations = -1;
556   adaptive_struct.global_max_iter_no = 0;
557   adaptive_struct.global_recv_iter_counter = 0;
558   adaptive_struct.in_progress = false;
559   adaptive_struct.prev_load = 0.0;
560   adaptive_struct.lb_strategy_cost = 0.0;
561   adaptive_struct.lb_migration_cost = 0.0;
562   adaptive_struct.lb_msg_send_no = 0;
563   adaptive_struct.lb_msg_recv_no = 0;
564   
565   max_load_vec.clear();
566   total_load_vec.clear();
567   total_contrib_vec.clear();
568
569   max_load_vec.resize(100, 0.0);
570   total_load_vec.resize(100, 0.0);
571   total_contrib_vec.resize(100, 0.0);
572
573   LDResumeClients(myLDHandle);
574 }
575
576 bool LBDatabase::AddLoad(int iteration, double load) {
577   total_contrib_vec[iteration]++;
578   //CkPrintf("At PE %d Total contribution for iteration %d is %lf total objs %d\n", CkMyPe(), iteration,
579   //total_contrib_vec[iteration], getLBDB()->ObjDataCount());
580
581   if (iteration > adaptive_struct.lb_no_iterations) {
582     adaptive_struct.lb_no_iterations = iteration;
583   }
584   total_load_vec[iteration] += load;
585   if (max_load_vec[iteration] < load) {
586     max_load_vec[iteration] = load;
587   }
588   if (total_contrib_vec[iteration] == getLBDB()->ObjDataCount()) {
589     double lb_data[4];
590     lb_data[0] = total_load_vec[iteration];
591     lb_data[1] = max_load_vec[iteration];
592     lb_data[2] = getLBDB()->ObjDataCount();
593     lb_data[3] = iteration;
594     //CkPrintf("[%d] sends total load %lf at iter %d\n", CkMyPe(), total_load_vec[iteration], adaptive_struct.lb_no_iterations);
595
596     CkCallback cb(CkIndex_LBDatabase::ReceiveMinStats((CkReductionMsg*)NULL), thisProxy[0]);
597     contribute(4*sizeof(double), lb_data, lbDataCollectionType, cb);
598   }
599   return true;
600 }
601
602 void LBDatabase::ReceiveMinStats(CkReductionMsg *msg) {
603   double* load = (double *) msg->getData();
604   double max = load[1];
605   double avg = load[0]/load[2];
606   int iteration_n = load[3];
607   CkPrintf("[%d] Iteration Total load : %lf Avg load: %lf Max load: %lf for %lf procs\n",iteration_n, load[0], load[0]/load[2], load[1], load[2]);
608   delete msg;
609
610   // Store the data for this iteration
611   adaptive_struct.lb_no_iterations = iteration_n;
612   AdaptiveData data;
613   data.iteration = adaptive_struct.lb_no_iterations;
614   data.max_load = max;
615   data.avg_load = avg;
616   adaptive_lbdb.history_data.push_back(data);
617
618   // If lb period inform is in progress, dont inform again
619   if (adaptive_struct.in_progress) {
620     return;
621   }
622
623 //  if (adaptive_struct.lb_period_informed) {
624 //    return;
625 //  }
626
627   // If the max/avg ratio is greater than the threshold and also this is not the
628   // step immediately after load balancing, carry out load balancing
629   //if (max/avg >= 1.1 && adaptive_lbdb.history_data.size() > 4) {
630   if (max/avg >= 1.5 && adaptive_lbdb.history_data.size() > 4) {
631     CkPrintf("Carry out load balancing step at iter max/avg(%lf) > 1.1\n", max/avg);
632 //    if (!adaptive_struct.lb_period_informed) {
633 //      // Just for testing
634 //      adaptive_struct.lb_calculated_period = 40;
635 //      adaptive_struct.lb_period_informed = true;
636 //      thisProxy.LoadBalanceDecision(adaptive_struct.lb_calculated_period);
637 //      return;
638 //    }
639
640     // If the new lb period is less than current set lb period
641     if (adaptive_struct.lb_calculated_period > iteration_n + 1) {
642       adaptive_struct.lb_calculated_period = iteration_n + 1;
643       adaptive_struct.lb_period_informed = true;
644       adaptive_struct.in_progress = true;
645       CkPrintf("Informing everyone the lb period is %d\n",
646           adaptive_struct.lb_calculated_period);
647       thisProxy.LoadBalanceDecision(adaptive_struct.lb_msg_send_no++, adaptive_struct.lb_calculated_period);
648     }
649     return;
650   }
651
652   // Generate the plan for the adaptive strategy
653   int period;
654   if (generatePlan(period)) {
655     //CkPrintf("Carry out load balancing step at iter\n");
656
657     // If the new lb period is less than current set lb period
658     if (adaptive_struct.lb_calculated_period > period) {
659       adaptive_struct.lb_calculated_period = period;
660       adaptive_struct.in_progress = true;
661       adaptive_struct.lb_period_informed = true;
662       CkPrintf("Informing everyone the lb period is %d\n",
663           adaptive_struct.lb_calculated_period);
664       thisProxy.LoadBalanceDecision(adaptive_struct.lb_msg_send_no++, adaptive_struct.lb_calculated_period);
665     }
666   }
667 }
668
669 bool LBDatabase::generatePlan(int& period) {
670   if (adaptive_lbdb.history_data.size() <= 8) {
671     return false;
672   }
673
674   // Some heuristics for lbperiod
675   // If constant load or almost constant,
676   // then max * new_lb_period > avg * new_lb_period + lb_cost
677   double max = 0.0;
678   double avg = 0.0;
679   AdaptiveData data;
680   for (int i = 0; i < adaptive_lbdb.history_data.size(); i++) {
681     data = adaptive_lbdb.history_data[i];
682     max += data.max_load;
683     avg += data.avg_load;
684     CkPrintf("max (%d, %lf) avg (%d, %lf)\n", i, data.max_load, i, data.avg_load);
685   }
686 //  max /= (adaptive_struct.lb_no_iterations - adaptive_lbdb.history_data[0].iteration);
687 //  avg /= (adaptive_struct.lb_no_iterations - adaptive_lbdb.history_data[0].iteration);
688 //
689 //  adaptive_struct.lb_ideal_period = (adaptive_struct.lb_strategy_cost +
690 //  adaptive_struct.lb_migration_cost) / (max - avg);
691 //  CkPrintf("max : %lf, avg: %lf, strat cost: %lf, migration_cost: %lf, idealperiod : %d \n",
692 //      max, avg, adaptive_struct.lb_strategy_cost, adaptive_struct.lb_migration_cost, adaptive_struct.lb_ideal_period);
693 //
694   // If linearly varying load, then find lb_period
695   // area between the max and avg curve 
696   double mslope, aslope, mc, ac;
697   getLineEq(aslope, ac, mslope, mc);
698   CkPrintf("\n max: %fx + %f; avg: %fx + %f\n", mslope, mc, aslope, ac);
699   double a = (mslope - aslope)/2;
700   double b = (mc - ac);
701   double c = -(adaptive_struct.lb_strategy_cost + adaptive_struct.lb_migration_cost);
702   //c = -2.5;
703   bool got_period = getPeriodForLinear(a, b, c, period);
704   if (!got_period) {
705     return false;
706   }
707   
708   if (mslope < 0) {
709     if (period > (-mc/mslope)) {
710       CkPrintf("Max < 0 Period set when max load is -ve\n");
711       return false;
712     }
713   }
714
715   if (aslope < 0) {
716     if (period > (-ac/aslope)) {
717       CkPrintf("Avg < 0 Period set when avg load is -ve\n");
718       return false;
719     }
720   }
721
722   int intersection_t = (mc-ac) / (aslope - mslope);
723   if (intersection_t > 0 && period > intersection_t) {
724     CkPrintf("Avg | Max Period set when curves intersect\n");
725     return false;
726   }
727   return true;
728 }
729
730 bool LBDatabase::getPeriodForLinear(double a, double b, double c, int& period) {
731   CkPrintf("Quadratic Equation %lf X^2 + %lf X + %lf\n", a, b, c);
732   if (a == 0.0) {
733     period = (-c / b);
734     CkPrintf("Ideal period for linear load %d\n", period);
735     return true;
736   }
737   int x;
738   double t = (b * b) - (4*a*c);
739   if (t < 0) {
740     CkPrintf("(b * b) - (4*a*c) is -ve sqrt : %lf\n", sqrt(t));
741     return false;
742   }
743   t = (-b + sqrt(t)) / (2*a);
744   x = t;
745   if (x < 0) {
746     CkPrintf("boo!!! x (%d) < 0\n", x);
747     x = 0;
748     return false;
749   }
750   period = x;
751   CkPrintf("Ideal period for linear load %d\n", period);
752   return true;
753 }
754
755 bool LBDatabase::getLineEq(double& aslope, double& ac, double& mslope, double& mc) {
756   int total = adaptive_lbdb.history_data.size();
757   int iterations = 1 + adaptive_lbdb.history_data[total - 1].iteration -
758       adaptive_lbdb.history_data[0].iteration;
759   double a1 = 0;
760   double m1 = 0;
761   double a2 = 0;
762   double m2 = 0;
763   AdaptiveData data;
764   int i = 0;
765   for (i = 0; i < total/2; i++) {
766     data = adaptive_lbdb.history_data[i];
767     m1 += data.max_load;
768     a1 += data.avg_load;
769   }
770   m1 /= i;
771   a1 /= i;
772
773   for (i = total/2; i < total; i++) {
774     data = adaptive_lbdb.history_data[i];
775     m2 += data.max_load;
776     a2 += data.avg_load;
777   }
778   m2 /= (i - total/2);
779   a2 /= (i - total/2);
780
781   aslope = 2 * (a2 - a1) / iterations;
782   mslope = 2 * (m2 - m1) / iterations;
783   ac = adaptive_lbdb.history_data[0].avg_load;
784   mc = adaptive_lbdb.history_data[0].max_load;
785   return true;
786 }
787
788 void LBDatabase::LoadBalanceDecision(int req_no, int period) {
789   if (req_no < adaptive_struct.lb_msg_recv_no) {
790     CkPrintf("Error!!! Received a request which was already sent or old\n");
791     return;
792   }
793   //CkPrintf("[%d] Load balance decision made cur iteration: %d period:%d state: %d\n",CkMyPe(), adaptive_struct.lb_no_iterations, period, local_state);
794   adaptive_struct.lb_ideal_period = period;
795   //local_state = ON;
796   adaptive_struct.lb_msg_recv_no = req_no;
797   thisProxy[0].ReceiveIterationNo(req_no, adaptive_struct.lb_no_iterations);
798 }
799
800 void LBDatabase::LoadBalanceDecisionFinal(int req_no, int period) {
801   if (req_no < adaptive_struct.lb_msg_recv_no) {
802     return;
803   }
804 //  CkPrintf("[%d] Final Load balance decision made cur iteration: %d period:%d \n",CkMyPe(), adaptive_struct.lb_no_iterations, period);
805   adaptive_struct.lb_ideal_period = period;
806   LDOMAdaptResumeSync(myLDHandle, period);
807
808 //  if (local_state == ON) {
809 //    local_state = DECIDED;
810 //    return;
811 //  }
812
813   // If the state is PAUSE, then its waiting for the final decision from central
814   // processor. If the decision is that the ideal period is in the future,
815   // resume. If the ideal period is now, then carry out load balancing.
816 //  if (local_state == PAUSE) {
817 //    if (adaptive_struct.lb_no_iterations < adaptive_struct.lb_ideal_period) {
818 //      local_state = DECIDED;
819 //      //SendMinStats();
820 //      //FIX ME!!! ResumeClients(0);
821 //    } else {
822 //      local_state = LOAD_BALANCE;
823 //      //FIX ME!!! ProcessAtSync();
824 //    }
825 //    return;
826 //  }
827 //  CkPrintf("Error!!! Final decision received but the state is invalid %d\n", local_state);
828 }
829
830
831 void LBDatabase::ReceiveIterationNo(int req_no, int local_iter_no) {
832   CmiAssert(CkMyPe() == 0);
833
834   adaptive_struct.global_recv_iter_counter++;
835   if (local_iter_no > adaptive_struct.global_max_iter_no) {
836     adaptive_struct.global_max_iter_no = local_iter_no;
837   }
838   if (CkNumPes() == adaptive_struct.global_recv_iter_counter) {
839     adaptive_struct.lb_ideal_period = (adaptive_struct.lb_ideal_period > adaptive_struct.global_max_iter_no) ? adaptive_struct.lb_ideal_period : adaptive_struct.global_max_iter_no + 1;
840     thisProxy.LoadBalanceDecisionFinal(req_no, adaptive_struct.lb_ideal_period);
841     CkPrintf("Final lb_period %d\n", adaptive_struct.lb_ideal_period);
842     adaptive_struct.in_progress = false;
843     adaptive_struct.global_max_iter_no = 0;
844     adaptive_struct.global_recv_iter_counter = 0;
845   }
846 }
847
848 int LBDatabase::getPredictedLBPeriod() {
849   return adaptive_struct.lb_ideal_period;
850 }
851
852 /*
853   callable from user's code
854 */
855 void TurnManualLBOn()
856 {
857 #if CMK_LBDB_ON
858    LBDatabase * myLbdb = LBDatabase::Object();
859    if (myLbdb) {
860      myLbdb->TurnManualLBOn();
861    }
862    else {
863      LBDatabase::manualOn = 1;
864    }
865 #endif
866 }
867
868 void TurnManualLBOff()
869 {
870 #if CMK_LBDB_ON
871    LBDatabase * myLbdb = LBDatabase::Object();
872    if (myLbdb) {
873      myLbdb->TurnManualLBOff();
874    }
875    else {
876      LBDatabase::manualOn = 0;
877    }
878 #endif
879 }
880
881 extern "C" void LBTurnInstrumentOn() { 
882 #if CMK_LBDB_ON
883   if (CkpvAccess(lbdatabaseInited))
884     LBDatabase::Object()->CollectStatsOn(); 
885   else
886     _lb_args.statsOn() = 1;
887 #endif
888 }
889
890 extern "C" void LBTurnInstrumentOff() { 
891 #if CMK_LBDB_ON
892   if (CkpvAccess(lbdatabaseInited))
893     LBDatabase::Object()->CollectStatsOff(); 
894   else
895     _lb_args.statsOn() = 0;
896 #endif
897 }
898 void LBClearLoads() {
899 #if CMK_LBDB_ON
900   LBDatabase::Object()->ClearLoads(); 
901 #endif
902 }
903
904 void LBTurnPredictorOn(LBPredictorFunction *model) {
905 #if CMK_LBDB_ON
906   LBDatabase::Object()->PredictorOn(model);
907 #endif
908 }
909
910 void LBTurnPredictorOn(LBPredictorFunction *model, int wind) {
911 #if CMK_LBDB_ON
912   LBDatabase::Object()->PredictorOn(model, wind);
913 #endif
914 }
915
916 void LBTurnPredictorOff() {
917 #if CMK_LBDB_ON
918   LBDatabase::Object()->PredictorOff();
919 #endif
920 }
921
922 void LBChangePredictor(LBPredictorFunction *model) {
923 #if CMK_LBDB_ON
924   LBDatabase::Object()->ChangePredictor(model);
925 #endif
926 }
927
928 void LBSetPeriod(double second) {
929 #if CMK_LBDB_ON
930   if (CkpvAccess(lbdatabaseInited))
931     LBDatabase::Object()->SetLBPeriod(second); 
932   else
933     _lb_args.lbperiod() = second;
934 #endif
935 }
936
937 #include "LBDatabase.def.h"
938
939 /*@}*/