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