Home Research Papers Posters Manuals Talks Download People Help Internal

ck-core/init.C

Go to the documentation of this file.

#include "ckcheckpoint.h"
#include "ck.h"
#include "trace.h"
#include "ckrdma.h"
#include "CkCheckpoint.decl.h"
#include "ckmulticast.h"
#include <sstream>
#include <limits.h>
#include "spanningTree.h"
#if CMK_CHARMPY
#include "GreedyRefineLB.h"
#endif

#if CMK_CUDA
#include "hapi_impl.h"
#endif

void CkRestartMain(const char* dirname, CkArgMsg* args);

#define  DEBUGF(x)     //CmiPrintf x;

#define CMK_WITH_WARNINGS 0

#include "TopoManager.h"

UChar _defaultQueueing = CK_QUEUEING_FIFO;

UInt  _printCS = 0;
UInt  _printSS = 0;

UInt  _numExpectInitMsgs = 0;
UInt  _numInitMsgs = 0;
CksvDeclare(UInt,_numInitNodeMsgs);

#if CMK_ONESIDED_IMPL
UInt  numZerocopyROops;
UInt  curROIndex;
NcpyROBcastAckInfo *roBcastAckInfo;
int   _roRdmaDoneHandlerIdx;
CksvDeclare(int,  _numPendingRORdmaTransfers);
#endif

int   _infoIdx;
int   _charmHandlerIdx;
int   _initHandlerIdx;
int   _roRestartHandlerIdx;
int   _bocHandlerIdx;
int   _qdHandlerIdx;
int   _qdCommHandlerIdx;
int   _triggerHandlerIdx;
bool  _mainDone = false;
CksvDeclare(bool, _triggersSent);

CkOutStream ckout;
CkErrStream ckerr;
CkInStream  ckin;

CkpvDeclare(void*,       _currentChare);
CkpvDeclare(int,         _currentChareType);
CkpvDeclare(CkGroupID,   _currentGroup);
CkpvDeclare(void*,       _currentNodeGroupObj);
CkpvDeclare(CkGroupID,   _currentGroupRednMgr);
CkpvDeclare(GroupTable*, _groupTable);
CkpvDeclare(GroupIDTable*, _groupIDTable);
CkpvDeclare(CmiImmediateLockType, _groupTableImmLock);
CkpvDeclare(UInt, _numGroups);

CkpvDeclare(CkCoreState *, _coreState);

CksvDeclare(UInt, _numNodeGroups);
CksvDeclare(GroupTable*, _nodeGroupTable);
CksvDeclare(GroupIDTable, _nodeGroupIDTable);
CksvDeclare(CmiImmediateLockType, _nodeGroupTableImmLock);
CksvDeclare(CmiNodeLock, _nodeLock);
CksvStaticDeclare(PtrVec*,_nodeBocInitVec);
CkpvDeclare(int, _charmEpoch);

CkpvDeclare(bool, _destroyingNodeGroup);


CkpvDeclare(Stats*, _myStats);
CkpvDeclare(MsgPool*, _msgPool);

CkpvDeclare(_CkOutStream*, _ckout);
CkpvDeclare(_CkErrStream*, _ckerr);

CkpvStaticDeclare(int,  _numInitsRecd);
CkpvStaticDeclare(bool,  _initdone);
CkpvStaticDeclare(PtrQ*, _buffQ);
CkpvStaticDeclare(PtrVec*, _bocInitVec);

//for interoperability
extern int userDrivenMode;
extern void _libExitHandler(envelope *env);
extern int _libExitHandlerIdx;
CpvCExtern(int,interopExitFlag);
void StopInteropScheduler();

#if CMK_SHRINK_EXPAND
//for shrink expand cleanup
int _ROGroupRestartHandlerIdx;
const char* _shrinkexpand_basedir;
#endif

#if CMK_FAULT_EVAC
CpvExtern(char *, _validProcessors);
CkpvDeclare(char ,startedEvac);
#endif

int    _exitHandlerIdx;

#if CMK_WITH_STATS
static Stats** _allStats = 0;
#endif
static bool   _exitStarted = false;
static int _exitcode;

static InitCallTable _initCallTable;

#if CMK_WITH_STATS
#define _STATS_ON(x) (x) = 1
#else
#define _STATS_ON(x) \
          if (CkMyPe()==0) CmiPrintf("stats unavailable in optimized version. ignoring...\n");
#endif

// fault tolerance
typedef void (*CkFtFn)(const char *, CkArgMsg *);
static CkFtFn  faultFunc = NULL;
static char* _restartDir;

#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
int teamSize=1;
int chkptPeriod=1000;
bool fastRecovery = false;
int parallelRecovery = 1;
extern int BUFFER_TIME; //time spent waiting for buffered messages
#endif

// flag for killing processes 
extern bool killFlag;
// file specifying the processes to be killed
extern char *killFile;
// function for reading the kill file
void readKillFile();
#if CMK_MESSAGE_LOGGING
// flag for disk checkpoint
extern bool diskCkptFlag;
#endif

int _defaultObjectQ = 0;            // for obejct queue
bool _ringexit = 0;         // for charm exit
int _ringtoken = 8;
extern int _messageBufferingThreshold;

#if CMK_FAULT_EVAC
static bool _raiseEvac=0; // whether or not to trigger the processor shutdowns
static char *_raiseEvacFile;
void processRaiseEvacFile(char *raiseEvacFile);
#endif

extern bool useNodeBlkMapping;

extern int quietMode;
extern int quietModeRequested;

// Modules are required to register command line opts they will parse. These
// options are stored in the _optSet, and then when parsing command line opts
// users will be warned about options starting with a '+' that are not in this
// table. This usually implies that they are attempting to use a Charm++ option
// without having compiled Charm++ to use the module that options belongs to.
std::set<std::string> _optSet;
void _registerCommandLineOpt(const char* opt) {
  // The command line options are only checked during init on PE0, so this makes
  // thread safety easy.
  if (CkMyPe() == 0) {
    _optSet.insert(opt);
  }
}

static inline void _parseCommandLineOpts(char **argv)
{
  if (CmiGetArgFlagDesc(argv,"+cs", "Print extensive statistics at shutdown"))
      _STATS_ON(_printCS);
  if (CmiGetArgFlagDesc(argv,"+ss", "Print summary statistics at shutdown"))
      _STATS_ON(_printSS);
  if (CmiGetArgFlagDesc(argv,"+fifo", "Default to FIFO queuing"))
      _defaultQueueing = CK_QUEUEING_FIFO;
  if (CmiGetArgFlagDesc(argv,"+lifo", "Default to LIFO queuing"))
      _defaultQueueing = CK_QUEUEING_LIFO;
  if (CmiGetArgFlagDesc(argv,"+ififo", "Default to integer-prioritized FIFO queuing"))
      _defaultQueueing = CK_QUEUEING_IFIFO;
  if (CmiGetArgFlagDesc(argv,"+ilifo", "Default to integer-prioritized LIFO queuing"))
      _defaultQueueing = CK_QUEUEING_ILIFO;
  if (CmiGetArgFlagDesc(argv,"+bfifo", "Default to bitvector-prioritized FIFO queuing"))
      _defaultQueueing = CK_QUEUEING_BFIFO;
  if (CmiGetArgFlagDesc(argv,"+blifo", "Default to bitvector-prioritized LIFO queuing"))
      _defaultQueueing = CK_QUEUEING_BLIFO;
  if (CmiGetArgFlagDesc(argv,"+objq", "Default to use object queue for every obejct"))
  {
#if CMK_OBJECT_QUEUE_AVAILABLE
      _defaultObjectQ = 1;
      if (CkMyPe()==0)
        CmiPrintf("Charm++> Create object queue for every Charm object.\n");
#else
      CmiAbort("Charm++> Object queue not enabled, recompile Charm++ with CMK_OBJECT_QUEUE_AVAILABLE defined to 1.");
#endif
  }

#if CMK_SHRINK_EXPAND
  if (!CmiGetArgStringDesc(argv, "+shrinkexpand_basedir", (char **)&_shrinkexpand_basedir,
                           "Checkpoint directory used for shrink-expand (defaults to /dev/shm)"))
# if defined __APPLE__
      _shrinkexpand_basedir = "/tmp";
# else
      _shrinkexpand_basedir = "/dev/shm";
# endif
#endif

  if(CmiGetArgString(argv,"+restart",&_restartDir))
      faultFunc = CkRestartMain;
#if __FAULT__
  if (CmiGetArgIntDesc(argv,"+restartaftercrash",&CpvAccess(_curRestartPhase),"restarting this processor after a crash")){  
# if CMK_MEM_CHECKPOINT
      faultFunc = CkMemRestart;
# endif
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
      faultFunc = CkMlogRestart;
#endif
      CmiPrintf("[%d] Restarting after crash \n",CmiMyPe());
  }
#if CMK_MESSAGE_LOGGING
    // reading +ftc_disk flag
    if (CmiGetArgFlagDesc(argv, "+ftc_disk", "Disk Checkpointing")) {
        diskCkptFlag = true;
    }
#endif
  // reading the killFile
  if(CmiGetArgStringDesc(argv,"+killFile", &killFile,"Generates SIGKILL on specified processors")){
    if(faultFunc == NULL){
      //do not read the killfile if this is a restarting processor
      killFlag = true;
      if(CmiMyPe() == 0){
        printf("[%d] killFlag set to true for file %s\n",CkMyPe(),killFile);
      }
    }
  }
#endif

  // shut down program in ring fashion to allow projections output w/o IO error
  if (CmiGetArgIntDesc(argv,"+ringexit",&_ringtoken, "Program exits in a ring fashion")) 
  {
    _ringexit = true;
    if (CkMyPe()==0)
      CkPrintf("Charm++> Program shutdown in token ring (%d).\n", _ringtoken);
    if (_ringtoken > CkNumPes())  _ringtoken = CkNumPes();
  }
#if CMK_FAULT_EVAC
  // if the argument +raiseevac is present then cause faults
    if(CmiGetArgStringDesc(argv,"+raiseevac", &_raiseEvacFile,"Generates processor evacuation on random processors")){
        _raiseEvac = 1;
    }
#endif
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
    if(!CmiGetArgIntDesc(argv,"+teamSize",&teamSize,"Set the team size for message logging")){
        teamSize = 1;
    }
    if(!CmiGetArgIntDesc(argv,"+chkptPeriod",&chkptPeriod,"Set the checkpoint period for the message logging fault tolerance algorithm in seconds")){
        chkptPeriod = 100;
    }
    if(CmiGetArgIntDesc(argv,"+fastRecovery", &parallelRecovery, "Parallel recovery with message logging protocol")){
        fastRecovery = true;
    }
#endif

        if (!CmiGetArgIntDesc(argv, "+messageBufferingThreshold",
                              &_messageBufferingThreshold,
                              "Message size above which the runtime will buffer messages directed at unlocated array elements")) {
          _messageBufferingThreshold = INT_MAX;
        }

    /* Anytime migration flag */
    _isAnytimeMigration = true;
    if (CmiGetArgFlagDesc(argv,"+noAnytimeMigration","The program does not require support for anytime migration")) {
      _isAnytimeMigration = false;
    }
    
    _isNotifyChildInRed = true;
    if (CmiGetArgFlagDesc(argv,"+noNotifyChildInReduction","The program has at least one element per processor for each charm array created")) {
      _isNotifyChildInRed = false;
    }

    _isStaticInsertion = false;
    if (CmiGetArgFlagDesc(argv,"+staticInsertion","Array elements are only inserted at construction")) {
      _isStaticInsertion = true;
    }

        useNodeBlkMapping = false;
        if (CmiGetArgFlagDesc(argv,"+useNodeBlkMapping","Array elements are block-mapped in SMP-node level")) {
          useNodeBlkMapping = true;
        }

#if ! CMK_WITH_CONTROLPOINT
    // Display a warning if charm++ wasn't compiled with control point support but user is expecting it
    if( CmiGetArgFlag(argv,"+CPSamplePeriod") || 
        CmiGetArgFlag(argv,"+CPSamplePeriodMs") || 
        CmiGetArgFlag(argv,"+CPSchemeRandom") || 
        CmiGetArgFlag(argv,"+CPExhaustiveSearch") || 
        CmiGetArgFlag(argv,"+CPAlwaysUseDefaults") || 
        CmiGetArgFlag(argv,"+CPSimulAnneal") || 
        CmiGetArgFlag(argv,"+CPCriticalPathPrio") || 
        CmiGetArgFlag(argv,"+CPBestKnown") || 
        CmiGetArgFlag(argv,"+CPSteering") || 
        CmiGetArgFlag(argv,"+CPMemoryAware") || 
        CmiGetArgFlag(argv,"+CPSimplex") || 
        CmiGetArgFlag(argv,"+CPDivideConquer") || 
        CmiGetArgFlag(argv,"+CPLDBPeriod") || 
        CmiGetArgFlag(argv,"+CPLDBPeriodLinear") || 
        CmiGetArgFlag(argv,"+CPLDBPeriodQuadratic") || 
        CmiGetArgFlag(argv,"+CPLDBPeriodOptimal") || 
        CmiGetArgFlag(argv,"+CPDefaultValues") || 
        CmiGetArgFlag(argv,"+CPGatherAll") || 
        CmiGetArgFlag(argv,"+CPGatherMemoryUsage") || 
        CmiGetArgFlag(argv,"+CPGatherUtilization") || 
        CmiGetArgFlag(argv,"+CPSaveData") || 
        CmiGetArgFlag(argv,"+CPNoFilterData") || 
        CmiGetArgFlag(argv,"+CPLoadData") || 
        CmiGetArgFlag(argv,"+CPDataFilename")    )
      {     
        CkAbort("You specified a control point command line argument, but compiled charm++ without control point support.\n");
      }
#endif
       
}

static void _bufferHandler(void *msg)
{
  DEBUGF(("[%d] _bufferHandler called.\n", CkMyPe()));
  CkpvAccess(_buffQ)->enq(msg);
}

static void _discardHandler(envelope *env)
{
//  MESSAGE_PHASE_CHECK(env);

  DEBUGF(("[%d] _discardHandler called.\n", CkMyPe()));
#if CMK_MEM_CHECKPOINT
  //CkPrintf("[%d] _discardHandler called!\n", CkMyPe());
  if (CkInRestarting()) CpvAccess(_qd)->process();
#endif
  CmiFree(env);
}

#if CMK_WITH_STATS
static inline void _printStats(void)
{
  DEBUGF(("[%d] _printStats\n", CkMyPe()));
  int i;
  if(_printSS || _printCS) {
    Stats *total = new Stats();
    _MEMCHECK(total);
    for(i=0;i<CkNumPes();i++)
      total->combine(_allStats[i]);
    CkPrintf("Charm Kernel Summary Statistics:\n");
    for(i=0;i<CkNumPes();i++) {
      CkPrintf("Proc %d: [%d created, %d processed]\n", i,
               _allStats[i]->getCharesCreated(),
               _allStats[i]->getCharesProcessed());
    }
    CkPrintf("Total Chares: [%d created, %d processed]\n",
             total->getCharesCreated(), total->getCharesProcessed());
  }
  if(_printCS) {
    CkPrintf("Charm Kernel Detailed Statistics (R=requested P=processed):\n\n");

    CkPrintf("         Create    Mesgs     Create    Mesgs     Create    Mesgs\n");
    CkPrintf("         Chare     for       Group     for       Nodegroup for\n");
    CkPrintf("PE   R/P Mesgs     Chares    Mesgs     Groups    Mesgs     Nodegroups\n");
    CkPrintf("---- --- --------- --------- --------- --------- --------- ----------\n");

    for(i=0;i<CkNumPes();i++) {
      CkPrintf("%4d  R  %9d %9d %9d %9d %9d %9d\n      P  %9d %9d %9d %9d %9d %9d\n",i,
               _allStats[i]->getCharesCreated(),
               _allStats[i]->getForCharesCreated(),
               _allStats[i]->getGroupsCreated(),
               _allStats[i]->getGroupMsgsCreated(),
               _allStats[i]->getNodeGroupsCreated(),
               _allStats[i]->getNodeGroupMsgsCreated(),
               _allStats[i]->getCharesProcessed(),
               _allStats[i]->getForCharesProcessed(),
               _allStats[i]->getGroupsProcessed(),
               _allStats[i]->getGroupMsgsProcessed(),
               _allStats[i]->getNodeGroupsProcessed(),
           _allStats[i]->getNodeGroupMsgsProcessed());
    }
  }
}
#else
static inline void _printStats(void) {}
#endif

typedef struct _statsHeader
{
  int n;
} statsHeader;

static void * mergeStats(int *size, void *data, void **remote, int count)
{
  envelope *newData;
  statsHeader *dataMsg = (statsHeader*)EnvToUsr((envelope*) data), *newDataMsg;
  int nPes = dataMsg->n, currentIndex = 0;

  for (int i = 0; i < count; ++i)
  {
    nPes += ((statsHeader *)EnvToUsr((envelope *)remote[i]))->n;
  }

  newData = _allocEnv(StatMsg, sizeof(statsHeader) + sizeof(Stats)*nPes);
  *size = newData->getTotalsize();
  newDataMsg = (statsHeader *)EnvToUsr(newData);
  newDataMsg->n = nPes;

  statsHeader *current = dataMsg;
  Stats *currentStats = (Stats*)(current + 1), *destination = (Stats*)(newDataMsg + 1);
  memcpy(destination + currentIndex, currentStats, sizeof(Stats) * current->n);
  currentIndex += current->n;

  for (int i = 0; i < count; ++i)
  {
    current = ((statsHeader *)EnvToUsr((envelope *)remote[i]));
    currentStats = (Stats *)(current + 1);
    memcpy(destination + currentIndex, currentStats, sizeof(Stats) * current->n);
    currentIndex += current->n;
  }

  CmiFree(data);
  return newData;
}

static inline void _sendStats(void)
{
  DEBUGF(("[%d] _sendStats\n", CkMyPe()));
  envelope *env = _allocEnv(StatMsg, sizeof(statsHeader) + sizeof(Stats));
  statsHeader* msg = (statsHeader*)EnvToUsr(env);
  msg->n = 1;
  memcpy(msg+1, CkpvAccess(_myStats), sizeof(Stats));
  CmiSetHandler(env, _exitHandlerIdx);
  CmiReduce(env, env->getTotalsize(), mergeStats);
}

#if CMK_LOCKLESS_QUEUE
typedef struct _WarningMsg{
  int queue_overflow_count;
} WarningMsg;

/* General purpose handler for runtime warnings post-execution */
static void *mergeWarningMsg(int * size, void * data, void ** remote, int count){
  int i;

  WarningMsg *msg = (WarningMsg*)EnvToUsr((envelope*) data), *m;

  /* Reduction on warning information gained from children */
  for(i = 0; i < count; ++i)
  {
    m = (WarningMsg*)EnvToUsr((envelope*) remote[i]);
    msg->queue_overflow_count += m->queue_overflow_count;
  }

  return data;
}

/* Initializes the reduction, called on each processor */
extern int messageQueueOverflow;
static inline void _sendWarnings(void)
{
  DEBUGF(("[%d] _sendWarnings\n", CkMyPe()));

  envelope *env = _allocEnv(WarnMsg, sizeof(WarningMsg));
  WarningMsg* msg = (WarningMsg*)EnvToUsr(env);

  /* Set processor specific warning information here */
  msg->queue_overflow_count = messageQueueOverflow;

  CmiSetHandler(env, _exitHandlerIdx);
  CmiReduce(env, env->getTotalsize(), mergeWarningMsg);
}

/* Reporting warnings to the user */
static inline void ReportWarnings(WarningMsg * msg)
{
  if(msg->queue_overflow_count > 0)
  {
    CmiPrintf("WARNING: Message queues overflowed during execution, this can negatively impact performance.\n");
    CmiPrintf("\tModify the size of the message queues using: +MessageQueueNodes and +MessageQueueNodeSize\n");
  }
}
#endif /* CMK_LOCKLESS_QUEUE */

#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
extern void _messageLoggingExit();
#endif

#if __FAULT__
//CpvExtern(int, CldHandlerIndex);
//void CldHandler(char *);
extern int index_skipCldHandler;
extern void _skipCldHandler(void *converseMsg);

void _discard_charm_message()
{
  CkNumberHandler(_charmHandlerIdx,_discardHandler);
//  CkNumberHandler(CpvAccess(CldHandlerIndex), _discardHandler);
  CkNumberHandler(index_skipCldHandler, _discardHandler);
}

void _resume_charm_message()
{
  CkNumberHandlerEx(_charmHandlerIdx, _processHandler, CkpvAccess(_coreState));
//  CkNumberHandler(CpvAccess(CldHandlerIndex), CldHandler);
  CkNumberHandler(index_skipCldHandler, _skipCldHandler);
}
#endif

static void _exitHandler(envelope *env)
{
  DEBUGF(("exitHandler called on %d msgtype: %d\n", CkMyPe(), env->getMsgtype()));
  switch(env->getMsgtype()) {
    case StartExitMsg: // Exit sequence trigger message
      CkAssert(CkMyPe()==0);
      if(_exitStarted) {
        CmiFree(env);
        return;
      }
      _exitStarted = true;

      // else fall through
    case ExitMsg: // Exit sequence trigger message using user registered exit function
      CkAssert(CkMyPe()==0);
      if (!_CkExitFnVec.isEmpty()) {
        CmiFree(env);
        CkExitFn fn = _CkExitFnVec.deq();
        fn();
        break;
      }

      CkNumberHandler(_charmHandlerIdx,_discardHandler);
      CkNumberHandler(_bocHandlerIdx, _discardHandler);
#if !CMK_BIGSIM_THREAD
      env->setMsgtype(ReqStatMsg);
      env->setSrcPe(CkMyPe());
      // if exit in ring, instead of broadcasting, send in ring
      if (_ringexit){
    DEBUGF(("[%d] Ring Exit \n",CkMyPe()));
        const int stride = CkNumPes()/_ringtoken;
        int pe = 0;
        while (pe<CkNumPes()) {
          CmiSyncSend(pe, env->getTotalsize(), (char *)env);
          pe += stride;
        }
        CmiFree(env);
      }else{
    CmiSyncBroadcastAllAndFree(env->getTotalsize(), (char *)env);
      }
#else
      CmiFree(env);
      ConverseExit(_exitcode);
#endif
      break;
    case ReqStatMsg: // Request stats and warnings message
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
      _messageLoggingExit();
#endif
      DEBUGF(("ReqStatMsg on %d\n", CkMyPe()));
      CkNumberHandler(_charmHandlerIdx,_discardHandler);
      CkNumberHandler(_bocHandlerIdx, _discardHandler);
#if CMK_FAULT_EVAC
      if(CmiNodeAlive(CkMyPe()))
#endif
      {
#if CMK_WITH_STATS
         _sendStats();
#endif
#if CMK_WITH_WARNINGS
         _sendWarnings();
#endif
      _mainDone = true; // This is needed because the destructors for
                        // readonly variables will be called when the program
                // exits. If the destructor is called while _mainDone
                // is false, it will assume that the readonly variable was
                // declared locally. On all processors other than 0,
                // _mainDone is never set to true before the program exits.
#if CMK_TRACE_ENABLED
      if (_ringexit) traceClose();
#endif
    }
      if (_ringexit) {
        int stride = CkNumPes()/_ringtoken;
        int pe = CkMyPe()+1;
        if (pe < CkNumPes() && pe % stride != 0)
          CmiSyncSendAndFree(pe, env->getTotalsize(), (char *)env);
        else
          CmiFree(env);
      }
      else
        CmiFree(env);
#if CMK_SHRINK_EXPAND
      ConverseCleanup();
#endif
      //everyone exits here - there may be issues with leftover messages in the queue
#if !CMK_WITH_STATS && !CMK_WITH_WARNINGS
      DEBUGF(("[%d] Calling converse exit from ReqStatMsg \n",CkMyPe()));
      ConverseExit(_exitcode);
      if(CharmLibInterOperate)
    CpvAccess(interopExitFlag) = 1;
#endif
      break;
#if CMK_WITH_STATS
    case StatMsg: // Stats data message (in response to ReqStatMsg)
    {
      CkAssert(CkMyPe()==0);
      statsHeader* header = (statsHeader*)EnvToUsr(env);
      int n = header->n;
      Stats* currentStats = (Stats*)(header + 1);
      for (int i = 0; i < n; ++i)
      {
        _allStats[currentStats->getPe()] = currentStats;
    currentStats++;
      }
      DEBUGF(("StatMsg on %d with %d\n", CkMyPe(), n));
      _printStats();
      // broadcast to all others that they can now exit
      envelope* env = _allocEnv(StatDoneMsg);
      CmiSetHandler(env, _exitHandlerIdx);
      CmiSyncBroadcastAllAndFree(env->getTotalsize(), (char*)env);
    }
    break;

    case StatDoneMsg: // Indicates completion of stats reduction
      DEBUGF(("[%d] Calling converse exit from StatDoneMsg \n",CkMyPe()));
      ConverseExit(_exitcode);
      if (CharmLibInterOperate)
        CpvAccess(interopExitFlag) = 1;
      break;
#endif
#if CMK_WITH_WARNINGS
    case WarnMsg: // Warnings data message (in reponse to ReqStatMsg)
    {
      CkAssert(CkMyPe()==0);
      WarningMsg* msg = (WarningMsg*)EnvToUsr(env);
      ReportWarnings(msg);

      envelope* env = _allocEnv(WarnDoneMsg);
      CmiSetHandler(env, _exitHandlerIdx);
      CmiSyncBroadcastAllAndFree(env->getTotalsize(), (char*)env);
      break;
    }
    case WarnDoneMsg: // Indicates completion of warnings reduction
      DEBUGF(("[%d] Calling converse exit from WarnDoneMsg \n",CkMyPe()));
      ConverseExit(_exitcode);
      if (CharmLibInterOperate)
        CpvAccess(interopExitFlag) = 1;
      break;
#endif
    default:
      CmiAbort("Internal Error(_exitHandler): Unknown-msg-type. Contact Developers.\n");
  }
}

#if CMK_SHRINK_EXPAND
void _ROGroupRestartHandler(void * msg){
  CkResumeRestartMain((char *)msg);
}
#endif

static inline void _processBufferedBocInits(void)
{
  CkCoreState *ck = CkpvAccess(_coreState);
  CkNumberHandlerEx(_bocHandlerIdx,_processHandler, ck);
  PtrVec &inits=*CkpvAccess(_bocInitVec);
  int len = inits.size();
  for(int i=1; i<len; i++) {
    envelope *env = inits[i];
    if(env==0) {
#if CMK_SHRINK_EXPAND
      if(_inrestart){
        CkPrintf("_processBufferedBocInits: empty message in restart, ignoring\n");
        break;
      }
      else
        CkAbort("_processBufferedBocInits: empty message");
#else
      CkAbort("_processBufferedBocInits: empty message");
#endif
    }
    if(env->isPacked())
      CkUnpackMessage(&env);
    _processBocInitMsg(ck,env);
  }
  delete &inits;
}

static inline void _processBufferedNodeBocInits(void)
{
  CkCoreState *ck = CkpvAccess(_coreState);
  PtrVec &inits=*CksvAccess(_nodeBocInitVec);
  int len = inits.size();
  for(int i=1; i<len; i++) {
    envelope *env = inits[i];
    if(env==0) CkAbort("_processBufferedNodeBocInits: empty message");
    if(env->isPacked())
      CkUnpackMessage(&env);
    _processNodeBocInitMsg(ck,env);
  }
  delete &inits;
}

static inline void _processBufferedMsgs(void)
{
  CkNumberHandlerEx(_charmHandlerIdx, _processHandler, CkpvAccess(_coreState));
  envelope *env;
  while(NULL!=(env=(envelope*)CkpvAccess(_buffQ)->deq())) {
    if(env->getMsgtype()==NewChareMsg || env->getMsgtype()==NewVChareMsg) {
      if(env->isForAnyPE())
        _CldEnqueue(CLD_ANYWHERE, env, _infoIdx);
      else
        _processHandler((void *)env, CkpvAccess(_coreState));
    } else {
      _processHandler((void *)env, CkpvAccess(_coreState));
    }
  }
}

static int _charmLoadEstimator(void)
{
  return CkpvAccess(_buffQ)->length();
}

static void _sendTriggers(void)
{
  int i, num, first;
  CmiImmediateLock(CksvAccess(_nodeGroupTableImmLock));
  if (!CksvAccess(_triggersSent))
  {
    CksvAccess(_triggersSent) = true;
    num = CmiMyNodeSize();
    envelope *env = _allocEnv(RODataMsg); // Notice that the type here is irrelevant
    env->setSrcPe(CkMyPe());
    CmiSetHandler(env, _triggerHandlerIdx);
    first = CmiNodeFirst(CmiMyNode());
    for (i=0; i < num; i++)
      if(first+i != CkMyPe())
    CmiSyncSend(first+i, env->getTotalsize(), (char *)env);
    CmiFree(env);
  }
  CmiImmediateUnlock(CksvAccess(_nodeGroupTableImmLock));
}

void _initDone(void)
{
  if (CkpvAccess(_initdone)) return;
  CkpvAccess(_initdone) = true;
  DEBUGF(("[%d] _initDone.\n", CkMyPe()));
  if (!CksvAccess(_triggersSent)) _sendTriggers();
  CkNumberHandler(_triggerHandlerIdx, _discardHandler);
  CmiNodeBarrier();
  if(CkMyRank() == 0) {
    _processBufferedNodeBocInits();
    quietMode = 0; // re-enable CmiPrintf's if they were disabled
  }
  CmiNodeBarrier(); // wait for all nodegroups to be created
  _processBufferedBocInits();
  DEBUGF(("Reached CmiNodeBarrier(), pe = %d, rank = %d\n", CkMyPe(), CkMyRank()));
  CmiNodeBarrier();
  DEBUGF(("Crossed CmiNodeBarrier(), pe = %d, rank = %d\n", CkMyPe(), CkMyRank()));
  _processBufferedMsgs();
  CkpvAccess(_charmEpoch)=1;
  if (userDrivenMode) {
    StopInteropScheduler();
  }
}

static void _triggerHandler(envelope *env)
{
  DEBUGF(("Calling Init Done from _triggerHandler\n"));
  checkForInitDone(true); // RO ZCPY Bcast operation is already complete
  if (env!=NULL) CmiFree(env);
}

static inline void _processROMsgMsg(envelope *env)
{
  if(!CmiMyRank()) {
    *((char **)(_readonlyMsgs[env->getRoIdx()]->pMsg))=(char *)EnvToUsr(env);
  }
}

static inline void _processRODataMsg(envelope *env)
{
  //Unpack each readonly:
  if(!CmiMyRank()) {
#if CMK_ONESIDED_IMPL && CMK_SMP
    if(CMI_IS_ZC_BCAST(env)) {
      // Send message to peers
      CmiForwardMsgToPeers(env->getTotalsize(), (char *)env);
    }
#endif

    //Unpack each readonly
    PUP::fromMem pu((char *)EnvToUsr(env));
    CmiSpanningTreeInfo &t = *_topoTree;

#if CMK_ONESIDED_IMPL
    pu|numZerocopyROops;

    // Set _numPendingRORdmaTransfers to numZerocopyROops
    CksvAccess(_numPendingRORdmaTransfers) = numZerocopyROops;

    // Allocate structure for ack tracking on intermediate nodes
    if(numZerocopyROops > 0 && t.child_count != 0) {
      readonlyAllocateOnSource();
    }
#endif

    for(size_t i=0;i<_readonlyTable.size();i++) {
      _readonlyTable[i]->pupData(pu);
    }
  } else {
    CmiFree(env);
  }
}

static void _roRestartHandler(void *msg)
{
  CkAssert(CkMyPe()!=0);
  envelope *env = (envelope *) msg;
  CkpvAccess(_numInitsRecd)++;
  _numExpectInitMsgs = env->getCount();
  _processRODataMsg(env);
  // in SMP, potentially there us a race condition between rank0 calling
  // initDone, which sendTriggers, and PE 0 calls bdcastRO which broadcast
  // readonlys
  // if this readonly message arrives later, we need to call trigger again
  // to trigger initDone() on all ranks
  // we therefore needs to make sure initDone() is exactly 
  _triggerHandler(NULL);
}

#if CMK_ONESIDED_IMPL
static void _roRdmaDoneHandler(envelope *env) {

  switch(env->getMsgtype()) {
    case ROPeerCompletionMsg:
      checkForInitDone(true); // Receiving a ROPeerCompletionMsg indicates that RO ZCPY Bcast
                              // operation is complete on the comm. thread
      if (env!=NULL) CmiFree(env);
      break;
    case ROChildCompletionMsg:
      roBcastAckInfo->counter++;
      if(roBcastAckInfo->counter == roBcastAckInfo->numChildren) {
        // deregister
        for(int i=0; i < roBcastAckInfo->numops; i++) {
          NcpyROBcastBuffAckInfo *buffAckInfo = &(roBcastAckInfo->buffAckInfo[i]);
          CmiDeregisterMem(buffAckInfo->ptr,
                           buffAckInfo->layerInfo +CmiGetRdmaCommonInfoSize(),
                           buffAckInfo->pe,
                           buffAckInfo->regMode);
        }

        if(roBcastAckInfo->isRoot != 1) {
          if(_topoTree == NULL) CkAbort("CkRdmaIssueRgets:: topo tree has not been calculated \n");
          CmiSpanningTreeInfo &t = *_topoTree;

          //forward message to root
          // Send a message to the parent to signal completion in order to deregister
          envelope *compEnv = _allocEnv(ROChildCompletionMsg);
          compEnv->setSrcPe(CkMyPe());
          CmiSetHandler(compEnv, _roRdmaDoneHandlerIdx);
          CmiSyncSendAndFree(t.parent, compEnv->getTotalsize(), (char *)compEnv);
        }
        // Free the roBcastAckInfo allocated inside readonlyAllocateOnSource
        CmiFree(roBcastAckInfo);
      }
      break;
    default:
      CmiAbort("Invalid msg type\n");
      break;
  }
}
#endif

void checkForInitDone(bool rdmaROCompleted) {

  bool noPendingRORdmaTransfers = true;
#if CMK_ONESIDED_IMPL
  // Ensure that there are no pending RO Rdma transfers on Rank 0 when numZerocopyROops > 0
  if(CmiMyRank() == 0 && numZerocopyROops > 0)
    noPendingRORdmaTransfers = rdmaROCompleted;
#endif
  if (_numExpectInitMsgs && CkpvAccess(_numInitsRecd) + CksvAccess(_numInitNodeMsgs) == _numExpectInitMsgs && noPendingRORdmaTransfers)
    _initDone();
}

static void _initHandler(void *msg, CkCoreState *ck)
{
  CkAssert(CkMyPe()!=0);
  envelope *env = (envelope *) msg;
  
  if (ck->watcher!=NULL) {
    if (!ck->watcher->processMessage(&env,ck)) return;
  }
  
  switch (env->getMsgtype()) {
    case BocInitMsg: // Group creation message
      if (env->getGroupEpoch()==0) {
        CkpvAccess(_numInitsRecd)++;
        // _qd->process() or ck->process() to update QD counters is called inside _processBocInitMsg
        if (CkpvAccess(_bocInitVec)->size() < env->getGroupNum().idx + 1) {
          CkpvAccess(_bocInitVec)->resize(env->getGroupNum().idx + 1);
        }
        (*CkpvAccess(_bocInitVec))[env->getGroupNum().idx] = env;
      } else _bufferHandler(msg);
      break;
    case NodeBocInitMsg: // Nodegroup creation message
      if (env->getGroupEpoch()==0) {
        CmiImmediateLock(CksvAccess(_nodeGroupTableImmLock));
        CksvAccess(_numInitNodeMsgs)++;
        if (CksvAccess(_nodeBocInitVec)->size() < env->getGroupNum().idx + 1) {
          CksvAccess(_nodeBocInitVec)->resize(env->getGroupNum().idx + 1);
        }
        (*CksvAccess(_nodeBocInitVec))[env->getGroupNum().idx] = env;
        CmiImmediateUnlock(CksvAccess(_nodeGroupTableImmLock));
        // _qd->process() or ck->process() to update QD counters is called inside _processNodeBocInitMsg
      } else _bufferHandler(msg);
      break;
    case ROMsgMsg: // Readonly Message (for user declared readonly messages)
      CkpvAccess(_numInitsRecd)++;
      CpvAccess(_qd)->process();
      if(env->isPacked()) CkUnpackMessage(&env);
      _processROMsgMsg(env);
      break;
    case RODataMsg: // Readonly Data Message (for user declared readonly variables)
      CkpvAccess(_numInitsRecd)++;
      CpvAccess(_qd)->process();
      _numExpectInitMsgs = env->getCount();
      _processRODataMsg(env);
      break;
    default:
      CmiAbort("Internal Error: Unknown-msg-type. Contact Developers.\n");
  }
  DEBUGF(("[%d,%.6lf] _numExpectInitMsgs %d CkpvAccess(_numInitsRecd)+CksvAccess(_numInitNodeMsgs) %d+%d\n",CmiMyPe(),CmiWallTimer(),_numExpectInitMsgs,CkpvAccess(_numInitsRecd),CksvAccess(_numInitNodeMsgs)));
  checkForInitDone(false); // RO ZCPY Bcast operation could still be incomplete
}

#if CMK_SHRINK_EXPAND
void CkCleanup()
{
    // always send to PE 0
    envelope *env = _allocEnv(StartExitMsg);
    env->setSrcPe(CkMyPe());
    CmiSetHandler(env, _exitHandlerIdx);
    CmiSyncSendAndFree(0, env->getTotalsize(), (char *)env);
}
#endif

CkQ<CkExitFn> _CkExitFnVec;

// Trigger exit on PE 0,
// which traverses _CkExitFnVec to call every registered user exit function.
// Every user exit function should end with CkExit() to continue the chain.
void CkExit(int exitcode)
{
  DEBUGF(("[%d] CkExit called \n",CkMyPe()));
    // always send to PE 0

  // Store exit code for use in ConverseExit
  _exitcode = exitcode;
  envelope *env = _allocEnv(StartExitMsg);
  env->setSrcPe(CkMyPe());
  CmiSetHandler(env, _exitHandlerIdx);
  CmiSyncSendAndFree(0, env->getTotalsize(), (char *)env);

#if ! CMK_BIGSIM_THREAD
  _TRACE_END_EXECUTE();
  //Wait for stats, which will call ConverseExit when finished:
    if(!CharmLibInterOperate)
  CsdScheduler(-1);
#endif
}

void CkContinueExit()
{
  envelope *env = _allocEnv(ExitMsg);
  env->setSrcPe(CkMyPe());
  CmiSetHandler(env, _exitHandlerIdx);
  CmiSyncSendAndFree(0, env->getTotalsize(), (char *)env);
}

/* This is a routine called in case the application is closing due to a signal.
   Tear down structures that must be cleaned up even when unclean exit happens.
   It is called by the machine layer whenever some problem occurs (it is thus up
   to the machine layer to call this function). */
void EmergencyExit(void) {
#ifndef __BIGSIM__
  /* Delete _coreState to force any CkMessageWatcher to close down. */
  if (CkpvAccess(_coreState) != NULL) {
    delete CkpvAccess(_coreState);
    CkpvAccess(_coreState) = NULL;
  }
#endif
}

static void _nullFn(void *, void *)
{
  CmiAbort("Null-Method Called. Program may have Unregistered Module!!\n");
}

extern void _registerLBDatabase(void);
extern void _registerMetaBalancer(void);
extern void _registerPathHistory(void);
#if CMK_WITH_CONTROLPOINT
extern void _registerControlPoints(void);
#endif
extern void _registerTraceControlPoints();
extern void _registerExternalModules(char **argv);
extern void _ckModuleInit(void);
extern void _loadbalancerInit();
extern void _metabalancerInit();
#if CMK_SMP && CMK_TASKQUEUE
extern void _taskqInit();
#endif
#if CMK_SMP
extern void LBTopoInit();
#endif
extern void _initChareTables();
#if CMK_MEM_CHECKPOINT
extern void init_memcheckpt(char **argv);
#endif
extern "C" void initCharmProjections();
void CmiInitCPUTopology(char **argv);
void CmiCheckAffinity();
void CmiInitMemAffinity(char **argv);
void CmiInitPxshm(char **argv);

//void CldCallback();

void _registerInitCall(CkInitCallFn fn, int isNodeCall)
{
  if (isNodeCall) _initCallTable.initNodeCalls.enq(fn);
  else _initCallTable.initProcCalls.enq(fn);
}

void InitCallTable::enumerateInitCalls()
{
  int i;
#ifdef __BIGSIM__
  if(BgNodeRank()==0)        // called only once on an emulating node
#else
  if(CkMyRank()==0) 
#endif
  {
    for (i=0; i<initNodeCalls.length(); i++) initNodeCalls[i]();
  }
  // initproc may depend on initnode calls.
  CmiNodeAllBarrier();
  for (i=0; i<initProcCalls.length(); i++) initProcCalls[i]();
}

CpvCExtern(int, cpdSuspendStartup);
void CpdFreeze(void);

extern int _dummy_dq;

void initQd(char **argv)
{
    CpvInitialize(QdState*, _qd);
    CpvAccess(_qd) = new QdState();
    if (CmiMyRank() == 0) {
#if !defined(CMK_CPV_IS_SMP) && !CMK_SHARED_VARS_UNIPROCESSOR
    CpvAccessOther(_qd, 1) = new QdState(); // for i/o interrupt
#endif
    }
    CmiAssignOnce(&_qdHandlerIdx, CmiRegisterHandler((CmiHandler)_qdHandler));
    CmiAssignOnce(&_qdCommHandlerIdx, CmiRegisterHandler((CmiHandler)_qdCommHandler));
        if (CmiGetArgIntDesc(argv,"+qd",&_dummy_dq, "QD time in seconds")) {
          if (CmiMyPe()==0)
            CmiPrintf("Charm++> Fake QD using %d seconds.\n", _dummy_dq);
        }
}

#if CMK_BIGSIM_CHARM && CMK_CHARMDEBUG
void CpdBgInit();
#endif
void CpdBreakPointInit();

extern void (*CkRegisterMainModuleCallback)();

void _sendReadonlies() {
  for(int i=0;i<_readonlyMsgs.size();i++) /* Send out readonly messages */
  {
    void *roMsg = (void *) *((char **)(_readonlyMsgs[i]->pMsg));
    if(roMsg==0)
      continue;
    //Pack the message and send it to all other processors
    envelope *env = UsrToEnv(roMsg);
    env->setSrcPe(CkMyPe());
    env->setMsgtype(ROMsgMsg);
    env->setRoIdx(i);
    CmiSetHandler(env, _initHandlerIdx);
    CkPackMessage(&env);
    CmiSyncBroadcast(env->getTotalsize(), (char *)env);
    CpvAccess(_qd)->create(CkNumPes()-1);

    //For processor 0, unpack and re-set the global
    CkUnpackMessage(&env);
    _processROMsgMsg(env);
    _numInitMsgs++;
  }

#if CMK_ONESIDED_IMPL
  numZerocopyROops = 0;
  curROIndex = 0;
#endif

  //Determine the size of the RODataMessage
  PUP::sizer ps;

#if CMK_ONESIDED_IMPL
  ps|numZerocopyROops;
#endif

  for(int i=0;i<_readonlyTable.size();i++) _readonlyTable[i]->pupData(ps);

#if CMK_ONESIDED_IMPL
  if(numZerocopyROops > 0) {
    readonlyAllocateOnSource();
  }
#endif

  //Allocate and fill out the RODataMessage
  envelope *env = _allocEnv(RODataMsg, ps.size());
  PUP::toMem pp((char *)EnvToUsr(env));
#if CMK_ONESIDED_IMPL
  pp|numZerocopyROops;
#endif
  for(int i=0;i<_readonlyTable.size();i++) _readonlyTable[i]->pupData(pp);

  env->setCount(++_numInitMsgs);
  env->setSrcPe(CkMyPe());
  CmiSetHandler(env, _initHandlerIdx);
  DEBUGF(("[%d,%.6lf] RODataMsg being sent of size %d \n",CmiMyPe(),CmiWallTimer(),env->getTotalsize()));
  CmiSyncBroadcast(env->getTotalsize(), (char *)env);
#if CMK_ONESIDED_IMPL && CMK_SMP
  if(numZerocopyROops > 0) {
    // Send message to peers
    CmiForwardMsgToPeers(env->getTotalsize(), (char *)env);
  }
#endif
  CmiFree(env);
  CpvAccess(_qd)->create(CkNumPes()-1);
  _initDone();
}

void _initCharm(int unused_argc, char **argv)
{ 
    int inCommThread = (CmiMyRank() == CmiMyNodeSize());

    DEBUGF(("[%d,%.6lf ] _initCharm started\n",CmiMyPe(),CmiWallTimer()));
    std::set_terminate([](){ CkAbort("Unhandled C++ exception in user code.\n");});

    CkpvInitialize(size_t *, _offsets);
    CkpvAccess(_offsets) = new size_t[32];
    CkpvInitialize(PtrQ*,_buffQ);
    CkpvInitialize(PtrVec*,_bocInitVec);
    CkpvInitialize(void*, _currentChare);
    CkpvInitialize(int,   _currentChareType);
    CkpvInitialize(CkGroupID, _currentGroup);
    CkpvInitialize(void *, _currentNodeGroupObj);
    CkpvInitialize(CkGroupID, _currentGroupRednMgr);
    CkpvInitialize(GroupTable*, _groupTable);
    CkpvInitialize(GroupIDTable*, _groupIDTable);
    CkpvInitialize(CmiImmediateLockType, _groupTableImmLock);
        CkpvInitialize(bool, _destroyingNodeGroup);
        CkpvAccess(_destroyingNodeGroup) = false;
    CkpvInitialize(UInt, _numGroups);
    CkpvInitialize(int, _numInitsRecd);
    CkpvInitialize(bool, _initdone);
    CkpvInitialize(char**, Ck_argv); CkpvAccess(Ck_argv)=argv;
    CkpvInitialize(MsgPool*, _msgPool);
    CkpvInitialize(CkCoreState *, _coreState);

#if CMK_FAULT_EVAC
#ifndef __BIGSIM__
    CpvInitialize(char *,_validProcessors);
#endif
    CkpvInitialize(char ,startedEvac);
#endif
    CpvInitialize(int,serializer);

    _initChareTables();            // for checkpointable plain chares

    CksvInitialize(UInt, _numNodeGroups);
    CksvInitialize(GroupTable*, _nodeGroupTable);
    CksvInitialize(GroupIDTable, _nodeGroupIDTable);
    CksvInitialize(CmiImmediateLockType, _nodeGroupTableImmLock);
    CksvInitialize(CmiNodeLock, _nodeLock);
    CksvInitialize(PtrVec*,_nodeBocInitVec);
    CksvInitialize(UInt,_numInitNodeMsgs);
    CkpvInitialize(int,_charmEpoch);
    CkpvAccess(_charmEpoch)=0;
    CksvInitialize(bool, _triggersSent);
    CksvAccess(_triggersSent) = false;

#if CMK_ONESIDED_IMPL
    CksvInitialize(int, _numPendingRORdmaTransfers);
#endif

    CkpvInitialize(_CkOutStream*, _ckout);
    CkpvInitialize(_CkErrStream*, _ckerr);
    CkpvInitialize(Stats*, _myStats);

    CkpvAccess(_groupIDTable) = new GroupIDTable(0);
    CkpvAccess(_groupTable) = new GroupTable;
    CkpvAccess(_groupTable)->init();
    CkpvAccess(_groupTableImmLock) = CmiCreateImmediateLock();
    CkpvAccess(_numGroups) = 1; // make 0 an invalid group number
    CkpvAccess(_buffQ) = new PtrQ();
    CkpvAccess(_bocInitVec) = new PtrVec();

    CkpvAccess(_currentNodeGroupObj) = NULL;

    if(CkMyRank()==0)
    {
        CksvAccess(_numNodeGroups) = 1; //make 0 an invalid group number
            CksvAccess(_numInitNodeMsgs) = 0;

#if CMK_ONESIDED_IMPL
        CksvAccess(_numPendingRORdmaTransfers) = 0;
#endif

        CksvAccess(_nodeLock) = CmiCreateLock();
        CksvAccess(_nodeGroupTable) = new GroupTable();
        CksvAccess(_nodeGroupTable)->init();
        CksvAccess(_nodeGroupTableImmLock) = CmiCreateImmediateLock();
        CksvAccess(_nodeBocInitVec) = new PtrVec();
    }

    CkCallbackInit();
    
    CmiNodeAllBarrier();

#if ! CMK_BIGSIM_CHARM
    initQd(argv);         // bigsim calls it in ConverseCommonInit
#endif

    CkpvAccess(_coreState)=new CkCoreState();

    CkpvAccess(_numInitsRecd) = 0;
    CkpvAccess(_initdone) = false;

    CkpvAccess(_ckout) = new _CkOutStream();
    CkpvAccess(_ckerr) = new _CkErrStream();

    CmiAssignOnce(&_charmHandlerIdx, CkRegisterHandler(_bufferHandler));
    CmiAssignOnce(&_initHandlerIdx, CkRegisterHandlerEx(_initHandler, CkpvAccess(_coreState)));
    CmiAssignOnce(&_roRestartHandlerIdx, CkRegisterHandler(_roRestartHandler));

#if CMK_ONESIDED_IMPL
    CmiAssignOnce(&_roRdmaDoneHandlerIdx, CkRegisterHandler(_roRdmaDoneHandler));
#endif

    CmiAssignOnce(&_exitHandlerIdx, CkRegisterHandler(_exitHandler));
    //added for interoperabilitY
    CmiAssignOnce(&_libExitHandlerIdx, CkRegisterHandler(_libExitHandler));
    CmiAssignOnce(&_bocHandlerIdx, CkRegisterHandlerEx(_initHandler, CkpvAccess(_coreState)));
#if CMK_SHRINK_EXPAND
    // for shrink expand cleanup
    CmiAssignOnce(&_ROGroupRestartHandlerIdx, CkRegisterHandler(_ROGroupRestartHandler));
#endif

#ifdef __BIGSIM__
    if(BgNodeRank()==0) 
#endif
    _infoIdx = CldRegisterInfoFn((CldInfoFn)_infoFn);

    CmiAssignOnce(&_triggerHandlerIdx, CkRegisterHandler(_triggerHandler));
    _ckModuleInit();

    CldRegisterEstimator((CldEstimator)_charmLoadEstimator);

    _futuresModuleInit(); // part of futures implementation is a converse module
    _loadbalancerInit();
        _metabalancerInit();

#if CMK_SMP
    if (CmiMyRank() == 0) {
        LBTopoInit();
    }
#endif
#if CMK_MEM_CHECKPOINT
        init_memcheckpt(argv);
#endif

    initCharmProjections();
#if CMK_TRACE_IN_CHARM
        // initialize trace module in ck
        traceCharmInit(argv);
#endif
    
    CkpvInitialize(int, envelopeEventID);
    CkpvAccess(envelopeEventID) = 0;
    CkMessageWatcherInit(argv,CkpvAccess(_coreState));
    
    // Set the ack handler function used for the direct nocopy api
    CmiSetDirectNcpyAckHandler(CkRdmaDirectAckHandler);

#if CMK_ONESIDED_IMPL
    // Set the ack handler function used for the entry method p2p api and entry method bcast api
    CmiSetEMNcpyAckHandler(CkRdmaEMAckHandler, CkRdmaEMBcastAckHandler, CkRdmaEMBcastPostAckHandler);
#endif

#ifdef __BIGSIM__
    if(BgNodeRank()==0) 
#else
    if(CkMyRank()==0)
#endif
    {
        SDAG::registerPUPables();
        CmiArgGroup("Charm++",NULL);
        _parseCommandLineOpts(argv);
        _registerInit();
        CkRegisterMsg("System", 0, 0, CkFreeMsg, sizeof(int));
        CkRegisterChareInCharm(CkRegisterChare("null", 0, TypeChare));
        CkIndex_Chare::__idx=CkRegisterChare("Chare", sizeof(Chare), TypeChare);
        CkRegisterChareInCharm(CkIndex_Chare::__idx);
        CkIndex_Group::__idx=CkRegisterChare("Group", sizeof(Group), TypeGroup);
                CkRegisterChareInCharm(CkIndex_Group::__idx);
        CkRegisterEp("null", (CkCallFnPtr)_nullFn, 0, 0, 0+CK_EP_INTRINSIC);
        
        _registerCkFutures();
        _registerCkArray();
        _registerLBDatabase();
    _registerMetaBalancer();
        _registerCkCallback();
        _registerwaitqd();
        _registerCkCheckpoint();
                _registerCkMulticast();
#if CMK_MEM_CHECKPOINT
        _registerCkMemCheckpoint();
#endif
#if CMK_CHARMPY

        _registerGreedyRefineLB();
#endif

#if !CMK_CHARMPY
        CkRegisterMainModule();
#else
                // CkRegisterMainModule doesn't exist in charm4py because there is no executable.
                // Instead, we go to Python to register user chares from there
        if (CkRegisterMainModuleCallback)
            CkRegisterMainModuleCallback();
        else
            CkAbort("No callback for CkRegisterMainModule");
#endif

#if !CMK_CHARMPY
        _registerExternalModules(argv);
#endif
    }

    /* The following will happen on every virtual processor in BigEmulator, not just on once per real processor */
    if (CkMyRank() == 0) {
      CpdBreakPointInit();
    }
    CmiNodeAllBarrier();

    // Execute the initcalls registered in modules
    _initCallTable.enumerateInitCalls();

#if CMK_CHARMDEBUG
    CpdFinishInitialization();
#endif
    if (CkMyRank() == 0)
      _registerDone();
    CmiNodeAllBarrier();

    CkpvAccess(_myStats) = new Stats();
    CkpvAccess(_msgPool) = new MsgPool();

    CmiNodeAllBarrier();

#if !(__FAULT__)
    CmiBarrier();
    CmiBarrier();
    CmiBarrier();
#endif
#if CMK_SMP_TRACE_COMMTHREAD
    _TRACE_BEGIN_COMPUTATION(); 
#else
    if (!inCommThread) {
      _TRACE_BEGIN_COMPUTATION();
    }
#endif

#ifdef ADAPT_SCHED_MEM
    if(CkMyRank()==0){
    memCriticalEntries = new int[numMemCriticalEntries];
    int memcnt=0;
    for(int i=0; i<_entryTable.size(); i++){
        if(_entryTable[i]->isMemCritical){
        memCriticalEntries[memcnt++] = i;
        }
    }
    }
#endif

#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
    _messageLoggingInit();
#endif

#if CMK_FAULT_EVAC
#ifndef __BIGSIM__
    CpvAccess(_validProcessors) = new char[CkNumPes()];
    for(int vProc=0;vProc<CkNumPes();vProc++){
        CpvAccess(_validProcessors)[vProc]=1;
    }
    CmiAssignOnce(&_ckEvacBcastIdx, CkRegisterHandler(_ckEvacBcast));
    CmiAssignOnce(&_ckAckEvacIdx, CkRegisterHandler(_ckAckEvac));
#endif

    CkpvAccess(startedEvac) = 0;
    evacuate = 0;
    CcdCallOnCondition(CcdSIGUSR1,(CcdVoidFn)CkDecideEvacPe,0);
#endif
    CpvAccess(serializer) = 0;

#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_)) 
    CcdCallOnCondition(CcdSIGUSR2,(CcdVoidFn)CkMlogRestart,0);
#endif

#if CMK_FAULT_EVAC
    if(_raiseEvac){
        processRaiseEvacFile(_raiseEvacFile);
        /*
        if(CkMyPe() == 2){
        //  CcdCallOnConditionKeep(CcdPERIODIC_10s,(CcdVoidFn)CkDecideEvacPe,0);
            CcdCallFnAfter((CcdVoidFn)CkDecideEvacPe, 0, 10000);
        }
        if(CkMyPe() == 3){
            CcdCallFnAfter((CcdVoidFn)CkDecideEvacPe, 0, 10000);
        }*/
    }   
#endif

    if (CkMyRank() == 0) {
      TopoManager_init();
    }
    CmiNodeAllBarrier();

    if (!_replaySystem) {
        CkFtFn  faultFunc_restart = CkRestartMain;
        if (faultFunc == NULL || faultFunc == faultFunc_restart) {         // this is not restart from memory
            // these two are blocking calls for non-bigsim
#if ! CMK_BIGSIM_CHARM
      CmiInitCPUAffinity(argv);
          CmiInitMemAffinity(argv);
#endif
        }
        CmiInitCPUTopology(argv);
        if (CkMyRank() == 0) {
          TopoManager_reset(); // initialize TopoManager singleton
#if !CMK_BIGSIM_CHARM
          _topoTree = ST_RecursivePartition_getTreeInfo(0);
#endif
        }
        CmiNodeAllBarrier(); // threads wait until _topoTree has been generated
#if CMK_SHARED_VARS_POSIX_THREADS_SMP
        if (CmiCpuTopologyEnabled()) {
            int *pelist;
            int num;
            CmiGetPesOnPhysicalNode(0, &pelist, &num);
#if !CMK_MULTICORE && !CMK_SMP_NO_COMMTHD
            // Count communication threads, if present
            // XXX: Assuming uniformity of node size here
            num += num/CmiMyNodeSize();
#endif
            if (!_Cmi_forceSpinOnIdle && num > CmiNumCores())
            {
              if (CmiMyPe() == 0)
                CmiPrintf("\nCharm++> Warning: the number of SMP threads (%d) is greater than the number of physical cores (%d), so threads will sleep while idling. Use +CmiSpinOnIdle or +CmiSleepOnIdle to control this directly.\n\n", num, CmiNumCores());
              CmiLock(CksvAccess(_nodeLock));
              if (! _Cmi_sleepOnIdle) _Cmi_sleepOnIdle = 1;
              CmiUnlock(CksvAccess(_nodeLock));
            }
        }
#endif
    }

#if CMK_CUDA
    if (CmiMyRank() == 0) {
      initHybridAPI();
    }
    else /* CmiMyRank() != 0 */ {
      setHybridAPIDevice();
    }
    initEventQueues();

    // ensure HAPI is initialized before registering callback functions
    if (CmiMyRank() < CmiMyNodeSize()) {
      CmiNodeBarrier();
    }
    hapiRegisterCallbacks();
#endif

    if(CmiMyPe() == 0) {
        char *topoFilename;
        if(CmiGetArgStringDesc(argv,"+printTopo",&topoFilename,"topo file name")) 
        {
            std::stringstream sstm;
            sstm << topoFilename << "." << CmiMyPartition();
            std::string result = sstm.str();
            FILE *fp;
            fp = fopen(result.c_str(), "w");
            if (fp == NULL) {
              CkPrintf("Error opening %s file, writing to stdout\n", topoFilename);
              fp = stdout;
            }
        TopoManager_printAllocation(fp);
            fclose(fp);
        }
    }

#if CMK_USE_PXSHM && ( CMK_CRAYXE || CMK_CRAYXC ) && CMK_SMP
      // for SMP on Cray XE6 (hopper) it seems pxshm has to be initialized
      // again after cpuaffinity is done
    if (CkMyRank() == 0) {
      CmiInitPxshm(argv);
    }
    CmiNodeAllBarrier();
#endif

    //CldCallback();
#if CMK_BIGSIM_CHARM && CMK_CHARMDEBUG
      // Register the BG handler for CCS. Notice that this is put into a variable shared by
      // the whole real processor. This because converse needs to find it. We check that all
      // virtual processors register the same index for this handler.
    CpdBgInit();
#endif

    if (faultFunc) {
#if CMK_WITH_STATS
        if (CkMyPe()==0) _allStats = new Stats*[CkNumPes()];
#endif
        if (!inCommThread) {
                  CkArgMsg *msg = (CkArgMsg *)CkAllocMsg(0, sizeof(CkArgMsg), 0, GroupDepNum{});
                  msg->argc = CmiGetArgc(argv);
                  msg->argv = argv;
                  faultFunc(_restartDir, msg);
                  CkFreeMsg(msg);
                }
    }else if(CkMyPe()==0){
#if CMK_WITH_STATS
        _allStats = new Stats*[CkNumPes()];
#endif
        size_t i, nMains=_mainTable.size();

        // Check CkArgMsg and warn if it contains any args starting with '+'.
        // These args may be args intended for Charm++ but because of the specific
        // build, were not parsed by the RTS.
        int count = 0;
        int argc = CmiGetArgc(argv);
        for (int i = 1; i < argc; i++) {
            // The +vp option for TCharm is a special case that needs to be checked
            // separately, because the number passed does not need a space after
            // the vp, and the option can be specified with a '+' or a '-'.
            if (strncmp(argv[i],"+vp",3) == 0) {
                if (_optSet.count("+vp") == 0) {
                    count++;
                    CmiPrintf("WARNING: %s is a TCharm command line argument, but you have not compiled with TCharm\n", argv[i]);
                }
            } else if (strncmp(argv[i],"-vp",3) == 0) {
                CmiPrintf("WARNING: %s is no longer valid because -vp has been deprecated. Please use +vp.\n", argv[i]);
            } else if (argv[i][0] == '+' && _optSet.count(argv[i]) == 0) {
                count++;
                CmiPrintf("WARNING: %s is a command line argument beginning with a '+' but was not parsed by the RTS.\n", argv[i]);
            } else if (argv[i][0] == '+' && _optSet.count(argv[i]) != 0) {
                fprintf(stderr,"%s is used more than once. Please remove duplicate arguments.\n", argv[i]);
                CmiAbort("Bad command-line argument\n");
            }
        }
        if (count) {
            CmiPrintf("If any of the above arguments were intended for the RTS you may need to recompile Charm++ with different options.\n");
        }

        CmiCheckAffinity(); // check for thread oversubscription

        for(i=0;i<nMains;i++)  /* Create all mainchares */
        {
            size_t size = _chareTable[_mainTable[i]->chareIdx]->size;
            void *obj = malloc(size);
            _MEMCHECK(obj);
            _mainTable[i]->setObj(obj);
            CkpvAccess(_currentChare) = obj;
            CkpvAccess(_currentChareType) = _mainTable[i]->chareIdx;
            CkArgMsg *msg = (CkArgMsg *)CkAllocMsg(0, sizeof(CkArgMsg), 0, GroupDepNum{});
            msg->argc = CmiGetArgc(argv);
            msg->argv = argv;
      quietMode = 0;  // allow printing any mainchare user messages
            _entryTable[_mainTable[i]->entryIdx]->call(msg, obj);
      if (quietModeRequested) quietMode = 1;
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
            CpvAccess(_currentObj) = (Chare *)obj;
#endif
        }
                _mainDone = true;

        _STATS_RECORD_CREATE_CHARE_N(nMains);
        _STATS_RECORD_PROCESS_CHARE_N(nMains);

        if (!userDrivenMode) {
            _sendReadonlies();
        }
    } else {
        // check for thread oversubscription
        CmiCheckAffinity();
        // NOTE: this assumes commthreads will not block from this point on
    }

    DEBUGF(("[%d,%d%.6lf] inCommThread %d\n",CmiMyPe(),CmiMyRank(),CmiWallTimer(),inCommThread));
    // when I am a communication thread, I don't participate initDone.
        if (inCommThread) {
                CkNumberHandlerEx(_bocHandlerIdx, _processHandler, CkpvAccess(_coreState));
                CkNumberHandlerEx(_charmHandlerIdx, _processHandler, CkpvAccess(_coreState));
                _processBufferedMsgs();
        }

#if CMK_CHARMDEBUG
        // Should not use CpdFreeze inside a thread (since this processor is really a user-level thread)
       if (CpvAccess(cpdSuspendStartup))
       { 
          //CmiPrintf("In Parallel Debugging mode .....\n");
          CpdFreeze();
       }
#endif


#if __FAULT__
    if(killFlag){                                                  
                readKillFile();                                        
        }
#endif

}

int charm_main(int argc, char **argv)
{
  int stack_top=0;
  memory_stack_top = &stack_top;

  ConverseInit(argc, argv, (CmiStartFn) _initCharm, 0, 0);

  return 0;
}

void FTN_NAME(CHARM_MAIN_FORTRAN_WRAPPER, charm_main_fortran_wrapper)(int *argc, char **argv)
{
  charm_main(*argc, argv);
}

// user callable function to register an exit function, this function
// will perform task of collecting of info from all pes to pe0, and call
// CkContinueExit() on pe0 again to recursively traverse the registered exitFn.
// see trace-summary for an example.
void registerExitFn(CkExitFn fn)
{
#if CMK_SHRINK_EXPAND
  CkAbort("registerExitFn is called when shrink-expand is enabled!");
#else
  _CkExitFnVec.enq(fn);
#endif
}

---