netlrts: replace some CMK_SHARED_VARS_UNAVAILABLE with not CMK_SMP
[charm.git] / src / arch / net / machine-persistent.c
1 /*
2  * =====================================================================================
3  *
4  *       Filename:  machine-persistent.c
5  *
6  *    Description:  
7  *
8  *        Version:  1.0
9  *        Created:  02/11/2013 14:33:51
10  *       Revision:  none
11  *       Compiler:  gcc
12  *
13  *         Author:  YOUR NAME (Yanhua), 
14  *   Organization:  
15  *
16  * =====================================================================================
17  */
18 #include "persist_impl.h"
19
20 #define CMI_DEST_RANK_NET(msg)  *(int *)(msg)
21 int persistentSendMsgHandlerIdx;
22
23 static void sendPerMsgHandler(char *msg)
24 {
25   int msgSize;
26   void *destAddr, *destSizeAddr;
27   int ep;
28
29   msgSize = CmiMsgHeaderGetLength(msg);
30   msgSize -= (2*sizeof(void *)+sizeof(int));
31   ep = *(int*)(msg+msgSize);
32   destAddr = *(void **)(msg + msgSize + sizeof(int));
33   destSizeAddr = *(void **)(msg + msgSize + sizeof(int) + sizeof(void*));
34 /*CmiPrintf("msgSize:%d destAddr:%p, destSizeAddr:%p\n", msgSize, destAddr, destSizeAddr);*/
35   CmiSetHandler(msg, ep);
36   *((int *)destSizeAddr) = msgSize;
37   memcpy(destAddr, msg, msgSize);
38 }
39
40 void CmiSendPersistentMsg(PersistentHandle h, int destPE, int size, void *m)
41 {
42   CmiAssert(h!=NULL);
43   PersistentSendsTable *slot = (PersistentSendsTable *)h;
44   CmiAssert(slot->used == 1);
45   CmiAssert(slot->destPE == destPE);
46   if (size > slot->sizeMax) {
47     CmiPrintf("size: %d sizeMax: %d\n", size, slot->sizeMax);
48     CmiAbort("Abort: Invalid size\n");
49   }
50
51 /*CmiPrintf("[%d] CmiSendPersistentMsg h=%p hdl=%d destpe=%d destAddress=%p size=%d\n", CmiMyPe(), *phs, CmiGetHandler(m), slot->destPE, slot->destAddress, size);*/
52
53   if (slot->destAddress[0]) {
54     int oldep = CmiGetHandler(m);
55     int newsize = size + sizeof(void *)*2 + sizeof(int);
56     char *newmsg = (char*)CmiAlloc(newsize);
57     memcpy(newmsg, m, size);
58     memcpy(newmsg+size, &oldep, sizeof(int));
59     memcpy(newmsg+size+sizeof(int), &slot->destAddress[0], sizeof(void *));
60     memcpy(newmsg+size+sizeof(int)+sizeof(void*), &slot->destSizeAddress[0], sizeof(void *));
61     CmiFree(m);
62     CmiMsgHeaderSetLength(newmsg, newsize);
63     CmiSetHandler(newmsg, persistentSendMsgHandlerIdx);
64     phs = NULL; phsSize = 0;
65     CmiSyncSendAndFree(slot->destPE, newsize, newmsg);
66   }
67   else {
68 #if 1
69     /* buffer until ready */
70     if (slot->messageBuf != NULL) {
71       CmiPrintf("Unexpected message in buffer on %d\n", CmiMyPe());
72       CmiAbort("");
73     }
74     slot->messageBuf = m;
75     slot->messageSize = size;
76 #else
77     /* normal send */
78     PersistentHandle  *phs_tmp = phs;
79     int phsSize_tmp = phsSize;
80     phs = NULL; phsSize = 0;
81     CmiPrintf("[%d]Slot sending message directly\n", CmiMyPe());
82     CmiSyncSendAndFree(slot->destPE, size, m);
83     phs = phs_tmp; phsSize = phsSize_tmp;
84 #endif
85   }
86 }
87
88 void CmiSyncSendPersistent(int destPE, int size, char *msg, PersistentHandle h)
89 {
90   char *dupmsg = (char *) CmiAlloc(size);
91   memcpy(dupmsg, msg, size);
92
93   /*  CmiPrintf("Setting root to %d\n", 0); */
94   if (CmiMyPe()==destPE) {
95     CQdCreate(CpvAccess(cQdState), 1);
96     CdsFifo_Enqueue(CpvAccess(CmiLocalQueue),dupmsg);
97   }
98   else
99     CmiSendPersistentMsg(h, destPE, size, dupmsg);
100 }
101
102 /* called in PumpMsgs */
103 int PumpPersistent()
104 {
105   PersistentReceivesTable *slot = persistentReceivesTableHead;
106   int status = 0;
107   while (slot) {
108     unsigned int size = *(slot->recvSizePtr[0]);
109     if (size > 0)
110     {
111       char *msg = slot->messagePtr[0];
112 /*CmiPrintf("[%d] size: %d rank:%d msg:%p %p\n", CmiMyPe(), size, CMI_DEST_RANK(msg), msg, slot->messagePtr);*/
113
114 #if 0
115       void *dupmsg;
116       dupmsg = CmiAlloc(size);
117       
118       _MEMCHECK(dupmsg);
119       memcpy(dupmsg, msg, size);
120       msg = dupmsg;
121 #else
122       /* return messagePtr directly and user MUST make sure not to delete it. */
123       /*CmiPrintf("[%d] %p size:%d rank:%d root:%d\n", CmiMyPe(), msg, size, CMI_DEST_RANK(msg), CMI_BROADCAST_ROOT(msg));*/
124
125       CmiReference(msg);
126 #endif
127       CmiPushPE(CMI_DEST_RANK_NET(msg), msg);
128 #if CMK_BROADCAST_SPANNING_TREE
129       if (CMI_BROADCAST_ROOT(msg))
130           SendSpanningChildrenNet(size, msg);
131 #endif
132       *(slot->recvSizePtr[0]) = 0;
133       status = 1;
134     }
135     slot = slot->next;
136   }
137   return status;
138 }
139
140 void *PerAlloc(int size)
141 {
142   return CmiAlloc(size);
143 }
144                                                                                 
145 void PerFree(char *msg)
146 {
147     CmiFree(msg);
148 }
149
150 void persist_machine_init() 
151 {
152   persistentSendMsgHandlerIdx =
153        CmiRegisterHandler((CmiHandler)sendPerMsgHandler);
154 }
155
156 void setupRecvSlot(PersistentReceivesTable *slot, int maxBytes)
157 {
158   int i;
159   for (i=0; i<PERSIST_BUFFERS_NUM; i++) {
160     char *buf = PerAlloc(maxBytes+sizeof(int)*2);
161     _MEMCHECK(buf);
162     memset(buf, 0, maxBytes+sizeof(int)*2);
163     slot->messagePtr[i] = buf;
164     slot->recvSizePtr[i] = (unsigned int*)CmiAlloc(sizeof(unsigned int));
165     *(slot->recvSizePtr[0]) = 0;
166   }
167   slot->sizeMax = maxBytes;
168 }
169
170