Home Research Papers Posters Manuals Talks Download People Help Wiki Internal

libs/ck-libs/pose/pose.C

Go to the documentation of this file.

#include "pose.h"
#include "pose.def.h"

CpvDeclare(int, stateRecovery);
CpvDeclare(eventID, theEventID);

void POSEreadCmdLine();
#ifdef POSE_COMM_ON
extern int comm_debug;
#endif
double busyWait;
double sim_timer;
int POSE_inactDetect;
POSE_TimeType POSE_endtime;
POSE_TimeType POSE_GlobalClock;
POSE_TimeType POSE_GlobalTS;
POSE_Config pose_config;
ComlibInstanceHandle POSE_commlib_insthndl;


const eventID& GetEventID() {
  //CpvStaticDeclare(eventID, theEventID);  // initializes to [0.pe]
  //  for each pe called on
  CpvAccess(theEventID).incEventID();
  CkAssert(CpvAccess(theEventID).getPE()>=0);
  return(CpvAccess(theEventID));
 }

// Main initialization for all of POSE
void POSE_init() // use inactivity detection by default
{
  POSE_init(1, POSE_UnsetTS);
}

void POSE_init(int ET) // a single parameter specifies endtime
{
  POSE_init(0, ET);
}

void POSE_init(int IDflag, int ET) // can specify both
{
  CkPrintf("Initializing POSE...  \n");
  POSEreadCmdLine();
  POSE_inactDetect = IDflag;
  POSE_endtime = ET;
#ifndef CMK_OPTIMIZE
  traceRegisterUserEvent("Forward Execution", 10);
  traceRegisterUserEvent("Cancellation", 20);
  traceRegisterUserEvent("Cancel Spawn", 30);
  traceRegisterUserEvent("Rollback", 40);
  traceRegisterUserEvent("Commit", 50);
  traceRegisterUserEvent("OptSync", 60);
#endif
#ifndef SEQUENTIAL_POSE
#ifdef POSE_COMM_ON
  // Create the communication library for POSE
  POSE_commlib_insthndl = CkGetComlibInstance();
  // Create the communication strategy for POSE
  StreamingStrategy *strategy = new StreamingStrategy(COMM_TIMEOUT,COMM_MAXMSG);
  //MeshStreamingStrategy *strategy = new MeshStreamingStrategy(COMM_TIMEOUT,COMM_MAXMSG);
  //PrioStreaming *strategy = new PrioStreaming(COMM_TIMEOUT,COMM_MAXMSG);
  //Register the strategy
  POSE_commlib_insthndl.setStrategy(strategy);
  //comm_debug=1;
  //CkPrintf("Simulation run with PrioStreaming(%d,%d) for communication optimization...\n", COMM_TIMEOUT, COMM_MAXMSG);
  CkPrintf("Simulation run with StreamingStrategy(%d,%d) for communication optimization...\n", COMM_TIMEOUT, COMM_MAXMSG);
  //CkPrintf("Simulation run with MeshStreaming(%d,%d) for communication optimization...\n", COMM_TIMEOUT, COMM_MAXMSG);
#endif
  // Create a MemoryPool with global handle for memory recycling 
  MemPoolID = CProxy_MemoryPool::ckNew();
  // Create a Temporal Memory Manager
  TempMemID = CProxy_TimePool::ckNew();
#endif
  // Initialize statistics collection if desired
#ifndef CMK_OPTIMIZE
  theLocalStats = CProxy_localStat::ckNew();
  CProxy_globalStat::ckNew(&theGlobalStats);
#endif
#ifndef SEQUENTIAL_POSE
  // Initialize global handles to GVT and PVT
  ThePVT = CProxy_PVT::ckNew(); 
  TheGVT = CProxy_GVT::ckNew();
  // Start off using normal forward execution
  if(pose_config.lb_on)
    {
      // Initialize the load balancer
      TheLBG = CProxy_LBgroup::ckNew();
      TheLBstrategy = CProxy_LBstrategy::ckNew();
      CkPrintf("Load balancing is ON.\n");
    }
#endif
  CProxy_pose::ckNew(&POSE_Coordinator_ID, 0);
  // Create array to hold all POSE objects
#ifdef POSE_COMM_ON  
  POSE_Objects_RO = CProxy_sim::ckNew(); 
  POSE_Objects = POSE_Objects_RO;
#else
  POSE_Objects = CProxy_sim::ckNew(); 
#endif
  //#ifndef SEQUENTIAL_POSE
  //#ifdef POSE_COMM_ON
  // Make POSE_Objects use the comm lib
  //  ComlibDelegateProxy(&POSE_Objects);
  //#endif
  //#endif

#ifdef SEQUENTIAL_POSE
  if (CkNumPes() > 1) CkAbort("ERROR: Cannot run a sequential simulation on more than one processor!\n");
  CkPrintf("NOTE: POSE running in sequential simulation mode!\n");
  int fnIdx = CkIndex_pose::stop();
  CkStartQD(fnIdx, &POSE_Coordinator_ID);
  POSE_GlobalClock = 0;
  POSE_GlobalTS = 0;
#else
  /*  CkPrintf("WARNING: Charm Quiescence termination enabled!\n");
  int fnIdx = CkIndex_pose::stop();
  CkStartQD(fnIdx, &POSE_Coordinator_ID);
  */
#endif  
  CkPrintf("POSE initialization complete.\n");
  if (POSE_inactDetect) CkPrintf("Using Inactivity Detection for termination.\n");
  else 
#if USE_LONG_TIMESTAMPS
    CkPrintf("Using endTime of %lld for termination.\n", POSE_endtime);
#else
    CkPrintf("Using endTime of %d for termination.\n", POSE_endtime);
#endif
  sim_timer = CmiWallTimer(); 
}

void POSE_startTimer() {
  CkPrintf("Starting simulation...\n"); 
  sim_timer = CmiWallTimer(); 
}

void POSE_useID() 
{
  CkPrintf("WARNING: POSE_useID obsolete. See POSE_init params.\n");
}

void POSE_useET(POSE_TimeType et) 
{
  CkPrintf("WARNING: POSE_useET obsolete. See POSE_init params.\n");
}

void POSE_registerCallBack(CkCallback cb)
{
  CProxy_pose POSE_Coordinator(POSE_Coordinator_ID);
  callBack *cbm = new callBack;
  cbm->callback = cb;
  POSE_Coordinator.registerCallBack(cbm);
}

void POSE_stop()
{
  CProxy_pose POSE_Coordinator(POSE_Coordinator_ID);
  POSE_Coordinator.stop();
}

void POSE_exit()
{
  CProxy_pose POSE_Coordinator(POSE_Coordinator_ID);
  POSE_Coordinator.exit();
}

void POSE_prepExit(void *param, void *msg)
{
  CkReductionMsg *m=(CkReductionMsg *)msg;
  delete m;
  CProxy_pose POSE_Coordinator(POSE_Coordinator_ID);
  POSE_Coordinator.prepExit();
}

void POSE_set_busy_wait(double n) { busyWait = n; }

void POSE_busy_wait()
{
  double start = CmiWallTimer();
  while (CmiWallTimer() - start < busyWait) ;
}

void POSE_busy_wait(double n)
{
  double start = CmiWallTimer();
  while (CmiWallTimer() - start < n) ;
}

void pose::registerCallBack(callBack *cbm) 
{
  callBackSet = 1;
  cb = cbm->callback;
}

void pose::stop(void) 
{ 
#ifdef SEQUENTIAL_POSE
#if USE_LONG_TIMESTAMPS
  CkPrintf("Sequential Endtime Approximation: %lld\n", POSE_GlobalClock);
#else
  CkPrintf("Sequential Endtime Approximation: %d\n", POSE_GlobalClock);
#endif
  // Call sequential termination here, when done it calls prepExit
  POSE_Objects.Terminate();
#endif
  // prepExit();
}

void pose::prepExit(void) 
{
#ifndef CMK_OPTIMIZE
  if(pose_config.stats)
    {
      CProxy_localStat stats(theLocalStats);
      CkPrintf("%d PE Simulation finished at %f. Gathering stats...\n", 
               CkNumPes(), CmiWallTimer() - sim_timer);
      stats.SendStats();
    }
  else
    {
      CkPrintf("%d PE Simulation finished at %f.\n", CkNumPes(), 
               CmiWallTimer() - sim_timer);
      POSE_exit();
    }
#else
  CkPrintf("%d PE Simulation finished at %f.\n", CkNumPes(), 
           CmiWallTimer() - sim_timer);
  POSE_exit();
#endif  
}

void pose::exit(void) 
{ 
  if (callBackSet)
    cb.send(); // need to make callback here
  else
    CkExit();
}

// this is a HACK to get module seqpose working
void _registerseqpose(void)
{
  _registerpose();
}

void POSEreadCmdLine()
{
  char **argv = CkGetArgv();
  CmiArgGroup("Charm++","POSE");
  pose_config.stats=CmiGetArgFlagDesc(argv, "+stats_pose",
                        "Gather timing information and other statistics");
  /*  semantic meaning for these still to be determined
  CmiGetArgIntDesc(argv, "+start_proj_pose",&pose_config.start_proj,
                        "GVT to initiate projections tracing");
  CmiGetArgIntDesc(argv, "+end_proj_pose",&pose_config.end_proj,
                        "GVT to end projections tracing");
  */
  pose_config.trace=CmiGetArgFlagDesc(argv, "+trace_pose",
                        "Traces key POSE operations like Forward Execution, Rollback, Cancellation, Fossil Collection, etc. via user events for display in projections");

  pose_config.dop=CmiGetArgFlagDesc(argv, "+dop_pose",
                        "Critical path analysis by measuring degree of parallelism");

  CmiGetArgIntDesc(argv, "+memman_pose", &pose_config.max_usage , "Coarse memory management: Restricts forward execution of objects with over <max_usage>/<checkpoint store_rate> checkpoints; default to 10");
  /*
  pose_config.msg_pool=CmiGetArgFlagDesc(argv, "+pose_msgpool",  "Store and reuse pools of messages under a certain size default 1000");
  CmiGetArgIntDesc(argv, "+msgpoolsize_pose", &pose_config.msg_pool_size , "Store and reuse pools of messages under a certain size default 1000");

  CmiGetArgIntDesc(argv, "+msgpoolmax_pose", &pose_config.max_pool_msg_size , "Store and reuse pools of messages under a certain size");
  char *strat;
  CmiGetArgStringDesc(argv, "+commlib_strat_pose", &strat , "Use commlib with strat in {stream|mesh|prio}");
  if(strcmp("stream",strat)==0)
    pose_config.commlib_strat=stream;
  if(strcmp("mesh",strat)==0)
    pose_config.commlib_strat=mesh;
  if(strcmp("prio",strat)==0)
    pose_config.commlib_strat=prio;
  CmiGetArgIntDesc(argv, "+commlib_timeout-pose", &pose_config.commlib_timeout , "Use commlib with timeout N; default 1ms");
  CmiGetArgIntDesc(argv, "+commlib_maxmsg_pose", &pose_config.commlib_maxmsg , "Use commlib with max msg N;  default 5");
  */
  pose_config.lb_on=CmiGetArgFlagDesc(argv, "+lb_on_pose", "Use load balancing");
  CmiGetArgIntDesc(argv, "+lb_skip_pose", &pose_config.lb_skip , "Load balancing skip N; default 51");
  CmiGetArgIntDesc(argv, "+lb_threshold_pose", &pose_config.lb_threshold , "Load balancing threshold N; default 4000");
  CmiGetArgIntDesc(argv, "+lb_diff_pose", &pose_config.lb_diff , "Load balancing  min diff between min and max load PEs; default 2000");
  CmiGetArgIntDesc(argv, "+checkpoint_rate_pose", &pose_config.store_rate , " Set checkpoint to 1 for every <rate> events. Default to 1. ");
  /* max_iteration seems to be defunct */
  //  CmiGetArgIntDesc(argv, "+FEmax_pose", &pose_config.max_iter , "Sets max events executed in single forward execution step.  Default to 100.");
  CmiGetArgIntDesc(argv, "+leash_specwindow_pose", &pose_config.spec_window , "Sets speculative window behavior.");
  CmiGetArgIntDesc(argv, "+leash_min_pose", &pose_config.min_leash , "Sets speculative window behavior minimum leash. Default 10.");
  CmiGetArgIntDesc(argv, "+leash_max_pose", &pose_config.max_leash , "Sets speculative window behavior maximum leash. Default 100.");
  CmiGetArgIntDesc(argv, "+leash_flex_pose", &pose_config.max_leash , "Sets speculative window behavior leash flex. Default 10.");
  if(pose_config.deterministic= CmiGetArgFlagDesc(argv, "+deterministic_pose",  "sorts events of same timestamp by event id for repeatable behavior "))
    {
      CkPrintf("enter at your own risk, this feature is broken\n");
    }


}  

---