Charm++: Charm Source Code Documentation

00001 /*
00002 Threaded Charm++ "Framework Framework"
00003 
00004 Orion Sky Lawlor, olawlor@acm.org, 11/19/2001
00005  */
00006 #include "tcharm_impl.h"
00007 #include "tcharm.h"
00008 #include "ckevacuation.h"
00009 #include <ctype.h>
00010 
00011 #if 0 
00012     /*Many debugging statements:*/
00013 #    define DBG(x) ckout<<"["<<thisIndex<<","<<CkMyPe()<<"] TCHARM> "<<x<<endl;
00014 #    define DBGX(x) ckout<<"PE("<<CkMyPe()<<") TCHARM> "<<x<<endl;
00015 #else
00016     /*No debugging statements*/
00017 #    define DBG(x) /*empty*/
00018 #    define DBGX(x) /*empty*/
00019 #endif
00020 
00021 CtvDeclare(TCharm *,_curTCharm);
00022 
00023 static int lastNumChunks=0;
00024 
00025 class TCharmTraceLibList {
00026         enum {maxLibs=20,maxLibNameLen=15};
00027         //List of libraries we want to trace:
00028         int curLibs;
00029         char libNames[maxLibs][maxLibNameLen];
00030         int checkIfTracing(const char *lib) const
00031         {
00032                 for (int i=0;i<curLibs;i++) 
00033                         if (0==strcmp(lib,libNames[i]))
00034                                 return 1;
00035                 return 0;
00036         }
00037 public:
00038         TCharmTraceLibList() {curLibs=0;}
00039         void addTracing(const char *lib) 
00040         { //We want to trace this library-- add its name to the list.
00041                 CkPrintf("TCHARM> Will trace calls to library %s\n",lib);
00042                 int i;
00043                 for (i=0;0!=*lib;i++,lib++)
00044                         libNames[curLibs][i]=tolower(*lib);
00045                 libNames[curLibs][i]=0;
00046                 // if already tracing, skip
00047                 if (checkIfTracing(libNames[curLibs])) return;
00048                 curLibs++;
00049         }
00050         inline int isTracing(const char *lib) const {
00051                 if (curLibs==0) return 0; //Common case
00052                 else return checkIfTracing(lib);
00053         }
00054 };
00055 static TCharmTraceLibList tcharm_tracelibs;
00056 static int tcharm_nomig=0, tcharm_nothreads=0;
00057 static int tcharm_stacksize=1*1024*1024; /*Default stack size is 1MB*/
00058 static int tcharm_initted=0;
00059 CkpvDeclare(int, mapCreated);
00060 static CkGroupID mapID;
00061 static char* mapping = NULL;
00062 
00063 void TCharm::nodeInit(void)
00064 {
00065 }
00066 
00067 void TCharm::procInit(void)
00068 {
00069   CtvInitialize(TCharm *,_curTCharm);
00070   CtvAccess(_curTCharm)=NULL;
00071   tcharm_initted=1;
00072   CtgInit();
00073 
00074   CkpvInitialize(int, mapCreated);
00075   CkpvAccess(mapCreated) = 0;
00076 
00077   // called on every pe to eat these arguments
00078   char **argv=CkGetArgv();
00079   tcharm_nomig=CmiGetArgFlagDesc(argv,"+tcharm_nomig","Disable migration support (debugging)");
00080   tcharm_nothreads=CmiGetArgFlagDesc(argv,"+tcharm_nothread","Disable thread support (debugging)");
00081   tcharm_nothreads|=CmiGetArgFlagDesc(argv,"+tcharm_nothreads",NULL);
00082   char *traceLibName=NULL;
00083   while (CmiGetArgStringDesc(argv,"+tcharm_trace",&traceLibName,"Print each call to this library"))
00084       tcharm_tracelibs.addTracing(traceLibName);
00085   // CmiGetArgIntDesc(argv,"+tcharm_stacksize",&tcharm_stacksize,"Set the thread stack size (default 1MB)");
00086   char *str;
00087   if (CmiGetArgStringDesc(argv,"+tcharm_stacksize",&str,"Set the thread stack size (default 1MB)"))  {
00088     if (strpbrk(str,"M")) {
00089       sscanf(str, "%dM", &tcharm_stacksize);
00090       tcharm_stacksize *= 1024*1024;
00091     }
00092     else if (strpbrk(str,"K")) {
00093       sscanf(str, "%dK", &tcharm_stacksize);
00094       tcharm_stacksize *= 1024;
00095     }
00096     else {
00097       sscanf(str, "%d", &tcharm_stacksize);
00098     }
00099     if (CkMyPe() == 0)
00100       CkPrintf("TCharm> stack size is set to %d.\n", tcharm_stacksize);
00101   }
00102   if (CkMyPe()!=0) { //Processor 0 eats "+vp<N>" and "-vp<N>" later:
00103         int ignored;
00104         while (CmiGetArgIntDesc(argv,"-vp",&ignored,NULL)) {}
00105         while (CmiGetArgIntDesc(argv,"+vp",&ignored,NULL)) {}
00106   }
00107   if (CkMyPe()==0) { // Echo various debugging options:
00108     if (tcharm_nomig) CmiPrintf("TCHARM> Disabling migration support, for debugging\n");
00109     if (tcharm_nothreads) CmiPrintf("TCHARM> Disabling thread support, for debugging\n");
00110   }
00111   if (CkpvAccess(mapCreated)==0) {
00112     if (0!=CmiGetArgString(argv, "+mapping", &mapping)){
00113     }
00114     CkpvAccess(mapCreated)=1;
00115   }
00116 }
00117 
00118 void TCHARM_Api_trace(const char *routineName,const char *libraryName)
00119 {
00120         if (!tcharm_tracelibs.isTracing(libraryName)) return;
00121         TCharm *tc=CtvAccess(_curTCharm);
00122         char where[100];
00123         if (tc==NULL) sprintf(where,"[serial context on %d]",CkMyPe());
00124         else sprintf(where,"[%p> vp %d, p %d]",(void *)tc,tc->getElement(),CkMyPe());
00125         CmiPrintf("%s Called routine %s\n",where,routineName);
00126         CmiPrintStackTrace(1);
00127         CmiPrintf("\n");
00128 }
00129 
00130 // register thread start functions to get a function handler
00131 // this is portable across heterogeneous platforms, or on machines with
00132 // random stack/function pointer
00133 
00134 static CkVec<TCHARM_Thread_data_start_fn> threadFnTable;
00135 
00136 int TCHARM_Register_thread_function(TCHARM_Thread_data_start_fn fn)
00137 {
00138   int idx = threadFnTable.size();
00139   threadFnTable.push_back(fn);
00140   return idx+1;                     // make 0 invalid number
00141 }
00142 
00143 TCHARM_Thread_data_start_fn getTCharmThreadFunction(int idx)
00144 {
00145   CmiAssert(idx > 0);
00146   return threadFnTable[idx-1];
00147 }
00148 
00149 static void startTCharmThread(TCharmInitMsg *msg)
00150 {
00151         DBGX("thread started");
00152         TCharm::activateThread();
00153        TCHARM_Thread_data_start_fn threadFn = getTCharmThreadFunction(msg->threadFn);
00154         threadFn(msg->data);
00155         TCharm::deactivateThread();
00156         CtvAccess(_curTCharm)->done();
00157 }
00158 
00159 TCharm::TCharm(TCharmInitMsg *initMsg_)
00160 {
00161   initMsg=initMsg_;
00162   initMsg->opts.sanityCheck();
00163   timeOffset=0.0;
00164   if (tcharm_nothreads)
00165   { //Don't even make a new thread-- just use main thread
00166     tid=CthSelf();
00167   }
00168   else /*Create a thread normally*/
00169   {
00170     if (tcharm_nomig) { /*Nonmigratable version, for debugging*/
00171       tid=CthCreate((CthVoidFn)startTCharmThread,initMsg,initMsg->opts.stackSize);
00172     } else {
00173       tid=CthCreateMigratable((CthVoidFn)startTCharmThread,initMsg,initMsg->opts.stackSize);
00174     }
00175 #if CMK_BIGSIM_CHARM
00176     BgAttach(tid);
00177     BgUnsetStartOutOfCore();
00178 #endif
00179   }
00180   threadGlobals=CtgCreate(tid);
00181   CtvAccessOther(tid,_curTCharm)=this;
00182   isStopped=true;
00183   resumeAfterMigration=false;
00184         /* FAULT_EVAC*/
00185         AsyncEvacuate(CmiTrue);
00186   skipResume=false;
00187   exitWhenDone=initMsg->opts.exitWhenDone;
00188   isSelfDone = false;
00189   threadInfo.tProxy=CProxy_TCharm(thisArrayID);
00190   threadInfo.thisElement=thisIndex;
00191   threadInfo.numElements=initMsg->numElements;
00192   if (1 || CmiMemoryIs(CMI_MEMORY_IS_ISOMALLOC)) {
00193         heapBlocks=CmiIsomallocBlockListNew(tid);
00194   } else
00195         heapBlocks=0;
00196   nUd=0;
00197   usesAtSync=CmiTrue;
00198   run();
00199 }
00200 
00201 TCharm::TCharm(CkMigrateMessage *msg)
00202         :CBase_TCharm(msg)
00203 {
00204   initMsg=NULL;
00205   tid=NULL;
00206   threadGlobals=NULL;
00207   threadInfo.tProxy=CProxy_TCharm(thisArrayID);
00208         AsyncEvacuate(CmiTrue);
00209   heapBlocks=0;
00210 }
00211 
00212 void checkPupMismatch(PUP::er &p,int expected,const char *where)
00213 {
00214         int v=expected;
00215         p|v;
00216         if (v!=expected) {
00217                 CkError("FATAL ERROR> Mismatch %s pup routine\n",where);
00218                 CkAbort("FATAL ERROR: Pup direction mismatch");
00219         }
00220 }
00221 
00222 void TCharm::pup(PUP::er &p) {
00223 //Pup superclass
00224   ArrayElement1D::pup(p);
00225 
00226   //BIGSIM_OOC DEBUGGING
00227   //if(!p.isUnpacking()){
00228   //  CmiPrintf("TCharm[%d] packing: ", thisIndex);
00229   //  CthPrintThdStack(tid);
00230   //}
00231 
00232   checkPupMismatch(p,5134,"before TCHARM");
00233 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
00234     if(!isStopped){
00235 //      resumeAfterMigration = true;
00236     }
00237     isStopped = true;
00238 #endif
00239   p(isStopped); p(resumeAfterMigration); p(exitWhenDone); p(isSelfDone); p(skipResume);
00240   p(threadInfo.thisElement);
00241   p(threadInfo.numElements);
00242   
00243   if (sema.size()>0){
00244         CkAbort("TCharm::pup> Cannot migrate with unconsumed semaphores!\n");
00245   }
00246 
00247 #ifndef CMK_OPTIMIZE
00248   DBG("Packing thread");
00249   if (!isStopped && !CmiMemoryIs(CMI_MEMORY_IS_ISOMALLOC)){
00250     if(_BgOutOfCoreFlag==0) //not doing out-of-core scheduling
00251         CkAbort("Cannot pup a running thread.  You must suspend before migrating.\n");
00252   }     
00253   if (tcharm_nomig) CkAbort("Cannot migrate with the +tcharm_nomig option!\n");
00254 #endif
00255 
00256   //This seekBlock allows us to reorder the packing/unpacking--
00257   // This is needed because the userData depends on the thread's stack
00258   // and heap data both at pack and unpack time.
00259   PUP::seekBlock s(p,2);
00260   
00261   if (p.isUnpacking())
00262   {//In this case, unpack the thread & heap before the user data
00263     s.seek(1);
00264     pupThread(p);
00265     //Restart our clock: set it up so packTime==CkWallTimer+timeOffset
00266     double packTime;
00267     p(packTime);
00268     timeOffset=packTime-CkWallTimer();
00269   }
00270   
00271 //Pack all user data
00272   // Set up TCHARM context for use during user's pup routines:
00273   CtvAccess(_curTCharm)=this;
00274   activateThread();
00275 
00276   s.seek(0);
00277   checkPupMismatch(p,5135,"before TCHARM user data");
00278   p(nUd);
00279   for(int i=0;i<nUd;i++) {
00280     if (p.isUnpacking()) ud[i].update(tid);
00281     ud[i].pup(p);
00282   }
00283   checkPupMismatch(p,5137,"after TCHARM_Register user data");
00284 
00285   if (CmiMemoryIs(CMI_MEMORY_IS_ISOMALLOC))
00286     deactivateThread();
00287   p|sud;           //  sud vector block can not be in isomalloc
00288   checkPupMismatch(p,5138,"after TCHARM_Global user data");
00289   
00290   // Tear down TCHARM context after calling user pup routines
00291   if (!CmiMemoryIs(CMI_MEMORY_IS_ISOMALLOC))
00292     deactivateThread();
00293   CtvAccess(_curTCharm)=NULL;
00294   
00295   if (!p.isUnpacking())
00296   {//In this case, pack the thread & heap after the user data
00297     s.seek(1);
00298     pupThread(p);
00299     //Stop our clock:
00300     double packTime=CkWallTimer()+timeOffset;
00301     p(packTime);
00302   }
00303   
00304   s.endBlock(); //End of seeking block
00305   checkPupMismatch(p,5140,"after TCHARM");
00306   
00307   //BIGSIM_OOC DEBUGGING
00308   //if(p.isUnpacking()){
00309   //  CmiPrintf("TCharm[%d] unpacking: ", thisIndex);
00310   //  CthPrintThdStack(tid);
00311   //}
00312 
00313 }
00314 
00315 // Pup our thread and related data
00316 void TCharm::pupThread(PUP::er &pc) {
00317     pup_er p=(pup_er)&pc;
00318     checkPupMismatch(pc,5138,"before TCHARM thread");
00319     if (1 || CmiMemoryIs(CMI_MEMORY_IS_ISOMALLOC))
00320       CmiIsomallocBlockListPup(p,&heapBlocks,tid);
00321     tid = CthPup(p, tid);
00322     if (pc.isUnpacking()) {
00323       CtvAccessOther(tid,_curTCharm)=this;
00324 #if CMK_BIGSIM_CHARM
00325       BgAttach(tid);
00326 #endif
00327     }
00328     threadGlobals=CtgPup(p,threadGlobals);
00329     checkPupMismatch(pc,5139,"after TCHARM thread");
00330 }
00331 
00332 //Pup one group of user data
00333 void TCharm::UserData::pup(PUP::er &p)
00334 {
00335   pup_er pext=(pup_er)(&p);
00336   p(mode);
00337   switch(mode) {
00338   case 'c': { /* C mode: userdata is on the stack, so keep address */
00339 //     p((char*)&data,sizeof(data));
00340      p(pos);
00341      //FIXME: function pointers may not be valid across processors
00342      p((char*)&cfn, sizeof(TCHARM_Pup_fn));
00343      char *data = CthPointer(t, pos);
00344      if (cfn) cfn(pext,data);
00345      } break;
00346   case 'g': { /* Global mode: zero out userdata on arrival */
00347      if (CmiMemoryIs(CMI_MEMORY_IS_ISOMALLOC))
00348      {
00349         // keep the pointer value if using isomalloc, no need to use pup
00350        p(pos);
00351      }
00352      else if (p.isUnpacking())      //  zero out userdata on arrival
00353        pos=0;
00354 
00355        //FIXME: function pointers may not be valid across processors
00356      p((char*)&gfn, sizeof(TCHARM_Pup_global_fn));
00357      if (gfn) gfn(pext);
00358      } break;
00359   default:
00360      break;
00361   };
00362 }
00363 
00364 TCharm::~TCharm()
00365 {
00366   //BIGSIM_OOC DEBUGGING
00367   //CmiPrintf("TCharm destructor called with heapBlocks=%p!\n", heapBlocks);
00368   
00369 #if !CMK_USE_MEMPOOL_ISOMALLOC
00370   if (heapBlocks) CmiIsomallocBlockListDelete(heapBlocks);
00371 #endif
00372   CthFree(tid);
00373   CtgFree(threadGlobals);
00374   delete initMsg;
00375 }
00376 
00377 void TCharm::migrateTo(int destPE) {
00378         if (destPE==CkMyPe()) return;
00379         if (CthMigratable() == 0) {
00380             CkPrintf("Warning> thread migration is not supported!\n");
00381             return;
00382         }
00383         // Make sure migrateMe gets called *after* we suspend:
00384         thisProxy[thisIndex].migrateDelayed(destPE);
00385 //      resumeAfterMigration=true;
00386         suspend();
00387 }
00388 void TCharm::migrateDelayed(int destPE) {
00389         migrateMe(destPE);
00390 }
00391 void TCharm::ckJustMigrated(void) {
00392         ArrayElement::ckJustMigrated();
00393 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
00394 //  resumeAfterMigration = true;
00395 #endif
00396         if (resumeAfterMigration) {
00397                 resumeAfterMigration=false;
00398                 resume(); //Start the thread running
00399         }
00400 }
00401 
00402 void TCharm::ckJustRestored(void) {
00403         //CkPrintf("call just restored from TCharm[%d]\n", thisIndex);
00404         ArrayElement::ckJustRestored();
00405         //if (resumeAfterMigration) {
00406         //      resumeAfterMigration=false;
00407         //      resume(); //Start the thread running
00408         //}
00409 }
00410 
00411 /*
00412         FAULT_EVAC
00413 
00414         If a Tcharm object is about to migrate it should be suspended first
00415 */
00416 void TCharm::ckAboutToMigrate(void){
00417         ArrayElement::ckAboutToMigrate();
00418         resumeAfterMigration = true;
00419         isStopped = true;
00420 //      suspend();
00421 }
00422 
00423 // clear the data before restarting from disk
00424 void TCharm::clear()
00425 {
00426   if (heapBlocks) CmiIsomallocBlockListDelete(heapBlocks);
00427   CthFree(tid);
00428   delete initMsg;
00429 }
00430 
00431 //Register user data to be packed with the thread
00432 int TCharm::add(const TCharm::UserData &d)
00433 {
00434   if (nUd>=maxUserData)
00435     CkAbort("TCharm: Registered too many user data fields!\n");
00436   int nu=nUd++;
00437   ud[nu]=d;
00438   return nu;
00439 }
00440 void *TCharm::lookupUserData(int i) {
00441         if (i<0 || i>=nUd)
00442                 CkAbort("Bad user data index passed to TCharmGetUserdata!\n");
00443         return ud[i].getData();
00444 }
00445 
00446 //Start the thread running
00447 void TCharm::run(void)
00448 {
00449   DBG("TCharm::run()");
00450   if (tcharm_nothreads) {/*Call user routine directly*/
00451           startTCharmThread(initMsg);
00452   } 
00453   else /* start the thread as usual */
00454           start();
00455 }
00456 
00457 //Block the thread until start()ed again.
00458 void TCharm::stop(void)
00459 {
00460 #ifndef CMK_OPTIMIZE
00461   if (tid != CthSelf())
00462     CkAbort("Called TCharm::stop from outside TCharm thread!\n");
00463   if (tcharm_nothreads)
00464     CkAbort("Cannot make blocking calls using +tcharm_nothreads!\n");
00465 #endif
00466   stopTiming();
00467   isStopped=true;
00468   DBG("thread suspended");
00469 
00470   CthSuspend();
00471 //      DBG("thread resumed");
00472   /*SUBTLE: We have to do the get() because "this" may have changed
00473     during a migration-suspend.  If you access *any* members
00474     from this point onward, you'll cause heap corruption if
00475     we're resuming from migration!  (OSL 2003/9/23)
00476    */
00477   TCharm *dis=TCharm::get();
00478 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_)) 
00479 /*  CpvAccess(_currentObj) = dis;
00480  *      printf("[%d] _currentObject set to TCharm index %d %p\n",CkMyPe(),dis->thisIndex,dis);*/
00481 #endif
00482   dis->isStopped=false;
00483   dis->startTiming();
00484   //CkPrintf("[%d] Thread resumed  for tid %p\n",dis->thisIndex,dis->tid);
00485 }
00486 
00487 //Resume the waiting thread
00488 void TCharm::start(void)
00489 {
00490   //  since this thread is scheduled, it is not a good idea to migrate 
00491   isStopped=false;
00492   DBG("thread resuming soon");
00493   //CkPrintf("TCharm[%d]::start()\n", thisIndex);
00494   //CmiPrintStackTrace(0);
00495 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
00496 //CthAwakenPrio(tid, CQS_QUEUEING_BFIFO, 1, &prio);
00497   CthAwaken(tid);
00498 #else
00499   CthAwaken(tid);
00500 #endif
00501 }
00502 
00503 //Block our thread, schedule, and come back:
00504 void TCharm::schedule(void) {
00505   DBG("thread schedule");
00506   start(); // Calls CthAwaken
00507   stop(); // Calls CthSuspend
00508 }
00509 
00510 //Go to sync, block, possibly migrate, and then resume
00511 void TCharm::migrate(void)
00512 {
00513 #if CMK_LBDB_ON
00514   DBG("going to sync");
00515   AtSync();
00516   stop();
00517 #else
00518   DBG("skipping sync, because there is no load balancer");
00519 #endif
00520 }
00521 
00522 
00523 
00524 void TCharm::evacuate(){
00525         /*
00526                 FAULT_EVAC
00527         */
00528         //CkClearAllArrayElementsCPP();
00529         if(CkpvAccess(startedEvac)){
00530 //              resumeAfterMigration=true;
00531                 CcdCallFnAfter((CcdVoidFn)CkEmmigrateElement, (void *)myRec, 1);
00532                 suspend();
00533                 return;
00534         }
00535         return;
00536 
00537 }
00538 
00539 //calls atsync with async mode
00540 void TCharm::async_migrate(void)
00541 {
00542 #if CMK_LBDB_ON
00543   DBG("going to sync at async mode");
00544   skipResume = true;            // we resume immediately
00545   ReadyMigrate(false);
00546   AtSync(0);
00547   schedule();
00548 //  allow_migrate();
00549 #else
00550   DBG("skipping sync, because there is no load balancer");
00551 #endif
00552 }
00553 
00554 /*
00555 Note:
00556  thread can only migrate at the point when this is called
00557 */
00558 void TCharm::allow_migrate(void)
00559 {
00560 #if CMK_LBDB_ON
00561 //  ReadyMigrate(true);
00562   int nextPe = MigrateToPe();
00563   if (nextPe != -1) {
00564     migrateTo(nextPe);
00565   }
00566 #else
00567   DBG("skipping sync, because there is no load balancer");
00568 #endif
00569 }
00570 
00571 //Resume from sync: start the thread again
00572 void TCharm::ResumeFromSync(void)
00573 {
00574   //if(isSelfDone) return;
00575   //if (exitWhenDone) return; //for bigsim ooc execution
00576   if (!skipResume) start();
00577 }
00578 
00579 
00580 /****** TcharmClient ******/
00581 void TCharmClient1D::ckJustMigrated(void) {
00582   ArrayElement1D::ckJustMigrated();
00583   findThread();
00584   tcharmClientInit();
00585 }
00586 
00587 void TCharmClient1D::pup(PUP::er &p) {
00588   ArrayElement1D::pup(p);
00589   p|threadProxy;
00590 }
00591 
00592 CkArrayID TCHARM_Get_threads(void) {
00593         TCHARMAPI("TCHARM_Get_threads");
00594         return TCharm::get()->getProxy();
00595 }
00596 
00597 /************* Startup/Shutdown Coordination Support ************/
00598 
00599 // Useless values to reduce over:
00600 int _vals[2]={0,1};
00601 
00602 //Called when we want to go to a barrier
00603 void TCharm::barrier(void) {
00604         //Contribute to a synchronizing reduction
00605         CkCallback cb(index_t::atBarrier(0), thisProxy[0]);
00606         contribute(sizeof(_vals),&_vals,CkReduction::sum_int,cb);
00607 #if CMK_BIGSIM_CHARM
00608         void *curLog;           // store current log in timeline
00609         _TRACE_BG_TLINE_END(&curLog);
00610         TRACE_BG_AMPI_BREAK(NULL, "TCharm_Barrier_START", NULL, 0, 1);
00611 #endif
00612         stop();
00613 #if CMK_BIGSIM_CHARM
00614          _TRACE_BG_SET_INFO(NULL, "TCHARM_Barrier_END",  &curLog, 1);
00615 #endif
00616 }
00617 
00618 //Called when we've reached the barrier
00619 void TCharm::atBarrier(CkReductionMsg *m) {
00620         DBGX("clients all at barrier");
00621         delete m;
00622         thisProxy.start(); //Just restart everybody
00623 }
00624 
00625 //Called when the thread is done running
00626 void TCharm::done(void) {
00627         //CmiPrintStackTrace(0);
00628         DBG("TCharm thread "<<thisIndex<<" done")
00629         if (exitWhenDone) {
00630                 //Contribute to a synchronizing reduction
00631                 CkCallback cb(index_t::atExit(0), thisProxy[0]);
00632                 contribute(sizeof(_vals),&_vals,CkReduction::sum_int,cb);
00633         }
00634         isSelfDone = true;
00635         stop();
00636 }
00637 //Called when all threads are done running
00638 void TCharm::atExit(CkReductionMsg *m) {
00639         DBGX("TCharm::atExit1> exiting");
00640         delete m;
00641         //thisProxy.unsetFlags();
00642         CkExit();
00643         //CkPrintf("After CkExit()!!!!!!!\n");
00644 }
00645 
00646 /************* Setup **************/
00647 
00648 //Globals used to control setup process
00649 static TCHARM_Fallback_setup_fn g_fallbackSetup=NULL;
00650 void TCHARM_Set_fallback_setup(TCHARM_Fallback_setup_fn f)
00651 {
00652         g_fallbackSetup=f;
00653 }
00654 void TCHARM_Call_fallback_setup(void) {
00655         if (g_fallbackSetup) 
00656                 (g_fallbackSetup)();
00657         else
00658                 CkAbort("TCHARM: Unexpected fallback setup--missing TCHARM_User_setup routine?");
00659 }
00660 
00661 /************** User API ***************/
00662 /**********************************
00663 Callable from UserSetup:
00664 */
00665 
00666 // Read the command line to figure out how many threads to create:
00667 CDECL int TCHARM_Get_num_chunks(void)
00668 {
00669         TCHARMAPI("TCHARM_Get_num_chunks");
00670         if (CkMyPe()!=0) CkAbort("TCHARM_Get_num_chunks should only be called on PE 0 during setup!");
00671         int nChunks=CkNumPes();
00672         char **argv=CkGetArgv();
00673         CmiGetArgIntDesc(argv,"-vp",&nChunks,"Set the total number of virtual processors");
00674         CmiGetArgIntDesc(argv,"+vp",&nChunks,NULL);
00675         lastNumChunks=nChunks;
00676         return nChunks;
00677 }
00678 FDECL int FTN_NAME(TCHARM_GET_NUM_CHUNKS,tcharm_get_num_chunks)(void)
00679 {
00680         return TCHARM_Get_num_chunks();
00681 }
00682 
00683 // Fill out the default thread options:
00684 TCHARM_Thread_options::TCHARM_Thread_options(int doDefault)
00685 {
00686         stackSize=0; /* default stacksize */
00687         exitWhenDone=0; /* don't exit when done by default. */
00688 }
00689 void TCHARM_Thread_options::sanityCheck(void) {
00690         if (stackSize<=0) stackSize=tcharm_stacksize;
00691 }
00692 
00693 
00694 TCHARM_Thread_options g_tcharmOptions(1);
00695 
00696 /*Set the size of the thread stack*/
00697 CDECL void TCHARM_Set_stack_size(int newStackSize)
00698 {
00699         TCHARMAPI("TCHARM_Set_stack_size");
00700         g_tcharmOptions.stackSize=newStackSize;
00701 }
00702 FDECL void FTN_NAME(TCHARM_SET_STACK_SIZE,tcharm_set_stack_size)
00703         (int *newSize)
00704 { TCHARM_Set_stack_size(*newSize); }
00705 
00706 CDECL void TCHARM_Set_exit(void) { g_tcharmOptions.exitWhenDone=1; }
00707 
00708 /*Create a new array of threads, which will be bound to by subsequent libraries*/
00709 CDECL void TCHARM_Create(int nThreads,
00710                         int threadFn)
00711 {
00712         TCHARMAPI("TCHARM_Create");
00713         TCHARM_Create_data(nThreads,
00714                          threadFn,NULL,0);
00715 }
00716 FDECL void FTN_NAME(TCHARM_CREATE,tcharm_create)
00717         (int *nThreads,int threadFn)
00718 { TCHARM_Create(*nThreads,threadFn); }
00719 
00720 static CProxy_TCharm TCHARM_Build_threads(TCharmInitMsg *msg);
00721 
00722 /*As above, but pass along (arbitrary) data to threads*/
00723 CDECL void TCHARM_Create_data(int nThreads,
00724                   int threadFn,
00725                   void *threadData,int threadDataLen)
00726 {
00727         TCHARMAPI("TCHARM_Create_data");
00728         TCharmInitMsg *msg=new (threadDataLen,0) TCharmInitMsg(
00729                 threadFn,g_tcharmOptions);
00730         msg->numElements=nThreads;
00731         memcpy(msg->data,threadData,threadDataLen);
00732         TCHARM_Build_threads(msg);
00733         
00734         // Reset the thread options:
00735         g_tcharmOptions=TCHARM_Thread_options(1);
00736 }
00737 
00738 FDECL void FTN_NAME(TCHARM_CREATE_DATA,tcharm_create_data)
00739         (int *nThreads,
00740                   int threadFn,
00741                   void *threadData,int *threadDataLen)
00742 { TCHARM_Create_data(*nThreads,threadFn,threadData,*threadDataLen); }
00743 
00744 CkGroupID CkCreatePropMap(void);
00745 
00746 static CProxy_TCharm TCHARM_Build_threads(TCharmInitMsg *msg)
00747 {
00748   CkArrayOptions opts(msg->numElements);
00749   CkAssert(CkpvAccess(mapCreated)==1);
00750 
00751   if(haveConfigurableRRMap()){
00752     CkPrintf("USING ConfigurableRRMap\n");
00753     mapID=CProxy_ConfigurableRRMap::ckNew();
00754   } else if(mapping==NULL){
00755 #if CMK_BIGSIM_CHARM
00756     mapID=CProxy_BlockMap::ckNew();
00757 #else
00758 #if __FAULT__
00759         mapID=CProxy_RRMap::ckNew();
00760 #else
00761     mapID=CkCreatePropMap();
00762 #endif
00763 #endif
00764   } else if(0 == strcmp(mapping,"BLOCK_MAP")) {
00765     CkPrintf("USING BLOCK_MAP\n");
00766     mapID = CProxy_BlockMap::ckNew();
00767   } else if(0 == strcmp(mapping,"RR_MAP")) {
00768     CkPrintf("USING RR_MAP\n");
00769     mapID = CProxy_RRMap::ckNew();
00770   } else if(0 == strcmp(mapping,"MAPFILE")) {
00771     CkPrintf("Reading map from file\n");
00772     mapID = CProxy_ReadFileMap::ckNew();
00773   } else {  // "PROP_MAP" or anything else
00774     mapID = CkCreatePropMap();
00775   }
00776   opts.setMap(mapID);
00777   return CProxy_TCharm::ckNew(msg,opts);
00778 }
00779 
00780 // Helper used when creating a new array bound to the TCHARM threads:
00781 CkArrayOptions TCHARM_Attach_start(CkArrayID *retTCharmArray,int *retNumElts)
00782 {
00783         TCharm *tc=TCharm::get();
00784         if (!tc)
00785                 CkAbort("You must call TCHARM initialization routines from a TCHARM thread!");
00786         int nElts=tc->getNumElements();
00787       
00788         //CmiPrintf("TCHARM Elements = %d\n", nElts);  
00789       
00790         if (retNumElts!=NULL) *retNumElts=nElts;
00791         *retTCharmArray=tc->getProxy();
00792         CkArrayOptions opts(nElts);
00793         opts.bindTo(tc->getProxy());
00794         return opts;
00795 }
00796 
00797 void TCHARM_Suspend(void) {
00798         TCharm *tc=TCharm::get();
00799         tc->suspend();
00800 }
00801 
00802 /***********************************
00803 Callable from worker thread
00804 */
00805 CDECL int TCHARM_Element(void)
00806 { 
00807         TCHARMAPI("TCHARM_Element");
00808         return TCharm::get()->getElement();
00809 }
00810 CDECL int TCHARM_Num_elements(void)
00811 { 
00812         TCHARMAPI("TCHARM_Num_elements");
00813         return TCharm::get()->getNumElements();
00814 }
00815 
00816 FDECL int FTN_NAME(TCHARM_ELEMENT,tcharm_element)(void) 
00817 { return TCHARM_Element();}
00818 FDECL int FTN_NAME(TCHARM_NUM_ELEMENTS,tcharm_num_elements)(void) 
00819 { return TCHARM_Num_elements();}
00820 
00821 //Make sure this address will migrate with us when we move:
00822 static void checkAddress(void *data)
00823 {
00824         if (tcharm_nomig||tcharm_nothreads) return; //Stack is not isomalloc'd
00825         if (CmiThreadIs(CMI_THREAD_IS_ALIAS)||CmiThreadIs(CMI_THREAD_IS_STACKCOPY)) return; // memory alias thread
00826         if (CmiIsomallocEnabled()) {
00827           if (!CmiIsomallocInRange(data))
00828             CkAbort("The UserData you register must be allocated on the stack!\n");
00829         }
00830         else {
00831           if(CkMyPe() == 0)
00832             CkPrintf("Warning> checkAddress failed because isomalloc not supported.\n");
00833         }
00834 }
00835 
00836 /* Old "register"-based userdata: */
00837 CDECL int TCHARM_Register(void *data,TCHARM_Pup_fn pfn)
00838 { 
00839         TCHARMAPI("TCHARM_Register");
00840         checkAddress(data);
00841         return TCharm::get()->add(TCharm::UserData(pfn,TCharm::get()->getThread(),data));
00842 }
00843 FDECL int FTN_NAME(TCHARM_REGISTER,tcharm_register)
00844         (void *data,TCHARM_Pup_fn pfn)
00845 { 
00846         TCHARMAPI("TCHARM_Register");
00847         checkAddress(data);
00848         return TCharm::get()->add(TCharm::UserData(pfn,TCharm::get()->getThread(),data));
00849 }
00850 
00851 CDECL void *TCHARM_Get_userdata(int id)
00852 {
00853         TCHARMAPI("TCHARM_Get_userdata");
00854         return TCharm::get()->lookupUserData(id);
00855 }
00856 FDECL void *FTN_NAME(TCHARM_GET_USERDATA,tcharm_get_userdata)(int *id)
00857 { return TCHARM_Get_userdata(*id); }
00858 
00859 /* New hardcoded-ID userdata: */
00860 CDECL void TCHARM_Set_global(int globalID,void *new_value,TCHARM_Pup_global_fn pup_or_NULL)
00861 {
00862         TCHARMAPI("TCHARM_Set_global");
00863         TCharm *tc=TCharm::get();
00864         if (tc->sud.length()<=globalID)
00865         { //We don't have room for this ID yet: make room
00866                 int newLen=2*globalID;
00867                 tc->sud.resize(newLen);
00868         }
00869         tc->sud[globalID]=TCharm::UserData(pup_or_NULL,tc->getThread(),new_value);
00870 }
00871 CDECL void *TCHARM_Get_global(int globalID)
00872 {
00873         //Skip TCHARMAPI("TCHARM_Get_global") because there's no dynamic allocation here,
00874         // and this routine should be as fast as possible.
00875         CkVec<TCharm::UserData> &v=TCharm::get()->sud;
00876         if (v.length()<=globalID) return NULL; //Uninitialized global
00877         return v[globalID].getData();
00878 }
00879 
00880 CDECL void TCHARM_Migrate(void)
00881 {
00882         TCHARMAPI("TCHARM_Migrate");
00883         if (CthMigratable() == 0) {
00884           if(CkMyPe() == 0)
00885             CkPrintf("Warning> thread migration is not supported!\n");
00886           return;
00887         }
00888         TCharm::get()->migrate();
00889 }
00890 FORTRAN_AS_C(TCHARM_MIGRATE,TCHARM_Migrate,tcharm_migrate,(void),())
00891 
00892 CDECL void TCHARM_Async_Migrate(void)
00893 {
00894         TCHARMAPI("TCHARM_Async_Migrate");
00895         TCharm::get()->async_migrate();
00896 }
00897 FORTRAN_AS_C(TCHARM_ASYNC_MIGRATE,TCHARM_Async_Migrate,tcharm_async_migrate,(void),())
00898 
00899 CDECL void TCHARM_Allow_Migrate(void)
00900 {
00901         TCHARMAPI("TCHARM_Allow_Migrate");
00902         TCharm::get()->allow_migrate();
00903 }
00904 FORTRAN_AS_C(TCHARM_ALLOW_MIGRATE,TCHARM_Allow_Migrate,tcharm_allow_migrate,(void),())
00905 
00906 CDECL void TCHARM_Migrate_to(int destPE)
00907 {
00908         TCHARMAPI("TCHARM_Migrate_to");
00909         TCharm::get()->migrateTo(destPE);
00910 }
00911 
00912 CDECL void TCHARM_Evacuate()
00913 {
00914         TCHARMAPI("TCHARM_Migrate_to");
00915         TCharm::get()->evacuate();
00916 }
00917 
00918 FORTRAN_AS_C(TCHARM_MIGRATE_TO,TCHARM_Migrate_to,tcharm_migrate_to,
00919         (int *destPE),(*destPE))
00920 
00921 CDECL void TCHARM_Yield(void)
00922 {
00923         TCHARMAPI("TCHARM_Yield");
00924         TCharm::get()->schedule();
00925 }
00926 FORTRAN_AS_C(TCHARM_YIELD,TCHARM_Yield,tcharm_yield,(void),())
00927 
00928 CDECL void TCHARM_Barrier(void)
00929 {
00930         TCHARMAPI("TCHARM_Barrier");
00931         TCharm::get()->barrier();
00932 }
00933 FORTRAN_AS_C(TCHARM_BARRIER,TCHARM_Barrier,tcharm_barrier,(void),())
00934 
00935 CDECL void TCHARM_Done(void)
00936 {
00937         TCHARMAPI("TCHARM_Done");
00938         TCharm *c=TCharm::getNULL();
00939         if (!c) CkExit();
00940         else c->done();
00941 }
00942 FORTRAN_AS_C(TCHARM_DONE,TCHARM_Done,tcharm_done,(void),())
00943 
00944 
00945 CDECL double TCHARM_Wall_timer(void)
00946 {
00947   TCHARMAPI("TCHARM_Wall_timer");
00948   TCharm *c=TCharm::getNULL();
00949   if(!c) return CkWallTimer();
00950   else { //Have to apply current thread's time offset
00951     return CkWallTimer()+c->getTimeOffset();
00952   }
00953 }
00954 
00955 #if 1
00956 /*Include Fortran-style "iargc" and "getarg" routines.
00957 These are needed to get access to the command-line arguments from Fortran.
00958 */
00959 FDECL int FTN_NAME(TCHARM_IARGC,tcharm_iargc)(void) {
00960   TCHARMAPI("tcharm_iargc");
00961   return CkGetArgc()-1;
00962 }
00963 
00964 FDECL void FTN_NAME(TCHARM_GETARG,tcharm_getarg)
00965         (int *i_p,char *dest,int destLen)
00966 {
00967   TCHARMAPI("tcharm_getarg");
00968   int i=*i_p;
00969   if (i<0) CkAbort("tcharm_getarg called with negative argument!");
00970   if (i>=CkGetArgc()) CkAbort("tcharm_getarg called with argument > iargc!");
00971   const char *src=CkGetArgv()[i];
00972   strcpy(dest,src);
00973   for (i=strlen(dest);i<destLen;i++) dest[i]=' ';
00974 }
00975 
00976 #endif
00977 
00978 //These silly routines are used for serial startup:
00979 extern void _initCharm(int argc, char **argv);
00980 CDECL void TCHARM_Init(int *argc,char ***argv) {
00981         if (!tcharm_initted) {
00982           ConverseInit(*argc, *argv, (CmiStartFn) _initCharm,1,1);
00983           _initCharm(*argc,*argv);
00984         }
00985 }
00986 
00987 FDECL void FTN_NAME(TCHARM_INIT,tcharm_init)(void)
00988 {
00989         int argc=1;
00990         const char *argv_sto[2]={"foo",NULL};
00991         char **argv=(char **)argv_sto;
00992         TCHARM_Init(&argc,&argv);
00993 }
00994 
00995 /***********************************
00996 * TCHARM Semaphores:
00997 * The idea is one side "puts", the other side "gets"; 
00998 * but the calls can come in any order--
00999 * if the "get" comes first, it blocks until the put.
01000 * This makes a convenient, race-condition-free way to do
01001 * onetime initializations.  
01002 */
01004 TCharm::TCharmSemaphore *TCharm::findSema(int id) {
01005         for (unsigned int s=0;s<sema.size();s++)
01006                 if (sema[s].id==id) 
01007                         return &sema[s];
01008         sema.push_back(TCharmSemaphore(id));
01009         return &sema[sema.size()-1];
01010 }
01012 void TCharm::freeSema(TCharmSemaphore *doomed) {
01013         int id=doomed->id;
01014         for (unsigned int s=0;s<sema.size();s++)
01015                 if (sema[s].id==id) {
01016                         sema[s]=sema[sema.length()-1];
01017                         sema.length()--;
01018                         return;
01019                 }
01020         CkAbort("Tried to free nonexistent TCharm semaphore");
01021 }
01022 
01024 TCharm::TCharmSemaphore *TCharm::getSema(int id) {
01025         TCharmSemaphore *s=findSema(id);
01026         if (s->data==NULL) 
01027         { //Semaphore isn't filled yet: wait until it is
01028                 s->thread=CthSelf();
01029                 suspend(); //Will be woken by semaPut
01030                 // Semaphore may have moved-- find it again
01031                 s=findSema(id);
01032                 if (s->data==NULL) CkAbort("TCharm::semaGet awoken too early!");
01033         }
01034         return s;
01035 }
01036 
01039 void TCharm::semaPut(int id,void *data) {
01040         TCharmSemaphore *s=findSema(id);
01041         if (s->data!=NULL) CkAbort("Duplicate calls to TCharm::semaPut!");
01042         s->data=data;
01043         DBG("semaPut "<<id<<" "<<data);
01044         if (s->thread!=NULL) {//Awaken the thread
01045                 s->thread=NULL;
01046                 resume();
01047         }
01048 }
01049 
01053 void *TCharm::semaGet(int id) {
01054         TCharmSemaphore *s=getSema(id);
01055         void *ret=s->data;
01056         DBG("semaGet "<<id<<" "<<ret);
01057         // Now remove the semaphore from the list:
01058         freeSema(s);
01059         return ret;
01060 }
01061 
01064 void *TCharm::semaGets(int id) {
01065         TCharmSemaphore *s=getSema(id);
01066         return s->data;
01067 }
01068 
01070 void *TCharm::semaPeek(int id) {
01071         TCharmSemaphore *s=findSema(id);
01072         return s->data;
01073 }
01074 
01075 /****** System Call support ******/
01076 /*
01077 TCHARM_System exists to work around a bug where Linux ia32
01078 glibc2.2.x with pthreads crashes at the fork() site when 
01079 called from a user-levelthread. 
01080 
01081 The fix is to call system from the main thread, by 
01082 passing the request out of the thread to our array element 
01083 before calling system().
01084 */
01085 
01086 CDECL int 
01087 TCHARM_System(const char *shell_command)
01088 {
01089         return TCharm::get()->system(shell_command);
01090 }
01091 int TCharm::system(const char *cmd)
01092 {
01093         int ret=-1778;
01094         callSystemStruct s;
01095         s.cmd=cmd;
01096         s.ret=&ret;
01097         thisProxy[thisIndex].callSystem(s);
01098         suspend();
01099         return ret;
01100 }
01101 
01102 void TCharm::callSystem(const callSystemStruct &s)
01103 {
01104         *s.ret = ::system(s.cmd);
01105         resume();
01106 }
01107 
01108 
01109 
01110 #include "tcharm.def.h"
libs/ck-libs/tcharm/tcharm.C
