0ab9a2a5712c4e5faac43901891393a1b4f2aedf
[charm.git] / src / arch / gemini_gni / machine.c
1 /*****************************************************************************
2  * $Source$
3  * $Author$  Yanhua Sun
4  * $Date$  07-01-2011
5  * $Revision$ 
6  *****************************************************************************/
7
8 /** @file
9  * Gemini GNI machine layer
10  */
11 /*@{*/
12
13 #include <stdio.h>
14 #include <stdlib.h>
15 #include <stdint.h>
16 #include <errno.h>
17 #include <malloc.h>
18
19 #include "gni_pub.h"
20 #include "pmi.h"
21
22 #include "converse.h"
23
24 /*Support for ++debug: */
25 #if defined(_WIN32) && ! defined(__CYGWIN__)
26 #include <windows.h>
27 #include <wincon.h>
28 #include <sys/types.h>
29 #include <sys/timeb.h>
30
31 static void sleep(int secs) {
32     Sleep(1000*secs);
33 }
34 #else
35 #include <unistd.h> /*For getpid()*/
36 #endif
37
38
39 #define REMOTE_EVENT         0
40 #define USE_LRTS_MEMPOOL   1
41
42 #if USE_LRTS_MEMPOOL
43 static size_t _mempool_size = 1024ll*1024*32;
44 #endif
45
46 #define PRINT_SYH  0
47
48 #if PRINT_SYH
49 int         lrts_smsg_success = 0;
50 int         lrts_send_msg_id = 0;
51 int         lrts_send_rdma_id = 0;
52 int         lrts_send_rdma_success = 0;
53 int         lrts_received_msg = 0;
54 int         lrts_local_done_msg = 0;
55 #endif
56
57 #include "machine.h"
58
59 #include "pcqueue.h"
60
61 #if CMK_PERSISTENT_COMM
62 #include "machine-persistent.h"
63 #endif
64
65 //#define  USE_ONESIDED 1
66 #ifdef USE_ONESIDED
67 //onesided implementation is wrong, since no place to restore omdh
68 #include "onesided.h"
69 onesided_hnd_t   onesided_hnd;
70 onesided_md_t    omdh;
71 #define MEMORY_REGISTER(handler, nic_hndl, msg, size, mem_hndl, myomdh)  omdh. onesided_mem_register(handler, (uint64_t)msg, size, 0, myomdh) 
72
73 #define MEMORY_DEREGISTER(handler, nic_hndl, mem_hndl, myomdh) onesided_mem_deregister(handler, myomdh)
74
75 #else
76 uint8_t   onesided_hnd, omdh;
77 #if REMOTE_EVENT
78 #define  MEMORY_REGISTER(handler, nic_hndl, msg, size, mem_hndl, myomdh) GNI_MemRegister(nic_hndl, (uint64_t)msg,  (uint64_t)size, smsg_rx_cqh,  GNI_MEM_READWRITE, -1, mem_hndl)
79 #else
80 #define  MEMORY_REGISTER(handler, nic_hndl, msg, size, mem_hndl, myomdh) GNI_MemRegister(nic_hndl, (uint64_t)msg,  (uint64_t)size, NULL,  GNI_MEM_READWRITE, -1, mem_hndl)
81 #endif
82 #define  MEMORY_DEREGISTER(handler, nic_hndl, mem_hndl, myomdh)  GNI_MemDeregister(nic_hndl, (mem_hndl))
83 #endif
84
85 #define CmiGetMsgSize(m)  ((CmiMsgHeaderExt*)m)->size
86 #define CmiSetMsgSize(m,s)  ((((CmiMsgHeaderExt*)m)->size)=(s))
87
88 #define ALIGNBUF                64
89
90 /* =======Beginning of Definitions of Performance-Specific Macros =======*/
91 /* If SMSG is not used */
92 #define FMA_PER_CORE  1024
93 #define FMA_BUFFER_SIZE 1024
94 /* If SMSG is used */
95 static int  SMSG_MAX_MSG;
96 //static int  log2_SMSG_MAX_MSG;
97 #define SMSG_MAX_CREDIT  36
98
99 #define MSGQ_MAXSIZE       2048
100 /* large message transfer with FMA or BTE */
101 #define LRTS_GNI_RDMA_THRESHOLD  2048
102
103 #define REMOTE_QUEUE_ENTRIES  2048 
104 #define LOCAL_QUEUE_ENTRIES   64 
105
106 #define PUT_DONE_TAG      0x29
107 #define ACK_TAG           0x30
108 /* SMSG is data message */
109 #define SMALL_DATA_TAG          0x31
110 /* SMSG is a control message to initialize a BTE */
111 #define MEDIUM_HEAD_TAG         0x32
112 #define MEDIUM_DATA_TAG         0x33
113 #define LMSG_INIT_TAG     0x39 
114
115 #define DEBUG
116 #ifdef GNI_RC_CHECK
117 #undef GNI_RC_CHECK
118 #endif
119 #ifdef DEBUG
120 #define GNI_RC_CHECK(msg,rc) do { if(rc != GNI_RC_SUCCESS) {           CmiPrintf("[%d] %s; err=%s\n",CmiMyPe(),msg,gni_err_str[rc]); CmiAbort("GNI_RC_CHECK"); } } while(0)
121 #else
122 #define GNI_RC_CHECK(msg,rc)
123 #endif
124
125 #define ALIGN64(x)       (size_t)((~63)&((x)+63))
126 #define ALIGN4(x)        (size_t)((~3)&((x)+3)) 
127
128 static int Mempool_MaxSize = 1024*1024*128;
129 static int useDynamicSMSG   = 0;
130 static int useStaticMSGQ = 0;
131 static int useStaticFMA = 0;
132 static int mysize, myrank;
133 static gni_nic_handle_t      nic_hndl;
134
135 typedef struct {
136     gni_mem_handle_t mdh;
137     uint64_t addr;
138 } mdh_addr_t;
139 // this is related to dynamic SMSG
140
141 #define     SMSG_CONN_SIZE     24
142 int     *smsg_connected_flag= 0;
143 char    *smsg_connection_addr = 0;
144 mdh_addr_t    *smsg_connection_vec = 0;
145 gni_mem_handle_t    smsg_connection_memhndl;
146
147 static void             *smsg_mailbox_base;
148 gni_msgq_attr_t         msgq_attrs;
149 gni_msgq_handle_t       msgq_handle;
150 gni_msgq_ep_attr_t      msgq_ep_attrs;
151 gni_msgq_ep_attr_t      msgq_ep_attrs_size;
152
153
154
155 /* preallocated DMA buffer */
156 int                     DMA_slots;
157 uint64_t                DMA_avail_tag = 0;
158 uint32_t                DMA_incoming_avail_tag = 0;
159 uint32_t                DMA_outgoing_avail_tag = 0;
160 void                    *DMA_incoming_base_addr;
161 void                    *DMA_outgoing_base_addr;
162 mdh_addr_t              DMA_buffer_base_mdh_addr;
163 mdh_addr_t              *DMA_buffer_base_mdh_addr_vec;
164 int                     DMA_buffer_size;
165 int                     DMA_max_single_msg = 131072;//524288 ;
166
167 #define                 DMA_SIZE_PER_SLOT       8192
168
169
170 typedef struct dma_msgid_map
171 {
172     uint64_t     msg_id;
173     int     msg_subid
174 } dma_msgid_map_t;
175
176 dma_msgid_map_t         *dma_map_list;
177
178 typedef struct msg_trace
179 {
180     uint64_t    msg_id;
181     int         done_num;
182 }msg_trace_t;
183
184 msg_trace_t             *pending_msg_list;
185 /* =====Beginning of Declarations of Machine Specific Variables===== */
186 static int cookie;
187 static int modes = 0;
188 static gni_cq_handle_t       smsg_rx_cqh = NULL;
189 static gni_cq_handle_t       smsg_tx_cqh = NULL;
190 static gni_cq_handle_t       post_rx_cqh = NULL;
191 static gni_cq_handle_t       post_tx_cqh = NULL;
192 static gni_ep_handle_t       *ep_hndl_array;
193
194 typedef    struct  pending_smg
195 {
196     int     inst_id;
197     struct  pending_smg *next;
198 } PENDING_GETNEXT;
199
200
201 typedef struct msg_list
202 {
203     uint32_t destNode;
204     uint32_t size;
205     void *msg;
206     struct msg_list *next;
207     struct msg_list *pehead_next;
208     uint8_t tag;
209 }MSG_LIST;
210
211 typedef struct medium_msg_list
212 {
213     uint32_t destNode;
214     uint32_t msg_id;
215     uint32_t msg_subid;
216     uint32_t remain_size;
217     void *msg;
218     struct medium_msg_list *next;
219 }MEDIUM_MSG_LIST;
220
221
222 typedef struct control_msg
223 {
224     uint64_t            source_addr;
225     int                 source;               /* source rank */
226     int                 length;
227     gni_mem_handle_t    source_mem_hndl;
228     struct control_msg *next;
229 }CONTROL_MSG;
230
231 typedef struct medium_msg_control
232 {
233     uint64_t            dma_offset;     //the dma_buffer for this block of msg
234     int                 msg_id;         //Id for the total index
235     int                 msg_subid;      //offset inside the message id 
236 }MEDIUM_MSG_CONTROL;
237
238 typedef struct  rmda_msg
239 {
240     int                   destNode;
241     gni_post_descriptor_t *pd;
242     struct  rmda_msg      *next;
243 }RDMA_REQUEST;
244
245 typedef struct  msg_list_index
246 {
247     int         next;
248     MSG_LIST    *head;
249     MSG_LIST    *tail;
250 } MSG_LIST_INDEX;
251
252 /* reuse PendingMsg memory */
253 static CONTROL_MSG          *control_freelist=0;
254 static MSG_LIST             *msglist_freelist=0;
255 static int                  smsg_head_index;
256 static MSG_LIST_INDEX       *smsg_msglist_index= 0;
257 static MSG_LIST             *smsg_free_head=0;
258 static MSG_LIST             *smsg_free_tail=0;
259
260 /*
261 #define FreeMsgList(msg_head, msg_tail, free_head, free_tail)       \
262     if(free_head == 0)  free_head = free_tail = msg_head;    \
263     else   free_tail = free_tail->next;    \
264     if( msg_head->next == msg_tail) msg_head =0;   \
265     else msg_head= msg_head->next;    
266
267 #define MallocMsgList(d, msg_head, msg_tail, free_head, free_tail, msgsize) \
268     if(free_head == 0) {d= malloc(msgsize);  \
269         if(msg_head == 0)   msg_head =msg_tail = msg_head->next = msg_tail->next = d; \
270         else { msg_tail->next = d; d->next = msg_head; msg_tail=d;} \
271     }else {d = free_head; free_head = free_head->next; if(free_tail->next == free_head) free_head =0;} \
272 */
273
274 #define FreeMsgList(d)  \
275   (d)->next = msglist_freelist;\
276   msglist_freelist = d;
277
278
279
280 #define MallocMsgList(d) \
281   d = msglist_freelist;\
282   if (d==0) {d = ((MSG_LIST*)malloc(sizeof(MSG_LIST)));\
283              _MEMCHECK(d);\
284   } else msglist_freelist = d->next;
285
286
287 #define FreeControlMsg(d)       \
288   (d)->next = control_freelist;\
289   control_freelist = d;
290
291 #define MallocControlMsg(d) \
292   d = control_freelist;\
293   if (d==0) {d = ((CONTROL_MSG*)malloc(sizeof(CONTROL_MSG)));\
294              _MEMCHECK(d);\
295   } else control_freelist = d->next;
296
297
298 static RDMA_REQUEST         *rdma_freelist = NULL;
299
300 #define FreeControlMsg(d)       \
301   (d)->next = control_freelist;\
302   control_freelist = d;
303
304 #define MallocControlMsg(d) \
305   d = control_freelist;\
306   if (d==0) {d = ((CONTROL_MSG*)malloc(sizeof(CONTROL_MSG)));\
307              _MEMCHECK(d);\
308   } else control_freelist = d->next;
309
310 #define FreeMediumControlMsg(d)       \
311   (d)->next = medium_control_freelist;\
312   medium_control_freelist = d;
313
314
315 #define MallocMediumControlMsg(d) \
316     d = medium_control_freelist;\
317     if (d==0) {d = ((MEDIUM_MSG_CONTROL*)malloc(sizeof(MEDIUM_MSG_CONTROL)));\
318     _MEMCHECK(d);\
319 } else mediumcontrol_freelist = d->next;
320
321 #define FreeRdmaRequest(d)       \
322   (d)->next = rdma_freelist;\
323   rdma_freelist = d;
324
325 #define MallocRdmaRequest(d) \
326   d = rdma_freelist;\
327   if (d==0) {d = ((RDMA_REQUEST*)malloc(sizeof(RDMA_REQUEST)));\
328              _MEMCHECK(d);\
329   } else rdma_freelist = d->next;
330
331 /* reuse gni_post_descriptor_t */
332 static gni_post_descriptor_t *post_freelist=0;
333
334 #if 1
335 #define FreePostDesc(d)       \
336     (d)->next_descr = post_freelist;\
337     post_freelist = d;
338
339 #define MallocPostDesc(d) \
340   d = post_freelist;\
341   if (d==0) { \
342      d = ((gni_post_descriptor_t*)malloc(sizeof(gni_post_descriptor_t)));\
343      _MEMCHECK(d);\
344   } else post_freelist = d->next_descr;
345 #else
346
347 #define FreePostDesc(d)     free(d);
348 #define MallocPostDesc(d)   d = ((gni_post_descriptor_t*)malloc(sizeof(gni_post_descriptor_t))); _MEMCHECK(d);
349
350 #endif
351
352 static PENDING_GETNEXT     *pending_smsg_head = 0;
353 static PENDING_GETNEXT     *pending_smsg_tail = 0;
354
355 /* LrtsSent is called but message can not be sent by SMSGSend because of mailbox full or no credit */
356 static int      buffered_smsg_counter = 0;
357
358 /* SmsgSend return success but message sent is not confirmed by remote side */
359
360 static RDMA_REQUEST  *pending_rdma_head = 0;
361 static RDMA_REQUEST  *pending_rdma_tail = 0;
362 static MSG_LIST *buffered_fma_head = 0;
363 static MSG_LIST *buffered_fma_tail = 0;
364
365 /* functions  */
366 #define IsFree(a,ind)  !( a& (1<<(ind) ))
367 #define SET_BITS(a,ind) a = ( a | (1<<(ind )) )
368 #define Reset(a,ind) a = ( a & (~(1<<(ind))) )
369
370 #include "mempool.c"
371
372 /* get the upper bound of log 2 */
373 int mylog2(int size)
374 {
375     int op = size;
376     unsigned int ret=0;
377     unsigned int mask = 0;
378     int i;
379     while(op>0)
380     {
381         op = op >> 1;
382         ret++;
383
384     }
385     for(i=1; i<ret; i++)
386     {
387         mask = mask << 1;
388         mask +=1;
389     }
390
391     ret -= ((size &mask) ? 0:1);
392     return ret;
393 }
394
395 static void
396 allgather(void *in,void *out, int len)
397 {
398     static int *ivec_ptr=NULL,already_called=0,job_size=0;
399     int i,rc;
400     int my_rank;
401     char *tmp_buf,*out_ptr;
402
403     if(!already_called) {
404
405         rc = PMI_Get_size(&job_size);
406         CmiAssert(rc == PMI_SUCCESS);
407         rc = PMI_Get_rank(&my_rank);
408         CmiAssert(rc == PMI_SUCCESS);
409
410         ivec_ptr = (int *)malloc(sizeof(int) * job_size);
411         CmiAssert(ivec_ptr != NULL);
412
413         rc = PMI_Allgather(&my_rank,ivec_ptr,sizeof(int));
414         CmiAssert(rc == PMI_SUCCESS);
415
416         already_called = 1;
417
418     }
419
420     tmp_buf = (char *)malloc(job_size * len);
421     CmiAssert(tmp_buf);
422
423     rc = PMI_Allgather(in,tmp_buf,len);
424     CmiAssert(rc == PMI_SUCCESS);
425
426     out_ptr = out;
427
428     for(i=0;i<job_size;i++) {
429
430         memcpy(&out_ptr[len * ivec_ptr[i]],&tmp_buf[i * len],len);
431
432     }
433
434     free(tmp_buf);
435 }
436 static void
437 allgather_2(void *in,void *out, int len)
438 {
439     //PMI_Allgather is out of order
440     int i,rc, extend_len;
441     int  rank_index;
442     char *out_ptr, *out_ref;
443     char *in2;
444
445     extend_len = sizeof(int) + len;
446     in2 = (char*)malloc(extend_len);
447
448     memcpy(in2, &myrank, sizeof(int));
449     memcpy(in2+sizeof(int), in, len);
450
451     out_ptr = (char*)malloc(mysize*extend_len);
452
453     rc = PMI_Allgather(in2, out_ptr, extend_len);
454     GNI_RC_CHECK("allgather", rc);
455
456     out_ref = out;
457
458     for(i=0;i<mysize;i++) {
459         //rank index 
460         memcpy(&rank_index, &(out_ptr[extend_len*i]), sizeof(int));
461         //copy to the rank index slot
462         memcpy(&out_ref[rank_index*len], &out_ptr[extend_len*i+sizeof(int)], len);
463     }
464
465     free(out_ptr);
466     free(in2);
467
468 }
469
470 static unsigned int get_gni_nic_address(int device_id)
471 {
472     unsigned int address, cpu_id;
473     gni_return_t status;
474     int i, alps_dev_id=-1,alps_address=-1;
475     char *token, *p_ptr;
476
477     p_ptr = getenv("PMI_GNI_DEV_ID");
478     if (!p_ptr) {
479         status = GNI_CdmGetNicAddress(device_id, &address, &cpu_id);
480        
481         GNI_RC_CHECK("GNI_CdmGetNicAddress", status);
482     } else {
483         while ((token = strtok(p_ptr,":")) != NULL) {
484             alps_dev_id = atoi(token);
485             if (alps_dev_id == device_id) {
486                 break;
487             }
488             p_ptr = NULL;
489         }
490         CmiAssert(alps_dev_id != -1);
491         p_ptr = getenv("PMI_GNI_LOC_ADDR");
492         CmiAssert(p_ptr != NULL);
493         i = 0;
494         while ((token = strtok(p_ptr,":")) != NULL) {
495             if (i == alps_dev_id) {
496                 alps_address = atoi(token);
497                 break;
498             }
499             p_ptr = NULL;
500             ++i;
501         }
502         CmiAssert(alps_address != -1);
503         address = alps_address;
504     }
505     return address;
506 }
507
508 static uint8_t get_ptag(void)
509 {
510     char *p_ptr, *token;
511     uint8_t ptag;
512
513     p_ptr = getenv("PMI_GNI_PTAG");
514     CmiAssert(p_ptr != NULL);
515     token = strtok(p_ptr, ":");
516     ptag = (uint8_t)atoi(token);
517     return ptag;
518         
519 }
520
521 static uint32_t get_cookie(void)
522 {
523     uint32_t cookie;
524     char *p_ptr, *token;
525
526     p_ptr = getenv("PMI_GNI_COOKIE");
527     CmiAssert(p_ptr != NULL);
528     token = strtok(p_ptr, ":");
529     cookie = (uint32_t)atoi(token);
530
531     return cookie;
532 }
533
534 /* =====Beginning of Definitions of Message-Corruption Related Macros=====*/
535 /* TODO: add any that are related */
536 /* =====End of Definitions of Message-Corruption Related Macros=====*/
537
538
539 #include "machine-lrts.h"
540 #include "machine-common-core.c"
541
542 /* Network progress function is used to poll the network when for
543    messages. This flushes receive buffers on some  implementations*/
544 #if CMK_MACHINE_PROGRESS_DEFINED
545 void CmiMachineProgressImpl() {
546 }
547 #endif
548
549 inline
550 static void delay_send_small_msg(void *msg, int size, int destNode, uint8_t tag)
551 {
552     MSG_LIST        *msg_tmp;
553     MallocMsgList(msg_tmp);
554     msg_tmp->destNode = destNode;
555     msg_tmp->size   = size;
556     msg_tmp->msg    = msg;
557     msg_tmp->tag    = tag;
558     msg_tmp->next   = 0;
559     if (smsg_msglist_index[destNode].head == 0 ) {
560         smsg_msglist_index[destNode].head = msg_tmp;
561         smsg_msglist_index[destNode].next = smsg_head_index;
562         smsg_head_index = destNode;
563     }
564     else {
565       (smsg_msglist_index[destNode].tail)->next    = msg_tmp;
566     }
567     smsg_msglist_index[destNode].tail          = msg_tmp;
568 #if PRINT_SYH
569     buffered_smsg_counter++;
570 #endif
571 }
572
573 inline 
574 static gni_return_t send_smsg_message(int destNode, void *header, int size_header, void *msg, int size, uint8_t tag, int inbuff )
575 {
576     gni_return_t status = GNI_RC_NOT_DONE;
577     
578     //if(useDynamicSMSG == 0 ) 
579     if(smsg_msglist_index[destNode].head == 0 || inbuff==1)
580     {
581         status = GNI_SmsgSendWTag(ep_hndl_array[destNode], header, size_header, msg, size, 0, tag);
582         if(status == GNI_RC_SUCCESS)
583         {
584 #if PRINT_SYH
585             lrts_smsg_success++;
586             if(lrts_smsg_success == lrts_send_msg_id)
587                 CmiPrintf("GOOD [%d==>%d] sent done%d (msgs=%d)\n", myrank, destNode, lrts_smsg_success, lrts_send_msg_id);
588             else
589                 CmiPrintf("BAD [%d==>%d] sent done%d (msgs=%d)\n", myrank, destNode, lrts_smsg_success, lrts_send_msg_id);
590 #endif
591             return status;
592         }
593     }else {
594         if(inbuff ==0)
595             delay_send_small_msg(msg, size, destNode, tag);
596     }
597     return status;
598 }
599
600 // Get first 0 in DMA_tags starting from index
601 static int get_first_avail_bit(uint64_t DMA_tags, int start_index)
602 {
603
604     uint64_t         mask = 0x1;
605     register    int     i=0;
606     while((DMA_tags & mask) && i<DMA_slots) {mask << 1; i++;}
607
608 }
609
610 static int send_medium_messages(int destNode, int size, char *msg)
611 {
612 #if 0
613     gni_return_t status = GNI_RC_SUCCESS;
614     int first_avail_bit=0;
615     uint64_t mask = 0x1;
616     MEDIUM_MSG_CONTROL  *medium_msg_control_tmp;
617     MEDIUM_MSG_LIST        *msg_tmp;
618     int blocksize, remain_size, pos;
619     int sub_id = 0;
620     remain_size = size;
621     pos = 0;  //offset before which data are sent
622     /* copy blocks of the message to DMA preallocated buffer and send SMSG */
623     //Check whether there is any available DMA buffer
624     
625     do{
626         while((DMA_avail_tag & mask) && first_avail_bit<DMA_slots) {mask << 1; first_avail_bit++;}
627         if(first_avail_bit == DMA_slots) //No available DMA, buffer this message
628         {
629             MallocMediumMsgList(msg_tmp);
630             msg_tmp->destNode = destNode;
631             msg_tmp->msg_id   = lrts_send_msg_id;
632             msg_tmp->msg_subid   = sub_id;
633             msg_tmp->size   = remain_size;
634             msg_tmp->msg    = msg+pos;
635             msg_tmp->next   = NULL;
636             break;
637         }else
638         {
639             //copy this part of the message into this DMA buffer
640             //TODO optimize here, some data can go with this SMSG
641             blocksize = (remain_size>DMA_SIZE_PER_SLOT)?DMA_SIZE_PER_SLOT: remain_size;
642             memcpy(DMA_buffer_base_mdh_addr.addr[first_avail_bit], msg+pos, blocksize);
643             pos += blocksize;
644             remain_size -= blocksize;
645             SET_BITS(DMA_avail_tag, first_avail_bit);
646            
647             MallocMediumControlMsg(medium_msg_control_tmp);
648             medium_msg_control_tmp->msg_id = lrts_send_msg_id;
649             medium_msg_control_tmp->msg_subid = sub_id;
650             if(status == GNI_RC_SUCCESS)
651             {
652                 if(sub_id==0)
653                     status = GNI_SmsgSendWTag(ep_hndl_array[destNode], NULL, 0, medium_msg_tmp, sizeof(MEDIUM_MSG_CONTROL), 0, MEDIUM_HEAD_TAG);
654                 else
655                     status = GNI_SmsgSendWTag(ep_hndl_array[destNode], NULL, 0, medium_msg_tmp, sizeof(MEDIUM_MSG_CONTROL), 0, MEDIUM_DATA_TAG);
656             }
657             //buffer this smsg
658             if(status != GNI_RC_SUCCESS)
659             {
660                 delay_send_small_msg(medium_msg_tmp, sizeof(MEDIUM_MSG_CONTROL), destNode, MEDIUM_HEAD_TAG);
661             }
662             sub_id++;
663         }while(remain_size > 0 );
664
665         }
666     }
667 #endif
668 }
669
670 // Large message, send control to receiver, receiver register memory and do a GET 
671 inline
672 static void send_large_messages(int destNode, int size, char *msg)
673 {
674 #if     USE_LRTS_MEMPOOL
675     gni_return_t        status  =   GNI_RC_SUCCESS;
676     CONTROL_MSG         *control_msg_tmp;
677     uint32_t            vmdh_index  = -1;
678     /* construct a control message and send */
679     MallocControlMsg(control_msg_tmp);
680     control_msg_tmp->source_addr    = (uint64_t)msg;
681     control_msg_tmp->source         = myrank;
682     control_msg_tmp->length         =ALIGN4(size); //for GET 4 bytes aligned 
683     //memcpy( &(control_msg_tmp->source_mem_hndl), GetMemHndl(msg), sizeof(gni_mem_handle_t)) ;
684     control_msg_tmp->source_mem_hndl = GetMemHndl(msg);
685 #if PRINT_SYH
686     lrts_send_msg_id++;
687     CmiPrintf("Large LrtsSend PE:%d==>%d, size=%d, messageid:%d LMSG\n", myrank, destNode, size, lrts_send_msg_id);
688 #endif
689     status = send_smsg_message( destNode, 0, 0, control_msg_tmp, sizeof(CONTROL_MSG), LMSG_INIT_TAG, 0);  
690     if(status == GNI_RC_SUCCESS)
691     {
692         FreeControlMsg(control_msg_tmp);
693     }
694 // NOT use mempool, should slow 
695 #else
696     gni_return_t        status  =   GNI_RC_SUCCESS;
697     CONTROL_MSG         *control_msg_tmp;
698     uint32_t            vmdh_index  = -1;
699     /* construct a control message and send */
700     MallocControlMsg(control_msg_tmp);
701     control_msg_tmp->source_addr    = (uint64_t)msg;
702     control_msg_tmp->source         = myrank;
703     control_msg_tmp->length         =ALIGN4(size); //for GET 4 bytes aligned 
704     control_msg_tmp->source_mem_hndl.qword1 = 0;
705     control_msg_tmp->source_mem_hndl.qword2 = 0;
706 #if PRINT_SYH
707     lrts_send_msg_id++;
708     CmiPrintf("Large LrtsSend PE:%d==>%d, size=%d, messageid:%d LMSG\n", myrank, destNode, size, lrts_send_msg_id);
709 #endif
710     status = MEMORY_REGISTER(onesided_hnd, nic_hndl,msg, ALIGN4(size), &(control_msg_tmp->source_mem_hndl), &omdh);
711     if(status == GNI_RC_SUCCESS)
712     {
713         status = send_smsg_message( destNode, 0, 0, control_msg_tmp, sizeof(CONTROL_MSG), LMSG_INIT_TAG, 0);  
714         if(status == GNI_RC_SUCCESS)
715         {
716             FreeControlMsg(control_msg_tmp);
717         }
718     } else if (status == GNI_RC_INVALID_PARAM || status == GNI_RC_PERMISSION_ERROR)
719     {
720         CmiAbort("Memory registor for large msg\n");
721     }else 
722     {
723         delay_send_small_msg(control_msg_tmp, sizeof(CONTROL_MSG), destNode, LMSG_INIT_TAG);
724     }
725 #endif
726 }
727
728 inline void LrtsPrepareEnvelope(char *msg, int size)
729 {
730     CmiSetMsgSize(msg, size);
731 }
732
733 static CmiCommHandle LrtsSendFunc(int destNode, int size, char *msg, int mode)
734 {
735     gni_return_t        status  =   GNI_RC_SUCCESS;
736     LrtsPrepareEnvelope(msg, size);
737
738     if(size <= SMSG_MAX_MSG)
739     {
740 #if PRINT_SYH
741         lrts_send_msg_id++;
742         CmiPrintf("SMSG LrtsSend PE:%d==>%d, size=%d, messageid:%d\n", myrank, destNode, size, lrts_send_msg_id);
743 #endif
744         status = send_smsg_message( destNode, 0, 0, msg, size, SMALL_DATA_TAG, 0);  
745         if(status == GNI_RC_SUCCESS)
746         {
747             CmiFree(msg);
748         }
749     }
750     else
751     {
752         send_large_messages(destNode, size, msg);
753     }
754     return 0;
755 }
756
757 static void LrtsPreCommonInit(int everReturn){}
758
759 /* Idle-state related functions: called in non-smp mode */
760 void CmiNotifyIdleForGemini(void) {
761     LrtsAdvanceCommunication();
762 }
763
764 static void LrtsPostCommonInit(int everReturn)
765 {
766 #if CMK_SMP
767     CmiIdleState *s=CmiNotifyGetState();
768     CcdCallOnConditionKeep(CcdPROCESSOR_BEGIN_IDLE,(CcdVoidFn)CmiNotifyBeginIdle,(void *)s);
769     CcdCallOnConditionKeep(CcdPROCESSOR_STILL_IDLE,(CcdVoidFn)CmiNotifyStillIdle,(void *)s);
770 #else
771     CcdCallOnConditionKeep(CcdPROCESSOR_STILL_IDLE,(CcdVoidFn)CmiNotifyIdleForGemini,NULL);
772 #endif
773
774 }
775
776
777 void LrtsPostNonLocal(){}
778 /* pooling CQ to receive network message */
779 static void PumpNetworkRdmaMsgs()
780 {
781     gni_cq_entry_t      event_data;
782     gni_return_t        status;
783     while( (status = GNI_CqGetEvent(post_rx_cqh, &event_data)) == GNI_RC_SUCCESS);
784 }
785
786 static int SendRdmaMsg();
787 static void getLargeMsgRequest(void* header, uint64_t inst_id);
788 static void PumpNetworkSmsg()
789 {
790     uint64_t            inst_id;
791     PENDING_GETNEXT     *pending_next;
792     int                 ret;
793     gni_cq_entry_t      event_data;
794     gni_return_t        status;
795     void                *header;
796     uint8_t             msg_tag;
797     int                 msg_nbytes;
798     void                *msg_data;
799     gni_mem_handle_t    msg_mem_hndl;
800  
801
802     while ((status =GNI_CqGetEvent(smsg_rx_cqh, &event_data)) == GNI_RC_SUCCESS)
803     {
804         inst_id = GNI_CQ_GET_INST_ID(event_data);
805         // GetEvent returns success but GetNext return not_done. caused by Smsg out-of-order transfer
806 #if PRINT_SYH
807         CmiPrintf("[%d] PumpNetworkMsgs small msgs is received from PE: %d,  status=%s\n", myrank, inst_id,  gni_err_str[status]);
808 #endif
809         msg_tag = GNI_SMSG_ANY_TAG;
810         while( (status = GNI_SmsgGetNextWTag(ep_hndl_array[inst_id], &header, &msg_tag)) == GNI_RC_SUCCESS)
811         {
812 #if PRINT_SYH
813             lrts_received_msg++;
814             CmiPrintf("+++[%d] PumpNetwork msg is received, messageid:%d tag=%d\n", myrank, lrts_received_msg, msg_tag);
815 #endif
816             /* copy msg out and then put into queue (small message) */
817             switch (msg_tag) {
818             case SMALL_DATA_TAG:
819             {
820                 msg_nbytes = CmiGetMsgSize(header);
821                 msg_data    = CmiAlloc(msg_nbytes);
822                 memcpy(msg_data, (char*)header, msg_nbytes);
823                 handleOneRecvedMsg(msg_nbytes, msg_data);
824                 break;
825             }
826             case LMSG_INIT_TAG:
827             {
828 #if PRINT_SYH
829                 CmiPrintf("+++[%d] from %d PumpNetwork Rdma Request msg is received, messageid:%d tag=%d\n", myrank, inst_id, lrts_received_msg, msg_tag);
830 #endif
831                 getLargeMsgRequest(header, inst_id);
832                 break;
833             }
834             case ACK_TAG:
835             {
836                 /* Get is done, release message . Now put is not used yet*/
837 #if         !USE_LRTS_MEMPOOL
838                 MEMORY_DEREGISTER(onesided_hnd, nic_hndl, &(((CONTROL_MSG *)header)->source_mem_hndl), &omdh);
839 #endif
840                 CmiFree((void*)((CONTROL_MSG *) header)->source_addr);
841                 SendRdmaMsg();
842                 break;
843             }
844             case PUT_DONE_TAG: //persistent message
845             {
846                 handleOneRecvedMsg(((CONTROL_MSG *) header)->length,((void*) (CONTROL_MSG *) header)); 
847                 break;
848             }
849             default: {
850                 CmiPrintf("weird tag problem\n");
851                 CmiAbort("Unknown tag\n");
852             }
853             }
854             GNI_SmsgRelease(ep_hndl_array[inst_id]);
855             msg_tag = GNI_SMSG_ANY_TAG;
856         } //endwhile getNext
857     }   //end while GetEvent
858     if(status == GNI_RC_ERROR_RESOURCE)
859     {
860         GNI_RC_CHECK("Smsg_rx_cq full", status);
861     }
862 }
863
864 static void getLargeMsgRequest(void* header, uint64_t inst_id)
865 {
866 #if     USE_LRTS_MEMPOOL
867     CONTROL_MSG         *request_msg;
868     gni_return_t        status;
869     void                *msg_data;
870     gni_post_descriptor_t *pd;
871     RDMA_REQUEST        *rdma_request_msg;
872     gni_mem_handle_t    msg_mem_hndl;
873     int source;
874     // initial a get to transfer data from the sender side */
875     request_msg = (CONTROL_MSG *) header;
876     source = request_msg->source;
877     msg_data = CmiAlloc(request_msg->length);
878     _MEMCHECK(msg_data);
879     //memcpy(&msg_mem_hndl, GetMemHndl(msg_data), sizeof(gni_mem_handle_t));
880     msg_mem_hndl = GetMemHndl(msg_data);
881
882     MallocPostDesc(pd);
883     if(request_msg->length < LRTS_GNI_RDMA_THRESHOLD) 
884         pd->type            = GNI_POST_FMA_GET;
885     else
886         pd->type            = GNI_POST_RDMA_GET;
887 #if REMOTE_EVENT
888     pd->cq_mode         = GNI_CQMODE_GLOBAL_EVENT |  GNI_CQMODE_REMOTE_EVENT;
889 #else
890     pd->cq_mode         = GNI_CQMODE_GLOBAL_EVENT;
891 #endif
892     pd->dlvr_mode       = GNI_DLVMODE_PERFORMANCE;
893     pd->length          = ALIGN4(request_msg->length);
894     pd->local_addr      = (uint64_t) msg_data;
895     pd->local_mem_hndl  = msg_mem_hndl;
896     pd->remote_addr     = request_msg->source_addr;
897     pd->remote_mem_hndl = request_msg->source_mem_hndl;
898     pd->src_cq_hndl     = 0;//post_tx_cqh;     /* smsg_tx_cqh;  */
899     pd->rdma_mode       = 0;
900
901     if(pd->type == GNI_POST_RDMA_GET) 
902         status = GNI_PostRdma(ep_hndl_array[source], pd);
903     else
904         status = GNI_PostFma(ep_hndl_array[source],  pd);
905     if(status == GNI_RC_ERROR_RESOURCE|| status == GNI_RC_ERROR_NOMEM )
906     {
907         MallocRdmaRequest(rdma_request_msg);
908         rdma_request_msg->next = 0;
909         rdma_request_msg->destNode = inst_id;
910         rdma_request_msg->pd = pd;
911         if(pending_rdma_head == 0)
912         {
913             pending_rdma_head = rdma_request_msg;
914         }else
915         {
916             pending_rdma_tail->next = rdma_request_msg;
917         }
918         pending_rdma_tail = rdma_request_msg;
919     }else
920         GNI_RC_CHECK("AFter posting", status);
921
922 #else
923     CONTROL_MSG         *request_msg;
924     gni_return_t        status;
925     void                *msg_data;
926     gni_post_descriptor_t *pd;
927     RDMA_REQUEST        *rdma_request_msg;
928     gni_mem_handle_t    msg_mem_hndl;
929     int source;
930     // initial a get to transfer data from the sender side */
931     request_msg = (CONTROL_MSG *) header;
932     source = request_msg->source;
933     msg_data = CmiAlloc(request_msg->length);
934     _MEMCHECK(msg_data);
935
936     status = MEMORY_REGISTER(onesided_hnd, nic_hndl, msg_data, request_msg->length, &msg_mem_hndl, &omdh);
937
938     if (status == GNI_RC_INVALID_PARAM || status == GNI_RC_PERMISSION_ERROR) 
939     {
940         GNI_RC_CHECK("Mem Register before post", status);
941     }
942
943     MallocPostDesc(pd);
944     if(request_msg->length < LRTS_GNI_RDMA_THRESHOLD) 
945         pd->type            = GNI_POST_FMA_GET;
946     else
947         pd->type            = GNI_POST_RDMA_GET;
948 #if REMOTE_EVENT
949     pd->cq_mode         = GNI_CQMODE_GLOBAL_EVENT |  GNI_CQMODE_REMOTE_EVENT;
950 #else
951     pd->cq_mode         = GNI_CQMODE_GLOBAL_EVENT;
952 #endif
953     pd->dlvr_mode       = GNI_DLVMODE_PERFORMANCE;
954     pd->length          = ALIGN4(request_msg->length);
955     pd->local_addr      = (uint64_t) msg_data;
956     pd->remote_addr     = request_msg->source_addr;
957     pd->remote_mem_hndl = request_msg->source_mem_hndl;
958     pd->src_cq_hndl     = 0;//post_tx_cqh;     /* smsg_tx_cqh;  */
959     pd->rdma_mode       = 0;
960
961     //memory registration successful
962     if(status == GNI_RC_SUCCESS)
963     {
964         pd->local_mem_hndl  = msg_mem_hndl;
965         if(pd->type == GNI_POST_RDMA_GET) 
966             status = GNI_PostRdma(ep_hndl_array[source], pd);
967         else
968             status = GNI_PostFma(ep_hndl_array[source],  pd);
969     }else
970     {
971         pd->local_mem_hndl.qword1  = 0; 
972         pd->local_mem_hndl.qword1  = 0; 
973     }
974     if(status == GNI_RC_ERROR_RESOURCE|| status == GNI_RC_ERROR_NOMEM )
975     {
976         MallocRdmaRequest(rdma_request_msg);
977         rdma_request_msg->next = 0;
978         rdma_request_msg->destNode = inst_id;
979         rdma_request_msg->pd = pd;
980         if(pending_rdma_head == 0)
981         {
982             pending_rdma_head = rdma_request_msg;
983         }else
984         {
985             pending_rdma_tail->next = rdma_request_msg;
986         }
987         pending_rdma_tail = rdma_request_msg;
988     }else
989         GNI_RC_CHECK("AFter posting", status);
990 #endif
991 }
992
993 /* Check whether message send or get is confirmed by remote */
994 static void PumpLocalSmsgTransactions()
995 {
996     gni_return_t            status;
997     gni_cq_entry_t          ev;
998     while ((status = GNI_CqGetEvent(smsg_tx_cqh, &ev)) == GNI_RC_SUCCESS)
999     {
1000 #if PRINT_SYH
1001         lrts_local_done_msg++;
1002         //CmiPrintf("*[%d]  PumpLocalSmsgTransactions GNI_CQ_GET_TYPE %d. Localdone=%d\n", myrank, GNI_CQ_GET_TYPE(ev), lrts_local_done_msg);
1003 #endif
1004         if(GNI_CQ_OVERRUN(ev))
1005         {
1006             CmiPrintf("Overrun detected in local CQ");
1007             CmiAbort("Overrun in TX");
1008         }
1009     }
1010     if(status == GNI_RC_ERROR_RESOURCE)
1011     {
1012         GNI_RC_CHECK("Smsg_tx_cq full", status);
1013     }
1014 }
1015
1016 static void PumpLocalRdmaTransactions()
1017 {
1018     gni_cq_entry_t          ev;
1019     gni_return_t            status;
1020     uint64_t                type, inst_id;
1021     gni_post_descriptor_t   *tmp_pd;
1022     MSG_LIST                *ptr;
1023     CONTROL_MSG             *ack_msg_tmp;
1024     uint8_t             msg_tag;
1025
1026    // while ( (status = GNI_CqGetEvent(post_tx_cqh, &ev)) == GNI_RC_SUCCESS) 
1027     while ( (status = GNI_CqGetEvent(smsg_tx_cqh, &ev)) == GNI_RC_SUCCESS) 
1028     {
1029         type        = GNI_CQ_GET_TYPE(ev);
1030 #if PRINT_SYH
1031         //lrts_local_done_msg++;
1032         CmiPrintf("**[%d] SMSGPumpLocalTransactions (type=%d)\n", myrank, type);
1033 #endif
1034         if (type == GNI_CQ_EVENT_TYPE_POST)
1035         {
1036             inst_id     = GNI_CQ_GET_INST_ID(ev);
1037 #if PRINT_SYH
1038             CmiPrintf("**[%d] SMSGPumpLocalTransactions localdone=%d\n", myrank,  lrts_local_done_msg);
1039 #endif
1040             //status = GNI_GetCompleted(post_tx_cqh, ev, &tmp_pd);
1041             status = GNI_GetCompleted(smsg_tx_cqh, ev, &tmp_pd);
1042             ////Message is sent, free message , put is not used now
1043             if(tmp_pd->type == GNI_POST_RDMA_PUT || tmp_pd->type == GNI_POST_FMA_PUT)
1044             {   //persistent message 
1045                 CmiFree((void *)tmp_pd->local_addr);
1046                 msg_tag = PUT_DONE_TAG;  
1047             }else if(tmp_pd->type == GNI_POST_RDMA_GET || tmp_pd->type == GNI_POST_FMA_GET)
1048             {
1049                 msg_tag = ACK_TAG;  
1050             }
1051             MallocControlMsg(ack_msg_tmp);
1052             ack_msg_tmp->source             = myrank;
1053             ack_msg_tmp->source_addr        = tmp_pd->remote_addr;
1054             ack_msg_tmp->length             = tmp_pd->length; 
1055             ack_msg_tmp->source_mem_hndl    = tmp_pd->remote_mem_hndl;
1056 #if PRINT_SYH
1057             lrts_send_msg_id++;
1058             CmiPrintf("ACK LrtsSend PE:%d==>%d, size=%d, messageid:%d ACK\n", myrank, inst_id, sizeof(CONTROL_MSG), lrts_send_msg_id);
1059 #endif
1060             status = send_smsg_message(inst_id, 0, 0, ack_msg_tmp, sizeof(CONTROL_MSG), msg_tag, 0);  
1061             if(status == GNI_RC_SUCCESS)
1062             {
1063                 FreeControlMsg(ack_msg_tmp);
1064             }
1065 #if     !USE_LRTS_MEMPOOL
1066             MEMORY_DEREGISTER(onesided_hnd, nic_hndl, &tmp_pd->local_mem_hndl, &omdh);
1067 #endif
1068             CmiAssert(SIZEFIELD((void*)(tmp_pd->local_addr)) <= tmp_pd->length);
1069             handleOneRecvedMsg(tmp_pd->length, (void*)tmp_pd->local_addr); 
1070             SendRdmaMsg(); 
1071             FreePostDesc(tmp_pd);
1072         }
1073     } //end while
1074 }
1075
1076 static int SendRdmaMsg()
1077 {
1078     gni_return_t            status = GNI_RC_SUCCESS;
1079     gni_mem_handle_t        msg_mem_hndl;
1080
1081     RDMA_REQUEST *ptr = pending_rdma_head;
1082     RDMA_REQUEST *prev = NULL;
1083
1084     while (ptr != NULL)
1085     {
1086         gni_post_descriptor_t *pd = ptr->pd;
1087         status = GNI_RC_SUCCESS;
1088         // register memory first
1089         if( pd->local_mem_hndl.qword1 == 0 && pd->local_mem_hndl.qword2 == 0)
1090         {
1091             status = MEMORY_REGISTER(onesided_hnd, nic_hndl, pd->local_addr, pd->length, &(pd->local_mem_hndl), &omdh);
1092         }
1093         if(status == GNI_RC_SUCCESS)
1094         {
1095             if(pd->type == GNI_POST_RDMA_GET || pd->type == GNI_POST_RDMA_PUT) 
1096                 status = GNI_PostRdma(ep_hndl_array[ptr->destNode], pd);
1097             else
1098                 status = GNI_PostFma(ep_hndl_array[ptr->destNode],  pd);
1099             if(status == GNI_RC_SUCCESS)
1100             {
1101                 RDMA_REQUEST *tmp = ptr;
1102                 if (prev)
1103                   prev->next = ptr->next;
1104                 else
1105                   pending_rdma_head = ptr->next;
1106                 ptr = ptr->next;
1107                 FreeRdmaRequest(tmp);
1108                 continue;
1109             }
1110         }
1111         prev = ptr;
1112         ptr = ptr->next;
1113     } //end while
1114     return 0;
1115 }
1116
1117 // return 1 if all messages are sent
1118 static int SendBufferMsg()
1119 {
1120     MSG_LIST            *ptr, *previous_head, *current_head;
1121     CONTROL_MSG         *control_msg_tmp;
1122     gni_return_t        status;
1123     int done = 1;
1124     register    int     i;
1125     int                 index_previous = -1;
1126     int                 index = smsg_head_index;
1127     //if( smsg_msglist_head == 0 && buffered_smsg_counter!= 0 ) {CmiPrintf("WRONGWRONG on rank%d, buffermsg=%d, (msgid-succ:%d)\n", myrank, buffered_smsg_counter, (lrts_send_msg_id-lrts_smsg_success)); CmiAbort("sendbuf");}
1128     /* can add flow control here to control the number of messages sent before handle message */
1129     while(index != -1)
1130     {
1131         ptr = smsg_msglist_index[index].head;
1132        
1133         while(ptr!=0)
1134         {
1135             CmiAssert(ptr!=NULL);
1136             if(ptr->tag == SMALL_DATA_TAG)
1137             {
1138                 status = send_smsg_message( ptr->destNode, 0, 0, ptr->msg, ptr->size, SMALL_DATA_TAG, 1);  
1139                 if(status == GNI_RC_SUCCESS)
1140                 {
1141                     CmiFree(ptr->msg);
1142                 }
1143             }
1144             else if(ptr->tag == LMSG_INIT_TAG)
1145             {
1146                 control_msg_tmp = (CONTROL_MSG*)ptr->msg;
1147 #if PRINT_SYH
1148                 CmiPrintf("[%d==>%d] LMSG buffer send call(%d)%s\n", myrank, ptr->destNode, lrts_smsg_success, gni_err_str[status] );
1149 #endif
1150                 if(control_msg_tmp->source_mem_hndl.qword1 == 0 && control_msg_tmp->source_mem_hndl.qword2 == 0)
1151                 {
1152                     MEMORY_REGISTER(onesided_hnd, nic_hndl, control_msg_tmp->source_addr, control_msg_tmp->length, &(control_msg_tmp->source_mem_hndl), &omdh);
1153                     if(status != GNI_RC_SUCCESS) {
1154                         done = 0;
1155                         break;
1156                     }
1157                 }
1158                 
1159                 status = send_smsg_message( ptr->destNode, 0, 0, ptr->msg, sizeof(CONTROL_MSG), LMSG_INIT_TAG, 1);  
1160                 if(status == GNI_RC_SUCCESS)
1161                 {
1162                     FreeControlMsg((CONTROL_MSG*)(ptr->msg));
1163                 }
1164             }else if (ptr->tag == ACK_TAG)
1165             {
1166                 status = send_smsg_message( ptr->destNode, 0, 0, ptr->msg, sizeof(CONTROL_MSG), ACK_TAG, 1);  
1167                 if(status == GNI_RC_SUCCESS)
1168                 {
1169                     FreeControlMsg((CONTROL_MSG*)ptr->msg);
1170                 }
1171             }else
1172             {
1173                 CmiPrintf("Weird tag\n");
1174                 CmiAbort("should not happen\n");
1175             }
1176         } //end while 
1177         if(status == GNI_RC_SUCCESS)
1178         {
1179             smsg_msglist_index[index].head = smsg_msglist_index[index].head->next;
1180             FreeMsgList(ptr);
1181             ptr= smsg_msglist_index[index].head;
1182 #if PRINT_SYH
1183             buffered_smsg_counter--;
1184             if(lrts_smsg_success == lrts_send_msg_id)
1185                 CmiPrintf("GOOD send buff [%d==>%d] send buffer sent done%d (msgs=%d)\n", myrank, ptr->destNode, lrts_smsg_success, lrts_send_msg_id);
1186             else
1187                 CmiPrintf("BAD send buff [%d==>%d] sent done%d (msgs=%d)\n", myrank, ptr->destNode, lrts_smsg_success, lrts_send_msg_id);
1188 #endif
1189         }else {
1190             done = 0;
1191             break;
1192         }
1193         if(ptr == 0)
1194         {
1195             if(index_previous != -1)
1196                 smsg_msglist_index[index_previous].next = smsg_msglist_index[index].next;
1197             else
1198                 smsg_head_index = smsg_msglist_index[index].next;
1199         }else
1200         {
1201             index_previous = index;
1202         }
1203         index = smsg_msglist_index[index].next;
1204     }   // end pooling for all cores
1205 #if PRINT_SYH
1206     if(lrts_send_msg_id-lrts_smsg_success !=0)
1207         CmiPrintf("WRONG [%d buffered msg is empty(%p) (actually=%d)(buffercounter=%d)\n", myrank, smsg_msglist_head[i], (lrts_send_msg_id-lrts_smsg_success, buffered_smsg_counter));
1208 #endif
1209     return done;
1210 }
1211
1212 static void LrtsAdvanceCommunication()
1213 {
1214     /*  Receive Msg first */
1215     //CmiPrintf("Calling Lrts Pump Msg PE:%d\n", CmiMyPe());
1216     PumpNetworkSmsg();
1217     //CmiPrintf("Calling Lrts Pump RdmaMsg PE:%d\n", CmiMyPe());
1218     //PumpNetworkRdmaMsgs();
1219     /* Release Sent Msg */
1220     //CmiPrintf("Calling Lrts Rlease Msg PE:%d\n", CmiMyPe());
1221     //PumpLocalSmsgTransactions();
1222     //CmiPrintf("Calling Lrts Rlease RdmaMsg PE:%d\n", CmiMyPe());
1223     PumpLocalRdmaTransactions();
1224     //CmiPrintf("Calling Lrts Send Buffmsg PE:%d\n", CmiMyPe());
1225     /* Send buffered Message */
1226     SendBufferMsg();
1227     SendRdmaMsg();
1228 }
1229
1230 static void _init_dynamic_smsg()
1231 {
1232     mdh_addr_t current_addr;
1233     gni_return_t status;
1234     smsg_connected_flag = (int*)malloc(sizeof(int)*mysize);
1235     memset(smsg_connected_flag, 0, mysize*sizeof(int));
1236
1237     current_addr.addr = (uint64_t)malloc(mysize * SMSG_CONN_SIZE);
1238     status = MEMORY_REGISTER(onesided_hnd, nic_hndl, smsg_connection_addr, mysize * SMSG_CONN_SIZE,  &(current_addr.mdh), &omdh);
1239     
1240     smsg_connection_vec = (mdh_addr_t*) malloc(mysize*sizeof(mdh_addr_t)); 
1241     allgather(&current_addr, smsg_connection_vec, sizeof(mdh_addr_t));
1242 }
1243
1244 static void _init_static_smsg()
1245 {
1246     gni_smsg_attr_t      *smsg_attr;
1247     gni_smsg_attr_t      remote_smsg_attr;
1248     gni_smsg_attr_t      *smsg_attr_vec;
1249     unsigned int         smsg_memlen;
1250     gni_mem_handle_t     my_smsg_mdh_mailbox;
1251     int      i;
1252     gni_return_t status;
1253     uint32_t              vmdh_index = -1;
1254     mdh_addr_t            base_infor;
1255     mdh_addr_t            *base_addr_vec;
1256     if(mysize <=4096)
1257     {
1258         SMSG_MAX_MSG = 1024;
1259         //log2_SMSG_MAX_MSG = 10;
1260     }else if (mysize > 4096 && mysize <= 16384)
1261     {
1262         SMSG_MAX_MSG = 512;
1263         //log2_SMSG_MAX_MSG = 9;
1264
1265     }else {
1266         SMSG_MAX_MSG = 256;
1267         //log2_SMSG_MAX_MSG = 8;
1268     }
1269     
1270     smsg_attr = malloc(mysize * sizeof(gni_smsg_attr_t));
1271     
1272     smsg_attr[0].msg_type = GNI_SMSG_TYPE_MBOX_AUTO_RETRANSMIT;
1273     smsg_attr[0].mbox_maxcredit = SMSG_MAX_CREDIT;
1274     smsg_attr[0].msg_maxsize = SMSG_MAX_MSG;
1275     status = GNI_SmsgBufferSizeNeeded(&smsg_attr[0], &smsg_memlen);
1276     GNI_RC_CHECK("GNI_GNI_MemRegister mem buffer", status);
1277     smsg_mailbox_base = memalign(64, smsg_memlen*(mysize));
1278     _MEMCHECK(smsg_mailbox_base);
1279     bzero(smsg_mailbox_base, smsg_memlen*(mysize));
1280     if (myrank == 0) CmiPrintf("Charm++> allocates %.2fMB for SMSG. \n", smsg_memlen*mysize/1e6);
1281     
1282     status = GNI_MemRegister(nic_hndl, (uint64_t)smsg_mailbox_base,
1283             smsg_memlen*(mysize), smsg_rx_cqh,
1284             GNI_MEM_READWRITE,   
1285             vmdh_index,
1286             &my_smsg_mdh_mailbox);
1287
1288     GNI_RC_CHECK("GNI_GNI_MemRegister mem buffer", status);
1289
1290 #if 1
1291     base_infor.addr =  (uint64_t)smsg_mailbox_base;
1292     base_infor.mdh =  my_smsg_mdh_mailbox;
1293     base_addr_vec = malloc(mysize * sizeof(mdh_addr_t));
1294
1295     allgather(&base_infor, base_addr_vec,  sizeof(mdh_addr_t));
1296  
1297     for(i=0; i<mysize; i++)
1298     {
1299         if(i==myrank)
1300             continue;
1301         smsg_attr[i].msg_type = GNI_SMSG_TYPE_MBOX_AUTO_RETRANSMIT;
1302         smsg_attr[i].mbox_maxcredit = SMSG_MAX_CREDIT;
1303         smsg_attr[i].msg_maxsize = SMSG_MAX_MSG;
1304         smsg_attr[i].mbox_offset = i*smsg_memlen;
1305         smsg_attr[i].buff_size = smsg_memlen;
1306         smsg_attr[i].msg_buffer = smsg_mailbox_base ;
1307         smsg_attr[i].mem_hndl = my_smsg_mdh_mailbox;
1308     }
1309
1310     for(i=0; i<mysize; i++)
1311     {
1312         if (myrank == i) continue;
1313
1314         remote_smsg_attr.msg_type = GNI_SMSG_TYPE_MBOX_AUTO_RETRANSMIT;
1315         remote_smsg_attr.mbox_maxcredit = SMSG_MAX_CREDIT;
1316         remote_smsg_attr.msg_maxsize = SMSG_MAX_MSG;
1317         remote_smsg_attr.mbox_offset = myrank*smsg_memlen;
1318         remote_smsg_attr.buff_size = smsg_memlen;
1319         remote_smsg_attr.msg_buffer = (void*)base_addr_vec[i].addr;
1320         remote_smsg_attr.mem_hndl = base_addr_vec[i].mdh;
1321
1322         /* initialize the smsg channel */
1323         status = GNI_SmsgInit(ep_hndl_array[i], &smsg_attr[i], &remote_smsg_attr);
1324         GNI_RC_CHECK("SMSG Init", status);
1325     } //end initialization
1326
1327     free(base_addr_vec);
1328 #else
1329     for(i=0; i<mysize; i++)
1330     {
1331         if(i==myrank)
1332             continue;
1333         smsg_attr[i].msg_type = GNI_SMSG_TYPE_MBOX_AUTO_RETRANSMIT;
1334         smsg_attr[i].mbox_maxcredit = SMSG_MAX_CREDIT;
1335         smsg_attr[i].msg_maxsize = SMSG_MAX_MSG;
1336         if(i<myrank)
1337             smsg_attr[i].mbox_offset = i*smsg_memlen;
1338         else
1339             smsg_attr[i].mbox_offset = (i-1)*smsg_memlen;
1340
1341         smsg_attr[i].msg_buffer = smsg_mailbox_base;
1342         smsg_attr[i].buff_size = smsg_memlen;
1343         smsg_attr[i].mem_hndl = my_smsg_mdh_mailbox;
1344 #if 0 
1345         if(i<2)
1346         {
1347             CmiPrintf("[assign %d==%d]  offset=%d buf=%p,credit=%d size=%d, bufsize=%d\n", myrank, i, (smsg_attr[i]).mbox_offset, (smsg_attr[i]).msg_buffer, (smsg_attr[i]).mbox_maxcredit, (smsg_attr[i]).msg_maxsize, (smsg_attr[i]).buff_size ); 
1348 #endif 
1349         }
1350     }
1351     smsg_attr_vec = (gni_smsg_attr_t*)malloc(mysize * mysize * sizeof(gni_smsg_attr_t));
1352     _MEMCHECK(smsg_attr_vec);
1353  
1354     if(myrank==0)
1355         CmiPrintf("total size=%d\n", mysize*mysize*sizeof(gni_smsg_attr_t));
1356     allgather(smsg_attr, smsg_attr_vec,  mysize*sizeof(gni_smsg_attr_t));
1357     
1358     CmiBarrier();
1359     for(i=0; i<mysize; i++)
1360     {
1361         if (myrank == i) continue;
1362         /* initialize the smsg channel */
1363         status = GNI_SmsgInit(ep_hndl_array[i], &smsg_attr[i], &smsg_attr_vec[i*mysize+myrank]);
1364 #if 0 
1365         if(i<2)
1366         {
1367             CmiPrintf("[%d==%d] SmsgInit rc=%s, offset=%d buf=%p,credit=%d size=%d, bufsize=%d\n", myrank, i, gni_err_str[status], (smsg_attr_vec[i*mysize+myrank]).mbox_offset, (smsg_attr_vec[i*mysize+myrank]).msg_buffer, (smsg_attr_vec[i*mysize+myrank]).mbox_maxcredit, (smsg_attr_vec[i*mysize+myrank]).msg_maxsize, (smsg_attr_vec[i*mysize+myrank]).buff_size ); 
1368         }
1369 #endif
1370         GNI_RC_CHECK("SMSG Init", status);
1371     } //end initialization
1372     CmiBarrier();
1373     free(smsg_attr_vec);
1374 #endif
1375     free(smsg_attr);
1376     status = GNI_SmsgSetMaxRetrans(nic_hndl, 4096);
1377     GNI_RC_CHECK("SmsgSetMaxRetrans Init", status);
1378     smsg_msglist_index = (MSG_LIST_INDEX*)malloc(mysize*sizeof(MSG_LIST_INDEX));
1379     for(i =0; i<mysize; i++)
1380     {
1381         smsg_msglist_index[i].next = -1;
1382         smsg_msglist_index[i].head = 0;
1383         smsg_msglist_index[i].tail = 0;
1384     }
1385     smsg_head_index = -1;
1386
1387
1388 static void _init_static_msgq()
1389 {
1390     gni_return_t status;
1391     /* MSGQ is to send and receive short messages for large jobs (exceeding 200,000 ranks). The          performance scales by the node count rather than rank count */
1392     msgq_attrs.max_msg_sz = MSGQ_MAXSIZE;
1393     msgq_attrs.smsg_q_sz = 1;
1394     msgq_attrs.rcv_pool_sz = 1;
1395     msgq_attrs.num_msgq_eps = 2;
1396     msgq_attrs.nloc_insts = 8;
1397     msgq_attrs.modes = 0;
1398     msgq_attrs.rcv_cq_sz = REMOTE_QUEUE_ENTRIES ;
1399
1400     status = GNI_MsgqInit(nic_hndl, NULL, NULL, NULL, &msgq_attrs, &msgq_handle);
1401     GNI_RC_CHECK("MSGQ Init", status);
1402
1403
1404 }
1405
1406 static void _init_DMA_buffer()
1407 {
1408     gni_return_t            status = GNI_RC_SUCCESS;
1409     /*AUTO tuning */
1410     /* suppose max_smsg is 1024, DMA buffer is split into 2048, 4096, 8192, ... */
1411     /*  This method might be better for SMP, but it is bad for Nonsmp since msgs are sharing same slots */
1412     /*
1413      * DMA_slots = 19-log2_SMSG_MAX_MSG;
1414     DMA_incoming_avail_tag = malloc(DMA_slots);
1415     DMA_buffer_size = 2*(DMA_max_single_msg - SMSG_MAX_MSG); 
1416     DMA_incoming_base_addr =  memalign(ALIGNBUF, DMA_buffer_size);
1417     DMA_outgoing_base_addr =  memalign(ALIGNBUF, DMA_buffer_size);
1418     
1419     status = GNI_MemRegister(nic_hndl, (uint64_t)DMA_incoming_base_addr,
1420             DMA_buffer_size, smsg_rx_cqh,
1421             GNI_MEM_READWRITE ,   
1422             vmdh_index,
1423             &);
1424             */
1425     // one is reserved to avoid deadlock
1426     DMA_slots           = 17; // each one is 8K  16*8K + 1 slot reserved to avoid deadlock
1427     DMA_buffer_size     = DMA_max_single_msg + 8192;
1428     DMA_buffer_base_mdh_addr.addr = (uint64_t)memalign(ALIGNBUF, DMA_buffer_size);
1429     status = GNI_MemRegister(nic_hndl, DMA_buffer_base_mdh_addr.addr,
1430         DMA_buffer_size, smsg_rx_cqh,
1431         GNI_MEM_READWRITE ,   
1432         -1,
1433         &(DMA_buffer_base_mdh_addr.mdh));
1434     GNI_RC_CHECK("GNI_MemRegister", status);
1435     DMA_buffer_base_mdh_addr_vec = (mdh_addr_t*) malloc(sizeof(mdh_addr_t) * mysize);
1436
1437     allgather(&DMA_buffer_base_mdh_addr, DMA_buffer_base_mdh_addr_vec, sizeof(mdh_addr_t) );
1438 }
1439
1440 static void LrtsInit(int *argc, char ***argv, int *numNodes, int *myNodeID)
1441 {
1442     register int            i;
1443     int                     rc;
1444     int                     device_id = 0;
1445     unsigned int            remote_addr;
1446     gni_cdm_handle_t        cdm_hndl;
1447     gni_return_t            status = GNI_RC_SUCCESS;
1448     uint32_t                vmdh_index = -1;
1449     uint8_t                 ptag;
1450     unsigned int            local_addr, *MPID_UGNI_AllAddr;
1451     int                     first_spawned;
1452     int                     physicalID;
1453     //void (*local_event_handler)(gni_cq_entry_t *, void *)       = &LocalEventHandle;
1454     //void (*remote_smsg_event_handler)(gni_cq_entry_t *, void *) = &RemoteSmsgEventHandle;
1455     //void (*remote_bte_event_handler)(gni_cq_entry_t *, void *)  = &RemoteBteEventHandle;
1456    
1457     //Mempool_MaxSize = CmiGetArgFlag(*argv, "+useMemorypoolSize");
1458     useDynamicSMSG = CmiGetArgFlag(*argv, "+useDynamicSmsg");
1459     if (myrank==0) 
1460     {
1461         if(useDynamicSMSG) 
1462             CmiPrintf("Charm++> use Dynamic SMSG\n"); 
1463         else 
1464             CmiPrintf("Charm++> use Static SMSG\n");
1465     };
1466     //useStaticMSGQ = CmiGetArgFlag(*argv, "+useStaticMsgQ");
1467     
1468     status = PMI_Init(&first_spawned);
1469     GNI_RC_CHECK("PMI_Init", status);
1470
1471     status = PMI_Get_size(&mysize);
1472     GNI_RC_CHECK("PMI_Getsize", status);
1473
1474     status = PMI_Get_rank(&myrank);
1475     GNI_RC_CHECK("PMI_getrank", status);
1476
1477     //physicalID = CmiPhysicalNodeID(myrank);
1478     
1479     //printf("Pysical Node ID:%d for PE:%d\n", physicalID, myrank);
1480
1481     *myNodeID = myrank;
1482     *numNodes = mysize;
1483   
1484     if(myrank == 0)
1485     {
1486         printf("Charm++> Running on Gemini (GNI) using %d  cores\n", mysize);
1487     }
1488 #ifdef USE_ONESIDED
1489     onesided_init(NULL, &onesided_hnd);
1490
1491     // this is a GNI test, so use the libonesided bypass functionality
1492     onesided_gni_bypass_get_nih(onesided_hnd, &nic_hndl);
1493     local_addr = gniGetNicAddress();
1494 #else
1495     ptag = get_ptag();
1496     cookie = get_cookie();
1497 #if 0
1498     modes = GNI_CDM_MODE_CQ_NIC_LOCAL_PLACEMENT;
1499 #endif
1500     //Create and attach to the communication  domain */
1501     status = GNI_CdmCreate(myrank, ptag, cookie, modes, &cdm_hndl);
1502     GNI_RC_CHECK("GNI_CdmCreate", status);
1503     //* device id The device id is the minor number for the device
1504     //that is assigned to the device by the system when the device is created.
1505     //To determine the device number, look in the /dev directory, which contains a list of devices. For a NIC, the device is listed as kgniX
1506     //where X is the device number 0 default 
1507     status = GNI_CdmAttach(cdm_hndl, device_id, &local_addr, &nic_hndl);
1508     GNI_RC_CHECK("GNI_CdmAttach", status);
1509     local_addr = get_gni_nic_address(0);
1510 #endif
1511     MPID_UGNI_AllAddr = (unsigned int *)malloc(sizeof(unsigned int) * mysize);
1512     _MEMCHECK(MPID_UGNI_AllAddr);
1513     allgather(&local_addr, MPID_UGNI_AllAddr, sizeof(unsigned int));
1514     /* create the local completion queue */
1515     /* the third parameter : The number of events the NIC allows before generating an interrupt. Setting this parameter to zero results in interrupt delivery with every event. When using this parameter, the mode parameter must be set to GNI_CQ_BLOCKING*/
1516     status = GNI_CqCreate(nic_hndl, LOCAL_QUEUE_ENTRIES, 0, GNI_CQ_NOBLOCK, NULL, NULL, &smsg_tx_cqh);
1517     GNI_RC_CHECK("GNI_CqCreate (tx)", status);
1518     
1519     //status = GNI_CqCreate(nic_hndl, LOCAL_QUEUE_ENTRIES, 0, GNI_CQ_NOBLOCK, NULL, NULL, &post_tx_cqh);
1520     //GNI_RC_CHECK("GNI_CqCreate post (tx)", status);
1521     /* create the destination completion queue for receiving micro-messages, make this queue considerably larger than the number of transfers */
1522
1523     status = GNI_CqCreate(nic_hndl, REMOTE_QUEUE_ENTRIES, 0, GNI_CQ_NOBLOCK, NULL, NULL, &smsg_rx_cqh);
1524     GNI_RC_CHECK("Create CQ (rx)", status);
1525     
1526     //status = GNI_CqCreate(nic_hndl, REMOTE_QUEUE_ENTRIES, 0, GNI_CQ_NOBLOCK, NULL, NULL, &post_rx_cqh);
1527     //GNI_RC_CHECK("Create Post CQ (rx)", status);
1528     
1529     //status = GNI_CqCreate(nic_hndl, REMOTE_QUEUE_ENTRIES, 0, GNI_CQ_NOBLOCK, NULL, NULL, &rdma_cqh);
1530     //GNI_RC_CHECK("Create BTE CQ", status);
1531
1532     /* create the endpoints. they need to be bound to allow later CQWrites to them */
1533     ep_hndl_array = (gni_ep_handle_t*)malloc(mysize * sizeof(gni_ep_handle_t));
1534     _MEMCHECK(ep_hndl_array);
1535
1536     for (i=0; i<mysize; i++) {
1537         if(i == myrank) continue;
1538         status = GNI_EpCreate(nic_hndl, smsg_tx_cqh, &ep_hndl_array[i]);
1539         GNI_RC_CHECK("GNI_EpCreate ", status);   
1540         remote_addr = MPID_UGNI_AllAddr[i];
1541         status = GNI_EpBind(ep_hndl_array[i], remote_addr, i);
1542         GNI_RC_CHECK("GNI_EpBind ", status);   
1543     }
1544     /* Depending on the number of cores in the job, decide different method */
1545     /* SMSG is fastest but not scale; Msgq is scalable, FMA is own implementation for small message */
1546     if(mysize > 1)
1547     {
1548         if(useDynamicSMSG == 0)
1549         {
1550             _init_static_smsg();
1551         }else
1552         {
1553             _init_dynamic_smsg();
1554         }
1555     }
1556 #if     USE_LRTS_MEMPOOL
1557     CmiGetArgLong(*argv, "+useMemorypoolSize", &_mempool_size);
1558     if (myrank==0) CmiPrintf("Charm++> use memorypool size: %1.fMB\n", _mempool_size/1024.0/1024);
1559     init_mempool(_mempool_size);
1560     //init_mempool(Mempool_MaxSize);
1561 #endif
1562
1563     /* init DMA buffer for medium message */
1564
1565     //_init_DMA_buffer();
1566     
1567     free(MPID_UGNI_AllAddr);
1568 }
1569
1570
1571 void* LrtsAlloc(int n_bytes, int header)
1572 {
1573     void *ptr;
1574 #if 0
1575     CmiPrintf("\n[PE:%d]Alloc Lrts for bytes=%d, head=%d\n", CmiMyPe(), n_bytes, header);
1576 #endif
1577     if(n_bytes <= SMSG_MAX_MSG)
1578     {
1579         int totalsize = n_bytes+header;
1580         ptr = malloc(totalsize);
1581     }else 
1582     {
1583
1584         CmiAssert(header <= ALIGNBUF);
1585 #if     USE_LRTS_MEMPOOL
1586         n_bytes = ALIGN64(n_bytes);
1587         char *res = syh_mempool_malloc(ALIGNBUF+n_bytes);
1588 #else
1589         n_bytes = ALIGN4(n_bytes);           /* make sure size if 4 aligned */
1590         char *res = memalign(ALIGNBUF, n_bytes+ALIGNBUF);
1591 #endif
1592         ptr = res + ALIGNBUF - header;
1593     }
1594 #if 0 
1595     CmiPrintf("Done Alloc Lrts for bytes=%d, head=%d\n", n_bytes, header);
1596 #endif
1597     return ptr;
1598 }
1599
1600 void  LrtsFree(void *msg)
1601 {
1602     int size = SIZEFIELD((char*)msg+sizeof(CmiChunkHeader));
1603     if (size <= SMSG_MAX_MSG)
1604       free(msg);
1605     else
1606     {
1607 #if 0
1608         CmiPrintf("[PE:%d] Free lrts for bytes=%d, ptr=%p\n", CmiMyPe(), size, (char*)msg + sizeof(CmiChunkHeader) - ALIGNBUF);
1609 #endif
1610 #if     USE_LRTS_MEMPOOL
1611         syh_mempool_free((char*)msg + sizeof(CmiChunkHeader) - ALIGNBUF);
1612 #else
1613         free((char*)msg + sizeof(CmiChunkHeader) - ALIGNBUF);
1614 #endif
1615     }
1616 #if 0 
1617     CmiPrintf("Done Free lrts for bytes=%d\n", size);
1618 #endif
1619 }
1620
1621 static void LrtsExit()
1622 {
1623     /* free memory ? */
1624 #if     USE_LRTS_MEMPOOL
1625     kill_allmempool();
1626 #endif
1627     PMI_Finalize();
1628     exit(0);
1629 }
1630
1631 static void LrtsDrainResources()
1632 {
1633     while (!SendBufferMsg()) {
1634         PumpNetworkSmsg();
1635         PumpNetworkRdmaMsgs();
1636         PumpLocalSmsgTransactions();
1637         PumpLocalRdmaTransactions();
1638     }
1639     PMI_Barrier();
1640 }
1641
1642 void CmiAbort(const char *message) {
1643
1644     CmiPrintf("CmiAbort is calling on PE:%d\n", myrank);
1645     PMI_Abort(-1, message);
1646 }
1647
1648 /**************************  TIMER FUNCTIONS **************************/
1649 #if CMK_TIMER_USE_SPECIAL
1650 /* MPI calls are not threadsafe, even the timer on some machines */
1651 static CmiNodeLock  timerLock = 0;
1652 static int _absoluteTime = 0;
1653 static int _is_global = 0;
1654 static struct timespec start_ns;
1655
1656 inline int CmiTimerIsSynchronized() {
1657     return 1;
1658 }
1659
1660 inline int CmiTimerAbsolute() {
1661     return _absoluteTime;
1662 }
1663
1664 double CmiStartTimer() {
1665     return 0.0;
1666 }
1667
1668 double CmiInitTime() {
1669     return (double)(start_ts.tv_sec)+(double)start_ts.tv_nsec/1000000000.0;
1670 }
1671
1672 void CmiTimerInit(char **argv) {
1673     _absoluteTime = CmiGetArgFlagDesc(argv,"+useAbsoluteTime", "Use system's absolute time as wallclock time.");
1674     if (_absoluteTime && CmiMyPe() == 0)
1675         printf("Charm++> absolute  timer is used\n");
1676     
1677     _is_global = CmiTimerIsSynchronized();
1678
1679
1680     if (_is_global) {
1681         if (CmiMyRank() == 0) {
1682             clock_gettime(CLOCK_MONOTONIC, &start_ts)
1683         }
1684     } else { /* we don't have a synchronous timer, set our own start time */
1685         CmiBarrier();
1686         CmiBarrier();
1687         CmiBarrier();
1688         clock_gettime(CLOCK_MONOTONIC, &start_ts)
1689     }
1690     CmiNodeAllBarrier();          /* for smp */
1691 }
1692
1693 /**
1694  * Since the timerLock is never created, and is
1695  * always NULL, then all the if-condition inside
1696  * the timer functions could be disabled right
1697  * now in the case of SMP.
1698  */
1699 double CmiTimer(void) {
1700     struct timespec now_ts;
1701     clock_gettime(CLOCK_MONOTONIC, &now_ts)
1702     return _absoluteTime?((double)(now_ts.tv_sec)+(double)now_ts.tv_nsec/1000000000.0)
1703         : (double)( now_ts.tv_sec - start_ts.tv_sec ) + (((double) now_ts.tv_nsec - (double) start_ts.tv_nsec)  / 1000000000.0);
1704 }
1705
1706 double CmiWallTimer(void) {
1707     struct timespec now_ts;
1708     clock_gettime(CLOCK_MONOTONIC, &now_ts)
1709     return _absoluteTime?((double)(now_ts.tv_sec)+(double)now_ts.tv_nsec/1000000000.0)
1710         : (double)( now_ts.tv_sec - start_ts.tv_sec ) + (((double) now_ts.tv_nsec - (double) start_ts.tv_nsec)  / 1000000000.0);
1711 }
1712
1713 double CmiCpuTimer(void) {
1714     struct timespec now_ts;
1715     clock_gettime(CLOCK_MONOTONIC, &now_ts)
1716     return _absoluteTime?((double)(now_ts.tv_sec)+(double)now_ts.tv_nsec/1000000000.0)
1717         : (double)( now_ts.tv_sec - start_ts.tv_sec ) + (((double) now_ts.tv_nsec - (double) start_ts.tv_nsec)  / 1000000000.0);
1718 }
1719
1720 #endif
1721 /************Barrier Related Functions****************/
1722
1723 int CmiBarrier()
1724 {
1725     int status;
1726     status = PMI_Barrier();
1727     return status;
1728
1729 }
1730
1731
1732 #if CMK_PERSISTENT_COMM
1733 #include "machine-persistent.c"
1734 #endif
1735
1736