use cmireduce during recovery
authorXiang Ni <xiangni2@illinois.edu>
Tue, 26 Mar 2013 04:04:56 +0000 (04:04 +0000)
committerXiang Ni <xiangni2@illinois.edu>
Tue, 26 Mar 2013 04:04:56 +0000 (04:04 +0000)
src/arch/mpi/machine.c
src/ck-core/cklocation.C
src/ck-core/ckmemcheckpoint.C

index 1648f95a05b51e9337bbf5f5a57c59999afde393..f844f28eef49814f40cfae226aabd557a597e518 100644 (file)
@@ -300,6 +300,7 @@ void (*signal_int)(int);
 
 static int _thread_provided = -1; /* Indicating MPI thread level */
 static int idleblock = 0;
 
 static int _thread_provided = -1; /* Indicating MPI thread level */
 static int idleblock = 0;
+
 #ifdef CMK_MEM_CHECKPOINT || (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_)
 typedef struct crashedrank{
   int rank;
 #ifdef CMK_MEM_CHECKPOINT || (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_)
 typedef struct crashedrank{
   int rank;
@@ -309,7 +310,6 @@ CpvDeclare(crashedRankList *, crashedRankHdr);
 CpvDeclare(crashedRankList *, crashedRankPtr);
 int isRankDie(int rank);
 #endif
 CpvDeclare(crashedRankList *, crashedRankPtr);
 int isRankDie(int rank);
 #endif
-
 /* A simple list for msgs that have been sent by MPI_Isend */
 typedef struct msg_list {
     char *msg;
 /* A simple list for msgs that have been sent by MPI_Isend */
 typedef struct msg_list {
     char *msg;
@@ -1394,12 +1394,12 @@ void LrtsInit(int *argc, char ***argv, int *numNodes, int *myNodeID) {
 
     if (*myNodeID >= num_workpes) {    /* is spare processor */
       //if isomalloc_sync call mpi_barrier
 
     if (*myNodeID >= num_workpes) {    /* is spare processor */
       //if isomalloc_sync call mpi_barrier
-               if(CmiGetArgFlag(largv,"+isomalloc_sync")){
+      if(CmiGetArgFlag(largv,"+isomalloc_sync")){
           MPI_Barrier(charmComm);
           MPI_Barrier(charmComm);
           MPI_Barrier(charmComm);
           MPI_Barrier(charmComm);
           MPI_Barrier(charmComm);
           MPI_Barrier(charmComm);
           MPI_Barrier(charmComm);
           MPI_Barrier(charmComm);
-           }
+      }
          MPI_Status sts;
       int vals[2];
       MPI_Recv(vals,2,MPI_INT,MPI_ANY_SOURCE,FAIL_TAG, charmComm,&sts);
          MPI_Status sts;
       int vals[2];
       MPI_Recv(vals,2,MPI_INT,MPI_ANY_SOURCE,FAIL_TAG, charmComm,&sts);
@@ -1987,6 +1987,7 @@ int isRankDie(int rank){
   crashedRankList * cur = CpvAccess(crashedRankHdr);
   while(cur!=NULL){
     if(rank == cur->rank){
   crashedRankList * cur = CpvAccess(crashedRankHdr);
   while(cur!=NULL){
     if(rank == cur->rank){
+      printf("die rank %d\n",cur->rank);
       return 1;
     }
     cur = cur->next;
       return 1;
     }
     cur = cur->next;
index 33f41c1e07a063df5c2b35e4f4fa96d9829baf42..99602f8827935d72c5a1fa1ba1914dcc987207a9 100644 (file)
@@ -1302,9 +1302,9 @@ void CkMigratable::recvChkpIter(void * _iter){
                int iter = *(int *)_iter;
                nextChkpIter = iter;
                nextChkpDecided = true;
                int iter = *(int *)_iter;
                nextChkpIter = iter;
                nextChkpDecided = true;
-               if(CkMyPe()==0){
-                 CkPrintf("receive chkp iter %d %d\n",atsync_chkp_iter, nextChkpIter);
-               }
+               //if(CkMyPe()==0){
+               //  CkPrintf("receive chkp iter %d %d\n",atsync_chkp_iter, nextChkpIter);
+               //}
                if(atsync_chkp_iter>nextChkpIter){
                        CkAbort("impossible state in notify\n");
                }
                if(atsync_chkp_iter>nextChkpIter){
                        CkAbort("impossible state in notify\n");
                }
@@ -2088,7 +2088,7 @@ inline void CkLocMgr::springCleaning(void)
     if (rec->isObsolete(nSprings,idx)) {
       //This record is obsolete-- remove it from the table
       DEBK((AA"Cleaning out old record %s\n"AB,idx2str(idx)));
     if (rec->isObsolete(nSprings,idx)) {
       //This record is obsolete-- remove it from the table
       DEBK((AA"Cleaning out old record %s\n"AB,idx2str(idx)));
-      CkPrintf("[%d][%d]Cleaning out old record %s\n",CmiMyPartition(),CkMyPe(),idx2str(idx));
+      //CkPrintf("[%d][%d]Cleaning out old record %s\n",CmiMyPartition(),CkMyPe(),idx2str(idx));
       hash.remove(*(CkArrayIndex *)&idx);
       delete rec;
       it->seek(-1);//retry this hash slot
       hash.remove(*(CkArrayIndex *)&idx);
       delete rec;
       it->seek(-1);//retry this hash slot
index bbfc9d22c2ded47111c7a48b5b3f8c97d22b2210..0635da00871956101a2d0c8b38636dd2f3d6abf1 100644 (file)
@@ -1188,22 +1188,22 @@ void CkMemCheckPT::RollBack(){
     void CkMemCheckPT::restart(int diePe)
     {
 #if CMK_MEM_CHECKPOINT
     void CkMemCheckPT::restart(int diePe)
     {
 #if CMK_MEM_CHECKPOINT
+      thisFailedPe = diePe;
       double curTime = CmiWallTimer();
       double curTime = CmiWallTimer();
-      if (CkMyPe() == diePe){
+      if (CkMyPe() == thisFailedPe){
         restartT = CmiWallTimer();
         CkPrintf("[%d] Process data restored in %f seconds\n", CkMyPe(), curTime - startTime);
       }
       stage = (char*)"resetLB";
       startTime = curTime;
         restartT = CmiWallTimer();
         CkPrintf("[%d] Process data restored in %f seconds\n", CkMyPe(), curTime - startTime);
       }
       stage = (char*)"resetLB";
       startTime = curTime;
-      if (CkMyPe() == diePe)
+      if (CkMyPe() == thisFailedPe)
         CkPrintf("[%d] CkMemCheckPT ----- restart.\n",CkMyPe());
 
         CkPrintf("[%d] CkMemCheckPT ----- restart.\n",CkMyPe());
 
-#if CK_NO_PROC_POOL
-      failed(diePe);   // add into the list of failed pes
-#endif
-      thisFailedPe = diePe;
+//#if CK_NO_PROC_POOL
+//      failed(diePe); // add into the list of failed pes
+//#endif
 
 
-      if (CkMyPe() == diePe) CmiAssert(ckTable.length() == 0);
+//      if (CkMyPe() == diePe) CmiAssert(ckTable.length() == 0);
 
       inRestarting = 1;
 
 
       inRestarting = 1;
 
@@ -1253,7 +1253,7 @@ void CkMemCheckPT::RollBack(){
     void CkMemCheckPT::resetReductionMgr()
     {
       if (CkMyPe() == thisFailedPe) 
     void CkMemCheckPT::resetReductionMgr()
     {
       if (CkMyPe() == thisFailedPe) 
-        CkPrintf("[%d] CkMemCheckPT ----- resetReductionMgr\n",CkMyPe());
+        CkPrintf("[%d] CkMemCheckPT ----- resetReductionMgr at %lf\n",CkMyPe(),CmiWallTimer());
       int numGroups = CkpvAccess(_groupIDTable)->size();
       for(int i=0;i<numGroups;i++) {
         CkGroupID gID = (*CkpvAccess(_groupIDTable))[i];
       int numGroups = CkpvAccess(_groupIDTable)->size();
       for(int i=0;i<numGroups;i++) {
         CkGroupID gID = (*CkpvAccess(_groupIDTable))[i];
@@ -1266,8 +1266,11 @@ void CkMemCheckPT::RollBack(){
       if(CmiNumPartition()==1){
         barrier(CkCallback(CkIndex_CkMemCheckPT::recoverBuddies(), thisProxy));
       }
       if(CmiNumPartition()==1){
         barrier(CkCallback(CkIndex_CkMemCheckPT::recoverBuddies(), thisProxy));
       }
-      else
+      else{
+       if (CkMyPe() == thisFailedPe) 
+         CkPrintf("[%d] CkMemCheckPT ----- resetReductionMgr ends at %lf\n",CkMyPe(),CmiWallTimer());
         barrier(CkCallback(CkReductionTarget(CkMemCheckPT, recoverArrayElements), thisProxy));
         barrier(CkCallback(CkReductionTarget(CkMemCheckPT, recoverArrayElements), thisProxy));
+      }
     }
 
     // recover the lost buddies
     }
 
     // recover the lost buddies
@@ -1358,8 +1361,6 @@ void CkMemCheckPT::RollBack(){
       int len = ckTable.length();
       //CkPrintf("[%d] CkMemCheckPT ----- %s len: %d in %f seconds \n",CkMyPe(), stage, len, curTime-startTime);
       stage = (char *)"recoverArrayElements";
       int len = ckTable.length();
       //CkPrintf("[%d] CkMemCheckPT ----- %s len: %d in %f seconds \n",CkMyPe(), stage, len, curTime-startTime);
       stage = (char *)"recoverArrayElements";
-      if (CkMyPe() == thisFailedPe)
-        CmiPrintf("[%d] CkMemCheckPT ----- %s starts at %f \n",CkMyPe(), stage, curTime);
       startTime = curTime;
       int flag = 0;
       // recover all array elements
       startTime = curTime;
       int flag = 0;
       // recover all array elements
@@ -1429,6 +1430,7 @@ void CkMemCheckPT::RollBack(){
       delete [] gmap;
 #endif
       DEBUGF("[%d] recoverArrayElements restore %d objects\n", CkMyPe(), count);
       delete [] gmap;
 #endif
       DEBUGF("[%d] recoverArrayElements restore %d objects\n", CkMyPe(), count);
+      //if (CkMyPe() == thisFailedPe)
 
       CKLOCMGR_LOOP(mgr->doneInserting(););
 
 
       CKLOCMGR_LOOP(mgr->doneInserting(););
 
@@ -1518,6 +1520,7 @@ void CkMemCheckPT::RollBack(){
         }
       }
       if (CmiMyPe() == BuddyPE(thisFailedPe)) {
         }
       }
       if (CmiMyPe() == BuddyPE(thisFailedPe)) {
+        lastPingTime = CmiWallTimer();
         CcdCallOnCondition(CcdPERIODIC_1s,(CcdVoidFn)pingCheckHandler,NULL);
       }
 #endif
         CcdCallOnCondition(CcdPERIODIC_1s,(CcdVoidFn)pingCheckHandler,NULL);
       }
 #endif
@@ -1635,7 +1638,7 @@ void CkMemCheckPT::RollBack(){
 #if CMK_MEM_CHECKPOINT
 #if CMK_USE_BARRIER
       if(CkMyPe()!=_diePE){
 #if CMK_MEM_CHECKPOINT
 #if CMK_USE_BARRIER
       if(CkMyPe()!=_diePE){
-        printf("restar begin on %d\n",CkMyPe());
+        printf("restart begin on %d at %lf\n",CkMyPe(),CmiWallTimer());
         char *restartmsg = (char*)CmiAlloc(CmiMsgHeaderSizeBytes);
         CmiSetHandler(restartmsg, restartBeginHandlerIdx);
         CmiSyncSendAndFree(_diePE, CmiMsgHeaderSizeBytes, (char *)restartmsg);
         char *restartmsg = (char*)CmiAlloc(CmiMsgHeaderSizeBytes);
         CmiSetHandler(restartmsg, restartBeginHandlerIdx);
         CmiSyncSendAndFree(_diePE, CmiMsgHeaderSizeBytes, (char *)restartmsg);
@@ -1871,7 +1874,17 @@ void CkMemCheckPT::RollBack(){
         _diePE = CpvAccess(_crashedNode);
         char *restartmsg = (char*)CmiAlloc(CmiMsgHeaderSizeBytes);
         CmiSetHandler(restartmsg, restartBeginHandlerIdx);
         _diePE = CpvAccess(_crashedNode);
         char *restartmsg = (char*)CmiAlloc(CmiMsgHeaderSizeBytes);
         CmiSetHandler(restartmsg, restartBeginHandlerIdx);
-        CmiSyncSendAndFree(_diePE, CmiMsgHeaderSizeBytes, (char *)restartmsg);
+        //CmiSyncSendAndFree(_diePE, CmiMsgHeaderSizeBytes, (char *)restartmsg);
+#if CMK_USE_BARRIER
+      //CmiPrintf("before reduce\n");  
+       if(CpvAccess(resilience)==1){
+         CmiSyncSendAndFree(_diePE, CmiMsgHeaderSizeBytes, (char *)restartmsg);
+       }else
+         CmiReduce(restartmsg,CmiMsgHeaderSizeBytes,doNothingMsg);
+      //CmiPrintf("after reduce\n");   
+#else
+       CmiSyncSendAndFree(_diePE, CmiMsgHeaderSizeBytes, (char *)restartmsg);
+#endif 
       }
     }
 
       }
     }
 
@@ -1893,7 +1906,17 @@ void CkMemCheckPT::RollBack(){
         //CmiPrintf("[%d] send to die pe %d\n",CkMyPe(),_diePE);
         char *restartmsg = (char*)CmiAlloc(CmiMsgHeaderSizeBytes);
         CmiSetHandler(restartmsg, restartBeginHandlerIdx);
         //CmiPrintf("[%d] send to die pe %d\n",CkMyPe(),_diePE);
         char *restartmsg = (char*)CmiAlloc(CmiMsgHeaderSizeBytes);
         CmiSetHandler(restartmsg, restartBeginHandlerIdx);
-        CmiSyncSendAndFree(_diePE, CmiMsgHeaderSizeBytes, (char *)restartmsg);
+        //CmiSyncSendAndFree(_diePE, CmiMsgHeaderSizeBytes, (char *)restartmsg);
+#if CMK_USE_BARRIER
+      //CmiPrintf("before reduce\n");  
+       if(CpvAccess(resilience)==1){
+         CmiSyncSendAndFree(_diePE, CmiMsgHeaderSizeBytes, (char *)restartmsg);
+       }else
+         CmiReduce(restartmsg,CmiMsgHeaderSizeBytes,doNothingMsg);
+      //CmiPrintf("after reduce\n");   
+#else
+       CmiSyncSendAndFree(_diePE, CmiMsgHeaderSizeBytes, (char *)restartmsg);
+#endif 
       }
     }
 
       }
     }