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