f09af887ed8f988e86311af1f3ee45490c17172e
[charm.git] / src / ck-ldb / MetaBalancer.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 "MetaBalancer.h"
13 #include "topology.h"
14
15 #include "limits.h"
16
17 #define VEC_SIZE 50
18 #define IMB_TOLERANCE 1.1
19 #define OUTOFWAY_TOLERANCE 2
20 #define UTILIZATION_THRESHOLD 0.7
21 #define NEGLECT_IDLE 2 // Should never be == 1
22 #define MIN_STATS 6
23 #define STATS_COUNT 8 // The number of stats collected during reduction
24
25 #   define DEBAD(x) /*CkPrintf x*/
26 #   define EXTRA_FEATURE 0
27
28 CkReductionMsg* lbDataCollection(int nMsg, CkReductionMsg** msgs) {
29   double lb_data[STATS_COUNT];
30   lb_data[1] = 0.0; // total number of processors contributing
31   lb_data[2] = 0.0; // total load
32   lb_data[3] = 0.0; // max load
33   lb_data[4] = 0.0; // idle time
34   lb_data[5] = 1.0; // utilization
35   lb_data[6] = 0.0; // total load with bg
36   lb_data[7] = 0.0; // max load with bg
37   for (int i = 0; i < nMsg; i++) {
38     CkAssert(msgs[i]->getSize() == STATS_COUNT*sizeof(double));
39     if (msgs[i]->getSize() != STATS_COUNT*sizeof(double)) {
40       CkPrintf("Error!!! Reduction not correct. Msg size is %d\n", msgs[i]->getSize());
41     }
42     double* m = (double *)msgs[i]->getData();
43     // Total count
44     lb_data[1] += m[1];
45     // Avg load
46     lb_data[2] += m[2];
47     // Max load
48     lb_data[3] = ((m[3] > lb_data[3])? m[3] : lb_data[3]);
49     // Avg idle
50     lb_data[4] += m[4];
51     // Get least utilization
52     lb_data[5] = ((m[5] < lb_data[5]) ? m[5] : lb_data[5]);
53     // Get Avg load with bg
54     lb_data[6] += m[6];
55     // Get Max load with bg
56     lb_data[7] = ((m[7] > lb_data[7])? m[7] : lb_data[7]);
57     if (i == 0) {
58       // Iteration no
59       lb_data[0] = m[0];
60     }
61     if (m[0] != lb_data[0]) {
62       CkPrintf("Error!!! Reduction is intermingled between iteration %lf and\
63       %lf\n", lb_data[0], m[0]);
64       CkAbort("Intermingling iterations\n");
65     }
66   }
67   return CkReductionMsg::buildNew(STATS_COUNT*sizeof(double), lb_data);
68 }
69
70 /*global*/ CkReduction::reducerType lbDataCollectionType;
71 /*initcall*/ void registerLBDataCollection(void) {
72   lbDataCollectionType = CkReduction::addReducer(lbDataCollection);
73 }
74
75 CkGroupID _metalb;
76
77 CkpvDeclare(int, metalbInited);  /**< true if metabalancer is inited */
78
79 // mainchare
80 MetaLBInit::MetaLBInit(CkArgMsg *m) {
81 #if CMK_LBDB_ON
82   _metalb = CProxy_MetaBalancer::ckNew();
83 #endif
84   delete m;
85 }
86
87 // called from init.C
88 void _metabalancerInit() {
89   CkpvInitialize(int, metalbInited);
90   CkpvAccess(metalbInited) = 0;
91 }
92
93 void MetaBalancer::initnodeFn() {
94 }
95
96 // called by my constructor
97 void MetaBalancer::init(void) {
98   lbdatabase = (LBDatabase *)CkLocalBranch(_lbdb);
99   CkpvAccess(metalbInited) = 1;
100   total_load_vec.resize(VEC_SIZE, 0.0);
101   total_count_vec.resize(VEC_SIZE, 0);
102   max_iteration = -1;
103   prev_idle = 0.0;
104   alpha_beta_cost_to_load = 1.0; // Some random value. TODO: Find the actual
105
106   adaptive_lbdb.lb_iter_no = -1;
107
108   // If metabalancer enabled, initialize the variables
109   adaptive_struct.tentative_period =  INT_MAX;
110   adaptive_struct.final_lb_period =  INT_MAX;
111   adaptive_struct.lb_calculated_period = INT_MAX;
112   adaptive_struct.lb_iteration_no = -1;
113   adaptive_struct.global_max_iter_no = 0;
114   adaptive_struct.tentative_max_iter_no = -1;
115   adaptive_struct.global_recv_iter_counter = 0;
116   adaptive_struct.in_progress = false;
117   adaptive_struct.lb_strategy_cost = 0.0;
118   adaptive_struct.lb_migration_cost = 0.0;
119   adaptive_struct.lb_msg_send_no = 0;
120   adaptive_struct.lb_msg_recv_no = 0;
121   adaptive_struct.total_syncs_called = 0;
122   adaptive_struct.last_lb_type = -1;
123
124
125   // This is indicating if the load balancing strategy and migration started.
126   // This is mainly used to register callbacks for noobj pes. They would
127   // register as soon as resumefromsync is called. On receiving the handles at
128   // the central pe, it clears the previous handlers and sets lb_in_progress
129   // to false so that it doesn't clear the handles.
130   lb_in_progress = false;
131
132   is_prev_lb_refine = -1;
133 }
134
135 void MetaBalancer::pup(PUP::er& p) {
136   CkPrintf("[%d] Metabalancer getting pupped\n", CkMyPe());
137         IrrGroup::pup(p);
138   if (p.isUnpacking()) {
139     lbdatabase = (LBDatabase *)CkLocalBranch(_lbdb);
140   }
141   // NOTE set lbdatabase using the id
142 }
143
144
145 void MetaBalancer::ResumeClients() {
146   // If metabalancer enabled, initialize the variables
147   adaptive_lbdb.history_data.clear();
148
149   adaptive_struct.tentative_period =  INT_MAX;
150   adaptive_struct.final_lb_period =  INT_MAX;
151   adaptive_struct.lb_calculated_period = INT_MAX;
152   adaptive_struct.lb_iteration_no = -1;
153   adaptive_struct.global_max_iter_no = 0;
154   adaptive_struct.tentative_max_iter_no = -1;
155   adaptive_struct.global_recv_iter_counter = 0;
156   adaptive_struct.in_progress = false;
157   adaptive_struct.lb_strategy_cost = 0.0;
158   adaptive_struct.lb_migration_cost = 0.0;
159   adaptive_struct.lb_msg_send_no = 0;
160   adaptive_struct.lb_msg_recv_no = 0;
161   adaptive_struct.total_syncs_called = 0;
162
163   prev_idle = 0.0;
164   if (lb_in_progress) {
165     lbdb_no_obj_callback.clear();
166     lb_in_progress = false;
167   }
168   // While resuming client, if we find that there are no objects, then handle
169   // the case accordingly.
170   if (lbdatabase->getLBDB()->ObjDataCount() == 0) {
171     CkPrintf("%d processor has 0 objs\n", CkMyPe());
172     HandleAdaptiveNoObj();
173   }
174 }
175
176 int MetaBalancer::get_iteration() {
177   CkPrintf("[%d] get iteration %d\n", CkMyPe(), adaptive_struct.lb_iteration_no);
178   return adaptive_struct.lb_iteration_no;
179 }
180
181 bool MetaBalancer::AddLoad(int it_n, double load) {
182   int index = it_n % VEC_SIZE;
183   total_count_vec[index]++;
184   adaptive_struct.total_syncs_called++;
185   DEBAD(("At PE %d Total contribution for iteration %d is %d total objs %d\n",
186       CkMyPe(), it_n, total_count_vec[index],
187       lbdatabase->getLBDB()->ObjDataCount()));
188
189   if (it_n < adaptive_struct.lb_iteration_no) {
190     CkAbort("Error!! Received load for previous iteration\n");
191   }
192   if (it_n > adaptive_struct.lb_iteration_no) {
193     adaptive_struct.lb_iteration_no = it_n;
194   }
195   total_load_vec[index] += load;
196   if (total_count_vec[index] > lbdatabase->getLBDB()->ObjDataCount()) {
197     CkPrintf("iteration %d received %d contributions and expected %d\n", it_n,
198         total_count_vec[index], lbdatabase->getLBDB()->ObjDataCount());
199     CkAbort("Abort!!! Received more contribution");
200   }
201
202   if (total_count_vec[index] == lbdatabase->getLBDB()->ObjDataCount()) {
203     double idle_time, bg_walltime, cpu_bgtime;
204     lbdatabase->IdleTime(&idle_time);
205     lbdatabase->BackgroundLoad(&bg_walltime, &cpu_bgtime);
206
207     int sync_for_bg = adaptive_struct.total_syncs_called +
208         lbdatabase->getLBDB()->ObjDataCount();
209     bg_walltime = bg_walltime * lbdatabase->getLBDB()->ObjDataCount() / sync_for_bg;
210
211     if (it_n < NEGLECT_IDLE) {
212       prev_idle = idle_time;
213     }
214     idle_time -= prev_idle;
215
216     // The chares do not contribute their 0th iteration load. So the total syncs
217     // in reality is total_syncs_called + obj_counts
218     int total_countable_syncs = adaptive_struct.total_syncs_called +
219         (1 - NEGLECT_IDLE) * lbdatabase->getLBDB()->ObjDataCount(); // TODO: Fix me! weird!
220     if (total_countable_syncs != 0) {
221       idle_time = idle_time * lbdatabase->getLBDB()->ObjDataCount() / total_countable_syncs;
222     }
223     //CkPrintf("[%d] Idle time %lf and countable %d for iteration %d\n", CkMyPe(), idle_time, total_countable_syncs, iteration);
224
225     double lb_data[STATS_COUNT];
226     lb_data[0] = it_n;
227     lb_data[1] = 1;
228     lb_data[2] = total_load_vec[index]; // For average load
229     lb_data[3] = total_load_vec[index]; // For max load
230     lb_data[4] = idle_time;
231     // Set utilization
232     if (total_load_vec[index] == 0.0) {
233       lb_data[5] = 0.0;
234     } else {
235       lb_data[5] = total_load_vec[index]/(idle_time + total_load_vec[index]);
236     }
237     lb_data[6] = lb_data[2] + bg_walltime; // For Avg load with bg
238     lb_data[7] = lb_data[6]; // For Max load with bg
239     total_load_vec[index] = 0.0;
240     total_count_vec[index] = 0;
241
242     //CkPrintf("   [%d] sends total load %lf idle time %lf ratio of idle/load %lf at iter %d\n", CkMyPe(),
243     //    total_load_vec[iteration], idle_time,
244     //    idle_time/total_load_vec[iteration], adaptive_struct.lb_iteration_no);
245
246     CkCallback cb(CkIndex_MetaBalancer::ReceiveMinStats((CkReductionMsg*)NULL), thisProxy[0]);
247     contribute(STATS_COUNT*sizeof(double), lb_data, lbDataCollectionType, cb);
248   }
249   return true;
250 }
251
252 void MetaBalancer::ReceiveMinStats(CkReductionMsg *msg) {
253   double* load = (double *) msg->getData();
254   double avg = load[2]/load[1];
255   double max = load[3];
256   double avg_idle = load[4]/load[1];
257   double utilization = load[5];
258   int iteration_n = load[0];
259   double avg_load_bg = load[6]/load[1];
260   double max_load_bg = load[7];
261   DEBAD(("** [%d] Iteration Avg load: %lf Max load: %lf Avg Idle : %lf Max Idle : %lf for %lf procs\n",iteration_n, avg, max, avg_idle, utilization, load[1]));
262   CkPrintf("** [%d] Iteration Avg load: %lf Max load: %lf With bg Avg load: %lf Max load: %lf Avg Idle : %lf Max Idle : %lf for %lf procs\n",iteration_n, avg, max, avg_load_bg, max_load_bg, avg_idle, utilization, load[1]);
263   delete msg;
264
265 #if EXTRA_FEATURE
266   if (adaptive_struct.final_lb_period != iteration_n) {
267     for (int i = 0; i < lbdb_no_obj_callback.size(); i++) {
268       thisProxy[lbdb_no_obj_callback[i]].TriggerAdaptiveReduction();
269     }
270   }
271 #endif
272
273   // Store the data for this iteration
274   adaptive_lbdb.lb_iter_no = iteration_n;
275   AdaptiveData data;
276   data.iteration = adaptive_lbdb.lb_iter_no;
277   data.max_load = max;
278   data.avg_load = avg;
279   data.utilization = utilization;
280   data.idle_time = avg_idle;
281   adaptive_lbdb.history_data.push_back(data);
282
283   // If lb period inform is in progress, dont inform again.
284   // If this is the lb data corresponding to the final lb period informed, then
285   // don't recalculate as some of the processors might have already gone into
286   // LB_STAGE.
287   if (adaptive_struct.in_progress || (adaptive_struct.final_lb_period == iteration_n)) {
288     return;
289   }
290
291   double utilization_threshold = UTILIZATION_THRESHOLD;
292
293 #if EXTRA_FEATURE
294   CkPrintf("alpha_beta_to_load %lf\n", alpha_beta_cost_to_load);
295   if (alpha_beta_cost_to_load < 0.1) {
296     // Ignore the effect of idle time and there by lesser utilization. So we
297     // assign utilization threshold to be 0.0
298     CkPrintf("Changing the idle load tolerance coz this isn't communication intensive benchmark\n");
299     utilization_threshold = 0.0;
300   }
301 #endif
302
303   // First generate the lb period based on the cost of lb. Find out what is the
304   // expected imbalance at the calculated lb period.
305   int period;
306   // This is based on the new max load after load balancing. So technically, it
307   // is calculated based on the shifter up avg curve.
308   double ratio_at_t = 1.0;
309   int tmp_lb_type;
310   double tmp_max_avg_ratio, tmp_comm_ratio;
311   GetPrevLBData(tmp_lb_type, tmp_max_avg_ratio, tmp_comm_ratio);
312   double tolerate_imb = IMB_TOLERANCE * tmp_max_avg_ratio;
313
314   if (generatePlan(period, ratio_at_t)) {
315     DEBAD(("Generated period and calculated %d and period %d max iter %d\n",
316       adaptive_struct.lb_calculated_period, period,
317       adaptive_struct.tentative_max_iter_no));
318     // set the imbalance tolerance to be ratio_at_calculated_lb_period
319     if (ratio_at_t != 1.0) {
320       CkPrintf("Changed tolerance to %lf after line eq whereas max/avg is %lf\n", ratio_at_t, max/avg);
321       // Since ratio_at_t is shifter up, max/(tmp_max_avg_ratio * avg) should be
322       // compared with the tolerance
323       tolerate_imb = ratio_at_t * tmp_max_avg_ratio * OUTOFWAY_TOLERANCE;
324     }
325
326     CkPrintf("Prev LB Data Type %d, max/avg %lf, local/remote %lf\n", tmp_lb_type, tmp_max_avg_ratio, tmp_comm_ratio);
327
328     if ((utilization < utilization_threshold || max/avg >= tolerate_imb) &&
329           adaptive_lbdb.history_data.size() > MIN_STATS) {
330       CkPrintf("Trigger soon even though we calculated lbperiod max/avg(%lf) and utilization ratio (%lf)\n", max/avg, utilization);
331       TriggerSoon(iteration_n, max/avg, tolerate_imb);
332       return;
333     }
334
335     // If the new lb period from linear extrapolation is greater than maximum
336     // iteration known from previously collected data, then inform all the
337     // processors about the new calculated period.
338     if (period > adaptive_struct.tentative_max_iter_no && period !=
339           adaptive_struct.final_lb_period) {
340       adaptive_struct.doCommStrategy = false;
341       adaptive_struct.lb_calculated_period = period;
342       adaptive_struct.in_progress = true;
343       CkPrintf("Sticking to the calculated period %d\n",
344         adaptive_struct.lb_calculated_period);
345       thisProxy.LoadBalanceDecision(adaptive_struct.lb_msg_send_no++,
346         adaptive_struct.lb_calculated_period);
347       return;
348     }
349     // TODO: Shouldn't we return from here??
350   }
351
352   CkPrintf("Prev LB Data Type %d, max/avg %lf, local/remote %lf\n", tmp_lb_type, tmp_max_avg_ratio, tmp_comm_ratio);
353
354   // This would be called when the datasize is not enough to calculate lb period
355   if ((utilization < utilization_threshold || max/avg >= tolerate_imb) && adaptive_lbdb.history_data.size() > 4) {
356     CkPrintf("Carry out load balancing step at iter max/avg(%lf) and utilization ratio (%lf)\n", max/avg, utilization);
357     TriggerSoon(iteration_n, max/avg, tolerate_imb);
358     return;
359   }
360
361 }
362
363 void MetaBalancer::TriggerSoon(int iteration_n, double imbalance_ratio,
364     double tolerate_imb) {
365
366   // If the previously calculated_period (not the final decision) is greater
367   // than the iter +1 and if it is greater than the maximum iteration we have
368   // seen so far, then we can inform this
369   if ((iteration_n + 1 > adaptive_struct.tentative_max_iter_no) &&
370       (iteration_n+1 < adaptive_struct.lb_calculated_period) &&
371       (iteration_n + 1 != adaptive_struct.final_lb_period)) {
372     if (imbalance_ratio < tolerate_imb) {
373       adaptive_struct.doCommStrategy = true;
374       CkPrintf("No load imbalance but idle time\n");
375     } else {
376       adaptive_struct.doCommStrategy = false;
377       CkPrintf("load imbalance \n");
378     }
379     adaptive_struct.lb_calculated_period = iteration_n + 1;
380     adaptive_struct.in_progress = true;
381     CkPrintf("Informing everyone the lb period is %d\n",
382         adaptive_struct.lb_calculated_period);
383     thisProxy.LoadBalanceDecision(adaptive_struct.lb_msg_send_no++,
384         adaptive_struct.lb_calculated_period);
385   }
386 }
387
388 bool MetaBalancer::generatePlan(int& period, double& ratio_at_t) {
389   if (adaptive_lbdb.history_data.size() <= 4) {
390     return false;
391   }
392
393   // Some heuristics for lbperiod
394   // If constant load or almost constant,
395   // then max * new_lb_period > avg * new_lb_period + lb_cost
396   double max = 0.0;
397   double avg = 0.0;
398   AdaptiveData data;
399   for (int i = 0; i < adaptive_lbdb.history_data.size(); i++) {
400     data = adaptive_lbdb.history_data[i];
401     max += data.max_load;
402     avg += data.avg_load;
403     //DEBAD(("max (%d, %lf) avg (%d, %lf)\n", i, data.max_load, i, data.avg_load));
404     //CkPrintf("max (%d, %lf) avg (%d, %lf)\n", i, data.max_load, i, data.avg_load);
405   }
406 //  max /= (adaptive_struct.lb_iteration_no - adaptive_lbdb.history_data[0].iteration);
407 //  avg /= (adaptive_struct.lb_iteration_no - adaptive_lbdb.history_data[0].iteration);
408
409   // If linearly varying load, then find lb_period
410   // area between the max and avg curve
411   // If we can attain perfect balance, then the new load is close to the
412   // average. Hence we pass 1, else pass in some other value which would be the
413   // new max_load after load balancing.
414   int tmp_lb_type;
415   double tmp_max_avg_ratio, tmp_comm_ratio;
416   double tolerate_imb;
417
418 #if EXTRA_FEATURE
419   // First get the data for refine.
420   GetLBDataForLB(1, tmp_max_avg_ratio, tmp_comm_ratio);
421   tolerate_imb = tmp_max_avg_ratio;
422
423   // If RefineLB does a good job, then find the period considering RefineLB
424   if (tmp_max_avg_ratio <= 1.01) {
425     if (max/avg < tolerate_imb) {
426       CkPrintf("Resorting to imb = 1.0 coz max/avg (%lf) < imb(%lf)\n", max/avg, tolerate_imb);
427       tolerate_imb = 1.0;
428     }
429     CkPrintf("Will generate plan for refine %lf imb and %lf overhead\n", tolerate_imb, 0.2);
430     return getPeriodForStrategy(tolerate_imb, 0.2, period, ratio_at_t);
431   }
432
433   GetLBDataForLB(0, tmp_max_avg_ratio, tmp_comm_ratio);
434 #endif
435
436   GetPrevLBData(tmp_lb_type, tmp_max_avg_ratio, tmp_comm_ratio);
437   tolerate_imb = tmp_max_avg_ratio;
438 //  if (max/avg < tolerate_imb) {
439 //    CkPrintf("Resorting to imb = 1.0 coz max/avg (%lf) < imb(%lf)\n", max/avg, tolerate_imb);
440 //    tolerate_imb = 1.0;
441 //  }
442   if (max/avg > tolerate_imb) {
443     if (getPeriodForStrategy(tolerate_imb, 1, period, ratio_at_t)) {
444       return true;
445     }
446   }
447
448   max = 0.0;
449   avg = 0.0;
450   for (int i = 0; i < adaptive_lbdb.history_data.size(); i++) {
451     data = adaptive_lbdb.history_data[i];
452     max += data.max_load;
453     avg += data.avg_load*tolerate_imb;
454     //DEBAD(("max (%d, %lf) avg (%d, %lf)\n", i, data.max_load, i, data.avg_load));
455     //CkPrintf("max (%d, %lf) avg (%d, %lf)\n", i, data.max_load, i, data.avg_load);
456   }
457   max /= adaptive_lbdb.history_data.size();
458   avg /= adaptive_lbdb.history_data.size();
459   double cost = adaptive_struct.lb_strategy_cost + adaptive_struct.lb_migration_cost;
460   period = cost/(max - avg); 
461   CkPrintf("Obtained period %d from constant prediction\n", period);
462   if (period < 0) { 
463     period = adaptive_struct.final_lb_period;
464     CkPrintf("Obtained -ve period from constant prediction so changing to prev %d\n", period);
465   } 
466   ratio_at_t = max / avg;
467   return true;
468 }
469
470 bool MetaBalancer::getPeriodForStrategy(double new_load_percent,
471     double overhead_percent, int& period, double& ratio_at_t) {
472   double mslope, aslope, mc, ac;
473   getLineEq(new_load_percent, aslope, ac, mslope, mc);
474   CkPrintf("new load percent %lf\n", new_load_percent);
475   CkPrintf("\n max: %fx + %f; avg: %fx + %f\n", mslope, mc, aslope, ac);
476   double a = (mslope - aslope)/2;
477   double b = (mc - ac);
478   double c = -(adaptive_struct.lb_strategy_cost +
479       adaptive_struct.lb_migration_cost) * overhead_percent;
480   bool got_period = getPeriodForLinear(a, b, c, period);
481   if (!got_period) {
482     return false;
483   }
484
485   if (mslope < 0) {
486     if (period > (-mc/mslope)) {
487       CkPrintf("Max < 0 Period set when max load is -ve\n");
488       return false;
489     }
490   }
491
492   if (aslope < 0) {
493     if (period > (-ac/aslope)) {
494       CkPrintf("Avg < 0 Period set when avg load is -ve\n");
495       return false;
496     }
497   }
498
499   int intersection_t = (mc-ac) / (aslope - mslope);
500   if (intersection_t > 0 && period > intersection_t) {
501     CkPrintf("Avg | Max Period set when curves intersect\n");
502     return false;
503   }
504   ratio_at_t = ((mslope*period + mc)/(aslope*period + ac));
505   CkPrintf("Ratio at t (%lf*%d + %lf) / (%lf*%d+%lf) = %lf\n", mslope, period, mc, aslope, period, ac, ratio_at_t);
506   return true;
507 }
508
509 bool MetaBalancer::getPeriodForLinear(double a, double b, double c, int& period) {
510   CkPrintf("Quadratic Equation %lf X^2 + %lf X + %lf\n", a, b, c);
511   if (a == 0.0) {
512     period = (-c / b);
513     if (period < 0) {
514       CkPrintf("-ve period for -c/b (%d)\n", period);
515       return false;
516     }
517     CkPrintf("Ideal period for linear load %d\n", period);
518     return true;
519   }
520   int x;
521   double t = (b * b) - (4*a*c);
522   if (t < 0) {
523     CkPrintf("(b * b) - (4*a*c) is -ve sqrt : %lf\n", sqrt(t));
524     return false;
525   }
526   t = (-b + sqrt(t)) / (2*a);
527   x = t;
528   if (x < 0) {
529     CkPrintf("boo!!! x (%d) < 0\n", x);
530     x = 0;
531     return false;
532   }
533   period = x;
534   CkPrintf("Ideal period for linear load %d\n", period);
535   return true;
536 }
537
538 bool MetaBalancer::getLineEq(double new_load_percent, double& aslope, double& ac, double& mslope, double& mc) {
539   int total = adaptive_lbdb.history_data.size();
540   int iterations = 1 + adaptive_lbdb.history_data[total - 1].iteration -
541       adaptive_lbdb.history_data[0].iteration;
542   double a1 = 0;
543   double m1 = 0;
544   double a2 = 0;
545   double m2 = 0;
546   AdaptiveData data;
547   int i = 0;
548   for (i = 0; i < total/2; i++) {
549     data = adaptive_lbdb.history_data[i];
550     m1 += data.max_load;
551     a1 += data.avg_load;
552     CkPrintf("max (%d, %lf) avg (%d, %lf) adjusted_avg (%d, %lf)\n", i, data.max_load, i, data.avg_load, i, new_load_percent*data.avg_load);
553   }
554   m1 /= i;
555   a1 = (a1 * new_load_percent) / i;
556
557   for (i = total/2; i < total; i++) {
558     data = adaptive_lbdb.history_data[i];
559     m2 += data.max_load;
560     a2 += data.avg_load;
561     CkPrintf("max (%d, %lf) avg (%d, %lf) adjusted_avg (%d, %lf)\n", i, data.max_load, i, data.avg_load, i, new_load_percent*data.avg_load);
562   }
563   m2 /= (i - total/2);
564   a2 = (a2 * new_load_percent) / (i - total/2);
565
566   aslope = 2 * (a2 - a1) / iterations;
567   mslope = 2 * (m2 - m1) / iterations;
568   ac = adaptive_lbdb.history_data[0].avg_load * new_load_percent;
569   mc = adaptive_lbdb.history_data[0].max_load;
570
571   ac = a1 - ((aslope * total)/4);
572   mc = m1 - ((mslope * total)/4);
573
574   //ac = (adaptive_lbdb.history_data[1].avg_load * new_load_percent - aslope);
575   //mc = (adaptive_lbdb.history_data[1].max_load - mslope);
576
577   return true;
578 }
579
580 void MetaBalancer::LoadBalanceDecision(int req_no, int period) {
581   if (req_no < adaptive_struct.lb_msg_recv_no) {
582     CkPrintf("Error!!! Received a request which was already sent or old\n");
583     return;
584   }
585   CkPrintf("[%d] Load balance decision made cur iteration: %d period:%d\n",CkMyPe(), adaptive_struct.lb_iteration_no, period);
586   adaptive_struct.tentative_period = period;
587   adaptive_struct.lb_msg_recv_no = req_no;
588   thisProxy[0].ReceiveIterationNo(req_no, adaptive_struct.lb_iteration_no);
589 }
590
591 void MetaBalancer::LoadBalanceDecisionFinal(int req_no, int period) {
592   if (req_no < adaptive_struct.lb_msg_recv_no) {
593     return;
594   }
595   DEBAD(("[%d] Final Load balance decision made cur iteration: %d period:%d \n",CkMyPe(), adaptive_struct.lb_iteration_no, period));
596   adaptive_struct.tentative_period = period;
597   adaptive_struct.final_lb_period = period;
598   // NOTE LDOMAdaptResumeSync(myLDHandle, period);
599   lbdatabase->AdaptResumeSync(period);
600 }
601
602
603 void MetaBalancer::ReceiveIterationNo(int req_no, int local_iter_no) {
604   CmiAssert(CkMyPe() == 0);
605
606   adaptive_struct.global_recv_iter_counter++;
607   if (local_iter_no > adaptive_struct.global_max_iter_no) {
608     adaptive_struct.global_max_iter_no = local_iter_no;
609   }
610
611   int period;
612   if (CkNumPes() == adaptive_struct.global_recv_iter_counter) {
613
614     if (adaptive_struct.global_max_iter_no > adaptive_struct.tentative_max_iter_no) {
615       adaptive_struct.tentative_max_iter_no = adaptive_struct.global_max_iter_no;
616     }
617     period = (adaptive_struct.tentative_period > adaptive_struct.global_max_iter_no) ? adaptive_struct.tentative_period : adaptive_struct.global_max_iter_no + 1;
618     // If no one has gone into load balancing stage, then we can safely change
619     // the period otherwise keep the old period.
620     if (adaptive_struct.global_max_iter_no < adaptive_struct.final_lb_period) {
621       adaptive_struct.tentative_period = period;
622       CkPrintf("Final lb_period CHANGED!%d\n", adaptive_struct.tentative_period);
623     } else {
624       adaptive_struct.tentative_period = adaptive_struct.final_lb_period;
625       CkPrintf("Final lb_period NOT CHANGED!%d\n", adaptive_struct.tentative_period);
626     }
627     thisProxy.LoadBalanceDecisionFinal(req_no, adaptive_struct.tentative_period);
628     adaptive_struct.in_progress = false;
629     adaptive_struct.global_recv_iter_counter = 0;
630   }
631 }
632
633 int MetaBalancer::getPredictedLBPeriod(bool& is_tentative) {
634   // If tentative and final_lb_period are the same, then the decision has been
635   // made but if not, they are in the middle of consensus, hence return the
636   // lease of the two
637   if (adaptive_struct.tentative_period != adaptive_struct.final_lb_period) {
638     is_tentative = true;
639   } else {
640     is_tentative = false;
641   }
642   if (adaptive_struct.tentative_period < adaptive_struct.final_lb_period) {
643     return adaptive_struct.tentative_period;
644    } else {
645      return adaptive_struct.final_lb_period;
646    }
647 }
648
649 // Called by CentralLB to indicate that the LB strategy and migration is in
650 // progress.
651 void MetaBalancer::ResetAdaptive() {
652   adaptive_lbdb.lb_iter_no = -1;
653   lb_in_progress = true;
654 }
655
656 void MetaBalancer::HandleAdaptiveNoObj() {
657 #if EXTRA_FEATURE
658   adaptive_struct.lb_iteration_no++;
659   //CkPrintf("HandleAdaptiveNoObj %d\n", adaptive_struct.lb_iteration_no);
660   thisProxy[0].RegisterNoObjCallback(CkMyPe());
661   TriggerAdaptiveReduction();
662 #endif
663 }
664
665 void MetaBalancer::RegisterNoObjCallback(int index) {
666 #if EXTRA_FEATURE
667   if (lb_in_progress) {
668     lbdb_no_obj_callback.clear();
669     //CkPrintf("Clearing and registering\n");
670     lb_in_progress = false;
671   }
672   lbdb_no_obj_callback.push_back(index);
673   CkPrintf("Registered %d to have no objs.\n", index);
674
675   // If collection has already happened and this is second iteration, then
676   // trigger reduction.
677   if (adaptive_lbdb.lb_iter_no != -1) {
678     //CkPrintf("Collection already started now %d so kick in\n", adaptive_struct.lb_iteration_no);
679     thisProxy[index].TriggerAdaptiveReduction();
680   }
681 #endif
682 }
683
684 void MetaBalancer::TriggerAdaptiveReduction() {
685 #if EXTRA_FEATURE
686   adaptive_struct.lb_iteration_no++;
687   //CkPrintf("Trigger adaptive for %d\n", adaptive_struct.lb_iteration_no);
688   double lb_data[STATS_COUNT];
689   lb_data[0] = adaptive_struct.lb_iteration_no;
690   lb_data[1] = 1;
691   lb_data[2] = 0.0;
692   lb_data[3] = 0.0;
693   lb_data[4] = 0.0;
694   lb_data[5] = 0.0;
695   lb_data[6] = 0.0;
696   lb_data[7] = 0.0;
697
698   // CkPrintf("   [%d] sends total load %lf idle time %lf ratio of idle/load %lf at iter %d\n", CkMyPe(),
699   //     total_load_vec[iteration], idle_time,
700   //     idle_time/total_load_vec[iteration], adaptive_struct.lb_iteration_no);
701
702   CkCallback cb(CkIndex_MetaBalancer::ReceiveMinStats((CkReductionMsg*)NULL), thisProxy[0]);
703   contribute(STATS_COUNT*sizeof(double), lb_data, lbDataCollectionType, cb);
704 #endif
705 }
706
707
708 bool MetaBalancer::isStrategyComm() {
709   return adaptive_struct.doCommStrategy;
710 }
711
712 void MetaBalancer::SetMigrationCost(double lb_migration_cost) {
713   adaptive_struct.lb_migration_cost = lb_migration_cost;
714 }
715
716 void MetaBalancer::SetStrategyCost(double lb_strategy_cost) {
717   adaptive_struct.lb_strategy_cost = lb_strategy_cost;
718 }
719
720 void MetaBalancer::UpdateAfterLBData(int lb, double lb_max, double lb_avg, double
721     local_comm, double remote_comm) {
722   adaptive_struct.last_lb_type = lb;
723   if (lb == 0) {
724     adaptive_struct.greedy_info.max_avg_ratio = lb_max/lb_avg;
725   } else if (lb == 1) {
726     adaptive_struct.refine_info.max_avg_ratio = lb_max/lb_avg;
727   } else if (lb == 2) {
728     adaptive_struct.comm_info.remote_local_ratio = remote_comm/local_comm;
729   } else if (lb == 3) {
730     adaptive_struct.comm_refine_info.remote_local_ratio =
731     remote_comm/local_comm;
732   }
733 }
734
735 void MetaBalancer::UpdateAfterLBData(double max_load, double max_cpu, double
736 avg_load) {
737   if (adaptive_struct.last_lb_type == -1) {
738     adaptive_struct.last_lb_type = 0;
739   }
740   int lb = adaptive_struct.last_lb_type;
741   //CkPrintf("Storing data after lb ratio %lf for lb %d\n", max_load/avg_load, lb);
742   if (lb == 0) {
743     adaptive_struct.greedy_info.max_avg_ratio = max_load/avg_load;
744   } else if (lb == 1) {
745     adaptive_struct.refine_info.max_avg_ratio = max_load/avg_load;
746   } else if (lb == 2) {
747     adaptive_struct.comm_info.max_avg_ratio = max_load/avg_load;
748   } else if (lb == 3) {
749     adaptive_struct.comm_refine_info.max_avg_ratio = max_load/avg_load;
750   }
751 }
752
753 void MetaBalancer::UpdateAfterLBComm(double alpha_beta_to_load) {
754   CkPrintf("Setting alpha beta %lf\n", alpha_beta_to_load);
755   alpha_beta_cost_to_load = alpha_beta_to_load;
756 }
757
758
759 void MetaBalancer::GetPrevLBData(int& lb_type, double& lb_max_avg_ratio, double&
760     remote_local_comm_ratio) {
761   lb_type = adaptive_struct.last_lb_type;
762   lb_max_avg_ratio = 1;
763   remote_local_comm_ratio = 1;
764   GetLBDataForLB(lb_type, lb_max_avg_ratio, remote_local_comm_ratio);
765 }
766
767 void MetaBalancer::GetLBDataForLB(int lb_type, double& lb_max_avg_ratio, double&
768     remote_local_comm_ratio) {
769   if (lb_type == 0) {
770     lb_max_avg_ratio = adaptive_struct.greedy_info.max_avg_ratio;
771   } else if (lb_type == 1) {
772     lb_max_avg_ratio = adaptive_struct.refine_info.max_avg_ratio;
773   } else if (lb_type == 2) {
774     remote_local_comm_ratio = adaptive_struct.comm_info.remote_local_ratio;
775   } else if (lb_type == 3) {
776     remote_local_comm_ratio =
777        adaptive_struct.comm_refine_info.remote_local_ratio;
778   }
779 }
780
781 #include "MetaBalancer.def.h"
782
783 /*@}*/