Charm++: Charm Source Code Documentation

00001 
00002 #include "pose.h"
00003 #include "poseMsgs.def.h"
00004 #include "sim.def.h"
00005 
00007 CProxy_sim POSE_Objects;
00008 CProxy_sim POSE_Objects_RO;
00010 CkChareID POSE_Coordinator_ID;
00011 
00013 sim::sim() 
00014 {
00015 #ifdef VERBOSE_DEBUG
00016   CkPrintf("[%d] constructing sim %d\n",CkMyPe(), thisIndex);
00017 #endif
00018 
00019 #ifndef SEQUENTIAL_POSE
00020   localPVT = (PVT *)CkLocalBranch(ThePVT);
00021   if(pose_config.lb_on)
00022     localLBG = TheLBG.ckLocalBranch();
00023 #endif
00024 #if !CMK_TRACE_DISABLED
00025   if(pose_config.stats)
00026     localStats = (localStat *)CkLocalBranch(theLocalStats);
00027 #endif
00028   basicStats[0] = basicStats[1] = 0LL;
00029   lastGVT = active = DOs = UNDOs = 0;
00030   srVector = (int *)malloc(CkNumPes() * sizeof(int));
00031   for (int i=0; i<CkNumPes(); i++) srVector[i] = 0;
00032   eq = new eventQueue();
00033   myStrat = NULL;
00034   objID = NULL;
00035 }
00036 
00038 sim::~sim() 
00039 {
00040   active = -1;
00041 #ifndef SEQUENTIAL_POSE
00042   localPVT->objRemove(myPVTidx);
00043 #endif
00044   if(pose_config.lb_on)
00045     localLBG->objRemove(myLBidx);
00046 
00047   delete(eq);
00048   delete(myStrat);
00049   delete(objID);
00050 }
00051 
00053 void sim::pup(PUP::er &p) {
00054   // pup simple types
00055   p(active); p(myPVTidx); p(myLBidx); p(sync); p(DOs); p(UNDOs);
00056   // pup event queue
00057   if (p.isUnpacking()) {
00058     eq = new eventQueue();
00059   }
00060   eq->pup(p);
00061   // pup cancellations
00062   cancels.pup(p);
00063   if (p.isUnpacking()) { // reactivate migrated object
00064 #if !CMK_TRACE_DISABLED
00065     localStats = (localStat *)CkLocalBranch(theLocalStats);
00066 #endif
00067 #ifndef SEQUENTIAL_POSE
00068     localPVT = (PVT *)CkLocalBranch(ThePVT);
00069     myPVTidx = localPVT->objRegister(thisIndex, localPVT->getGVT(), sync, this);
00070     if(pose_config.lb_on){
00071       localLBG = TheLBG.ckLocalBranch();
00072       myLBidx = localLBG->objRegister(thisIndex, sync, this);
00073     }
00074 #endif
00075     active = 0;
00076   }
00077   PUParray(p, basicStats, 2);
00078   // pup checkpoint info for sequential mode using sim 0 only
00079 #ifdef SEQUENTIAL_POSE
00080   if (thisIndex == 0) {
00081     p|seqCheckpointInProgress;
00082     p|seqLastCheckpointGVT;
00083     p|seqLastCheckpointTime;
00084     p|seqStartTime;
00085     p|POSE_Skipped_Events;
00086     p|poseIndexOfStopEvent;
00087     if (p.isUnpacking()) {
00088       seqStartTime = seqLastCheckpointTime;
00089     }
00090   }
00091 #endif
00092 }
00093 
00095 void sim::Step()
00096 {
00097   if (active < 0) return; // object is migrating; deactivate it 
00098 
00099 #if !CMK_TRACE_DISABLED
00100   double critStart;
00101   if(pose_config.trace)
00102     critStart=CmiWallTimer();  // trace timing
00103   int tstat;
00104   if(pose_config.stats)
00105     {
00106       tstat= localStats->TimerRunning();
00107       if (!tstat)  localStats->TimerStart(SIM_TIMER);
00108       else localStats->SwitchTimer(SIM_TIMER);
00109     }
00110 #endif
00111 
00112   prioMsg *pm;
00113   switch (myStrat->STRAT_T) { // step based on strategy type
00114   case SEQ_T:
00115   case CONS_T:
00116   case OPT_T:
00117   case OPT2_T: // pass this step call directly to strategy
00118   case OPT3_T: // prioritize this step call if work exists
00119   case SPEC_T:
00120   case ADAPT_T:
00121   case ADAPT2_T:
00122   case ADAPT3_T:
00123   case ADAPT4_T:
00124   case ADAPT5_T: // pass this step call directly to strategy
00125     myStrat->Step();
00126     break;
00127   default: 
00128     CkPrintf("Invalid strategy type: %d\n", myStrat->STRAT_T); 
00129     break;
00130   }
00131 #if !CMK_TRACE_DISABLED
00132   if(pose_config.stats)  
00133     {
00134       if (!tstat)  localStats->TimerStop();
00135       else localStats->SwitchTimer(tstat);
00136     }
00137   if(pose_config.trace)
00138     traceUserBracketEvent(60, critStart, CmiWallTimer());
00139 #endif
00140 }
00141 
00143 void sim::Step(prioMsg *m)
00144 {
00145   CkFreeMsg(m);
00146   if (active < 0) return; // object is migrating; deactivate it 
00147 #if !CMK_TRACE_DISABLED
00148   int tstat;
00149   if(pose_config.stats)
00150     {
00151       tstat= localStats->TimerRunning();
00152       if (!tstat)
00153     localStats->TimerStart(SIM_TIMER);
00154       else localStats->SwitchTimer(SIM_TIMER);
00155     }
00156 #endif
00157 
00158   myStrat->Step(); // Call Step on strategy
00159 
00160 #if !CMK_TRACE_DISABLED
00161   if(pose_config.stats)
00162     {
00163       if (!tstat)
00164     localStats->TimerStop();
00165       else localStats->SwitchTimer(tstat);
00166     }
00167 #endif
00168 }
00169 
00171 void sim::CheckpointStep(eventMsg *m) {
00172   CkFreeMsg(m);
00173   if (active < 0) return; // object is migrating; deactivate it 
00174 #if !CMK_TRACE_DISABLED
00175   int tstat;
00176   if (pose_config.stats) {
00177     tstat = localStats->TimerRunning();
00178     if (!tstat)
00179       localStats->TimerStart(SIM_TIMER);
00180     else localStats->SwitchTimer(SIM_TIMER);
00181   }
00182 #endif
00183 
00184   // ensure sequential mode
00185   CkAssert(myStrat->STRAT_T == SEQ_T);
00186   myStrat->Step(); // Call Step on strategy
00187 
00188 #if !CMK_TRACE_DISABLED
00189   if (pose_config.stats) {
00190     if (!tstat)
00191       localStats->TimerStop();
00192     else localStats->SwitchTimer(tstat);
00193   }
00194 #endif
00195 }
00196 
00198 void sim::Commit()
00199 {
00200   if (active < 0)  return; // object is migrating
00201 #if !CMK_TRACE_DISABLED
00202   double critStart;
00203   if(pose_config.trace)
00204     critStart= CmiWallTimer();  // trace timing
00205   int tstat;
00206   if(pose_config.stats) {
00207     tstat = localStats->TimerRunning();
00208     if (!tstat)  localStats->TimerStart(SIM_TIMER);
00209     else localStats->SwitchTimer(SIM_TIMER);
00210   }
00211   if(pose_config.stats)
00212     localStats->SwitchTimer(FC_TIMER);
00213 #endif
00214   int isDone=localPVT->done(); 
00215   int curGVT=localPVT->getGVT();
00216   if (isDone) { // simulation inactive
00217     eq->CommitEvents(this, POSE_endtime); // commit all events in queue
00218     Terminate();// call terminus on all posers
00219   }
00220   else if (curGVT > lastGVT + 100) {  // What's the constant doing to us?
00221     lastGVT = curGVT;
00222     eq->CommitEvents(this, lastGVT); // commit events up to GVT
00223   }
00224 #if !CMK_TRACE_DISABLED
00225   if(pose_config.trace) {
00226     traceUserBracketEvent(50, critStart, CmiWallTimer());
00227     critStart = CmiWallTimer();
00228   }
00229   if(pose_config.stats)
00230     localStats->SwitchTimer(SIM_TIMER);
00231 #endif
00232   if (!isDone && (eq->currentPtr->timestamp > -1))
00233     Step(); // not done; try stepping again
00234 
00235 #if !CMK_TRACE_DISABLED
00236   if(pose_config.stats) {
00237     if (!tstat)  localStats->TimerStop();
00238     else localStats->SwitchTimer(tstat);
00239   }
00240   if(pose_config.trace)
00241     traceUserBracketEvent(60, critStart, CmiWallTimer());
00242 #endif
00243 }
00244 
00246 /*  This is necessary to ensure a minimal state at checkpoint time.
00247     sim::Commit() requires a minimum advancement in GVT (currently 100
00248     ticks) before the event queue commits its events.  As a result,
00249     some events could be left in the queue.  This function takes care
00250     of that.
00251 */
00252 void sim::CheckpointCommit() {
00253   if (active < 0)  return; // object is migrating
00254 #if !CMK_TRACE_DISABLED
00255   double critStart;
00256   if(pose_config.trace)
00257     critStart= CmiWallTimer();  // trace timing
00258   int tstat;
00259   if(pose_config.stats) {
00260     tstat = localStats->TimerRunning();
00261     if (!tstat)  localStats->TimerStart(SIM_TIMER);
00262     else localStats->SwitchTimer(SIM_TIMER);
00263   }
00264   if(pose_config.stats)
00265     localStats->SwitchTimer(FC_TIMER);
00266 #endif
00267   int curGVT = localPVT->getGVT();
00268   lastGVT = curGVT;
00269   eq->CommitEvents(this, lastGVT); // commit everything up to the current GVT
00270 #if !CMK_TRACE_DISABLED
00271   if(pose_config.trace) {
00272     traceUserBracketEvent(50, critStart, CmiWallTimer());
00273     critStart = CmiWallTimer();
00274   }
00275   if(pose_config.stats)
00276     localStats->SwitchTimer(SIM_TIMER);
00277 #endif
00278 #if !CMK_TRACE_DISABLED
00279   if(pose_config.stats) {
00280     if (!tstat)  localStats->TimerStop();
00281     else localStats->SwitchTimer(tstat);
00282   }
00283   if(pose_config.trace)
00284     traceUserBracketEvent(60, critStart, CmiWallTimer());
00285 #endif
00286 }
00287 
00289 void sim::ReportLBdata()
00290 {
00291   if(pose_config.lb_on){
00292     double rbOh;
00293     int numEvents = 0;
00294     Event *tmp = eq->currentPtr;
00295 
00296     if (DOs-UNDOs == 0) rbOh = 1.0;
00297     else rbOh = ((double)DOs)/((double)(DOs-UNDOs));
00298     while (tmp->timestamp > POSE_UnsetTS) {
00299       numEvents++;
00300       tmp = tmp->next;
00301     }
00302     localLBG->objUpdate(myLBidx, objID->ovt, eq->currentPtr->timestamp,
00303             numEvents, rbOh, srVector);
00304     DOs = UNDOs = 0;
00305     for (int i=0; i<CkNumPes(); i++) srVector[i] = 0;
00306   }
00307 }
00308 
00310 void sim::Cancel(cancelMsg *m) 
00311 {
00312 #if !CMK_TRACE_DISABLED
00313   if(pose_config.stats)
00314     localStats->TimerStart(CAN_TIMER);
00315 #endif
00316   //  char str[20];
00317   //  CkPrintf("[%d] RECV(cancel) %s at %d...\n", CkMyPe(), m->evID.sdump(str), m->timestamp);      
00318   //localPVT = (PVT *)CkLocalBranch(ThePVT);
00319   cancels.Insert(m->timestamp, m->evID); // add to cancellations list
00320   localPVT->objUpdate(m->timestamp, RECV); // tell PVT branch about recv
00321   CkFreeMsg(m);
00322 
00323 #if !CMK_TRACE_DISABLED
00324   double critStart;
00325   if(pose_config.trace)
00326     critStart= CmiWallTimer();  // trace timing
00327   if(pose_config.stats)
00328     localStats->SwitchTimer(SIM_TIMER);      
00329 #endif
00330 
00331   myStrat->Step(); // call Step to handle cancellation
00332 
00333 #if !CMK_TRACE_DISABLED
00334   if(pose_config.stats)
00335     localStats->TimerStop();
00336   if(pose_config.trace)
00337     traceUserBracketEvent(60, critStart, CmiWallTimer());
00338 #endif
00339 }
00340 
00341 // Sequential checkpointing: Two functions, SeqBeginCheckpoint and
00342 // SeqResumeAfterCheckpoint, were added to the sim class to
00343 // handle this.  Only initiate the checkpointing procedure on sim
00344 // 0, after commits have occurred.  This should minimize the
00345 // amount of data written to disk.  In order to ensure a stable
00346 // state, we wait for quiescence to be reached before beginning
00347 // the checkpoint.  Once this happens, sim 0 checkpoints and then
00348 // resumes the simulation in SeqResumeAfterCheckpoint.  This
00349 // Callback function is also the first POSE function to be called
00350 // when restarting from a checkpoint.
00351 
00352 // While waiting for quiescence to be reached, all events with
00353 // timestamps less than the checkpoint GVT are allowed to
00354 // execute.  All others are skipped, instead storing their sim handles (indices
00355 // into the POSE_Objects array) in POSE_Skipped_Events.  Even
00356 // though execution is skipped, the events still remain in their
00357 // event queues.  When resuming the simulation, both after
00358 // checkpointing and after a restart, sim::Step() (which calls
00359 // seq::Step()) is called on each poser listed in
00360 // POSE_Skipped_Events to execute the skipped events.
00361 
00362 // Checkpoints are initiated approximately every
00363 // pose_config.checkpoint_gvt_interval GVT ticks or
00364 // pose_config.checkpoint_time_interval seconds (both defined in
00365 // pose_config.h).
00366 
00368 void sim::SeqBeginCheckpoint() {
00369   // Ensure this only happens on sim 0
00370   CkAssert(thisIndex == 0);
00371   // Ensure we're checkpointing
00372   CkAssert(seqCheckpointInProgress);
00373   CkPrintf("POSE: quiescence detected\n");
00374   CkPrintf("POSE: beginning checkpoint on sim %d at GVT=%lld sim time=%.1f sec\n", thisIndex, seqLastCheckpointGVT, CmiWallTimer() + seqStartTime);
00375   CkCallback cb(CkIndex_sim::SeqResumeAfterCheckpoint(), CkArrayIndex1D(thisIndex), thisProxy);
00376   CkStartCheckpoint(POSE_CHECKPOINT_DIRECTORY, cb);
00377 }
00378 
00380 void sim::SeqResumeAfterCheckpoint() {
00381   // Ensure this only happens on sim 0
00382   CkAssert(thisIndex == 0);
00383   // Ensure this function is only called once after a checkpoint
00384   CkAssert(seqCheckpointInProgress);
00385   seqCheckpointInProgress = 0;
00386   POSE_GlobalClock = seqLastCheckpointGVT;
00387   CkPrintf("POSE: checkpoint/restart complete on sim %d at GVT=%lld sim time=%.1f sec\n", thisIndex, POSE_GlobalClock, CmiWallTimer() + seqStartTime);
00388 
00389   // restart simulation
00390   while (POSE_Skipped_Events.length() > 0) {
00391     // These Step iterations must all be enqueued now, before any messages
00392     // are delivered, or else the event queues will break.  Because
00393     // messages spawned by these events may need to be inserted, these
00394     // events should be enqueued in the same manner as that used by
00395     // the translated verion of POSE_invoke
00396     eventMsg *evtMsg = new eventMsg;
00397     Skipped_Event se = POSE_Skipped_Events.deq();
00398     int _POSE_handle = se.simIndex;
00399     POSE_TimeType _POSE_timeOffset = se.timestamp - objID->ovt;
00400     objID->registerTimestamp(_POSE_handle, evtMsg, _POSE_timeOffset);
00401     if (pose_config.dop) {
00402       ct = CmiWallTimer();
00403       evtMsg->rst = ct - st + eq->currentPtr->srt;
00404     }
00405 #if !CMK_TRACE_DISABLED
00406     evtMsg->sanitize();
00407 #endif
00408 #if !CMK_TRACE_DISABLED
00409     if(pose_config.stats)
00410       localStats->SwitchTimer(COMM_TIMER);
00411 #endif
00412     thisProxy[_POSE_handle].CheckpointStep(evtMsg);
00413 #if !CMK_TRACE_DISABLED
00414     if (pose_config.stats)
00415       localStats->SwitchTimer(DO_TIMER);
00416 #endif
00417   }
00418 
00419   // restart quiescence detection, which is used for termination of
00420   // sequential POSE
00421   CkStartQD(CkIndex_pose::stop(), &POSE_Coordinator_ID);
00422 }
00423 
00424 void sim::ResumeFromSync()
00425 {
00426 //  PVT *localPVT = (PVT *)CkLocalBranch(ThePVT);
00427 //  localPVT->doneLB();
00428 }
00429 
00431 void sim::dump()
00432 {
00433   CkPrintf("[SIM: thisIndex=%d active=%d sync=%d myPVTidx=%d ", thisIndex, active, sync, myPVTidx);
00434   if (objID) objID->dump();
00435   else CkPrintf("objID=NULL\n");
00436   eq->dump();
00437   cancels.dump();
00438   CkPrintf("end SIM]\n");
00439 }
00440
libs/ck-libs/pose/sim.C
