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_LOGGING 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_PE_NOW(x,y)  if(CkMyPe() == x) y
#define DEBUG_RECOVERY(x) //x

#define FAIL_DET_THRESHOLD 10

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);

bool _recoveryFlag=false;
bool _restartFlag=false;
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;
int diskCkptFlag = 0;
static char fName[100];

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);

// storage for remove determinants header
CpvDeclare(RemoveDeterminantsHeader *, _removeDetsHeader);
// storage for store determinants header
CpvDeclare(StoreDeterminantsHeader *, _storeDetsHeader);
// storage for the sizes in vector-send to store determinants
CpvDeclare(int *, _storeDetsSizes);
// storage for the pointers in vector-send to store determinants
CpvDeclare(char **, _storeDetsPtrs);

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

/***** VARIABLES FOR PARALLEL RECOVERY *****/
CpvDeclare(int, _numEmigrantRecObjs);
CpvDeclare(int, _numImmigrantRecObjs);
CpvDeclare(std::vector<CkLocation *> *, _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_LOGGING
float MLOGFT_totalLogSize = 0.0;
float MLOGFT_totalMessages = 0.0;
float MLOGFT_totalMcastLogSize = 0.0;
float MLOGFT_totalReductionLogSize = 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 _sendBackLocationHandlerIdx;
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;

std::vector<MigrationRecord> migratedNoticeList;
std::vector<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;

#if CMK_CONVERSE_MPI
static int heartBeatHandlerIdx;
static int heartBeatCheckHandlerIdx;
static int partnerFailureHandlerIdx;
static double lastPingTime = -1;

void mpi_restart_crashed(int pe, int rank);
int  find_spare_mpirank(int pe, int partition);

void heartBeatPartner();
void heartBeatHandler(void *msg);
void heartBeatCheckHandler();
void partnerFailureHandler(char *msg);
int getReverseCheckPointPE();
int inCkptFlag = 0;
#endif

static void *doNothingMsg(int * size, void * data, void ** remote, int count){
    return data;
}

void _messageLoggingInit(){
    if(CkMyPe() == 0)
        CkPrintf("[%d] Causal Message Logging Support\n",CkMyPe());

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

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

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

    // handlers for causal message logging
    _storeDeterminantsHandlerIdx = CkRegisterHandler(_storeDeterminantsHandler);
    _removeDeterminantsHandlerIdx = CkRegisterHandler(_removeDeterminantsHandler);
    
    //handlers for load balancing
    _receiveMlogLocationHandlerIdx=CkRegisterHandler(_receiveMlogLocationHandler);
    _receiveMigrationNoticeHandlerIdx=CkRegisterHandler(_receiveMigrationNoticeHandler);
    _receiveMigrationNoticeAckHandlerIdx=CkRegisterHandler(_receiveMigrationNoticeAckHandler);
    _getGlobalStepHandlerIdx=CkRegisterHandler(_getGlobalStepHandler);
    _recvGlobalStepHandlerIdx=CkRegisterHandler(_recvGlobalStepHandler);
    _checkpointBarrierHandlerIdx=CkRegisterHandler(_checkpointBarrierHandler);
    _checkpointBarrierAckHandlerIdx=CkRegisterHandler(_checkpointBarrierAckHandler);
    
    //handlers for updating locations
    _receiveLocationHandlerIdx=CkRegisterHandler(_receiveLocationHandler);

   // handlers for failure detection in MPI layer
#if CMK_CONVERSE_MPI
    heartBeatHandlerIdx = CkRegisterHandler(heartBeatHandler);
    heartBeatCheckHandlerIdx = CkRegisterHandler(heartBeatCheckHandler);
    partnerFailureHandlerIdx = CkRegisterHandler(partnerFailureHandler);
#endif

    //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>, std::vector<Determinant> *> *),_remoteDets);
    CpvInitialize(char *, _incarnation);
    CpvInitialize(RemoveDeterminantsHeader *, _removeDetsHeader);
    CpvInitialize(StoreDeterminantsHeader *, _storeDetsHeader);
    CpvInitialize(int *, _storeDetsSizes);
    CpvInitialize(char **, _storeDetsPtrs);
    CpvAccess(_localDets) = (char *) CmiAlloc(_maxBufferedDets * sizeof(Determinant));
    CpvAccess(_remoteDets) = new CkHashtableT<CkHashtableAdaptorT<CkObjID>, std::vector<Determinant> *>(100, 0.4);
    CpvAccess(_incarnation) = (char *) CmiAlloc(CmiNumPes() * sizeof(int));
    for(int i=0; i<CmiNumPes(); i++){
        CpvAccess(_incarnation)[i] = 0;
    }
    CpvAccess(_removeDetsHeader) = (RemoveDeterminantsHeader *) CmiAlloc(sizeof(RemoveDeterminantsHeader));
    CpvAccess(_storeDetsHeader) = (StoreDeterminantsHeader *) CmiAlloc(sizeof(StoreDeterminantsHeader));
    CpvAccess(_storeDetsSizes) = (int *) CmiAlloc(sizeof(int) * 2);
    CpvAccess(_storeDetsPtrs) = (char **) CmiAlloc(sizeof(char *) * 2);

    // Cpv variables for parallel recovery
    CpvInitialize(int, _numEmigrantRecObjs);
    CpvAccess(_numEmigrantRecObjs) = 0;
    CpvInitialize(int, _numImmigrantRecObjs);
    CpvAccess(_numImmigrantRecObjs) = 0;

    CpvInitialize(std::vector<CkLocation *> *, _immigrantRecObjs);
    CpvAccess(_immigrantRecObjs) = new std::vector<CkLocation *>;

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

    // Disk checkpoint
    if(diskCkptFlag){
#if CMK_USE_MKSTEMP
        sprintf(fName, "/tmp/ckpt%d-XXXXXX", CkMyPe());
        mkstemp(fName);
#else
        fName=tmpnam(NULL);
#endif
    }

    // 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();

    // fault detection for MPI layer
#if CMK_CONVERSE_MPI
    void heartBeatPartner();
    void heartBeatCheckHandler();
    CcdCallOnCondition(CcdPERIODIC_100ms,(CcdVoidFn)heartBeatPartner,NULL);
    CcdCallOnCondition(CcdPERIODIC_5s,(CcdVoidFn)heartBeatCheckHandler,NULL);
#endif

#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

}

#if CMK_CONVERSE_MPI

void partnerFailureHandler(char *msg)
{
   int diepe = *(int *)(msg+CmiMsgHeaderSizeBytes);

   // send message to crash pe to let it restart
   int newrank = find_spare_mpirank(diepe, CmiMyPartition());
   int buddy = getReverseCheckPointPE();
   if (buddy == diepe)  {
     mpi_restart_crashed(diepe, newrank);
     CcdCallOnCondition(CcdPERIODIC_5s,(CcdVoidFn)heartBeatCheckHandler,NULL);
   }
}

void heartBeatHandler(void *msg)
{
    lastPingTime = CmiWallTimer();
    CmiFree(msg);
}

void heartBeatCheckHandler()
{
    double now = CmiWallTimer();
    if (lastPingTime > 0 && now - lastPingTime > FAIL_DET_THRESHOLD && !inCkptFlag) {
        int i, pe, buddy;
        // tell everyone that PE is down
        buddy = getReverseCheckPointPE();
        CmiPrintf("[%d] detected buddy processor %d died %f %f. \n", CmiMyPe(), buddy, now, lastPingTime);
        
        for (int pe = 0; pe < CmiNumPes(); pe++) {
            if (pe == buddy) continue;
            char *msg = (char*)CmiAlloc(CmiMsgHeaderSizeBytes+sizeof(int));
            *(int *)(msg+CmiMsgHeaderSizeBytes) = buddy;
            CmiSetHandler(msg, partnerFailureHandlerIdx);
            CmiSyncSendAndFree(pe, CmiMsgHeaderSizeBytes+sizeof(int), (char *)msg);
        }
    }
    else 
        CcdCallOnCondition(CcdPERIODIC_5s,(CcdVoidFn)heartBeatCheckHandler,NULL);
}

void heartBeatPartner()
{
    int buddy = getCheckPointPE();
    //printf("[%d] heartBeatPartner %d\n", CmiMyPe(), buddy);
    char *msg = (char*)CmiAlloc(CmiMsgHeaderSizeBytes+sizeof(int));
    *(int *)(msg+CmiMsgHeaderSizeBytes) = CmiMyPe();
    CmiSetHandler(msg, heartBeatHandlerIdx);
    CmiSyncSendAndFree(buddy, CmiMsgHeaderSizeBytes+sizeof(int), (char *)msg);
    CcdCallOnCondition(CcdPERIODIC_100ms,(CcdVoidFn)heartBeatPartner,NULL);
}
#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{
#if CMK_CONVERSE_MPI
        CkDieNow(); 
#else
        kill(getpid(),SIGKILL);
#endif
    }
}

#if ! CMK_CONVERSE_MPI
void CkDieNow()
{
    // kill -9 for non-mpi version
    CmiPrintf("[%d] die now.\n", CmiMyPe());
    killTime = CmiWallTimer()+0.001;
    CcdCallFnAfter(killLocal,NULL,1);
}
#endif

/*** 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->getArrayMgr();
    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];
    char senderString[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;
    }

    // checking if this message should bypass determinants in message-logging
    if(env->flags & CK_BYPASS_DET_MLOG){
        env->sender = CpvAccess(_currentObj)->mlogData->objID;
        env->recver = recver;
        DEBUG(CkPrintf("[%d] Bypassing determinants from %s to %s PE %d\n",CkMyPe(),CpvAccess(_currentObj)->mlogData->objID.toString(senderString),recver.toString(recverName),destPE));
        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;
    }
    
//  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));
    }*/
        
//  CkPackMessage(&(mEntry->env));
//  traceUserBracketEvent(32,_startTime,CkWallTimer());
    
    _startTime = CkWallTimer();

    // uses the proper ticketing mechanism for local, team and general messages
    if(isLocal(destPE)){
        sendLocalMsg(env, _infoIdx);
    }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
        MlogEntry *mEntry = new MlogEntry(env,destPE,_infoIdx);
        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;

    envelope *env = entry->env;
    DEBUG(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_LOGGING
            MLOGFT_totalMessages += 1.0;
            MLOGFT_totalLogSize += entry->env->getTotalsize();
            if(entry->env->flags & CK_MULTICAST_MSG_MLOG){
                MLOGFT_totalMcastLogSize += entry->env->getTotalsize(); 
            }
            if(entry->env->flags & CK_REDUCTION_MSG_MLOG){
                MLOGFT_totalReductionLogSize += entry->env->getTotalsize(); 
            }

#endif
        }else{
            // the message has to be deleted after it has been sent
            entry->env->flags = entry->env->flags | CK_FREE_MSG_MLOG;
        }
    }

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

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

    // 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(envelope *env, int _infoIdx){
    DEBUG_PERF(double _startTime=CkWallTimer());
    DEBUG_MEM(CmiMemoryCheck());
    DEBUG(Chare *senderObj = (Chare *)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(),env->SN,env->sender.toString(senderString),env->recver.toString(recverString)));

    // getting the receiver local object
    Chare *recverObj = (Chare *)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(),env,_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;
    Determinant *detPtr, det;
    int i, n, index, phase, destPE;
    std::vector<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.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 std::vector<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
    CpvAccess(_removeDetsHeader)->index = index;
    CpvAccess(_removeDetsHeader)->phase = phase;
    CmiSetHandler(CpvAccess(_removeDetsHeader),_removeDeterminantsHandlerIdx);
    CmiSyncSend(destPE, sizeof(RemoveDeterminantsHeader), (char *)CpvAccess(_removeDetsHeader));
    
}

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;

    // checking for determinants bypass in message logging
    if(env->flags & CK_BYPASS_DET_MLOG){
        DEBUG(printf("[%d] Bypassing message sender %s recver %s \n",CkMyPe(),env->sender.toString(senderString), recver.toString(recverString)));
        return 1;   
    }

    // checking if receiver is fault aware
    if(!fault_aware(recver)){
        CkPrintf("[%d] Receiver NOT fault aware\n",CkMyPe());
        return 1;
    }

    // checking if receiver is NULL
    Chare *obj = (Chare *)recver.getObject();
    *objPointer = obj;
    if(obj == NULL){
        if(_recoveryFlag){
            int possiblePE = recver.guessPE();
            if(possiblePE != CkMyPe()){
                int totalSize = env->getTotalsize();
                CmiSyncSendAndFree(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;
        } else {
            return 1;
        }
    } 

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

    DEBUG_MEM(CmiMemoryCheck());
    DEBUG_PE(2,printf("[%d] Message received, sender = %s SN %d TN %d tProcessed %d for recver %s at %.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(2,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
*****************************************************************************/
std::vector<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;
    CmiInitMsgHeader(request.header, sizeof(CheckpointRequest));
    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++;
    _recoveryFlag = false;

#if CMK_CONVERSE_MPI
    inCkptFlag = 1;
#endif
    
#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,true);
    DEBUG_MEM(CmiMemoryCheck());
    CkPupNodeGroupData(psizer,true);
    DEBUG_MEM(CmiMemoryCheck());
    pupArrayElementsSkip(psizer,true,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,true);
    CkPupNodeGroupData(pBuf,true);
    pupArrayElementsSkip(pBuf,true,NULL);

    unAckedCheckpoint=1;
    CmiSetHandler(msg,_storeCheckpointHandlerIdx);
    CmiSyncSendAndFree(getCheckPointPE(),totalSize,msg);
    
    /*
        Store the highest Ticket number processed for each chare on this processor
    */
    processedTicketLog.clear();
    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]);

    std::vector<TProcessedLog> *log = (std::vector<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, bool create, MigrationRecord *listToSkip,int listsize){
    int numElements,i;
    int numGroups = CkpvAccess(_groupIDTable)->size();  
    if(!p.isUnpacking()){
        numElements = CkCountArrayElements();
    }   
    //cppcheck-suppress uninitvar
    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 readCheckpointFromDisk(int size, char *data){
    FILE *f = fopen(fName,"rb");
    fread(data,1,size,f);
    fclose(f);
}

void writeCheckpointToDisk(int size, char *data){
    FILE *f = fopen(fName,"wb");
    fwrite(data, 1, size, f);
    fclose(f);
}

//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);
    }
    
    int sendingPE = chkMsg->PE;
    
    CpvAccess(_storedCheckpointData)->buf = chkpt;
    CpvAccess(_storedCheckpointData)->bufSize = chkMsg->dataSize;
    CpvAccess(_storedCheckpointData)->PE = sendingPE;

    // storing checkpoint in disk
    if(diskCkptFlag){
        writeCheckpointToDisk(chkMsg->dataSize,chkpt);
        CpvAccess(_storedCheckpointData)->buf = NULL;
        CmiFree(msg);
    }

    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.erase(migratedNoticeList.begin() + j);
            count++;
        }
        
    }
    DEBUG(printf("[%d] For proc %d from number of migratedNoticeList cleared %d checkpointAckHandler %d\n",CmiMyPe(),sendingPE,count,_checkpointAckHandlerIdx));
    
    CheckPointAck ackMsg;
    CmiInitMsgHeader(ackMsg.header, sizeof(CheckPointAck));
    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.data(),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);  //make it broadcast
    }
    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.erase(migratedNoticeList.begin() + 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.erase(retainedObjectList.begin() + j);
            if(obj->msg != NULL){
                CmiMemoryCheck();
                CmiFree(obj->msg);
            }
            delete obj;
        }
    }
}

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

void CkMlogRestart(const char * dummy, CkArgMsg * dummyMsg){
    RestartRequest msg;
    CmiInitMsgHeader(msg.header, sizeof(RestartRequest));
    fprintf(stderr,"[%d] Restart started at %.6lf \n",CkMyPe(),CmiWallTimer());

    // setting the restart flag
    _restartFlag = true;
    _recoveryFlag = true;
    _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;
    CmiInitMsgHeader(msg.header, sizeof(RestartRequest));
    fprintf(stderr,"[%d] Restart-team started at %.6lf \n",CkMyPe(),CmiWallTimer());

    // setting the restart flag
    _restartFlag = true;
    _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;
                CmiInitMsgHeader(msg.header, sizeof(VerifyAckMsg));
                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,false);
    CkPupNodeGroupData(pBuf,false);
    pupArrayElementsSkip(pBuf,false,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
    std::vector<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); // make it a broadcast
        }   
    }
    _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;
                CmiInitMsgHeader(msg.header, sizeof(VerifyAckMsg));
                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) {
            //this location exists on this processor
            //and needs to be removed   
            CmiPrintf("[%d] Found element gid %d idx %s that needs to be removed\n",CmiMyPe(),verifyRequest->migRecord.gID.idx,idx2str(verifyRequest->migRecord.idx));
            
            LBDatabase *lbdb = rec->getLBDB();
            LDObjHandle ldHandle = rec->getLdHandle();
                
            locMgr->setDuringMigration(true);
            
            locMgr->reclaim(verifyRequest->migRecord.idx);
            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.data(),migratedNoticeList.size()*sizeof(MigrationRecord));
        buf = &buf[migratedNoticeList.size()*sizeof(MigrationRecord)];
    }

    if(diskCkptFlag){
        readCheckpointFromDisk(storedChkpt->bufSize,buf);
    } else {    
        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 std::vector<TProcessedLog> *
void createObjIDList(void *data,ChareMlogData *mlogData){
    std::vector<TProcessedLog> *list = (std::vector<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,true);
    CkPupNodeGroupData(pBuf,true);
    pupArrayElementsSkip(pBuf,true,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
    std::vector<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);
    CmiSyncBroadcastAndFree(totalSize,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
    std::vector<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);
    CmiSyncBroadcastAllAndFree(totalSize, resendMsg);

};

void initializeRestart(void *data, ChareMlogData *mlogData){
    mlogData->resendReplyRecvd = 0;
    mlogData->receivedTNs = new std::vector<MCount>;
    mlogData->restartFlag = true;
};

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;
                CmiInitMsgHeader(migMsg.header, sizeof(MigrationNotice));
                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 = true;
}

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

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;
    std::vector<Determinant> *detVec;
    ResendRequest *resendReq = (ResendRequest *)msg;

    // CkPrintf("[%d] Sending determinants\n",CkMyPe());
    
    // cleaning global reduction sequence number
    CmiResetGlobalReduceSeqID();

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

    // adding the remote determinants to the resendReplyMsg
    // traversing all the remote determinants
    std::vector<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].data(),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(std::vector<MCount> *TNvec);
void sortVec(std::vector<MCount> *TNvec);
int searchVec(std::vector<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
    std::vector<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);
    CmiSyncBroadcastAllAndFree(totalSize, 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 = false;

        // 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(std::vector<MCount> *TNvec){
    MCount max = 0;
    for(int i=0; i<TNvec->size(); i++){
        if((*TNvec)[i] > max)
            max = (*TNvec)[i];
    }
    return max;
}

void sortVec(std::vector<MCount> *TNvec){
    std::sort(TNvec->begin(), TNvec->end());
    std::erase(std::unique(TNvec->begin(), TNvec->end()), TNvec->end());
}   

int searchVec(std::vector<MCount> *TNVec,MCount searchTN){
    // Find first element greater than or equal to searchTN
    auto it = std::lower_bound(TNvec->begin(), TNvec->end(), searchTN);
    if (it == TNvec.end() || *it != searchTN)
        return -1;
    else
        return it - TNvec->begin();
};


/*
    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 *rec = loc.getLocalRecord();
        CkLocMgr *locMgr = loc.getManager();
        std::vector<CkMigratable *> eltList;
            
        CkPrintf("[%d] Distributing objects to Processor %d: ",CkMyPe(),*targetPE);
        idx.print();

        // incrementing number of emigrant objects
        CpvAccess(_numEmigrantRecObjs)++;
        locMgr->migratableList((CkLocRec *)rec,eltList);
        CkReductionMgr *reductionMgr = (CkReductionMgr*)CkpvAccess(_groupTable)->find(eltList[0]->mlogData->objID.data.array.id).getObj();
        
        // let everybody else know the object is leaving
        locMgr->callMethod(rec,&CkMigratable::ckAboutToMigrate);
        reductionMgr->incNumEmigrantRecObjs();
    
        //pack up this location and send it across
        PUP::sizer psizer;
        pupLocation(loc,psizer);
        int totalSize = psizer.size() + sizeof(DistributeObjectMsg);
        char *msg = (char *)CmiAlloc(totalSize);
        DistributeObjectMsg *distributeMsg = (DistributeObjectMsg *)msg;
        distributeMsg->PE = CkMyPe();
        char *buf = &msg[sizeof(DistributeObjectMsg)];
        PUP::toMem pmem(buf);
        pmem.becomeDeleting();
        pupLocation(loc,pmem);
            
        locMgr->setDuringMigration(true);
        delete rec;
        locMgr->setDuringMigration(false);
        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 _sendBackLocationHandler(char *receivedMsg){
    printf("Array element received at processor %d after recovery\n",CkMyPe());
    DistributeObjectMsg *distributeMsg = (DistributeObjectMsg *)receivedMsg;
    int sourcePE = distributeMsg->PE;
    char *buf = &receivedMsg[sizeof(DistributeObjectMsg)];
    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,true);
    donotCountMigration=0;
    informLocationHome(gID,idx,mgr->homePe(idx),CkMyPe());
    printf("Array element inserted at processor %d after parallel recovery\n",CkMyPe());
    idx.print();

    // decrementing number of emigrant objects at reduction manager
    std::vector<CkMigratable *> eltList;
    CkLocRec *rec = mgr->elementRec(idx);
    mgr->migratableList((CkLocRec *)rec,eltList);
    CkReductionMgr *reductionMgr = (CkReductionMgr*)CkpvAccess(_groupTable)->find(eltList[0]->mlogData->objID.data.array.id).getObj();
    reductionMgr->decNumEmigrantRecObjs();
    reductionMgr->decGCount();

    // checking if it has received all emigrant recovering objects
    CpvAccess(_numEmigrantRecObjs)--;
    if(CpvAccess(_numEmigrantRecObjs) == 0){
        (*resumeLbFnPtr)(centralLb);
    }

}

void _distributedLocationHandler(char *receivedMsg){
    printf("Array element received at processor %d after distribution at restart\n",CkMyPe());
    DistributeObjectMsg *distributeMsg = (DistributeObjectMsg *)receivedMsg;
    int sourcePE = distributeMsg->PE;
    char *buf = &receivedMsg[sizeof(DistributeObjectMsg)];
    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,true);
    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 != NULL);

    // adding object to the list of immigrant recovery objects
    CpvAccess(_immigrantRecObjs)->push_back(new CkLocation(mgr,rec));
    CpvAccess(_numImmigrantRecObjs)++;
    
    std::vector<CkMigratable *> eltList;
    mgr->migratableList((CkLocRec *)rec,eltList);
    for(int i=0;i<eltList.size();i++){
        if(eltList[i]->mlogData->toResumeOrNot && eltList[i]->mlogData->resumeCount < globalResumeCount){
            CpvAccess(_currentObj) = eltList[i];
            eltList[i]->mlogData->immigrantRecFlag = true;
            eltList[i]->mlogData->immigrantSourcePE = sourcePE;

            // incrementing immigrant counter at reduction manager
            CkReductionMgr *reductionMgr = (CkReductionMgr*)CkpvAccess(_groupTable)->find(eltList[i]->mlogData->objID.data.array.id).getObj();
            reductionMgr->incNumImmigrantRecObjs();
            reductionMgr->decGCount();

            eltList[i]->ResumeFromSync();
        }
    }
}


void sendDummyMigration(int restartPE,CkGroupID lbID,CkGroupID locMgrID,CkArrayIndexMax &idx,int locationPE){
    DummyMigrationMsg buf;
    CmiInitMsgHeader(buf.header, sizeof(DummyMigrationMsg));
    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;
    CmiInitMsgHeader(buf.header, sizeof(DummyMigrationMsg));
    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){
        std::vector<CkMigratable *> list;
        CkLocRec *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 pupLocation(CkLocation *loc, CkLocMgr *locMgr, 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;
};

void sendBackImmigrantRecObjs(){
    CkLocation *loc;
    CkLocMgr *locMgr;
    CkArrayIndexMax idx;
    CkLocRec *rec;
    PUP::sizer psizer;
    int targetPE;
    std::vector<CkMigratable *> eltList;
    CkReductionMgr *reductionMgr;
 
    // looping through all elements in immigrant recovery objects vector
    for(int i=0; i<CpvAccess(_numImmigrantRecObjs); i++){

        // getting the components of each location
        loc = (*CpvAccess(_immigrantRecObjs))[i];
        idx = loc->getIndex();
        rec = loc->getLocalRecord();
        locMgr = loc->getManager();
        locMgr->migratableList(rec,eltList);
        targetPE = eltList[i]->mlogData->immigrantSourcePE;

        // decrement counter at array manager
        reductionMgr = (CkReductionMgr*)CkpvAccess(_groupTable)->find(eltList[i]->mlogData->objID.data.array.id).getObj();
        reductionMgr->decNumImmigrantRecObjs();

        CkPrintf("[%d] Sending back object to %d: ",CkMyPe(),targetPE);
        idx.print();

        // let everybody else know the object is leaving
        locMgr->callMethod(rec,&CkMigratable::ckAboutToMigrate);
            
        //pack up this location and send it across
        pupLocation(loc,locMgr,psizer);
        int totalSize = psizer.size() + sizeof(DistributeObjectMsg);
        char *msg = (char *)CmiAlloc(totalSize);
        DistributeObjectMsg *distributeMsg = (DistributeObjectMsg *)msg;
        distributeMsg->PE = CkMyPe();
        char *buf = &msg[sizeof(DistributeObjectMsg)];
        PUP::toMem pmem(buf);
        pmem.becomeDeleting();
        pupLocation(loc,locMgr,pmem);
        
        locMgr->setDuringMigration(true);
        delete rec;
        locMgr->setDuringMigration(false);
        locMgr->inform(idx,targetPE);

        // sending the object
        CmiSetHandler(msg,_sendBackLocationHandlerIdx);
        CmiSyncSendAndFree(targetPE,totalSize,msg);

        // freeing memory
        delete loc;

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

    // cleaning up all data structures
    CpvAccess(_immigrantRecObjs)->clear();
    CpvAccess(_numImmigrantRecObjs) = 0;

}

void restoreParallelRecovery(void (*_fnPtr)(void *),void *_centralLb){
    resumeLbFnPtr = _fnPtr;
    centralLb = _centralLb;

    // sending back the immigrant recovering objects
    if(CpvAccess(_numImmigrantRecObjs) > 0){
        sendBackImmigrantRecObjs(); 
    }

    // checking whether it needs to wait for emigrant recovery objects
    if(CpvAccess(_numEmigrantRecObjs) > 0)
        return;

    // otherwise, load balancing process is finished
    (*resumeLbFnPtr)(centralLb);
}

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());)

    char *msg = (char*)CmiAlloc(CmiMsgHeaderSizeBytes);
    CmiSetHandler(msg,_checkpointBarrierHandlerIdx);
    CmiReduce(msg,CmiMsgHeaderSizeBytes,doNothingMsg);
};


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;
    CmiInitMsgHeader(migMsg.header, sizeof(MigrationNotice));
    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;
    CmiInitMsgHeader(buf.header, sizeof(MigrationNoticeAck));
    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();
    }*/
}

void _checkpointBarrierHandler(CheckpointBarrierMsg *barrierMsg){

    // deleting the received message
    CmiFree(barrierMsg);

    // sending a broadcast to resume execution
    CheckpointBarrierMsg *msg = (CheckpointBarrierMsg *)CmiAlloc(sizeof(CheckpointBarrierMsg));
    CmiSetHandler(msg,_checkpointBarrierAckHandlerIdx);
    CmiSyncBroadcastAllAndFree(sizeof(CheckpointBarrierMsg),(char *)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

#if CMK_CONVERSE_MPI
    inCkptFlag = 0;
#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;
    std::vector<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 = *(std::vector<Determinant> **)iterator->next();
        detArray->clear();
    }
    
    // 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;
    CmiInitMsgHeader(msg.header, sizeof(CurrentLocationMsg));
    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()){
            if(data->fromPE == data->locationPE){
                CmiAbort("Another processor has the same object");
            }
        }
    }
    if(rec!= NULL && 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;
    CmiInitMsgHeader(msg.header, sizeof(LBStepMsg));
    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());

#if COLLECT_STATS_LOGGING
    printf("[%d] LOGGED MESSAGES: %.0f\n",CkMyPe(),MLOGFT_totalMessages);
    printf("[%d] MESSAGE LOG SIZE: %.2f MB\n",CkMyPe(),MLOGFT_totalLogSize/(float)MEGABYTE);
    printf("[%d] MULTICAST MESSAGE LOG SIZE: %.2f MB\n",CkMyPe(),MLOGFT_totalMcastLogSize/(float)MEGABYTE);
    printf("[%d] REDUCTION MESSAGE LOG SIZE: %.2f MB\n",CkMyPe(),MLOGFT_totalReductionLogSize/(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 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();
        }   
    }
    //cppcheck-suppress uninitvar
    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 = 0;
    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();        
        }
    }
    //cppcheck-suppress uninitvar
    p | lengthReceivedTNs;
    if(p.isUnpacking()){
        if(lengthReceivedTNs == -1){
            receivedTNs = NULL;
        }else{
            receivedTNs = new std::vector<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();
        }
        //cppcheck-suppress uninitvar
        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();
}

int getReverseCheckPointPE(){
    return (CmiMyPe() - 1 + CmiNumPes()) % 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 bool isSameDet(Determinant *first, Determinant *second){
    return first->sender == second->sender && first->receiver == second->receiver && first->SN == second->SN && first->TN == second->TN;
}

#endif

---