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