Make GetNumElements callable from init().
[charm.git] / src / libs / ck-libs / tcharm / tcharm.C
1 /*
2 Threaded Charm++ "Framework Framework"
3
4 Orion Sky Lawlor, olawlor@acm.org, 11/19/2001
5  */
6 #include "tcharm.h"
7
8 #if 0
9     /*Many debugging statements:*/
10 #    define DBG(x) ckout<<"["<<thisIndex<<"] TCHARM> "<<x<<endl;
11 #    define DBGX(x) ckout<<"PE("<<CkMyPe()<<") TCHARM> "<<x<<endl;
12 #else
13     /*No debugging statements*/
14 #    define DBG(x) /*empty*/
15 #    define DBGX(x) /*empty*/
16 #endif
17
18 CtvDeclare(TCharm *,_curTCharm);
19 CpvDeclare(inState,_stateTCharm);
20
21 static int lastNumChunks=0;
22 /*readonly*/ int tcharm_nomig=0, tcharm_nothreads=0;
23
24 void TCharm::nodeInit(void)
25 {
26   CtvInitialize(TCharm *,_curTCharm);
27   CpvInitialize(inState,_stateTCharm);
28   TCharm::setState(inNodeSetup);
29   TCharmUserNodeSetup();
30   FTN_NAME(TCHARM_USER_NODE_SETUP,tcharm_user_node_setup)();
31   TCharm::setState(inInit);
32 }
33
34 static void startTCharmThread(TCharmInitMsg *msg)
35 {
36         TCharm::setState(inDriver);
37         typedef void (*threadFn_t)(void *);
38         ((threadFn_t)msg->threadFn)(msg->data);
39         CtvAccess(_curTCharm)->done();
40 }
41
42 TCharm::TCharm(TCharmInitMsg *initMsg_)
43 {
44   initMsg=initMsg_;
45   if (tcharm_nothreads) 
46   { //Don't even make a new thread-- just use main thread
47     tid=CthSelf();
48   } 
49   else /*Create a thread normally*/
50   {
51     if (tcharm_nomig) /*Nonmigratable version, for debugging*/ 
52       tid=CthCreate((CthVoidFn)startTCharmThread,initMsg,initMsg->stackSize);
53     else
54       tid=CthCreateMigratable((CthVoidFn)startTCharmThread,initMsg,initMsg->stackSize);
55   }
56   CtvAccessOther(tid,_curTCharm)=this;
57   TCharm::setState(inInit);
58   isStopped=true;
59   threadInfo.tProxy=CProxy_TCharm(thisArrayID);
60   threadInfo.thisElement=thisIndex;
61   threadInfo.numElements=initMsg->numElements;
62   nUd=0;
63   usesAtSync=CmiTrue;
64   ready();
65 }
66
67 TCharm::TCharm(CkMigrateMessage *msg)
68         :ArrayElement1D(msg)
69 {
70   initMsg=NULL;
71   tid=NULL;
72   threadInfo.tProxy=CProxy_TCharm(thisArrayID);  
73 }
74
75 void TCharm::pup(PUP::er &p) {
76 //Pup superclass
77   ArrayElement1D::pup(p);  
78
79   p(isStopped);
80   p(threadInfo.thisElement);
81   p(threadInfo.numElements);
82
83 #ifndef CMK_OPTIMIZE
84   DBG("Packing thread");
85   if (!isStopped)
86     CkAbort("Cannot pup a running thread.  You must suspend before migrating.\n");
87   if (tcharm_nomig) CkAbort("Cannot migrate with the +tcharm_nomig option!\n");
88 #endif
89
90 //Pup thread (EVIL & UGLY):
91   //This seekBlock allows us to reorder the packing/unpacking--
92   // This is needed because the userData depends on the thread's stack
93   // both at pack and unpack time.
94   PUP::seekBlock s(p,2);
95   if (p.isUnpacking()) 
96   {//In this case, unpack the thread before the user data
97     s.seek(1);
98     tid = CthPup((pup_er) &p, tid);
99     CtvAccessOther(tid,_curTCharm)=this;
100   }
101   
102   //Pack all user data
103   TCharm::setState(inPup);
104   s.seek(0);
105   p(nUd);
106   for(int i=0;i<nUd;i++) 
107     ud[i].pup(p);
108   TCharm::setState(inFramework);
109
110   if (!p.isUnpacking()) 
111   {//In this case, pack the thread after the user data
112     s.seek(1);
113     tid = CthPup((pup_er) &p, tid);
114   }
115   s.endBlock(); //End of seeking block
116 }
117
118 //Pup one group of user data
119 void TCharm::UserData::pup(PUP::er &p)
120 {
121   pup_er pext=(pup_er)(&p);
122   p(isC);
123   //Save address of userdata-- assumes user data is on the stack
124   p((void*)&data,sizeof(data));
125   if (isC) { //C version
126     //FIXME: function pointers may not be valid across processors
127     p((void*)&cfn, sizeof(TCpupUserDataC));
128     cfn(pext,data);
129   } 
130   else { //Fortran version
131     //FIXME: function pointers may not be valid across processors
132     p((void*)&ffn, sizeof(TCpupUserDataF));        
133     ffn(pext,data);
134   }
135 }
136
137 TCharm::~TCharm() 
138 {
139   CthFree(tid);
140   delete initMsg;
141 }
142
143 //Register user data to be packed with the thread
144 int TCharm::add(const TCharm::UserData &d)
145 {
146   if (nUd>=maxUserData)
147     CkAbort("TCharm: Registered too many user data fields!\n");
148   int nu=nUd++;
149   ud[nu]=d;
150   return nu;
151 }
152 void *TCharm::lookupUserData(int i) {
153         if (i<0 || i>=nUd)
154                 CkAbort("Bad user data index passed to TCharmGetUserdata!\n");
155         return ud[i].getData();
156 }
157
158 //Start the thread running
159 void TCharm::run(void)
160 {
161   DBG("TCharm::run()");
162   start();
163 }
164
165 //Block the thread until start()ed again.
166 void TCharm::stop(void)
167 {
168   if (isStopped) return; //Nothing to do
169 #ifndef CMK_OPTIMIZE
170   DBG("suspending thread");
171   if (tid != CthSelf())
172     CkAbort("Called TCharm::stop from outside TCharm thread!\n");
173   if (tcharm_nothreads)
174     CkAbort("Cannot make blocking calls using +tcharm_nothreads!\n");
175 #endif
176   isStopped=true;
177   stopTiming();
178   TCharm::setState(inFramework);
179   CthSuspend();
180   TCharm::setState(inDriver);
181   /*We have to do the get() because "this" may have changed
182     during a migration-suspend.*/
183   TCharm::get()->startTiming();
184 }
185
186 //Resume the waiting thread
187 void TCharm::start(void)
188 {
189   if (!isStopped) return; //Already started
190   isStopped=false;
191   TCharm::setState(inDriver);
192   DBG("awakening thread");
193   if (tcharm_nothreads) /*Call user routine directly*/
194           startTCharmThread(initMsg);
195   else /*Jump to thread normally*/
196           CthAwaken(tid);
197 }
198
199 //Go to sync, block, possibly migrate, and then resume
200 void TCharm::migrate(void)
201 {
202 #if CMK_LBDB_ON
203   DBG("going to sync");  
204   AtSync();
205   stop();
206 #else
207   DBG("skipping sync, because there is no load balancer");
208 #endif
209 }
210
211 //Resume from sync: start the thread again
212 void TCharm::ResumeFromSync(void)
213 {
214   start();
215 }
216
217 #ifndef CMK_OPTIMIZE
218 //Make sure we're actually in driver
219 void TCharm::check(void)
220 {
221         if (getState()!=inDriver)
222                 CkAbort("TCharm> Can only use that routine from within driver!\n");
223 }
224 #endif
225
226 static int propMapCreated=0;
227 static CkGroupID propMapID;
228 CkGroupID CkCreatePropMap(void);
229
230 static void TCharmBuildThreads(TCharmInitMsg *msg,TCharmSetupCookie &cook)
231 {
232         CkArrayOptions opts(msg->numElements);
233         if (!propMapCreated) {
234                 propMapCreated=1;
235                 propMapID=CkCreatePropMap();
236         }
237         opts.setMap(propMapID);
238         int nElem=msg->numElements; //<- save it because msg will be deleted.
239         CkArrayID id=CProxy_TCharm::ckNew(msg,opts);
240         cook.setThreads(id,nElem);
241 }
242
243 /****** Readonlys *****/
244 CkVec<TCpupReadonlyGlobal> TCharmReadonlys::entries;
245 void TCharmReadonlys::add(TCpupReadonlyGlobal fn)
246 {
247         entries.push_back(fn);
248 }
249 //Pups all registered readonlys
250 void TCharmReadonlys::pup(PUP::er &p) {
251         if (p.isUnpacking()) {
252                 //HACK: Rather than sending this message only where its needed,
253                 // we send it everywhere and just ignore it if it's not needed.
254                 if (CkMyPe()==0) return; //Processor 0 is the source-- no unpacking needed
255                 if (CkMyRank()!=0) return; //Some other processor will do the unpacking
256         }
257         //Pup the globals for this node:
258         int i,n=entries.length();
259         p(n);
260         if (n!=entries.length())
261                 CkAbort("TCharmReadonly list length mismatch!\n");
262         for (i=0;i<n;i++)
263                 (entries[i])((pup_er)&p);
264 }
265
266 CDECL void TCharmReadonlyGlobals(TCpupReadonlyGlobal fn)
267 {
268         if (TCharm::getState()!=inNodeSetup)
269                 CkAbort("Can only call TCharmReadonlyGlobals from in TCharmUserNodeSetup!\n");
270         TCharmReadonlys::add(fn);
271 }
272 FDECL void FTN_NAME(TCHARM_READONLY_GLOBALS,tcharm_readonly_globals)
273         (TCpupReadonlyGlobal fn)
274 {
275         TCharmReadonlyGlobals(fn);
276 }
277
278 /************* Startup/Shutdown Coordination Support ************/
279
280 enum {TC_READY=23, TC_BARRIER=87, TC_DONE=42};
281
282 //Called when a client is ready to run
283 void TCharm::ready(void) {
284         DBG("TCharm thread "<<thisIndex<<" ready")
285         int vals[2]={0,1};
286         if (thisIndex==0) vals[0]=TC_READY;
287         //Contribute to a synchronizing reduction
288         contribute(sizeof(vals),&vals,CkReduction::sum_int);
289 }
290
291 //Called when we want to go to a barrier
292 void TCharm::barrier(void) {
293         int vals[2]={0,1};
294         if (thisIndex==0) vals[0]=TC_BARRIER;
295         //Contribute to a synchronizing reduction
296         contribute(sizeof(vals),&vals,CkReduction::sum_int);
297         stop();
298 }
299
300 //Called when the thread is done running
301 void TCharm::done(void) {
302         DBG("TCharm thread "<<thisIndex<<" done")
303         int vals[2]={0,1};
304         if (thisIndex==0) vals[0]=TC_DONE;
305         //Contribute to a synchronizing reduction
306         contribute(sizeof(vals),&vals,CkReduction::sum_int);
307         stop();
308 }
309
310 //Called when an array reduction is complete
311 static void coordinatorReduction(void *coord_,int dataLen,void *reductionData)
312 {
313         TCharmCoordinator *coord=(TCharmCoordinator *)coord_;
314         int *vals=(int *)reductionData;
315         if (dataLen!=2*sizeof(int))
316                 CkAbort("Unexpected length in TCharm array reduction!\n");
317         DBGX("Finished coordinator reduction: "<<vals[0]<<", "<<vals[1]);
318         switch (vals[0]) {
319         case TC_READY: coord->clientReady(); break;
320         case TC_BARRIER: coord->clientBarrier(); break;
321         case TC_DONE: coord->clientDone(); break;
322         default:
323                 CkAbort("Unexpected value from TCharm array reduction!\n");
324         };
325 }
326
327 int TCharmCoordinator::nArrays=0; //Total number of running thread arrays
328 TCharmCoordinator *TCharmCoordinator::head=NULL; //List of coordinators
329
330
331 TCharmCoordinator::TCharmCoordinator(CkArrayID threads_,int nThreads_)
332         :threads(threads_), nThreads(nThreads_), nClients(0), nReady(0)
333 {
334         nArrays++;
335         //Link into the coordinator list
336         next=head;
337         head=this;
338
339         threads.setReductionClient(coordinatorReduction,this);
340         nClients=1; //Thread array itself is a client
341 }
342 TCharmCoordinator::~TCharmCoordinator()
343 {
344         //Coordinators never get deleted
345 }
346 void TCharmCoordinator::addClient(const CkArrayID &client)
347 {
348         nClients++;
349 }
350 void TCharmCoordinator::clientReady(void)
351 {
352         DBGX("client "<<nReady+1<<" of "<<nClients<<" ready");
353         nReady++;
354         if (nReady>=nClients) { //All client arrays are ready-- start threads
355                 DBGX("starting threads");
356                 threads.run();
357         }
358 }
359 void TCharmCoordinator::clientBarrier(void)
360 {
361         DBGX("clients all at barrier");
362         threads.run();
363 }
364 void TCharmCoordinator::clientDone(void)
365 {
366         DBGX("clientDone");     
367         nArrays--;
368         if (nArrays<=0) { //All arrays have exited
369                 DBGX("done with computation");
370                 CkExit();
371         }
372 }
373
374 /************* Setup **************/
375
376 //Cookie used during setup
377 TCharmSetupCookie *TCharmSetupCookie::theCookie;
378
379 //Globals used to control setup process
380 static int g_numDefaultSetups=0;
381 static TCharmFallbackSetupFn g_fallbackSetup=NULL;
382 void TCharmSetFallbackSetup(TCharmFallbackSetupFn f)
383 {
384         g_fallbackSetup=f;
385 }
386 CDECL void TCharmInDefaultSetup(void) {
387         g_numDefaultSetups++;
388 }
389
390 //Tiny simple main chare
391 class TCharmMain : public Chare {
392 public:
393   TCharmMain(CkArgMsg *msg) {
394     if (0!=(tcharm_nomig=CmiGetArgFlag(msg->argv,"+tcharm_nomig")))
395         CmiPrintf("TCHARM> Disabling migration support, for debugging\n");
396     tcharm_nothreads=CmiGetArgFlag(msg->argv,"+tcharm_nothread");
397     tcharm_nothreads|=CmiGetArgFlag(msg->argv,"+tcharm_nothreads");
398     if (0!=tcharm_nothreads)
399        CmiPrintf("TCHARM> Disabling thread support, for debugging\n");
400     
401     TCharmSetupCookie cookie(msg->argv);
402     TCharmSetupCookie::theCookie=&cookie;
403     g_numDefaultSetups=0;
404     
405     /*Call user-overridable C setup*/
406     TCharmUserSetup();
407     /*Call user-overridable Fortran setup*/
408     FTN_NAME(TCHARM_USER_SETUP,tcharm_user_setup)();
409     
410     if (g_numDefaultSetups==2) 
411     { //User didn't override either setup routine
412             if (g_fallbackSetup)
413                     (g_fallbackSetup)();
414             else
415                     CmiAbort("You need to override TCharmUserSetup to start your computation, or else link in a framework module\n");
416     }       
417     
418     delete msg;
419     
420     if (0==TCharmCoordinator::getTotal())
421             CkAbort("You didn't create any TCharm arrays in TCharmUserSetup!\n");
422
423     //Send out the readonly globals:
424     TCharmReadonlys r;
425     CProxy_TCharmReadonlyGroup::ckNew(r);
426   }
427 };
428
429 #ifndef CMK_OPTIMIZE
430 /*The setup cookie, used to store global initialization state*/
431 TCharmSetupCookie &TCharmSetupCookie::check(void)
432 {
433         if (magic!=correctMagic)
434                 CkAbort("TCharm setup cookie is damaged!\n");
435         return *this;
436 }
437 #endif
438
439 void TCharmSetupCookie::setThreads(const CkArrayID &aid,int nel)
440 {
441         coord=new TCharmCoordinator(aid,nel);
442         tc=aid; numElements=nel;
443 }
444
445 TCharmSetupCookie::TCharmSetupCookie(char **argv_)
446 {
447         magic=correctMagic;
448         stackSize=0;
449         argv=argv_;
450         coord=NULL;
451 }
452
453
454 /************** User API ***************/
455
456 #define cookie (*TCharmSetupCookie::get())
457
458 /**********************************
459 Callable from UserSetup: 
460 */
461
462 /*Set the size of the thread stack*/
463 CDECL void TCharmSetStackSize(int newStackSize)
464 {
465         if (TCharm::getState()!=inInit)
466                 CkAbort("TCharm> Can only set stack size from in init!\n");
467         cookie.setStackSize(newStackSize);
468 }
469 FDECL void FTN_NAME(TCHARM_SET_STACK_SIZE,tcharm_set_stack_size)
470         (int *newSize)
471 { TCharmSetStackSize(*newSize); }
472
473
474 /*Create a new array of threads, which will be bound to by subsequent libraries*/
475 CDECL void TCharmCreate(int nThreads,
476                         TCharmThreadStartFn threadFn)
477 {
478         TCharmCreateData(nThreads,
479                          (TCharmThreadDataStartFn)threadFn,NULL,0);
480 }
481 FDECL void FTN_NAME(TCHARM_CREATE,tcharm_create)
482         (int *nThreads,TCharmThreadStartFn threadFn)
483 { TCharmCreate(*nThreads,threadFn); }
484
485
486 /*As above, but pass along (arbitrary) data to threads*/
487 CDECL void TCharmCreateData(int nThreads,
488                   TCharmThreadDataStartFn threadFn,
489                   void *threadData,int threadDataLen)
490 {
491         if (TCharm::getState()!=inInit)
492                 CkAbort("TCharm> Can only create threads from in init!\n");
493         TCharmSetupCookie &cook=cookie;
494         TCharmInitMsg *msg=new (threadDataLen,0) TCharmInitMsg(
495                 (CthVoidFn)threadFn,cook.getStackSize());
496         msg->numElements=nThreads;
497         memcpy(msg->data,threadData,threadDataLen);
498         TCharmBuildThreads(msg,cook);
499 }
500
501 FDECL void FTN_NAME(TCHARM_CREATE_DATA,tcharm_create_data)
502         (int *nThreads,
503                   TCharmThreadDataStartFn threadFn,
504                   void *threadData,int *threadDataLen)
505 { TCharmCreateData(*nThreads,threadFn,threadData,*threadDataLen); }
506
507
508 /*Get the unconsumed command-line arguments*/
509 CDECL char **TCharmArgv(void)
510 {
511         if (TCharm::getState()!=inInit)
512                 CkAbort("TCharm> Can only get arguments from in init!\n");
513         return cookie.getArgv();
514 }
515 CDECL int TCharmArgc(void)
516 {
517         if (TCharm::getState()!=inInit)
518                 CkAbort("TCharm> Can only get arguments from in init!\n");
519         return CmiGetArgc(cookie.getArgv());
520 }
521
522 CDECL int TCharmGetNumChunks(void)
523 {
524         int nChunks=CkNumPes();
525         char **argv=TCharmArgv();
526         CmiGetArgInt(argv,"-vp",&nChunks);
527         CmiGetArgInt(argv,"+vp",&nChunks);
528         lastNumChunks=nChunks;
529         return nChunks;
530 }
531 FDECL int FTN_NAME(TCHARM_GET_NUM_CHUNKS,tcharm_get_num_chunks)(void)
532 {
533         return TCharmGetNumChunks();
534 }
535
536
537 /***********************************
538 Callable from worker thread
539 */
540 CDECL int TCharmElement(void)
541 { return TCharm::get()->getElement();}
542 CDECL int TCharmNumElements(void)
543
544         if (TCharm::getState()==inDriver)
545                 return TCharm::get()->getNumElements();
546         else
547                 return lastNumChunks;
548 }
549
550 FDECL int FTN_NAME(TCHARM_ELEMENT,tcharm_element)(void) 
551 { return TCharmElement();}
552 FDECL int FTN_NAME(TCHARM_NUM_ELEMENTS,tcharm_num_elements)(void) 
553 { return TCharmNumElements();}
554
555 //Make sure this address will migrate with us when we move:
556 static void checkAddress(void *data)
557 {
558         if (tcharm_nomig||tcharm_nothreads) return; //Stack is not isomalloc'd
559         if (!CmiIsomallocInRange(data))
560             CkAbort("The UserData you register must be allocated on the stack!\n");
561 }
562
563 CDECL int TCharmRegister(void *data,TCharmPupFn pfn)
564
565         checkAddress(data);
566         return TCharm::get()->add(TCharm::UserData(pfn,data));
567 }
568 FDECL int FTN_NAME(TCHARM_REGISTER,tcharm_register)
569         (void *data,TCpupUserDataF pfn)
570
571         checkAddress(data);
572         return TCharm::get()->add(TCharm::UserData(
573                 pfn,data,TCharm::UserData::isFortran()));
574 }
575
576 CDECL void *TCharmGetUserdata(int id)
577 {
578         return TCharm::get()->lookupUserData(id);
579 }
580 FDECL void *FTN_NAME(TCHARM_GET_USERDATA,tcharm_get_userdata)(int *id)
581 { return TCharmGetUserdata(*id); }
582
583 CDECL void TCharmMigrate(void)
584 {
585         TCharm::get()->migrate();
586 }
587 FDECL void FTN_NAME(TCHARM_MIGRATE,tcharm_migrate)(void)
588 {
589         TCharm::get()->migrate();
590 }
591
592 CDECL void TCharmBarrier(void)
593 {
594         TCharm::get()->barrier();
595 }
596 FDECL void FTN_NAME(TCHARM_BARRIER,tcharm_barrier)(void)
597 {
598         TCharmBarrier();
599 }
600
601 CDECL void TCharmDone(void)
602 {
603         TCharm::get()->done();
604 }
605 FDECL void FTN_NAME(TCHARM_DONE,tcharm_done)(void)
606 {
607         TCharmDone();
608 }
609
610
611
612 #include "tcharm.def.h"