add in log summary the meaning of the thread_resume_ep, so that loadlog can tell...
[charm.git] / src / langs / bluegene / blue.C
1 /*****************************************************************************
2  * $Source$
3  * $Author$
4  * $Date$
5  * $Revision$
6  *****************************************************************************/
7
8
9 /*
10   File: Blue.C -- Converse BlueGene Emulator Code
11   Emulator written by Gengbin Zheng, gzheng@uiuc.edu on 2/20/2001
12 */ 
13 #include <stdio.h>
14 #include <string.h>
15 #include <stdlib.h>
16
17 #include "cklists.h"
18 #include "queueing.h"
19 #include "blue.h"
20 #include "blue_impl.h"          // implementation header file
21 //#include "blue_timing.h"      // timing module
22
23 #define  DEBUGF(x)      //CmiPrintf x;
24
25 /* node level variables */
26 CpvDeclare(nodeInfo*, nodeinfo);                /* represent a bluegene node */
27
28 /* thread level variables */
29 CtvDeclare(threadInfo *, threadinfo);   /* represent a bluegene thread */
30
31 CpvStaticDeclare(CthThread, mainThread);
32
33 /* BG machine parameter */
34 CpvDeclare(BGMach, bgMach);     /* BG machine size description */
35 CpvDeclare(int, numNodes);        /* number of bg nodes on this PE */
36
37 /* emulator node level variables */
38 CpvDeclare(SimState, simState);
39
40 CpvDeclare(int      , CthResumeBigSimThreadIdx);
41
42 static int arg_argc;
43 static char **arg_argv;
44
45 CmiNodeLock initLock;     // used for BnvInitialize
46
47 int _bgSize = 0;                        // short cut of blue gene node size
48 int delayCheckFlag = 1;          // when enabled, only check correction 
49                                         // messages after some interval
50 int programExit = 0;
51
52 static int bgstats_flag = 0;            // flag print stats at end of simulation
53
54 // for debugging log
55 FILE *bgDebugLog;                       // for debugging
56
57 #ifdef CMK_ORIGIN2000
58 extern "C" int start_counters(int e0, int e1);
59 extern "C" int read_counters(int e0, long long *c0, int e1, long long *c1);
60 inline double Count2Time(long long c) { return c*5.e-7; }
61 #elif CMK_HAS_COUNTER_PAPI
62 #include <papi.h>
63 int papiEvents[4];
64 int numPapiEvents;
65 long_long papiValues[4];
66 #endif
67
68
69 /****************************************************************************
70      little utility functions
71 ****************************************************************************/
72
73 char **BgGetArgv() { return arg_argv; }
74 int    BgGetArgc() { return arg_argc; }
75
76 /***************************************************************************
77      Implementation of the same counter interface currently used for the
78      Origin2000 using PAPI in the underlying layer.
79
80      Because of the difference in counter numbering, it is assumed that
81      the two counters desired are CYCLES and FLOPS and hence the numbers
82      e0 and e1 are ignored.
83
84      start_counters is also not implemented because PAPI does things in
85      a different manner.
86 ****************************************************************************/
87
88 #if CMK_HAS_COUNTER_PAPI
89 int init_counters()
90 {
91     numPapiEvents = 0;
92     // PAPI high level API does not require explicit library intialization
93     papiEvents[numPapiEvents++] = PAPI_TOT_CYC;
94     if (PAPI_query_event(PAPI_FP_INS) == PAPI_OK) {
95       if (CmiMyPe()== 0) printf("PAPI_FP_INS used\n");
96       papiEvents[numPapiEvents++] = PAPI_FP_INS;
97     } else {
98       if (CmiMyPe()== 0) printf("PAPI_TOT_INS used\n");
99       papiEvents[numPapiEvents++] = PAPI_TOT_INS;
100     }
101     int status = PAPI_start_counters(papiEvents, numPapiEvents);
102     if (status != PAPI_OK) {
103       CmiPrintf("PAPI_start_counters error code: %d\n", status);
104       CmiAbort("Unable to start PAPI counters!\n");
105     }
106 }
107
108 int read_counters(long_long *papiValues, int n) 
109 {
110   // PAPI_read_counters resets the counter, hence it behaves like the perfctr
111   // code for Origin2000
112   int status;
113   status = PAPI_read_counters(papiValues, n);
114   if (status != PAPI_OK) {
115     CmiPrintf("PAPI_read_counters error: %d\n", status);
116     CmiAbort("Failed to read PAPI counters!\n");
117   }
118   return 0;
119 }
120
121 inline double Count2Time(long_long *papiValues, int n) { 
122   return papiValues[1]*cva(bgMach).fpfactor; 
123 }
124 #endif
125
126 /*****************************************************************************
127      Handler Table, one per thread
128 ****************************************************************************/
129
130 extern "C" void defaultBgHandler(char *null, void *uPtr)
131 {
132   CmiAbort("BG> Invalid Handler called!\n");
133 }
134
135 HandlerTable::HandlerTable()
136 {
137     handlerTableCount = 1;
138     handlerTable = new BgHandlerInfo [MAX_HANDLERS];
139     for (int i=0; i<MAX_HANDLERS; i++) {
140       handlerTable[i].fnPtr = defaultBgHandler;
141       handlerTable[i].userPtr = NULL;
142     }
143 }
144
145 inline int HandlerTable::registerHandler(BgHandler h)
146 {
147     ASSERT(!cva(simState).inEmulatorInit);
148     /* leave 0 as blank, so it can report error luckily */
149     int cur = handlerTableCount++;
150     if (cur >= MAX_HANDLERS)
151       CmiAbort("BG> HandlerID exceed the maximum.\n");
152     handlerTable[cur].fnPtr = (BgHandlerEx)h;
153     handlerTable[cur].userPtr = NULL;
154     return cur;
155 }
156
157 inline void HandlerTable::numberHandler(int idx, BgHandler h)
158 {
159     ASSERT(!cva(simState).inEmulatorInit);
160     if (idx >= handlerTableCount || idx < 1)
161       CmiAbort("BG> HandlerID exceed the maximum!\n");
162     handlerTable[idx].fnPtr = (BgHandlerEx)h;
163     handlerTable[idx].userPtr = NULL;
164 }
165
166 inline void HandlerTable::numberHandlerEx(int idx, BgHandlerEx h, void *uPtr)
167 {
168     ASSERT(!cva(simState).inEmulatorInit);
169     if (idx >= handlerTableCount || idx < 1)
170       CmiAbort("BG> HandlerID exceed the maximum!\n");
171     handlerTable[idx].fnPtr = h;
172     handlerTable[idx].userPtr = uPtr;
173 }
174
175 inline BgHandlerInfo * HandlerTable::getHandle(int handler)
176 {
177 #if 0
178     if (handler >= handlerTableCount) {
179       CmiPrintf("[%d] handler: %d handlerTableCount:%d. \n", tMYNODEID, handler, handlerTableCount);
180       CmiAbort("Invalid handler!");
181     }
182 #endif
183     if (handler >= handlerTableCount) return NULL;
184     return &handlerTable[handler];
185 }
186
187 /*****************************************************************************
188       low level API
189 *****************************************************************************/
190
191 int BgRegisterHandler(BgHandler h)
192 {
193   ASSERT(!cva(simState).inEmulatorInit);
194   int cur;
195 #if CMK_BLUEGENE_NODE
196   return tMYNODE->handlerTable.registerHandler(h);
197 #else
198   if (tTHREADTYPE == COMM_THREAD) {
199     return tMYNODE->handlerTable.registerHandler(h);
200   }
201   else {
202     return tHANDLETAB.registerHandler(h);
203   }
204 #endif
205 }
206
207 void BgNumberHandler(int idx, BgHandler h)
208 {
209   ASSERT(!cva(simState).inEmulatorInit);
210 #if CMK_BLUEGENE_NODE
211   tMYNODE->handlerTable.numberHandler(idx,h);
212 #else
213   if (tTHREADTYPE == COMM_THREAD) {
214     tMYNODE->handlerTable.numberHandler(idx, h);
215   }
216   else {
217     tHANDLETAB.numberHandler(idx, h);
218   }
219 #endif
220 }
221
222 void BgNumberHandlerEx(int idx, BgHandlerEx h, void *uPtr)
223 {
224   ASSERT(!cva(simState).inEmulatorInit);
225 #if CMK_BLUEGENE_NODE
226   tMYNODE->handlerTable.numberHandlerEx(idx,h,uPtr);
227 #else
228   if (tTHREADTYPE == COMM_THREAD) {
229     tMYNODE->handlerTable.numberHandlerEx(idx,h,uPtr);
230   }
231   else {
232     tHANDLETAB.numberHandlerEx(idx,h,uPtr);
233   }
234 #endif
235 }
236
237 /*****************************************************************************
238       BG Timing Functions
239 *****************************************************************************/
240
241 void resetVTime()
242 {
243   /* reset start time */
244   int timingMethod = cva(bgMach).timingMethod;
245   if (timingMethod == BG_WALLTIME) {
246     double ct = CmiWallTimer();
247     if (tTIMERON) CmiAssert(ct >= tSTARTTIME);
248     tSTARTTIME = ct;
249   }
250   else if (timingMethod == BG_ELAPSE)
251     tSTARTTIME = tCURRTIME;
252 #ifdef CMK_ORIGIN2000
253   else if (timingMethod == BG_COUNTER) {
254     if (start_counters(0, 21) <0) {
255       perror("start_counters");;
256     }
257   }
258 #elif CMK_HAS_COUNTER_PAPI
259   else if (timingMethod == BG_COUNTER) {
260     // do a fake read to reset the counters. It would be more efficient
261     // to use the low level API, but that would be a lot more code to
262     // write for now.
263     if (read_counters(papiValues, numPapiEvents) < 0) perror("read_counters");
264   }
265 #endif
266 }
267
268 void startVTimer()
269 {
270   CmiAssert(tTIMERON == 0);
271   resetVTime();
272   tTIMERON = 1;
273 }
274
275 // should be used only when BG_WALLTIME
276 static inline void advanceTime(double inc)
277 {
278   if (BG_ABS(inc) < 1e-10) inc = 0.0;    // ignore floating point errors
279   CmiAssert(inc>=0.0);
280   inc *= cva(bgMach).cpufactor;
281   tCURRTIME += inc;
282 }
283
284 void stopVTimer()
285 {
286   int k;
287   if (tTIMERON != 1) {
288     CmiAbort("stopVTimer called without startVTimer!\n");
289   }
290   CmiAssert(tTIMERON == 1);
291   tTIMERON = 0;
292   const int timingMethod = cva(bgMach).timingMethod;
293   if (timingMethod == BG_WALLTIME) {
294     const double tp = CmiWallTimer();
295     double inc = tp-tSTARTTIME;
296     advanceTime(inc);
297 //    tSTARTTIME = CmiWallTimer();      // skip the above time
298   }
299   else if (timingMethod == BG_ELAPSE) {
300     // if no bgelapse called, assume it takes 1us
301     if (tCURRTIME-tSTARTTIME < 1E-9) {
302 //      tCURRTIME += 1e-6;
303     }
304   }
305   else if (timingMethod == BG_COUNTER)  {
306 #if CMK_ORIGIN2000
307     long long c0, c1;
308     if (read_counters(0, &c0, 21, &c1) < 0) perror("read_counters");
309     tCURRTIME += Count2Time(c1);
310 #elif CMK_HAS_COUNTER_PAPI
311     if (read_counters(papiValues, numPapiEvents) < 0) perror("read_counters");
312     tCURRTIME += Count2Time(papiValues, numPapiEvents);
313 #endif
314   }
315 }
316
317 double BgGetTime()
318 {
319 #if 1
320   const int timingMethod = cva(bgMach).timingMethod;
321   if (timingMethod == BG_WALLTIME) {
322     /* accumulate time since last starttime, and reset starttime */
323     if (tTIMERON) {
324       const double tp2= CmiWallTimer();
325       double &startTime = tSTARTTIME;
326       double inc = tp2 - startTime;
327       advanceTime(inc);
328       startTime = CmiWallTimer();
329     }
330     return tCURRTIME;
331   }
332   else if (timingMethod == BG_ELAPSE) {
333     return tCURRTIME;
334   }
335   else if (timingMethod == BG_COUNTER) {
336     if (tTIMERON) {
337 #if CMK_ORIGIN2000
338       long long c0, c1;
339       if (read_counters(0, &c0, 21, &c1) <0) perror("read_counters");;
340       tCURRTIME += Count2Time(c1);
341       if (start_counters(0, 21)<0) perror("start_counters");;
342 #elif CMK_HAS_COUNTER_PAPI
343     if (read_counters(papiValues, numPapiEvents) < 0) perror("read_counters");
344     tCURRTIME += Count2Time(papiValues, numPapiEvents);
345 #endif
346     }
347     return tCURRTIME;
348   }
349   else 
350     CmiAbort("Unknown Timing Method.");
351   return -1;
352 #else
353   /* sometime I am interested in real wall time */
354   tCURRTIME = CmiWallTimer();
355   return tCURRTIME;
356 #endif
357 }
358
359 // moved to blue_logs.C
360 double BgGetCurTime()
361 {
362   ASSERT(tTHREADTYPE == WORK_THREAD);
363   return tCURRTIME;
364 }
365
366 extern "C" 
367 void BgElapse(double t)
368 {
369 //  ASSERT(tTHREADTYPE == WORK_THREAD);
370   if (cva(bgMach).timingMethod == BG_ELAPSE)
371     tCURRTIME += t;
372 }
373
374 // advance virtual timer no matter what scheme is used
375 extern "C" 
376 void BgAdvance(double t)
377 {
378 //  ASSERT(tTHREADTYPE == WORK_THREAD);
379   tCURRTIME += t;
380 }
381
382 /* BG API Func
383  * called by a communication thread to test if poll data 
384  * in the node's INBUFFER for its own queue 
385  */
386 char * getFullBuffer()
387 {
388   /* I must be a communication thread */
389   if (tTHREADTYPE != COMM_THREAD) 
390     CmiAbort("GetFullBuffer called by a non-communication thread!\n");
391
392   return tMYNODE->getFullBuffer();
393 }
394
395 /**  BG API Func
396  * called by a Converse handler or sendPacket()
397  * add message msgPtr to a bluegene node's inbuffer queue 
398  */
399 void addBgNodeInbuffer(char *msgPtr, int lnodeID)
400 {
401 #ifndef CMK_OPTIMIZE
402   if (lnodeID >= cva(numNodes)) CmiAbort("NodeID is out of range!");
403 #endif
404   nodeInfo &nInfo = cva(nodeinfo)[lnodeID];
405   nInfo.addBgNodeInbuffer(msgPtr);
406 }
407
408 /** BG API Func 
409  *  called by a comm thread
410  *  add a message to a thread's affinity queue in same node 
411  */
412 void addBgThreadMessage(char *msgPtr, int threadID)
413 {
414 #ifndef CMK_OPTIMIZE
415   if (!cva(bgMach).isWorkThread(threadID)) CmiAbort("ThreadID is out of range!");
416 #endif
417   workThreadInfo *tInfo = (workThreadInfo *)tMYNODE->threadinfo[threadID];
418   tInfo->addAffMessage(msgPtr);
419 }
420
421 /** BG API Func 
422  *  called by a comm thread, add a message to a node's non-affinity queue 
423  */
424 void addBgNodeMessage(char *msgPtr)
425 {
426   tMYNODE->addBgNodeMessage(msgPtr);
427 }
428
429 /** BG API Func 
430  *  check if inBuffer on this node has msg available
431  */
432 int checkReady()
433 {
434   if (tTHREADTYPE != COMM_THREAD)
435     CmiAbort("checkReady called by a non-communication thread!\n");
436   return !tINBUFFER.isEmpty();
437 }
438
439
440 /* handler to process the msg */
441 void msgHandlerFunc(char *msg)
442 {
443   /* bgmsg is CmiMsgHeaderSizeBytes offset of original message pointer */
444   int gnodeID = CmiBgMsgNodeID(msg);
445   if (gnodeID >= 0) {
446 #ifndef CMK_OPTIMIZE
447     if (nodeInfo::Global2PE(gnodeID) != CmiMyPe())
448       CmiAbort("msgHandlerFunc received wrong message!");
449 #endif
450     int lnodeID = nodeInfo::Global2Local(gnodeID);
451     addBgNodeInbuffer(msg, lnodeID);
452   }
453   else {
454     CmiAbort("Invalid message!");
455   }
456 }
457
458 /* Converse handler for node level broadcast message */
459 void nodeBCastMsgHandlerFunc(char *msg)
460 {
461   /* bgmsg is CmiMsgHeaderSizeBytes offset of original message pointer */
462   int gnodeID = CmiBgMsgNodeID(msg);
463   CmiInt2 threadID = CmiBgMsgThreadID(msg);
464   int lnodeID;
465
466   if (gnodeID < -1) {
467     gnodeID = - (gnodeID+100);
468     if (nodeInfo::Global2PE(gnodeID) == CmiMyPe())
469       lnodeID = nodeInfo::Global2Local(gnodeID);
470     else
471       lnodeID = -1;
472   }
473   else {
474     ASSERT(gnodeID == -1);
475     lnodeID = gnodeID;
476   }
477   // broadcast except lnodeID:threadId
478   int len = CmiBgMsgLength(msg);
479   int count = 0;
480   for (int i=0; i<cva(numNodes); i++)
481   {
482     if (i==lnodeID) continue;
483     char *dupmsg;
484     if (count == 0) dupmsg = msg;
485     else dupmsg = CmiCopyMsg(msg, len);
486     DEBUGF(("addBgNodeInbuffer to %d\n", i));
487     CmiBgMsgNodeID(dupmsg) = nodeInfo::Local2Global(i);         // updated
488     addBgNodeInbuffer(dupmsg, i);
489     count ++;
490   }
491   if (count == 0) CmiFree(msg);
492 }
493
494 // clone a msg, only has a valid header, plus a pointer to the real msg
495 char *BgCloneMsg(char *msg)
496 {
497   int size = CmiBlueGeneMsgHeaderSizeBytes + sizeof(char *);
498   char *dupmsg = (char *)CmiAlloc(size);
499   memcpy(dupmsg, msg, CmiBlueGeneMsgHeaderSizeBytes);
500   *(char **)(dupmsg + CmiBlueGeneMsgHeaderSizeBytes) = msg;
501   CmiBgMsgRefCount(msg) ++;
502   CmiBgMsgFlag(dupmsg) = BG_CLONE;
503   return dupmsg;
504 }
505
506 // expand the cloned msg to the full size msg
507 char *BgExpandMsg(char *msg)
508 {
509   char *origmsg = *(char **)(msg + CmiBlueGeneMsgHeaderSizeBytes);
510   int size = CmiBgMsgLength(origmsg);
511   char *dupmsg = (char *)CmiAlloc(size);
512   memcpy(dupmsg, msg, CmiBlueGeneMsgHeaderSizeBytes);
513   memcpy(dupmsg+CmiBlueGeneMsgHeaderSizeBytes, origmsg+CmiBlueGeneMsgHeaderSizeBytes, size-CmiBlueGeneMsgHeaderSizeBytes);
514   CmiFree(msg);
515   CmiBgMsgRefCount(origmsg) --;
516   if (CmiBgMsgRefCount(origmsg) == 0) CmiFree(origmsg);
517   return dupmsg;
518 }
519
520 /* Converse handler for thread level broadcast message */
521 void threadBCastMsgHandlerFunc(char *msg)
522 {
523   /* bgmsg is CmiMsgHeaderSizeBytes offset of original message pointer */
524   int gnodeID = CmiBgMsgNodeID(msg);
525   CmiInt2 threadID = CmiBgMsgThreadID(msg);
526   int lnodeID;
527   if (gnodeID < -1) {
528       gnodeID = - (gnodeID+100);
529       if (nodeInfo::Global2PE(gnodeID) == CmiMyPe())
530         lnodeID = nodeInfo::Global2Local(gnodeID);
531       else
532         lnodeID = -1;
533       CmiAssert(threadID != ANYTHREAD);
534   }
535   else {
536     ASSERT(gnodeID == -1);
537     lnodeID = gnodeID;
538   }
539   // broadcast except nodeID:threadId
540   int len = CmiBgMsgLength(msg);
541   // optimization needed if the message size is big
542   // making duplications can easily run out of memory
543   int bigOpt = (len > 4096);
544   for (int i=0; i<cva(numNodes); i++)
545   {
546       for (int j=0; j<cva(bgMach).numWth; j++) {
547         if (i==lnodeID && j==threadID) continue;
548         // for big message, clone a message token instead of a real msg
549         char *dupmsg = bigOpt? BgCloneMsg(msg) : CmiCopyMsg(msg, len);
550         CmiBgMsgNodeID(dupmsg) = nodeInfo::Local2Global(i);
551         CmiBgMsgThreadID(dupmsg) = j;
552         DEBUGF(("[%d] addBgNodeInbuffer to %d tid:%d\n", CmiMyPe(), i, j));
553         addBgNodeInbuffer(dupmsg, i);
554       }
555   }
556   // for big message, will free after all tokens are done
557   if (!bigOpt) CmiFree(msg);
558 }
559
560 /**
561  *              BG Messaging Functions
562  */
563
564 static inline double MSGTIME(int ox, int oy, int oz, int nx, int ny, int nz, int bytes)
565 {
566   return cva(bgMach).network->latency(ox, oy, oz, nx, ny, nz, bytes);
567 }
568
569 /**
570  *   a simple message streaming on demand and special purpose
571  *   user call  BgStartStreaming() and BgEndStreaming()
572  *   each worker thread call one send, and all sends are sent
573  *   via multiplesend at the end
574  */
575
576 static int bg_streaming = 0;
577
578 class BgStreaming {
579 public:
580   char **streamingMsgs;
581   int  *streamingMsgSizes;
582   int count;
583   int totalWorker;
584   int pe;
585 public:
586   BgStreaming() {
587     streamingMsgs = NULL;
588     streamingMsgSizes = NULL;
589     count = 0;
590     totalWorker = 0;
591     pe = -1;
592   }
593   ~BgStreaming() {
594     if (streamingMsgs) {
595       delete [] streamingMsgs;
596       delete [] streamingMsgSizes;
597     }
598   }
599   void init(int nNodes) {
600     totalWorker = nNodes * BgGetNumWorkThread();
601     streamingMsgs = new char *[totalWorker];
602     streamingMsgSizes = new int [totalWorker];
603   }
604   void depositMsg(int p, int size, char *m) {
605     streamingMsgs[count] = m;
606     streamingMsgSizes[count] = size;
607     count ++;
608     if (pe == -1) pe = p;
609     else CmiAssert(pe == p);
610     if (count == totalWorker) {
611       // CkPrintf("streaming send\n");
612       CmiMultipleSend(pe, count, streamingMsgSizes, streamingMsgs);
613       for (int i=0; i<count; i++) CmiFree(streamingMsgs[i]);
614       pe = -1;
615       count = 0;
616     }
617   }
618 };
619
620 BgStreaming bgstreaming;
621
622 void BgStartStreaming()
623 {
624   bg_streaming = 1;
625 }
626
627 void BgEndStreaming()
628 {
629   bg_streaming = 0;
630 }
631
632 void CmiSendPacketWrapper(int pe, int msgSize,char *msg, int streaming)
633 {
634   if (streaming && pe != CmiMyPe())
635     bgstreaming.depositMsg(pe, msgSize, msg);
636   else
637     CmiSyncSendAndFree(pe, msgSize, msg);
638 }
639
640
641 void CmiSendPacket(int x, int y, int z, int msgSize,char *msg)
642 {
643 //  CmiSyncSendAndFree(nodeInfo::XYZ2PE(x,y,z),msgSize,(char *)msg);
644 #if !DELAY_SEND
645   const int pe = nodeInfo::XYZ2PE(x,y,z);
646   CmiSendPacketWrapper(pe, msgSize, msg, bg_streaming);
647 #else
648   if (!correctTimeLog) {
649     const int pe = nodeInfo::XYZ2PE(x,y,z);
650     CmiSendPacketWrapper(pe, msgSize, msg, bg_streaming);
651   }
652   // else messages are kept in the log (MsgEntry), and only will be sent
653   // after timing correction has done on that log.
654   // TODO: streaming has no effect if time correction is on.
655 #endif
656 }
657
658 /* send will copy data to msg buffer */
659 /* user data is not free'd in this routine, user can reuse the data ! */
660 void sendPacket_(nodeInfo *myNode, int x, int y, int z, int threadID, int handlerID, WorkType type, int numbytes, char* sendmsg, int local)
661 {
662   double latency;
663   CmiSetHandler(sendmsg, cva(simState).msgHandler);
664   CmiBgMsgNodeID(sendmsg) = nodeInfo::XYZ2Global(x,y,z);
665   CmiBgMsgThreadID(sendmsg) = threadID;
666   CmiBgMsgHandle(sendmsg) = handlerID;
667   CmiBgMsgType(sendmsg) = type;
668   CmiBgMsgLength(sendmsg) = numbytes;
669   CmiBgMsgFlag(sendmsg) = 0;
670   CmiBgMsgRefCount(sendmsg) = 0;
671   if (local) {
672     BgAdvance(cva(bgMach).network->charmcost());
673     latency = 0.0;
674   }
675   else {
676     BgAdvance(cva(bgMach).network->alphacost());
677     latency = MSGTIME(myNode->x, myNode->y, myNode->z, x,y,z, numbytes);
678     CmiAssert(latency >= 0);
679   }
680   CmiBgMsgRecvTime(sendmsg) = latency + BgGetTime();
681   
682   // timing
683   BG_ADDMSG(sendmsg, CmiBgMsgNodeID(sendmsg), threadID, local, 1);
684
685   if (local)
686     addBgNodeInbuffer(sendmsg, myNode->id);
687   else
688     CmiSendPacket(x, y, z, numbytes, sendmsg);
689
690   // bypassing send time
691   resetVTime();
692 }
693
694 /* broadcast will copy data to msg buffer */
695 static inline void nodeBroadcastPacketExcept_(int node, CmiInt2 threadID, int handlerID, WorkType type, int numbytes, char* sendmsg)
696 {
697   nodeInfo *myNode = cta(threadinfo)->myNode;
698   CmiSetHandler(sendmsg, cva(simState).nBcastMsgHandler);
699   if (node >= 0)
700     CmiBgMsgNodeID(sendmsg) = -node-100;
701   else
702     CmiBgMsgNodeID(sendmsg) = node;
703   CmiBgMsgThreadID(sendmsg) = threadID; 
704   CmiBgMsgHandle(sendmsg) = handlerID;  
705   CmiBgMsgType(sendmsg) = type; 
706   CmiBgMsgLength(sendmsg) = numbytes;
707   CmiBgMsgFlag(sendmsg) = 0;
708   CmiBgMsgRefCount(sendmsg) = 0;
709   /* FIXME */
710   CmiBgMsgRecvTime(sendmsg) = MSGTIME(myNode->x, myNode->y, myNode->z, 0,0,0, numbytes) + BgGetTime();
711
712   // timing
713   // FIXME
714   BG_ADDMSG(sendmsg, CmiBgMsgNodeID(sendmsg), threadID, 0, 1);
715
716   DEBUGF(("[%d]CmiSyncBroadcastAllAndFree node: %d\n", BgMyNode(), node));
717 #if DELAY_SEND
718   if (!correctTimeLog)
719 #endif
720   CmiSyncBroadcastAllAndFree(numbytes,sendmsg);
721
722   resetVTime();
723 }
724
725 /* broadcast will copy data to msg buffer */
726 static inline void threadBroadcastPacketExcept_(int node, CmiInt2 threadID, int handlerID, WorkType type, int numbytes, char* sendmsg)
727 {
728   CmiSetHandler(sendmsg, cva(simState).tBcastMsgHandler);       
729   if (node >= 0)
730     CmiBgMsgNodeID(sendmsg) = -node-100;
731   else
732     CmiBgMsgNodeID(sendmsg) = node;
733   CmiBgMsgThreadID(sendmsg) = threadID; 
734   CmiBgMsgHandle(sendmsg) = handlerID;  
735   CmiBgMsgType(sendmsg) = type; 
736   CmiBgMsgLength(sendmsg) = numbytes;
737   CmiBgMsgFlag(sendmsg) = 0;
738   CmiBgMsgRefCount(sendmsg) = 0;
739   /* FIXME */
740   BgAdvance(cva(bgMach).network->alphacost());
741   CmiBgMsgRecvTime(sendmsg) = BgGetTime();      
742
743   // timing
744 #if 0
745   if (node == BG_BROADCASTALL) {
746     for (int i=0; i<_bgSize; i++) {
747       for (int j=0; j<cva(numWth); j++) {
748         BG_ADDMSG(sendmsg, node);
749       }
750     }
751   }
752   else {
753     CmiAssert(node >= 0);
754     BG_ADDMSG(sendmsg, (node+100));
755   }
756 #else
757   // FIXME
758   BG_ADDMSG(sendmsg, CmiBgMsgNodeID(sendmsg), threadID, 0, 1);
759 #endif
760
761   DEBUGF(("[%d]CmiSyncBroadcastAllAndFree node: %d tid:%d recvT:%f\n", BgMyNode(), node, threadID, CmiBgMsgRecvTime(sendmsg)));
762 #if DELAY_SEND
763   if (!correctTimeLog)
764 #endif
765   CmiSyncBroadcastAllAndFree(numbytes,sendmsg);
766
767   resetVTime();
768 }
769
770 /* sendPacket to route */
771 /* this function can be called by any thread */
772 void BgSendNonLocalPacket(nodeInfo *myNode, int x, int y, int z, int threadID, int handlerID, WorkType type, int numbytes, char * data)
773 {
774 #ifndef CMK_OPTIMIZE
775   if (x<0 || y<0 || z<0 || x>=cva(bgMach).x || y>=cva(bgMach).y || z>=cva(bgMach).z) {
776     CmiPrintf("Trying to send packet to a nonexisting node: (%d %d %d)!\n", x,y,z);
777     CmiAbort("Abort!\n");
778   }
779 #endif
780
781   sendPacket_(myNode, x, y, z, threadID, handlerID, type, numbytes, data, 0);
782 }
783
784 static void _BgSendLocalPacket(nodeInfo *myNode, int threadID, int handlerID, WorkType type, int numbytes, char * data)
785 {
786   sendPacket_(myNode, myNode->x, myNode->y, myNode->z, threadID, handlerID, type, numbytes, data, 1);
787 }
788
789 void BgSendLocalPacket(int threadID, int handlerID, WorkType type,
790                        int numbytes, char* data)
791 {
792   nodeInfo *myNode = cta(threadinfo)->myNode;
793   _BgSendLocalPacket(myNode, threadID, handlerID, type, numbytes, data);
794 }
795
796 /* wrapper of the previous two functions */
797 void BgSendPacket(int x, int y, int z, int threadID, int handlerID, WorkType type, int numbytes, char * data)
798 {
799   nodeInfo *myNode = cta(threadinfo)->myNode;
800   if (myNode->x == x && myNode->y == y && myNode->z == z)
801     _BgSendLocalPacket(myNode,threadID, handlerID, type, numbytes, data);
802   else
803     BgSendNonLocalPacket(myNode,x,y,z,threadID,handlerID, type, numbytes, data);
804 }
805
806 void BgBroadcastPacketExcept(int node, CmiInt2 threadID, int handlerID, WorkType type, int numbytes, char * data)
807 {
808   nodeBroadcastPacketExcept_(node, threadID, handlerID, type, numbytes, data);
809 }
810
811 void BgBroadcastAllPacket(int handlerID, WorkType type, int numbytes, char * data)
812 {
813   nodeBroadcastPacketExcept_(BG_BROADCASTALL, ANYTHREAD, handlerID, type, numbytes, data);
814 }
815
816 void BgThreadBroadcastPacketExcept(int node, CmiInt2 threadID, int handlerID, WorkType type, int numbytes, char * data)
817 {
818   threadBroadcastPacketExcept_(node, threadID, handlerID, type, numbytes, data);
819 }
820
821 void BgThreadBroadcastAllPacket(int handlerID, WorkType type, int numbytes, char * data)
822 {
823   threadBroadcastPacketExcept_(BG_BROADCASTALL, ANYTHREAD, handlerID, type, numbytes, data);
824 }
825
826 /**
827  send a msg to a list of processors (processors represented in global seq #
828 */
829 void BgSyncListSend(int npes, int *pes, int handlerID, WorkType type, int numbytes, char *msg)
830 {
831   nodeInfo *myNode = cta(threadinfo)->myNode;
832
833   CmiSetHandler(msg, cva(simState).msgHandler);
834   CmiBgMsgHandle(msg) = handlerID;
835   CmiBgMsgType(msg) = type;
836   CmiBgMsgLength(msg) = numbytes;
837   CmiBgMsgFlag(msg) = 0;
838   CmiBgMsgRefCount(msg) = 0;
839
840   BgAdvance(cva(bgMach).network->alphacost());
841
842   double now = BgGetTime();
843
844   // send one by one
845   for (int i=0; i<npes; i++)
846   {
847     int local = 0;
848     int x,y,z,t;
849     int pe = pes[i];
850 #if CMK_BLUEGENE_NODE
851     CmiAbort("Not implemented yet!");
852 #else
853     t = pe%BgGetNumWorkThread();
854     pe = pe/BgGetNumWorkThread();
855     BgGetXYZ(pe, &x, &y, &z);
856 #endif
857
858     char *sendmsg = CmiCopyMsg(msg, numbytes);
859     CmiBgMsgNodeID(sendmsg) = nodeInfo::XYZ2Global(x,y,z);
860     CmiBgMsgThreadID(sendmsg) = t;
861     double latency = MSGTIME(myNode->x, myNode->y, myNode->z, x,y,z, numbytes);
862     CmiAssert(latency >= 0);
863     CmiBgMsgRecvTime(sendmsg) = latency + now;
864
865     if (myNode->x == x && myNode->y == y && myNode->z == z) local = 1;
866
867     // timing and make sure all msgID are the same
868     if (i!=0) CpvAccess(msgCounter) --;
869     BG_ADDMSG(sendmsg, CmiBgMsgNodeID(sendmsg), t, local, i==0?npes:-1);
870
871     if (myNode->x == x && myNode->y == y && myNode->z == z)
872       addBgNodeInbuffer(sendmsg, myNode->id);
873     else
874       CmiSendPacket(x, y, z, numbytes, sendmsg);
875   }
876
877   CmiFree(msg);
878
879   resetVTime();
880 }
881
882 /*****************************************************************************
883       BG node level API - utilities
884 *****************************************************************************/
885
886 /* must be called in a communication or worker thread */
887 void BgGetMyXYZ(int *x, int *y, int *z)
888 {
889   ASSERT(!cva(simState).inEmulatorInit);
890   *x = tMYX; *y = tMYY; *z = tMYZ;
891 }
892
893 void BgGetXYZ(int seq, int *x, int *y, int *z)
894 {
895   nodeInfo::Global2XYZ(seq, x, y, z);
896 }
897
898 void BgGetSize(int *sx, int *sy, int *sz)
899 {
900   cva(bgMach).getSize(sx, sy, sz);
901 }
902
903 int BgTraceProjectionOn(int pe)
904 {
905   return cva(bgMach).traceProejctions(pe);
906 }
907
908 /* return the total number of Blue gene nodes */
909 int BgNumNodes()
910 {
911   return _bgSize;
912 }
913
914 /* can only called in emulatorinit */
915 void BgSetSize(int sx, int sy, int sz)
916 {
917   ASSERT(cva(simState).inEmulatorInit);
918   cva(bgMach).setSize(sx, sy, sz);
919 }
920
921 /* return number of bg nodes on this emulator node */
922 int BgNodeSize()
923 {
924   ASSERT(!cva(simState).inEmulatorInit);
925   return cva(numNodes);
926 }
927
928 /* return the bg node ID (local array index) */
929 int BgMyRank()
930 {
931 #ifndef CMK_OPTIMIZE
932   if (tMYNODE == NULL) CmiAbort("Calling BgMyRank in the main thread!");
933 #endif
934   ASSERT(!cva(simState).inEmulatorInit);
935   return tMYNODEID;
936 }
937
938 /* return my serialed blue gene node number */
939 int BgMyNode()
940 {
941 #ifndef CMK_OPTIMIZE
942   if (tMYNODE == NULL) CmiAbort("Calling BgMyNode in the main thread!");
943 #endif
944   return nodeInfo::XYZ2Global(tMYX, tMYY, tMYZ);
945 }
946
947 /* return a real processor number from a bg node */
948 int BgNodeToPE(int node)
949 {
950   return nodeInfo::Global2PE(node);
951 }
952
953 // thread ID on a BG node
954 int BgGetThreadID()
955 {
956   ASSERT(tTHREADTYPE == WORK_THREAD || tTHREADTYPE == COMM_THREAD);
957 //  if (cva(bgMach).numWth == 1) return 0;   // accessing ctv is expensive
958   return tMYID;
959 }
960
961 int BgGetGlobalThreadID()
962 {
963   ASSERT(tTHREADTYPE == WORK_THREAD || tTHREADTYPE == COMM_THREAD);
964   return nodeInfo::Local2Global(tMYNODEID)*(cva(bgMach).numTh())+tMYID;
965   //return tMYGLOBALID;
966 }
967
968 int BgGetGlobalWorkerThreadID()
969 {
970   ASSERT(tTHREADTYPE == WORK_THREAD);
971 //  return nodeInfo::Local2Global(tMYNODEID)*cva(bgMach).numWth+tMYID;
972   return tMYGLOBALID;
973 }
974
975 char *BgGetNodeData()
976 {
977   return tUSERDATA;
978 }
979
980 void BgSetNodeData(char *data)
981 {
982   ASSERT(!cva(simState).inEmulatorInit);
983   tUSERDATA = data;
984 }
985
986 int BgGetNumWorkThread()
987 {
988   return cva(bgMach).numWth;
989 }
990
991 void BgSetNumWorkThread(int num)
992 {
993   ASSERT(cva(simState).inEmulatorInit);
994   cva(bgMach).numWth = num;
995 }
996
997 int BgGetNumCommThread()
998 {
999   return cva(bgMach).numCth;
1000 }
1001
1002 void BgSetNumCommThread(int num)
1003 {
1004   ASSERT(cva(simState).inEmulatorInit);
1005   cva(bgMach).numCth = num;
1006 }
1007
1008 /*****************************************************************************
1009       Communication and Worker threads
1010 *****************************************************************************/
1011
1012 BgStartHandler  workStartFunc = NULL;
1013
1014 void BgSetWorkerThreadStart(BgStartHandler f)
1015 {
1016   workStartFunc = f;
1017 }
1018
1019 // kernel function for processing a bluegene message
1020 void BgProcessMessage(char *msg)
1021 {
1022   int handler = CmiBgMsgHandle(msg);
1023   DEBUGF(("[%d] call handler %d\n", BgMyNode(), handler));
1024
1025   BgHandlerInfo *handInfo;
1026 #if  CMK_BLUEGENE_NODE
1027   handInfo = tMYNODE->handlerTable.getHandle(handler);
1028 #else
1029   handInfo = tHANDLETAB.getHandle(handler);
1030   if (handInfo == NULL) handInfo = tMYNODE->handlerTable.getHandle(handler);
1031 #endif
1032
1033   if (handInfo == NULL) {
1034     CmiPrintf("[%d] invalid handler: %d. \n", tMYNODEID, handler);
1035     CmiAbort("BgProcessMessage Failed!");
1036   }
1037   BgHandlerEx entryFunc = handInfo->fnPtr;
1038
1039   CmiSetHandler(msg, CmiBgMsgHandle(msg));
1040
1041   // optimization for broadcast messages:
1042   // if the msg is a broadcast token msg, expand it to a real msg
1043   if (CmiBgMsgFlag(msg) == BG_CLONE) {
1044     msg = BgExpandMsg(msg);
1045   }
1046
1047   // don't count thread overhead and timing overhead
1048   startVTimer();
1049
1050   entryFunc(msg, handInfo->userPtr);
1051
1052   stopVTimer();
1053 }
1054
1055
1056 void scheduleWorkerThread(char *msg)
1057 {
1058   CthThread tid = (CthThread)msg;
1059 //CmiPrintf("scheduleWorkerThread %p\n", tid);
1060   CthAwaken(tid);
1061 }
1062
1063 // thread entry
1064 // for both comm and work thread, virtual function
1065 void run_thread(threadInfo *tinfo)
1066 {
1067   /* set the thread-private threadinfo */
1068   cta(threadinfo) = tinfo;
1069   tinfo->run();
1070 }
1071
1072 /* should be done only once per bg node */
1073 void BgNodeInitialize(nodeInfo *ninfo)
1074 {
1075   CthThread t;
1076   int i;
1077
1078   /* this is here because I will put a message to node inbuffer */
1079   tCURRTIME = 0.0;
1080   tSTARTTIME = CmiWallTimer();
1081
1082   /* creat work threads */
1083   for (i=0; i< cva(bgMach).numWth; i++)
1084   {
1085     threadInfo *tinfo = ninfo->threadinfo[i];
1086     t = CthCreate((CthVoidFn)run_thread, tinfo, cva(bgMach).stacksize);
1087     if (t == NULL) CmiAbort("BG> Failed to create worker thread. \n");
1088     tinfo->setThread(t);
1089     /* put to thread table */
1090     tTHREADTABLE[tinfo->id] = t;
1091     CthAwaken(t);
1092   }
1093
1094   /* creat communication thread */
1095   for (i=0; i< cva(bgMach).numCth; i++)
1096   {
1097     threadInfo *tinfo = ninfo->threadinfo[i+cva(bgMach).numWth];
1098     t = CthCreate((CthVoidFn)run_thread, tinfo, cva(bgMach).stacksize);
1099     if (t == NULL) CmiAbort("BG> Failed to create communication thread. \n");
1100     tinfo->setThread(t);
1101     /* put to thread table */
1102     tTHREADTABLE[tinfo->id] = t;
1103     CthAwaken(t);
1104   }
1105
1106 }
1107
1108 static void beginExitHandlerFunc(void *msg);
1109 static void writeToDisk();
1110 static void sendCorrectionStats();
1111
1112 static CmiHandler exitHandlerFunc(char *msg)
1113 {
1114   // TODO: free memory before exit
1115   int i,j;
1116
1117   programExit = 2;
1118 #if BLUEGENE_TIMING
1119   // timing
1120   if (0)        // detail
1121   if (genTimeLog) {
1122     for (j=0; j<cva(numNodes); j++)
1123     for (i=0; i<cva(bgMach).numWth; i++) {
1124       BgTimeLine &log = cva(nodeinfo)[j].timelines[i].timeline; 
1125 //      BgPrintThreadTimeLine(nodeInfo::Local2Global(j), i, log);
1126       int x,y,z;
1127       nodeInfo::Local2XYZ(j, &x, &y, &z);
1128       BgWriteThreadTimeLine(arg_argv[0], x, y, z, i, log);
1129     }
1130
1131   }
1132 #endif
1133   if (genTimeLog) sendCorrectionStats();
1134
1135   if (genTimeLog) writeToDisk();
1136
1137 //  if (tTHREADTYPE == WORK_THREAD)
1138   {
1139   int origPe = -2;
1140   // close all tracing modules
1141   for (j=0; j<cva(numNodes); j++)
1142     for (i=0; i<cva(bgMach).numWth; i++) {
1143       int oldPe = CmiSwitchToPE(nodeInfo::Local2Global(j)*cva(bgMach).numWth+i);
1144       if (origPe == -2) origPe = oldPe;
1145       traceCharmClose();
1146 //      CmiSwitchToPE(oldPe);
1147     }
1148     if (origPe!=-2) CmiSwitchToPE(origPe);
1149   }
1150
1151 #if 0
1152   delete [] cva(nodeinfo);
1153   delete [] cva(inBuffer);
1154   for (i=0; i<cva(numNodes); i++) CmmFree(cva(msgBuffer)[i]);
1155   delete [] cva(msgBuffer);
1156 #endif
1157
1158   //ConverseExit();
1159   if (genTimeLog)
1160     { if (CmiMyPe() != 0) CsdExitScheduler(); }
1161   else
1162     CsdExitScheduler();
1163
1164   //if (CmiMyPe() == 0) CmiPrintf("\nBG> BlueGene emulator shutdown gracefully!\n");
1165
1166   return 0;
1167 }
1168
1169 static void sanityCheck()
1170 {
1171   if (cva(bgMach).x==0 || cva(bgMach).y==0 || cva(bgMach).z==0)  {
1172     if (CmiMyPe() == 0)
1173       CmiPrintf("\nMissing parameters for BlueGene machine size!\n<tip> use command line options: +x, +y, or +z.\n");
1174     BgShutdown(); 
1175   } 
1176   else if (cva(bgMach).numCth==0 || cva(bgMach).numWth==0) { 
1177 #if 1
1178     if (cva(bgMach).numCth==0) cva(bgMach).numCth=1;
1179     if (cva(bgMach).numWth==0) cva(bgMach).numWth=1;
1180 #else
1181     if (CmiMyPe() == 0)
1182       CmiPrintf("\nMissing parameters for number of communication/worker threads!\n<tip> use command line options: +cth or +wth.\n");
1183     BgShutdown(); 
1184 #endif
1185   }
1186   else if (cva(bgMach).getNodeSize()<CmiNumPes()) {
1187     CmiAbort("\nToo few BlueGene nodes!\n");
1188   }
1189 }
1190
1191 #undef CmiSwitchToPE
1192 extern "C" int CmiSwitchToPEFn(int pe);
1193
1194 // main
1195 CmiStartFn bgMain(int argc, char **argv)
1196 {
1197   int i;
1198   char *configFile = NULL;
1199
1200 #if CMK_CONDS_USE_SPECIAL_CODE
1201   // overwrite possible implementation in machine.c
1202   CmiSwitchToPE = CmiSwitchToPEFn;
1203 #endif
1204
1205   /* initialize all processor level data */
1206   CpvInitialize(BGMach,bgMach);
1207   cva(bgMach).nullify();
1208
1209   CmiArgGroup("Charm++","BlueGene Simulator");
1210   if (CmiGetArgStringDesc(argv, "+bgconfig", &configFile, "BlueGene machine config file")) {
1211    cva(bgMach). read(configFile);
1212   }
1213   CmiGetArgIntDesc(argv, "+x", &cva(bgMach).x, 
1214                 "The x size of the grid of nodes");
1215   CmiGetArgIntDesc(argv, "+y", &cva(bgMach).y, 
1216                 "The y size of the grid of nodes");
1217   CmiGetArgIntDesc(argv, "+z", &cva(bgMach).z, 
1218                 "The z size of the grid of nodes");
1219   CmiGetArgIntDesc(argv, "+cth", &cva(bgMach).numCth, 
1220                 "The number of simulated communication threads per node");
1221   CmiGetArgIntDesc(argv, "+wth", &cva(bgMach).numWth, 
1222                 "The number of simulated worker threads per node");
1223
1224   CmiGetArgIntDesc(argv, "+bgstacksize", &cva(bgMach).stacksize, 
1225                 "Blue Gene thread stack size");
1226
1227   if (CmiGetArgFlagDesc(argv, "+bglog", "Write events to log file"))
1228      genTimeLog = 1;
1229   if (CmiGetArgFlagDesc(argv, "+bgcorrect", "Apply timestamp correction to logs"))
1230     correctTimeLog = 1;
1231   schedule_flag = 0;
1232   if (correctTimeLog) {
1233     genTimeLog = 1;
1234     schedule_flag = 1;
1235   }
1236
1237   if (CmiGetArgFlagDesc(argv, "+bgverbose", "Print debug info verbosely"))
1238     bgverbose = 1;
1239
1240   // for timing method, default using elapse calls.
1241   if(CmiGetArgFlagDesc(argv, "+bgelapse", 
1242                        "Use user provided BgElapse for time prediction")) 
1243       cva(bgMach).timingMethod = BG_ELAPSE;
1244   if(CmiGetArgFlagDesc(argv, "+bgwalltime", 
1245                        "Use walltime method for time prediction")) 
1246       cva(bgMach).timingMethod = BG_WALLTIME;
1247 #ifdef CMK_ORIGIN2000
1248   if(CmiGetArgFlagDesc(argv, "+bgcounter", "Use performance counter")) 
1249       cva(bgMach).timingMethod = BG_COUNTER;
1250 #elif CMK_HAS_COUNTER_PAPI
1251   if (CmiGetArgFlagDesc(argv, "+bgpapi", "Use PAPI Performance counters")) {
1252     cva(bgMach).timingMethod = BG_COUNTER;
1253   }
1254   if (cva(bgMach).timingMethod == BG_COUNTER) {
1255     init_counters();
1256   }
1257 #endif
1258   CmiGetArgDoubleDesc(argv,"+bgcpufactor", &cva(bgMach).cpufactor, 
1259                       "scale factor for wallclock time measured");
1260   
1261   char *networkModel;
1262   if (CmiGetArgStringDesc(argv, "+bgnetwork", &networkModel, "Network model")) {
1263    cva(bgMach).setNetworkModel(networkModel);
1264   }
1265
1266   bgcorroff = 0;
1267   if(CmiGetArgFlagDesc(argv, "+bgcorroff", "Start with correction off")) 
1268     bgcorroff = 1;
1269
1270   bgstats_flag=0;
1271   if(CmiGetArgFlagDesc(argv, "+bgstats", "Print correction statistics")) 
1272     bgstats_flag = 1;
1273
1274   if (CmiGetArgStringDesc(argv, "+bgtraceroot", &cva(bgMach).traceroot, "Directory to write bgTrace files to"))
1275   {
1276     char *root = (char*)malloc(strlen(cva(bgMach).traceroot) + 10);
1277     sprintf(root, "%s/", cva(bgMach).traceroot);
1278     cva(bgMach).traceroot = root;
1279   }
1280
1281 #if BLUEGENE_DEBUG_LOG
1282   {
1283     char ln[200];
1284     sprintf(ln,"bgdebugLog.%d",CmiMyPe());
1285     bgDebugLog=fopen(ln,"w");
1286   }
1287 #endif
1288
1289   arg_argv = argv;
1290   arg_argc = CmiGetArgc(argv);
1291
1292   /* msg handler */
1293   CpvInitialize(SimState, simState);
1294   cva(simState).msgHandler = CmiRegisterHandler((CmiHandler) msgHandlerFunc);
1295   cva(simState).nBcastMsgHandler = CmiRegisterHandler((CmiHandler)nodeBCastMsgHandlerFunc);
1296   cva(simState).tBcastMsgHandler = CmiRegisterHandler((CmiHandler)threadBCastMsgHandlerFunc);
1297   cva(simState).exitHandler = CmiRegisterHandler((CmiHandler) exitHandlerFunc);
1298
1299   cva(simState).beginExitHandler = CmiRegisterHandler((CmiHandler) beginExitHandlerFunc);
1300   cva(simState).inEmulatorInit = 1;
1301   /* call user defined BgEmulatorInit */
1302   BgEmulatorInit(arg_argc, arg_argv);
1303   cva(simState).inEmulatorInit = 0;
1304
1305   /* check if all bluegene node size and thread information are set */
1306   sanityCheck();
1307
1308   _bgSize = cva(bgMach).getNodeSize(); 
1309
1310   timerFunc = BgGetTime;
1311
1312   BgInitTiming();               // timing module
1313
1314   if (CmiMyPe() == 0) {
1315     CmiPrintf("BG info> Simulating %dx%dx%d nodes with %d comm + %d work threads each.\n", cva(bgMach).x, cva(bgMach).y, cva(bgMach).z, cva(bgMach).numCth, cva(bgMach).numWth);
1316     CmiPrintf("BG info> Network type: %s.\n", cva(bgMach).network->name());
1317     cva(bgMach).network->print();
1318     CmiPrintf("BG info> cpufactor is %f.\n", cva(bgMach).cpufactor);
1319     CmiPrintf("BG info> floating point factor is %f.\n", cva(bgMach).fpfactor);
1320     if (cva(bgMach).stacksize)
1321       CmiPrintf("BG info> BG stack size: %d bytes. \n", cva(bgMach).stacksize);
1322     if (cva(bgMach).timingMethod == BG_ELAPSE) 
1323       CmiPrintf("BG info> Using BgElapse calls for timing method. \n");
1324     else if (cva(bgMach).timingMethod == BG_WALLTIME)
1325       CmiPrintf("BG info> Using WallTimer for timing method. \n");
1326     else if (cva(bgMach).timingMethod == BG_COUNTER)
1327       CmiPrintf("BG info> Using performance counter for timing method. \n");
1328     if (genTimeLog)
1329       CmiPrintf("BG info> Generating timing log. \n");
1330     if (correctTimeLog)
1331       CmiPrintf("BG info> Perform timestamp correction. \n");
1332     if (cva(bgMach).traceroot)
1333       CmiPrintf("BG info> bgTrace root is '%s'. \n", cva(bgMach).traceroot);
1334   }
1335
1336   CtvInitialize(threadInfo *, threadinfo);
1337
1338   /* number of bg nodes on this PE */
1339   CpvInitialize(int, numNodes);
1340   cva(numNodes) = nodeInfo::numLocalNodes();
1341
1342   initLock = CmiCreateLock();     // used for BnvInitialize
1343
1344   bgstreaming.init(cva(numNodes));
1345
1346   /* create BG nodes */
1347   CpvInitialize(nodeInfo *, nodeinfo);
1348   cva(nodeinfo) = new nodeInfo[cva(numNodes)];
1349   _MEMCHECK(cva(nodeinfo));
1350
1351   cta(threadinfo) = new threadInfo(-1, UNKNOWN_THREAD, NULL);
1352   _MEMCHECK(cta(threadinfo));
1353
1354   /* create BG processors for each node */
1355   for (i=0; i<cva(numNodes); i++)
1356   {
1357     nodeInfo *ninfo = cva(nodeinfo) + i;
1358     // create threads
1359     ninfo->initThreads(i);
1360
1361     /* pretend that I am a thread */
1362     cta(threadinfo)->myNode = ninfo;
1363
1364     /* initialize a BG node and fire all threads */
1365     BgNodeInitialize(ninfo);
1366   }
1367   // clear main thread.
1368   cta(threadinfo)->myNode = NULL;
1369   CpvInitialize(CthThread, mainThread);
1370   cva(mainThread) = CthSelf();
1371
1372   CpvInitialize(int, CthResumeBigSimThreadIdx);
1373
1374   cva(simState).simStartTime = CmiWallTimer();
1375
1376   return 0;
1377 }
1378
1379 // for conv-conds:
1380 // if -2 untouch
1381 // if -1 main thread
1382 #if CMK_BLUEGENE_THREAD
1383 extern "C" int CmiSwitchToPEFn(int pe)
1384 {
1385   if (pe == -2) return -2;
1386   int oldpe;
1387 //  ASSERT(tTHREADTYPE != COMM_THREAD);
1388   if (tMYNODE == NULL) oldpe = -1;
1389   else if (tTHREADTYPE == COMM_THREAD) oldpe = -BgGetThreadID();
1390   else if (tTHREADTYPE == WORK_THREAD) oldpe = BgGetGlobalWorkerThreadID();
1391   else oldpe = -1;
1392 //CmiPrintf("CmiSwitchToPE from %d to %d\n", oldpe, pe);
1393   if (pe == -1) {
1394     CthSwitchThread(cva(mainThread));
1395   }
1396   else if (pe < 0) {
1397   }
1398   else {
1399     int t = pe%cva(bgMach).numWth;
1400     int newpe = nodeInfo::Global2Local(pe/cva(bgMach).numWth);
1401     nodeInfo *ninfo = cva(nodeinfo) + newpe;;
1402     threadInfo *tinfo = ninfo->threadinfo[t];
1403     CthSwitchThread(tinfo->me);
1404   }
1405   return oldpe;
1406 }
1407 #else
1408 extern "C" int CmiSwitchToPEFn(int pe)
1409 {
1410   if (pe == -2) return -2;
1411   int oldpe;
1412   if (tMYNODE == NULL) oldpe = -1;
1413   else oldpe = BgMyNode();
1414   if (pe == -1) {
1415     cta(threadinfo)->myNode = NULL;
1416   }
1417   else {
1418     int newpe = nodeInfo::Global2Local(pe);
1419     cta(threadinfo)->myNode = cva(nodeinfo) + newpe;;
1420   }
1421   return oldpe;
1422 }
1423 #endif
1424
1425
1426 /*****************************************************************************
1427                         TimeLog correction
1428 *****************************************************************************/
1429
1430 extern void processCorrectionMsg(int nodeidx);
1431
1432 // return the msg pointer, and the index of the message in the affinity queue.
1433 static inline char* searchInAffinityQueue(int nodeidx, BgMsgID &msgId, CmiInt2 tID, int &index)
1434 {
1435   CmiAssert(tID != ANYTHREAD);
1436   ckMsgQueue &affinityQ = cva(nodeinfo)[nodeidx].affinityQ[tID];
1437   for (int i=0; i<affinityQ.length(); i++)  {
1438       char *msg = affinityQ[i];
1439       BgMsgID md = BgMsgID(CmiBgMsgSrcPe(msg), CmiBgMsgID(msg));
1440       if (msgId == md) {
1441         index = i;
1442         return msg;
1443       }
1444   }
1445   return NULL;
1446 }
1447
1448 // return the msg pointer, thread id and the index of the message in the affinity queue.
1449 static char* searchInAffinityQueueInNode(int nodeidx, BgMsgID &msgId, CmiInt2 &tID, int &index)
1450 {
1451   for (tID=0; tID<cva(bgMach).numWth; tID++) {
1452     char *msg = searchInAffinityQueue(nodeidx, msgId, tID, index);
1453     if (msg) return msg;
1454   }
1455   return NULL;
1456 }
1457
1458 StateCounters stateCounters;
1459
1460 int updateRealMsgs(bgCorrectionMsg *cm, int nodeidx)
1461 {
1462   char *msg;
1463   CmiInt2 tID = cm->tID;
1464   int index;
1465   if (tID == ANYTHREAD) {
1466     msg = searchInAffinityQueueInNode(nodeidx, cm->msgId, tID, index);
1467   }
1468   else {
1469     msg = searchInAffinityQueue(nodeidx, cm->msgId, tID, index);
1470   }
1471   if (msg == NULL) return 0;
1472
1473   ckMsgQueue &affinityQ = cva(nodeinfo)[nodeidx].affinityQ[tID];
1474   CmiBgMsgRecvTime(msg) = cm->tAdjust;
1475   affinityQ.update(index);
1476   CthThread tid = cva(nodeinfo)[nodeidx].threadTable[tID];
1477   unsigned int prio = (unsigned int)(cm->tAdjust*PRIO_FACTOR)+1;
1478   CthAwakenPrio(tid, CQS_QUEUEING_IFIFO, sizeof(int), &prio);
1479   stateCounters.corrMsgCRCnt++;
1480   return 1;       /* invalidate this msg */
1481 }
1482
1483 extern void processBufferCorrectionMsgs(void *ignored);
1484
1485 // Coverse handler for begin exit
1486 // flush and process all correction messages
1487 static void beginExitHandlerFunc(void *msg)
1488 {
1489   CmiFree(msg);
1490   delayCheckFlag = 0;
1491 //CmiPrintf("\n\n\nbeginExitHandlerFunc called on %d\n", CmiMyPe());
1492   programExit = 1;
1493 #if LIMITED_SEND
1494   CQdCreate(CpvAccess(cQdState), BgNodeSize());
1495 #endif
1496
1497 #if 1
1498   for (int i=0; i<BgNodeSize(); i++) 
1499     processCorrectionMsg(i); 
1500 #if USE_MULTISEND
1501   BgSendBufferedCorrMsgs();
1502 #endif
1503 #else
1504   // the msg queue should be empty now.
1505   // don't do insert adjustment, but start to do all timing correction from here
1506   int nodeidx, tID;
1507   for (nodeidx=0; nodeidx<BgNodeSize(); nodeidx++) {
1508     BgTimeLineRec *tlines = cva(nodeinfo)[nodeidx].timelines;
1509     for (tID=0; tID<cva(numWth); tID++) {
1510         BgTimeLineRec &tlinerec = tlines[tID];
1511         BgAdjustTimeLine(tlinerec, nodeidx, tID);
1512     }
1513   }
1514
1515 #endif
1516
1517 #if !THROTTLE_WORK
1518 #if DELAY_CHECK
1519   CcdCallFnAfter(processBufferCorrectionMsgs,NULL,CHECK_INTERVAL);
1520 #endif
1521 #endif
1522 }
1523
1524 #define HISTOGRAM_SIZE  100
1525 // compute total CPU utilization for each timeline and 
1526 // return the number of real msgs
1527 void computeUtilForAll(int* array, int *nReal)
1528 {
1529   double scale = 1.0e3;         // scale to ms
1530
1531   //We measure from 1ms to 5001 ms in steps of 100 ms
1532   int min = 0, step = 1;
1533   int max = min + HISTOGRAM_SIZE*step;
1534   // [min, max: step]  HISTOGRAM_SIZE slots
1535
1536   if (CmiMyPe()==0)
1537     CmiPrintf("computeUtilForAll: min:%d max:%d step:%d scale:%f.\n", min, max, step, scale);
1538   int size = (max-min)/step;
1539   CmiAssert(size == HISTOGRAM_SIZE);
1540   int allmin = -1, allmax = -1;
1541   for(int i=0;i<size;i++) array[i] = 0;
1542
1543   for (int nodeidx=0; nodeidx<cva(numNodes); nodeidx++) {
1544     BgTimeLineRec *tlinerec = cva(nodeinfo)[nodeidx].timelines;
1545     for (int tID=0; tID<cva(bgMach).numWth; tID++) {
1546       int util = (int)(scale*(tlinerec[tID].computeUtil(nReal)));
1547
1548       if (util >= max) { if (util>allmax||allmax==-1) allmax=util; util=max-1;}
1549       if (util < min) { if (util<allmin||allmin==-1) allmin=util; util=min; }
1550       array[(util-min)/step]++;
1551     }
1552   }
1553   if (allmin!=-1 || allmax!=-1)
1554     CmiPrintf("[%d] Warning: computeUtilForAll out of range %f - %f.\n", CmiMyPe(), (allmin==-1)?-1:allmin/scale, (allmax==-1)?-1:allmax/scale);
1555 }
1556
1557 class StatsMessage {
1558   char core[CmiBlueGeneMsgHeaderSizeBytes];
1559 public:
1560   int processCount;
1561   int corrMsgCount;
1562   int realMsgCount;
1563   int maxTimelineLen, minTimelineLen;
1564 };
1565
1566 extern int processCount, corrMsgCount;
1567
1568 static void sendCorrectionStats()
1569 {
1570   int msgSize = sizeof(StatsMessage)+sizeof(int)*HISTOGRAM_SIZE;
1571   StatsMessage *statsMsg = (StatsMessage *)CmiAlloc(msgSize);
1572   statsMsg->processCount = processCount;
1573   statsMsg->corrMsgCount = corrMsgCount;
1574   int numMsgs=0;
1575   int maxTimelineLen=-1, minTimelineLen=CMK_MAXINT;
1576   int totalMem = 0;
1577   if (bgstats_flag) {
1578   for (int nodeidx=0; nodeidx<cva(numNodes); nodeidx++) {
1579     BgTimeLineRec *tlines = cva(nodeinfo)[nodeidx].timelines;
1580     for (int tID=0; tID<cva(bgMach).numWth; tID++) {
1581         BgTimeLineRec &tlinerec = tlines[tID];
1582         int tlen = tlinerec.length();
1583         if (tlen>maxTimelineLen) maxTimelineLen=tlen;
1584         if (tlen<minTimelineLen) minTimelineLen=tlen;
1585         totalMem = tlen*sizeof(BgTimeLog);
1586 //CmiPrintf("[%d node:%d] BgTimeLog: %dK len:%d size of bglog: %d bytes\n", CmiMyPe(), nodeidx, totalMem/1000, tlen, sizeof(BgTimeLog));
1587 #if 0
1588         for (int i=0; i< tlinerec.length(); i++) {
1589           numMsgs += tlinerec[i]->msgs.length();
1590         }
1591 #endif
1592     }
1593   }
1594   computeUtilForAll((int*)(statsMsg+1), &numMsgs);
1595   statsMsg->realMsgCount = numMsgs;
1596   statsMsg->maxTimelineLen = maxTimelineLen;
1597   statsMsg->minTimelineLen = minTimelineLen;
1598   }  // end if
1599
1600   CmiSetHandler(statsMsg, cva(simState).bgStatCollectHandler);
1601   CmiSyncSendAndFree(0, msgSize, statsMsg);
1602 }
1603
1604 // Converse handler for collecting stats
1605 void statsCollectionHandlerFunc(void *msg)
1606 {
1607   static int count=0;
1608   static int pc=0, cc=0, realMsgCount=0;
1609   static int maxTimelineLen=0, minTimelineLen=CMK_MAXINT;
1610   static int *histArray = NULL;
1611   int i;
1612
1613   count++;
1614   if (histArray == NULL) {
1615     histArray = new int[HISTOGRAM_SIZE];
1616     for (i=0; i<HISTOGRAM_SIZE; i++) histArray[i]=0;
1617   }
1618   StatsMessage *m = (StatsMessage *)msg;
1619   pc += m->processCount;
1620   cc += m->corrMsgCount;
1621   realMsgCount += m->realMsgCount;
1622   if (minTimelineLen> m->minTimelineLen) minTimelineLen=m->minTimelineLen;
1623   if (maxTimelineLen< m->maxTimelineLen) maxTimelineLen=m->maxTimelineLen;
1624   int *array = (int *)(m+1);
1625   for (i=0; i<HISTOGRAM_SIZE; i++) histArray[i] += array[i];
1626   if (count == CmiNumPes()) {
1627     if (bgstats_flag) {
1628       CmiPrintf("Total procCount:%d corrMsgCount:%d realMsg:%d timeline:%d-%d\n", pc, cc, realMsgCount, minTimelineLen, maxTimelineLen);
1629       for (i=0; i<HISTOGRAM_SIZE; i++) {
1630         CmiPrintf("%d ", histArray[i]);
1631         if (i%20 == 19) CmiPrintf("\n");
1632       }
1633       CmiPrintf("\n");
1634     }
1635     CsdExitScheduler();
1636   }
1637   CmiFree(msg);
1638 }
1639
1640 // update arrival time from buffer messages
1641 // before start an entry, check message time against buffered timing
1642 // correction message to update to the correct time.
1643 void correctMsgTime(char *msg)
1644 {
1645    if (!correctTimeLog) return;
1646 //   if (CkMsgDoCorrect(msg) == 0) return;
1647
1648    BgMsgID msgId(CmiBgMsgSrcPe(msg), CmiBgMsgID(msg));
1649    CmiInt2 tid = CmiBgMsgThreadID(msg);
1650
1651    bgCorrectionQ &cmsg = cva(nodeinfo)[tMYNODEID].cmsg;
1652    int len = cmsg.length();
1653    for (int i=0; i<len; i++) {
1654      bgCorrectionMsg* m = cmsg[i];
1655      if (msgId == m->msgId && tid == m->tID) {
1656         if (m->tAdjust < 0.0) return;
1657         //CmiPrintf("correctMsgTime from %e to %e\n", CmiBgMsgRecvTime(msg), m->tAdjust);
1658         CmiBgMsgRecvTime(msg) = m->tAdjust;
1659         m->tAdjust = -1.0;       /* invalidate this msg */
1660 //      cmsg.update(i);
1661         stateCounters.corrMsgRCCnt++;
1662         break;
1663      }
1664    }
1665 }
1666
1667
1668 //TODO: write disk bgTraceFiles
1669 static void writeToDisk()
1670 {
1671
1672   char* d = new char[1025];
1673   //Num of simulated procs on this real pe
1674   int numLocalProcs = cva(numNodes)*cva(bgMach).numWth;
1675
1676   const PUP::machineInfo &machInfo = PUP::machineInfo::current();
1677
1678   // write summary file on PE0
1679   if(CmiMyPe()==0){
1680     
1681     sprintf(d, "%sbgTrace", cva(bgMach).traceroot?cva(bgMach).traceroot:""); 
1682     FILE *f2 = fopen(d,"w");
1683     //Total emulating processors and total target BG processors
1684     int numPes=CmiNumPes();
1685     int totalProcs = BgNumNodes()*cva(bgMach).numWth;
1686
1687     if(f2==NULL) {
1688       CmiPrintf("[%d] Creating trace file %s  failed\n", CmiMyPe(), d);
1689       CmiAbort("BG> Abort");
1690     }
1691     PUP::toDisk p(f2);
1692     p((char *)&machInfo, sizeof(machInfo));
1693     p|totalProcs;
1694     p|cva(bgMach);
1695     p|numPes;
1696     p|bglog_version;
1697     p|CpvAccess(CthResumeBigSimThreadIdx);
1698     
1699     CmiPrintf("[0] Number is numX:%d numY:%d numZ:%d numCth:%d numWth:%d numPes:%d totalProcs:%d bglog_ver:%d\n",cva(bgMach).x,cva(bgMach).y,cva(bgMach).z,cva(bgMach).numCth,cva(bgMach).numWth,numPes,totalProcs,bglog_version);
1700     
1701     fclose(f2);
1702   }
1703   
1704   sprintf(d, "%sbgTrace%d", cva(bgMach).traceroot?cva(bgMach).traceroot:"", CmiMyPe()); 
1705   FILE *f = fopen(d,"w");
1706  
1707   if(f==NULL)
1708     CmiPrintf("Creating bgTrace%d failed\n",CmiMyPe());
1709   PUP::toDisk p(f);
1710   
1711   p((char *)&machInfo, sizeof(machInfo));       // machine info
1712   p|numLocalProcs;
1713
1714   // CmiPrintf("Timelines are: \n");
1715   int procTablePos = ftell(f);
1716
1717   int *procOffsets = new int[numLocalProcs];
1718   int procTableSize = (numLocalProcs)*sizeof(int);
1719   fseek(f,procTableSize,SEEK_CUR); 
1720
1721   for (int j=0; j<cva(numNodes); j++){
1722     for(int i=0;i<cva(bgMach).numWth;i++){
1723       BgTimeLineRec &t = cva(nodeinfo)[j].timelines[i];
1724       procOffsets[j*cva(bgMach).numWth + i] = ftell(f);
1725       t.pup(p);
1726     }
1727   }
1728   
1729   fseek(f,procTablePos,SEEK_SET);
1730   p(procOffsets,numLocalProcs);
1731   fclose(f);
1732
1733   CmiPrintf("[%d] Wrote to disk for %d BG nodes. \n", CmiMyPe(), cva(numNodes));
1734 }
1735
1736
1737 // application Converse thread hook
1738
1739 CpvExtern(int      , CthResumeBigSimThreadIdx);
1740
1741 void CthEnqueueBigSimThread(CthThreadToken* token, int s,
1742                                    int pb,unsigned int *prio)
1743 {
1744 /*
1745   CmiSetHandler(token, CpvAccess(CthResumeNormalThreadIdx));
1746   CsdEnqueueGeneral(token, s, pb, prio);
1747 */
1748 #if CMK_BLUEGENE_THREAD
1749   int x, y, z;
1750   BgGetMyXYZ(&x, &y, &z);
1751   int t = BgGetThreadID();
1752 #else
1753   #error "ERROR HERE"
1754 #endif
1755     // local message into queue
1756   BgSendPacket(x,y,z, t, CpvAccess(CthResumeBigSimThreadIdx), LARGE_WORK, sizeof(CthThreadToken), (char *)token);
1757
1758
1759 CthThread CthSuspendBigSimThread()
1760
1761   return  cta(threadinfo)->me;
1762 }
1763
1764
1765 void BgSetStrategyBigSimDefault(CthThread t)
1766
1767   CthSetStrategy(t,
1768                  CthEnqueueBigSimThread,
1769                  CthSuspendBigSimThread);
1770 }
1771