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