Home Research Papers Posters Manuals Talks Download People Help Internal

ck-core/ckcausalmlog.C

Go to the documentation of this file.

#include "charm.h"
#include "ck.h"
#include "ckcausalmlog.h"
#include "queueing.h"
#include <sys/types.h>
#include <signal.h>
#include "CentralLB.h"

#ifdef _FAULT_CAUSAL_

// Collects some statistics about message logging. Beware of the high cost of accounting for
// duplicated determinants (for every determinant received, it consists in a linear search 
// through a potentially big list).
#define COLLECT_STATS_MSGS 0
#define COLLECT_STATS_MSGS_TOTAL 0
#define COLLECT_STATS_MSG_COUNT 0
#define COLLECT_STATS_DETS 0
#define COLLECT_STATS_DETS_DUP 0
#define COLLECT_STATS_MEMORY 0
#define COLLECT_STATS_TEAM 0

#define RECOVERY_SEND "SEND"
#define RECOVERY_PROCESS "PROCESS"

#define DEBUG_MEM(x)  //x
#define DEBUG(x) // x
#define DEBUG_RESTART(x)  //x
#define DEBUGLB(x)   // x
#define DEBUG_TEAM(x)  // x
#define DEBUG_PERF(x) // x
#define DEBUG_CHECKPOINT 1
#define DEBUG_NOW(x) x
#define DEBUG_PE(x,y) // if(CkMyPe() == x) y
#define DEBUG_RECOVERY(x) //x

extern const char *idx2str(const CkArrayIndex &ind);
extern const char *idx2str(const ArrayElement *el);

void getGlobalStep(CkGroupID gID);

bool fault_aware(CkObjID &recver);
void sendCheckpointData(int mode);
void createObjIDList(void *data,ChareMlogData *mlogData);
inline bool isLocal(int destPE);
inline bool isTeamLocal(int destPE);
void printLog(TProcessedLog *log);

int _restartFlag=0;
int _numRestartResponses=0;

//TODO: remove for perf runs
int countHashRefs=0; //count the number of gets
int countHashCollisions=0;

char *checkpointDirectory=".";
int unAckedCheckpoint=0;

int countLocal=0,countBuffered=0;
int countPiggy=0;
int countClearBufferedLocalCalls=0;

int countUpdateHomeAcks=0;

extern int teamSize;
extern int chkptPeriod;
extern bool fastRecovery;
extern int parallelRecovery;

char *killFile;
char *faultFile;
int killFlag=0;
int faultFlag=0;
int restartingMlogFlag=0;
void readKillFile();
double killTime=0.0;
double faultMean;
int checkpointCount=0;

CpvDeclare(Chare *,_currentObj);
CpvDeclare(StoredCheckpoint *,_storedCheckpointData);
CpvDeclare(CkQ<MlogEntry *> *,_delayedLocalMsgs);
CpvDeclare(Queue, _outOfOrderMessageQueue);
CpvDeclare(Queue, _delayedRemoteMessageQueue);
CpvDeclare(char **,_bufferedTicketRequests);
CpvDeclare(int *,_numBufferedTicketRequests);

/***** VARIABLES FOR CAUSAL MESSAGE LOGGING *****/
// temporal storage for the outgoing determinants
CpvDeclare(char *, _localDets);
// number of buffered determinants
int _numBufferedDets;
// current index of first determinant in _localDets
int _indexBufferedDets;
// current phase of determinants
int _phaseBufferedDets;

// stores the determinants from other nodes, to send them in case of a crash
CpvDeclare(CkDeterminantHashtableT *, _remoteDets);
// max number of buffered determinants
int _maxBufferedDets;

// stores the incarnation number from every other processor
CpvDeclare(char *, _incarnation);

/***** *****/

/***** VARIABLES FOR PARALLEL RECOVERY *****/
CpvDeclare(CkVec<LocationID *> *, _emigrantRecObjs);
CpvDeclare(CkVec<CkLocRec_local *> *, _immigrantRecObjs);
/***** *****/

#if COLLECT_STATS_MSGS
int *numMsgsTarget;
int *sizeMsgsTarget;
int totalMsgsTarget;
float totalMsgsSize;
#endif
#if COLLECT_STATS_DETS
int numPiggyDets;
int numDets;
int numDupDets;
#endif
#if COLLECT_STATS_MEMORY
int msgLogSize;
int bufferedDetsSize;
int storedDetsSize;
#endif
//TML: variables for measuring savings with teams in message logging
#if COLLECT_STATS_TEAM
float MLOGFT_totalLogSize = 0.0;
float MLOGFT_totalMessages = 0.0;
#endif

static double adjustChkptPeriod=0.0; //in ms
static double nextCheckpointTime=0.0;//in seconds
static CkHashtableT<CkHashtableAdaptorT<CkObjID>,CkHashtableT<CkHashtableAdaptorT<CkObjID>,SNToTicket *> *> detTable (1000,0.3);

int _pingHandlerIdx;

char objString[100];
int _checkpointRequestHandlerIdx;
int _storeCheckpointHandlerIdx;
int _checkpointAckHandlerIdx;
int _getCheckpointHandlerIdx;
int _recvCheckpointHandlerIdx;
int _removeProcessedLogHandlerIdx;

int _verifyAckRequestHandlerIdx;
int _verifyAckHandlerIdx;
int _dummyMigrationHandlerIdx;


int     _getGlobalStepHandlerIdx;
int     _recvGlobalStepHandlerIdx;

int _updateHomeRequestHandlerIdx;
int _updateHomeAckHandlerIdx;
int _resendMessagesHandlerIdx;
int _sendDetsHandlerIdx;
int _sendDetsReplyHandlerIdx;
int _receivedTNDataHandlerIdx;
int _receivedDetDataHandlerIdx;
int _distributedLocationHandlerIdx;
int _storeDeterminantsHandlerIdx;
int _removeDeterminantsHandlerIdx;

//TML: integer constants for team-based message logging
int _restartHandlerIdx;
int _getRestartCheckpointHandlerIdx;
int _recvRestartCheckpointHandlerIdx;
void setTeamRecovery(void *data, ChareMlogData *mlogData);
void unsetTeamRecovery(void *data, ChareMlogData *mlogData);

int verifyAckTotal;
int verifyAckCount;

int verifyAckedRequests=0;

RestartRequest *storedRequest;

int _falseRestart =0; 
//Load balancing globals
int onGoingLoadBalancing=0;
void *centralLb;
void (*resumeLbFnPtr)(void *);
int _receiveMlogLocationHandlerIdx;
int _receiveMigrationNoticeHandlerIdx;
int _receiveMigrationNoticeAckHandlerIdx;
int _checkpointBarrierHandlerIdx;
int _checkpointBarrierAckHandlerIdx;

CkVec<MigrationRecord> migratedNoticeList;
CkVec<RetainedMigratedObject *> retainedObjectList;
int donotCountMigration=0;
int countLBMigratedAway=0;
int countLBToMigrate=0;
int migrationDoneCalled=0;
int checkpointBarrierCount=0;
int globalResumeCount=0;
CkGroupID globalLBID;
int restartDecisionNumber=-1;

double lastCompletedAlarm=0;
double lastRestart=0;

//update location globals
int _receiveLocationHandlerIdx;

void _messageLoggingInit(){
        //current object
        CpvInitialize(Chare *,_currentObj);
        
        //registering handlers for message logging
        _pingHandlerIdx = CkRegisterHandler((CmiHandler)_pingHandler);
                
        //handlers for checkpointing
        _storeCheckpointHandlerIdx = CkRegisterHandler((CmiHandler)_storeCheckpointHandler);
        _checkpointAckHandlerIdx = CkRegisterHandler((CmiHandler) _checkpointAckHandler);
        _removeProcessedLogHandlerIdx  = CkRegisterHandler((CmiHandler)_removeProcessedLogHandler);
        _checkpointRequestHandlerIdx =  CkRegisterHandler((CmiHandler)_checkpointRequestHandler);

        //handlers for restart
        _getCheckpointHandlerIdx = CkRegisterHandler((CmiHandler)_getCheckpointHandler);
        _recvCheckpointHandlerIdx = CkRegisterHandler((CmiHandler)_recvCheckpointHandler);
        _updateHomeRequestHandlerIdx =CkRegisterHandler((CmiHandler)_updateHomeRequestHandler);
        _updateHomeAckHandlerIdx =  CkRegisterHandler((CmiHandler) _updateHomeAckHandler);
        _resendMessagesHandlerIdx = CkRegisterHandler((CmiHandler)_resendMessagesHandler);
        _sendDetsHandlerIdx = CkRegisterHandler((CmiHandler)_sendDetsHandler);
        _sendDetsReplyHandlerIdx = CkRegisterHandler((CmiHandler)_sendDetsReplyHandler);
        _receivedTNDataHandlerIdx=CkRegisterHandler((CmiHandler)_receivedTNDataHandler);
        _receivedDetDataHandlerIdx = CkRegisterHandler((CmiHandler)_receivedDetDataHandler);
        _distributedLocationHandlerIdx=CkRegisterHandler((CmiHandler)_distributedLocationHandler);
        _verifyAckRequestHandlerIdx = CkRegisterHandler((CmiHandler)_verifyAckRequestHandler);
        _verifyAckHandlerIdx = CkRegisterHandler((CmiHandler)_verifyAckHandler);
        _dummyMigrationHandlerIdx = CkRegisterHandler((CmiHandler)_dummyMigrationHandler);

        //TML: handlers for team-based message logging
        _restartHandlerIdx = CkRegisterHandler((CmiHandler)_restartHandler);
        _getRestartCheckpointHandlerIdx = CkRegisterHandler((CmiHandler)_getRestartCheckpointHandler);
        _recvRestartCheckpointHandlerIdx = CkRegisterHandler((CmiHandler)_recvRestartCheckpointHandler);

        // handlers for causal message logging
        _storeDeterminantsHandlerIdx = CkRegisterHandler((CmiHandler)_storeDeterminantsHandler);
        _removeDeterminantsHandlerIdx = CkRegisterHandler((CmiHandler)_removeDeterminantsHandler);
        
        //handlers for load balancing
        _receiveMlogLocationHandlerIdx=CkRegisterHandler((CmiHandler)_receiveMlogLocationHandler);
        _receiveMigrationNoticeHandlerIdx=CkRegisterHandler((CmiHandler)_receiveMigrationNoticeHandler);
        _receiveMigrationNoticeAckHandlerIdx=CkRegisterHandler((CmiHandler)_receiveMigrationNoticeAckHandler);
        _getGlobalStepHandlerIdx=CkRegisterHandler((CmiHandler)_getGlobalStepHandler);
        _recvGlobalStepHandlerIdx=CkRegisterHandler((CmiHandler)_recvGlobalStepHandler);
        _checkpointBarrierHandlerIdx=CkRegisterHandler((CmiHandler)_checkpointBarrierHandler);
        _checkpointBarrierAckHandlerIdx=CkRegisterHandler((CmiHandler)_checkpointBarrierAckHandler);
        
        //handlers for updating locations
        _receiveLocationHandlerIdx=CkRegisterHandler((CmiHandler)_receiveLocationHandler);
        
        //Cpv variables for message logging
        CpvInitialize(CkQ<MlogEntry *>*,_delayedLocalMsgs);
        CpvAccess(_delayedLocalMsgs) = new CkQ<MlogEntry *>;
        CpvInitialize(Queue, _outOfOrderMessageQueue);
        CpvInitialize(Queue, _delayedRemoteMessageQueue);
        CpvAccess(_outOfOrderMessageQueue) = CqsCreate();
        CpvAccess(_delayedRemoteMessageQueue) = CqsCreate();
        
        CpvInitialize(char **,_bufferedTicketRequests);
        CpvAccess(_bufferedTicketRequests) = new char *[CkNumPes()];
        CpvAccess(_numBufferedTicketRequests) = new int[CkNumPes()];
        for(int i=0;i<CkNumPes();i++){
                CpvAccess(_bufferedTicketRequests)[i]=NULL;
                CpvAccess(_numBufferedTicketRequests)[i]=0;
        }
 
        // Cpv variables for causal protocol
        _numBufferedDets = 0;
        _indexBufferedDets = 0;
        _phaseBufferedDets = 0;
        _maxBufferedDets = INITIAL_BUFFERED_DETERMINANTS;
        CpvInitialize(char *, _localDets);
        CpvInitialize((CkHashtableT<CkHashtableAdaptorT<CkObjID>, CkVec<Determinant> *> *),_remoteDets);
        CpvInitialize(char *, _incarnation);
        CpvAccess(_localDets) = (char *) CmiAlloc(_maxBufferedDets * sizeof(Determinant));
        CpvAccess(_remoteDets) = new CkHashtableT<CkHashtableAdaptorT<CkObjID>, CkVec<Determinant> *>(100, 0.4);
        CpvAccess(_incarnation) = (char *) CmiAlloc(CmiNumPes() * sizeof(int));
        for(int i=0; i<CmiNumPes(); i++){
                CpvAccess(_incarnation)[i] = 0;
        }

        // Cpv variables for parallel recovery
        CpvInitialize(CkVec<LocationID *> *, _emigrantRecObjs);
        CpvAccess(_emigrantRecObjs) = new CkVec<LocationID *>;
        CpvInitialize(CkVec<CkLocRec_local *> *, _immigrantRecObjs);
        CpvAccess(_immigrantRecObjs) = new CkVec<CkLocRec_local *>;

        //Cpv variables for checkpoint
        CpvInitialize(StoredCheckpoint *,_storedCheckpointData);
        CpvAccess(_storedCheckpointData) = new StoredCheckpoint;

        // registering user events for projections      
        traceRegisterUserEvent("Remove Logs", 20);
        traceRegisterUserEvent("Ticket Request Handler", 21);
        traceRegisterUserEvent("Ticket Handler", 22);
        traceRegisterUserEvent("Local Message Copy Handler", 23);
        traceRegisterUserEvent("Local Message Ack Handler", 24);        
        traceRegisterUserEvent("Preprocess current message",25);
        traceRegisterUserEvent("Preprocess past message",26);
        traceRegisterUserEvent("Preprocess future message",27);
        traceRegisterUserEvent("Checkpoint",28);
        traceRegisterUserEvent("Checkpoint Store",29);
        traceRegisterUserEvent("Checkpoint Ack",30);
        traceRegisterUserEvent("Send Ticket Request",31);
        traceRegisterUserEvent("Generalticketrequest1",32);
        traceRegisterUserEvent("TicketLogLocal",33);
        traceRegisterUserEvent("next_ticket and SN",34);
        traceRegisterUserEvent("Timeout for buffered remote messages",35);
        traceRegisterUserEvent("Timeout for buffered local messages",36);
        traceRegisterUserEvent("Inform Location Home",37);
        traceRegisterUserEvent("Receive Location Handler",38);
        
        lastCompletedAlarm=CmiWallTimer();
        lastRestart = CmiWallTimer();

#if COLLECT_STATS_MSGS
#if COLLECT_STATS_MSGS_TOTAL
        totalMsgsTarget = 0;
        totalMsgsSize = 0.0;
#else
        numMsgsTarget = (int *)CmiAlloc(sizeof(int) * CmiNumPes());
        sizeMsgsTarget = (int *)CmiAlloc(sizeof(int) * CmiNumPes());
        for(int i=0; i<CmiNumPes(); i++){
                numMsgsTarget[i] = 0;
                sizeMsgsTarget[i] = 0;
        }
#endif
#endif
#if COLLECT_STATS_DETS
        numPiggyDets = 0;
        numDets = 0;
        numDupDets = 0;
#endif
#if COLLECT_STATS_MEMORY
        msgLogSize = 0;
        bufferedDetsSize = 0;
        storedDetsSize = 0;
#endif


}

void killLocal(void *_dummy,double curWallTime);        

void readKillFile(){
        FILE *fp=fopen(killFile,"r");
        if(!fp){
                return;
        }
        int proc;
        double sec;
        while(fscanf(fp,"%d %lf",&proc,&sec)==2){
                if(proc == CkMyPe()){
                        killTime = CmiWallTimer()+sec;
                        printf("[%d] To be killed after %.6lf s (MLOG) \n",CkMyPe(),sec);
                        CcdCallFnAfter(killLocal,NULL,sec*1000);        
                }
        }
        fclose(fp);
}

void readFaultFile(){
        FILE *fp=fopen(faultFile,"r");
        if(!fp){
                return;
        }
        int proc;
        double sec;
        fscanf(fp,"%d %lf",&proc,&sec);
        faultMean = sec;
        if(proc == CkMyPe()){
                printf("[%d] PE %d to be killed every %.6lf s (MEMCKPT) \n",CkMyPe(),proc,sec);
                CcdCallFnAfter(killLocal,NULL,sec*1000);
        }
        fclose(fp);
}

void killLocal(void *_dummy,double curWallTime){
        printf("[%d] KillLocal called at %.6lf \n",CkMyPe(),CmiWallTimer());
        if(CmiWallTimer()<killTime-1){
                CcdCallFnAfter(killLocal,NULL,(killTime-CmiWallTimer())*1000);  
        }else{  
                kill(getpid(),SIGKILL);
        }
}

/*** Auxiliary Functions ***/

inline void addBufferedDeterminant(CkObjID sender, CkObjID receiver, MCount SN, MCount TN){
        Determinant *det, *auxDet;
        char *aux;

        DEBUG(CkPrintf("[%d]Adding determinant\n",CkMyPe()));

        // checking if we are overflowing the buffered determinants array
        if(_indexBufferedDets >= _maxBufferedDets){
                aux = CpvAccess(_localDets);
                _maxBufferedDets *= 2;
                CpvAccess(_localDets) = (char *) CmiAlloc(_maxBufferedDets * sizeof(Determinant));
                memcpy(CpvAccess(_localDets), aux, _indexBufferedDets * sizeof(Determinant));
                CmiFree(aux);
        }

        // adding the new determinant at the end
        det = (Determinant *) (CpvAccess(_localDets) + _indexBufferedDets * sizeof(Determinant));
        det->sender = sender;
        det->receiver = receiver;
        det->SN = SN;
        det->TN = TN;
        _numBufferedDets++;
        _indexBufferedDets++;

#if COLLECT_STATS_MEMORY
        bufferedDetsSize++;
#endif
#if COLLECT_STATS_DETS
        numDets++;
#endif
}

/************************ Message logging methods ****************/

void sendGroupMsg(envelope *env, int destPE, int _infoIdx){
        if(destPE == CLD_BROADCAST || destPE == CLD_BROADCAST_ALL){
                DEBUG(printf("[%d] Group Broadcast \n",CkMyPe()));
                void *origMsg = EnvToUsr(env);
                for(int i=0;i<CmiNumPes();i++){
                        if(!(destPE == CLD_BROADCAST && i == CmiMyPe())){
                                void *copyMsg = CkCopyMsg(&origMsg);
                                envelope *copyEnv = UsrToEnv(copyMsg);
                                copyEnv->SN=0;
                                copyEnv->TN=0;
                                copyEnv->sender.type = TypeInvalid;
                                DEBUG(printf("[%d] Sending group broadcast message to proc %d \n",CkMyPe(),i));
                                sendGroupMsg(copyEnv,i,_infoIdx);
                        }
                }
                return;
        }

        // initializing values of envelope
        env->SN=0;
        env->TN=0;
        env->sender.type = TypeInvalid;

        CkObjID recver;
        recver.type = TypeGroup;
        recver.data.group.id = env->getGroupNum();
        recver.data.group.onPE = destPE;
        sendCommonMsg(recver,env,destPE,_infoIdx);
}

void sendNodeGroupMsg(envelope *env, int destNode, int _infoIdx){
        if(destNode == CLD_BROADCAST || destNode == CLD_BROADCAST_ALL){
                DEBUG(printf("[%d] NodeGroup Broadcast \n",CkMyPe()));
                void *origMsg = EnvToUsr(env);
                for(int i=0;i<CmiNumNodes();i++){
                        if(!(destNode == CLD_BROADCAST && i == CmiMyNode())){
                                void *copyMsg = CkCopyMsg(&origMsg);
                                envelope *copyEnv = UsrToEnv(copyMsg);
                                copyEnv->SN=0;
                                copyEnv->TN=0;
                                copyEnv->sender.type = TypeInvalid;
                                sendNodeGroupMsg(copyEnv,i,_infoIdx);
                        }
                }
                return;
        }

        // initializing values of envelope
        env->SN=0;
        env->TN=0;
        env->sender.type = TypeInvalid;

        CkObjID recver;
        recver.type = TypeNodeGroup;
        recver.data.group.id = env->getGroupNum();
        recver.data.group.onPE = destNode;
        sendCommonMsg(recver,env,destNode,_infoIdx);
}

void sendArrayMsg(envelope *env,int destPE,int _infoIdx){
        CkObjID recver;
        recver.type = TypeArray;
        recver.data.array.id = env->getsetArrayMgr();
        recver.data.array.idx.asChild() = *(&env->getsetArrayIndex());

        if(CpvAccess(_currentObj)!=NULL &&  CpvAccess(_currentObj)->mlogData->objID.type != TypeArray){
                char recverString[100],senderString[100];
                
                DEBUG(printf("[%d] %s being sent message from non-array %s \n",CkMyPe(),recver.toString(recverString),CpvAccess(_currentObj)->mlogData->objID.toString(senderString)));
        }

        // initializing values of envelope
        env->SN = 0;
        env->TN = 0;

        sendCommonMsg(recver,env,destPE,_infoIdx);
};

void sendChareMsg(envelope *env,int destPE,int _infoIdx, const CkChareID *pCid){
        CkObjID recver;
        recver.type = TypeChare;
        recver.data.chare.id = *pCid;

        if(CpvAccess(_currentObj)!=NULL &&  CpvAccess(_currentObj)->mlogData->objID.type != TypeArray){
                char recverString[100],senderString[100];
                
                DEBUG(printf("[%d] %s being sent message from non-array %s \n",CkMyPe(),recver.toString(recverString),CpvAccess(_currentObj)->mlogData->objID.toString(senderString)));
        }

        // initializing values of envelope
        env->SN = 0;
        env->TN = 0;

        sendCommonMsg(recver,env,destPE,_infoIdx);
};

void sendCommonMsg(CkObjID &recver,envelope *_env,int destPE,int _infoIdx){
        envelope *env = _env;
        MCount ticketNumber = 0;
        int resend=0; //is it a resend
        char recverName[100];
        double _startTime=CkWallTimer();
        
        DEBUG_MEM(CmiMemoryCheck());

        if(CpvAccess(_currentObj) == NULL){
//              CkAssert(0);
                DEBUG(printf("[%d] !!!!WARNING: _currentObj is NULL while message is being sent\n",CkMyPe());)
                generalCldEnqueue(destPE,env,_infoIdx);
                return;
        }
        
        // setting message logging data in the envelope
        env->incarnation = CpvAccess(_incarnation)[CkMyPe()];
        if(env->sender.type == TypeInvalid){
                env->sender = CpvAccess(_currentObj)->mlogData->objID;
        }else{
                envelope *copyEnv = copyEnvelope(env);
                env = copyEnv;
                env->sender = CpvAccess(_currentObj)->mlogData->objID;
                env->SN = 0;
        }
        
        DEBUG_MEM(CmiMemoryCheck());

        CkObjID &sender = env->sender;
        env->recver = recver;

        Chare *obj = (Chare *)env->sender.getObject();
          
        if(env->SN == 0){
                DEBUG_MEM(CmiMemoryCheck());
                env->SN = obj->mlogData->nextSN(recver);
        }else{
                resend = 1;
        }
        
        char senderString[100];
//      if(env->SN != 1){
                DEBUG(printf("[%d] Generate Ticket Request to %s from %s PE %d SN %d \n",CkMyPe(),env->recver.toString(recverName),env->sender.toString(senderString),destPE,env->SN));
        //      CmiPrintStackTrace(0);
/*      }else{
                DEBUG_RESTART(printf("[%d] Generate Ticket Request to %s from %s PE %d SN %d \n",CkMyPe(),env->recver.toString(recverName),env->sender.toString(senderString),destPE,env->SN));
        }*/
                
        MlogEntry *mEntry = new MlogEntry(env,destPE,_infoIdx);
//      CkPackMessage(&(mEntry->env));
//      traceUserBracketEvent(32,_startTime,CkWallTimer());
        
        _startTime = CkWallTimer();

        // uses the proper ticketing mechanism for local, team and general messages
        if(isLocal(destPE)){
                sendLocalMsg(mEntry);
        }else{
                if((teamSize > 1) && isTeamLocal(destPE)){

                        // look to see if this message already has a ticket in the team-table
                        Chare *senderObj = (Chare *)sender.getObject();
                        SNToTicket *ticketRow = senderObj->mlogData->teamTable.get(recver);
                        if(ticketRow != NULL){
                                Ticket ticket = ticketRow->get(env->SN);
                                if(ticket.TN != 0){
                                        ticketNumber = ticket.TN;
                                        DEBUG(CkPrintf("[%d] Found a team preticketed message\n",CkMyPe()));
                                }
                        }
                }
                
                // sending the message
                sendMsg(sender,recver,destPE,mEntry,env->SN,ticketNumber,resend);
                
        }
}

inline bool isLocal(int destPE){
        // both the destination and the origin are the same PE
        if(destPE == CkMyPe())
                return true;

        return false;
}

inline bool isTeamLocal(int destPE){

        // they belong to the same group
        if(teamSize > 1 && destPE/teamSize == CkMyPe()/teamSize)
                return true;

        return false;
}

void sendMsg(CkObjID &sender,CkObjID &recver,int destPE,MlogEntry *entry,MCount SN,MCount TN,int resend){
        DEBUG_NOW(char recverString[100]);
        DEBUG_NOW(char senderString[100]);

        int totalSize;
        StoreDeterminantsHeader header;
        int sizes[2];
        char *ptrs[2];

        envelope *env = entry->env;
        DEBUG_PE(3,printf("[%d] Sending message to %s from %s PE %d SN %d time %.6lf \n",CkMyPe(),env->recver.toString(recverString),env->sender.toString(senderString),destPE,env->SN,CkWallTimer()));

        // setting all the information
        Chare *obj = (Chare *)entry->env->sender.getObject();
        entry->env->recver = recver;
        entry->env->SN = SN;
        entry->env->TN = TN;
        if(!resend){
                //TML: only stores message if either it goes to this processor or to a processor in a different group
                if(!isTeamLocal(entry->destPE)){
                        obj->mlogData->addLogEntry(entry);
#if COLLECT_STATS_TEAM
                        MLOGFT_totalMessages += 1.0;
                        MLOGFT_totalLogSize += entry->env->getTotalsize();
#endif
                }else{
                        // the message has to be deleted after it has been sent
                        entry->env->freeMsg = true;
                }
        }

        // sending the determinants that causally affect this message
        if(_numBufferedDets > 0){

                // modifying the actual number of determinants sent in message
                header.number = _numBufferedDets;
                header.index = _indexBufferedDets;
                header.phase = _phaseBufferedDets;
                header.PE = CmiMyPe();
        
                // sending the message
                sizes[0] = sizeof(StoreDeterminantsHeader);
                sizes[1] = _numBufferedDets * sizeof(Determinant);
                ptrs[0] = (char *) &header;
                ptrs[1] = CpvAccess(_localDets) + (_indexBufferedDets - _numBufferedDets) * sizeof(Determinant);
                DEBUG(CkPrintf("[%d] Sending %d determinants\n",CkMyPe(),_numBufferedDets));
                CmiSetHandler(&header, _storeDeterminantsHandlerIdx);
                CmiSyncVectorSend(destPE, 2, &sizes[0], &ptrs[0]);
        }

        // updating its message log entry
        entry->indexBufDets = _indexBufferedDets;
        entry->numBufDets = _numBufferedDets;

        // sending the message
        generalCldEnqueue(destPE, entry->env, entry->_infoIdx);

        DEBUG_MEM(CmiMemoryCheck());
#if COLLECT_STATS_MSGS
#if COLLECT_STATS_MSGS_TOTAL
        totalMsgsTarget++;
        totalMsgsSize += (float)env->getTotalsize();
#else
        numMsgsTarget[destPE]++;
        sizeMsgsTarget[destPE] += env->getTotalsize();
#endif
#endif
#if COLLECT_STATS_DETS
        numPiggyDets += _numBufferedDets;
#endif
#if COLLECT_STATS_MEMORY
        msgLogSize += env->getTotalsize();
#endif
};


void sendLocalMsg(MlogEntry *entry){
        DEBUG_PERF(double _startTime=CkWallTimer());
        DEBUG_MEM(CmiMemoryCheck());
        DEBUG(Chare *senderObj = (Chare *)entry->env->sender.getObject();)
        DEBUG(char senderString[100]);
        DEBUG(char recverString[100]);
        Ticket ticket;

        DEBUG(printf("[%d] Local Message being sent for SN %d sender %s recver %s \n",CmiMyPe(),entry->env->SN,entry->env->sender.toString(senderString),entry->env->recver.toString(recverString)));

        // getting the receiver local object
        Chare *recverObj = (Chare *)entry->env->recver.getObject();

        // if receiver object is not NULL, we will ask it for a ticket
        if(recverObj){

/*HERE          // if we are recovery, then we must put this off for later
                if(recverObj->mlogData->restartFlag){
                        CpvAccess(_delayedLocalMsgs)->enq(entry);
                        return;
                }

                // check if this combination of sender and SN has been already a ticket
                ticket = recverObj->mlogData->getTicket(entry->env->sender, entry->env->SN);
                        
                // assigned a new ticket in case this message is completely new
                if(ticket.TN == 0){
                        ticket = recverObj->mlogData->next_ticket(entry->env->sender,entry->env->SN);

                        // if TN == 0 we enqueue this message since we are recovering from a crash
                        if(ticket.TN == 0){
                                CpvAccess(_delayedLocalMsgs)->enq(entry);
                                DEBUG(printf("[%d] Local Message request enqueued for SN %d sender %s recver %s \n",CmiMyPe(),entry->env->SN,entry->env->sender.toString(senderString),entry->env->recver.toString(recverString)));
                                
//                              traceUserBracketEvent(33,_startTime,CkWallTimer());
                                return;
                        }
                }

                //TODO: check for the case when an invalid ticket is returned
                //TODO: check for OLD or RECEIVED TICKETS
                entry->env->TN = ticket.TN;
                CkAssert(entry->env->TN > 0);
                DEBUG(printf("[%d] Local Message gets TN %d for SN %d sender %s recver %s \n",CmiMyPe(),entry->env->TN,entry->env->SN,entry->env->sender.toString(senderString),entry->env->recver.toString(recverString)));
        
                DEBUG_MEM(CmiMemoryCheck());

                // adding the determinant to _localDets
                addBufferedDeterminant(entry->env->sender, entry->env->recver, entry->env->SN, entry->env->TN);
*/
                // sends the local message
                _skipCldEnqueue(CmiMyPe(),entry->env,entry->_infoIdx);  

                DEBUG_MEM(CmiMemoryCheck());
        }else{
                DEBUG(printf("[%d] Local recver object is NULL \n",CmiMyPe()););
        }
//      traceUserBracketEvent(33,_startTime,CkWallTimer());
};

/****
        The handler functions
*****/

/*** CAUSAL PROTOCOL HANDLERS ***/

void _removeDeterminantsHandler(char *buffer){
        RemoveDeterminantsHeader *header;
        int index, phase;

        // getting the header from the message
        header = (RemoveDeterminantsHeader *)buffer;
        index = header->index;
        phase = header->phase;

        // fprintf(stderr,"ACK index=%d\n",index);

        // updating the number of buffered determinants
        if(phase == _phaseBufferedDets){
                if(index > _indexBufferedDets)
                        CkPrintf("phase: %d %d, index:%d %d\n",phase, _phaseBufferedDets, index, _indexBufferedDets);
                CmiAssert(index <= _indexBufferedDets);
                _numBufferedDets = _indexBufferedDets - index;
        }

        // releasing memory
        CmiFree(buffer);

}

void _storeDeterminantsHandler(char *buffer){
        StoreDeterminantsHeader *header;
        RemoveDeterminantsHeader removeHeader;
        Determinant *detPtr, *det;
        int i, n, index, phase, destPE;
        CkVec<Determinant> *vec;


        // getting the header from the message and pointing to the first determinant
        header = (StoreDeterminantsHeader *)buffer;
        n = header->number;
        index = header->index;
        phase = header->phase;
        destPE = header->PE;
        detPtr = (Determinant *)(buffer + sizeof(StoreDeterminantsHeader));

        DEBUG(CkPrintf("[%d] Storing %d determinants\n",CkMyPe(),header->number));
        DEBUG_MEM(CmiMemoryCheck());

        // going through all determinants and storing them into _remoteDets
        for(i = 0; i < n; i++){
                det = new Determinant();
                det->sender = detPtr->sender;
                det->receiver = detPtr->receiver;
                det->SN = detPtr->SN;
                det->TN = detPtr->TN;

                // getting the determinant array
                vec = CpvAccess(_remoteDets)->get(det->receiver);
                if(vec == NULL){
                        vec = new CkVec<Determinant>();
                        CpvAccess(_remoteDets)->put(det->receiver) = vec;
                }
#if COLLECT_STATS_DETS
#if COLLECT_STATS_DETS_DUP
                for(int j=0; j<vec->size(); j++){
                        if(isSameDet(&(*vec)[j],det)){
                                numDupDets++;
                                break;
                        }
                }
#endif
#endif
#if COLLECT_STATS_MEMORY
                storedDetsSize++;
#endif
                vec->push_back(*det);
                detPtr = detPtr++;
        }

        DEBUG_MEM(CmiMemoryCheck());

        // freeing buffer
        CmiFree(buffer);

        // sending the ACK back to the sender
        removeHeader.index = index;
        removeHeader.phase = phase;
        CmiSetHandler(&removeHeader,_removeDeterminantsHandlerIdx);
        CmiSyncSend(destPE, sizeof(RemoveDeterminantsHeader), (char *)&removeHeader);
        
}

inline void _ticketRequestHandler(TicketRequest *ticketRequest){
        DEBUG(printf("[%d] Ticket Request handler started \n",CkMyPe()));
        double  _startTime = CkWallTimer();
        CmiFree(ticketRequest);
//      traceUserBracketEvent(21,_startTime,CkWallTimer());                     
}


inline bool _getTicket(envelope *env, int *flag){
        DEBUG_MEM(CmiMemoryCheck());
        DEBUG(char recverName[100]);
        DEBUG(char senderString[100]);
        Ticket ticket;
        
        // getting information from request
        CkObjID sender = env->sender;
        CkObjID recver = env->recver;
        MCount SN = env->SN;
        MCount TN = env->TN;
        Chare *recverObj = (Chare *)recver.getObject();
        
        DEBUG(recver.toString(recverName);)

        // verifying whether we are recovering from a failure or not
        if(recverObj->mlogData->restartFlag)
                return false;

        // checking ticket table to see if this message already received a ticket
        ticket = recverObj->mlogData->getTicket(sender, SN);
        TN = ticket.TN;
        
        // checking if the message is team local and if it has a ticket already assigned
        if(teamSize > 1 && TN != 0){
                DEBUG(CkPrintf("[%d] Message has a ticket already assigned\n",CkMyPe()));
                recverObj->mlogData->verifyTicket(sender,SN,TN);
        }

        //check if a ticket for this has been already handed out to an object that used to be local but 
        // is no longer so.. need for parallel restart
        if(TN == 0){
                ticket = recverObj->mlogData->next_ticket(sender,SN);
                *flag = NEW_TICKET;
        } else {
                *flag = OLD_TICKET;
        }
        if(ticket.TN > recverObj->mlogData->tProcessed){
                ticket.state = NEW_TICKET;
        } else {
                ticket.state = OLD_TICKET;
        }
        // @todo: check for the case when an invalid ticket is returned
        if(ticket.TN == 0){
                DEBUG(printf("[%d] Ticket request to %s SN %d from %s delayed mesg %p\n",CkMyPe(),recverName, SN,sender.toString(senderString),ticketRequest));
                return false;
        }
                
        // setting the ticket number in the envelope
        env->TN = ticket.TN;
        DEBUG(printf("[%d] TN %d handed out to %s SN %d by %s sent to PE %d mesg %p at %.6lf\n",CkMyPe(),ticket.TN,sender.toString(senderString),SN,recverName,ticketRequest->senderPE,ticketRequest,CmiWallTimer()));
        return true;

};

bool fault_aware(CkObjID &recver){
        switch(recver.type){
                case TypeChare:
                        return true;
                case TypeMainChare:
                        return false;
                case TypeGroup:
                case TypeNodeGroup:
                case TypeArray:
                        return true;
                default:
                        return false;
        }
};

/* Preprocesses a received message */
int preProcessReceivedMessage(envelope *env, Chare **objPointer, MlogEntry **logEntryPointer){
        DEBUG_NOW(char recverString[100]);
        DEBUG_NOW(char senderString[100]);
        DEBUG_MEM(CmiMemoryCheck());
        int flag;
        bool ticketSuccess;

        // getting the receiver object
        CkObjID recver = env->recver;
        if(!fault_aware(recver)){
                return 1;
                CkPrintf("[%d] Receiver NOT fault aware\n",CkMyPe());
        }

        Chare *obj = (Chare *)recver.getObject();
        *objPointer = obj;
        if(obj == NULL){
                int possiblePE = recver.guessPE();
                if(possiblePE != CkMyPe()){
                        int totalSize = env->getTotalsize();
                        CmiSyncSend(possiblePE,totalSize,(char *)env);
                        
                        DEBUG_PE(0,printf("[%d] Forwarding message SN %d sender %s recver %s to %d\n",CkMyPe(),env->SN,env->sender.toString(senderString), recver.toString(recverString), possiblePE));
                }else{
                        // this is the case where a message is received and the object has not been initialized
                        // we delayed the delivery of the message
                        CqsEnqueue(CpvAccess(_outOfOrderMessageQueue),env);
                        
                        DEBUG_PE(0,printf("[%d] Message SN %d TN %d sender %s recver %s, receiver NOT found\n",CkMyPe(),env->SN,env->TN,env->sender.toString(senderString), recver.toString(recverString)));
                }
                return 0;
        }

        // checking if message comes from an old incarnation
        // message must be discarded
        if(env->incarnation < CpvAccess(_incarnation)[env->getSrcPe()]){
                CmiFree(env);
                return 0;
        }

        DEBUG_MEM(CmiMemoryCheck());
        DEBUG_PE(0,printf("[%d] Message received, sender = %s SN %d TN %d tProcessed %d for recver %s stored for future time %.6lf \n",CkMyPe(),env->sender.toString(senderString),env->SN,env->TN,obj->mlogData->tProcessed, recver.toString(recverString),CkWallTimer()));

        // getting a ticket for this message
        ticketSuccess = _getTicket(env,&flag);

        // we might be during recovery when this message arrives
        if(!ticketSuccess){
                
                // adding the message to a delay queue
                CqsEnqueue(CpvAccess(_delayedRemoteMessageQueue),env);
                DEBUG(printf("[%d] Adding to delayed remote message queue\n",CkMyPe()));

                return 0;
        }
        
        //printf("[%d] ----------> SN = %d, TN = %d\n",CkMyPe(),env->SN,env->TN);
        //printf("[%d] ----------> numBufferedDeterminants = %d \n",CkMyPe(),_numBufferedDets);
        
        if(flag == NEW_TICKET){
                // storing determinant of message in data structure
                addBufferedDeterminant(env->sender, env->recver, env->SN, env->TN);
        }

        DEBUG_MEM(CmiMemoryCheck());

        double _startTime = CkWallTimer();
//env->sender.updatePosition(env->getSrcPe());
        if(env->TN == obj->mlogData->tProcessed+1){
                //the message that needs to be processed now
                DEBUG_PE(3,printf("[%d] Message SN %d TN %d sender %s recver %s being processed recvPointer %p\n",CkMyPe(),env->SN,env->TN,env->sender.toString(senderString), recver.toString(recverString),obj));
                // once we find a message that we can process we put back all the messages in the out of order queue
                // back into the main scheduler queue. 
        DEBUG_MEM(CmiMemoryCheck());
                while(!CqsEmpty(CpvAccess(_outOfOrderMessageQueue))){
                        void *qMsgPtr;
                        CqsDequeue(CpvAccess(_outOfOrderMessageQueue),&qMsgPtr);
                        envelope *qEnv = (envelope *)qMsgPtr;
                        CqsEnqueueGeneral((Queue)CpvAccess(CsdSchedQueue),qEnv,CQS_QUEUEING_FIFO,qEnv->getPriobits(),(unsigned int *)qEnv->getPrioPtr());
        DEBUG_MEM(CmiMemoryCheck());
                }
//              traceUserBracketEvent(25,_startTime,CkWallTimer());
                //TODO: this might be a problem.. change made for leanMD
//              CpvAccess(_currentObj) = obj;
        DEBUG_MEM(CmiMemoryCheck());
                return 1;
        }

        // checking if message has already been processed
        // message must be discarded
        if(env->TN <= obj->mlogData->tProcessed){
                DEBUG_PE(3,printf("[%d] Message SN %d TN %d sender %s for recver %s being ignored tProcessed %d \n",CkMyPe(),env->SN,env->TN,env->sender.toString(senderString),recver.toString(recverString),obj->mlogData->tProcessed));
                
                CmiFree(env);
                return 0;
        }
        //message that needs to be processed in the future

        DEBUG_PE(3,printf("[%d] Early Message sender = %s SN %d TN %d tProcessed %d for recver %s stored for future time %.6lf \n",CkMyPe(),env->sender.toString(senderString),env->SN,env->TN,obj->mlogData->tProcessed, recver.toString(recverString),CkWallTimer()));
        //the message cant be processed now put it back in the out of order message Q, 
        //It will be transferred to the main queue later
        CqsEnqueue(CpvAccess(_outOfOrderMessageQueue),env);
//              traceUserBracketEvent(27,_startTime,CkWallTimer());
        DEBUG_MEM(CmiMemoryCheck());
        
        return 0;
}

void postProcessReceivedMessage(Chare *obj, CkObjID &sender, MCount SN, MlogEntry *entry){
        DEBUG(char senderString[100]);
        if(obj){
                if(sender.guessPE() == CkMyPe()){
                        if(entry != NULL){
                                entry->env = NULL;
                        }
                }
                obj->mlogData->tProcessed++;
/*              DEBUG(int qLength = CqsLength((Queue )CpvAccess(CsdSchedQueue)));               
                DEBUG(printf("[%d] Message SN %d %s has been processed  tProcessed %d scheduler queue length %d\n",CkMyPe(),SN,obj->mlogData->objID.toString(senderString),obj->mlogData->tProcessed,qLength));         */
//              CpvAccess(_currentObj)= NULL;
        }
        DEBUG_MEM(CmiMemoryCheck());
}

/***
        Helpers for the handlers and message logging methods
***/

void generalCldEnqueue(int destPE, envelope *env, int _infoIdx){
//      double _startTime = CkWallTimer();
        if(env->recver.type != TypeNodeGroup){
        //This repeats a step performed in skipCldEnq for messages sent to
        //other processors. I do this here so that messages to local processors
        //undergo the same transformation.. It lets the resend be uniform for 
        //all messages
//              CmiSetXHandler(env,CmiGetHandler(env));
                _skipCldEnqueue(destPE,env,_infoIdx);
        }else{
                _noCldNodeEnqueue(destPE,env);
        }
//      traceUserBracketEvent(22,_startTime,CkWallTimer());
}
//extern "C" int CmiGetNonLocalLength();
int calledRetryTicketRequest=0;

void _pingHandler(CkPingMsg *msg){
        printf("[%d] Received Ping from %d\n",CkMyPe(),msg->PE);
        CmiFree(msg);
}


/*****************************************************************************
        Checkpointing methods..
                Pack all the data on a processor and send it to the buddy periodically
                Also used to throw away message logs
*****************************************************************************/
CkVec<TProcessedLog> processedTicketLog;
void buildProcessedTicketLog(void *data,ChareMlogData *mlogData);
void clearUpMigratedRetainedLists(int PE);

void checkpointAlarm(void *_dummy,double curWallTime){
        double diff = curWallTime-lastCompletedAlarm;
        DEBUG(printf("[%d] calling for checkpoint %.6lf after last one\n",CkMyPe(),diff));
/*      if(CkWallTimer()-lastRestart < 50){
                CcdCallFnAfter(checkpointAlarm,NULL,chkptPeriod);
                return;
        }*/
        if(diff < ((chkptPeriod) - 2)){
                CcdCallFnAfter(checkpointAlarm,NULL,(chkptPeriod-diff)*1000);
                return;
        }
        CheckpointRequest request;
        request.PE = CkMyPe();
        CmiSetHandler(&request,_checkpointRequestHandlerIdx);
        CmiSyncBroadcastAll(sizeof(CheckpointRequest),(char *)&request);
};

void _checkpointRequestHandler(CheckpointRequest *request){
        startMlogCheckpoint(NULL,CmiWallTimer());
}

void startMlogCheckpoint(void *_dummy, double curWallTime){
        double _startTime = CkWallTimer();

        // increasing the checkpoint counter
        checkpointCount++;
        
#if DEBUG_CHECKPOINT
        if(CmiMyPe() == 0){
                printf("[%d] starting checkpoint at %.6lf CmiTimer %.6lf \n",CkMyPe(),CmiWallTimer(),CmiTimer());
        }
#endif

        DEBUG_MEM(CmiMemoryCheck());

        PUP::sizer psizer;
        psizer | checkpointCount;
        for(int i=0; i<CmiNumPes(); i++){
                psizer | CpvAccess(_incarnation)[i];
        }
        CkPupROData(psizer);
        DEBUG_MEM(CmiMemoryCheck());
        CkPupGroupData(psizer,CmiTrue);
        DEBUG_MEM(CmiMemoryCheck());
        CkPupNodeGroupData(psizer,CmiTrue);
        DEBUG_MEM(CmiMemoryCheck());
        pupArrayElementsSkip(psizer,CmiTrue,NULL);
        DEBUG_MEM(CmiMemoryCheck());

        int dataSize = psizer.size();
        int totalSize = sizeof(CheckPointDataMsg)+dataSize;
        char *msg = (char *)CmiAlloc(totalSize);
        CheckPointDataMsg *chkMsg = (CheckPointDataMsg *)msg;
        chkMsg->PE = CkMyPe();
        chkMsg->dataSize = dataSize;
        
        char *buf = &msg[sizeof(CheckPointDataMsg)];
        PUP::toMem pBuf(buf);

        pBuf | checkpointCount;
        for(int i=0; i<CmiNumPes(); i++){
                pBuf | CpvAccess(_incarnation)[i];
        }
        CkPupROData(pBuf);
        CkPupGroupData(pBuf,CmiTrue);
        CkPupNodeGroupData(pBuf,CmiTrue);
        pupArrayElementsSkip(pBuf,CmiTrue,NULL);

        unAckedCheckpoint=1;
        CmiSetHandler(msg,_storeCheckpointHandlerIdx);
        CmiSyncSendAndFree(getCheckPointPE(),totalSize,msg);
        
        /*
                Store the highest Ticket number processed for each chare on this processor
        */
        processedTicketLog.removeAll();
        forAllCharesDo(buildProcessedTicketLog,(void *)&processedTicketLog);

#if DEBUG_CHECKPOINT
        if(CmiMyPe() == 0){
                printf("[%d] finishing checkpoint at %.6lf CmiTimer %.6lf with dataSize %d\n",CkMyPe(),CmiWallTimer(),CmiTimer(),dataSize);
        }
#endif

#if COLLECT_STATS_MEMORY
        CkPrintf("[%d] CKP=%d BUF_DET=%d STO_DET=%d MSG_LOG=%d\n",CkMyPe(),totalSize,bufferedDetsSize*sizeof(Determinant),storedDetsSize*sizeof(Determinant),msgLogSize);
        msgLogSize = 0;
        bufferedDetsSize = 0;
        storedDetsSize = 0;
#endif

        if(CkMyPe() ==  0 && onGoingLoadBalancing==0 ){
                lastCompletedAlarm = curWallTime;
                CcdCallFnAfter(checkpointAlarm,NULL,chkptPeriod);
        }
        traceUserBracketEvent(28,_startTime,CkWallTimer());
};

void buildProcessedTicketLog(void *data, ChareMlogData *mlogData){
        DEBUG(char objString[100]);

        CkVec<TProcessedLog> *log = (CkVec<TProcessedLog> *)data;
        TProcessedLog logEntry;
        logEntry.recver = mlogData->objID;
        logEntry.tProcessed = mlogData->tProcessed;
        log->push_back(logEntry);

        DEBUG(printf("[%d] Tickets lower than %d to be thrown away for %s \n",CkMyPe(),logEntry.tProcessed,logEntry.recver.toString(objString)));
}

class ElementPacker : public CkLocIterator {
private:
        CkLocMgr *locMgr;
        PUP::er &p;
public:
        ElementPacker(CkLocMgr* mgr_, PUP::er &p_):locMgr(mgr_),p(p_){};
        void addLocation(CkLocation &loc) {
                CkArrayIndexMax idx=loc.getIndex();
                CkGroupID gID = locMgr->ckGetGroupID();
                p|gID;      // store loc mgr's GID as well for easier restore
                p|idx;
                p|loc;
        }
};

void pupArrayElementsSkip(PUP::er &p, CmiBool create, MigrationRecord *listToSkip,int listsize){
        int numElements,i;
        int numGroups = CkpvAccess(_groupIDTable)->size();      
        if(!p.isUnpacking()){
                numElements = CkCountArrayElements();
        }       
        p | numElements;
        DEBUG(printf("[%d] Number of arrayElements %d \n",CkMyPe(),numElements));
        if(!p.isUnpacking()){
                CKLOCMGR_LOOP(ElementPacker packer(mgr, p); mgr->iterate(packer););
        }else{
                //Flush all recs of all LocMgrs before putting in new elements
//              CKLOCMGR_LOOP(mgr->flushAllRecs(););
                for(int j=0;j<listsize;j++){
                        if(listToSkip[j].ackFrom == 0 && listToSkip[j].ackTo == 1){
                                printf("[%d] Array element to be skipped gid %d idx",CmiMyPe(),listToSkip[j].gID.idx);
                                listToSkip[j].idx.print();
                        }
                }
                
                printf("numElements = %d\n",numElements);
        
                for (int i=0; i<numElements; i++) {
                        CkGroupID gID;
                        CkArrayIndexMax idx;
                        p|gID;
                p|idx;
                        int flag=0;
                        int matchedIdx=0;
                        for(int j=0;j<listsize;j++){
                                if(listToSkip[j].ackFrom == 0 && listToSkip[j].ackTo == 1){
                                        if(listToSkip[j].gID == gID && listToSkip[j].idx == idx){
                                                matchedIdx = j;
                                                flag = 1;
                                                break;
                                        }
                                }
                        }
                        if(flag == 1){
                                printf("[%d] Array element being skipped gid %d idx %s\n",CmiMyPe(),gID.idx,idx2str(idx));
                        }else{
                                printf("[%d] Array element being recovered gid %d idx %s\n",CmiMyPe(),gID.idx,idx2str(idx));
                        }
                                
                        CkLocMgr *mgr = (CkLocMgr*)CkpvAccess(_groupTable)->find(gID).getObj();
                        CkPrintf("numLocalElements = %d\n",mgr->numLocalElements());
                        mgr->resume(idx,p,create,flag);
                        if(flag == 1){
                                int homePE = mgr->homePe(idx);
                                informLocationHome(gID,idx,homePE,listToSkip[matchedIdx].toPE);
                        }
                }
        }
};


void writeCheckpointToDisk(int size,char *chkpt){
        char fNameTemp[100];
        sprintf(fNameTemp,"%s/mlogCheckpoint%d_tmp",checkpointDirectory,CkMyPe());
        int fd = creat(fNameTemp,S_IRWXU);
        int ret = write(fd,chkpt,size);
        CkAssert(ret == size);
        close(fd);
        
        char fName[100];
        sprintf(fName,"%s/mlogCheckpoint%d",checkpointDirectory,CkMyPe());
        unlink(fName);

        rename(fNameTemp,fName);
}

//handler that receives the checkpoint from a processor
//it stores it and acks it
void _storeCheckpointHandler(char *msg){
        double _startTime=CkWallTimer();
                
        CheckPointDataMsg *chkMsg = (CheckPointDataMsg *)msg;
        DEBUG(printf("[%d] Checkpoint Data from %d stored with datasize %d\n",CkMyPe(),chkMsg->PE,chkMsg->dataSize);)
        
        char *chkpt = &msg[sizeof(CheckPointDataMsg)];  
        
        char *oldChkpt =        CpvAccess(_storedCheckpointData)->buf;
        if(oldChkpt != NULL){
                char *oldmsg = oldChkpt - sizeof(CheckPointDataMsg);
                CmiFree(oldmsg);
        }
        //turning off storing checkpoints
        
        int sendingPE = chkMsg->PE;
        
        CpvAccess(_storedCheckpointData)->buf = chkpt;
        CpvAccess(_storedCheckpointData)->bufSize = chkMsg->dataSize;
        CpvAccess(_storedCheckpointData)->PE = sendingPE;

        int count=0;
        for(int j=migratedNoticeList.size()-1;j>=0;j--){
                if(migratedNoticeList[j].fromPE == sendingPE){
                        migratedNoticeList[j].ackFrom = 1;
                }else{
                        CmiAssert("migratedNoticeList entry for processor other than buddy");
                }
                if(migratedNoticeList[j].ackFrom == 1 && migratedNoticeList[j].ackTo == 1){
                        migratedNoticeList.remove(j);
                        count++;
                }
                
        }
        DEBUG(printf("[%d] For proc %d from number of migratedNoticeList cleared %d checkpointAckHandler %d\n",CmiMyPe(),sendingPE,count,_checkpointAckHandlerIdx));
        
        CheckPointAck ackMsg;
        ackMsg.PE = CkMyPe();
        ackMsg.dataSize = CpvAccess(_storedCheckpointData)->bufSize;
        CmiSetHandler(&ackMsg,_checkpointAckHandlerIdx);
        CmiSyncSend(sendingPE,sizeof(CheckPointAck),(char *)&ackMsg);
        
        traceUserBracketEvent(29,_startTime,CkWallTimer());
};


void sendRemoveLogRequests(){
#if SYNCHRONIZED_CHECKPOINT
        CmiAbort("Remove log requests should not be sent in a synchronized checkpoint");
#endif
        double _startTime = CkWallTimer();      

        // computing total message size
        int totalSize = sizeof(RemoveLogRequest) + processedTicketLog.size()*sizeof(TProcessedLog) + sizeof(int) + sizeof(Determinant);
        char *requestMsg = (char *)CmiAlloc(totalSize);

        // filling up the message
        RemoveLogRequest *request = (RemoveLogRequest *)requestMsg;
        request->PE = CkMyPe();
        request->numberObjects = processedTicketLog.size();
        char *listProcessedLogs = &requestMsg[sizeof(RemoveLogRequest)];
        memcpy(listProcessedLogs,(char *)processedTicketLog.getVec(),processedTicketLog.size()*sizeof(TProcessedLog));
        char *listDeterminants = &listProcessedLogs[processedTicketLog.size()*sizeof(TProcessedLog)];
        int *numDeterminants = (int *)listDeterminants;
        numDeterminants[0] = 0; 
        listDeterminants = (char *)&numDeterminants[1];

        // setting the handler in the message
        CmiSetHandler(requestMsg,_removeProcessedLogHandlerIdx);
        
        DEBUG_MEM(CmiMemoryCheck());
        for(int i=0;i<CkNumPes();i++){
                CmiSyncSend(i,totalSize,requestMsg);
        }
        CmiFree(requestMsg);

        clearUpMigratedRetainedLists(CmiMyPe());
        //TODO: clear ticketTable
        
        traceUserBracketEvent(30,_startTime,CkWallTimer());

        DEBUG_MEM(CmiMemoryCheck());
}


void _checkpointAckHandler(CheckPointAck *ackMsg){
        DEBUG_MEM(CmiMemoryCheck());
        unAckedCheckpoint=0;
        DEBUGLB(printf("[%d] CheckPoint Acked from PE %d with size %d onGoingLoadBalancing %d \n",CkMyPe(),ackMsg->PE,ackMsg->dataSize,onGoingLoadBalancing));
        DEBUGLB(CkPrintf("[%d] ACK HANDLER with %d\n",CkMyPe(),onGoingLoadBalancing));  
        if(onGoingLoadBalancing){
                onGoingLoadBalancing = 0;
                finishedCheckpointLoadBalancing();
        }else{
                sendRemoveLogRequests();
        }
        CmiFree(ackMsg);
        
};


inline void populateDeterminantTable(char *data){
        DEBUG(char recverString[100]);
        DEBUG(char senderString[100]);
        int numDets, *numDetsPtr;
        Determinant *detList;
        CkHashtableT<CkHashtableAdaptorT<CkObjID>,SNToTicket *> *table;
        SNToTicket *tickets;
        Ticket ticket;

        RemoveLogRequest *request = (RemoveLogRequest *)data;
        TProcessedLog *list = (TProcessedLog *)(&data[sizeof(RemoveLogRequest)]);
        
        numDetsPtr = (int *)&list[request->numberObjects];
        numDets = numDetsPtr[0];
        detList = (Determinant *)&numDetsPtr[1];

        // inserting determinants into hashtable
        for(int i=0; i<numDets; i++){
                table = detTable.get(detList[i].sender);
                if(table == NULL){
                        table = new CkHashtableT<CkHashtableAdaptorT<CkObjID>,SNToTicket *>();
                        detTable.put(detList[i].sender) = table;
                }
                tickets = table->get(detList[i].receiver);
                if(tickets == NULL){
                        tickets = new SNToTicket();
                        table->put(detList[i].receiver) = tickets; 
                }
                ticket.TN = detList[i].TN;
                tickets->put(detList[i].SN) = ticket;
        }

        DEBUG_MEM(CmiMemoryCheck());    

}

void removeProcessedLogs(void *_data,ChareMlogData *mlogData){
        int total;
        DEBUG(char nameString[100]);
        DEBUG_MEM(CmiMemoryCheck());
        CmiMemoryCheck();
        char *data = (char *)_data;
        RemoveLogRequest *request = (RemoveLogRequest *)data;
        TProcessedLog *list = (TProcessedLog *)(&data[sizeof(RemoveLogRequest)]);
        CkQ<MlogEntry *> *mlog = mlogData->getMlog();
        CkHashtableT<CkHashtableAdaptorT<CkObjID>,SNToTicket *> *table;
        SNToTicket *tickets;
        MCount TN;

        int count=0;
        total = mlog->length();
        for(int i=0; i<total; i++){
                MlogEntry *logEntry = mlog->deq();

                // looking message in determinant table
                table = detTable.get(logEntry->env->sender);
                if(table != NULL){
                        tickets = table->get(logEntry->env->recver);
                        if(tickets != NULL){
                                TN = tickets->get(logEntry->env->SN).TN;
                                if(TN != 0){
                                        logEntry->env->TN = TN;
                                }
                        }
                }
        
                int match=0;
                for(int j=0;j<request->numberObjects;j++){
                        if(logEntry->env == NULL || (logEntry->env->recver == list[j].recver && logEntry->env->TN > 0 && logEntry->env->TN < list[j].tProcessed)){
                                //this log Entry should be removed
                                match = 1;
                                break;
                        }
                }
                char senderString[100],recverString[100];
//              DEBUG(CkPrintf("[%d] Message sender %s recver %s TN %d removed %d PE %d\n",CkMyPe(),logEntry->env->sender.toString(senderString),logEntry->env->recver.toString(recverString),logEntry->env->TN,match,request->PE));
                if(match){
                        count++;
                        delete logEntry;
                }else{
                        mlog->enq(logEntry);
                }
        }
        if(count > 0){
                DEBUG(printf("[%d] Removed %d processed Logs for %s\n",CkMyPe(),count,mlogData->objID.toString(nameString)));
        }
        DEBUG_MEM(CmiMemoryCheck());
        CmiMemoryCheck();
}

void _removeProcessedLogHandler(char *requestMsg){
        double start = CkWallTimer();

        // building map for determinants
        populateDeterminantTable(requestMsg);

        // removing processed messages from the message log
        forAllCharesDo(removeProcessedLogs,requestMsg);

        // @todo: clean determinant table 
        
        // printf("[%d] Removing Processed logs took %.6lf \n",CkMyPe(),CkWallTimer()-start);
        RemoveLogRequest *request = (RemoveLogRequest *)requestMsg;
        DEBUG(printf("[%d] Removing Processed logs for proc %d took %.6lf \n",CkMyPe(),request->PE,CkWallTimer()-start));
        //this assumes the buddy relationship between processors is symmetric. TODO:remove this assumption later
        /*
                Clear up the retainedObjectList and the migratedNoticeList that were created during load balancing
        */
        DEBUG_MEM(CmiMemoryCheck());
        clearUpMigratedRetainedLists(request->PE);
        
        traceUserBracketEvent(20,start,CkWallTimer());
        CmiFree(requestMsg);    
};


void clearUpMigratedRetainedLists(int PE){
        int count=0;
        CmiMemoryCheck();
        
        for(int j=migratedNoticeList.size()-1;j>=0;j--){
                if(migratedNoticeList[j].toPE == PE){
                        migratedNoticeList[j].ackTo = 1;
                }
                if(migratedNoticeList[j].ackFrom == 1 && migratedNoticeList[j].ackTo == 1){
                        migratedNoticeList.remove(j);
                        count++;
                }
        }
        DEBUG(printf("[%d] For proc %d to number of migratedNoticeList cleared %d \n",CmiMyPe(),PE,count));
        
        for(int j=retainedObjectList.size()-1;j>=0;j--){
                if(retainedObjectList[j]->migRecord.toPE == PE){
                        RetainedMigratedObject *obj = retainedObjectList[j];
                        DEBUG(printf("[%d] Clearing retainedObjectList %d to PE %d obj %p msg %p\n",CmiMyPe(),j,PE,obj,obj->msg));
                        retainedObjectList.remove(j);
                        if(obj->msg != NULL){
                                CmiMemoryCheck();
                                CmiFree(obj->msg);
                        }
                        delete obj;
                }
        }
}

/***************************************************************
        Restart Methods and handlers
***************************************************************/        

void CkMlogRestart(const char * dummy, CkArgMsg * dummyMsg){
        RestartRequest msg;

        fprintf(stderr,"[%d] Restart started at %.6lf \n",CkMyPe(),CmiWallTimer());

        // setting the restart flag
        _restartFlag = 1;
        _numRestartResponses = 0;

        // if we are using team-based message logging, all members of the group have to be restarted
        if(teamSize > 1){
                for(int i=(CkMyPe()/teamSize)*teamSize; i<((CkMyPe()/teamSize)+1)*teamSize; i++){
                        if(i != CkMyPe() && i < CkNumPes()){
                                // sending a message to the team member
                                msg.PE = CkMyPe();
                            CmiSetHandler(&msg,_restartHandlerIdx);
                            CmiSyncSend(i,sizeof(RestartRequest),(char *)&msg);
                        }
                }
        }

        // requesting the latest checkpoint from its buddy
        msg.PE = CkMyPe();
        CmiSetHandler(&msg,_getCheckpointHandlerIdx);
        CmiSyncSend(getCheckPointPE(),sizeof(RestartRequest),(char *)&msg);
};

void _restartHandler(RestartRequest *restartMsg){
        int i;
        int numGroups = CkpvAccess(_groupIDTable)->size();
        RestartRequest msg;
        
        fprintf(stderr,"[%d] Restart-team started at %.6lf \n",CkMyPe(),CmiWallTimer());

    // setting the restart flag
        _restartFlag = 1;
        _numRestartResponses = 0;

        // flushing all buffers
        //TEST END
/*      CkPrintf("[%d] HERE numGroups = %d\n",CkMyPe(),numGroups);
        CKLOCMGR_LOOP(mgr->flushAllRecs(););    
        for(int i=0;i<numGroups;i++){
        CkGroupID gID = (*CkpvAccess(_groupIDTable))[i];
                IrrGroup *obj = CkpvAccess(_groupTable)->find(gID).getObj();
                obj->flushStates();
                obj->ckJustMigrated();
        }*/

    // requesting the latest checkpoint from its buddy
        msg.PE = CkMyPe();
        CmiSetHandler(&msg,_getRestartCheckpointHandlerIdx);
        CmiSyncSend(getCheckPointPE(),sizeof(RestartRequest),(char *)&msg);
}


void _getRestartCheckpointHandler(RestartRequest *restartMsg){

        // retrieving the stored checkpoint
        StoredCheckpoint *storedChkpt = CpvAccess(_storedCheckpointData);

        // making sure it is its buddy who is requesting the checkpoint
        CkAssert(restartMsg->PE == storedChkpt->PE);

        storedRequest = restartMsg;
        verifyAckTotal = 0;

        for(int i=0;i<migratedNoticeList.size();i++){
                if(migratedNoticeList[i].fromPE == restartMsg->PE){
//                      if(migratedNoticeList[i].ackFrom == 0 && migratedNoticeList[i].ackTo == 0){
                        if(migratedNoticeList[i].ackFrom == 0){
                                //need to verify if the object actually exists .. it might not
                                //have been acked but it might exist on it
                                VerifyAckMsg msg;
                                msg.migRecord = migratedNoticeList[i];
                                msg.index = i;
                                msg.fromPE = CmiMyPe();
                                CmiPrintf("[%d] Verify  gid %d idx %s from proc %d\n",CmiMyPe(),migratedNoticeList[i].gID.idx,idx2str(migratedNoticeList[i].idx),migratedNoticeList[i].toPE);
                                CmiSetHandler(&msg,_verifyAckRequestHandlerIdx);
                                CmiSyncSend(migratedNoticeList[i].toPE,sizeof(VerifyAckMsg),(char *)&msg);
                                verifyAckTotal++;
                        }
                }
        }

        // sending the checkpoint back to its buddy     
        if(verifyAckTotal == 0){
                sendCheckpointData(MLOG_RESTARTED);
        }
        verifyAckCount = 0;
}

void _recvRestartCheckpointHandler(char *_restartData){
        RestartProcessorData *restartData = (RestartProcessorData *)_restartData;
        MigrationRecord *migratedAwayElements;

        globalLBID = restartData->lbGroupID;
        
        restartData->restartWallTime *= 1000;
        adjustChkptPeriod = restartData->restartWallTime/(double) chkptPeriod - floor(restartData->restartWallTime/(double) chkptPeriod);
        adjustChkptPeriod = (double )chkptPeriod*(adjustChkptPeriod);
        if(adjustChkptPeriod < 0) adjustChkptPeriod = 0;

        
        printf("[%d] Team Restart Checkpointdata received from PE %d at %.6lf with checkpointSize %d\n",CkMyPe(),restartData->PE,CmiWallTimer(),restartData->checkPointSize);
        char *buf = &_restartData[sizeof(RestartProcessorData)];
        
        if(restartData->numMigratedAwayElements != 0){
                migratedAwayElements = new MigrationRecord[restartData->numMigratedAwayElements];
                memcpy(migratedAwayElements,buf,restartData->numMigratedAwayElements*sizeof(MigrationRecord));
                printf("[%d] Number of migratedaway elements %d\n",CmiMyPe(),restartData->numMigratedAwayElements);
                buf = &buf[restartData->numMigratedAwayElements*sizeof(MigrationRecord)];
        }

        // turning on the team recovery flag
        forAllCharesDo(setTeamRecovery,NULL);
        
        PUP::fromMem pBuf(buf);
        pBuf | checkpointCount;
        for(int i=0; i<CmiNumPes(); i++){
                pBuf | CpvAccess(_incarnation)[i];
        }
        CkPupROData(pBuf);
        CkPupGroupData(pBuf,CmiFalse);
        CkPupNodeGroupData(pBuf,CmiFalse);
        pupArrayElementsSkip(pBuf,CmiFalse,NULL);
        CkAssert(pBuf.size() == restartData->checkPointSize);
        printf("[%d] Restart Objects created from CheckPointData at %.6lf \n",CkMyPe(),CmiWallTimer());

        // increasing the incarnation number of all the team members
        for(int i=(CmiMyPe()/teamSize)*teamSize; i<(CmiMyPe()/teamSize+1)*(teamSize); i++){
                CpvAccess(_incarnation)[i]++;
        }
        
        // turning off the team recovery flag
        forAllCharesDo(unsetTeamRecovery,NULL);

        // initializing a few variables for handling local messages
        forAllCharesDo(initializeRestart,NULL);
        
        CmiFree(_restartData);  

        /*HERE _initDone();

        getGlobalStep(globalLBID);
        
        countUpdateHomeAcks = 0;
        RestartRequest updateHomeRequest;
        updateHomeRequest.PE = CmiMyPe();
        CmiSetHandler (&updateHomeRequest,_updateHomeRequestHandlerIdx);
        for(int i=0;i<CmiNumPes();i++){
                if(i != CmiMyPe()){
                        CmiSyncSend(i,sizeof(RestartRequest),(char *)&updateHomeRequest);
                }
        }
*/


        // Send out the request to resend logged messages to all other processors
        CkVec<TProcessedLog> objectVec;
        forAllCharesDo(createObjIDList, (void *)&objectVec);
        int numberObjects = objectVec.size();
        
        /*
                resendMsg layout |ResendRequest|Array of TProcessedLog|
        */
        int totalSize = sizeof(ResendRequest)+numberObjects*sizeof(TProcessedLog);
        char *resendMsg = (char *)CmiAlloc(totalSize);  

        ResendRequest *resendReq = (ResendRequest *)resendMsg;
        resendReq->PE =CkMyPe(); 
        resendReq->numberObjects = numberObjects;
        char *objList = &resendMsg[sizeof(ResendRequest)];
        memcpy(objList,objectVec.getVec(),numberObjects*sizeof(TProcessedLog));
        

        /* test for parallel restart migrate away object**/
//      if(fastRecovery){
//              distributeRestartedObjects();
//              printf("[%d] Redistribution of objects done at %.6lf \n",CkMyPe(),CmiWallTimer());
//      }
        
        /*      To make restart work for load balancing.. should only
        be used when checkpoint happens along with load balancing
        **/
//      forAllCharesDo(resumeFromSyncRestart,NULL);

        CentralLB *lb = (CentralLB *)CkpvAccess(_groupTable)->find(globalLBID).getObj();
        CpvAccess(_currentObj) = lb;
        lb->ReceiveDummyMigration(restartDecisionNumber);

        sleep(10);
        
        CmiSetHandler(resendMsg,_resendMessagesHandlerIdx);
        for(int i=0;i<CkNumPes();i++){
                if(i != CkMyPe()){
                        CmiSyncSend(i,totalSize,resendMsg);
                }       
        }
        _resendMessagesHandler(resendMsg);

}


void CkMlogRestartDouble(void *,double){
        CkMlogRestart(NULL,NULL);
};

//TML: restarting from local (group) failure
void CkMlogRestartLocal(){
    CkMlogRestart(NULL,NULL);
};

void _getCheckpointHandler(RestartRequest *restartMsg){

        // retrieving the stored checkpoint
        StoredCheckpoint *storedChkpt = CpvAccess(_storedCheckpointData);

        // making sure it is its buddy who is requesting the checkpoint
        CkAssert(restartMsg->PE == storedChkpt->PE);

        storedRequest = restartMsg;
        verifyAckTotal = 0;

        for(int i=0;i<migratedNoticeList.size();i++){
                if(migratedNoticeList[i].fromPE == restartMsg->PE){
//                      if(migratedNoticeList[i].ackFrom == 0 && migratedNoticeList[i].ackTo == 0){
                        if(migratedNoticeList[i].ackFrom == 0){
                                //need to verify if the object actually exists .. it might not
                                //have been acked but it might exist on it
                                VerifyAckMsg msg;
                                msg.migRecord = migratedNoticeList[i];
                                msg.index = i;
                                msg.fromPE = CmiMyPe();
                                CmiPrintf("[%d] Verify  gid %d idx %s from proc %d\n",CmiMyPe(),migratedNoticeList[i].gID.idx,idx2str(migratedNoticeList[i].idx),migratedNoticeList[i].toPE);
                                CmiSetHandler(&msg,_verifyAckRequestHandlerIdx);
                                CmiSyncSend(migratedNoticeList[i].toPE,sizeof(VerifyAckMsg),(char *)&msg);
                                verifyAckTotal++;
                        }
                }
        }

        // sending the checkpoint back to its buddy     
        if(verifyAckTotal == 0){
                sendCheckpointData(MLOG_CRASHED);
        }
        verifyAckCount = 0;
}


void _verifyAckRequestHandler(VerifyAckMsg *verifyRequest){
        CkLocMgr *locMgr =  (CkLocMgr*)CkpvAccess(_groupTable)->find(verifyRequest->migRecord.gID).getObj();
        CkLocRec *rec = locMgr->elementNrec(verifyRequest->migRecord.idx);
        if(rec != NULL && rec->type() == CkLocRec::local){
                        //this location exists on this processor
                        //and needs to be removed       
                        CkLocRec_local *localRec = (CkLocRec_local *) rec;
                        CmiPrintf("[%d] Found element gid %d idx %s that needs to be removed\n",CmiMyPe(),verifyRequest->migRecord.gID.idx,idx2str(verifyRequest->migRecord.idx));
                        
                        int localIdx = localRec->getLocalIndex();
                        LBDatabase *lbdb = localRec->getLBDB();
                        LDObjHandle ldHandle = localRec->getLdHandle();
                                
                        locMgr->setDuringMigration(true);
                        
                        locMgr->reclaim(verifyRequest->migRecord.idx,localIdx);
                        lbdb->UnregisterObj(ldHandle);
                        
                        locMgr->setDuringMigration(false);
                        
                        verifyAckedRequests++;

        }
        CmiSetHandler(verifyRequest, _verifyAckHandlerIdx);
        CmiSyncSendAndFree(verifyRequest->fromPE,sizeof(VerifyAckMsg),(char *)verifyRequest);
};


void _verifyAckHandler(VerifyAckMsg *verifyReply){
        int index =     verifyReply->index;
        migratedNoticeList[index] = verifyReply->migRecord;
        verifyAckCount++;
        CmiPrintf("[%d] VerifyReply received %d for  gid %d idx %s from proc %d\n",CmiMyPe(),migratedNoticeList[index].ackTo, migratedNoticeList[index].gID,idx2str(migratedNoticeList[index].idx),migratedNoticeList[index].toPE);
        if(verifyAckCount == verifyAckTotal){
                sendCheckpointData(MLOG_CRASHED);
        }
}


void sendCheckpointData(int mode){      
        RestartRequest *restartMsg = storedRequest;
        StoredCheckpoint *storedChkpt = CpvAccess(_storedCheckpointData);
        int numMigratedAwayElements = migratedNoticeList.size();
        if(migratedNoticeList.size() != 0){
                        printf("[%d] size of migratedNoticeList %d\n",CmiMyPe(),migratedNoticeList.size());
//                      CkAssert(migratedNoticeList.size() == 0);
        }
        
        
        int totalSize = sizeof(RestartProcessorData)+storedChkpt->bufSize;
        
        DEBUG_RESTART(CkPrintf("[%d] Sending out checkpoint for processor %d size %d \n",CkMyPe(),restartMsg->PE,totalSize);)
        CkPrintf("[%d] Sending out checkpoint for processor %d size %d \n",CkMyPe(),restartMsg->PE,totalSize);
        
        totalSize += numMigratedAwayElements*sizeof(MigrationRecord);
        
        char *msg = (char *)CmiAlloc(totalSize);
        
        RestartProcessorData *dataMsg = (RestartProcessorData *)msg;
        dataMsg->PE = CkMyPe();
        dataMsg->restartWallTime = CmiTimer();
        dataMsg->checkPointSize = storedChkpt->bufSize;
        
        dataMsg->numMigratedAwayElements = numMigratedAwayElements;
//      dataMsg->numMigratedAwayElements = 0;
        
        dataMsg->numMigratedInElements = 0;
        dataMsg->migratedElementSize = 0;
        dataMsg->lbGroupID = globalLBID;
        /*msg layout 
                |RestartProcessorData|List of Migrated Away ObjIDs|CheckpointData|CheckPointData for objects migrated in|
                Local MessageLog|
        */
        //store checkpoint data
        char *buf = &msg[sizeof(RestartProcessorData)];

        if(dataMsg->numMigratedAwayElements != 0){
                memcpy(buf,migratedNoticeList.getVec(),migratedNoticeList.size()*sizeof(MigrationRecord));
                buf = &buf[migratedNoticeList.size()*sizeof(MigrationRecord)];
        }
        

        memcpy(buf,storedChkpt->buf,storedChkpt->bufSize);
        buf = &buf[storedChkpt->bufSize];

        if(mode == MLOG_RESTARTED){
                CmiSetHandler(msg,_recvRestartCheckpointHandlerIdx);
                CmiSyncSendAndFree(restartMsg->PE,totalSize,msg);
                CmiFree(restartMsg);
        }else{
                CmiSetHandler(msg,_recvCheckpointHandlerIdx);
                CmiSyncSendAndFree(restartMsg->PE,totalSize,msg);
                CmiFree(restartMsg);
        }
};


// this list is used to create a vector of the object ids of all
//the chares on this processor currently and the highest TN processed by them 
//the first argument is actually a CkVec<TProcessedLog> *
void createObjIDList(void *data,ChareMlogData *mlogData){
        CkVec<TProcessedLog> *list = (CkVec<TProcessedLog> *)data;
        TProcessedLog entry;
        entry.recver = mlogData->objID;
        entry.tProcessed = mlogData->tProcessed;
        list->push_back(entry);
        DEBUG_TEAM(char objString[100]);
        DEBUG_TEAM(CkPrintf("[%d] %s restored with tProcessed set to %d \n",CkMyPe(),mlogData->objID.toString(objString),mlogData->tProcessed));
        DEBUG_RECOVERY(printLog(&entry));
}


void _recvCheckpointHandler(char *_restartData){
        RestartProcessorData *restartData = (RestartProcessorData *)_restartData;
        MigrationRecord *migratedAwayElements;

        globalLBID = restartData->lbGroupID;
        
        restartData->restartWallTime *= 1000;
        adjustChkptPeriod = restartData->restartWallTime/(double) chkptPeriod - floor(restartData->restartWallTime/(double) chkptPeriod);
        adjustChkptPeriod = (double )chkptPeriod*(adjustChkptPeriod);
        if(adjustChkptPeriod < 0) adjustChkptPeriod = 0;

        
        printf("[%d] Restart Checkpointdata received from PE %d at %.6lf with checkpointSize %d\n",CkMyPe(),restartData->PE,CmiWallTimer(),restartData->checkPointSize);
        char *buf = &_restartData[sizeof(RestartProcessorData)];
        
        if(restartData->numMigratedAwayElements != 0){
                migratedAwayElements = new MigrationRecord[restartData->numMigratedAwayElements];
                memcpy(migratedAwayElements,buf,restartData->numMigratedAwayElements*sizeof(MigrationRecord));
                printf("[%d] Number of migratedaway elements %d\n",CmiMyPe(),restartData->numMigratedAwayElements);
                buf = &buf[restartData->numMigratedAwayElements*sizeof(MigrationRecord)];
        }
        
        PUP::fromMem pBuf(buf);

        pBuf | checkpointCount;
        for(int i=0; i<CmiNumPes(); i++){
                pBuf | CpvAccess(_incarnation)[i];
        }
        CkPupROData(pBuf);
        CkPupGroupData(pBuf,CmiTrue);
        CkPupNodeGroupData(pBuf,CmiTrue);
        pupArrayElementsSkip(pBuf,CmiTrue,NULL);
        CkAssert(pBuf.size() == restartData->checkPointSize);
        printf("[%d] Restart Objects created from CheckPointData at %.6lf \n",CkMyPe(),CmiWallTimer());

        // increases the incarnation number
        CpvAccess(_incarnation)[CmiMyPe()]++;
        
        forAllCharesDo(initializeRestart,NULL);
        
        CmiFree(_restartData);
        
        _initDone();

        getGlobalStep(globalLBID);

        // sending request to send determinants
        _numRestartResponses = 0;
        
        // Send out the request to resend logged determinants to all other processors
        CkVec<TProcessedLog> objectVec;
        forAllCharesDo(createObjIDList, (void *)&objectVec);
        int numberObjects = objectVec.size();
        
        //      resendMsg layout: |ResendRequest|Array of TProcessedLog|
        int totalSize = sizeof(ResendRequest)+numberObjects*sizeof(TProcessedLog);
        char *resendMsg = (char *)CmiAlloc(totalSize);  

        ResendRequest *resendReq = (ResendRequest *)resendMsg;
        resendReq->PE =CkMyPe(); 
        resendReq->numberObjects = numberObjects;
        char *objList = &resendMsg[sizeof(ResendRequest)];
        memcpy(objList,objectVec.getVec(),numberObjects*sizeof(TProcessedLog)); 

        CmiSetHandler(resendMsg,_sendDetsHandlerIdx);
        for(int i=0;i<CkNumPes();i++){
                if(i != CkMyPe()){
                        CmiSyncSend(i,totalSize,resendMsg);
                }
        }
        CmiFree(resendMsg);
        
}

void _updateHomeAckHandler(RestartRequest *updateHomeAck){
        countUpdateHomeAcks++;
        CmiFree(updateHomeAck);
        // one is from the recvglobal step handler .. it is a dummy updatehomeackhandler
        if(countUpdateHomeAcks != CmiNumPes()){
                return;
        }

        // Send out the request to resend logged messages to all other processors
        CkVec<TProcessedLog> objectVec;
        forAllCharesDo(createObjIDList, (void *)&objectVec);
        int numberObjects = objectVec.size();
        
        //      resendMsg layout: |ResendRequest|Array of TProcessedLog|
        int totalSize = sizeof(ResendRequest)+numberObjects*sizeof(TProcessedLog);
        char *resendMsg = (char *)CmiAlloc(totalSize);  

        ResendRequest *resendReq = (ResendRequest *)resendMsg;
        resendReq->PE =CkMyPe(); 
        resendReq->numberObjects = numberObjects;
        char *objList = &resendMsg[sizeof(ResendRequest)];
        memcpy(objList,objectVec.getVec(),numberObjects*sizeof(TProcessedLog)); 

        /* test for parallel restart migrate away object**/
        if(fastRecovery){
                distributeRestartedObjects();
                printf("[%d] Redistribution of objects done at %.6lf \n",CkMyPe(),CmiWallTimer());
        }
        
        /*      To make restart work for load balancing.. should only
        be used when checkpoint happens along with load balancing
        **/
//      forAllCharesDo(resumeFromSyncRestart,NULL);

        CentralLB *lb = (CentralLB *)CkpvAccess(_groupTable)->find(globalLBID).getObj();
        CpvAccess(_currentObj) = lb;
        lb->ReceiveDummyMigration(restartDecisionNumber);

//HERE  sleep(10);
        
        CmiSetHandler(resendMsg,_resendMessagesHandlerIdx);
        for(int i=0;i<CkNumPes();i++){
                if(i != CkMyPe()){
                        CmiSyncSend(i,totalSize,resendMsg);
                }
        }
        _resendMessagesHandler(resendMsg);
        CmiFree(resendMsg);

};

void initializeRestart(void *data, ChareMlogData *mlogData){
        mlogData->resendReplyRecvd = 0;
        mlogData->receivedTNs = new CkVec<MCount>;
        mlogData->restartFlag = 1;
};

void updateHomePE(void *data,ChareMlogData *mlogData){
        RestartRequest *updateRequest = (RestartRequest *)data;
        int PE = updateRequest->PE; //restarted PE
        //if this object is an array Element and its home is the restarted processor
        // the home processor needs to know its current location
        if(mlogData->objID.type == TypeArray){
                //it is an array element
                CkGroupID myGID = mlogData->objID.data.array.id;
                CkArrayIndexMax myIdx =  mlogData->objID.data.array.idx.asChild();
                CkArrayID aid(mlogData->objID.data.array.id);           
                //check if the restarted processor is the home processor for this object
                CkLocMgr *locMgr = aid.ckLocalBranch()->getLocMgr();
                if(locMgr->homePe(myIdx) == PE){
                        DEBUG_RESTART(printf("[%d] Tell %d of current location of array element",CkMyPe(),PE));
                        DEBUG_RESTART(myIdx.print());
                        informLocationHome(locMgr->getGroupID(),myIdx,PE,CkMyPe());
                }
        }
};


void _updateHomeRequestHandler(RestartRequest *updateRequest){
        int sender = updateRequest->PE;
        
        forAllCharesDo(updateHomePE,updateRequest);
        
        updateRequest->PE = CmiMyPe();
        CmiSetHandler(updateRequest,_updateHomeAckHandlerIdx);
        CmiSyncSendAndFree(sender,sizeof(RestartRequest),(char *)updateRequest);
        if(sender == getCheckPointPE() && unAckedCheckpoint==1){
                CmiPrintf("[%d] Crashed processor did not ack so need to checkpoint again\n",CmiMyPe());
                checkpointCount--;
                startMlogCheckpoint(NULL,0);
        }
        if(sender == getCheckPointPE()){
                for(int i=0;i<retainedObjectList.size();i++){
                        if(retainedObjectList[i]->acked == 0){
                                MigrationNotice migMsg;
                                migMsg.migRecord = retainedObjectList[i]->migRecord;
                                migMsg.record = retainedObjectList[i];
                                CmiSetHandler((void *)&migMsg,_receiveMigrationNoticeHandlerIdx);
                                CmiSyncSend(getCheckPointPE(),sizeof(migMsg),(char *)&migMsg);
                        }
                }
        }
}

void fillTicketForChare(void *data, ChareMlogData *mlogData){
        DEBUG(char name[100]);
        ResendData *resendData = (ResendData *)data;
        int PE = resendData->PE; //restarted PE
        int count=0;
        CkHashtableIterator *iterator;
        void *objp;
        void *objkey;
        CkObjID *objID;
        SNToTicket *snToTicket;
        Ticket ticket;
        
        // traversing the team table looking up for the maximum TN received     
        iterator = mlogData->teamTable.iterator();
        while( (objp = iterator->next(&objkey)) != NULL ){
                objID = (CkObjID *)objkey;
        
                // traversing the resendData structure to add ticket numbers
                for(int j=0;j<resendData->numberObjects;j++){
                        if((*objID) == (resendData->listObjects)[j].recver){
                                snToTicket = *(SNToTicket **)objp;
//CkPrintf("[%d] ---> Traversing the resendData for %s start=%u finish=%u \n",CkMyPe(),objID->toString(name),snToTicket->getStartSN(),snToTicket->getFinishSN());
                                for(MCount snIndex=snToTicket->getStartSN(); snIndex<=snToTicket->getFinishSN(); snIndex++){
                                        ticket = snToTicket->get(snIndex);      
                                if(ticket.TN >= (resendData->listObjects)[j].tProcessed){
                                                //store the TNs that have been since the recver last checkpointed
                                                resendData->ticketVecs[j].push_back(ticket.TN);
                                        }
                                }
                        }
                }
        }

        //releasing the memory for the iterator
        delete iterator;
}


void setTeamRecovery(void *data, ChareMlogData *mlogData){
        char name[100];
        mlogData->teamRecoveryFlag = 1; 
}

void unsetTeamRecovery(void *data, ChareMlogData *mlogData){
        mlogData->teamRecoveryFlag = 0;
}

void printLog(TProcessedLog *log){
        char recverString[100];
        CkPrintf("[RECOVERY] [%d] OBJECT=\"%s\" TN=%d\n",CkMyPe(),log->recver.toString(recverString),log->tProcessed);
}

void printMsg(envelope *env, const char* par){
        char senderString[100];
        char recverString[100];
        CkPrintf("[RECOVERY] [%d] MSG-%s FROM=\"%s\" TO=\"%s\" SN=%d\n",CkMyPe(),par,env->sender.toString(senderString),env->recver.toString(recverString),env->SN);
}

void printDet(Determinant *det, const char* par){
        char senderString[100];
        char recverString[100];
        CkPrintf("[RECOVERY] [%d] DET-%s FROM=\"%s\" TO=\"%s\" SN=%d TN=%d\n",CkMyPe(),par,det->sender.toString(senderString),det->receiver.toString(recverString),det->SN,det->TN);
}

void resendMessageForChare(void *data, ChareMlogData *mlogData){
        DEBUG_RESTART(char nameString[100]);
        DEBUG_RESTART(char recverString[100]);
        DEBUG_RESTART(char senderString[100]);

        ResendData *resendData = (ResendData *)data;
        int PE = resendData->PE; //restarted PE
        int count=0;
        int ticketRequests=0;
        CkQ<MlogEntry *> *log = mlogData->getMlog();

        DEBUG_RESTART(printf("[%d] Resend message from %s to processor %d \n",CkMyPe(),mlogData->objID.toString(nameString),PE);)

        // traversing the message log to see if we must resend a message        
        for(int i=0;i<log->length();i++){
                MlogEntry *logEntry = (*log)[i];
                
                // if we sent out the logs of a local message to buddy and it crashed
                //before acknowledging 
                envelope *env = logEntry->env;
                if(env == NULL){
                        continue;
                }
        
                // resend if type is not invalid        
                if(env->recver.type != TypeInvalid){
                        for(int j=0;j<resendData->numberObjects;j++){
                                if(env->recver == (resendData->listObjects)[j].recver){
                                        if(PE != CkMyPe()){
                                                DEBUG_RECOVERY(printMsg(env,RECOVERY_SEND));
                                                if(env->recver.type == TypeNodeGroup){
                                                        CmiSyncNodeSend(PE,env->getTotalsize(),(char *)env);
                                                }else{
                                                        CmiSetHandler(env,CmiGetXHandler(env));
                                                        CmiSyncSend(PE,env->getTotalsize(),(char *)env);
                                                }
                                        }else{
                                                envelope *copyEnv = copyEnvelope(env);
                                                CqsEnqueueGeneral((Queue)CpvAccess(CsdSchedQueue),copyEnv, copyEnv->getQueueing(),copyEnv->getPriobits(),(unsigned int *)copyEnv->getPrioPtr());
                                        }
                                        DEBUG_RESTART(printf("[%d] Resent message sender %s recver %s SN %d TN %d \n",CkMyPe(),env->sender.toString(senderString),env->recver.toString(nameString),env->SN,env->TN));
                                        count++;
                                }
                        }//end of for loop of objects
                        
                }       
        }
        DEBUG_RESTART(printf("[%d] Resent  %d/%d (%d) messages  from %s to processor %d \n",CkMyPe(),count,log->length(),ticketRequests,mlogData->objID.toString(nameString),PE);)      
}

void _sendDetsHandler(char *msg){
        ResendData d;
        CkVec<Determinant> *detVec;
        ResendRequest *resendReq = (ResendRequest *)msg;

        // CkPrintf("[%d] Sending determinants\n",CkMyPe());

        // building the reply message
        char *listObjects = &msg[sizeof(ResendRequest)];
        d.numberObjects = resendReq->numberObjects;
        d.PE = resendReq->PE;
        d.listObjects = (TProcessedLog *)listObjects;
        d.ticketVecs = new CkVec<MCount>[d.numberObjects];
        detVec = new CkVec<Determinant>[d.numberObjects];

        // adding the remote determinants to the resendReplyMsg
        // traversing all the remote determinants
        CkVec<Determinant> *vec;
        for(int i=0; i<d.numberObjects; i++){
                vec = CpvAccess(_remoteDets)->get(d.listObjects[i].recver);
                if(vec != NULL){
                        for(int j=0; j<vec->size(); j++){

                                // only relevant determinants are to be sent
                                if((*vec)[j].TN > d.listObjects[i].tProcessed){

                                        // adding the ticket in the ticket vector
                                        d.ticketVecs[i].push_back((*vec)[j].TN);

                                        // adding the determinant in the determinant vector
                                        detVec[i].push_back((*vec)[j]);

                                        DEBUG_RECOVERY(printDet(&(*vec)[j],RECOVERY_SEND));
                                }
                        }
                }
        }

        int totalDetStored = 0;
        for(int i=0;i<d.numberObjects;i++){
                totalDetStored += detVec[i].size();
        }

        //send back the maximum ticket number for a message sent to each object on the 
        //restarted processor
        //Message: |ResendRequest|List of CkObjIDs|List<#number of objects in vec,TN of tickets seen>|
        
        int totalTNStored=0;
        for(int i=0;i<d.numberObjects;i++){
                totalTNStored += d.ticketVecs[i].size();
        }
        
        int totalSize = sizeof(ResendRequest) + d.numberObjects*(sizeof(CkObjID)+sizeof(int)+sizeof(int)) + totalTNStored*sizeof(MCount) + totalDetStored * sizeof(Determinant);
        char *resendReplyMsg = (char *)CmiAlloc(totalSize);
        
        ResendRequest *resendReply = (ResendRequest *)resendReplyMsg;
        resendReply->PE = CkMyPe();
        resendReply->numberObjects = d.numberObjects;
        
        char *replyListObjects = &resendReplyMsg[sizeof(ResendRequest)];
        CkObjID *replyObjects = (CkObjID *)replyListObjects;
        for(int i=0;i<d.numberObjects;i++){
                replyObjects[i] = d.listObjects[i].recver;
        }
        
        char *ticketList = &replyListObjects[sizeof(CkObjID)*d.numberObjects];
        for(int i=0;i<d.numberObjects;i++){
                int vecsize = d.ticketVecs[i].size();
                memcpy(ticketList,&vecsize,sizeof(int));
                ticketList = &ticketList[sizeof(int)];
                memcpy(ticketList,d.ticketVecs[i].getVec(),sizeof(MCount)*vecsize);
                ticketList = &ticketList[sizeof(MCount)*vecsize];
        }

        // adding the stored remote determinants to the message
        for(int i=0;i<d.numberObjects;i++){
                int vecsize = detVec[i].size();
                memcpy(ticketList,&vecsize,sizeof(int));
                ticketList = &ticketList[sizeof(int)];
                memcpy(ticketList,detVec[i].getVec(),sizeof(Determinant)*vecsize);
                ticketList = &ticketList[sizeof(Determinant)*vecsize];
        }

        CmiSetHandler(resendReplyMsg,_sendDetsReplyHandlerIdx);
        CmiSyncSendAndFree(d.PE,totalSize,(char *)resendReplyMsg);

        delete [] detVec;
        delete [] d.ticketVecs;

        DEBUG_MEM(CmiMemoryCheck());

        if(resendReq->PE != CkMyPe()){
                CmiFree(msg);
        }       
//      CmiPrintf("[%d] End of resend Request \n",CmiMyPe());
        lastRestart = CmiWallTimer();

}

void _resendMessagesHandler(char *msg){
        ResendData d;
        ResendRequest *resendReq = (ResendRequest *)msg;

        //CkPrintf("[%d] Resending messages\n",CkMyPe());

        // building the reply message
        char *listObjects = &msg[sizeof(ResendRequest)];
        d.numberObjects = resendReq->numberObjects;
        d.PE = resendReq->PE;
        d.listObjects = (TProcessedLog *)listObjects;
        
        DEBUG(printf("[%d] Received request to Resend Messages to processor %d numberObjects %d at %.6lf\n",CkMyPe(),resendReq->PE,resendReq->numberObjects,CmiWallTimer()));

        //TML: examines the origin processor to determine if it belongs to the same group.
        // In that case, it only returns the maximum ticket received for each object in the list.
        if(isTeamLocal(resendReq->PE) && CkMyPe() != resendReq->PE)
                forAllCharesDo(fillTicketForChare,&d);
        else
                forAllCharesDo(resendMessageForChare,&d);

        DEBUG_MEM(CmiMemoryCheck());

        if(resendReq->PE != CkMyPe()){
                CmiFree(msg);
        }       
//      CmiPrintf("[%d] End of resend Request \n",CmiMyPe());
        lastRestart = CmiWallTimer();
}

MCount maxVec(CkVec<MCount> *TNvec);
void sortVec(CkVec<MCount> *TNvec);
int searchVec(CkVec<MCount> *TNVec,MCount searchTN);

void processDelayedRemoteMsgQueue(){
        DEBUG(printf("[%d] Processing delayed remote messages\n",CkMyPe()));
        
        while(!CqsEmpty(CpvAccess(_delayedRemoteMessageQueue))){
                void *qMsgPtr;
                CqsDequeue(CpvAccess(_delayedRemoteMessageQueue),&qMsgPtr);
                envelope *qEnv = (envelope *)qMsgPtr;
                DEBUG_RECOVERY(printMsg(qEnv,RECOVERY_PROCESS));
                CqsEnqueueGeneral((Queue)CpvAccess(CsdSchedQueue),qEnv,CQS_QUEUEING_FIFO,qEnv->getPriobits(),(unsigned int *)qEnv->getPrioPtr());
                DEBUG_MEM(CmiMemoryCheck());
        }

}

void _sendDetsReplyHandler(char *msg){
        ResendRequest *resendReply = (ResendRequest *)msg;
        CkObjID *listObjects = (CkObjID *)(&msg[sizeof(ResendRequest)]);
        char *listTickets = (char *)(&listObjects[resendReply->numberObjects]);
        
//      DEBUG_RESTART(printf("[%d] _resendReply from %d \n",CmiMyPe(),resendReply->PE));
        DEBUG_TEAM(printf("[%d] _resendReply from %d \n",CmiMyPe(),resendReply->PE));
        
        for(int i =0; i< resendReply->numberObjects;i++){
                Chare *obj = (Chare *)listObjects[i].getObject();
                
                int vecsize;
                memcpy(&vecsize,listTickets,sizeof(int));
                listTickets = &listTickets[sizeof(int)];
                MCount *listTNs = (MCount *)listTickets;
                listTickets = &listTickets[vecsize*sizeof(MCount)];
        
                if(obj != NULL){
                        //the object was restarted on the processor on which it existed
                        processReceivedTN(obj,vecsize,listTNs);
                }else{
                        //pack up objID vecsize and listTNs and send it to the correct processor
                        int totalSize = sizeof(ReceivedTNData)+vecsize*sizeof(MCount);
                        char *TNMsg = (char *)CmiAlloc(totalSize);
                        ReceivedTNData *receivedTNData = (ReceivedTNData *)TNMsg;
                        receivedTNData->recver = listObjects[i];
                        receivedTNData->numTNs = vecsize;
                        char *tnList = &TNMsg[sizeof(ReceivedTNData)];
                        memcpy(tnList,listTNs,sizeof(MCount)*vecsize);
                        CmiSetHandler(TNMsg,_receivedTNDataHandlerIdx);
                        CmiSyncSendAndFree(listObjects[i].guessPE(),totalSize,TNMsg);
                }

        }

        // traversing all the retrieved determinants
        for(int i = 0; i < resendReply->numberObjects; i++){
                Chare *obj = (Chare *)listObjects[i].getObject();
                
                int vecsize;
                memcpy(&vecsize,listTickets,sizeof(int));
                listTickets = &listTickets[sizeof(int)];
                Determinant *listDets = (Determinant *)listTickets;     
                listTickets = &listTickets[vecsize*sizeof(Determinant)];
        
                if(obj != NULL){
                        //the object was restarted on the processor on which it existed
                        processReceivedDet(obj,vecsize,listDets);
                } else {
                        // pack the determinants and ship them to the other processor
                        // pack up objID vecsize and listDets and send it to the correct processor
                        int totalSize = sizeof(ReceivedDetData) + vecsize*sizeof(Determinant);
                        char *detMsg = (char *)CmiAlloc(totalSize);
                        ReceivedDetData *receivedDetData = (ReceivedDetData *)detMsg;
                        receivedDetData->recver = listObjects[i];
                        receivedDetData->numDets = vecsize;
                        char *detList = &detMsg[sizeof(ReceivedDetData)];
                        memcpy(detList,listDets,sizeof(Determinant)*vecsize);
                        CmiSetHandler(detMsg,_receivedDetDataHandlerIdx);
                        CmiSyncSendAndFree(listObjects[i].guessPE(),totalSize,detMsg);
                }

        }

        // checking if we have received all replies
        _numRestartResponses++;
        if(_numRestartResponses != CkNumPes())
                return;
        else 
                _numRestartResponses = 0;

        // continuing with restart process; send out the request to resend logged messages to all other processors
        CkVec<TProcessedLog> objectVec;
        forAllCharesDo(createObjIDList, (void *)&objectVec);
        int numberObjects = objectVec.size();
        
        //      resendMsg layout: |ResendRequest|Array of TProcessedLog|
        int totalSize = sizeof(ResendRequest)+numberObjects*sizeof(TProcessedLog);
        char *resendMsg = (char *)CmiAlloc(totalSize);  

        ResendRequest *resendReq = (ResendRequest *)resendMsg;
        resendReq->PE =CkMyPe(); 
        resendReq->numberObjects = numberObjects;
        char *objList = &resendMsg[sizeof(ResendRequest)];
        memcpy(objList,objectVec.getVec(),numberObjects*sizeof(TProcessedLog)); 

        CentralLB *lb = (CentralLB *)CkpvAccess(_groupTable)->find(globalLBID).getObj();
        CpvAccess(_currentObj) = lb;
        lb->ReceiveDummyMigration(restartDecisionNumber);

//HERE  sleep(10);
//      CkPrintf("[%d] RESUMING RECOVERY with %d \n",CkMyPe(),restartDecisionNumber);
        
        CmiSetHandler(resendMsg,_resendMessagesHandlerIdx);
        for(int i=0;i<CkNumPes();i++){
                if(i != CkMyPe()){
                        CmiSyncSend(i,totalSize,resendMsg);
                }
        }
        _resendMessagesHandler(resendMsg);
        CmiFree(resendMsg);

        /* test for parallel restart migrate away object**/
        if(fastRecovery){
                distributeRestartedObjects();
                printf("[%d] Redistribution of objects done at %.6lf \n",CkMyPe(),CmiWallTimer());
        }

//      processDelayedRemoteMsgQueue();

};

void _receivedDetDataHandler(ReceivedDetData *msg){
        DEBUG_NOW(char objName[100]);
        Chare *obj = (Chare *) msg->recver.getObject();
        if(obj){                
                char *_msg = (char *)msg;
                DEBUG(printf("[%d] receivedDetDataHandler for %s\n",CmiMyPe(),obj->mlogData->objID.toString(objName)));
                Determinant *listDets = (Determinant *)(&_msg[sizeof(ReceivedDetData)]);
                processReceivedDet(obj,msg->numDets,listDets);
                CmiFree(msg);
        }else{
                int totalSize = sizeof(ReceivedDetData)+sizeof(Determinant)*msg->numDets;
                CmiSyncSendAndFree(msg->recver.guessPE(),totalSize,(char *)msg);
        }
}

void _receivedTNDataHandler(ReceivedTNData *msg){
        DEBUG_NOW(char objName[100]);
        Chare *obj = (Chare *) msg->recver.getObject();
        if(obj){                
                char *_msg = (char *)msg;
                DEBUG(printf("[%d] receivedTNDataHandler for %s\n",CmiMyPe(),obj->mlogData->objID.toString(objName)));
                MCount *listTNs = (MCount *)(&_msg[sizeof(ReceivedTNData)]);
                processReceivedTN(obj,msg->numTNs,listTNs);
                CmiFree(msg);
        }else{
                int totalSize = sizeof(ReceivedTNData)+sizeof(MCount)*msg->numTNs;
                CmiSyncSendAndFree(msg->recver.guessPE(),totalSize,(char *)msg);
        }
};

void processReceivedDet(Chare *obj, int listSize, Determinant *listDets){
        Determinant *det;

        // traversing the whole list of determinants
        for(int i=0; i<listSize; i++){
                det = &listDets[i];
                if(CkMyPe() == 4) printDet(det,"RECOVERY");
                obj->mlogData->verifyTicket(det->sender, det->SN, det->TN);
                DEBUG_RECOVERY(printDet(det,RECOVERY_PROCESS));
        }
        
        DEBUG_MEM(CmiMemoryCheck());
}
        
void processReceivedTN(Chare *obj, int listSize, MCount *listTNs){
        // increases the number of resendReply received
        obj->mlogData->resendReplyRecvd++;

        DEBUG(char objName[100]);
        DEBUG(CkPrintf("[%d] processReceivedTN obj->mlogData->resendReplyRecvd=%d CkNumPes()=%d\n",CkMyPe(),obj->mlogData->resendReplyRecvd,CkNumPes()));
        //CkPrintf("[%d] processReceivedTN with %d listSize by %s\n",CkMyPe(),listSize,obj->mlogData->objID.toString(objName));
        //if(obj->mlogData->receivedTNs == NULL)
        //      CkPrintf("NULL\n");     
        //CkPrintf("using %d entries\n",obj->mlogData->receivedTNs->length());  

        // includes the tickets into the receivedTN structure
        for(int j=0;j<listSize;j++){
                obj->mlogData->receivedTNs->push_back(listTNs[j]);
        }
        
        //if this object has received all the replies find the ticket numbers
        //that senders know about. Those less than the ticket number processed 
        //by the receiver can be thrown away. The rest need not be consecutive
        // ie there can be holes in the list of ticket numbers seen by senders
        if(obj->mlogData->resendReplyRecvd == (CkNumPes() -1)){
                obj->mlogData->resendReplyRecvd = 0;

#if VERIFY_DETS
                //sort the received TNS
                sortVec(obj->mlogData->receivedTNs);
        
                //after all the received tickets are in we need to sort them and then 
                // calculate the holes  
                if(obj->mlogData->receivedTNs->size() > 0){
                        int tProcessedIndex = searchVec(obj->mlogData->receivedTNs,obj->mlogData->tProcessed);
                        int vecsize = obj->mlogData->receivedTNs->size();
                        int numberHoles = ((*obj->mlogData->receivedTNs)[vecsize-1] - obj->mlogData->tProcessed)-(vecsize -1 - tProcessedIndex);
                        
                        // updating tCount with the highest ticket handed out
                        if(teamSize > 1){
                                if(obj->mlogData->tCount < (*obj->mlogData->receivedTNs)[vecsize-1])
                                        obj->mlogData->tCount = (*obj->mlogData->receivedTNs)[vecsize-1];
                        }else{
                                obj->mlogData->tCount = (*obj->mlogData->receivedTNs)[vecsize-1];
                        }
                        
                        if(numberHoles == 0){
                        }else{
                                char objName[100];                                      
                                printf("[%d] Holes detected in the TNs for %s number %d \n",CkMyPe(),obj->mlogData->objID.toString(objName),numberHoles);
                                obj->mlogData->numberHoles = numberHoles;
                                obj->mlogData->ticketHoles = new MCount[numberHoles];
                                int countHoles=0;
                                for(int k=tProcessedIndex+1;k<vecsize;k++){
                                        if((*obj->mlogData->receivedTNs)[k] != (*obj->mlogData->receivedTNs)[k-1]+1){
                                                //the TNs are not consecutive at this point
                                                for(MCount newTN=(*obj->mlogData->receivedTNs)[k-1]+1;newTN<(*obj->mlogData->receivedTNs)[k];newTN++){
                                                        DEBUG(CKPrintf("hole no %d at %d next available ticket %d \n",countHoles,newTN,(*obj->mlogData->receivedTNs)[k]));
                                                        obj->mlogData->ticketHoles[countHoles] = newTN;
                                                        countHoles++;
                                                }       
                                        }
                                }
                                //Holes have been given new TN
                                if(countHoles != numberHoles){
                                        char str[100];
                                        printf("[%d] Obj %s countHoles %d numberHoles %d\n",CmiMyPe(),obj->mlogData->objID.toString(str),countHoles,numberHoles);
                                }
                                CkAssert(countHoles == numberHoles);                                    
                                obj->mlogData->currentHoles = numberHoles;
                        }
                }
#else
                if(obj->mlogData->receivedTNs->size() > 0){
                        obj->mlogData->tCount = maxVec(obj->mlogData->receivedTNs);
                }
#endif
                // cleaning up structures and getting ready to continue execution       
                delete obj->mlogData->receivedTNs;
                DEBUG(CkPrintf("[%d] Resetting receivedTNs\n",CkMyPe()));
                obj->mlogData->receivedTNs = NULL;
                obj->mlogData->restartFlag = 0;

                // processDelayedRemoteMsgQueue();

                DEBUG_RESTART(char objString[100]);
                DEBUG_RESTART(CkPrintf("[%d] Can restart handing out tickets again at %.6lf for %s\n",CkMyPe(),CmiWallTimer(),obj->mlogData->objID.toString(objString)));
        }

}

MCount maxVec(CkVec<MCount> *TNvec){
        MCount max = 0;
        for(int i=0; i<TNvec->size(); i++){
                if((*TNvec)[i] > max)
                        max = (*TNvec)[i];
        }
        return max;
}

void sortVec(CkVec<MCount> *TNvec){
        //sort it ->its bloddy bubble sort
        //TODO: use quicksort
        for(int i=0;i<TNvec->size();i++){
                for(int j=i+1;j<TNvec->size();j++){
                        if((*TNvec)[j] < (*TNvec)[i]){
                                MCount temp;
                                temp = (*TNvec)[i];
                                (*TNvec)[i] = (*TNvec)[j];
                                (*TNvec)[j] = temp;
                        }
                }
        }
        //make it unique .. since its sorted all equal units will be consecutive
        MCount *tempArray = new MCount[TNvec->size()];
        int     uniqueCount=-1;
        for(int i=0;i<TNvec->size();i++){
                tempArray[i] = 0;
                if(uniqueCount == -1 || tempArray[uniqueCount] != (*TNvec)[i]){
                        uniqueCount++;
                        tempArray[uniqueCount] = (*TNvec)[i];
                }
        }
        uniqueCount++;
        TNvec->removeAll();
        for(int i=0;i<uniqueCount;i++){
                TNvec->push_back(tempArray[i]);
        }
        delete [] tempArray;
}       

int searchVec(CkVec<MCount> *TNVec,MCount searchTN){
        if(TNVec->size() == 0){
                return -1; //not found in an empty vec
        }
        //binary search to find 
        int left=0;
        int right = TNVec->size();
        int mid = (left +right)/2;
        while(searchTN != (*TNVec)[mid] && left < right){
                if((*TNVec)[mid] > searchTN){
                        right = mid-1;
                }else{
                        left = mid+1;
                }
                mid = (left + right)/2;
        }
        if(left < right){
                //mid is the element to be returned
                return mid;
        }else{
                if(mid < TNVec->size() && mid >=0){
                        if((*TNVec)[mid] == searchTN){
                                return mid;
                        }else{
                                return -1;
                        }
                }else{
                        return -1;
                }
        }
};


/*
        Method to do parallel restart. Distribute some of the array elements to other processors.
        The problem is that we cant use to charm entry methods to do migration as it will get
        stuck in the protocol that is going to restart
        Note: in order to avoid interference between the objects being recovered, the current PE
    will NOT keep any object. It will be devoted to forward the messages to recovering objects.    Otherwise, the current PE has to do both things, recover objects and forward messages and 
    objects end up stepping into each other's shoes (interference).
*/

class ElementDistributor: public CkLocIterator{
        CkLocMgr *locMgr;
        int *targetPE;

        void pupLocation(CkLocation &loc,PUP::er &p){
                CkArrayIndexMax idx=loc.getIndex();
                CkGroupID gID = locMgr->ckGetGroupID();
                p|gID;      // store loc mgr's GID as well for easier restore
                p|idx;
                p|loc;
        };
public:
        ElementDistributor(CkLocMgr *mgr_,int *toPE_):locMgr(mgr_),targetPE(toPE_){};

        void addLocation(CkLocation &loc){

                // leaving object on this PE
                if(*targetPE == CkMyPe()){
                        *targetPE = (*targetPE +1)%CkNumPes();
                        return;
                }
                        
                CkArrayIndexMax idx = loc.getIndex();
                CkLocRec_local *rec = loc.getLocalRecord();
                        
                CkPrintf("[%d] Distributing objects to Processor %d: ",CkMyPe(),*targetPE);
                idx.print();

                // inserting the emigrant object to the list of emigrant recovery objects
                // every broadcast message will be sent to these objects
                LocationID *location = new LocationID();
                location->idx = idx;
                location->gid = loc.getManager()->getGroupID();
                location->PE = *targetPE;
                CpvAccess(_emigrantRecObjs)->push_back(location);       
                        
                //pack up this location and send it across
                PUP::sizer psizer;
                pupLocation(loc,psizer);
                int totalSize = psizer.size()+CmiMsgHeaderSizeBytes;
                char *msg = (char *)CmiAlloc(totalSize);
                char *buf = &msg[CmiMsgHeaderSizeBytes];
                PUP::toMem pmem(buf);
                pmem.becomeDeleting();
                pupLocation(loc,pmem);
                        
                locMgr->setDuringMigration(CmiTrue);
                delete rec;
                locMgr->setDuringMigration(CmiFalse);
                locMgr->inform(idx,*targetPE);

                CmiSetHandler(msg,_distributedLocationHandlerIdx);
                CmiSyncSendAndFree(*targetPE,totalSize,msg);

                CmiAssert(locMgr->lastKnown(idx) == *targetPE);

                //decide on the target processor for the next object
                *targetPE = *targetPE + 1;
                if(*targetPE > (CkMyPe() + parallelRecovery)){
                        *targetPE = CkMyPe() + 1;
                }
        }

};

void distributeRestartedObjects(){
        int numGroups = CkpvAccess(_groupIDTable)->size();      
        int i;
        int targetPE=CkMyPe()+1;
        CKLOCMGR_LOOP(ElementDistributor distributor(mgr,&targetPE);mgr->iterate(distributor););
};

void _distributedLocationHandler(char *receivedMsg){
        printf("Array element received at processor %d after distribution at restart\n",CkMyPe());
        char *buf = &receivedMsg[CmiMsgHeaderSizeBytes];
        PUP::fromMem pmem(buf);
        CkGroupID gID;
        CkArrayIndexMax idx;
        pmem |gID;
        pmem |idx;
        CkLocMgr *mgr = (CkLocMgr*)CkpvAccess(_groupTable)->find(gID).getObj();
        donotCountMigration=1;
        mgr->resume(idx,pmem,CmiTrue);
        donotCountMigration=0;
        informLocationHome(gID,idx,mgr->homePe(idx),CkMyPe());
        printf("Array element inserted at processor %d after distribution at restart ",CkMyPe());
        idx.print();

        CkLocRec *rec = mgr->elementRec(idx);
        CmiAssert(rec->type() == CkLocRec::local);

        // adding object to the list of immigrant recovery objects
        CpvAccess(_immigrantRecObjs)->push_back((CkLocRec_local *)rec);
        
        CkVec<CkMigratable *> eltList;
        mgr->migratableList((CkLocRec_local *)rec,eltList);
        for(int i=0;i<eltList.size();i++){
                if(eltList[i]->mlogData->toResumeOrNot == 1 && eltList[i]->mlogData->resumeCount < globalResumeCount){
                        CpvAccess(_currentObj) = eltList[i];
                        eltList[i]->mlogData->immigrantRecFlag = 1;
                        eltList[i]->ResumeFromSync();
                }
        }
}


void sendDummyMigration(int restartPE,CkGroupID lbID,CkGroupID locMgrID,CkArrayIndexMax &idx,int locationPE){
        DummyMigrationMsg buf;
        buf.flag = MLOG_OBJECT;
        buf.lbID = lbID;
        buf.mgrID = locMgrID;
        buf.idx = idx;
        buf.locationPE = locationPE;
        CmiSetHandler(&buf,_dummyMigrationHandlerIdx);
        CmiSyncSend(restartPE,sizeof(DummyMigrationMsg),(char *)&buf);
};


void sendDummyMigrationCounts(int *dummyCounts){
        DummyMigrationMsg buf;
        buf.flag = MLOG_COUNT;
        buf.lbID = globalLBID;
        CmiSetHandler(&buf,_dummyMigrationHandlerIdx);
        for(int i=0;i<CmiNumPes();i++){
                if(i != CmiMyPe() && dummyCounts[i] != 0){
                        buf.count = dummyCounts[i];
                        CmiSyncSend(i,sizeof(DummyMigrationMsg),(char *)&buf);
                }
        }
}


void _dummyMigrationHandler(DummyMigrationMsg *msg){
        CentralLB *lb = (CentralLB *)CkpvAccess(_groupTable)->find(msg->lbID).getObj();
        if(msg->flag == MLOG_OBJECT){
                DEBUG_RESTART(CmiPrintf("[%d] dummy Migration received from pe %d for %d:%s \n",CmiMyPe(),msg->locationPE,msg->mgrID.idx,idx2str(msg->idx)));
                LDObjHandle h;
                lb->Migrated(h,1);
        }
        if(msg->flag == MLOG_COUNT){
                DEBUG_RESTART(CmiPrintf("[%d] dummyMigration count %d received from restarted processor\n",CmiMyPe(),msg->count));
                msg->count -= verifyAckedRequests;
                for(int i=0;i<msg->count;i++){
                        LDObjHandle h;
                        lb->Migrated(h,1);
                }
        }
        verifyAckedRequests=0;
        CmiFree(msg);
};

/*****************************************************
        Implementation of a method that can be used to call
        any method on the ChareMlogData of all the chares on
        a processor currently
******************************************************/


class ElementCaller :  public CkLocIterator {
private:
        CkLocMgr *locMgr;
        MlogFn fnPointer;
        void *data;
public:
        ElementCaller(CkLocMgr * _locMgr, MlogFn _fnPointer,void *_data){
                locMgr = _locMgr;
                fnPointer = _fnPointer;
                data = _data;
        };
        void addLocation(CkLocation &loc){
                CkVec<CkMigratable *> list;
                CkLocRec_local *local = loc.getLocalRecord();
                locMgr->migratableList (local,list);
                for(int i=0;i<list.size();i++){
                        CkMigratable *migratableElement = list[i];
                        fnPointer(data,migratableElement->mlogData);
                }
        }
};

void forAllCharesDo(MlogFn fnPointer, void *data){
        int numGroups = CkpvAccess(_groupIDTable)->size();
        for(int i=0;i<numGroups;i++){
                Chare *obj = (Chare *)CkpvAccess(_groupTable)->find((*CkpvAccess(_groupIDTable))[i]).getObj();
                fnPointer(data,obj->mlogData);
        }
        int numNodeGroups = CksvAccess(_nodeGroupIDTable).size();
        for(int i=0;i<numNodeGroups;i++){
                Chare *obj = (Chare *)CksvAccess(_nodeGroupTable)->find(CksvAccess(_nodeGroupIDTable)[i]).getObj();
                fnPointer(data,obj->mlogData);
        }
        int i;
        CKLOCMGR_LOOP(ElementCaller caller(mgr, fnPointer,data); mgr->iterate(caller););
};


/******************************************************************
 Load Balancing
******************************************************************/

void initMlogLBStep(CkGroupID gid){
        DEBUGLB(CkPrintf("[%d] INIT MLOG STEP\n",CkMyPe()));
        countLBMigratedAway = 0;
        countLBToMigrate=0;
        onGoingLoadBalancing=1;
        migrationDoneCalled=0;
        checkpointBarrierCount=0;
        if(globalLBID.idx != 0){
                CmiAssert(globalLBID.idx == gid.idx);
        }
        globalLBID = gid;
#if SYNCHRONIZED_CHECKPOINT
        garbageCollectMlog();
#endif
}

void startLoadBalancingMlog(void (*_fnPtr)(void *),void *_centralLb){
        DEBUGLB(printf("[%d] start Load balancing section of message logging \n",CmiMyPe()));
        DEBUG_TEAM(printf("[%d] start Load balancing section of message logging \n",CmiMyPe()));
        
        resumeLbFnPtr = _fnPtr;
        centralLb = _centralLb;
        migrationDoneCalled = 1;
        if(countLBToMigrate == countLBMigratedAway){
                DEBUGLB(printf("[%d] calling startMlogCheckpoint in startLoadBalancingMlog countLBToMigrate %d countLBMigratedAway %d \n",CmiMyPe(),countLBToMigrate,countLBMigratedAway));
                startMlogCheckpoint(NULL,CmiWallTimer());       
        }
};

void finishedCheckpointLoadBalancing(){
        DEBUGLB(printf("[%d] finished checkpoint after lb \n",CmiMyPe());)
        CheckpointBarrierMsg msg;
        msg.fromPE = CmiMyPe();
        msg.checkpointCount = checkpointCount;

        CmiSetHandler(&msg,_checkpointBarrierHandlerIdx);
        CmiSyncSend(0,sizeof(CheckpointBarrierMsg),(char *)&msg);
        
};


void sendMlogLocation(int targetPE, envelope *env){
#if !SYNCHRONIZED_CHECKPOINT
        void *_msg = EnvToUsr(env);
        CkArrayElementMigrateMessage *msg = (CkArrayElementMigrateMessage *)_msg;

        int existing = 0;
        //if this object is already in the retainedobjectlust destined for this
        //processor it should not be sent
        
        for(int i=0;i<retainedObjectList.size();i++){
                MigrationRecord &migRecord = retainedObjectList[i]->migRecord;
                if(migRecord.gID == msg->gid && migRecord.idx == msg->idx){
                        DEBUG(CmiPrintf("[%d] gid %d idx %s being sent to %d exists in retainedObjectList with toPE %d\n",CmiMyPe(),msg->gid.idx,idx2str(msg->idx),targetPE,migRecord.toPE));
                        existing = 1;
                        break;
                }
        }

        if(existing){
                return;
        }
        
        countLBToMigrate++;
        
        MigrationNotice migMsg;
        migMsg.migRecord.gID = msg->gid;
        migMsg.migRecord.idx = msg->idx;
        migMsg.migRecord.fromPE = CkMyPe();
        migMsg.migRecord.toPE =  targetPE;
        
        DEBUGLB(printf("[%d] Sending array to proc %d gid %d idx %s\n",CmiMyPe(),targetPE,msg->gid.idx,idx2str(msg->idx)));
        
        RetainedMigratedObject  *retainedObject = new RetainedMigratedObject;
        retainedObject->migRecord = migMsg.migRecord;
        retainedObject->acked  = 0;
        
        CkPackMessage(&env);
        
        migMsg.record = retainedObject;
        retainedObject->msg = env;
        int size = retainedObject->size = env->getTotalsize();
        
        retainedObjectList.push_back(retainedObject);
        
        CmiSetHandler((void *)&migMsg,_receiveMigrationNoticeHandlerIdx);
        CmiSyncSend(getCheckPointPE(),sizeof(migMsg),(char *)&migMsg);
        
        DEBUGLB(printf("[%d] Location in message of size %d being sent to PE %d\n",CkMyPe(),size,targetPE));
#endif
}

void _receiveMigrationNoticeHandler(MigrationNotice *msg){
        msg->migRecord.ackFrom = msg->migRecord.ackTo = 0;
        migratedNoticeList.push_back(msg->migRecord);

        MigrationNoticeAck buf;
        buf.record = msg->record;
        CmiSetHandler((void *)&buf,_receiveMigrationNoticeAckHandlerIdx);
        CmiSyncSend(getCheckPointPE(),sizeof(MigrationNoticeAck),(char *)&buf);
}

void _receiveMigrationNoticeAckHandler(MigrationNoticeAck *msg){
        
        RetainedMigratedObject *retainedObject = (RetainedMigratedObject *)(msg->record);
        retainedObject->acked = 1;

        CmiSetHandler(retainedObject->msg,_receiveMlogLocationHandlerIdx);
        CmiSyncSend(retainedObject->migRecord.toPE,retainedObject->size,(char *)retainedObject->msg);

        //inform home about the new location of this object
        CkGroupID gID = retainedObject->migRecord.gID ;
        CkLocMgr *mgr = (CkLocMgr*)CkpvAccess(_groupTable)->find(gID).getObj();
        informLocationHome(gID,retainedObject->migRecord.idx, mgr->homePe(retainedObject->migRecord.idx),retainedObject->migRecord.toPE);
        
        countLBMigratedAway++;
        if(countLBMigratedAway == countLBToMigrate && migrationDoneCalled == 1){
                DEBUGLB(printf("[%d] calling startMlogCheckpoint in _receiveMigrationNoticeAckHandler countLBToMigrate %d countLBMigratedAway %d \n",CmiMyPe(),countLBToMigrate,countLBMigratedAway));
                startMlogCheckpoint(NULL,CmiWallTimer());
        }
};

void _receiveMlogLocationHandler(void *buf){
        envelope *env = (envelope *)buf;
        DEBUG(printf("[%d] Location received in message of size %d\n",CkMyPe(),env->getTotalsize()));
        CkUnpackMessage(&env);
        void *_msg = EnvToUsr(env);
        CkArrayElementMigrateMessage *msg = (CkArrayElementMigrateMessage *)_msg;
        CkGroupID gID= msg->gid;
        DEBUG(printf("[%d] Object to be inserted into location manager %d\n",CkMyPe(),gID.idx));
        CkLocMgr *mgr = (CkLocMgr*)CkpvAccess(_groupTable)->find(gID).getObj();
        CpvAccess(_currentObj)=mgr;
        mgr->immigrate(msg);
};


void resumeFromSyncRestart(void *data,ChareMlogData *mlogData){
/*      if(mlogData->objID.type == TypeArray){
                CkMigratable *elt = (CkMigratable *)mlogData->objID.getObject();
        //      TODO: make sure later that atSync has been called and it needs 
        //      to be resumed from sync
        //
                CpvAccess(_currentObj) = elt;
                elt->ResumeFromSync();
        }*/
}

inline void checkAndSendCheckpointBarrierAcks(CheckpointBarrierMsg *msg){
        if(checkpointBarrierCount == CmiNumPes()){
                CmiSetHandler(msg,_checkpointBarrierAckHandlerIdx);
                for(int i=0;i<CmiNumPes();i++){
                        CmiSyncSend(i,sizeof(CheckpointBarrierMsg),(char *)msg);
                }
        }
}

void _checkpointBarrierHandler(CheckpointBarrierMsg *msg){
        DEBUG(CmiPrintf("[%d] msg->checkpointCount %d pe %d checkpointCount %d checkpointBarrierCount %d \n",CmiMyPe(),msg->checkpointCount,msg->fromPE,checkpointCount,checkpointBarrierCount));
        if(msg->checkpointCount == checkpointCount){
                checkpointBarrierCount++;
                checkAndSendCheckpointBarrierAcks(msg);
        }else{
                if(msg->checkpointCount-1 == checkpointCount){
                        checkpointBarrierCount++;
                        checkAndSendCheckpointBarrierAcks(msg);
                }else{
                        printf("[%d] msg->checkpointCount %d checkpointCount %d\n",CmiMyPe(),msg->checkpointCount,checkpointCount);
                        CmiAbort("msg->checkpointCount and checkpointCount differ by more than 1");
                }
        }

        // deleting the received message
        CmiFree(msg);
}

void _checkpointBarrierAckHandler(CheckpointBarrierMsg *msg){
        DEBUG(CmiPrintf("[%d] _checkpointBarrierAckHandler \n",CmiMyPe()));
        DEBUGLB(CkPrintf("[%d] Reaching this point\n",CkMyPe()));

#if !SYNCHRONIZED_CHECKPOINT
        // sending a notice to all senders to remove message logs
        sendRemoveLogRequests();
#endif

        // resuming LB function pointer
        (*resumeLbFnPtr)(centralLb);

        // deleting message
        CmiFree(msg);
}

void garbageCollectMlogForChare(void *data, ChareMlogData *mlogData){
        int total;
        MlogEntry *logEntry;
        CkQ<MlogEntry *> *mlog = mlogData->getMlog();

        // traversing the whole message log and removing all elements
        total = mlog->length();
        for(int i=0; i<total; i++){
                logEntry = mlog->deq();
                delete logEntry;
        }

}

void garbageCollectMlog(){
        CkHashtableIterator *iterator;
        CkVec<Determinant> *detArray;

        DEBUG(CkPrintf("[%d] Garbage collecting message log and data structures\n", CkMyPe()));

        // cleaning up the buffered determinants, since they belong to a previous checkpoint period
        _indexBufferedDets = 0;
        _numBufferedDets = 0;
        _phaseBufferedDets++;

        // cleaning up remote determinants, since they belong to a previous checkpoint period
        iterator = CpvAccess(_remoteDets)->iterator();
        while(iterator->hasNext()){
                detArray = *(CkVec<Determinant> **)iterator->next();
                detArray->removeAll();
        }
        
        // deleting the iterator
        delete iterator;

        // removing all messages in message log for every chare
        forAllCharesDo(garbageCollectMlogForChare, NULL);
}

void informLocationHome(CkGroupID locMgrID,CkArrayIndexMax idx,int homePE,int currentPE){
        double _startTime = CmiWallTimer();
        CurrentLocationMsg msg;
        msg.mgrID = locMgrID;
        msg.idx = idx;
        msg.locationPE = currentPE;
        msg.fromPE = CkMyPe();

        DEBUG(CmiPrintf("[%d] informing home %d of location %d of gid %d idx %s \n",CmiMyPe(),homePE,currentPE,locMgrID.idx,idx2str(idx)));
        CmiSetHandler(&msg,_receiveLocationHandlerIdx);
        CmiSyncSend(homePE,sizeof(CurrentLocationMsg),(char *)&msg);
        traceUserBracketEvent(37,_startTime,CmiWallTimer());
}


void _receiveLocationHandler(CurrentLocationMsg *data){
        double _startTime = CmiWallTimer();
        CkLocMgr *mgr =  (CkLocMgr*)CkpvAccess(_groupTable)->find(data->mgrID).getObj();
        if(mgr == NULL){
                CmiFree(data);
                return;
        }
        CkLocRec *rec = mgr->elementNrec(data->idx);
        DEBUG(CmiPrintf("[%d] location from %d is %d for gid %d idx %s rec %p \n",CkMyPe(),data->fromPE,data->locationPE,data->mgrID,idx2str(data->idx),rec));
        if(rec != NULL){
                if(mgr->lastKnown(data->idx) == CmiMyPe() && data->locationPE != CmiMyPe() && rec->type() == CkLocRec::local){
                        if(data->fromPE == data->locationPE){
                                CmiAbort("Another processor has the same object");
                        }
                }
        }
        if(rec!= NULL && rec->type() == CkLocRec::local && data->fromPE != CmiMyPe()){
                int targetPE = data->fromPE;
                data->fromPE = CmiMyPe();
                data->locationPE = CmiMyPe();
                DEBUG(printf("[%d] WARNING!! informing proc %d of current location\n",CmiMyPe(),targetPE));
                CmiSyncSend(targetPE,sizeof(CurrentLocationMsg),(char *)data);
        }else{
                mgr->inform(data->idx,data->locationPE);
        }
        CmiFree(data);
        traceUserBracketEvent(38,_startTime,CmiWallTimer());
}



void getGlobalStep(CkGroupID gID){
        LBStepMsg msg;
        int destPE = 0;
        msg.lbID = gID;
        msg.fromPE = CmiMyPe();
        msg.step = -1;
        CmiSetHandler(&msg,_getGlobalStepHandlerIdx);
        CmiSyncSend(destPE,sizeof(LBStepMsg),(char *)&msg);
};

void _getGlobalStepHandler(LBStepMsg *msg){
        CentralLB *lb = (CentralLB *)CkpvAccess(_groupTable)->find(msg->lbID).getObj();
        msg->step = lb->step();
        CmiAssert(msg->fromPE != CmiMyPe());
        CmiPrintf("[%d] getGlobalStep called from %d step %d gid %d \n",CmiMyPe(),msg->fromPE,lb->step(),msg->lbID.idx);
        CmiSetHandler(msg,_recvGlobalStepHandlerIdx);
        CmiSyncSend(msg->fromPE,sizeof(LBStepMsg),(char *)msg);
};

void _recvGlobalStepHandler(LBStepMsg *msg){
        
        // updating restart decision number
        restartDecisionNumber = msg->step;
        CmiFree(msg);

        CmiPrintf("[%d] recvGlobalStepHandler \n",CmiMyPe());

        // sending a dummy message to sendDetsReplyHandler
        ResendRequest *resendReplyMsg = (ResendRequest *)CmiAlloc(sizeof(ResendRequest));
        resendReplyMsg->PE = CkMyPe();
        resendReplyMsg->numberObjects = 0;
        _sendDetsReplyHandler((char *)resendReplyMsg);
};

void _messageLoggingExit(){
        
        // printing the signature for causal message logging
        if(CkMyPe() == 0)
                printf("[%d] _causalMessageLoggingExit \n",CmiMyPe());

        //TML: printing some statistics for group approach
#if COLLECT_STATS_TEAM
        printf("[%d] LOGGED MESSAGES: %.0f\n",CkMyPe(),MLOGFT_totalMessages);
        printf("[%d] MESSAGE LOG SIZE: %.2f MB\n",CkMyPe(),MLOGFT_totalLogSize/(float)MEGABYTE);
#endif

#if COLLECT_STATS_MSGS
#if COLLECT_STATS_MSGS_TOTAL
        printf("[%d] TOTAL MESSAGES SENT: %d\n",CmiMyPe(),totalMsgsTarget);
        printf("[%d] TOTAL MESSAGES SENT SIZE: %.2f MB\n",CmiMyPe(),totalMsgsSize/(float)MEGABYTE);
#else
        printf("[%d] TARGETS: ",CmiMyPe());
        for(int i=0; i<CmiNumPes(); i++){
#if COLLECT_STATS_MSG_COUNT
                printf("%d ",numMsgsTarget[i]);
#else
                printf("%d ",sizeMsgsTarget[i]);
#endif
        }
        printf("\n");
#endif
#endif

#if COLLECT_STATS_DETS
        printf("\n");
        printf("[%d] DETS: %d\n",CmiMyPe(),numDets);
        printf("[%d] PIGGYBACKED DETS: %d\n",CmiMyPe(),numPiggyDets);
#if COLLECT_STATS_DETS_DUP
        printf("[%d] DUPLICATED DETS: %d\n",CmiMyPe(),numDupDets);
#endif
#endif

}

void* CkObjID::getObject(){
        
                switch(type){
                        case TypeChare: 
                                return CkLocalChare(&data.chare.id);
                        case TypeMainChare:
                                return CkLocalChare(&data.chare.id);
                        case TypeGroup:
        
                                CkAssert(data.group.onPE == CkMyPe());
                                return CkLocalBranch(data.group.id);
                        case TypeNodeGroup:
                                CkAssert(data.group.onPE == CkMyNode());
                                //CkLocalNodeBranch(data.group.id);
                                {
                                        CmiImmediateLock(CksvAccess(_nodeGroupTableImmLock));
                                  void *retval = CksvAccess(_nodeGroupTable)->find(data.group.id).getObj();
                                  CmiImmediateUnlock(CksvAccess(_nodeGroupTableImmLock));                                       
        
                                        return retval;
                                }       
                        case TypeArray:
                                {
        
        
                                        CkArrayID aid(data.array.id);
        
                                        if(aid.ckLocalBranch() == NULL){ return NULL;}
        
                                        CProxyElement_ArrayBase aProxy(aid,data.array.idx.asChild());
        
                                        return aProxy.ckLocal();
                                }
                        default:
                                CkAssert(0);
                }
}


int CkObjID::guessPE(){
                switch(type){
                        case TypeChare:
                        case TypeMainChare:
                                return data.chare.id.onPE;
                        case TypeGroup:
                        case TypeNodeGroup:
                                return data.group.onPE;
                        case TypeArray:
                                {
                                        CkArrayID aid(data.array.id);
                                        if(aid.ckLocalBranch() == NULL){
                                                return -1;
                                        }
                                        return aid.ckLocalBranch()->lastKnown(data.array.idx.asChild());
                                }
                        default:
                                CkAssert(0);
                }
};

char *CkObjID::toString(char *buf) const {
        
        switch(type){
                case TypeChare:
                        sprintf(buf,"Chare %p PE %d \0",data.chare.id.objPtr,data.chare.id.onPE);
                        break;
                case TypeMainChare:
                        sprintf(buf,"Chare %p PE %d \0",data.chare.id.objPtr,data.chare.id.onPE);       
                        break;
                case TypeGroup:
                        sprintf(buf,"Group %d   PE %d \0",data.group.id.idx,data.group.onPE);
                        break;
                case TypeNodeGroup:
                        sprintf(buf,"NodeGroup %d       Node %d \0",data.group.id.idx,data.group.onPE);
                        break;
                case TypeArray:
                        {
                                const CkArrayIndexMax &idx = data.array.idx.asChild();
                                const int *indexData = idx.data();
                                sprintf(buf,"Array |%d %d %d| id %d \0",indexData[0],indexData[1],indexData[2],data.array.id.idx);
                                break;
                        }
                default:
                        CkAssert(0);
        }
        
        return buf;
};

void CkObjID::updatePosition(int PE){
        if(guessPE() == PE){
                return;
        }
        switch(type){
                case TypeArray:
                        {
                                        CkArrayID aid(data.array.id);
                                        if(aid.ckLocalBranch() == NULL){
                                                
                                        }else{
                                                char str[100];
                                                CkLocMgr *mgr = aid.ckLocalBranch()->getLocMgr();
//                                              CmiPrintf("[%d] location for object %s is %d\n",CmiMyPe(),toString(str),PE);
                                                CkLocRec *rec = mgr->elementNrec(data.array.idx.asChild());
                                                if(rec != NULL){
                                                        if(rec->type() == CkLocRec::local){
                                                                CmiPrintf("[%d] local object %s can not exist on another processor %d\n",CmiMyPe(),str,PE);
                                                                return;
                                                        }
                                                }
                                                mgr->inform(data.array.idx.asChild(),PE);
                                        }       
                                }

                        break;
                case TypeChare:
                case TypeMainChare:
                        CkAssert(data.chare.id.onPE == PE);
                        break;
                case TypeGroup:
                case TypeNodeGroup:
                        CkAssert(data.group.onPE == PE);
                        break;
                default:
                        CkAssert(0);
        }
}



void MlogEntry::pup(PUP::er &p){
        p | destPE;
        p | _infoIdx;
        int size;
        if(!p.isUnpacking()){
/*              CkAssert(env);
                if(!env->isPacked()){
                        CkPackMessage(&env);
                }*/
                if(env == NULL){
                        //message was probably local and has been removed from logs
                        size = 0;
                }else{
                        size = env->getTotalsize();
                }       
        }
        p | size;
        if(p.isUnpacking()){
                if(size > 0){
                        env = (envelope *)_allocEnv(ForChareMsg,size);
                }else{
                        env = NULL;
                }
        }
        if(size > 0){
                p((char *)env,size);
        }
};

void RestoredLocalMap::pup(PUP::er &p){
        p | minSN;
        p | maxSN;
        p | count;
        if(p.isUnpacking()){
                TNArray = new MCount[count];
        }
        p(TNArray,count);
};

/**********************************
        * The methods of the message logging
        * data structure stored in each chare
        ********************************/

MCount ChareMlogData::nextSN(const CkObjID &recver){
        MCount *SN = snTable.getPointer(recver);
        if(SN==NULL){
                snTable.put(recver) = 1;
                return 1;
        }else{
                (*SN)++;
                return *SN;
        }
};
 
MCount ChareMlogData::newTN(){
        MCount TN;
        if(currentHoles > 0){
                int holeidx = numberHoles-currentHoles;
                TN = ticketHoles[holeidx];
                currentHoles--;
                if(currentHoles == 0){
                        delete []ticketHoles;
                        numberHoles = 0;
                }
        }else{
                TN = ++tCount;
        }
        return TN;
};

inline Ticket ChareMlogData::getTicket(CkObjID &sender, MCount SN){
        Ticket ticket;

        SNToTicket *ticketRow = ticketTable.get(sender);
        if(ticketRow != NULL){
                return ticketRow->get(SN);
        }
        return ticket;
}

inline void ChareMlogData::verifyTicket(CkObjID &sender, MCount SN, MCount TN){
        Ticket ticket;

        SNToTicket *ticketRow = ticketTable.get(sender);
        if(ticketRow != NULL){
                Ticket earlierTicket = ticketRow->get(SN);
                if(earlierTicket.TN != 0){
                        CkAssert(earlierTicket.TN == TN);
                        return;
                }
        }else{
                ticketRow = new SNToTicket();
                ticketTable.put(sender) = ticketRow;
        }
        ticket.TN = TN;
        ticketRow->put(SN) = ticket;
}

inline Ticket ChareMlogData::next_ticket(CkObjID &sender, MCount SN){
        DEBUG(char senderName[100];)
        DEBUG(char recverName[100];)
        double _startTime = CmiWallTimer();
        Ticket ticket;

        // if a ticket is requested during restart, 0 is returned to make the requester to ask for it later.
        if(restartFlag){
                ticket.TN = 0;
                return ticket;
        }
        
        SNToTicket *ticketRow = ticketTable.get(sender);
        if(ticketRow != NULL){
                Ticket earlierTicket = ticketRow->get(SN);
                if(earlierTicket.TN == 0){
                        ticket.TN = newTN();
                        ticketRow->put(SN) = ticket;
                        DEBUG(CkAssert((ticketRow->get(SN)).TN == ticket.TN));
                }else{
                        ticket.TN = earlierTicket.TN;
                        if(ticket.TN > tCount){
                                DEBUG(CmiPrintf("[%d] next_ticket old row ticket sender %s recver %s SN %d TN %d tCount %d\n",CkMyPe(),sender.toString(senderName),objID.toString(recverName),SN,ticket.TN,tCount));
                        }
                                CmiAssert(ticket.TN <= tCount);
                }
                DEBUG(CmiPrintf("[%d] next_ticket old row ticket sender %s recver %s SN %d TN %d tCount %d\n",CkMyPe(),sender.toString(senderName),objID.toString(recverName),SN,ticket.TN,tCount));
        }else{
                SNToTicket *newRow = new SNToTicket;            
                ticket.TN = newTN();
                newRow->put(SN) = ticket;
                ticketTable.put(sender) = newRow;
                DEBUG(printf("[%d] next_ticket new row ticket sender %s recver %s SN %d TN %d\n",CkMyPe(),sender.toString(senderName),objID.toString(recverName),SN,ticket.TN));
        }
        ticket.state = NEW_TICKET;

//      traceUserBracketEvent(34,_startTime,CkWallTimer());
        return ticket;  
};

void ChareMlogData::addLogEntry(MlogEntry *entry){
        DEBUG(char nameString[100]);
        DEBUG(printf("[%d] Adding logEntry %p to the log of %s with SN %d\n",CkMyPe(),entry,objID.toString(nameString),entry->env->SN));
        DEBUG_MEM(CmiMemoryCheck());

        // enqueuing the entry in the message log
        mlog.enq(entry);
};

double totalSearchRestoredTime=0;
double totalSearchRestoredCount=0;

MCount ChareMlogData::searchRestoredLocalQ(CkObjID &sender,CkObjID &recver,MCount SN){
        double start= CkWallTimer();
        MCount TN=0;    
        
        DEBUG(char senderName[100]);
        DEBUG(char recverName[100]);
        DEBUG(if(TN != 0){ CmiPrintf("[%d] searchRestoredLocalQ found match sender %s recver %s SN %d TN %d\n",CmiMyPe(),sender.toString(senderName),recver.toString(recverName),SN,TN);});

        totalSearchRestoredTime += CkWallTimer()-start;
        totalSearchRestoredCount++;
        return TN;
}

void ChareMlogData::pup(PUP::er &p){
        int tCountAux;
        int startSize=0;
        char nameStr[100];
        if(p.isSizing()){
                PUP::sizer *sizep = (PUP::sizer *)&p;
                startSize = sizep->size();
        }
        double _startTime = CkWallTimer();
        
        p | objID;
        if(teamRecoveryFlag)
                p | tCountAux;
        else
                p | tCount;
        p | tProcessed;
        if(p.isUnpacking()){
                DEBUG(CmiPrintf("[%d] Obj %s being unpacked with tCount %d tProcessed %d \n",CmiMyPe(),objID.toString(nameStr),tCount,tProcessed));
        }
        p | toResumeOrNot;
        p | resumeCount;
        DEBUG(CmiPrintf("[%d] Obj %s toResumeOrNot %d resumeCount %d \n",CmiMyPe(),objID.toString(nameStr),toResumeOrNot,resumeCount));
        

        /*pack the receivedTN vector*/
        int lengthReceivedTNs;
        if(!p.isUnpacking()){
                if(receivedTNs == NULL){
                        lengthReceivedTNs = -1;
                }else{
                        lengthReceivedTNs = receivedTNs->size();                
                }
        }
        p | lengthReceivedTNs;
        if(p.isUnpacking()){
                if(lengthReceivedTNs == -1){
                        receivedTNs = NULL;
                }else{
                        receivedTNs = new CkVec<MCount>;
                        for(int i=0;i<lengthReceivedTNs;i++){
                                MCount tempTicket;
                                p | tempTicket;
                                CkAssert(tempTicket > 0);
                                receivedTNs->push_back(tempTicket);
                        }
                }
        }else{
                for(int i=0;i<lengthReceivedTNs;i++){
                        p | (*receivedTNs)[i];
                }
        }
        
        p | currentHoles;
        p | numberHoles;
        if(p.isUnpacking()){
                if(numberHoles > 0){
                        ticketHoles = new MCount[numberHoles];                  
                }else{
                        ticketHoles = NULL;
                }
        }
        if(numberHoles > 0){
                p(ticketHoles,numberHoles);
        }
        
        snTable.pup(p);


/*      int length;
        if(!p.isUnpacking()){           
                length = mlog.length(); 
                if(length > 0)
                        DEBUG(printf("[%d] Mlog length %d \n",CkMyPe(),length));
        }
        p | length;
        for(int i=0;i<length;i++){
                MlogEntry *entry;
                if(p.isUnpacking()){
                        entry = new MlogEntry();
                        mlog.enq(entry);
                }else{
                        entry = mlog[i];
                }
                entry->pup(p);
        }*/
        
        p | resendReplyRecvd;
        p | restartFlag;

        // pup the mapTable
/*      int tableSize;
        if(!p.isUnpacking()){
                tableSize = mapTable.numObjects();
        }
        p | tableSize;
        if(!p.isUnpacking()){
                CkHashtableIterator *iter = mapTable.iterator();
                while(iter->hasNext()){
                        CkObjID *objID;
                        RestoredLocalMap **map = (RestoredLocalMap **) iter->next((void **)&objID);
                        p | (*objID);
                        (*map)->pup(p);
                }
                // releasing memory for iterator
                delete iter;
        }else{
                for(int i=0;i<tableSize;i++){
                        CkObjID objID;
                        p | objID;
                        RestoredLocalMap *map = new RestoredLocalMap;
                        map->pup(p);
                        mapTable.put(objID) = map;
                }
        }*/

        //pup the ticketTable
        {
                int ticketTableSize;
                if(!p.isUnpacking()){
                        ticketTableSize = ticketTable.numObjects();
                }
                p | ticketTableSize;
                if(!p.isUnpacking()){
                        CkHashtableIterator *iter = ticketTable.iterator();
                        while(iter->hasNext()){
                                CkObjID *objID;
                                SNToTicket **ticketRow = (SNToTicket **)iter->next((void **)&objID);
                                p | (*objID);
                                (*ticketRow)->pup(p);
                        }
                        //releasing memory for iterator
                        delete iter;
                }else{
                        for(int i=0;i<ticketTableSize;i++){
                                CkObjID objID;
                                p | objID;
                                SNToTicket *ticketRow = new SNToTicket;
                                ticketRow->pup(p);
                                if(!teamRecoveryFlag)
                                        ticketTable.put(objID) = ticketRow;
                                else
                                        delete ticketRow;
                        }
                }
        }       
        
        if(p.isSizing()){
                PUP::sizer *sizep = (PUP::sizer *)&p;
                int pupSize = sizep->size()-startSize;
                DEBUG(char name[40]);
                DEBUG(CkPrintf("[%d]PUP::sizer of %s shows size %d\n",CkMyPe(),objID.toString(name),pupSize));
        //      CkAssert(pupSize <100000000);
        }
        
        double _finTime = CkWallTimer();
        DEBUG(CkPrintf("[%d] Pup took %.6lf\n",CkMyPe(),_finTime - _startTime));
};

/****************
*****************/

int getCheckPointPE(){
        return (CmiMyPe() + 1) % CmiNumPes();
}

//assume it is a packed envelope
envelope *copyEnvelope(envelope *env){
        envelope *newEnv = (envelope *)CmiAlloc(env->getTotalsize());
        memcpy(newEnv,env,env->getTotalsize());
        return newEnv;
}

/* Checks if two determinants are the same */
inline int isSameDet(Determinant *first, Determinant *second){
        return first->sender == second->sender && first->receiver == second->receiver && first->SN == second->SN && first->TN == second->TN;
}

#endif

---