Merging
[charm.git] / src / ck-core / cklocation.C
index 82872b3d38e2bba1b35f83f137dfe5ee49d9f4ed..6a297f81543503c6b2be02b2ed064b0831636f0c 100644 (file)
@@ -55,8 +55,6 @@ static const char *idx2str(const CkArrayMessage *m) {
 #   define DEBUG(x)   /**/
 #endif
 
-
-
 #if CMK_LBDB_ON
 /*LBDB object handles are fixed-sized, and not necc.
 the same size as ArrayIndices.
@@ -973,6 +971,10 @@ void CkMigratable::commonInit(void) {
        usesAtSync=CmiFalse;
        usesAutoMeasure=CmiTrue;
        barrierRegistered=CmiFalse;
+  atsync_iteration = -1;
+  //CkPrintf("%s in init and off\n", idx2str(thisIndexMax));
+  local_state = OFF;
+  prev_load = 0.0;
        /*
        FAULT_EVAC
        */
@@ -1072,6 +1074,31 @@ double CkMigratable::getObjTime() {
        return myRec->getObjTime();
 }
 
+void CkMigratable::recvLBPeriod(void *data) {
+  int lb_period = *((int *) data);
+  //CkPrintf("--[pe %s] Received the LB Period %d current iter %d state %d\n",
+   //   idx2str(thisIndexMax), lb_period, atsync_iteration, local_state);
+  if (local_state == PAUSE) {
+    if (atsync_iteration < lb_period) {
+    //  CkPrintf("---[pe %s] pause and decided\n", idx2str(thisIndexMax));
+      local_state = DECIDED;
+      ResumeFromSync();
+      return;
+    }
+   // CkPrintf("---[pe %s] load balance\n", idx2str(thisIndexMax));
+    local_state = LOAD_BALANCE;
+
+    local_state = OFF;
+    atsync_iteration = -1;
+    prev_load = 0.0;
+
+    myRec->getLBDB()->AtLocalBarrier(ldBarrierHandle);
+    return;
+  }
+ // CkPrintf("---[pe %s] decided\n", idx2str(thisIndexMax));
+  local_state = DECIDED;
+}
+
 void CkMigratable::ckFinishConstruction(void)
 {
 //     if ((!usesAtSync) || barrierRegistered) return;
@@ -1086,6 +1113,7 @@ void CkMigratable::ckFinishConstruction(void)
                (LDBarrierFn)staticResumeFromSync,(void*)(this));
        barrierRegistered=CmiTrue;
 }
+
 void CkMigratable::AtSync(int waitForMigration)
 {
        if (!usesAtSync)
@@ -1096,11 +1124,47 @@ void CkMigratable::AtSync(int waitForMigration)
        myRec->AsyncMigrate(!waitForMigration);
        if (waitForMigration) ReadyMigrate(CmiTrue);
        ckFinishConstruction();
-       DEBL((AA"Element %s going to sync\n"AB,idx2str(thisIndexMax)));
-          // model-based load balancing, ask user to provide cpu load
-        if (usesAutoMeasure == CmiFalse) UserSetLBLoad();
-       myRec->getLBDB()->AtLocalBarrier(ldBarrierHandle);
+  DEBL((AA"Element %s going to sync\n"AB,idx2str(thisIndexMax)));
+  // model-based load balancing, ask user to provide cpu load
+  if (usesAutoMeasure == CmiFalse) UserSetLBLoad();
+  //   myRec->getLBDB()->AtLocalBarrier(ldBarrierHandle);
+
+  atsync_iteration++;
+  // CkPrintf("[pe %s] atsync_iter %d && predicted period %d state: %d\n",
+  //     idx2str(thisIndexMax), atsync_iteration,
+  //     myRec->getLBDB()->getPredictedLBPeriod(), local_state);
+  double tmp = prev_load;
+  prev_load = myRec->getObjTime();
+  double current_load = prev_load - tmp;
+
+  if (atsync_iteration != 0) {
+    myRec->getLBDB()->AddLoad(atsync_iteration, current_load);
+  }
+
+//
+//  if (atsync_iteration == 3) {
+//    myRec->getLBDB()->AtLocalBarrier(ldBarrierHandle);
+//    return;
+//  } else {
+//    ResumeFromSync();
+//    return;
+//  }
+
+  if (atsync_iteration < myRec->getLBDB()->getPredictedLBPeriod()) {
+    ResumeFromSync();
+  } else if (local_state == DECIDED) {
+//    CkPrintf("[pe %s] Went to load balance\n", idx2str(thisIndexMax));
+    local_state = LOAD_BALANCE;
+    local_state = OFF;
+    atsync_iteration = -1;
+    prev_load = 0.0;
+    myRec->getLBDB()->AtLocalBarrier(ldBarrierHandle);
+  } else {
+//    CkPrintf("[pe %s] Went to pause state\n", idx2str(thisIndexMax));
+    local_state = PAUSE;
+  }
 }
+
 void CkMigratable::ReadyMigrate(CmiBool ready)
 {
        myRec->ReadyMigrate(ready);
@@ -1266,6 +1330,10 @@ void CkLocRec_local::migrateMe(int toPe) //Leaving this processor
        myLocMgr->emigrate(this,toPe);
 }
 
+void CkLocRec_local::informIdealLBPeriod(int lb_ideal_period) {
+  myLocMgr->informLBPeriod(this, lb_ideal_period);
+}
+
 #if CMK_LBDB_ON
 void CkLocRec_local::startTiming(int ignore_running) {
        if (!ignore_running) running=CmiTrue;
@@ -1484,6 +1552,17 @@ CmiBool CkLocRec_local::deliver(CkArrayMessage *msg,CkDeliver_t type,int opts)
 }
 
 #if CMK_LBDB_ON
+
+void CkLocRec_local::staticAdaptResumeSync(LDObjHandle h, int lb_ideal_period) {
+       CkLocRec_local *el=(CkLocRec_local *)LDObjUserData(h);
+       DEBL((AA"Load balancer wants to migrate %s to %d\n"AB,idx2str(el->idx),dest));
+       el->adaptResumeSync(lb_ideal_period);
+}
+
+void CkLocRec_local::adaptResumeSync(int lb_ideal_period) {
+  informIdealLBPeriod(lb_ideal_period);
+}
+
 void CkLocRec_local::staticMigrate(LDObjHandle h, int dest)
 {
        CkLocRec_local *el=(CkLocRec_local *)LDObjUserData(h);
@@ -2534,6 +2613,16 @@ void CkLocMgr::callMethod(CkLocRec_local *rec,CkMigratable_voidfn_t fn)
        }
 }
 
+/// Call this member function on each element of this location:
+void CkLocMgr::callMethod(CkLocRec_local *rec,CkMigratable_voidfn_arg_t fn,     void * data)
+{
+       int localIdx=rec->getLocalIndex();
+       for (ManagerRec *m=firstManager;m!=NULL;m=m->next) {
+               CkMigratable *el=m->element(localIdx);
+               if (el) (el->* fn)(data);
+       }
+}
+
 /// return a list of migratables in this local record
 void CkLocMgr::migratableList(CkLocRec_local *rec, CkVec<CkMigratable *> &list)
 {
@@ -2636,6 +2725,10 @@ void CkLocMgr::emigrate(CkLocRec_local *rec,int toPe)
        CK_MAGICNUMBER_CHECK
 }
 
+void CkLocMgr::informLBPeriod(CkLocRec_local *rec, int lb_ideal_period) {
+       callMethod(rec,&CkMigratable::recvLBPeriod, (void *)&lb_ideal_period);
+}
+
 /**
   Migrating array element is arriving on this processor.
 */
@@ -2926,6 +3019,8 @@ void CkLocMgr::initLB(CkGroupID lbdbID_)
        myCallbacks.migrate = (LDMigrateFn)CkLocRec_local::staticMigrate;
        myCallbacks.setStats = NULL;
        myCallbacks.queryEstLoad = NULL;
+  myCallbacks.adaptResumeSync =
+      (LDAdaptResumeSyncFn)CkLocRec_local::staticAdaptResumeSync;
        myLBHandle = the_lbdb->RegisterOM(myId,this,myCallbacks);
 
        // Tell the lbdb that I'm registering objects