d9d112e803a7fad57b3f312e6158def462b2f261
[charm.git] / src / conv-core / convcore.c
1 /*****************************************************************************
2  * $Source$
3  * $Author$
4  * $Date$
5  * $Revision$
6  *****************************************************************************/
7 /** 
8   @defgroup Converse
9   \brief Converse--a parallel portability layer.
10
11   Converse is the lowest level inside the Charm++ hierarchy. It stands on top
12   of the machine layer, and it provides all the common functionality across
13   platforms.
14
15   One converse program is running on every processor (or node in the smp
16   version). it manages the message transmission, and the memory allocation.
17   Charm++, which is on top of Converse, uses its functionality for
18   interprocess *communication.
19
20   In order to maintain multiple independent objects inside a single user space
21   program, it uses a personalized version of threads, which can be executed,
22   suspended, and migrated across processors.
23
24   It provides a scheduler for message delivery: methods can be registered to
25   the scheduler, and then messages allocated through CmiAlloc can be sent to
26   the correspondent method in a remote processor. This is done through the
27   converse header (which has few common fields, but is architecture dependent).
28
29   @defgroup ConverseScheduler 
30   \brief The portion of Converse responsible for scheduling the execution of 
31   incoming messages.
32   
33   Converse provides a scheduler for message delivery: methods can be registered to 
34   the scheduler, and then messages allocated through CmiAlloc can be sent to 
35   the correspondent method in a remote processor. This is done through the
36   converse header (which has few common fields, but is architecture dependent).
37
38   In converse the CsdScheduleForever() routine will run an infinite while loop that 
39   looks for available messages to processes from the message queue. Messages in the 
40   queue are dequeued by the CsdNextMessage() routine. When a 
41   message is taken from the queue it is then passed into CmiHandleMessage() which
42   calls the handler associated with the message.
43
44   Incoming messages that are destined for Charm++ will be passed to the 
45   \ref CharmScheduler "charm scheduling routines".
46
47   @file
48   converse main core
49   @ingroup Converse
50   @ingroup ConverseScheduler
51  
52   @addtogroup Converse
53   @{
54
55 */
56
57 #include <stdio.h>
58 #include <stdlib.h>
59 #include <string.h>
60 #include <errno.h>
61 #ifndef _WIN32
62 #include <sys/time.h>
63 #include <sys/resource.h>
64 #endif
65
66 #include "converse.h"
67 #include "conv-trace.h"
68 #include "sockRoutines.h"
69 #include "queueing.h"
70 #include "conv-ccs.h"
71 #include "ccs-server.h"
72 #include "memory-isomalloc.h"
73 #include "converseEvents.h"             /* projector */
74 #include "traceCoreCommon.h"    /* projector */
75 #include "machineEvents.h"     /* projector */
76
77 #if CMK_OUT_OF_CORE
78 #include "conv-ooc.h"
79 #endif
80
81 #if CONVERSE_POOL
82 #include "cmipool.h"
83 #endif
84
85 #if CMK_CONDS_USE_SPECIAL_CODE
86 CmiSwitchToPEFnPtr CmiSwitchToPE;
87 #endif
88
89 CpvExtern(int, _traceCoreOn);   /* projector */
90 extern void CcdModuleInit(char **);
91 extern void CmiMemoryInit(char **);
92 extern void CldModuleInit(char **);
93
94 #if CMK_WHEN_PROCESSOR_IDLE_USLEEP
95 #include <sys/types.h>
96 #include <sys/time.h>
97 #endif
98
99 #if CMK_TIMER_USE_TIMES
100 #include <sys/times.h>
101 #include <limits.h>
102 #include <unistd.h>
103 #endif
104
105 #if CMK_TIMER_USE_GETRUSAGE
106 #include <sys/time.h>
107 #include <sys/resource.h>
108 #endif
109
110 #if CMK_TIMER_USE_RDTSC
111 #include <string.h>
112 #include <unistd.h>
113 #include <time.h>
114 #include <stdio.h>
115 #include <sys/time.h>
116 #include <sys/resource.h>
117 #endif
118
119 #ifdef CMK_TIMER_USE_WIN32API
120 #include <stdlib.h>
121 #include <time.h>
122 #include <sys/types.h>
123 #include <sys/timeb.h>
124 #endif
125
126 #ifdef CMK_CUDA
127 #include "cuda-hybrid-api.h"
128 #endif
129
130 #include "quiescence.h"
131
132 int cur_restart_phase = 1;      /* checkpointing/restarting phase counter */
133
134 static int CsdLocalMax = CSD_LOCAL_MAX_DEFAULT;
135
136 CpvStaticDeclare(int, CmiMainHandlerIDP); /* Main handler for _CmiMultipleSend that is run on every node */
137
138 /*****************************************************************************
139  *
140  * Unix Stub Functions
141  *
142  ****************************************************************************/
143
144 #ifdef MEMMONITOR
145 typedef unsigned long mmulong;
146 CpvDeclare(mmulong,MemoryUsage);
147 CpvDeclare(mmulong,HiWaterMark);
148 CpvDeclare(mmulong,ReportedHiWaterMark);
149 CpvDeclare(int,AllocCount);
150 CpvDeclare(int,BlocksAllocated);
151 #endif
152
153 #define MAX_HANDLERS 512
154
155 #if ! CMK_CMIPRINTF_IS_A_BUILTIN
156 CpvDeclare(int,expIOFlushFlag);
157 #if CMI_IO_BUFFER_EXPLICIT
158 /* 250k not too large depending on how slow terminal IO is */
159 #define DEFAULT_IO_BUFFER_SIZE 250000
160 CpvDeclare(char*,explicitIOBuffer);
161 CpvDeclare(int,expIOBufferSize);
162 #endif
163 #endif
164
165 #if CMK_NODE_QUEUE_AVAILABLE
166 void  *CmiGetNonLocalNodeQ();
167 #endif
168
169 CpvDeclare(void*, CsdSchedQueue);
170
171 #if CMK_OUT_OF_CORE
172 /* The Queue where the Prefetch Thread puts the messages from CsdSchedQueue  */
173 CpvDeclare(void*, CsdPrefetchQueue);
174 pthread_mutex_t prefetchLock;
175 #endif
176
177 #if CMK_NODE_QUEUE_AVAILABLE
178 CsvDeclare(void*, CsdNodeQueue);
179 CsvDeclare(CmiNodeLock, CsdNodeQueueLock);
180 #endif
181 CpvDeclare(int,   CsdStopFlag);
182 CpvDeclare(int,   CsdLocalCounter);
183
184 CmiNodeLock smp_mutex;               /* for smp */
185
186 #if CONVERSE_VERSION_VMI
187 void *CMI_VMI_CmiAlloc (int size);
188 void CMI_VMI_CmiFree (void *ptr);
189 #endif
190
191 #if CONVERSE_VERSION_ELAN
192 void* elan_CmiAlloc(int size);
193 #endif
194
195 #if CMK_USE_IBVERBS | CMK_USE_IBUD
196 void *infi_CmiAlloc(int size);
197 void infi_CmiFree(void *ptr);
198 void infi_freeMultipleSend(void *ptr);
199 void infi_unregAndFreeMeta(void *ch);
200 #endif
201
202
203 #if CMK_GRID_QUEUE_AVAILABLE
204 CpvDeclare(void *, CkGridObject);
205 CpvDeclare(void *, CsdGridQueue);
206 #endif
207
208
209 /*****************************************************************************
210  *
211  * Command-Line Argument (CLA) parsing routines.
212  *
213  *****************************************************************************/
214
215 static int usageChecked=0; /* set when argv has been searched for a usage request */
216 static int printUsage=0; /* if set, print command-line usage information */
217 static const char *CLAformatString="%20s %10s %s\n";
218
219 /** This little list of CLA's holds the argument descriptions until it's
220    safe to print them--it's needed because the net- versions don't have 
221    printf until they're pretty well started.
222  */
223 typedef struct {
224         const char *arg; /* Flag name, like "-foo"*/
225         const char *param; /* Argument's parameter type, like "integer" or "none"*/
226         const char *desc; /* Human-readable description of what it does */
227 } CLA;
228 static int CLAlistLen=0;
229 static int CLAlistMax=0;
230 static CLA *CLAlist=NULL;
231
232 /** Add this CLA */
233 static void CmiAddCLA(const char *arg,const char *param,const char *desc) {
234         int i;
235         if (CmiMyPe()!=0) return; /*Don't bother if we're not PE 0*/
236         if (desc==NULL) return; /*It's an internal argument*/
237         if (usageChecked) { /* Printf should work now */
238                 if (printUsage)
239                         CmiPrintf(CLAformatString,arg,param,desc);
240         }
241         else { /* Printf doesn't work yet-- just add to the list.
242                 This assumes the const char *'s are static references,
243                 which is probably reasonable. */
244                 i=CLAlistLen++;
245                 if (CLAlistLen>CLAlistMax) { /*Grow the CLA list */
246                         CLAlistMax=16+2*CLAlistLen;
247                         CLAlist=realloc(CLAlist,sizeof(CLA)*CLAlistMax);\
248                 }
249                 CLAlist[i].arg=arg;
250                 CLAlist[i].param=param;
251                 CLAlist[i].desc=desc;
252         }
253 }
254
255 /** Print out the stored list of CLA's */
256 static void CmiPrintCLAs(void) {
257         int i;
258         if (CmiMyPe()!=0) return; /*Don't bother if we're not PE 0*/
259         CmiPrintf("Converse Machine Command-line Parameters:\n ");
260         CmiPrintf(CLAformatString,"Option:","Parameter:","Description:");
261         for (i=0;i<CLAlistLen;i++) {
262                 CLA *c=&CLAlist[i];
263                 CmiPrintf(CLAformatString,c->arg,c->param,c->desc);
264         }
265 }
266
267 /**
268  * Determines if command-line usage information should be printed--
269  * that is, if a "-?", "-h", or "--help" flag is present.
270  * Must be called after printf is setup.
271  */
272 void CmiArgInit(char **argv) {
273         int i;
274         for (i=0;argv[i]!=NULL;i++)
275         {
276                 if (0==strcmp(argv[i],"-?") ||
277                     0==strcmp(argv[i],"-h") ||
278                     0==strcmp(argv[i],"--help")) 
279                 {
280                         printUsage=1;
281                         /* Don't delete arg:  CmiDeleteArgs(&argv[i],1);
282                           Leave it there for user program to see... */
283                         CmiPrintCLAs();
284                 }
285         }
286         if (CmiMyPe()==0) { /* Throw away list of stored CLA's */
287                 CLAlistLen=CLAlistMax=0;
288                 free(CLAlist); CLAlist=NULL;
289         }
290         usageChecked=1;
291 }
292
293 /** Return 1 if we're currently printing command-line usage information. */
294 int CmiArgGivingUsage(void) {
295         return (CmiMyPe()==0) && printUsage;
296 }
297
298 /** Identifies the module that accepts the following command-line parameters */
299 void CmiArgGroup(const char *parentName,const char *groupName) {
300         if (CmiArgGivingUsage()) {
301                 if (groupName==NULL) groupName=parentName; /* Start of a new group */
302                 CmiPrintf("\n%s Command-line Parameters:\n",groupName);
303         }
304 }
305
306 /** Count the number of non-NULL arguments in list*/
307 int CmiGetArgc(char **argv)
308 {
309         int i=0,argc=0;
310         if (argv)
311         while (argv[i++]!=NULL)
312                 argc++;
313         return argc;
314 }
315
316 /** Return a new, heap-allocated copy of the argv array*/
317 char **CmiCopyArgs(char **argv)
318 {
319         int argc=CmiGetArgc(argv);
320         char **ret=(char **)malloc(sizeof(char *)*(argc+1));
321         int i;
322         for (i=0;i<=argc;i++)
323                 ret[i]=argv[i];
324         return ret;
325 }
326
327 /** Delete the first k argument from the given list, shifting
328 all other arguments down by k spaces.
329 e.g., argv=={"a","b","c","d",NULL}, k==3 modifies
330 argv={"d",NULL,"c","d",NULL}
331 */
332 void CmiDeleteArgs(char **argv,int k)
333 {
334         int i=0;
335         while ((argv[i]=argv[i+k])!=NULL)
336                 i++;
337 }
338
339 /** Find the given argment and string option in argv.
340 If the argument is present, set the string option and
341 delete both from argv.  If not present, return NULL.
342 e.g., arg=="-name" returns "bob" from
343 argv=={"a.out","foo","-name","bob","bar"},
344 and sets argv={"a.out","foo","bar"};
345 */
346 int CmiGetArgStringDesc(char **argv,const char *arg,char **optDest,const char *desc)
347 {
348         int i;
349         CmiAddCLA(arg,"string",desc);
350         for (i=0;argv[i]!=NULL;i++)
351                 if (0==strcmp(argv[i],arg))
352                 {/*We found the argument*/
353                         if (argv[i+1]==NULL) CmiAbort("Argument not complete!");
354                         *optDest=argv[i+1];
355                         CmiDeleteArgs(&argv[i],2);
356                         return 1;
357                 }
358         return 0;/*Didn't find the argument*/
359 }
360 int CmiGetArgString(char **argv,const char *arg,char **optDest) {
361         return CmiGetArgStringDesc(argv,arg,optDest,"");
362 }
363
364 /** Find the given argument and floating-point option in argv.
365 Remove it and return 1; or return 0.
366 */
367 int CmiGetArgDoubleDesc(char **argv,const char *arg,double *optDest,const char *desc) {
368         char *number=NULL;
369         CmiAddCLA(arg,"number",desc);
370         if (!CmiGetArgStringDesc(argv,arg,&number,NULL)) return 0;
371         if (1!=sscanf(number,"%lg",optDest)) return 0;
372         return 1;
373 }
374 int CmiGetArgDouble(char **argv,const char *arg,double *optDest) {
375         return CmiGetArgDoubleDesc(argv,arg,optDest,"");
376 }
377
378 /** Find the given argument and integer option in argv.
379 If the argument is present, parse and set the numeric option,
380 delete both from argv, and return 1. If not present, return 0.
381 e.g., arg=="-pack" matches argv=={...,"-pack","27",...},
382 argv=={...,"-pack0xf8",...}, and argv=={...,"-pack=0777",...};
383 but not argv=={...,"-packsize",...}.
384 */
385 int CmiGetArgIntDesc(char **argv,const char *arg,int *optDest,const char *desc)
386 {
387         int i;
388         int argLen=strlen(arg);
389         CmiAddCLA(arg,"integer",desc);
390         for (i=0;argv[i]!=NULL;i++)
391                 if (0==strncmp(argv[i],arg,argLen))
392                 {/*We *may* have found the argument*/
393                         const char *opt=NULL;
394                         int nDel=0;
395                         switch(argv[i][argLen]) {
396                         case 0: /* like "-p","27" */
397                                 opt=argv[i+1]; nDel=2; break;
398                         case '=': /* like "-p=27" */
399                                 opt=&argv[i][argLen+1]; nDel=1; break;
400                         case '-':case '+':
401                         case '0':case '1':case '2':case '3':case '4':
402                         case '5':case '6':case '7':case '8':case '9':
403                                 /* like "-p27" */
404                                 opt=&argv[i][argLen]; nDel=1; break;
405                         default:
406                                 continue; /*False alarm-- skip it*/
407                         }
408                         if (opt==NULL) continue; /*False alarm*/
409                         if (sscanf(opt,"%i",optDest)<1) {
410                         /*Bad command line argument-- die*/
411                                 fprintf(stderr,"Cannot parse %s option '%s' "
412                                         "as an integer.\n",arg,opt);
413                                 CmiAbort("Bad command-line argument\n");
414                         }
415                         CmiDeleteArgs(&argv[i],nDel);
416                         return 1;
417                 }
418         return 0;/*Didn't find the argument-- dest is unchanged*/       
419 }
420 int CmiGetArgInt(char **argv,const char *arg,int *optDest) {
421         return CmiGetArgIntDesc(argv,arg,optDest,"");
422 }
423
424 /** Find the given argument in argv.  If present, delete
425 it and return 1; if not present, return 0.
426 e.g., arg=="-foo" matches argv=={...,"-foo",...} but not
427 argv={...,"-foobar",...}.
428 */
429 int CmiGetArgFlagDesc(char **argv,const char *arg,const char *desc)
430 {
431         int i;
432         CmiAddCLA(arg,"",desc);
433         for (i=0;argv[i]!=NULL;i++)
434                 if (0==strcmp(argv[i],arg))
435                 {/*We found the argument*/
436                         CmiDeleteArgs(&argv[i],1);
437                         return 1;
438                 }
439         return 0;/*Didn't find the argument*/
440 }
441 int CmiGetArgFlag(char **argv,const char *arg) {
442         return CmiGetArgFlagDesc(argv,arg,"");
443 }
444
445
446 /*****************************************************************************
447  *
448  * Stack tracing routines.
449  *
450  *****************************************************************************/
451 #include "cmibacktrace.c"
452
453 /*
454 Convert "X(Y) Z" to "Y Z"-- remove text prior to first '(', and supress
455 the next parenthesis.  Operates in-place on the character data.
456 or Convert X(Y) to "Y" only, when trimname=1
457 */
458 static char *_implTrimParenthesis(char *str, int trimname) {
459   char *lParen=str, *ret=NULL, *rParen=NULL;
460   while (*lParen!='(') {
461     if (*lParen==0) return str; /* No left parenthesis at all. */
462     lParen++;
463   }
464   /* now *lParen=='(', so trim it*/
465   ret=lParen+1;
466   rParen=ret;
467   while (*rParen!=')') {
468     if (*rParen==0) return ret; /* No right parenthesis at all. */
469     rParen++;
470   }
471   /* now *rParen==')', so trim it*/
472   *rParen=trimname?0:' ';
473   return ret;  
474 }
475
476 /*
477 Return the text description of this trimmed routine name, if 
478 it's a system-generated routine where we should stop printing. 
479 This is probably overkill, but improves the appearance of callbacks.
480 */
481 static const char* _implGetBacktraceSys(const char *name) {
482   if (0==strncmp(name,"_call",5)) 
483   { /*it might be something we're interested in*/
484     if (0==strncmp(name,"_call_",6)) return "Call Entry Method";
485     if (0==strncmp(name,"_callthr_",9)) return "Call Threaded Entry Method";
486   }
487   if (0==strncmp(name,"CthResume",9)) return "Resumed thread";
488   if (0==strncmp(name,"qt_args",7)) return "Converse thread";
489   
490   return 0; /*ordinary user routine-- just print normally*/
491 }
492
493 /** Print out the names of these function pointers. */
494 void CmiBacktracePrint(void **retPtrs,int nLevels) {
495   if (nLevels>0) {
496     int i;
497     char **names=CmiBacktraceLookup(retPtrs,nLevels);
498     if (names==NULL) return;
499     CmiPrintf("[%d] Stack Traceback:\n", CmiMyPe());
500     for (i=0;i<nLevels;i++) {
501       if (names[i] == NULL) continue;
502       {
503       const char *trimmed=_implTrimParenthesis(names[i], 0);
504       const char *print=trimmed;
505       const char *sys=_implGetBacktraceSys(print);
506       if (sys) {
507           CmiPrintf("  [%d] Charm++ Runtime: %s (%s)\n",i,sys,print);
508           break; /*Stop when we hit Charm++ runtime.*/
509       } else {
510           CmiPrintf("  [%d:%d] %s\n",CmiMyPe(),i,print);
511       }
512      }
513     }
514     free(names);
515   }
516 }
517
518 /* Print (to stdout) the names of the functions that have been 
519    called up to this point. nSkip is the number of routines on the
520    top of the stack to *not* print out. */
521 void CmiPrintStackTrace(int nSkip) {
522 #if CMK_USE_BACKTRACE
523         int nLevels=max_stack;
524         void *stackPtrs[max_stack];
525         CmiBacktraceRecord(stackPtrs,1+nSkip,&nLevels);
526         CmiBacktracePrint(stackPtrs,nLevels);
527 #endif
528 }
529
530 int CmiIsFortranLibraryCall() {
531 #if CMK_USE_BACKTRACE
532   int ret = 0;
533   int nLevels=9;
534   void *stackPtrs[18];
535   CmiBacktraceRecord(stackPtrs,1,&nLevels);
536   if (nLevels>0) {
537     int i;
538     char **names=CmiBacktraceLookup(stackPtrs,nLevels);
539     if (names==NULL) return 0;
540     for (i=0;i<nLevels;i++) {
541       if (names[i] == NULL) continue;
542       const char *trimmed=_implTrimParenthesis(names[i], 1);
543       if (strncmp(trimmed, "for__", 5) == 0                /* ifort */
544           || strncmp(trimmed, "_xlf", 4) == 0               /* xlf90 */
545           || strncmp(trimmed, "_xlfBeginIO", 11) == 0 
546          )
547           {  /* CmiPrintf("[%d] NAME:%s\n", CmiMyPe(), trimmed); */
548              ret = 1; break; }
549     }
550     free(names);
551   }
552   return ret;
553 #else
554   return 0;
555 #endif
556 }
557
558 /*****************************************************************************
559  *
560  * Statistics: currently, the following statistics are not updated by converse.
561  *
562  *****************************************************************************/
563
564 CpvDeclare(int, CstatsMaxChareQueueLength);
565 CpvDeclare(int, CstatsMaxForChareQueueLength);
566 CpvDeclare(int, CstatsMaxFixedChareQueueLength);
567 CpvStaticDeclare(int, CstatPrintQueueStatsFlag);
568 CpvStaticDeclare(int, CstatPrintMemStatsFlag);
569
570 void CstatsInit(argv)
571 char **argv;
572 {
573
574 #ifdef MEMMONITOR
575   CpvInitialize(mmulong,MemoryUsage);
576   CpvAccess(MemoryUsage) = 0;
577   CpvInitialize(mmulong,HiWaterMark);
578   CpvAccess(HiWaterMark) = 0;
579   CpvInitialize(mmulong,ReportedHiWaterMark);
580   CpvAccess(ReportedHiWaterMark) = 0;
581   CpvInitialize(int,AllocCount);
582   CpvAccess(AllocCount) = 0;
583   CpvInitialize(int,BlocksAllocated);
584   CpvAccess(BlocksAllocated) = 0;
585 #endif
586
587   CpvInitialize(int, CstatsMaxChareQueueLength);
588   CpvInitialize(int, CstatsMaxForChareQueueLength);
589   CpvInitialize(int, CstatsMaxFixedChareQueueLength);
590   CpvInitialize(int, CstatPrintQueueStatsFlag);
591   CpvInitialize(int, CstatPrintMemStatsFlag);
592
593   CpvAccess(CstatsMaxChareQueueLength) = 0;
594   CpvAccess(CstatsMaxForChareQueueLength) = 0;
595   CpvAccess(CstatsMaxFixedChareQueueLength) = 0;
596   CpvAccess(CstatPrintQueueStatsFlag) = 0;
597   CpvAccess(CstatPrintMemStatsFlag) = 0;
598
599 #if 0
600   if (CmiGetArgFlagDesc(argv,"+mems", "Print memory statistics at shutdown"))
601     CpvAccess(CstatPrintMemStatsFlag)=1;
602   if (CmiGetArgFlagDesc(argv,"+qs", "Print queue statistics at shutdown"))
603     CpvAccess(CstatPrintQueueStatsFlag)=1;
604 #endif
605 }
606
607 int CstatMemory(i)
608 int i;
609 {
610   return 0;
611 }
612
613 int CstatPrintQueueStats()
614 {
615   return CpvAccess(CstatPrintQueueStatsFlag);
616 }
617
618 int CstatPrintMemStats()
619 {
620   return CpvAccess(CstatPrintMemStatsFlag);
621 }
622
623 /*****************************************************************************
624  *
625  * Cmi handler registration
626  *
627  *****************************************************************************/
628
629 CpvDeclare(CmiHandlerInfo*, CmiHandlerTable);
630 CpvStaticDeclare(int  , CmiHandlerCount);
631 CpvStaticDeclare(int  , CmiHandlerLocal);
632 CpvStaticDeclare(int  , CmiHandlerGlobal);
633 CpvDeclare(int,         CmiHandlerMax);
634
635 static void CmiExtendHandlerTable(int atLeastLen) {
636     int max = CpvAccess(CmiHandlerMax);
637     int newmax = (atLeastLen+(atLeastLen>>2)+32);
638     int bytes = max*sizeof(CmiHandlerInfo);
639     int newbytes = newmax*sizeof(CmiHandlerInfo);
640     CmiHandlerInfo *nu = (CmiHandlerInfo*)malloc(newbytes);
641     CmiHandlerInfo *tab = CpvAccess(CmiHandlerTable);
642     _MEMCHECK(nu);
643     memcpy(nu, tab, bytes);
644     memset(((char *)nu)+bytes, 0, (newbytes-bytes));
645     free(tab); tab=nu;
646     CpvAccess(CmiHandlerTable) = tab;
647     CpvAccess(CmiHandlerMax) = newmax;
648 }
649
650 void CmiNumberHandler(int n, CmiHandler h)
651 {
652   CmiHandlerInfo *tab;
653   if (n >= CpvAccess(CmiHandlerMax)) CmiExtendHandlerTable(n);
654   tab = CpvAccess(CmiHandlerTable);
655   tab[n].hdlr = (CmiHandlerEx)h; /* LIE!  This assumes extra pointer will be ignored!*/
656   tab[n].userPtr = 0;
657 }
658 void CmiNumberHandlerEx(int n, CmiHandlerEx h,void *userPtr) {
659   CmiHandlerInfo *tab;
660   if (n >= CpvAccess(CmiHandlerMax)) CmiExtendHandlerTable(n);
661   tab = CpvAccess(CmiHandlerTable);
662   tab[n].hdlr = h;
663   tab[n].userPtr=userPtr;
664 }
665
666 #if CMI_LOCAL_GLOBAL_AVAILABLE /*Leave room for local and global handlers*/
667 #  define DIST_BETWEEN_HANDLERS 3
668 #else /*No local or global handlers; ordinary handlers are back-to-back*/
669 #  define DIST_BETWEEN_HANDLERS 1
670 #endif
671
672 int CmiRegisterHandler(CmiHandler h)
673 {
674   int Count = CpvAccess(CmiHandlerCount);
675   CmiNumberHandler(Count, h);
676   CpvAccess(CmiHandlerCount) = Count+DIST_BETWEEN_HANDLERS;
677   return Count;
678 }
679 int CmiRegisterHandlerEx(CmiHandlerEx h,void *userPtr)
680 {
681   int Count = CpvAccess(CmiHandlerCount);
682   CmiNumberHandlerEx(Count, h, userPtr);
683   CpvAccess(CmiHandlerCount) = Count+DIST_BETWEEN_HANDLERS;
684   return Count;
685 }
686
687 #if CMI_LOCAL_GLOBAL_AVAILABLE
688 int CmiRegisterHandlerLocal(h)
689 CmiHandler h;
690 {
691   int Local = CpvAccess(CmiHandlerLocal);
692   CmiNumberHandler(Local, h);
693   CpvAccess(CmiHandlerLocal) = Local+3;
694   return Local;
695 }
696
697 int CmiRegisterHandlerGlobal(h)
698 CmiHandler h;
699 {
700   int Global = CpvAccess(CmiHandlerGlobal);
701   if (CmiMyPe()!=0) 
702     CmiError("CmiRegisterHandlerGlobal must only be called on PE 0.\n");
703   CmiNumberHandler(Global, h);
704   CpvAccess(CmiHandlerGlobal) = Global+3;
705   return Global;
706 }
707 #endif
708
709 static void _cmiZeroHandler(void *msg) {
710         CmiAbort("Converse zero handler executed-- was a message corrupted?\n");
711 }
712
713 static void CmiHandlerInit()
714 {
715   CpvInitialize(CmiHandlerInfo *, CmiHandlerTable);
716   CpvInitialize(int         , CmiHandlerCount);
717   CpvInitialize(int         , CmiHandlerLocal);
718   CpvInitialize(int         , CmiHandlerGlobal);
719   CpvInitialize(int         , CmiHandlerMax);
720   CpvAccess(CmiHandlerCount)  = 0;
721   CpvAccess(CmiHandlerLocal)  = 1;
722   CpvAccess(CmiHandlerGlobal) = 2;
723   CpvAccess(CmiHandlerMax) = 0; /* Table will be extended on the first registration*/
724   CpvAccess(CmiHandlerTable) = NULL;
725   CmiRegisterHandler((CmiHandler)_cmiZeroHandler);
726 }
727
728
729 /******************************************************************************
730  *
731  * CmiTimer
732  *
733  * Here are two possible implementations of CmiTimer.  Some machines don't
734  * select either, and define the timer in machine.c instead.
735  *
736  *****************************************************************************/
737
738 #if CMK_TIMER_USE_TIMES
739
740 CpvStaticDeclare(double, clocktick);
741 CpvStaticDeclare(int,inittime_wallclock);
742 CpvStaticDeclare(int,inittime_virtual);
743
744 int CmiTimerIsSynchronized()
745 {
746   return 0;
747 }
748
749 void CmiTimerInit()
750 {
751   struct tms temp;
752   CpvInitialize(double, clocktick);
753   CpvInitialize(int, inittime_wallclock);
754   CpvInitialize(int, inittime_virtual);
755   CpvAccess(inittime_wallclock) = times(&temp);
756   CpvAccess(inittime_virtual) = temp.tms_utime + temp.tms_stime;
757   CpvAccess(clocktick) = 1.0 / (sysconf(_SC_CLK_TCK));
758 }
759
760 double CmiWallTimer()
761 {
762   struct tms temp;
763   double currenttime;
764   int now;
765
766   now = times(&temp);
767   currenttime = (now - CpvAccess(inittime_wallclock)) * CpvAccess(clocktick);
768   return (currenttime);
769 }
770
771 double CmiCpuTimer()
772 {
773   struct tms temp;
774   double currenttime;
775   int now;
776
777   times(&temp);
778   now = temp.tms_stime + temp.tms_utime;
779   currenttime = (now - CpvAccess(inittime_virtual)) * CpvAccess(clocktick);
780   return (currenttime);
781 }
782
783 double CmiTimer()
784 {
785   return CmiCpuTimer();
786 }
787
788 #endif
789
790 #if CMK_TIMER_USE_GETRUSAGE
791
792 static double inittime_wallclock;
793 CpvStaticDeclare(double, inittime_virtual);
794
795 int CmiTimerIsSynchronized()
796 {
797   return 0;
798 }
799
800 void CmiTimerInit()
801 {
802   struct timeval tv;
803   struct rusage ru;
804   CpvInitialize(double, inittime_virtual);
805
806   /* try to synchronize calling barrier */
807   CmiBarrier();
808   CmiBarrier();
809   CmiBarrier();
810
811   gettimeofday(&tv,0);
812   inittime_wallclock = (tv.tv_sec * 1.0) + (tv.tv_usec*0.000001);
813   getrusage(0, &ru); 
814   CpvAccess(inittime_virtual) =
815     (ru.ru_utime.tv_sec * 1.0)+(ru.ru_utime.tv_usec * 0.000001) +
816     (ru.ru_stime.tv_sec * 1.0)+(ru.ru_stime.tv_usec * 0.000001);
817
818   CmiBarrier();
819 /*  CmiBarrierZero(); */
820 }
821
822 double CmiCpuTimer()
823 {
824   struct rusage ru;
825   double currenttime;
826
827   getrusage(0, &ru);
828   currenttime =
829     (ru.ru_utime.tv_sec * 1.0)+(ru.ru_utime.tv_usec * 0.000001) +
830     (ru.ru_stime.tv_sec * 1.0)+(ru.ru_stime.tv_usec * 0.000001);
831   return currenttime - CpvAccess(inittime_virtual);
832 }
833
834 static double lastT = -1.0;
835
836 double CmiWallTimer()
837 {
838   struct timeval tv;
839   double currenttime;
840
841   gettimeofday(&tv,0);
842   currenttime = (tv.tv_sec * 1.0) + (tv.tv_usec * 0.000001);
843 #ifndef CMK_OPTIMIZE
844   if (lastT > 0.0 && currenttime < lastT) {
845     currenttime = lastT;
846   }
847   lastT = currenttime;
848 #endif
849   return currenttime - inittime_wallclock;
850 }
851
852 double CmiTimer()
853 {
854   return CmiCpuTimer();
855 }
856
857 #endif
858
859 #if CMK_TIMER_USE_RDTSC
860
861 static double readMHz(void)
862 {
863   double x;
864   char str[1000];
865   char buf[100];
866   FILE *fp;
867   CmiLock(smp_mutex);
868   fp = fopen("/proc/cpuinfo", "r");
869   if (fp != NULL)
870   while(fgets(str, 1000, fp)!=0) {
871     if(sscanf(str, "cpu MHz%[^:]",buf)==1)
872     {
873       char *s = strchr(str, ':'); s=s+1;
874       sscanf(s, "%lf", &x);
875       fclose(fp);
876       CmiUnlock(smp_mutex);
877       return x;
878     }
879   }
880   CmiUnlock(smp_mutex);
881   CmiAbort("Cannot read CPU MHz from /proc/cpuinfo file.");
882   return 0.0;
883 }
884
885 double _cpu_speed_factor;
886 CpvStaticDeclare(double, inittime_virtual);
887 CpvStaticDeclare(double, inittime_walltime);
888
889 double  CmiStartTimer(void)
890 {
891   return CpvAccess(inittime_walltime);
892 }
893
894 void CmiTimerInit()
895 {
896   struct rusage ru;
897
898   CmiBarrier();
899   CmiBarrier();
900
901   _cpu_speed_factor = 1.0/(readMHz()*1.0e6); 
902   rdtsc(); rdtsc(); rdtsc(); rdtsc(); rdtsc();
903   CpvInitialize(double, inittime_walltime);
904   CpvAccess(inittime_walltime) = CmiWallTimer();
905   CpvInitialize(double, inittime_virtual);
906   getrusage(0, &ru); 
907   CpvAccess(inittime_virtual) =
908     (ru.ru_utime.tv_sec * 1.0)+(ru.ru_utime.tv_usec * 0.000001) +
909     (ru.ru_stime.tv_sec * 1.0)+(ru.ru_stime.tv_usec * 0.000001);
910
911   CmiBarrierZero();
912 }
913
914 double CmiCpuTimer()
915 {
916   struct rusage ru;
917   double currenttime;
918
919   getrusage(0, &ru);
920   currenttime =
921     (ru.ru_utime.tv_sec * 1.0)+(ru.ru_utime.tv_usec * 0.000001) +
922     (ru.ru_stime.tv_sec * 1.0)+(ru.ru_stime.tv_usec * 0.000001);
923   return currenttime - CpvAccess(inittime_virtual);
924 }
925
926 #endif
927
928 #if CMK_BLUEGENEL || CMK_BLUEGENEP
929 #include "dcopy.h"
930 #endif
931
932 #if CMK_TIMER_USE_BLUEGENEL
933
934 #include "rts.h"
935
936 #if 0 
937 #define SPRN_TBRL 0x10C  /* Time Base Read Lower Register (user & sup R/O) */
938 #define SPRN_TBRU 0x10D  /* Time Base Read Upper Register (user & sup R/O) */
939 #define SPRN_PIR  0x11E  /* CPU id */
940
941 static inline unsigned long long BGLTimebase(void)
942 {
943   unsigned volatile u1, u2, lo;
944   union
945   {
946     struct { unsigned hi, lo; } w;
947     unsigned long long d;
948   } result;
949                                                                                 
950   do {
951     asm volatile ("mfspr %0,%1" : "=r" (u1) : "i" (SPRN_TBRU));
952     asm volatile ("mfspr %0,%1" : "=r" (lo) : "i" (SPRN_TBRL));
953     asm volatile ("mfspr %0,%1" : "=r" (u2) : "i" (SPRN_TBRU));
954   } while (u1!=u2);
955                                                                                 
956   result.w.lo = lo;
957   result.w.hi = u2;
958   return result.d;
959 }
960 #endif
961
962 static unsigned long long inittime_wallclock = 0;
963 CpvStaticDeclare(double, clocktick);
964
965 int CmiTimerIsSynchronized()
966 {
967   return 0;
968 }
969
970 void CmiTimerInit()
971 {
972   BGLPersonality dst;
973   CpvInitialize(double, clocktick);
974   int size = sizeof(BGLPersonality);
975   rts_get_personality(&dst, size);
976   CpvAccess(clocktick) = 1.0 / dst.clockHz;
977
978   /* try to synchronize calling barrier */
979   CmiBarrier();
980   CmiBarrier();
981   CmiBarrier();
982
983   /* inittime_wallclock = rts_get_timebase(); */
984   inittime_wallclock = 0.0;    /* use bgl absolute time */
985 }
986
987 double CmiWallTimer()
988 {
989   unsigned long long currenttime;
990   currenttime = rts_get_timebase();
991   return CpvAccess(clocktick)*(currenttime-inittime_wallclock);
992 }
993
994 double CmiCpuTimer()
995 {
996   return CmiWallTimer();
997 }
998
999 double CmiTimer()
1000 {
1001   return CmiWallTimer();
1002 }
1003
1004 #endif
1005
1006 #if CMK_TIMER_USE_BLUEGENEP  /* This module just compiles with GCC charm. */
1007
1008 void CmiTimerInit() {}
1009
1010 #if 0
1011 #include "common/bgp_personality.h"
1012 #include <spi/bgp_SPI.h>
1013
1014 #define SPRN_TBRL 0x10C  /* Time Base Read Lower Register (user & sup R/O) */
1015 #define SPRN_TBRU 0x10D  /* Time Base Read Upper Register (user & sup R/O) */
1016 #define SPRN_PIR  0x11E  /* CPU id */
1017
1018 static inline unsigned long long BGPTimebase(void)
1019 {
1020   unsigned volatile u1, u2, lo;
1021   union
1022   {
1023     struct { unsigned hi, lo; } w;
1024     unsigned long long d;
1025   } result;
1026                                                                          
1027   do {
1028     asm volatile ("mfspr %0,%1" : "=r" (u1) : "i" (SPRN_TBRU));
1029     asm volatile ("mfspr %0,%1" : "=r" (lo) : "i" (SPRN_TBRL));
1030     asm volatile ("mfspr %0,%1" : "=r" (u2) : "i" (SPRN_TBRU));
1031   } while (u1!=u2);
1032                                                                          
1033   result.w.lo = lo;
1034   result.w.hi = u2;
1035   return result.d;
1036 }
1037
1038 static unsigned long long inittime_wallclock = 0;
1039 CpvStaticDeclare(double, clocktick);
1040
1041 int CmiTimerIsSynchronized()
1042 {
1043   return 0;
1044 }
1045
1046 void CmiTimerInit()
1047 {
1048   _BGP_Personality_t dst;
1049   CpvInitialize(double, clocktick);
1050   int size = sizeof(_BGP_Personality_t);
1051   rts_get_personality(&dst, size);
1052
1053   CpvAccess(clocktick) = 1.0 / (dst.Kernel_Config.FreqMHz * 1e6);
1054
1055   /* try to synchronize calling barrier */
1056   CmiBarrier();
1057   CmiBarrier();
1058   CmiBarrier();
1059
1060   inittime_wallclock = BGPTimebase (); 
1061 }
1062
1063 double CmiWallTimer()
1064 {
1065   unsigned long long currenttime;
1066   currenttime = BGPTimebase();
1067   return CpvAccess(clocktick)*(currenttime-inittime_wallclock);
1068 }
1069 #endif
1070
1071 #include "dcmf.h"
1072
1073 double CmiWallTimer () {
1074   return DCMF_Timer();
1075 }
1076
1077 double CmiCpuTimer()
1078 {
1079   return CmiWallTimer();
1080 }
1081
1082 double CmiTimer()
1083 {
1084   return CmiWallTimer();
1085 }
1086
1087 #endif
1088
1089
1090 #if CMK_TIMER_USE_WIN32API
1091
1092 CpvStaticDeclare(double, inittime_wallclock);
1093 CpvStaticDeclare(double, inittime_virtual);
1094
1095 void CmiTimerInit()
1096 {
1097 #ifdef __CYGWIN__
1098         struct timeb tv;
1099 #else
1100         struct _timeb tv;
1101 #endif
1102         clock_t       ru;
1103
1104         CpvInitialize(double, inittime_wallclock);
1105         CpvInitialize(double, inittime_virtual);
1106         _ftime(&tv);
1107         CpvAccess(inittime_wallclock) = tv.time*1.0 + tv.millitm*0.001;
1108         ru = clock();
1109         CpvAccess(inittime_virtual) = ((double) ru)/CLOCKS_PER_SEC;
1110 }
1111
1112 double CmiCpuTimer()
1113 {
1114         clock_t ru;
1115         double currenttime;
1116
1117         ru = clock();
1118         currenttime = (double) ru/CLOCKS_PER_SEC;
1119
1120         return currenttime - CpvAccess(inittime_virtual);
1121 }
1122
1123 double CmiWallTimer()
1124 {
1125 #ifdef __CYGWIN__
1126         struct timeb tv;
1127 #else
1128         struct _timeb tv;
1129 #endif
1130         double currenttime;
1131
1132         _ftime(&tv);
1133         currenttime = tv.time*1.0 + tv.millitm*0.001;
1134
1135         return currenttime - CpvAccess(inittime_wallclock);
1136 }
1137         
1138
1139 double CmiTimer()
1140 {
1141         return CmiCpuTimer();
1142 }
1143
1144 #endif
1145
1146 #if CMK_TIMER_USE_RTC
1147
1148 #if __crayx1
1149  /* For _rtc() on Cray X1 */
1150 #include <intrinsics.h>
1151 #endif
1152
1153 static double clocktick;
1154 CpvStaticDeclare(long long, inittime_wallclock);
1155
1156 void CmiTimerInit()
1157 {
1158   CpvInitialize(long long, inittime_wallclock);
1159   CpvAccess(inittime_wallclock) = _rtc();
1160   clocktick = 1.0 / (double)(sysconf(_SC_SV2_USER_TIME_RATE));
1161 }
1162
1163 double CmiWallTimer()
1164 {
1165   long long now;
1166
1167   now = _rtc();
1168   return (clocktick * (now - CpvAccess(inittime_wallclock)));
1169 }
1170
1171 double CmiCpuTimer()
1172 {
1173   return CmiWallTimer();
1174 }
1175
1176 double CmiTimer()
1177 {
1178   return CmiCpuTimer();
1179 }
1180
1181 #endif
1182
1183 #if CMK_TIMER_USE_AIX_READ_TIME
1184
1185 #include <sys/time.h>
1186
1187 static timebasestruct_t inittime_wallclock;
1188 static double clocktick;
1189 CpvStaticDeclare(double, inittime_virtual);
1190
1191 void CmiTimerInit()
1192 {
1193   struct rusage ru;
1194
1195   if (CmiMyRank() == 0) {
1196     read_wall_time(&inittime_wallclock, TIMEBASE_SZ);
1197     time_base_to_time(&inittime_wallclock, TIMEBASE_SZ);
1198   }
1199
1200   CpvInitialize(double, inittime_virtual);
1201   getrusage(0, &ru);
1202   CpvAccess(inittime_virtual) =
1203     (ru.ru_utime.tv_sec * 1.0)+(ru.ru_utime.tv_usec * 0.000001) +
1204     (ru.ru_stime.tv_sec * 1.0)+(ru.ru_stime.tv_usec * 0.000001);
1205 }
1206
1207 double CmiWallTimer()
1208 {
1209   int secs, n_secs;
1210   double curt;
1211   timebasestruct_t now;
1212   read_wall_time(&now, TIMEBASE_SZ);
1213   time_base_to_time(&now, TIMEBASE_SZ);
1214
1215   secs = now.tb_high - inittime_wallclock.tb_high;
1216   n_secs = now.tb_low - inittime_wallclock.tb_low;
1217   if (n_secs < 0)  {
1218     secs--;
1219     n_secs += 1000000000;
1220   }
1221   curt = secs*1.0 + n_secs*1e-9;
1222   return curt;
1223 }
1224
1225 double CmiCpuTimer()
1226 {
1227   struct rusage ru;
1228   double currenttime;
1229
1230   getrusage(0, &ru);
1231   currenttime =
1232     (ru.ru_utime.tv_sec * 1.0)+(ru.ru_utime.tv_usec * 0.000001) +
1233     (ru.ru_stime.tv_sec * 1.0)+(ru.ru_stime.tv_usec * 0.000001);
1234   return currenttime - CpvAccess(inittime_virtual);
1235 }
1236
1237 double CmiTimer()
1238 {
1239   return CmiWallTimer();
1240 }
1241
1242 #endif
1243
1244 #ifndef CMK_USE_SPECIAL_MESSAGE_QUEUE_CHECK
1245 /** Return 1 if our outgoing message queue 
1246    for this node is longer than this many bytes. */
1247 int CmiLongSendQueue(int forNode,int longerThanBytes) {
1248   return 0;
1249 }
1250 #endif
1251
1252 #if CMK_SIGNAL_USE_SIGACTION
1253 #include <signal.h>
1254 void CmiSignal(int sig1, int sig2, int sig3, void (*handler)())
1255 {
1256   struct sigaction in, out ;
1257   in.sa_handler = handler;
1258   sigemptyset(&in.sa_mask);
1259   if (sig1) sigaddset(&in.sa_mask, sig1);
1260   if (sig2) sigaddset(&in.sa_mask, sig2);
1261   if (sig3) sigaddset(&in.sa_mask, sig3);
1262   in.sa_flags = 0;
1263   if (sig1) if (sigaction(sig1, &in, &out)<0) exit(1);
1264   if (sig2) if (sigaction(sig2, &in, &out)<0) exit(1);
1265   if (sig3) if (sigaction(sig3, &in, &out)<0) exit(1);
1266 }
1267 #endif
1268
1269 #if CMK_SIGNAL_USE_SIGACTION_WITH_RESTART
1270 #include <signal.h>
1271 void CmiSignal(sig1, sig2, sig3, handler)
1272 int sig1, sig2, sig3;
1273 void (*handler)();
1274 {
1275   struct sigaction in, out ;
1276   in.sa_handler = handler;
1277   sigemptyset(&in.sa_mask);
1278   if (sig1) sigaddset(&in.sa_mask, sig1);
1279   if (sig2) sigaddset(&in.sa_mask, sig2);
1280   if (sig3) sigaddset(&in.sa_mask, sig3);
1281   in.sa_flags = SA_RESTART;
1282   if (sig1) if (sigaction(sig1, &in, &out)<0) exit(1);
1283   if (sig2) if (sigaction(sig2, &in, &out)<0) exit(1);
1284   if (sig3) if (sigaction(sig3, &in, &out)<0) exit(1);
1285 }
1286 #endif
1287
1288 /** 
1289  *  @addtogroup ConverseScheduler
1290  *  @{
1291  */
1292
1293 /*****************************************************************************
1294  * 
1295  * The following is the CsdScheduler function.  A common
1296  * implementation is provided below.  The machine layer can provide an
1297  * alternate implementation if it so desires.
1298  *
1299  * void CmiDeliversInit()
1300  *
1301  *      - CmiInit promises to call this before calling CmiDeliverMsgs
1302  *        or any of the other functions in this section.
1303  *
1304  * int CmiDeliverMsgs(int maxmsgs)
1305  *
1306  *      - CmiDeliverMsgs will retrieve up to maxmsgs that were transmitted
1307  *        with the Cmi, and will invoke their handlers.  It does not wait
1308  *        if no message is unavailable.  Instead, it returns the quantity
1309  *        (maxmsgs-delivered), where delivered is the number of messages it
1310  *        delivered.
1311  *
1312  * void CmiDeliverSpecificMsg(int handlerno)
1313  *
1314  *      - Waits for a message with the specified handler to show up, then
1315  *        invokes the message's handler.  Note that unlike CmiDeliverMsgs,
1316  *        This function _does_ wait.
1317  *
1318  * For this common implementation to work, the machine layer must provide the
1319  * following:
1320  *
1321  * void *CmiGetNonLocal()
1322  *
1323  *      - returns a message just retrieved from some other PE, not from
1324  *        local.  If no such message exists, returns 0.
1325  *
1326  * CpvExtern(CdsFifo, CmiLocalQueue);
1327  *
1328  *      - a FIFO queue containing all messages from the local processor.
1329  *
1330  *****************************************************************************/
1331
1332 void CsdBeginIdle(void)
1333 {
1334   CcdCallBacks();
1335   _LOG_E_PROC_IDLE();   /* projector */
1336   CcdRaiseCondition(CcdPROCESSOR_BEGIN_IDLE) ;
1337 }
1338
1339 void CsdStillIdle(void)
1340 {
1341   CcdRaiseCondition(CcdPROCESSOR_STILL_IDLE);
1342 }
1343
1344 void CsdEndIdle(void)
1345 {
1346   _LOG_E_PROC_BUSY();   /* projector */
1347   CcdRaiseCondition(CcdPROCESSOR_BEGIN_BUSY) ;
1348 }
1349
1350 #if CMK_MEM_CHECKPOINT
1351 #define MESSAGE_PHASE_CHECK     \
1352         {       \
1353           int phase = CmiGetRestartPhase(msg);  \
1354           if (phase < cur_restart_phase) {      \
1355             /*CmiPrintf("[%d] discard message of phase %d cur_restart_phase:%d handler:%d. \n", CmiMyPe(), phase, cur_restart_phase, handler);*/        \
1356             CmiFree(msg);       \
1357             return;     \
1358           }     \
1359         }
1360 #else
1361 #define MESSAGE_PHASE_CHECK
1362 #endif
1363
1364 extern int _exitHandlerIdx;
1365
1366 /** Takes a message and calls its corresponding handler. */
1367 void CmiHandleMessage(void *msg)
1368 {
1369 /* this is wrong because it counts the Charm++ messages in sched queue
1370         CpvAccess(cQdState)->mProcessed++;
1371 */
1372         CmiHandlerInfo *h;
1373 #ifndef CMK_OPTIMIZE
1374         CmiUInt2 handler=CmiGetHandler(msg); /* Save handler for use after msg is gone */
1375         _LOG_E_HANDLER_BEGIN(handler); /* projector */
1376         /* setMemoryStatus(1) */ /* charmdebug */
1377 #endif
1378
1379 /*
1380         FAULT_EVAC
1381 */
1382 /*      if((!CpvAccess(_validProcessors)[CmiMyPe()]) && handler != _exitHandlerIdx){
1383                 return;
1384         }*/
1385         
1386         MESSAGE_PHASE_CHECK
1387
1388         h=&CmiGetHandlerInfo(msg);
1389         (h->hdlr)(msg,h->userPtr);
1390 #ifndef CMK_OPTIMIZE
1391         /* setMemoryStatus(0) */ /* charmdebug */
1392         _LOG_E_HANDLER_END(handler);    /* projector */
1393 #endif
1394 }
1395
1396 #if CMK_CMIDELIVERS_USE_COMMON_CODE
1397
1398 void CmiDeliversInit()
1399 {
1400 }
1401
1402 int CmiDeliverMsgs(int maxmsgs)
1403 {
1404   return CsdScheduler(maxmsgs);
1405 }
1406
1407 #if CMK_OBJECT_QUEUE_AVAILABLE
1408 CpvDeclare(void *, CsdObjQueue);
1409 #endif
1410
1411 void CsdSchedulerState_new(CsdSchedulerState_t *s)
1412 {
1413 #if CMK_OBJECT_QUEUE_AVAILABLE
1414         s->objQ=CpvAccess(CsdObjQueue);
1415 #endif
1416         s->localQ=CpvAccess(CmiLocalQueue);
1417         s->schedQ=CpvAccess(CsdSchedQueue);
1418         s->localCounter=&(CpvAccess(CsdLocalCounter));
1419 #if CMK_NODE_QUEUE_AVAILABLE
1420         s->nodeQ=CsvAccess(CsdNodeQueue);
1421         s->nodeLock=CsvAccess(CsdNodeQueueLock);
1422 #endif
1423 #if CMK_GRID_QUEUE_AVAILABLE
1424         s->gridQ=CpvAccess(CsdGridQueue);
1425 #endif
1426 }
1427
1428
1429 /** Dequeue and return the next message from the message queue. */
1430 void *CsdNextMessage(CsdSchedulerState_t *s) {
1431         void *msg;
1432         if((*(s->localCounter))-- >0)
1433           {
1434               /* This avoids a race condition with migration detected by megatest*/
1435               msg=CdsFifo_Dequeue(s->localQ);
1436               if (msg!=NULL)
1437                 {
1438                   CpvAccess(cQdState)->mProcessed++;
1439                   return msg;       
1440                 }
1441               CqsDequeue(s->schedQ,(void **)&msg);
1442               if (msg!=NULL) return msg;
1443           }
1444         
1445         *(s->localCounter)=CsdLocalMax;
1446         if ( NULL!=(msg=CmiGetNonLocal()) || 
1447              NULL!=(msg=CdsFifo_Dequeue(s->localQ)) ) {
1448             CpvAccess(cQdState)->mProcessed++;
1449             return msg;
1450         }
1451 #if CMK_GRID_QUEUE_AVAILABLE
1452         /*#warning "CsdNextMessage: CMK_GRID_QUEUE_AVAILABLE" */
1453         CqsDequeue (s->gridQ, (void **) &msg);
1454         if (msg != NULL) {
1455           return (msg);
1456         }
1457 #endif
1458 #if CMK_NODE_QUEUE_AVAILABLE
1459         /*#warning "CsdNextMessage: CMK_NODE_QUEUE_AVAILABLE" */
1460         if (NULL!=(msg=CmiGetNonLocalNodeQ())) return msg;
1461         if (!CqsEmpty(s->nodeQ)
1462          && !CqsPrioGT(CqsGetPriority(s->nodeQ),
1463                        CqsGetPriority(s->schedQ))) {
1464           CmiLock(s->nodeLock);
1465           CqsDequeue(s->nodeQ,(void **)&msg);
1466           CmiUnlock(s->nodeLock);
1467           if (msg!=NULL) return msg;
1468         }
1469 #endif
1470 #if CMK_OBJECT_QUEUE_AVAILABLE
1471         /*#warning "CsdNextMessage: CMK_OBJECT_QUEUE_AVAILABLE"   */
1472         if (NULL!=(msg=CdsFifo_Dequeue(s->objQ))) {
1473           return msg;
1474         }
1475 #endif
1476         if(!CsdLocalMax) {
1477           CqsDequeue(s->schedQ,(void **)&msg);
1478           if (msg!=NULL) return msg;        
1479         }
1480         return NULL;
1481 }
1482
1483 int CsdScheduler(int maxmsgs)
1484 {
1485         if (maxmsgs<0) CsdScheduleForever();    
1486         else if (maxmsgs==0)
1487                 CsdSchedulePoll();
1488         else /*(maxmsgs>0)*/ 
1489                 return CsdScheduleCount(maxmsgs);
1490         return 0;
1491 }
1492
1493 /*Declare the standard scheduler housekeeping*/
1494 #define SCHEDULE_TOP \
1495       void *msg;\
1496       int *CsdStopFlag_ptr = &CpvAccess(CsdStopFlag); \
1497       int cycle = CpvAccess(CsdStopFlag); \
1498       CsdSchedulerState_t state;\
1499       CsdSchedulerState_new(&state);\
1500
1501 /*A message is available-- process it*/
1502 #define SCHEDULE_MESSAGE \
1503       CmiHandleMessage(msg);\
1504       if (*CsdStopFlag_ptr != cycle) break;\
1505
1506 /*No message available-- go (or remain) idle*/
1507 #define SCHEDULE_IDLE \
1508       if (!isIdle) {isIdle=1;CsdBeginIdle();}\
1509       else CsdStillIdle();\
1510       if (*CsdStopFlag_ptr != cycle) {\
1511         CsdEndIdle();\
1512         break;\
1513       }\
1514 /*
1515         EVAC
1516 */
1517 extern void CkClearAllArrayElements();
1518
1519
1520 extern void machine_OffloadAPIProgress();
1521
1522 /** The main scheduler loop that repeatedly executes messages from a queue, forever. */
1523 void CsdScheduleForever(void)
1524 {
1525   #if CMK_CELL
1526     #define CMK_CELL_PROGRESS_FREQ  96  /* (MSG-Q Entries x1.5) */
1527     int progressCount = CMK_CELL_PROGRESS_FREQ;
1528   #endif
1529
1530   #ifdef CMK_CUDA
1531     #define CMK_CUDA_PROGRESS_FREQ 50
1532     int cudaProgressCount = CMK_CUDA_PROGRESS_FREQ;
1533   #endif
1534
1535   int isIdle=0;
1536   SCHEDULE_TOP
1537   while (1) {
1538     msg = CsdNextMessage(&state);
1539     if (msg!=NULL) { /*A message is available-- process it*/
1540       if (isIdle) {isIdle=0;CsdEndIdle();}
1541       SCHEDULE_MESSAGE
1542
1543       #if CMK_CELL
1544         if (progressCount <= 0) {
1545           /*OffloadAPIProgress();*/
1546           machine_OffloadAPIProgress();
1547           progressCount = CMK_CELL_PROGRESS_FREQ;
1548         }
1549         progressCount--;
1550       #endif
1551
1552       #ifdef CMK_CUDA
1553         if (cudaProgressCount == 0) {
1554           gpuProgressFn(); 
1555           cudaProgressCount = CMK_CUDA_PROGRESS_FREQ; 
1556         }
1557         cudaProgressCount--; 
1558       #endif
1559
1560     } else { /*No message available-- go (or remain) idle*/
1561       SCHEDULE_IDLE
1562
1563       #if CMK_CELL
1564         /*OffloadAPIProgress();*/
1565         machine_OffloadAPIProgress();
1566         progressCount = CMK_CELL_PROGRESS_FREQ;
1567       #endif
1568
1569       #ifdef CMK_CUDA
1570         gpuProgressFn(); 
1571         cudaProgressCount = CMK_CUDA_PROGRESS_FREQ;
1572       #endif
1573
1574     }
1575     CsdPeriodic();
1576   }
1577 }
1578 int CsdScheduleCount(int maxmsgs)
1579 {
1580   int isIdle=0;
1581   SCHEDULE_TOP
1582   while (1) {
1583     msg = CsdNextMessage(&state);
1584     if (msg!=NULL) { /*A message is available-- process it*/
1585       if (isIdle) {isIdle=0;CsdEndIdle();}
1586       maxmsgs--; 
1587       SCHEDULE_MESSAGE
1588       if (maxmsgs==0) break;
1589     } else { /*No message available-- go (or remain) idle*/
1590       SCHEDULE_IDLE
1591     }
1592     CsdPeriodic();
1593   }
1594   return maxmsgs;
1595 }
1596
1597 void CsdSchedulePoll(void)
1598 {
1599   SCHEDULE_TOP
1600   while (1)
1601   {
1602         CsdPeriodic();
1603         /*CmiMachineProgressImpl(); ??? */
1604         if (NULL!=(msg = CsdNextMessage(&state)))
1605         {
1606              SCHEDULE_MESSAGE 
1607         }
1608         else break;
1609   }
1610 }
1611
1612 void CmiDeliverSpecificMsg(handler)
1613 int handler;
1614 {
1615   int *msg; int side;
1616   void *localqueue = CpvAccess(CmiLocalQueue);
1617  
1618   side = 0;
1619   while (1) {
1620     CsdPeriodic();
1621     side ^= 1;
1622     if (side) msg = CmiGetNonLocal();
1623     else      msg = CdsFifo_Dequeue(localqueue);
1624     if (msg) {
1625       if (CmiGetHandler(msg)==handler) {
1626         CpvAccess(cQdState)->mProcessed++;
1627         CmiHandleMessage(msg);
1628         return;
1629       } else {
1630         CdsFifo_Enqueue(localqueue, msg);
1631       }
1632     }
1633   }
1634 }
1635  
1636 #endif /* CMK_CMIDELIVERS_USE_COMMON_CODE */
1637
1638 /***************************************************************************
1639  *
1640  * Standin Schedulers.
1641  *
1642  * We use the following strategy to make sure somebody's always running
1643  * the scheduler (CsdScheduler).  Initially, we assume the main thread
1644  * is responsible for this.  If the main thread blocks, we create a
1645  * "standin scheduler" thread to replace it.  If the standin scheduler
1646  * blocks, we create another standin scheduler to replace that one,
1647  * ad infinitum.  Collectively, the main thread and all the standin
1648  * schedulers are called "scheduling threads".
1649  *
1650  * Suppose the main thread is blocked waiting for data, and a standin
1651  * scheduler is running instead.  Suppose, then, that the data shows
1652  * up and the main thread is CthAwakened.  This causes a token to be
1653  * pushed into the queue.  When the standin pulls the token from the
1654  * queue and handles it, the standin goes to sleep, and control shifts
1655  * back to the main thread.  In this way, unnecessary standins are put
1656  * back to sleep.  These sleeping standins are stored on the
1657  * CthSleepingStandins list.
1658  *
1659  ***************************************************************************/
1660
1661 CpvStaticDeclare(CthThread, CthMainThread);
1662 CpvStaticDeclare(CthThread, CthSchedulingThread);
1663 CpvStaticDeclare(CthThread, CthSleepingStandins);
1664 CpvDeclare(int      , CthResumeNormalThreadIdx);
1665 CpvStaticDeclare(int      , CthResumeSchedulingThreadIdx);
1666
1667
1668 void CthStandinCode()
1669 {
1670   while (1) CsdScheduler(0);
1671 }
1672
1673 /* this fix the function pointer for thread migration and pup */
1674 static CthThread CthSuspendNormalThread()
1675 {
1676   return CpvAccess(CthSchedulingThread);
1677 }
1678
1679 void CthEnqueueSchedulingThread(CthThreadToken *token, int, int, unsigned int*);
1680 CthThread CthSuspendSchedulingThread();
1681
1682 CthThread CthSuspendSchedulingThread()
1683 {
1684   CthThread succ = CpvAccess(CthSleepingStandins);
1685
1686   if (succ) {
1687     CpvAccess(CthSleepingStandins) = CthGetNext(succ);
1688   } else {
1689     succ = CthCreate(CthStandinCode, 0, 256000);
1690     CthSetStrategy(succ,
1691                    CthEnqueueSchedulingThread,
1692                    CthSuspendSchedulingThread);
1693   }
1694   
1695   CpvAccess(CthSchedulingThread) = succ;
1696   return succ;
1697 }
1698
1699 /* Notice: For changes to the following function, make sure the function CthResumeNormalThreadDebug is also kept updated. */
1700 void CthResumeNormalThread(CthThreadToken* token)
1701 {
1702   CthThread t = token->thread;
1703
1704   /* BIGSIM_OOC DEBUGGING
1705   CmiPrintf("Resume normal thread with token[%p] ==> thread[%p]\n", token, t);
1706   */
1707
1708   if(t == NULL){
1709     free(token);
1710     return;
1711   }
1712 #ifndef CMK_OPTIMIZE
1713 #if ! CMK_TRACE_IN_CHARM
1714   if(CpvAccess(traceOn))
1715     CthTraceResume(t);
1716 /*    if(CpvAccess(_traceCoreOn)) 
1717                 resumeTraceCore();*/
1718 #endif
1719 #endif
1720   
1721   /* BIGSIM_OOC DEBUGGING
1722   CmiPrintf("In CthResumeNormalThread:   ");
1723   CthPrintThdMagic(t);
1724   */
1725
1726   CthResume(t);
1727 }
1728
1729 void CthResumeSchedulingThread(CthThreadToken  *token)
1730 {
1731   CthThread t = token->thread;
1732   CthThread me = CthSelf();
1733   if (me == CpvAccess(CthMainThread)) {
1734     CthEnqueueSchedulingThread(CthGetToken(me),CQS_QUEUEING_FIFO, 0, 0);
1735   } else {
1736     CthSetNext(me, CpvAccess(CthSleepingStandins));
1737     CpvAccess(CthSleepingStandins) = me;
1738   }
1739   CpvAccess(CthSchedulingThread) = t;
1740 #ifndef CMK_OPTIMIZE
1741 #if ! CMK_TRACE_IN_CHARM
1742   if(CpvAccess(traceOn))
1743     CthTraceResume(t);
1744 /*    if(CpvAccess(_traceCoreOn)) 
1745                 resumeTraceCore();*/
1746 #endif
1747 #endif
1748   CthResume(t);
1749 }
1750
1751 void CthEnqueueNormalThread(CthThreadToken* token, int s, 
1752                                    int pb,unsigned int *prio)
1753 {
1754   CmiSetHandler(token, CpvAccess(CthResumeNormalThreadIdx));
1755   CsdEnqueueGeneral(token, s, pb, prio);
1756 }
1757
1758 void CthEnqueueSchedulingThread(CthThreadToken* token, int s, 
1759                                        int pb,unsigned int *prio)
1760 {
1761   CmiSetHandler(token, CpvAccess(CthResumeSchedulingThreadIdx));
1762   CsdEnqueueGeneral(token, s, pb, prio);
1763 }
1764
1765 void CthSetStrategyDefault(CthThread t)
1766 {
1767   CthSetStrategy(t,
1768                  CthEnqueueNormalThread,
1769                  CthSuspendNormalThread);
1770 }
1771
1772 void CthSchedInit()
1773 {
1774   CpvInitialize(CthThread, CthMainThread);
1775   CpvInitialize(CthThread, CthSchedulingThread);
1776   CpvInitialize(CthThread, CthSleepingStandins);
1777   CpvInitialize(int      , CthResumeNormalThreadIdx);
1778   CpvInitialize(int      , CthResumeSchedulingThreadIdx);
1779
1780   CpvAccess(CthMainThread) = CthSelf();
1781   CpvAccess(CthSchedulingThread) = CthSelf();
1782   CpvAccess(CthSleepingStandins) = 0;
1783   CpvAccess(CthResumeNormalThreadIdx) =
1784     CmiRegisterHandler((CmiHandler)CthResumeNormalThread);
1785   CpvAccess(CthResumeSchedulingThreadIdx) =
1786     CmiRegisterHandler((CmiHandler)CthResumeSchedulingThread);
1787   CthSetStrategy(CthSelf(),
1788                  CthEnqueueSchedulingThread,
1789                  CthSuspendSchedulingThread);
1790 }
1791
1792 void CsdInit(argv)
1793   char **argv;
1794 {
1795   CpvInitialize(void *, CsdSchedQueue);
1796   CpvInitialize(int,   CsdStopFlag);
1797   CpvInitialize(int,   CsdLocalCounter);
1798   if(!CmiGetArgIntDesc(argv,"+csdLocalMax",&CsdLocalMax,"Set the max number of local messages to process before forcing a check for remote messages."))
1799     {
1800       CsdLocalMax= CSD_LOCAL_MAX_DEFAULT;
1801     }
1802   CpvAccess(CsdLocalCounter) = CsdLocalMax;
1803   CpvAccess(CsdSchedQueue) = (void *)CqsCreate();
1804
1805 #if CMK_OBJECT_QUEUE_AVAILABLE
1806   CpvInitialize(void *,CsdObjQueue);
1807   CpvAccess(CsdObjQueue) = CdsFifo_Create();
1808 #endif
1809
1810 #if CMK_NODE_QUEUE_AVAILABLE
1811   CsvInitialize(CmiLock, CsdNodeQueueLock);
1812   CsvInitialize(void *, CsdNodeQueue);
1813   if (CmiMyRank() ==0) {
1814         CsvAccess(CsdNodeQueueLock) = CmiCreateLock();
1815         CsvAccess(CsdNodeQueue) = (void *)CqsCreate();
1816   }
1817   CmiNodeAllBarrier();
1818 #endif
1819
1820 #if CMK_GRID_QUEUE_AVAILABLE
1821   CsvInitialize(void *, CsdGridQueue);
1822   CpvAccess(CsdGridQueue) = (void *)CqsCreate();
1823 #endif
1824
1825   CpvAccess(CsdStopFlag)  = 0;
1826 }
1827
1828
1829
1830 /** 
1831  *  @}
1832  */
1833
1834
1835 /*****************************************************************************
1836  *
1837  * Vector Send
1838  *
1839  * The last parameter "system" is by default at zero, in which case the normal
1840  * messages are sent. If it is set to 1, the CmiChunkHeader prepended to every
1841  * CmiAllocced message will also be sent (except for the first one). Useful for
1842  * AllToAll communication, and other system features. If system is 1, also all
1843  * the messages will be padded to 8 bytes. Thus, the caller must be aware of
1844  * that.
1845  *
1846  ****************************************************************************/
1847
1848 #if CMK_VECTOR_SEND_USES_COMMON_CODE
1849
1850 void CmiSyncVectorSend(int destPE, int n, int *sizes, char **msgs) {
1851   int total;
1852   char *mesg;
1853 #if CMK_USE_IBVERBS
1854   VECTOR_COMPACT(total, mesg, n, sizes, msgs,sizeof(infiCmiChunkHeader));
1855 #else
1856   VECTOR_COMPACT(total, mesg, n, sizes, msgs,sizeof(CmiChunkHeader));
1857 #endif  
1858   CmiSyncSendAndFree(destPE, total, mesg);
1859 }
1860
1861 CmiCommHandle CmiASyncVectorSend(int destPE, int n, int *sizes, char **msgs) {
1862   CmiSyncVectorSend(destPE, n, sizes, msgs);
1863   return NULL;
1864 }
1865
1866 void CmiSyncVectorSendAndFree(int destPE, int n, int *sizes, char **msgs) {
1867   int i;
1868   CmiSyncVectorSend(destPE, n, sizes, msgs);
1869   for(i=0;i<n;i++) CmiFree(msgs[i]);
1870   CmiFree(sizes);
1871   CmiFree(msgs);
1872 }
1873
1874 #endif
1875
1876 /*****************************************************************************
1877  *
1878  * Reduction management
1879  *
1880  * Only one reduction can be active at a single time in the program.
1881  * Moreover, since every call is supposed to pass in the same arguments,
1882  * having some static variables is not a problem for multithreading.
1883  * 
1884  * Except for "data" and "size", all the other parameters (which are all function
1885  * pointers) MUST be the same in every processor. Having different processors
1886  * pass in different function pointers results in an undefined behaviour.
1887  * 
1888  * The data passed in to CmiReduce and CmiNodeReduce is deleted by the system,
1889  * and MUST be allocated with CmiAlloc. The data passed in to the "Struct"
1890  * functions is deleted with the provided function, or it is left intact if no
1891  * function is specified.
1892  * 
1893  * The destination handler for the the first form MUST be embedded into the
1894  * message's header.
1895  * 
1896  * The pup function is used to pup the input data structure into a message to
1897  * be sent to the parent processor. This pup routine is currently used only
1898  * for sizing and packing, NOT unpacking. It MUST be non-null.
1899  * 
1900  * The merge function receives as first parameter the input "data", being it
1901  * a message or a complex data structure (it is up to the user to interpret it
1902  * correctly), and a list of incoming (packed) messages from the children.
1903  * The merge function is responsible to delete "data" if this is no longer needed.
1904  * The system will be in charge of deleting the messages passed in as the second
1905  * argument, and the return value of the function (using the provided deleteFn in
1906  * the second version, or CmiFree in the first). The merge function can return
1907  * data if the merge can be performed in-place. It MUST be non-null.
1908  * 
1909  * At the destination, on processor zero, the final data returned by the last
1910  * merge call will not be deleted by the system, and the CmiHandler function
1911  * will be in charge of its deletion.
1912  * 
1913  * CmiReduce/CmiReduceStruct MUST be called once by every processor,
1914  * CmiNodeReduce/CmiNodeReduceStruct MUST be called once by every node, and in
1915  * particular by the rank zero in each node.
1916  ****************************************************************************/
1917
1918 CpvStaticDeclare(int, CmiReductionMessageHandler);
1919 CpvStaticDeclare(int, CmiReductionDynamicRequestHandler);
1920
1921 CpvStaticDeclare(CmiReduction**, _reduce_info);
1922 CpvStaticDeclare(int, _reduce_info_size); /* This is the log2 of the size of the array */
1923 CpvStaticDeclare(CmiUInt2, _reduce_seqID_global); /* This is used only by global reductions */
1924 CpvStaticDeclare(CmiUInt2, _reduce_seqID_request);
1925 CpvStaticDeclare(CmiUInt2, _reduce_seqID_dynamic);
1926
1927 enum {
1928   CmiReductionID_globalOffset = 0, /* Reductions that involve the whole set of processors */
1929   CmiReductionID_requestOffset = 1, /* Reductions IDs that are requested by all the processors (i.e during intialization) */
1930   CmiReductionID_dynamicOffset = 2, /* Reductions IDs that are requested by only one processor (typically at runtime) */
1931   CmiReductionID_multiplier = 3
1932 };
1933
1934 CmiReduction* CmiGetReductionCreate(int id, short int numChildren) {
1935   int index = id & ~((~0)<<CpvAccess(_reduce_info_size));
1936   CmiReduction *red = CpvAccess(_reduce_info)[index];
1937   if (red != NULL && red->seqID != id) {
1938     /* The table needs to be expanded */
1939     CmiAbort("Too many simultaneous reductions");
1940   }
1941   if (red == NULL || red->numChildren < numChildren) {
1942     CmiReduction *newred;
1943     if (red != NULL) CmiPrintf("[%d] Reduction structure reallocated\n",CmiMyPe());
1944     CmiAssert(red == NULL || red->localContributed == 0);
1945     if (numChildren == 0) numChildren = 4;
1946     newred = (CmiReduction*)malloc(sizeof(CmiReduction)+numChildren*sizeof(void*));
1947     newred->numRemoteReceived = 0;
1948     newred->localContributed = 0;
1949     newred->seqID = id;
1950     if (red != NULL) {
1951       memcpy(newred, red, sizeof(CmiReduction)+red->numChildren*sizeof(void*));
1952       free(red);
1953     }
1954     red = newred;
1955     red->numChildren = numChildren;
1956     red->remoteData = (char**)(red+1);
1957     CpvAccess(_reduce_info)[index] = red;
1958   }
1959   return red;
1960 }
1961
1962 CmiReduction* CmiGetReduction(int id) {
1963   return CmiGetReductionCreate(id, 0);
1964 }
1965
1966 void CmiClearReduction(int id) {
1967   int index = id & ~((~0)<<CpvAccess(_reduce_info_size));
1968   free(CpvAccess(_reduce_info)[index]);
1969   CpvAccess(_reduce_info)[index] = NULL;
1970 }
1971
1972 CmiReduction* CmiGetNextReduction(short int numChildren) {
1973   int id = CpvAccess(_reduce_seqID_global);
1974   CpvAccess(_reduce_seqID_global) += CmiReductionID_multiplier;
1975   if (id > 0xFFF0) CpvAccess(_reduce_seqID_global) = CmiReductionID_globalOffset;
1976   return CmiGetReductionCreate(id, numChildren);
1977 }
1978
1979 CmiReductionID CmiGetGlobalReduction() {
1980   return CpvAccess(_reduce_seqID_request)+=CmiReductionID_multiplier;
1981 }
1982
1983 CmiReductionID CmiGetDynamicReduction() {
1984   if (CmiMyPe() != 0) CmiAbort("Cannot call CmiGetDynamicReduction on processors other than zero!\n");
1985   return CpvAccess(_reduce_seqID_dynamic)+=CmiReductionID_multiplier;
1986 }
1987
1988 void CmiReductionHandleDynamicRequest(char *msg) {
1989   int *values = (int*)(msg+CmiMsgHeaderSizeBytes);
1990   int pe = values[0];
1991   int size = CmiMsgHeaderSizeBytes+2*sizeof(int)+values[1];
1992   values[0] = CmiGetDynamicReduction();
1993   CmiSetHandler(msg, CmiGetXHandler(msg));
1994   if (pe >= 0) {
1995     CmiSyncSendAndFree(pe, size, msg);
1996   } else {
1997     CmiSyncBroadcastAllAndFree(size, msg);
1998   }
1999 }
2000
2001 void CmiGetDynamicReductionRemote(int handlerIdx, int pe, int dataSize, void *data) {
2002   int size = CmiMsgHeaderSizeBytes+2*sizeof(int)+dataSize;
2003   char *msg = (char*)CmiAlloc(size);
2004   int *values = (int*)(msg+CmiMsgHeaderSizeBytes);
2005   values[0] = pe;
2006   values[1] = dataSize;
2007   CmiSetXHandler(msg, handlerIdx);
2008   if (dataSize) memcpy(msg+CmiMsgHeaderSizeBytes+2*sizeof(int), data, dataSize);
2009   if (CmiMyPe() == 0) {
2010     CmiReductionHandleDynamicRequest(msg);
2011   } else {
2012     /* send the request to processor 0 */
2013     CmiSetHandler(msg, CpvAccess(CmiReductionDynamicRequestHandler));
2014     CmiSyncSendAndFree(0, size, msg);
2015   }
2016 }
2017
2018 void CmiSendReduce(CmiReduction *red) {
2019   void *mergedData, *msg;
2020   int msg_size;
2021   if (!red->localContributed || red->numChildren != red->numRemoteReceived) return;
2022   mergedData = red->localData;
2023   msg_size = red->localSize;
2024   if (red->numChildren > 0) {
2025     int i, offset=0;
2026     if (red->ops.pupFn != NULL) {
2027       offset = CmiReservedHeaderSize;
2028       for (i=0; i<red->numChildren; ++i) red->remoteData[i] += offset;
2029     }
2030     mergedData = (red->ops.mergeFn)(&msg_size, red->localData, (void **)red->remoteData, red->numChildren);
2031     for (i=0; i<red->numChildren; ++i) CmiFree(red->remoteData[i] - offset);
2032   }
2033   /*CpvAccess(_reduce_num_children) = 0;*/
2034   /*CpvAccess(_reduce_received) = 0;*/
2035   msg = mergedData;
2036   if (red->parent != -1) {
2037     if (red->ops.pupFn != NULL) {
2038       pup_er p = pup_new_sizer();
2039       (red->ops.pupFn)(p, mergedData);
2040       msg_size = pup_size(p) + CmiReservedHeaderSize;
2041       pup_destroy(p);
2042       msg = CmiAlloc(msg_size);
2043       p = pup_new_toMem((void*)(((char*)msg)+CmiReservedHeaderSize));
2044       (red->ops.pupFn)(p, mergedData);
2045       pup_destroy(p);
2046       if (red->ops.deleteFn != NULL) (red->ops.deleteFn)(red->localData);
2047     }
2048     CmiSetHandler(msg, CpvAccess(CmiReductionMessageHandler));
2049     CmiSetRedID(msg, red->seqID);
2050     /*CmiPrintf("CmiSendReduce(%d): sending %d bytes to %d\n",CmiMyPe(),msg_size,CpvAccess(_reduce_parent));*/
2051     CmiSyncSendAndFree(red->parent, msg_size, msg);
2052   } else {
2053     (red->ops.destination)(msg);
2054   }
2055   CmiClearReduction(red->seqID);
2056 }
2057
2058 void *CmiReduceMergeFn_random(int *size, void *data, void** remote, int n) {
2059   return data;
2060 }
2061
2062 static void CmiGlobalReduce(void *msg, int size, CmiReduceMergeFn mergeFn, CmiReduction *red) {
2063   CmiAssert(red->localContributed == 0);
2064   red->localContributed = 1;
2065   red->localData = msg;
2066   red->localSize = size;
2067   red->numChildren = CmiNumSpanTreeChildren(CmiMyPe());
2068   red->parent = CmiSpanTreeParent(CmiMyPe());
2069   red->ops.destination = (CmiHandler)CmiGetHandlerFunction(msg);
2070   red->ops.mergeFn = mergeFn;
2071   red->ops.pupFn = NULL;
2072   /*CmiPrintf("[%d] CmiReduce::local %hd parent=%d, numChildren=%d\n",CmiMyPe(),red->seqID,red->parent,red->numChildren);*/
2073   CmiSendReduce(red);
2074 }
2075
2076 static void CmiGlobalReduceStruct(void *data, CmiReducePupFn pupFn,
2077                      CmiReduceMergeFn mergeFn, CmiHandler dest,
2078                      CmiReduceDeleteFn deleteFn, CmiReduction *red) {
2079   CmiAssert(red->localContributed == 0);
2080   red->localContributed = 1;
2081   red->localData = data;
2082   red->localSize = 0;
2083   red->numChildren = CmiNumSpanTreeChildren(CmiMyPe());
2084   red->parent = CmiSpanTreeParent(CmiMyPe());
2085   red->ops.destination = dest;
2086   red->ops.mergeFn = mergeFn;
2087   red->ops.pupFn = pupFn;
2088   red->ops.deleteFn = deleteFn;
2089   /*CmiPrintf("[%d] CmiReduceStruct::local %hd parent=%d, numChildren=%d\n",CmiMyPe(),red->seqID,red->parent,red->numChildren);*/
2090   CmiSendReduce(red);
2091 }
2092
2093 void CmiReduce(void *msg, int size, CmiReduceMergeFn mergeFn) {
2094   CmiReduction *red = CmiGetNextReduction(CmiNumSpanTreeChildren(CmiMyPe()));
2095   CmiGlobalReduce(msg, size, mergeFn, red);
2096 }
2097
2098 void CmiReduceStruct(void *data, CmiReducePupFn pupFn,
2099                      CmiReduceMergeFn mergeFn, CmiHandler dest,
2100                      CmiReduceDeleteFn deleteFn) {
2101   CmiReduction *red = CmiGetNextReduction(CmiNumSpanTreeChildren(CmiMyPe()));
2102   CmiGlobalReduceStruct(data, pupFn, mergeFn, dest, deleteFn, red);
2103 }
2104
2105 void CmiReduceID(void *msg, int size, CmiReduceMergeFn mergeFn, CmiReductionID id) {
2106   CmiReduction *red = CmiGetReductionCreate(id, CmiNumSpanTreeChildren(CmiMyPe()));
2107   CmiGlobalReduce(msg, size, mergeFn, red);
2108 }
2109
2110 void CmiReduceStructID(void *data, CmiReducePupFn pupFn,
2111                      CmiReduceMergeFn mergeFn, CmiHandler dest,
2112                      CmiReduceDeleteFn deleteFn, CmiReductionID id) {
2113   CmiReduction *red = CmiGetReductionCreate(id, CmiNumSpanTreeChildren(CmiMyPe()));
2114   CmiGlobalReduceStruct(data, pupFn, mergeFn, dest, deleteFn, red);
2115 }
2116
2117 void CmiListReduce(int npes, int *pes, void *msg, int size, CmiReduceMergeFn mergeFn, CmiReductionID id) {
2118   CmiReduction *red = CmiGetReductionCreate(id, CmiNumSpanTreeChildren(CmiMyPe()));
2119   int myPos;
2120   CmiAssert(red->localContributed == 0);
2121   red->localContributed = 1;
2122   red->localData = msg;
2123   red->localSize = size;
2124   for (myPos=0; myPos<npes; ++myPos) {
2125     if (pes[myPos] == CmiMyPe()) break;
2126   }
2127   CmiAssert(myPos < npes);
2128   red->numChildren = npes - (myPos << 2) - 1;
2129   if (red->numChildren > 4) red->numChildren = 4;
2130   if (red->numChildren < 0) red->numChildren = 0;
2131   if (myPos == 0) red->parent = -1;
2132   else red->parent = pes[(myPos - 1) >> 2];
2133   red->ops.destination = (CmiHandler)CmiGetHandlerFunction(msg);
2134   red->ops.mergeFn = mergeFn;
2135   red->ops.pupFn = NULL;
2136   /*CmiPrintf("[%d] CmiListReduce::local %hd parent=%d, numChildren=%d\n",CmiMyPe(),red->seqID,red->parent,red->numChildren);*/
2137   CmiSendReduce(red);
2138 }
2139
2140 void CmiListReduceStruct(int npes, int *pes,
2141                      void *data, CmiReducePupFn pupFn,
2142                      CmiReduceMergeFn mergeFn, CmiHandler dest,
2143                      CmiReduceDeleteFn deleteFn, CmiReductionID id) {
2144   CmiReduction *red = CmiGetReductionCreate(id, CmiNumSpanTreeChildren(CmiMyPe()));
2145   int myPos;
2146   CmiAssert(red->localContributed == 0);
2147   red->localContributed = 1;
2148   red->localData = data;
2149   red->localSize = 0;
2150   for (myPos=0; myPos<npes; ++myPos) {
2151     if (pes[myPos] == CmiMyPe()) break;
2152   }
2153   CmiAssert(myPos < npes);
2154   red->numChildren = npes - (myPos << 2) - 1;
2155   if (red->numChildren > 4) red->numChildren = 4;
2156   if (red->numChildren < 0) red->numChildren = 0;
2157   red->parent = (myPos - 1) >> 2;
2158   if (myPos == 0) red->parent = -1;
2159   red->ops.destination = dest;
2160   red->ops.mergeFn = mergeFn;
2161   red->ops.pupFn = pupFn;
2162   red->ops.deleteFn = deleteFn;
2163   CmiSendReduce(red);
2164 }
2165
2166 void CmiGroupReduce(CmiGroup grp, void *msg, int size, CmiReduceMergeFn mergeFn, CmiReductionID id) {
2167   int npes, *pes;
2168   CmiLookupGroup(grp, &npes, &pes);
2169   CmiListReduce(npes, pes, msg, size, mergeFn, id);
2170 }
2171
2172 void CmiGroupReduceStruct(CmiGroup grp, void *data, CmiReducePupFn pupFn,
2173                      CmiReduceMergeFn mergeFn, CmiHandler dest,
2174                      CmiReduceDeleteFn deleteFn, CmiReductionID id) {
2175   int npes, *pes;
2176   CmiLookupGroup(grp, &npes, &pes);
2177   CmiListReduceStruct(npes, pes, data, pupFn, mergeFn, dest, deleteFn, id);
2178 }
2179
2180 void CmiNodeReduce(void *data, int size, CmiReduceMergeFn mergeFn, int redID, int numChildren, int parent) {
2181   CmiAbort("Feel free to implement CmiNodeReduce...");
2182   /*
2183   CmiAssert(CmiRankOf(CmiMyPe()) == 0);
2184   CpvAccess(_reduce_data) = data;
2185   CpvAccess(_reduce_data_size) = size;
2186   CpvAccess(_reduce_parent) = CmiNodeFirst(CmiNodeSpanTreeParent(CmiMyNode()));
2187   _reduce_destination = (CmiHandler)CmiGetHandlerFunction(data);
2188   _reduce_pupFn = NULL;
2189   _reduce_mergeFn = mergeFn;
2190   CpvAccess(_reduce_num_children) = CmiNumNodeSpanTreeChildren(CmiMyNode());
2191   if (CpvAccess(_reduce_received) == CpvAccess(_reduce_num_children)) CmiSendReduce(size);
2192   */
2193 }
2194 #if 0
2195 void CmiNodeReduce(void *data, int size, void * (*mergeFn)(void*,void**,int), int redID) {
2196   CmiNodeReduce(data, size, mergeFn, redID, CmiNumNodeSpanTreeChildren(CmiMyNode()),
2197       CmiNodeFirst(CmiNodeSpanTreeParent(CmiMyNode())));
2198 }
2199 void CmiNodeReduce(void *data, int size, void * (*mergeFn)(void*,void**,int), int numChildren, int parent) {
2200   CmiNodeReduce(data, size, mergeFn, CmiReduceNextID(), numChildren, parent);
2201 }
2202 void CmiNodeReduce(void *data, int size, void * (*mergeFn)(void*,void**,int)) {
2203   CmiNodeReduce(data, size, mergeFn, CmiReduceNextID(), CmiNumNodeSpanTreeChildren(CmiMyNode()),
2204       CmiNodeFirst(CmiNodeSpanTreeParent(CmiMyNode())));
2205 }
2206 #endif
2207
2208 void CmiNodeReduceStruct(void *data, CmiReducePupFn pupFn,
2209                          CmiReduceMergeFn mergeFn, CmiHandler dest,
2210                          CmiReduceDeleteFn deleteFn) {
2211   CmiAbort("Feel free to implement CmiNodeReduceStruct...");
2212 /*
2213   CmiAssert(CmiRankOf(CmiMyPe()) == 0);
2214   CpvAccess(_reduce_data) = data;
2215   CpvAccess(_reduce_parent) = CmiNodeFirst(CmiNodeSpanTreeParent(CmiMyNode()));
2216   _reduce_destination = dest;
2217   _reduce_pupFn = pupFn;
2218   _reduce_mergeFn = mergeFn;
2219   _reduce_deleteFn = deleteFn;
2220   CpvAccess(_reduce_num_children) = CmiNumNodeSpanTreeChildren(CmiMyNode());
2221   if (CpvAccess(_reduce_received) == CpvAccess(_reduce_num_children)) CmiSendReduce(0);
2222   */
2223 }
2224
2225 void CmiHandleReductionMessage(void *msg) {
2226   CmiReduction *red = CmiGetReduction(CmiGetRedID(msg));
2227   if (red->numRemoteReceived == red->numChildren) red = CmiGetReductionCreate(CmiGetRedID(msg), red->numChildren+4);
2228   red->remoteData[red->numRemoteReceived++] = msg;
2229   /*CmiPrintf("[%d] CmiReduce::remote %hd\n",CmiMyPe(),red->seqID);*/
2230   CmiSendReduce(red);
2231 /*
2232   CpvAccess(_reduce_msg_list)[CpvAccess(_reduce_received)++] = msg;
2233   if (CpvAccess(_reduce_received) == CpvAccess(_reduce_num_children)) CmiSendReduce();
2234   / *else CmiPrintf("CmiHandleReductionMessage(%d): %d - %d\n",CmiMyPe(),CpvAccess(_reduce_received),CpvAccess(_reduce_num_children));*/
2235 }
2236
2237 void CmiReductionsInit() {
2238   int i;
2239   CpvInitialize(int, CmiReductionMessageHandler);
2240   CpvAccess(CmiReductionMessageHandler) = CmiRegisterHandler((CmiHandler)CmiHandleReductionMessage);
2241   CpvInitialize(int, CmiReductionDynamicRequestHandler);
2242   CpvAccess(CmiReductionDynamicRequestHandler) = CmiRegisterHandler((CmiHandler)CmiReductionHandleDynamicRequest);
2243   CpvInitialize(CmiUInt2, _reduce_seqID_global);
2244   CpvAccess(_reduce_seqID_global) = CmiReductionID_globalOffset;
2245   CpvInitialize(CmiUInt2, _reduce_seqID_request);
2246   CpvAccess(_reduce_seqID_request) = CmiReductionID_requestOffset;
2247   CpvInitialize(CmiUInt2, _reduce_seqID_dynamic);
2248   CpvAccess(_reduce_seqID_dynamic) = CmiReductionID_dynamicOffset;
2249   CpvInitialize(int, _reduce_info_size);
2250   CpvAccess(_reduce_info_size) = 4;
2251   CpvInitialize(CmiReduction**, _reduce_info);
2252   CpvAccess(_reduce_info) = malloc(16*sizeof(CmiReduction*));
2253   for (i=0; i<16; ++i) CpvAccess(_reduce_info)[i] = NULL;
2254 }
2255
2256 /*****************************************************************************
2257  *
2258  * Multicast groups
2259  *
2260  ****************************************************************************/
2261
2262 #if CMK_MULTICAST_DEF_USE_COMMON_CODE
2263
2264 typedef struct GroupDef
2265 {
2266   union {
2267     char core[CmiMsgHeaderSizeBytes];
2268     struct GroupDef *next;
2269   } core;
2270   CmiGroup group;
2271   int npes;
2272   int pes[1];
2273 }
2274 *GroupDef;
2275
2276 #define GROUPTAB_SIZE 101
2277
2278 CpvStaticDeclare(int, CmiGroupHandlerIndex);
2279 CpvStaticDeclare(int, CmiGroupCounter);
2280 CpvStaticDeclare(GroupDef *, CmiGroupTable);
2281
2282 void CmiGroupHandler(GroupDef def)
2283 {
2284   /* receive group definition, insert into group table */
2285   GroupDef *table = CpvAccess(CmiGroupTable);
2286   unsigned int hashval, bucket;
2287   hashval = (def->group.id ^ def->group.pe);
2288   bucket = hashval % GROUPTAB_SIZE;
2289   def->core.next = table[bucket];
2290   table[bucket] = def;
2291 }
2292
2293 CmiGroup CmiEstablishGroup(int npes, int *pes)
2294 {
2295   /* build new group definition, broadcast it */
2296   CmiGroup grp; GroupDef def; int len, i;
2297   grp.id = CpvAccess(CmiGroupCounter)++;
2298   grp.pe = CmiMyPe();
2299   len = sizeof(struct GroupDef)+(npes*sizeof(int));
2300   def = (GroupDef)CmiAlloc(len);
2301   def->group = grp;
2302   def->npes = npes;
2303   for (i=0; i<npes; i++)
2304     def->pes[i] = pes[i];
2305   CmiSetHandler(def, CpvAccess(CmiGroupHandlerIndex));
2306   CmiSyncBroadcastAllAndFree(len, def);
2307   return grp;
2308 }
2309
2310 void CmiLookupGroup(CmiGroup grp, int *npes, int **pes)
2311 {
2312   unsigned int hashval, bucket;  GroupDef def;
2313   GroupDef *table = CpvAccess(CmiGroupTable);
2314   hashval = (grp.id ^ grp.pe);
2315   bucket = hashval % GROUPTAB_SIZE;
2316   for (def=table[bucket]; def; def=def->core.next) {
2317     if ((def->group.id == grp.id)&&(def->group.pe == grp.pe)) {
2318       *npes = def->npes;
2319       *pes = def->pes;
2320       return;
2321     }
2322   }
2323   *npes = 0; *pes = 0;
2324 }
2325
2326 void CmiGroupInit()
2327 {
2328   CpvInitialize(int, CmiGroupHandlerIndex);
2329   CpvInitialize(int, CmiGroupCounter);
2330   CpvInitialize(GroupDef *, CmiGroupTable);
2331   CpvAccess(CmiGroupHandlerIndex) = CmiRegisterHandler((CmiHandler)CmiGroupHandler);
2332   CpvAccess(CmiGroupCounter) = 0;
2333   CpvAccess(CmiGroupTable) =
2334     (GroupDef*)calloc(GROUPTAB_SIZE, sizeof(GroupDef));
2335   if (CpvAccess(CmiGroupTable) == 0)
2336     CmiAbort("Memory Allocation Error");
2337 }
2338
2339 #endif
2340
2341 /*****************************************************************************
2342  *
2343  * Common List-Cast and Multicast Code
2344  *
2345  ****************************************************************************/
2346
2347 #if CMK_MULTICAST_LIST_USE_COMMON_CODE
2348
2349 void CmiSyncListSendFn(int npes, int *pes, int len, char *msg)
2350 {
2351   int i;
2352   for(i=0;i<npes;i++) {
2353     CmiSyncSend(pes[i], len, msg);
2354   }
2355 }
2356
2357 CmiCommHandle CmiAsyncListSendFn(int npes, int *pes, int len, char *msg)
2358 {
2359   /* A better asynchronous implementation may be wanted, but at least it works */
2360   CmiSyncListSendFn(npes, pes, len, msg);
2361   return (CmiCommHandle) 0;
2362 }
2363
2364 void CmiFreeListSendFn(int npes, int *pes, int len, char *msg)
2365 {
2366   int i;
2367   for(i=0;i<npes-1;i++) {
2368     CmiSyncSend(pes[i], len, msg);
2369   }
2370   if (npes>0)
2371     CmiSyncSendAndFree(pes[npes-1], len, msg);
2372   else 
2373     CmiFree(msg);
2374 }
2375
2376 #endif
2377
2378 #if CMK_MULTICAST_GROUP_USE_COMMON_CODE
2379
2380 typedef struct MultiMsg
2381 {
2382   char core[CmiMsgHeaderSizeBytes];
2383   CmiGroup group;
2384   int pos;
2385   int origlen;
2386 }
2387 *MultiMsg;
2388
2389
2390 CpvDeclare(int, CmiMulticastHandlerIndex);
2391
2392 void CmiMulticastDeliver(MultiMsg msg)
2393 {
2394   int npes, *pes; int olen, nlen, pos, child1, child2;
2395   olen = msg->origlen;
2396   nlen = olen + sizeof(struct MultiMsg);
2397   CmiLookupGroup(msg->group, &npes, &pes);
2398   if (pes==0) {
2399     CmiSyncSendAndFree(CmiMyPe(), nlen, msg);
2400     return;
2401   }
2402   if (npes==0) {
2403     CmiFree(msg);
2404     return;
2405   }
2406   if (msg->pos == -1) {
2407     msg->pos=0;
2408     CmiSyncSendAndFree(pes[0], nlen, msg);
2409     return;
2410   }
2411   pos = msg->pos;
2412   child1 = ((pos+1)<<1);
2413   child2 = child1-1;
2414   if (child1 < npes) {
2415     msg->pos = child1;
2416     CmiSyncSend(pes[child1], nlen, msg);
2417   }
2418   if (child2 < npes) {
2419     msg->pos = child2;
2420     CmiSyncSend(pes[child2], nlen, msg);
2421   }
2422   if(olen < sizeof(struct MultiMsg)) {
2423     memcpy(msg, msg+1, olen);
2424   } else {
2425     memcpy(msg, (((char*)msg)+olen), sizeof(struct MultiMsg));
2426   }
2427   CmiSyncSendAndFree(CmiMyPe(), olen, msg);
2428 }
2429
2430 void CmiMulticastHandler(MultiMsg msg)
2431 {
2432   CmiMulticastDeliver(msg);
2433 }
2434
2435 void CmiSyncMulticastFn(CmiGroup grp, int len, char *msg)
2436 {
2437   int newlen; MultiMsg newmsg;
2438   newlen = len + sizeof(struct MultiMsg);
2439   newmsg = (MultiMsg)CmiAlloc(newlen);
2440   if(len < sizeof(struct MultiMsg)) {
2441     memcpy(newmsg+1, msg, len);
2442   } else {
2443     memcpy(newmsg+1, msg+sizeof(struct MultiMsg), len-sizeof(struct MultiMsg));
2444     memcpy(((char *)newmsg+len), msg, sizeof(struct MultiMsg));
2445   }
2446   newmsg->group = grp;
2447   newmsg->origlen = len;
2448   newmsg->pos = -1;
2449   CmiSetHandler(newmsg, CpvAccess(CmiMulticastHandlerIndex));
2450   CmiMulticastDeliver(newmsg);
2451 }
2452
2453 void CmiFreeMulticastFn(CmiGroup grp, int len, char *msg)
2454 {
2455   CmiSyncMulticastFn(grp, len, msg);
2456   CmiFree(msg);
2457 }
2458
2459 CmiCommHandle CmiAsyncMulticastFn(CmiGroup grp, int len, char *msg)
2460 {
2461   CmiError("Async Multicast not implemented.");
2462   return (CmiCommHandle) 0;
2463 }
2464
2465 void CmiMulticastInit()
2466 {
2467   CpvInitialize(int, CmiMulticastHandlerIndex);
2468   CpvAccess(CmiMulticastHandlerIndex) =
2469     CmiRegisterHandler((CmiHandler)CmiMulticastHandler);
2470 }
2471
2472 #endif
2473
2474 #if CONVERSE_VERSION_SHMEM && CMK_ARENA_MALLOC
2475 extern void *arena_malloc(int size);
2476 extern void arena_free(void *blockPtr);
2477 #endif
2478
2479 /***************************************************************************
2480  *
2481  * Memory Allocation routines 
2482  *
2483  * A block of memory can consist of multiple chunks.  Each chunk has
2484  * a sizefield and a refcount.  The first chunk's refcount is a reference
2485  * count.  That's how many CmiFrees it takes to free the message.
2486  * Subsequent chunks have a refcount which is less than zero.  This is
2487  * the offset back to the start of the first chunk.
2488  *
2489  * Each chunk has a CmiChunkHeader before the user data, with the fields:
2490  *
2491  *  size: The user-allocated size of the chunk, in bytes.
2492  *
2493  *  ref: A magic reference count object. Ordinary blocks start with
2494  *     reference count 1.  When the reference count reaches zero,
2495  *     the block is deleted.  To support nested buffers, the 
2496  *     reference count can also be negative, which means it is 
2497  *     a byte offset to the enclosing buffer's reference count.
2498  *
2499  ***************************************************************************/
2500
2501
2502 void *CmiAlloc(int size)
2503 {
2504   char *res;
2505
2506 #if CONVERSE_VERSION_ELAN
2507   res = (char *) elan_CmiAlloc(size+sizeof(CmiChunkHeader));
2508 #elif CONVERSE_VERSION_VMI
2509   res = (char *) CMI_VMI_CmiAlloc(size+sizeof(CmiChunkHeader));
2510 #elif CONVERSE_VERSION_SHMEM && CMK_ARENA_MALLOC
2511   res = (char*) arena_malloc(size+sizeof(CmiChunkHeader));
2512 #elif CMK_USE_IBVERBS | CMK_USE_IBUD
2513   res = (char *) infi_CmiAlloc(size+sizeof(CmiChunkHeader));
2514 #elif CONVERSE_POOL
2515   res =(char *) CmiPoolAlloc(size+sizeof(CmiChunkHeader));
2516 #else
2517   res =(char *) malloc_nomigrate(size+sizeof(CmiChunkHeader));
2518 #endif
2519
2520   _MEMCHECK(res);
2521
2522 #ifdef MEMMONITOR
2523   CpvAccess(MemoryUsage) += size+sizeof(CmiChunkHeader);
2524   CpvAccess(AllocCount)++;
2525   CpvAccess(BlocksAllocated)++;
2526   if (CpvAccess(MemoryUsage) > CpvAccess(HiWaterMark)) {
2527     CpvAccess(HiWaterMark) = CpvAccess(MemoryUsage);
2528   }
2529   if (CpvAccess(MemoryUsage) > 1.1 * CpvAccess(ReportedHiWaterMark)) {
2530     CmiPrintf("HIMEM STAT PE%d: %d Allocs, %d blocks, %lu K, Max %lu K\n",
2531             CmiMyPe(), CpvAccess(AllocCount), CpvAccess(BlocksAllocated),
2532             CpvAccess(MemoryUsage)/1024, CpvAccess(HiWaterMark)/1024);
2533     CpvAccess(ReportedHiWaterMark) = CpvAccess(MemoryUsage);
2534   }
2535   if ((CpvAccess(AllocCount) % 1000) == 0) {
2536     CmiPrintf("MEM STAT PE%d: %d Allocs, %d blocks, %lu K, Max %lu K\n",
2537             CmiMyPe(), CpvAccess(AllocCount), CpvAccess(BlocksAllocated),
2538             CpvAccess(MemoryUsage)/1024, CpvAccess(HiWaterMark)/1024);
2539   }
2540 #endif
2541
2542   res+=sizeof(CmiChunkHeader);
2543   SIZEFIELD(res)=size;
2544   REFFIELD(res)=1;
2545   return (void *)res;
2546 }
2547
2548 /** Follow the header links out to the most enclosing block */
2549 static void *CmiAllocFindEnclosing(void *blk) {
2550   int refCount = REFFIELD(blk);
2551   while (refCount < 0) {
2552     blk = (void *)((char*)blk+refCount); /* Jump to enclosing block */
2553     refCount = REFFIELD(blk);
2554   }
2555   return blk;
2556 }
2557
2558 /** Increment the reference count for this block's owner.
2559     This call must be matched by an equivalent CmiFree. */
2560 void CmiReference(void *blk)
2561 {
2562   REFFIELD(CmiAllocFindEnclosing(blk))++;
2563 }
2564
2565 /** Return the size of the user portion of this block. */
2566 int CmiSize(void *blk)
2567 {
2568   return SIZEFIELD(blk);
2569 }
2570
2571 /** Decrement the reference count for this block. */
2572 void CmiFree(void *blk)
2573 {
2574   void *parentBlk=CmiAllocFindEnclosing(blk);
2575   int refCount=REFFIELD(parentBlk);
2576 #ifndef CMK_OPTIMIZE
2577   if(refCount==0) /* Logic error: reference count shouldn't already have been zero */
2578     CmiAbort("CmiFree reference count was zero-- is this a duplicate free?");
2579 #endif
2580   refCount--;
2581   REFFIELD(parentBlk) = refCount;
2582   if(refCount==0) { /* This was the last reference to the block-- free it */
2583 #ifdef MEMMONITOR
2584     int size=SIZEFIELD(parentBlk);
2585     if (size > 1000000000) /* Absurdly large size field-- warning */
2586       CmiPrintf("MEMSTAT Uh-oh -- SIZEFIELD=%d\n",size);
2587     CpvAccess(MemoryUsage) -= (size + sizeof(CmiChunkHeader));
2588     CpvAccess(BlocksAllocated)--;
2589 #endif
2590
2591 #if CONVERSE_VERSION_ELAN
2592     elan_CmiFree(BLKSTART(parentBlk));
2593 #elif CONVERSE_VERSION_VMI
2594     CMI_VMI_CmiFree(BLKSTART(parentBlk));
2595 #elif CONVERSE_VERSION_SHMEM && CMK_ARENA_MALLOC
2596     arena_free(BLKSTART(parentBlk));
2597 #elif CMK_USE_IBVERBS | CMK_USE_IBUD
2598     /* is this message the head of a MultipleSend that we received?
2599        Then the parts with INFIMULTIPOOL have metadata which must be 
2600        unregistered and freed.  */
2601 #ifdef CMK_IBVERS_CLEAN_MULTIPLESEND
2602     if(CmiGetHandler(parentBlk)==CpvAccess(CmiMainHandlerIDP))
2603       {
2604         infi_freeMultipleSend(parentBlk);
2605       }
2606 #endif
2607     infi_CmiFree(BLKSTART(parentBlk));
2608 #elif CONVERSE_POOL
2609     CmiPoolFree(BLKSTART(parentBlk));
2610 #else
2611     free_nomigrate(BLKSTART(parentBlk));
2612 #endif
2613   }
2614 }
2615
2616
2617 /***************************************************************************
2618  *
2619  * Temporary-memory Allocation routines 
2620  *
2621  *  This buffer augments the storage available on the regular machine stack
2622  * for fairly large temporary buffers, which allows us to use smaller machine
2623  * stacks.
2624  *
2625  ***************************************************************************/
2626
2627 #define CMI_TMP_BUF_MAX 128*1024 /* Allow this much temporary storage. */
2628
2629 typedef struct {
2630   char *buf; /* Start of temporary buffer */
2631   int cur; /* First unused location in temporary buffer */
2632   int max; /* Length of temporary buffer */
2633 } CmiTmpBuf_t;
2634 CpvDeclare(CmiTmpBuf_t,CmiTmpBuf); /* One temporary buffer per PE */
2635
2636 static void CmiTmpSetup(CmiTmpBuf_t *b) {
2637   b->buf=malloc(CMI_TMP_BUF_MAX);
2638   b->cur=0;
2639   b->max=CMI_TMP_BUF_MAX;
2640 }
2641
2642 void *CmiTmpAlloc(int size) {
2643   if (!CpvInitialized(CmiTmpBuf)) {
2644     return malloc(size);
2645   }
2646   else { /* regular case */
2647     CmiTmpBuf_t *b=&CpvAccess(CmiTmpBuf);
2648     void *t;
2649     if (b->cur+size>b->max) {
2650       if (b->max==0) /* We're just uninitialized */
2651         CmiTmpSetup(b);
2652       else /* We're really out of space! */
2653         CmiAbort("CmiTmpAlloc: asked for too much temporary buffer space");
2654     }
2655     t=b->buf+b->cur;
2656     b->cur+=size;
2657     return t;
2658   }
2659 }
2660 void CmiTmpFree(void *t) {
2661   if (!CpvInitialized(CmiTmpBuf)) {
2662     free(t);
2663   }
2664   else { /* regular case */
2665     CmiTmpBuf_t *b=&CpvAccess(CmiTmpBuf);
2666     /* t should point into our temporary buffer: figure out where */
2667     int cur=((const char *)t)-b->buf;
2668 #ifndef CMK_OPTIMIZE
2669     if (cur<0 || cur>b->max)
2670       CmiAbort("CmiTmpFree: called with an invalid pointer");
2671 #endif
2672     b->cur=cur;
2673   }
2674 }
2675
2676 void CmiTmpInit(char **argv) {
2677   CpvInitialize(CmiTmpBuf_t,CmiTmpBuf);
2678   /* Set up this processor's temporary buffer */
2679   CmiTmpSetup(&CpvAccess(CmiTmpBuf));
2680 }
2681
2682 /******************************************************************************
2683
2684   Cross-platform directory creation
2685
2686   ****************************************************************************/
2687 #ifdef _MSC_VER
2688 /* Windows directory creation: */
2689 #include <windows.h>
2690
2691 void CmiMkdir(const char *dirName) {
2692         CreateDirectory(dirName,NULL);
2693 }
2694
2695 #else /* !_MSC_VER */
2696 /* UNIX directory creation */
2697 #include <unistd.h> 
2698 #include <sys/stat.h> /* from "mkdir" man page */
2699 #include <sys/types.h>
2700
2701 void CmiMkdir(const char *dirName) {
2702 #ifndef __MINGW_H
2703         mkdir(dirName,0777);
2704 #else
2705         mkdir(dirName);
2706 #endif
2707 }
2708
2709 #endif
2710
2711
2712 /******************************************************************************
2713
2714   Multiple Send function                               
2715
2716   ****************************************************************************/
2717
2718
2719
2720
2721
2722 /****************************************************************************
2723 * DESCRIPTION : This function call allows the user to send multiple messages
2724 *               from one processor to another, all intended for differnet 
2725 *               handlers.
2726 *
2727 *               Parameters :
2728 *
2729 *               destPE, len, int sizes[0..len-1], char *messages[0..len-1]
2730 *
2731 ****************************************************************************/
2732 /* Round up message size to the message granularity. 
2733    Does this by adding, then truncating.
2734 */
2735 static int roundUpSize(unsigned int s) {
2736   return (int)((s+sizeof(double)-1)&~(sizeof(double)-1));
2737 }
2738 /* Return the amount of message padding required for a message
2739    with this many user bytes. 
2740  */
2741 static int paddingSize(unsigned int s) {
2742   return roundUpSize(s)-s;
2743 }
2744
2745 /* Message header for a bundle of multiple-sent messages */
2746 typedef struct {
2747   char convHeader[CmiMsgHeaderSizeBytes];
2748   int nMessages; /* Number of distinct messages bundled below. */
2749   double pad; /* To align the first message, which follows this header */
2750 } CmiMultipleSendHeader;
2751
2752 #if CMK_USE_IBVERBS | CMK_USE_IBUD
2753 /* given a pointer to a multisend message clean up the metadata */
2754
2755 void infi_freeMultipleSend(void *msgWhole)
2756 {
2757   int len=((CmiMultipleSendHeader *)msgWhole)->nMessages;
2758   double pad=((CmiMultipleSendHeader *)msgWhole)->pad;
2759   int offset=sizeof(CmiMultipleSendHeader);
2760   int m;
2761   void *thisMsg=NULL;
2762   if (pad != 1234567.89) return;
2763   for(m=0;m<len;m++)
2764     {
2765       /*unreg meta, free meta, move the ptr */
2766       /* note these weird little things are not pooled */
2767       /* do NOT free the message here, we are only a part of this buffer*/
2768       infiCmiChunkHeader *ch=(infiCmiChunkHeader *)(msgWhole+offset);
2769       char *msg=(msgWhole+offset+sizeof(infiCmiChunkHeader));
2770       int msgSize=ch->chunkHeader.size; /* Size of user portion of message (plus padding at end) */
2771       infi_unregAndFreeMeta(ch->metaData);
2772       offset+= sizeof(infiCmiChunkHeader) + msgSize;
2773     }
2774 }
2775 #endif
2776
2777
2778 static void _CmiMultipleSend(unsigned int destPE, int len, int sizes[], char *msgComps[], int immed)
2779 {
2780   CmiMultipleSendHeader header;
2781   int m; /* Outgoing message */
2782
2783 #if CMK_USE_IBVERBS
2784   infiCmiChunkHeader *msgHdr;
2785 #else
2786   CmiChunkHeader *msgHdr; /* Chunk headers for each message */
2787 #endif
2788         
2789   double pad = 0; /* padding required */
2790   int vecLen; /* Number of pieces in outgoing message vector */
2791   int *vecSizes; /* Sizes of each piece we're sending out. */
2792   char **vecPtrs; /* Pointers to each piece we're sending out. */
2793   int vec; /* Entry we're currently filling out in above array */
2794         
2795 #if CMK_USE_IBVERBS
2796   msgHdr = (infiCmiChunkHeader *)CmiTmpAlloc(len * sizeof(infiCmiChunkHeader));
2797 #else
2798   msgHdr = (CmiChunkHeader *)CmiTmpAlloc(len * sizeof(CmiChunkHeader));
2799 #endif
2800         
2801   /* Allocate memory for the outgoing vector*/
2802   vecLen=1+3*len; /* Header and 3 parts per message */
2803   vecSizes = (int *)CmiTmpAlloc(vecLen * sizeof(int));
2804   vecPtrs = (char **)CmiTmpAlloc(vecLen * sizeof(char *));
2805   vec=0;
2806   
2807   /* Build the header */
2808   header.nMessages=len;
2809   CmiSetHandler(&header, CpvAccess(CmiMainHandlerIDP));
2810   header.pad = 1234567.89;
2811 #if CMK_IMMEDIATE_MSG
2812   if (immed) CmiBecomeImmediate(&header);
2813 #endif
2814   vecSizes[vec]=sizeof(header); vecPtrs[vec]=(char *)&header;
2815   vec++;
2816
2817   /* Build an entry for each message: 
2818          | CmiChunkHeader | Message data | Message padding | ...next message entry ...
2819   */
2820   for (m=0;m<len;m++) {
2821 #if CMK_USE_IBVERBS
2822     msgHdr[m].chunkHeader.size=roundUpSize(sizes[m]); /* Size of message and padding */
2823     msgHdr[m].chunkHeader.ref=0; /* Reference count will be filled out on receive side */
2824     msgHdr[m].metaData=NULL;
2825 #else
2826     msgHdr[m].size=roundUpSize(sizes[m]); /* Size of message and padding */
2827     msgHdr[m].ref=0; /* Reference count will be filled out on receive side */
2828 #endif          
2829     
2830     /* First send the message's CmiChunkHeader (for use on receive side) */
2831 #if CMK_USE_IBVERBS
2832     vecSizes[vec]=sizeof(infiCmiChunkHeader);
2833 #else
2834     vecSizes[vec]=sizeof(CmiChunkHeader); 
2835 #endif          
2836                 vecPtrs[vec]=(char *)&msgHdr[m];
2837     vec++;
2838     
2839     /* Now send the actual message data */
2840     vecSizes[vec]=sizes[m]; vecPtrs[vec]=msgComps[m];
2841     vec++;
2842     
2843     /* Now send padding to align the next message on a double-boundary */
2844     vecSizes[vec]=paddingSize(sizes[m]); vecPtrs[vec]=(char *)&pad;
2845     vec++;
2846   }
2847   CmiAssert(vec==vecLen);
2848   
2849   CmiSyncVectorSend(destPE, vecLen, vecSizes, vecPtrs);
2850   
2851   CmiTmpFree(vecPtrs); /* CmiTmp: Be sure to throw away in opposite order of allocation */
2852   CmiTmpFree(vecSizes);
2853   CmiTmpFree(msgHdr);
2854 }
2855
2856 void CmiMultipleSend(unsigned int destPE, int len, int sizes[], char *msgComps[])
2857 {
2858   _CmiMultipleSend(destPE, len, sizes, msgComps, 0);
2859 }
2860
2861 void CmiMultipleIsend(unsigned int destPE, int len, int sizes[], char *msgComps[])
2862 {
2863   _CmiMultipleSend(destPE, len, sizes, msgComps, 1);
2864 }
2865
2866 /****************************************************************************
2867 * DESCRIPTION : This function initializes the main handler required for the
2868 *               CmiMultipleSend() function to work. 
2869 *               
2870 *               This function should be called once in any Converse program
2871 *               that uses CmiMultipleSend()
2872 *
2873 ****************************************************************************/
2874
2875 static void CmiMultiMsgHandler(char *msgWhole);
2876
2877 void CmiInitMultipleSend(void)
2878 {
2879   CpvInitialize(int,CmiMainHandlerIDP); 
2880   CpvAccess(CmiMainHandlerIDP) =
2881     CmiRegisterHandler((CmiHandler)CmiMultiMsgHandler);
2882 }
2883
2884 /****************************************************************************
2885 * DESCRIPTION : This function is the main handler that splits up the messages
2886 *               CmiMultipleSend() pastes together. 
2887 *
2888 ****************************************************************************/
2889
2890 static void CmiMultiMsgHandler(char *msgWhole)
2891 {
2892   int len=((CmiMultipleSendHeader *)msgWhole)->nMessages;
2893   int offset=sizeof(CmiMultipleSendHeader);
2894   int m;
2895   for (m=0;m<len;m++) {
2896 #if CMK_USE_IBVERBS
2897     infiCmiChunkHeader *ch=(infiCmiChunkHeader *)(msgWhole+offset);
2898     char *msg=(msgWhole+offset+sizeof(infiCmiChunkHeader));
2899     int msgSize=ch->chunkHeader.size; /* Size of user portion of message (plus padding at end) */
2900     ch->chunkHeader.ref=msgWhole-msg; 
2901     ch->metaData =  registerMultiSendMesg(msg,msgSize);
2902 #else
2903     CmiChunkHeader *ch=(CmiChunkHeader *)(msgWhole+offset);
2904     char *msg=(msgWhole+offset+sizeof(CmiChunkHeader));
2905     int msgSize=ch->size; /* Size of user portion of message (plus padding at end) */
2906     ch->ref=msgWhole-msg; 
2907 #endif          
2908     /* Link new message to owner via a negative ref pointer */
2909     CmiReference(msg); /* Follows link & increases reference count of *msgWhole* */
2910     CmiSyncSendAndFree(CmiMyPe(), msgSize, msg);
2911 #if CMK_USE_IBVERBS
2912     offset+= sizeof(infiCmiChunkHeader) + msgSize;
2913 #else
2914     offset+= sizeof(CmiChunkHeader) + msgSize;
2915 #endif          
2916   }
2917   /* Release our reference to the whole message.  The message will
2918      only actually be deleted once all its sub-messages are free'd as well. */
2919   CmiFree(msgWhole);
2920 }
2921
2922 /****************************************************************************
2923 * Hypercube broadcast message passing.
2924 ****************************************************************************/
2925
2926 int HypercubeGetBcastDestinations(int mype, int total_pes, int k, int *dest_pes) {
2927   int num_pes = 0;
2928   for ( ; k>=0; --k) {
2929     /* add the processor destination at level k if it exist */
2930     dest_pes[num_pes] = mype ^ (1<<k);
2931     if (dest_pes[num_pes] >= total_pes) {
2932       /* find the first proc in the other part of the current dimention */
2933       dest_pes[num_pes] &= (-1)<<k;
2934       /* if the first proc there is over CmiNumPes() then there is no other
2935          dimension, otherwise if it is valid compute my correspondent in such
2936          a way to minimize the load for every processor */
2937       if (total_pes>dest_pes[num_pes]) dest_pes[num_pes] += (mype - (mype & ((-1)<<k))) % (total_pes - dest_pes[num_pes]);
2938       }
2939     if (dest_pes[num_pes] < total_pes) {
2940       /* if the destination is in the acceptable range increment num_pes */
2941       ++num_pes;
2942     }
2943   }
2944   return num_pes;
2945 }
2946
2947
2948 /****************************************************************************
2949 * DESCRIPTION : This function initializes the main handler required for the
2950 *               Immediate message
2951 *               
2952 *               This function should be called once in any Converse program
2953 *
2954 ****************************************************************************/
2955
2956 int _immediateLock = 0; /* if locked, all immediate message handling will be delayed. */
2957 int _immediateFlag = 0; /* if set, there is delayed immediate message. */
2958
2959 CpvDeclare(int, CmiImmediateMsgHandlerIdx); /* Main handler that is run on every node */
2960
2961 /* xdl is the real handler */
2962 static void CmiImmediateMsgHandler(char *msg)
2963 {
2964   CmiSetHandler(msg, CmiGetXHandler(msg));
2965   CmiHandleMessage(msg);
2966 }
2967
2968 void CmiInitImmediateMsg(void)
2969 {
2970   CpvInitialize(int,CmiImmediateMsgHandlerIdx); 
2971   CpvAccess(CmiImmediateMsgHandlerIdx) =
2972     CmiRegisterHandler((CmiHandler)CmiImmediateMsgHandler);
2973 }
2974
2975 /*#if !CMK_IMMEDIATE_MSG
2976 #if !CMK_MACHINE_PROGRESS_DEFINED
2977 void CmiProbeImmediateMsg()
2978 {
2979 }
2980 #endif
2981 #endif*/
2982
2983 /******** Idle timeout module (+idletimeout=30) *********/
2984
2985 typedef struct {
2986   int idle_timeout;/*Milliseconds to wait idle before aborting*/
2987   int is_idle;/*Boolean currently-idle flag*/
2988   int call_count;/*Number of timeout calls currently in flight*/
2989 } cmi_cpu_idlerec;
2990
2991 static void on_timeout(cmi_cpu_idlerec *rec,double curWallTime)
2992 {
2993   rec->call_count--;
2994   if(rec->call_count==0 && rec->is_idle==1) {
2995     CmiError("Idle time on PE %d exceeded specified timeout.\n", CmiMyPe());
2996     CmiAbort("Exiting.\n");
2997   }
2998 }
2999 static void on_idle(cmi_cpu_idlerec *rec,double curWallTime)
3000 {
3001   CcdCallFnAfter((CcdVoidFn)on_timeout, rec, rec->idle_timeout);
3002   rec->call_count++; /*Keeps track of overlapping timeout calls.*/  
3003   rec->is_idle = 1;
3004 }
3005 static void on_busy(cmi_cpu_idlerec *rec,double curWallTime)
3006 {
3007   rec->is_idle = 0;
3008 }
3009 static void CIdleTimeoutInit(char **argv)
3010 {
3011   int idle_timeout=0; /*Seconds to wait*/
3012   CmiGetArgIntDesc(argv,"+idle-timeout",&idle_timeout,"Abort if idle for this many seconds");
3013   if(idle_timeout != 0) {
3014     cmi_cpu_idlerec *rec=(cmi_cpu_idlerec *)malloc(sizeof(cmi_cpu_idlerec));
3015     _MEMCHECK(rec);
3016     rec->idle_timeout=idle_timeout*1000;
3017     rec->is_idle=0;
3018     rec->call_count=0;
3019     CcdCallOnCondition(CcdPROCESSOR_BEGIN_IDLE, (CcdVoidFn)on_idle, rec);
3020     CcdCallOnCondition(CcdPROCESSOR_BEGIN_BUSY, (CcdVoidFn)on_busy, rec);
3021   }
3022 }
3023
3024 /*****************************************************************************
3025  *
3026  * Converse Initialization
3027  *
3028  *****************************************************************************/
3029
3030 extern void CrnInit(void);
3031 extern void CmiIsomallocInit(char **argv);
3032 #if ! CMK_CMIPRINTF_IS_A_BUILTIN
3033 void CmiIOInit(char **argv);
3034 #endif
3035
3036 static void CmiProcessPriority(char **argv)
3037 {
3038   int dummy, nicelevel=-100;      /* process priority */
3039   CmiGetArgIntDesc(argv,"+nice",&nicelevel,"Set the process priority level");
3040   /* ignore others */
3041   while (CmiGetArgIntDesc(argv,"+nice",&dummy,"Set the process priority level"));
3042   /* call setpriority once on each process to set process's priority */
3043   if (CmiMyRank() == 0 && nicelevel != -100)  {
3044 #ifndef _WIN32
3045     if (0!=setpriority(PRIO_PROCESS, 0, nicelevel))  {
3046       CmiPrintf("[%d] setpriority failed with value %d. \n", CmiMyPe(), nicelevel);
3047       perror("setpriority");
3048       CmiAbort("setpriority failed.");
3049     }
3050     else
3051       CmiPrintf("[%d] Charm++: setpriority %d\n", CmiMyPe(), nicelevel);
3052 #else
3053     HANDLE hProcess = GetCurrentProcess();
3054     DWORD dwPriorityClass = NORMAL_PRIORITY_CLASS;
3055     char *prio_str = "NORMAL_PRIORITY_CLASS";
3056     BOOL status;
3057     /*
3058        <-20:      real time
3059        -20--10:   high 
3060        -10-0:     above normal
3061        0:         normal
3062        0-10:      below normal
3063        10-:       idle
3064     */
3065     if (0) ;
3066 #ifdef BELOW_NORMAL_PRIORITY_CLASS
3067     else if (nicelevel<10 && nicelevel>0) {
3068       dwPriorityClass = BELOW_NORMAL_PRIORITY_CLASS;
3069       prio_str = "BELOW_NORMAL_PRIORITY_CLASS";
3070     }
3071 #endif
3072     else if (nicelevel>0) {
3073       dwPriorityClass = IDLE_PRIORITY_CLASS;
3074       prio_str = "IDLE_PRIORITY_CLASS";
3075     }
3076     else if (nicelevel<=-20) {
3077       dwPriorityClass = REALTIME_PRIORITY_CLASS;
3078       prio_str = "REALTIME_PRIORITY_CLASS";
3079     }
3080 #ifdef ABOVE_NORMAL_PRIORITY_CLASS
3081     else if (nicelevel>-10 && nicelevel<0) {
3082       dwPriorityClass = ABOVE_NORMAL_PRIORITY_CLASS;
3083       prio_str = "ABOVE_NORMAL_PRIORITY_CLASS";
3084     }
3085 #endif
3086     else if (nicelevel<0) {
3087       dwPriorityClass = HIGH_PRIORITY_CLASS;
3088       prio_str = "HIGH_PRIORITY_CLASS";
3089     }
3090     status = SetPriorityClass(hProcess, dwPriorityClass);
3091     if (!status)  {
3092         int err=GetLastError();
3093         CmiPrintf("SetPriorityClass failed errno=%d, WSAerr=%d\n",errno, err);
3094         CmiAbort("SetPriorityClass failed.");
3095     }
3096     else
3097       CmiPrintf("[%d] Charm++: setpriority %s\n", CmiMyPe(), prio_str);
3098 #endif
3099   }
3100 }
3101
3102 void CommunicationServerInit()
3103 {
3104 #if CMK_IMMEDIATE_MSG
3105   CQdCpvInit();
3106   CpvInitialize(int,CmiImmediateMsgHandlerIdx); 
3107 #endif
3108 }
3109
3110
3111 static int testEndian(void)
3112 {
3113         int test=0x1c;
3114         unsigned char *c=(unsigned char *)&test;
3115         if (c[sizeof(int)-1]==0x1c)
3116                 /* Macintosh and most workstations are big-endian */
3117                 return 1;   /* Big-endian machine */
3118         if (c[0]==0x1c)
3119                 /* Intel x86 PC's, and DEC VAX are little-endian */
3120                 return 0;  /* Little-endian machine */
3121         return -2;  /*Unknown integer type */
3122 }
3123
3124 int CmiEndianness()
3125 {
3126   static int _cmi_endianness = -1;
3127   if (_cmi_endianness == -1) _cmi_endianness = testEndian();
3128   CmiAssert(_cmi_endianness != -2);
3129   return  _cmi_endianness;
3130 }
3131
3132 /**
3133   Main Converse initialization routine.  This routine is 
3134   called by the machine file (machine.c) to set up Converse.
3135   It's "Common" because it's shared by all the machine.c files. 
3136   
3137   The main task of this routine is to set up all the Cpv's
3138   (message queues, handler tables, etc.) used during main execution.
3139   
3140   On SMP versions, this initialization routine is called by 
3141   *all* processors of a node simultaniously.  It's *also* called
3142   by the communication thread, which is rather strange but needed
3143   for immediate messages.  Each call to this routine expects a 
3144   different copy of the argv arguments, so use CmiCopyArgs(argv).
3145   
3146   Requires:
3147     - A working network layer.
3148     - Working Cpv's and CmiNodeBarrier.
3149     - CthInit to already have been called.  CthInit is called
3150       from the machine layer directly, because some machine layers
3151       (like uth) use Converse threads internally.
3152
3153   Initialization is somewhat subtle, in that various modules
3154   won't work properly until they're initialized.  For example,
3155   nobody can register handlers before calling CmiHandlerInit.
3156 */
3157 void ConverseCommonInit(char **argv)
3158 {
3159
3160 /**
3161  * The reason to initialize this variable here:
3162  * cmiArgDebugFlag is possibly accessed in CmiPrintf/CmiError etc.,
3163  * therefore, we have to initialize this variable before any calls
3164  * to those functions (such as CmiPrintf). Otherwise, we may encounter
3165  * a memory segmentation fault (bad memory access). Note, even
3166  * testing CpvInitialized(cmiArgDebugFlag) doesn't help to solve
3167  * this problem because the variable indicating whether cmiArgDebugFlag is 
3168  * initialized or not is not initialized, thus possibly causing another
3169  * bad memory access. --Chao Mei
3170  */
3171 #if CMK_CCS_AVAILABLE
3172   CpvInitialize(int, cmiArgDebugFlag);
3173 #endif
3174
3175   CmiArgInit(argv);
3176   CmiMemoryInit(argv);
3177 #if ! CMK_CMIPRINTF_IS_A_BUILTIN
3178   CmiIOInit(argv);
3179 #endif
3180 #if CONVERSE_POOL
3181   CmiPoolAllocInit(30);  
3182 #endif
3183   CmiTmpInit(argv);
3184   CmiTimerInit();
3185   CstatsInit(argv);
3186
3187   CcdModuleInit(argv);
3188   CmiHandlerInit();
3189   CmiReductionsInit();
3190   CIdleTimeoutInit(argv);
3191   
3192 #if CMK_SHARED_VARS_POSIX_THREADS_SMP /*Used by the net-*-smp versions*/
3193   if(CmiGetArgFlagDesc(argv,"+CmiNoProcForComThread","Is there an extra processor for the communication thread on each node(only for net-smp-*) ?")){
3194     if(CmiMyRank() == 0) _Cmi_noprocforcommthread=1;
3195    }
3196 #endif
3197         
3198 #ifndef CMK_OPTIMIZE
3199   traceInit(argv);
3200 /*initTraceCore(argv);*/ /* projector */
3201 #endif
3202   CmiProcessPriority(argv);
3203
3204 #if CMK_CCS_AVAILABLE
3205   CcsInit(argv);
3206 #endif
3207   CmiPersistentInit();
3208   CmiIsomallocInit(argv);
3209   CpdInit();
3210   CmiDeliversInit();
3211   CsdInit(argv);
3212   CthSchedInit();
3213   CmiGroupInit();
3214   CmiMulticastInit();
3215   CmiInitMultipleSend();
3216   CQdInit();
3217
3218   CrnInit();
3219   CmiInitImmediateMsg();
3220   CldModuleInit(argv);
3221   
3222 #if CMK_CELL
3223   void CmiInitCell();
3224   CmiInitCell();
3225 #endif
3226
3227 #ifdef CMK_CUDA
3228   initHybridAPI(CmiMyPe()); 
3229 #endif
3230
3231   /* main thread is suspendable */
3232 /*
3233   CthSetSuspendable(CthSelf(), 0);
3234 */
3235
3236 #if CMK_BLUEGENE_CHARM
3237    /* have to initialize QD here instead of _initCharm */
3238   extern void initQd(char **argv);
3239   initQd(argv);
3240 #endif
3241 }
3242
3243 void ConverseCommonExit(void)
3244 {
3245   CcsImpl_kill();
3246
3247 #ifndef CMK_OPTIMIZE
3248   traceClose();
3249 /*closeTraceCore();*/ /* projector */
3250 #endif
3251
3252 #if CMI_IO_BUFFER_EXPLICIT
3253   CmiFlush(stdout);  /* end of program, always flush */
3254 #endif
3255
3256 #if CMK_CELL
3257   CloseOffloadAPI();
3258 #endif
3259
3260 #if CMK_CUDA
3261   exitHybridAPI(); 
3262 #endif
3263
3264 }
3265
3266
3267 #if CMK_CELL != 0
3268
3269 extern void register_accel_spe_funcs(void);
3270
3271 void CmiInitCell()
3272 {
3273   // Create a unique string for each PPE to use for the timing
3274   //   data file's name
3275   char fileNameBuf[64];
3276   sprintf(fileNameBuf, "speTiming.%d", CmiMyPe());
3277
3278   InitOffloadAPI(offloadCallback, NULL, NULL, fileNameBuf);
3279   //CcdCallOnConditionKeep(CcdPERIODIC, 
3280   //      (CcdVoidFn) OffloadAPIProgress, NULL);
3281   CcdCallOnConditionKeep(CcdPROCESSOR_STILL_IDLE,
3282       (CcdVoidFn) OffloadAPIProgress, NULL);
3283
3284   // Register accelerated entry methods on the PPE
3285   register_accel_spe_funcs();
3286 }
3287
3288 #include "cell-api.c"
3289
3290 #endif
3291
3292 /****
3293  * CW Lee - 9/14/2005
3294  * Added a mechanism to allow some control over machines with extremely
3295  * inefficient terminal IO mechanisms. Case in point: the XT3 has a
3296  * 20ms flush overhead along with about 25MB/s bandwidth for IO. This,
3297  * coupled with a default setup using unbuffered stdout introduced
3298  * severe overheads (and hence limiting scaling) for applications like 
3299  * NAMD.
3300  */
3301 #if ! CMK_CMIPRINTF_IS_A_BUILTIN
3302 void CmiIOInit(char **argv) {
3303   CpvInitialize(int, expIOFlushFlag);
3304 #if CMI_IO_BUFFER_EXPLICIT
3305   /* 
3306      Support for an explicit buffer only makes sense if the machine
3307      layer does not wish to make its own implementation.
3308
3309      Placing this after CmiMemoryInit() means that CmiMemoryInit()
3310      MUST NOT make use of stdout if an explicit buffer is requested.
3311
3312      The setvbuf function may only be used after opening a stream and
3313      before any other operations have been performed on it
3314   */
3315   CpvInitialize(char*, explicitIOBuffer);
3316   CpvInitialize(int, expIOBufferSize);
3317   if (!CmiGetArgIntDesc(argv,"+io_buffer_size", &CpvAccess(expIOBufferSize),
3318                         "Explicit IO Buffer Size")) {
3319     CpvAccess(expIOBufferSize) = DEFAULT_IO_BUFFER_SIZE;
3320   }
3321   if (CpvAccess(expIOBufferSize) <= 0) {
3322     CpvAccess(expIOBufferSize) = DEFAULT_IO_BUFFER_SIZE;
3323   }
3324   CpvAccess(explicitIOBuffer) = (char*)CmiAlloc(CpvAccess(expIOBufferSize)*
3325                                                 sizeof(char));
3326   if (setvbuf(stdout, CpvAccess(explicitIOBuffer), _IOFBF, 
3327               CpvAccess(expIOBufferSize))) {
3328     CmiAbort("Explicit IO Buffering failed\n");
3329   }
3330 #endif
3331 #if CMI_IO_FLUSH_USER
3332   /* system default to have user control flushing of IO */
3333   /* Now look for user override */
3334   CpvAccess(expIOFlushFlag) = !CmiGetArgFlagDesc(argv,"+io_flush_system",
3335                                                  "System Controls IO Flush");
3336 #else
3337   /* system default to have system handle IO flushing */
3338   /* Now look for user override */
3339   CpvAccess(expIOFlushFlag) = CmiGetArgFlagDesc(argv,"+io_flush_user",
3340                                                 "User Controls IO Flush");
3341 #endif
3342 }
3343 #endif
3344
3345 #if ! CMK_CMIPRINTF_IS_A_BUILTIN
3346
3347 void CmiPrintf(const char *format, ...)
3348 {
3349   CpdSystemEnter();
3350   {
3351   va_list args;
3352   va_start(args,format);
3353   vfprintf(stdout,format, args);
3354   if (CpvInitialized(expIOFlushFlag) && !CpvAccess(expIOFlushFlag)) {
3355     CmiFlush(stdout);
3356   }
3357   va_end(args);
3358 #if CMK_CCS_AVAILABLE
3359   if (CpvAccess(cmiArgDebugFlag)) {
3360     va_start(args,format);
3361     print_node0(format, args);
3362     va_end(args);
3363   }
3364 #endif
3365   }
3366   CpdSystemExit();
3367 }
3368
3369 void CmiError(const char *format, ...)
3370 {
3371   CpdSystemEnter();
3372   {
3373   va_list args;
3374   va_start(args,format);
3375   vfprintf(stderr,format, args);
3376   CmiFlush(stderr);  /* stderr is always flushed */
3377   va_end(args);
3378 #if CMK_CCS_AVAILABLE
3379   if (CpvAccess(cmiArgDebugFlag)) {
3380     va_start(args,format);
3381     print_node0(format, args);
3382     va_end(args);
3383   }
3384 #endif
3385   }
3386   CpdSystemExit();
3387 }
3388
3389 #endif
3390
3391 void __cmi_assert(const char *expr, const char *file, int line)
3392 {
3393   CmiError("[%d] Assertion \"%s\" failed in file %s line %d.\n",
3394       CmiMyPe(), expr, file, line);
3395   CmiAbort("");
3396 }
3397
3398 char *CmiCopyMsg(char *msg, int len)
3399 {
3400   char *copy = (char *)CmiAlloc(len);
3401   _MEMCHECK(copy);
3402   memcpy(copy, msg, len);
3403   return copy;
3404 }
3405
3406 unsigned char computeCheckSum(unsigned char *data, int len)
3407 {
3408   int i;
3409   unsigned char ret = 0;
3410   for (i=0; i<len; i++) ret ^= (unsigned char)data[i];
3411   return ret;
3412 }
3413
3414 /* Flag for bigsim's out-of-core emulation */
3415 int BgOutOfCoreFlag=0; /*indicate the type of memory operation (in or out) */
3416 int BgInOutOfCoreMode=0; /*indicate whether the emulation is in the out-of-core emulation mode */
3417
3418 #if !CMK_HAS_LOG2
3419 unsigned int CmiLog2(unsigned int val) {
3420   unsigned int log = 0u;
3421   if ( val != 0u ) {
3422       while ( val > (1u<<log) ) { log++; }
3423   }
3424   return log;
3425 }
3426 #endif
3427
3428 /*@}*/