Cleaned up Mesh Streaming strategy. Also added some more optimizations to MesgPacker
[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
42     p | epIdx;
43     p | size;        
44     //p | idx;
45     
46     if(p.isUnpacking())
47         idx.nInts = 0;
48
49     p((char *)&(idx.nInts), 1);
50     p((int *)(idx.data()), idx.nInts);
51
52     p.pupCmiAllocBuf((void **)&data, size);
53 }
54
55 struct CombinedMessage{
56
57     char header[CmiReservedHeaderSize];
58     CkArrayID aid;
59     int srcPE;
60     int nmsgs;
61 };
62
63 PUPbytes(CombinedMessage);
64
65 class MsgPacker {        
66     CkArrayID aid;
67     short_envelope * msgList;
68     int nShortMsgs;   
69
70  public:
71     MsgPacker();
72     ~MsgPacker();    
73     
74     //Makes a message out of a queue of CharmMessageHolders
75     MsgPacker(CkQ<CharmMessageHolder*> &cmsg_list, int n_msgs);
76     
77     //Takes a queue of envelopes as char* ptrs and not charm message holders
78     //Used by mesh streaming strategy
79     MsgPacker::MsgPacker(CkQ<char *> &msgq, int n_msgs);
80     
81     void getMessage(CombinedMessage *&msg, int &size);
82     static void deliver(CombinedMessage *cmb_msg);
83 };
84
85 inline void MsgPacker::deliver(CombinedMessage *cmb_msg){
86
87     CombinedMessage cmb_hdr;
88
89     PUP_fromCmiAllocMem fp(cmb_msg);
90     fp | cmb_hdr;
91
92     int nmsgs = cmb_hdr.nmsgs;
93
94     ComlibPrintf("In MsgPacker::deliver\n");
95     CkArrayID aid = cmb_hdr.aid;
96     int src_pe = cmb_hdr.srcPE;
97
98     for(int count = 0; count < nmsgs; count ++){
99         short_envelope senv;
100         fp | senv;
101         
102         int ep = senv.epIdx;
103         CkArrayIndexMax idx = senv.idx;
104         int size = senv.size;
105
106         CProxyElement_ArrayBase ap(aid, idx);
107         ArrayElement *a_elem = ap.ckLocal();
108         CkArray *a=(CkArray *)_localBranch(aid);
109
110         int msgIdx = _entryTable[ep]->msgIdx;
111         if(_entryTable[ep]->noKeep && a_elem != NULL) {
112             //Unpack the message
113             senv.data = (char *)_msgTable[msgIdx]->unpack(senv.data); 
114             CkDeliverMessageReadonly(ep, senv.data, a_elem);            
115             CmiFree(senv.data);
116         }
117         else {
118             //envelope *env = (envelope *)CmiAlloc(sizeof(envelope) + size);
119             envelope *env = _allocEnv(ForArrayEltMsg, 
120                                       sizeof(envelope) + size);
121
122             void *data = EnvToUsr(env);
123             memcpy(data, senv.data, size);
124             
125             //Unpack the message
126             data = (char *)_msgTable[msgIdx]->unpack(data); 
127             
128             env->getsetArrayMgr() = aid;
129             env->getsetArrayIndex() = idx;
130             env->getsetArrayEp() = ep;
131             env->setPacked(0); 
132             env->getsetArraySrcPe()=src_pe;  
133             env->getsetArrayHops()=1;  
134             env->setQueueing(CK_QUEUEING_FIFO);            
135             env->setUsed(0);
136             env->setMsgIdx(msgIdx);
137
138             env->setTotalsize(sizeof(envelope) + size);
139
140             //if(a_elem)
141             //  CkDeliverMessageFree(ep, data, a_elem);                     
142             //else
143             //ap.ckSend((CkArrayMessage *)data, ep);
144             
145             a->deliver((CkArrayMessage *)data, CkDeliver_queue);
146
147             CmiFree(senv.data);
148         }        
149     }      
150         
151     CmiFree(cmb_msg);
152 }
153
154
155 #endif