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