Fixing more bugs in the bracketed strategies including: 1) destination and source...
authorIsaac Dooley <idooley2@illinois.edu>
Wed, 15 Jul 2009 18:25:55 +0000 (18:25 +0000)
committerIsaac Dooley <idooley2@illinois.edu>
Wed, 15 Jul 2009 18:25:55 +0000 (18:25 +0000)
src/ck-com/ComlibManager.C
src/ck-com/ComlibManager.ci
src/ck-com/ComlibManager.h
src/ck-com/ComlibStrategy.C
src/ck-com/ComlibStrategy.h
src/ck-com/EachToManyMulticastStrategy.C
src/conv-com/convcomlibmanager.C
src/conv-com/convcomlibmanager.h
src/conv-com/convcomlibstrategy.C
src/conv-com/convcomlibstrategy.h

index 2a45b6b95448e8cea3fbd250f9fd786c7ed75dd0..44f233f07b8c17061f53f37b15457bb9f07db43d 100644 (file)
@@ -185,7 +185,7 @@ void ComlibManager::resumeFromSetupBarrier(){
 ***************************************************************************/
 bool ComlibManager::shouldBufferMessagesNow(int instid){
   StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
-  return (!setupComplete) || myEntry->errorMode == ERROR_MODE || myEntry->errorMode == CONFIRM_MODE || myEntry->bufferOutgoing;
+  return (!setupComplete) || myEntry->getErrorMode() == ERROR_MODE || myEntry->getErrorMode() == CONFIRM_MODE || myEntry->bufferOutgoing;
 }
 
 
@@ -204,16 +204,16 @@ void ComlibManager::sendBufferedMessagesAllStrategies(){
    Send all the buffered messages once startup has completed, and we have 
    recovered from all errors.
 ***************************************************************************/
-void ComlibManager::sendBufferedMessages(int instid){
+void ComlibManager::sendBufferedMessages(int instid, int step){
   StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
   
   if(shouldBufferMessagesNow(instid)){
-    ComlibManagerPrintf("[%d] sendBufferedMessages is not flushing buffered messages for strategy %d because shouldBufferMessagesNow()==true\n", CkMyPe(), instid);  
+    ComlibPrintf("[%d] sendBufferedMessages is not flushing buffered messages for strategy %d because shouldBufferMessagesNow()==true step %d\n", CkMyPe(), instid, step);  
   } else if(delayMessageSendBuffer[instid].size() == 0){
-    ComlibManagerPrintf("[%d] sendBufferedMessages: no bufferedmessages to send for strategy %d\n", CkMyPe(), instid);  
+    ComlibPrintf("[%d] sendBufferedMessages: no bufferedmessages to send for strategy %d step %d\n", CkMyPe(), instid, step);  
   } else{
-    ComlibManagerPrintf("[%d] sendBufferedMessages Sending %d buffered messages for instid=%d\n", CkMyPe(), delayMessageSendBuffer[instid].size(), instid);
-
+    ComlibPrintf("[%d] sendBufferedMessages Sending %d buffered messages for instid=%d step %d\n", CkMyPe(), delayMessageSendBuffer[instid].size(), instid, step);
     for (std::set<CharmMessageHolder*>::iterator iter = delayMessageSendBuffer[instid].begin(); iter != delayMessageSendBuffer[instid].end(); ++iter) {
       CharmMessageHolder* cmsg = *iter;
          
@@ -266,52 +266,26 @@ void ComlibManager::beginIteration(int instid, int iteration){
                            
        ComlibManagerPrintf("[%d] beginIteration iter=%d lastKnownIteration=%d  %s %s %s\n", CkMyPe(), iteration, myEntry->lastKnownIteration, myEntry->errorModeString(),  myEntry->errorModeServerString(),  myEntry->discoveryModeString() );
 
-
+       CkAssert(myEntry->getErrorMode() == NORMAL_MODE || myEntry->getErrorMode() == ERROR_MODE);
+       if(CkMyPe()==0)
+         CkAssert(myEntry->getErrorModeServer() == NORMAL_MODE_SERVER ||  myEntry->getErrorModeServer() == ERROR_MODE_SERVER);
 
        if(iteration > myEntry->lastKnownIteration){
              ComlibManagerPrintf("[%d] beginIteration Starting Next Iteration ( # %d )\n", CkMyPe(), iteration);
 
-             // Verify & update errorModeServer:
-             if(CkMyPe()==0){
-               CkAssert(myEntry->errorModeServer == NORMAL_MODE_SERVER || 
-                        myEntry->errorModeServer == STARTUP_MODE_SERVER ||  
-                        myEntry->errorModeServer == ERROR_FIXED_MODE_SERVER);
-               myEntry->errorModeServer = NORMAL_MODE_SERVER;
-             } else {
-               CkAssert(myEntry->errorModeServer == NON_SERVER_MODE_SERVER || 
-                        myEntry->errorModeServer == STARTUP_MODE_SERVER);
-               myEntry->errorModeServer = NON_SERVER_MODE_SERVER;
-             }
-             // Verify & update errorMode:
-             CkAssert(myEntry->errorMode == ERROR_FIXED_MODE || myEntry->errorMode == NORMAL_MODE  );
-             myEntry->errorMode = NORMAL_MODE;   
-             
-             ComlibArrayInfo *myInfo = &dynamic_cast<CharmStrategy*>(myEntry->strategy)->ainfo;
-
-             if(myEntry->lastKnownIteration == STARTUP_ITERATION){
-               // At this point, myEntry->numElements == 0
-               // An error will be detected below, so messages will be buffered.
-               // Since this is startup, bracketedStartErrorRecoveryProcess 
-               // will wait until the comlib layer is setup before continuing
-             } else {
-               // switch to using the newly updated source and destination lists (describing which objects are local to a PE)
-               // This must be done here because the list can only be updated once all sends 
-               // have completed from the iteration during which the error was detected.
-               myInfo->useNewSourceAndDestinations();  
-             }
-
-
              myEntry->lastKnownIteration = iteration;
              myEntry->nBeginItr = 1; // we are the first time to be called this iteration
              myEntry->nEndItr = 0;
              myEntry->nProcSync = 0;
              myEntry->totalEndCounted = 0;
-             myEntry->discoveryMode = NORMAL_DISCOVERY_MODE;
              myEntry->nEndSaved = 0;
                      
-       } else {
+       } else if(iteration == myEntry->lastKnownIteration){
                ComlibManagerPrintf("[%d] beginIteration continuing iteration # %d\n", CkMyPe(), iteration);
                myEntry->nBeginItr++;
+       } else {
+         CkPrintf("[%d] ERROR: ComlibManager::beginIteration iteration=%d < myEntry->lastKnownIteration=%d", iteration, myEntry->lastKnownIteration);
+         CkAbort("[%d] ERROR: ComlibManager::beginIteration iteration < myEntry->lastKnownIteration");
        }
        
        
@@ -345,6 +319,9 @@ void ComlibManager::endIteration(int instid, int step){
        CkAssert(myEntry->nEndItr <= myEntry->nBeginItr);
        CkAssert(step == myEntry->lastKnownIteration);
 
+       CkAssert(myEntry->getErrorMode() == NORMAL_MODE || myEntry->getErrorMode() == ERROR_MODE);
+       if(CkMyPe()==0)
+         CkAssert(myEntry->getErrorModeServer() == NORMAL_MODE_SERVER ||  myEntry->getErrorModeServer() == ERROR_MODE_SERVER);
 
        myEntry->nEndItr++;
        
@@ -376,25 +353,25 @@ void ComlibManager::endIteration(int instid, int step){
 void ComlibManager::bracketedStartErrorRecoveryProcess(int instid, int step){  
   CProxy_ComlibManager myProxy(thisgroup);
   StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);     
+  CkAssert(step >= myEntry->lastKnownIteration);
 
-  
 
   if(converseManager->isReady(instid)){
     ComlibManagerPrintf("[%d] bracketedStartErrorRecoveryProcess(instid=%d step=%d) %s %s %s\n", CkMyPe(), instid, step, myEntry->errorModeString(),  myEntry->errorModeServerString(),  myEntry->discoveryModeString() );
 
     CkAssert(myEntry->strategy != NULL);
-    CkAssert(myEntry->errorMode == NORMAL_MODE || myEntry->errorMode == ERROR_MODE);
+    CkAssert(myEntry->getErrorMode() == NORMAL_MODE || myEntry->getErrorMode() == ERROR_MODE);
          
     if (!myEntry->strategy->isBracketed()) {
       CkPrintf("[%d] endIteration called unecessarily for a non-bracketed strategy\n", CkMyPe());
       return;
     }
        
-    if (myEntry->errorMode == NORMAL_MODE) {
+    if (myEntry->getErrorMode() == NORMAL_MODE) {
       ComlibManagerPrintf("[%d] bracketedStartErrorRecoveryProcess()\n", CkMyPe());
       myEntry->nEndSaved = myEntry->nEndItr;
       myProxy[0].bracketedReceiveCount(instid, CkMyPe(), myEntry->nEndSaved, 1, step);
-      myEntry->errorMode = ERROR_MODE;
+      myEntry->setErrorMode(ERROR_MODE);
       bracketedStartDiscovery(instid);
     } else {
       // Send the updated count
@@ -421,17 +398,17 @@ void ComlibManager::bracketedErrorDetected(int instid, int step) {
        
        bracketedCatchUpPE(instid,step);
        CkAssert(step == myEntry->lastKnownIteration);
-       CkAssert(myEntry->errorMode == NORMAL_MODE || myEntry->errorMode == ERROR_MODE);
+       CkAssert(myEntry->getErrorMode() == NORMAL_MODE || myEntry->getErrorMode() == ERROR_MODE);
 
        ComlibManagerPrintf("[%d] bracketedErrorDetected()\n", CkMyPe());
 
-       if (myEntry->errorMode == NORMAL_MODE) {
+       if (myEntry->getErrorMode() == NORMAL_MODE) {
          // save the value we are sending to bracketedReceiveCount
          myEntry->nEndSaved = myEntry->nEndItr; // save the value we are sending to bracketedReceiveCount
          CProxy_ComlibManager myProxy(thisgroup);
          myProxy[0].bracketedReceiveCount(instid, CkMyPe(), myEntry->nEndSaved, 1, step);
          bracketedStartDiscovery(instid);
-         myEntry->errorMode = ERROR_MODE;
+         myEntry->setErrorMode(ERROR_MODE);
 
        } else { // ERROR_MODE
          // If we have an update count, send it
@@ -449,13 +426,13 @@ void ComlibManager::bracketedErrorDetected(int instid, int step) {
 
 /// Invoked on all processors. After processor 0 has a count match, it sends out
 /// a broadcast on this entry method to get a confirmation from all others.
-void ComlibManager::bracketedConfirmCount(int instid) {
+void ComlibManager::bracketedConfirmCount(int instid, int step) {
         StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
-       CkAssert(myEntry->errorMode == ERROR_MODE);
-       myEntry->errorMode = CONFIRM_MODE;
+       CkAssert(myEntry->getErrorMode() == ERROR_MODE);
+       myEntry->setErrorMode(CONFIRM_MODE);
        CProxy_ComlibManager myProxy(thisgroup);
        ComlibManagerPrintf("[%d] bracketedConfirmCount\n", CkMyPe());
-       myProxy[0].bracketedCountConfirmed(instid, myEntry->nEndSaved, myEntry->lastKnownIteration);
+       myProxy[0].bracketedCountConfirmed(instid, myEntry->nEndSaved, step);
 }
 
 /// Invoked on processor 0 upon a request to confirm the count. If the count is
@@ -466,7 +443,7 @@ void ComlibManager::bracketedCountConfirmed(int instid, int count, int step) {
        CkAssert(CkMyPe() == 0);
        // Advance PE0 to current step if we had no local objects
        bracketedCatchUpPE(instid, step);
-       CkAssert(myEntry->errorModeServer == CONFIRM_MODE_SERVER);
+       CkAssert(myEntry->getErrorModeServer() == CONFIRM_MODE_SERVER);
        CkAssert(step == myEntry->lastKnownIteration);
 
        myEntry->total += count;
@@ -478,10 +455,10 @@ void ComlibManager::bracketedCountConfirmed(int instid, int count, int step) {
 
                CkAssert(myEntry->total == myEntry->totalEndCounted);
                  
-               CProxy_ComlibManager(thisgroup).bracketedReceiveNewCount(instid);
-               myEntry->errorModeServer = ERROR_FIXED_MODE_SERVER;
+               CProxy_ComlibManager(thisgroup).bracketedReceiveNewCount(instid, step);
+               myEntry->setErrorModeServer(ERROR_FIXED_MODE_SERVER);
                  
-               myEntry->total = 0;     
+               myEntry->total = 0;
        }
 
 }
@@ -505,18 +482,13 @@ void ComlibManager::bracketedCatchUpPE(int instid, int step){
     myEntry->nProcSync = 0;
     myEntry->totalEndCounted = 0;
     myEntry->nEndSaved = 0;
-    myEntry->errorMode = NORMAL_MODE;
-    myEntry->discoveryMode = NORMAL_DISCOVERY_MODE;
+    myEntry->setErrorMode(NORMAL_MODE);
+    myEntry->setDiscoveryMode(NORMAL_DISCOVERY_MODE);
 
     if(CkMyPe()==0){
-      CkAssert(myEntry->errorModeServer == NORMAL_MODE_SERVER || myEntry->errorModeServer == ERROR_FIXED_MODE_SERVER);
-      ComlibManagerPrintf("[%d] NORMAL_MODE_SERVER *******************AAAAAAAAAAAA**********************\n", CkMyPe());
-      myEntry->errorModeServer = NORMAL_MODE_SERVER;
-    } else {
-      CkAssert(myEntry->errorModeServer == NON_SERVER_MODE_SERVER || myEntry->errorModeServer == STARTUP_MODE_SERVER);
-      myEntry->errorModeServer == NON_SERVER_MODE_SERVER;
-    }
-    
+      CkAssert(myEntry->getErrorModeServer() == NORMAL_MODE_SERVER || myEntry->getErrorModeServer() == ERROR_FIXED_MODE_SERVER);
+      myEntry->setErrorModeServer(NORMAL_MODE_SERVER);
+    }    
     
   }
 }
@@ -532,11 +504,12 @@ void ComlibManager::bracketedCatchUpPE(int instid, int step){
 /// be called there.
 void ComlibManager::bracketedReceiveCount(int instid, int pe, int count, int isFirst, int step) {
        StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
-       CkAssert(CkMyPe() == 0);
-       CkAssert(myEntry->errorModeServer == NORMAL_MODE_SERVER || myEntry->errorModeServer == ERROR_MODE_SERVER );
-
-       ComlibManagerPrintf("[%d] bracketedReceiveCount step=%d \n", CkMyPe(), step);
+       ComlibPrintf("[%d] bracketedReceiveCount begins step=%d lastKnownIteration=%d totalEndCounted=%d count=%d\n", CkMyPe(), step, myEntry->lastKnownIteration, myEntry->totalEndCounted, count);
+       CkAssert(step >= myEntry->lastKnownIteration);
 
+       CkAssert(CkMyPe() == 0);
+       CkAssert(myEntry->getErrorModeServer() == NORMAL_MODE_SERVER || myEntry->getErrorModeServer() == ERROR_MODE_SERVER );
+       
        
        // Advance PE0 to current step if we had no local objects
        bracketedCatchUpPE(instid, step);
@@ -546,31 +519,31 @@ void ComlibManager::bracketedReceiveCount(int instid, int pe, int count, int isF
 
        myEntry->totalEndCounted += count;
 
-       ComlibManagerPrintf("[%d] bracketedReceiveCount step=%d totalEndCounted=%d count=%d\n", CkMyPe(), step, myEntry->totalEndCounted, count);
 
        
        myEntry->nProcSync += isFirst; // isFirst is 1 the first time a processor send a message,
        CkAssert(myEntry->nProcSync <= CkNumPes());
        
-       if (myEntry->errorModeServer == NORMAL_MODE_SERVER) { 
+       if (myEntry->getErrorModeServer() == NORMAL_MODE_SERVER) { 
                // first time this is called
                CkAssert(myEntry->nProcSync == 1);
                CProxy_ComlibManager(thisgroup).bracketedErrorDetected(instid, step);
-               myEntry->errorModeServer = ERROR_MODE_SERVER;
+               myEntry->setErrorModeServer(ERROR_MODE_SERVER);
                ComlibManagerPrintf("[%d] bracketedReceiveCount first time\n", CkMyPe());
-       } else { // ERROR_MODE
-       
-         CharmStrategy* s = dynamic_cast<CharmStrategy*>(myEntry->strategy);
-         ComlibArrayInfo ainfo = s->ainfo;
-         int totalsrc =  ainfo.getTotalSrc() ;
+       }
+
+
+
+       CharmStrategy* s = dynamic_cast<CharmStrategy*>(myEntry->strategy);
+       ComlibArrayInfo ainfo = s->ainfo;
+       int totalsrc =  ainfo.getTotalSrc() ;
          
-         if(myEntry->nProcSync == CkNumPes() && myEntry->totalEndCounted == totalsrc) {
-           // ok, we received notifications from all PEs and all objects have called endIteration
-           myEntry->errorModeServer = CONFIRM_MODE_SERVER
-           ComlibManagerPrintf("[%d] bracketedReceiveCount errorModeServer is now CONFIRM_MODE calling bracketedConfirmCount totalsrc=%d\n", CkMyPe(), (int)totalsrc);
-           CProxy_ComlibManager(thisgroup).bracketedConfirmCount(instid);
+       if(myEntry->nProcSync == CkNumPes() && myEntry->totalEndCounted == totalsrc) {
+         // ok, we received notifications from all PEs and all objects have called endIteration
+         myEntry->setErrorModeServer(CONFIRM_MODE_SERVER)
+         ComlibManagerPrintf("[%d] bracketedReceiveCount errorModeServer is now CONFIRM_MODE calling bracketedConfirmCount totalsrc=%d\n", CkMyPe(), (int)totalsrc);
+         CProxy_ComlibManager(thisgroup).bracketedConfirmCount(instid, step);
                        
-         }
        }
                
 }
@@ -580,18 +553,18 @@ void ComlibManager::bracketedReceiveCount(int instid, int pe, int count, int isF
 /// the number of emements involved in the bracketed operation, processor 0
 /// sends a broadcast to acknowledge the success. 
 /// The strategy is disabled until a new Pe list is received.
-void ComlibManager::bracketedReceiveNewCount(int instid) {
+void ComlibManager::bracketedReceiveNewCount(int instid, int step) {
        StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
-       CkAssert(myEntry->errorMode == CONFIRM_MODE);
+       CkAssert(myEntry->getErrorMode() == CONFIRM_MODE);
 
-       myEntry->errorMode = ERROR_FIXED_MODE;
+       myEntry->setErrorMode(ERROR_FIXED_MODE);
        
        myEntry->nEndItr -= myEntry->nEndSaved;
        myEntry->nBeginItr -= myEntry->nEndSaved;
 
        myEntry->nEndSaved = 0;
 
-       bracketedFinalBarrier(instid);
+       bracketedFinalBarrier(instid, step);
 }
 
 
@@ -603,10 +576,10 @@ void ComlibManager::bracketedReceiveNewCount(int instid) {
 /// together with the counting process, otherwise the strategy is updated at an
 /// undetermined state. In future it may be useful to implement the discovery
 /// process alone.
-void ComlibManager::bracketedReceiveNewPeList(int instid, int *count) {
+void ComlibManager::bracketedReceiveNewPeList(int instid, int step, int *count) {
        StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
-       CkAssert(myEntry->discoveryMode == STARTED_DISCOVERY_MODE);
-       myEntry->discoveryMode = FINISHED_DISCOVERY_MODE;
+       CkAssert(myEntry->getDiscoveryMode() == STARTED_DISCOVERY_MODE);
+       myEntry->setDiscoveryMode(FINISHED_DISCOVERY_MODE);
 
        myEntry->strategy->bracketedUpdatePeKnowledge(count);
        
@@ -618,7 +591,7 @@ void ComlibManager::bracketedReceiveNewPeList(int instid, int *count) {
        ComlibManagerPrintf("[%d] delayMessageSendBuffer[%d].size()=%d\n",CkMyPe(), instid, delayMessageSendBuffer[instid].size() );
        ComlibManagerPrintf("[%d] delayMessageSendBuffer[%d].size()=%d\n", CkMyPe(), instid, delayMessageSendBuffer[instid].size());
                
-       bracketedFinalBarrier(instid);
+       bracketedFinalBarrier(instid, step);
 }
 
 
@@ -629,14 +602,21 @@ void ComlibManager::bracketedReceiveNewPeList(int instid, int *count) {
     ComlibManager::bracketedReceiveNewPeList is called as an array broadcast to thisProxy, 
     so every PE will call this method.
  */
-void ComlibManager::bracketedFinalBarrier(int instid) {
+void ComlibManager::bracketedFinalBarrier(int instid, int step) {
   StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
   ComlibManagerPrintf("[%d] ComlibManager::bracketedFinalBarrier %s %s %s\n", CkMyPe(), myEntry->errorModeString(),  myEntry->errorModeServerString(),  myEntry->discoveryModeString() );
 
+  
+  if (myEntry->getDiscoveryMode() == FINISHED_DISCOVERY_MODE &&  myEntry->getErrorMode() == ERROR_FIXED_MODE) {       
+    myEntry->setDiscoveryMode(NORMAL_DISCOVERY_MODE);
+    myEntry->setErrorMode(NORMAL_MODE);
 
-  if (myEntry->discoveryMode == FINISHED_DISCOVERY_MODE &&  myEntry->errorMode == ERROR_FIXED_MODE) {    
+    // Update destination and source element lists for use in the next step
+    ComlibArrayInfo *myInfo = &dynamic_cast<CharmStrategy*>(myEntry->strategy)->ainfo;
+    myInfo->useNewSourceAndDestinations();     
+    
     CProxy_ComlibManager myProxy(thisgroup);
-    myProxy[0].bracketedReleaseCount(instid);
+    myProxy[0].bracketedReleaseCount(instid, step);
   }
 }
 
@@ -644,17 +624,18 @@ void ComlibManager::bracketedFinalBarrier(int instid) {
 /** 
     Once all PEs report here, we will allow them to release the buffered messages from this iteration
  */
-void ComlibManager::bracketedReleaseCount(int instid) {
-    ComlibManagerPrintf("[%d] ComlibManager::bracketedReleaseCount\n", CkMyPe());
+void ComlibManager::bracketedReleaseCount(int instid, int step) {
 
     StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
+    ComlibPrintf("[%d] ComlibManager::bracketedReleaseCount myEntry->numBufferReleaseReady was %d\n", CkMyPe(), myEntry->numBufferReleaseReady);
+
     CkAssert(CkMyPe() == 0);
-    CkAssert(myEntry->errorModeServer == ERROR_FIXED_MODE_SERVER);
+    CkAssert(myEntry->getErrorModeServer() == ERROR_FIXED_MODE_SERVER);
     
     myEntry->numBufferReleaseReady++;
     if(myEntry->numBufferReleaseReady == CkNumPes()) {
-      myEntry->errorModeServer = NORMAL_MODE_SERVER;
-      CProxy_ComlibManager(thisgroup).bracketedReleaseBufferedMessages(instid);
+      myEntry->setErrorModeServer(NORMAL_MODE_SERVER);
+      CProxy_ComlibManager(thisgroup).bracketedReleaseBufferedMessages(instid, step);
       myEntry->numBufferReleaseReady = 0;
     }
 }
@@ -662,13 +643,16 @@ void ComlibManager::bracketedReleaseCount(int instid) {
 /** 
     Release any buffered messages.
  */
-void ComlibManager::bracketedReleaseBufferedMessages(int instid) {
-  ComlibManagerPrintf("[%d] ComlibManager::bracketedReleaseBufferedMessages\n", CkMyPe());
-
+void ComlibManager::bracketedReleaseBufferedMessages(int instid, int step) {
+  ComlibManagerPrintf("[%d] ComlibManager::bracketedReleaseBufferedMessages step=%d\n", CkMyPe(), step);
   StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
+
+  CkAssert(myEntry->getErrorModeServer() == NORMAL_MODE_SERVER);
+  CkAssert(myEntry->getErrorMode() == NORMAL_MODE);
+  CkAssert(myEntry->getDiscoveryMode() == NORMAL_DISCOVERY_MODE);
   
   myEntry->bufferOutgoing = 0;
-  sendBufferedMessages(instid);
+  sendBufferedMessages(instid, step);
   
   converseManager->doneInserting(instid);
 }
@@ -692,15 +676,11 @@ void ComlibManager::bracketedReleaseBufferedMessages(int instid) {
     it is still here. If the array element has migrated away, then the
     bracketedDiscover() method is called on the new PE.
 
-    the new source and destination element arrays will be empty at this point,
-    so myInfo->addNewDestinationList(e); will add a new record that
-    will be used for future iterations.
-
 */
 void ComlibManager::bracketedStartDiscovery(int instid) {
        StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
-       CkAssert(myEntry->discoveryMode == NORMAL_DISCOVERY_MODE);
-       myEntry->discoveryMode = STARTED_DISCOVERY_MODE;
+       CkAssert(myEntry->getDiscoveryMode() == NORMAL_DISCOVERY_MODE);
+       myEntry->setDiscoveryMode(STARTED_DISCOVERY_MODE);
        ComlibArrayInfo *myInfo = &dynamic_cast<CharmStrategy*>(myEntry->strategy)->ainfo;
        const CProxy_ComlibManager myProxy(thisgroup);
 
@@ -710,7 +690,6 @@ void ComlibManager::bracketedStartDiscovery(int instid) {
        int countDest = 0;
 
        if (myInfo->isSourceArray()) {
-         CkAssert(myInfo->newSourceListSize() == 0);
 
          const CkVec<CkArrayIndexMax> & srcElements = myInfo->getSourceElements();
          const int nelem = srcElements.size();
@@ -731,7 +710,7 @@ void ComlibManager::bracketedStartDiscovery(int instid) {
        }
 
        if (myInfo->isDestinationArray()) {
-         CkAssert(myInfo->newDestinationListSize() == 0);
+//       CkAssert(myInfo->newDestinationListSize() == 0);
 
          const CkVec<CkArrayIndexMax> & destElements = myInfo->getDestinationElements();
          const int nelem = destElements.size();
@@ -745,6 +724,7 @@ void ComlibManager::bracketedStartDiscovery(int instid) {
              myInfo->addNewLocalDestination(destElements[i]);
            }
            else {
+             ComlibPrintf("[%d] destination element %d is no longer local\n", CkMyPe(), (int)destElements[i].data()[0]);
              myProxy[pe].bracketedDiscover(instid, aid, destElements[i], false);
            }
          }
@@ -861,7 +841,7 @@ void ComlibManager::bracketedContributeDiscovery(int instid, int pe, int nsrc, i
                                myEntry->peList[CkNumPes()+1], myInfo->getTotalDest() );
                
                printPeList("bracketedContributeDiscovery peList=", myEntry->peList);
-               myProxy.bracketedReceiveNewPeList(instid, myEntry->peList);
+               myProxy.bracketedReceiveNewPeList(instid, step, myEntry->peList);
                delete myEntry->peList;
                myEntry->peList = NULL;
        } else {
@@ -945,7 +925,9 @@ void ComlibManager::ArraySend(CkDelegateData *pd,int ep, void *msg,
        
        if(shouldBufferMessagesNow(instid)){
          delayMessageSendBuffer[instid].insert(cmsg);
-         ComlibManagerPrintf("[%d] ComlibManager::ArraySend BUFFERED OUTGOING: now buffer contains %d messages\n",CkMyPe(), delayMessageSendBuffer[instid].size() );
+         StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
+         int step = myEntry->lastKnownIteration;
+         ComlibManagerPrintf("[%d] ComlibManager::ArraySend BUFFERED OUTGOING: now buffer contains %d messages step=%d\n",CkMyPe(), delayMessageSendBuffer[instid].size(), step);
        } else {
          ComlibPrintf("ComlibManager::ArraySend NOT BUFFERING inserting message into strategy %d\n",instid);
          
@@ -999,7 +981,9 @@ void ComlibManager::GroupSend(CkDelegateData *pd,int ep, void *msg, int onPE, Ck
        CharmMessageHolder *cmsg = new CharmMessageHolder((char *)msg, dest_proc, CMH_GROUPSEND); 
 
        if(shouldBufferMessagesNow(instid)){
-         ComlibPrintf("ComlibManager::GroupSend Buffering message for %d\n",instid);
+         StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
+         int step = myEntry->lastKnownIteration;
+         ComlibPrintf("ComlibManager::GroupSend Buffering message for %d step=%d\n",instid, step);
          delayMessageSendBuffer[instid].insert(cmsg);
        } else {
          ComlibPrintf("ComlibManager::GroupSend inserting message into strategy %d\n",instid);
@@ -1113,7 +1097,7 @@ void ComlibManager::multicast(CharmMessageHolder *cmsg, int instid) {
 
 #if DEBUG
        StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
-       ComlibPrintf("[%d] multicast setupComplete=%d errorMode=%d errorModeServer=%d\n", CkMyPe(), setupComplete, myEntry->errorMode, myEntry->errorModeServer);
+       ComlibPrintf("[%d] multicast setupComplete=%d %s %s\n", CkMyPe(), setupComplete, myEntry->getErrorModeString(), myEntry->getErrorModeServerString());
 #endif
  
        env->setUsed(0);
@@ -1121,6 +1105,9 @@ void ComlibManager::multicast(CharmMessageHolder *cmsg, int instid) {
        
        if(shouldBufferMessagesNow(instid)){
          cmsg->saveCopyOf_sec_id();
+         StrategyTableEntry *myEntry = converseManager->getStrategyTable(instid);
+         int step = myEntry->lastKnownIteration;
+         ComlibPrintf("[%d] ComlibManager::multicast Buffering message for %d lastKnownIteration=%d\n", CkMyPe(),instid, step);
          delayMessageSendBuffer[instid].insert(cmsg);
        } else {
          converseManager->insertMessage(cmsg, instid);
index d0219e974903f5034a7d15478d9e89dc81b34002..b6b34a78fafcf0832a7310ea57ac42301ffa7c9f 100644 (file)
@@ -19,14 +19,14 @@ module comlib {
     entry void bracketedStartErrorRecoveryProcess(int instid, int step);        
     entry void bracketedErrorDetected(int instid, int step);
     entry void bracketedReceiveCount(int instid, int pe, int count, int isFirst, int step);
-    entry void bracketedConfirmCount(int instid);
+    entry void bracketedConfirmCount(int instid, int step);
     entry void bracketedCountConfirmed(int instid, int count, int step);
-    entry void bracketedReceiveNewCount(int instid);
+    entry void bracketedReceiveNewCount(int instid, int step);
     entry void bracketedDiscover(int instid, CkArrayID aid, CkArrayIndexMax &idx, int isSrc);
     entry void bracketedContributeDiscovery(int instid, int pe, int nsrc, int ndest, int step);
-    entry void bracketedReceiveNewPeList(int instid, int count[CkNumPes()]);
-    entry void bracketedReleaseCount(int instid);
-    entry void bracketedReleaseBufferedMessages(int instid);
+    entry void bracketedReceiveNewPeList(int instid, int step, int count[CkNumPes()]);
+    entry void bracketedReleaseCount(int instid, int step);
+    entry void bracketedReleaseBufferedMessages(int instid, int step);
 
     entry void collectStats(ComlibLocalStats s, int src);
   }
index b28f45d0c38cf1025609ec1acea9d9c382e042d8..a4604697cd4657134fd551c9687d7f9d8b37ea0c 100644 (file)
@@ -288,7 +288,7 @@ class ComlibManager: public CkDelegateMgr {
 
     
     bool shouldBufferMessagesNow(int instid);
-    void sendBufferedMessages(int instid);
+    void sendBufferedMessages(int instid, int step=-1);
     void sendBufferedMessagesAllStrategies();
 
  public:
@@ -338,14 +338,14 @@ class ComlibManager: public CkDelegateMgr {
     void bracketedReceiveCount(int instid, int pe, int count, int isFirst, int step);
     void bracketedStartErrorRecoveryProcess(int instid, int step);
     void bracketedErrorDetected(int instid, int step);
-    void bracketedConfirmCount(int instid);
+    void bracketedConfirmCount(int instid, int step);
     void bracketedCountConfirmed(int instid, int count, int step);
-    void bracketedReceiveNewCount(int instid);
-    void bracketedReceiveNewPeList(int instid, int *count);
+    void bracketedReceiveNewCount(int instid, int step);
+    void bracketedReceiveNewPeList(int instid, int step, int *count);
 
-    void bracketedFinalBarrier(int instid);
-    void bracketedReleaseCount(int instid);
-    void bracketedReleaseBufferedMessages(int instid);
+    void bracketedFinalBarrier(int instid, int step);
+    void bracketedReleaseCount(int instid, int step);
+    void bracketedReleaseBufferedMessages(int instid, int step);
 
     void bracketedStartDiscovery(int instid);
     void bracketedDiscover(int instid, CkArrayID aid, CkArrayIndexMax &idx, int isSrc);
index a13ccf51cfdadbdb8b3a6c80cd30cf8bf7d516a6..3bc767ae5f02b1be1320f7e18311de50ad032931 100644 (file)
@@ -381,139 +381,7 @@ int *ComlibArrayInfo::getCombinedCountList() {
   return result;
 }
 
-/*
-//Get the list of destination processors
-void ComlibArrayInfo::getDestinationPeList(int *&destpelist, int &ndestpes) {
-    
-    int count = 0, acount =0;
-    
-    //Destination has not been set
-    if(nDestIndices < 0) {
-        destpelist = 0;
-        ndestpes = 0;
-        return;
-    }
-
-    //Create an array of size CkNumPes()
-    //Inefficient in space
-    ndestpes = CkNumPes();
-    destpelist = new int[ndestpes];
-
-    memset(destpelist, 0, ndestpes * sizeof(int));    
-
-    if(nDestIndices == 0){
-        for(count =0; count < CkNumPes(); count ++) 
-            destpelist[count] = count;             
-        return;
-    }
-
-    ndestpes = 0;
-    CkArray *amgr = CkArrayID::CkLocalBranch(dest_aid);
-
-    //Find the last known processors of the array elements
-    for(acount = 0; acount < nDestIndices; acount++) {
-
-      //int p = ComlibGetLastKnown(dest_aid, dest_elements[acount]); 
-        int p = amgr->lastKnown(dest_elements[acount]);
-        
-        for(count = 0; count < ndestpes; count ++)
-            if(destpelist[count] == p)
-                break;
-        
-        if(count == ndestpes) {
-            destpelist[ndestpes ++] = p; 
-        }       
-    }                            
-}
-
-void ComlibArrayInfo::getSourcePeList(int *&srcpelist, int &nsrcpes) {
-    
-    int count = 0, acount =0;
-
-    if(nSrcIndices < 0) {
-        srcpelist = 0;
-        nsrcpes = 0;
-        return;
-    }
-
-    nsrcpes = CkNumPes();
-    srcpelist = new int[nsrcpes];
-
-    memset(srcpelist, 0, nsrcpes * sizeof(int));    
 
-    if(nSrcIndices == 0){
-        for(count =0; count < CkNumPes(); count ++) 
-            srcpelist[count] = count;             
-        return;
-    }
-
-    nsrcpes = 0;
-    CkArray *amgr = CkArrayID::CkLocalBranch(src_aid);
-
-    for(acount = 0; acount < nSrcIndices; acount++) {
-        
-      //int p = ComlibGetLastKnown(src_aid, src_elements[acount]); 
-        int p = amgr->lastKnown(src_elements[acount]);
-        
-        for(count = 0; count < nsrcpes; count ++)
-            if(srcpelist[count] == p)
-                break;
-        
-        if(count == nsrcpes) {
-            srcpelist[nsrcpes ++] = p; 
-        }       
-    }                            
-}
-
-void ComlibArrayInfo::getCombinedPeList(int *&pelist, int &npes) {
-
-    int count = 0;        
-    pelist = 0;
-    npes = 0;
-    
-    //Both arrays empty;
-    //Sanity check, this should really not happen
-    if(nSrcIndices < 0 && nDestIndices < 0) {
-        CkAbort("Arrays have not been set\n");
-        return;
-    }
-    
-    //One of them is the entire array Hence set the number of
-    //processors to all Currently does not work for the case where
-    //number of array elements less than number of processors
-    //Will fix it later!
-    if(nSrcIndices == 0 || nDestIndices == 0) {
-        npes = CkNumPes();        
-        pelist = new int[npes];
-        for(count = 0; count < CkNumPes(); count ++) 
-            pelist[count] = count;                         
-    }
-    else {
-        getDestinationPeList(pelist, npes);
-        
-        //Destination has not been set
-        //Strategy does not care about destination
-        //Is an error case
-        if(npes == 0)
-            pelist = new int[CkNumPes()];
-        
-       CkArray *amgr = CkArrayID::CkLocalBranch(src_aid);
-
-        //Add source processors to the destination processors
-        //already obtained
-        for(int acount = 0; acount < nSrcIndices; acount++) {
-         //int p = ComlibGetLastKnown(src_aid, src_elements[acount]);
-           int p = amgr->lastKnown(src_elements[acount]);
-
-            for(count = 0; count < npes; count ++)
-                if(pelist[count] == p)
-                    break;
-            if(count == npes)
-                pelist[npes ++] = p;
-        }                        
-    }
-}
-*/
 
 /**  Broadcast the message to all local elements (as listed in dest_elements) */
 void ComlibArrayInfo::localBroadcast(envelope *env) {
index c95f70ed0438aedd5e6277d018784800d6c1f0a8..10c0845f02b5222a4ef499e2d56bb4dee32ca193 100644 (file)
@@ -285,18 +285,17 @@ class ComlibArrayInfo {
   
     /// Switch to using the new destination and source lists if the previous iteration found an error and constructed the new list
     void useNewSourceAndDestinations() {
-      if(new_dest_elements.size() > 0) {
-       dest_elements.removeAll();
-       dest_elements = new_dest_elements;
-       CkAssert(dest_elements.size() == new_dest_elements.size());
-       new_dest_elements.removeAll();
-      }
-      if(new_src_elements.size() > 0) {
-       src_elements.removeAll();
-       src_elements = new_src_elements;
-       CkAssert(src_elements.size() == new_src_elements.size());
-       new_src_elements.removeAll();
-      }
+      ComlibPrintf("[%d] useNewSourceAndDestinations previously had %d elements, now have %d elements\n", CkMyPe(), dest_elements.size(), new_src_elements.size() );
+      dest_elements.removeAll();
+      dest_elements = new_dest_elements;
+      CkAssert(dest_elements.size() == new_dest_elements.size());
+      new_dest_elements.removeAll();
+      src_elements.removeAll();
+      src_elements = new_src_elements;
+      CkAssert(src_elements.size() == new_src_elements.size());
+      new_src_elements.removeAll();
+      if(com_debug)
+       printDestElementList();
     }
 
     int newDestinationListSize(){ return new_dest_elements.size(); }
index c71d29e8f223cdf5adcf449603eea02fac2dda29..75d87456dac786d4ee58a64525103691beb55a0b 100644 (file)
@@ -33,7 +33,7 @@ EachToManyMulticastStrategy::EachToManyMulticastStrategy(int substrategy,
        //ginfo.getDestinationGroup(gid, destpelist, ndestpes);
        //ginfo.getCombinedPeList(pelist, npes);
 
-       commonInit(ginfo.getCombinedCountList());
+       commonInit(ginfo.getCombinedCountList()); // The array returned by getCombinedCountList is deleted inside commonInit
 }
 
 //Array Constructor
index d7c58256f652efd270a908041ff518cd079a30cb..c58ae310dcc8870c75d7bfd27be5bb1b2b5e12b4 100644 (file)
@@ -55,7 +55,9 @@ void *strategyHandler(void *msg) {
     return NULL;
 }
 
-/* Strategies synchronization process:
+/** @file
+ * 
+ *  Strategies synchronization process:
  *
  * 1) the strategies are inserted on processor 0 (and possibly in other
  *    processors with the same order. The strategies are marked as "new"
index 95d6f6d9edb7eccaafc2e4f99c5b80cb984a17e9..66e8a92b4babaca9ea047c586d42f4d8823d766f 100644 (file)
@@ -90,7 +90,7 @@ class ConvComlibManager {
     strategyTable[loc].isNew = 0;
     strategyTable[loc].isInSync = 1;
   }
-  inline int getErrorMode(int loc) {return strategyTable[loc].errorMode;}
+/*   inline int getErrorMode(int loc) {return strategyTable[loc].errorMode;} */
   inline CkQ<MessageHolder*> *getTmpList(int loc) {return &strategyTable[loc].tmplist;}
 
   
index 5e392351a6156970c12601e0fa713b86f4a1a5eb..2a3dcb43f885f3ff784ad9707bec307b81e62a80 100644 (file)
@@ -98,7 +98,7 @@ StrategyTableEntry::StrategyTableEntry() {
     isReady = 0;
     // WARNING: This constructor is called before CkMyPe() returns the correct results
     errorMode = NORMAL_MODE;
-    errorModeServer = STARTUP_MODE_SERVER; 
+    errorModeServer = NORMAL_MODE_SERVER; 
     discoveryMode = NORMAL_DISCOVERY_MODE;
     bracketedSetupFinished = 0;
 
@@ -150,21 +150,23 @@ const char *StrategyTableEntry::errorModeString(){
 
 
 const char *StrategyTableEntry::errorModeServerString(){
-  switch(errorModeServer) {
-  case STARTUP_MODE_SERVER:
-    return "STARTUP_MODE_SERVER    ";
-  case NORMAL_MODE_SERVER:
-    return "NORMAL_MODE_SERVER     ";
-  case ERROR_MODE_SERVER:
-    return "ERROR_MODE_SERVER      ";
-  case CONFIRM_MODE_SERVER:
-    return "CONFIRM_MODE_SERVER    ";
-  case ERROR_FIXED_MODE_SERVER:
-    return "ERROR_FIXED_MODE_SERVER";
-  case NON_SERVER_MODE_SERVER:
-    return "NON_SERVER_MODE_SERVER ";
-  default:
-    return "Unknown Server Error Mode";
+  if(CkMyPe() == 0){
+    switch(errorModeServer) {
+    case NORMAL_MODE_SERVER:
+      return "NORMAL_MODE_SERVER     ";
+    case ERROR_MODE_SERVER:
+      return "ERROR_MODE_SERVER      ";
+    case CONFIRM_MODE_SERVER:
+      return "CONFIRM_MODE_SERVER    ";
+    case ERROR_FIXED_MODE_SERVER:
+      return "ERROR_FIXED_MODE_SERVER";
+    case NON_SERVER_MODE_SERVER:
+      return "NON_SERVER_MODE_SERVER ";
+    default:
+      return "Unknown Server Error Mode";
+    }
+  } else {
+    return "";
   }
 }
 
index 15850772691d50b7417ef99cb67c4a805faa2eee..f2ccdc484c81cc07a21afd21e7d34b35d3a80ec1 100644 (file)
@@ -291,8 +291,8 @@ enum CountingErrorMode { NORMAL_MODE = 0,
                         CONFIRM_MODE = 2,   
                         ERROR_FIXED_MODE = 3  };
 
-enum CountingServerErrorMode { STARTUP_MODE_SERVER = 100,
-                              NORMAL_MODE_SERVER, 
+
+enum CountingServerErrorMode { NORMAL_MODE_SERVER, 
                               ERROR_MODE_SERVER,
                               CONFIRM_MODE_SERVER,  
                               ERROR_FIXED_MODE_SERVER,
@@ -345,34 +345,6 @@ public:
   int bufferOutgoing;
   
   
-  /** 
-   A flag to specify that this bracketed strategy is in an error mode,
-   due to an object migration causing more objects to call BeginIteration 
-   on one of the PEs than was expected. The detecting PE immediately enter the error
-   mode state, while other PEs will be notified of the error, and they will then 
-   enter the error state.
-   
-   This flag represents the state of the DFA followed by all processors. See the
-   hand drawn document describing these states and their transitions.
-      
-   There are three stated:
-     0) Normal mode: everything is fine
-     1) Error mode: miscount detected, trying to fix it
-     2) Confirm mode: a count has matched, trying to confirm it
-  
-  */
-  CountingErrorMode errorMode;
-  
-  /** 
-   The state of the coordinator (PE 0 of the a chare group).
-  */
-  CountingServerErrorMode errorModeServer;
-  
-  
-  /// A flag to specify what stage of Discovery process is underway for the PE
-  DiscoveryErrorMode discoveryMode;
-  
-
   /// A flag to specify that the bracketed strategy is fully setup, and it can operate normally.
   /// A value of 0 means strategy is fully setup
   int bracketedSetupFinished;
@@ -417,13 +389,58 @@ public:
   void reset();
   
 
-  const char *errorModeString();
-  const char *errorModeServerString();
-  const char *discoveryModeString();
-
 
   StrategyTableEntry();
+
+
+
+ private:
+  /** 
+   A flag to specify that this bracketed strategy is in an error mode,
+   due to an object migration causing more objects to call BeginIteration 
+   on one of the PEs than was expected. The detecting PE immediately enter the error
+   mode state, while other PEs will be notified of the error, and they will then 
+   enter the error state.
+  */
+  CountingErrorMode errorMode;
   
+  /** 
+   The state of the coordinator (PE 0 of the chare group).
+  */
+  CountingServerErrorMode errorModeServer;
+  
+  
+  /** 
+      A flag to specify what stage of Discovery process is underway for the PE 
+  */
+  DiscoveryErrorMode discoveryMode;
+
+
+ public:
+
+  CountingErrorMode getErrorMode(){return errorMode;}
+  CountingServerErrorMode getErrorModeServer(){return errorModeServer;}
+  DiscoveryErrorMode getDiscoveryMode(){return discoveryMode;}
+   
+  void setErrorMode(CountingErrorMode mode){
+    errorMode=mode;
+    ComlibPrintf("[%d] %s\n", CmiMyPe(), errorModeString());
+  }
+  void setErrorModeServer(CountingServerErrorMode mode){
+    errorModeServer=mode;
+    ComlibPrintf("[%d] %s lastKnownIteration=%d\n", CmiMyPe(), errorModeServerString(), lastKnownIteration);
+  }
+  void setDiscoveryMode(DiscoveryErrorMode mode){
+    discoveryMode=mode;
+    ComlibPrintf("[%d] %s\n", CmiMyPe(), discoveryModeString());
+  }
+
+  const char *errorModeString();
+  const char *errorModeServerString();
+  const char *discoveryModeString();  
+
 };
 
 typedef CkVec<StrategyTableEntry> StrategyTable;