c30744cfc8c60e9fee410474b377642174f5b34a
[charm.git] / src / ck-com / MsgPacker.h
1 #ifndef MESSAGE_PACKER_H
2 #define MESSAGE_PACKER_H
3
4 #include "charm++.h"
5 #include "envelope.h"
6 #include "ComlibManager.h"
7 #include "register.h"
8 #include "pup_cmialloc.h"
9
10 #define MAX_MESSAGE_SIZE 32768
11
12 class short_envelope {
13  public:
14     UShort epIdx;
15     UShort size;  //Can only send messages up to 64KB :)    
16     
17     CkArrayIndexMax idx;
18     char *data;
19
20     short_envelope();
21     ~short_envelope();
22     inline short_envelope(CkMigrateMessage *){}
23     
24     void pup(PUP::er &p);
25 };
26
27 inline short_envelope::short_envelope(){
28     epIdx = 0;
29     data = NULL;
30 }
31
32 inline short_envelope::~short_envelope(){
33     /*
34       if(data) 
35       CmiFree(data);        
36       data = NULL;
37     */
38 }
39
40 inline void short_envelope::pup(PUP::er &p){    
41   char nints = 0;
42
43   p | epIdx;
44   p | size;        
45   //p | idx;
46   
47   //Complex pup of arrays, even want to save 3 bytes, GREEDY, GREEDY :)
48   if(!p.isUnpacking()) 
49     nints = idx.nInts;
50
51   p | nints;
52   idx.nInts = nints;
53   p((int *)(idx.data()), idx.nInts);
54   
55   p.pupCmiAllocBuf((void **)&data, size);
56 }
57
58 struct CombinedMessage{
59
60     char header[CmiReservedHeaderSize];
61     CkArrayID aid;
62     int srcPE;
63     int nmsgs;
64 };
65
66 PUPbytes(CombinedMessage);
67
68 class MsgPacker {        
69     CkArrayID aid;
70     short_envelope * msgList;
71     int nShortMsgs;   
72
73  public:
74     MsgPacker();
75     ~MsgPacker();    
76     
77     //Makes a message out of a queue of CharmMessageHolders
78     MsgPacker(CkQ<CharmMessageHolder*> &cmsg_list, int n_msgs);
79     
80     //Takes a queue of envelopes as char* ptrs and not charm message holders
81     //Used by mesh streaming strategy
82     MsgPacker::MsgPacker(CkQ<char *> &msgq, int n_msgs);
83     
84     void getMessage(CombinedMessage *&msg, int &size);
85     static void deliver(CombinedMessage *cmb_msg);
86 };
87
88 inline void MsgPacker::deliver(CombinedMessage *cmb_msg){
89
90     CombinedMessage cmb_hdr;
91
92     PUP_fromCmiAllocMem fp(cmb_msg);
93     fp | cmb_hdr;
94
95     int nmsgs = cmb_hdr.nmsgs;
96
97     ComlibPrintf("In MsgPacker::deliver\n");
98     CkArrayID aid = cmb_hdr.aid;
99     int src_pe = cmb_hdr.srcPE;
100     CkArray *a=(CkArray *)_localBranch(aid);
101
102     ArrayElement *a_elem=NULL, *prev_elem=NULL;
103     CkArrayIndexMax prev_idx;
104     prev_idx.nInts = -1;
105
106     for(int count = 0; count < nmsgs; count ++){
107         short_envelope senv;
108         fp | senv;
109         
110         int ep = senv.epIdx;
111         int size = senv.size;
112
113         if(senv.idx == prev_idx) {
114             a_elem = prev_elem;
115         }
116         else {
117             CProxyElement_ArrayBase ap(aid, senv.idx);
118             a_elem = ap.ckLocal();
119         }
120
121         int msgIdx = _entryTable[ep]->msgIdx;
122         if(_entryTable[ep]->noKeep && a_elem != NULL) {
123             //Unpack the message
124             senv.data = (char *)_msgTable[msgIdx]->unpack(senv.data); 
125             CkDeliverMessageReadonly(ep, senv.data, a_elem);            
126
127             prev_elem = a_elem;
128             prev_idx = senv.idx;
129             CmiFree(senv.data);
130         }
131         else {
132             //envelope *env = (envelope *)CmiAlloc(sizeof(envelope) + size);
133             envelope *env = _allocEnv(ForArrayEltMsg, 
134                                       sizeof(envelope) + size);
135
136             void *data = EnvToUsr(env);
137             memcpy(data, senv.data, size);
138             
139             //Unpack the message
140             data = (char *)_msgTable[msgIdx]->unpack(data); 
141             
142             env->getsetArrayMgr() = aid;
143             env->getsetArrayIndex() = senv.idx;
144             env->getsetArrayEp() = ep;
145             env->setPacked(0); 
146             env->getsetArraySrcPe()=src_pe;  
147             env->getsetArrayHops()=1;  
148             env->setQueueing(CK_QUEUEING_FIFO);            
149             env->setUsed(0);
150             env->setMsgIdx(msgIdx);
151
152             env->setTotalsize(sizeof(envelope) + size);
153
154             //if(a_elem)
155             //  CkDeliverMessageFree(ep, data, a_elem);                     
156             //else
157             //ap.ckSend((CkArrayMessage *)data, ep);
158             
159             a->deliver((CkArrayMessage *)data, CkDeliver_queue);
160
161             prev_elem = a_elem;
162             prev_idx = senv.idx;
163             CmiFree(senv.data);
164         }   
165     }      
166         
167     CmiFree(cmb_msg);
168 }
169
170
171 #endif