Merge branch 'ramv/cleanup-arridx-hierarchy' into charm
[charm.git] / src / ck-com / EachToManyMulticastStrategy.C
1 /**
2    @addtogroup ComlibCharmStrategy
3    @{
4    @file 
5 */
6
7 #include "EachToManyMulticastStrategy.h"
8 #include "string.h"
9
10 #include "AAPLearner.h"
11 #include "AAMLearner.h"
12
13
14 //Group Constructor
15 EachToManyMulticastStrategy::EachToManyMulticastStrategy(int substrategy,
16                 CkGroupID src,
17                 CkGroupID dest,
18                 int n_srcpes, 
19                 int *src_pelist,
20                 int n_destpes, 
21                 int *dest_pelist) 
22 : RouterStrategy(substrategy), CharmStrategy() {
23         ComlibPrintf("[%d] EachToManyMulticast group constructor\n",CkMyPe());
24
25         setType(GROUP_STRATEGY);
26
27         //CkGroupID gid;
28         //gid.setZero();
29         ginfo.setSourceGroup(src, src_pelist, n_srcpes);    
30         ginfo.setDestinationGroup(dest, dest_pelist, n_destpes);
31
32         //Written in this funny way to be symmetric with the array case.
33         //ginfo.getDestinationGroup(gid, destpelist, ndestpes);
34         //ginfo.getCombinedPeList(pelist, npes);
35
36         commonInit(ginfo.getCombinedCountList()); // The array returned by getCombinedCountList is deleted inside commonInit
37 }
38
39 //Array Constructor
40 EachToManyMulticastStrategy::EachToManyMulticastStrategy(int substrategy, 
41                 CkArrayID src, 
42                 CkArrayID dest, 
43                 int nsrc, 
44                 CkArrayIndex *srcelements, 
45                 int ndest, 
46                 CkArrayIndex *destelements)
47 : RouterStrategy(substrategy), CharmStrategy() {
48         ComlibPrintf("[%d] EachToManyMulticast array constructor. nsrc=%d ndest=%d\n",CkMyPe(), nsrc, ndest);
49
50         setType(ARRAY_STRATEGY);
51         ainfo.setSourceArray(src, srcelements, nsrc);
52         ainfo.setDestinationArray(dest, destelements, ndest);
53
54         int *count = ainfo.getCombinedCountList();
55         //ainfo.getSourcePeList(nsrcPe, srcPe);
56         //ainfo.getDestinationPeList(ndestPe, destPe);
57
58         commonInit(count);
59 }
60
61 extern char *routerName;
62 //Common initialization for both group and array constructors
63 void EachToManyMulticastStrategy::commonInit(int *count) {
64
65         setBracketed();
66         //setForwardOnMigration(1);
67
68         if(CkMyPe() == 0 && routerName != NULL){
69                 if(strcmp(routerName, "USE_MESH") == 0)
70                         routerIDsaved = USE_MESH;
71                 else if(strcmp(routerName, "USE_GRID") == 0)
72                         routerIDsaved = USE_GRID;
73                 else  if(strcmp(routerName, "USE_HYPERCUBE") == 0)
74                         routerIDsaved = USE_HYPERCUBE;
75                 else  if(strcmp(routerName, "USE_DIRECT") == 0)
76                         routerIDsaved = USE_DIRECT;        
77                 else  if(strcmp(routerName, "USE_PREFIX") == 0)
78                         routerIDsaved = USE_PREFIX;        
79
80                 //Just for the first step. After learning the learned
81                 //strategies will be chosen
82                 //router = NULL;
83         }
84
85         //ComlibPrintf("Creating Strategy %d\n", routerID);
86
87         // finish the creation of the RouterStrategy superclass
88         bracketedUpdatePeKnowledge(count);
89         delete [] count;
90         //rstrat = NULL;
91 }
92
93 EachToManyMulticastStrategy::~EachToManyMulticastStrategy() {
94 }
95
96
97 void EachToManyMulticastStrategy::insertMessage(CharmMessageHolder *cmsg){
98
99   cmsg -> checkme();
100
101         ComlibPrintf("[%d] EachToManyMulticast: insertMessage\n", CkMyPe());
102
103         envelope *env = UsrToEnv(cmsg->getCharmMessage());
104
105         if(cmsg->dest_proc == IS_BROADCAST) {
106                 //All to all multicast
107
108                 // not necessary to set cmsg since the superclass will take care
109                 //cmsg->npes = ndestPes;
110                 //cmsg->pelist = destPes;
111
112                 //Use Multicast Learner (Foobar will not work for combinations
113                 //of personalized and multicast messages
114
115                 // this handler will ensure that handleMessage is called if direcly
116                 // sent, or deliver is called if normal routing path is taken
117                 CmiSetHandler(env, CkpvAccess(comlib_handler));
118
119                 //Collect Multicast Statistics
120                 RECORD_SENDM_STATS(getInstance(), env->getTotalsize(), 
121                                 cmsg->pelist, cmsg->npes);
122         }
123         else {
124                 //All to all personalized
125
126                 //Collect Statistics
127                 RECORD_SEND_STATS(getInstance(), env->getTotalsize(), 
128                                 cmsg->dest_proc);
129         }
130
131         RouterStrategy::insertMessage(cmsg);
132 }
133
134
135 void EachToManyMulticastStrategy::pup(PUP::er &p){
136
137         ComlibPrintf("[%d] EachToManyMulticastStrategy::pup called for %s\n", CkMyPe(), 
138                         p.isPacking()?"packing":(p.isUnpacking()?"unpacking":"sizing"));
139
140         RouterStrategy::pup(p);
141         CharmStrategy::pup(p);
142
143 }
144
145
146
147 void EachToManyMulticastStrategy::deliver(char *msg, int size) {
148         ComlibPrintf("[%d] EachToManyMulticastStrategy::deliver for %s\n",
149                         CkMyPe(), isAllToAll()?"multicast":"personalized");
150         
151         envelope *env = (envelope *)msg;
152         RECORD_RECV_STATS(myHandle, env->getTotalsize(), env->getSrcPe());
153
154         if (isAllToAll()){
155                 ComlibPrintf("Delivering via localMulticast()\n");
156                 localMulticast(msg);
157         }
158         else { 
159                 if (getType() == GROUP_STRATEGY) {
160                         ComlibPrintf("Delivering via personalized CkSendMsgBranchInline\n");
161                         CkUnpackMessage(&env);
162                         CkSendMsgBranchInline(env->getEpIdx(), EnvToUsr(env), CkMyPe(), env->getGroupNum());
163                 }
164                 else if (getType() == ARRAY_STRATEGY) {
165                         //        ComlibPrintf("[%d] Delivering via ComlibArrayInfo::deliver()\n", CkMyPe());
166                         //      ComlibArrayInfo::deliver(env);
167
168                         ComlibPrintf("[%d] Delivering via ComlibArrayInfo::localBroadcast()\n", CkMyPe());
169                         ainfo.localBroadcast(env);
170
171                 }
172         }
173 }
174
175 void EachToManyMulticastStrategy::localMulticast(void *msg){
176         register envelope *env = (envelope *)msg;
177         CkUnpackMessage(&env);
178         ComlibPrintf("localMulticast calls ainfo.localBroadcast()\n");
179         ainfo.localBroadcast(env);
180 }
181
182 void EachToManyMulticastStrategy::notifyDone() {
183         if (!getOnFinish().isInvalid()) getOnFinish().send(0);
184         RouterStrategy::notifyDone();
185 }
186
187 /*@}*/