src/ck-com/NodeMulticast.C

   1 // #ifdef filippo
   2
   3 // #include "NodeMulticast.h"
   4 // #include "converse.h"
   5
   6 // #define MAX_BUF_SIZE 165000
   7 // #define MAX_SENDS_PER_BATCH 16
   8 // #define MULTICAST_DELAY 5
   9
  10 // static NodeMulticast *nm_mgr;
  11
  12 // static void call_doneInserting(void *ptr,double curWallTime){
  13 //     NodeMulticast *mgr = (NodeMulticast *)ptr;
  14 //     mgr->doneInserting();
  15 // }
  16
  17 // static void* NodeMulticastHandler(void *msg){
  18 //     ComlibPrintf("In Node MulticastHandler\n");
  19 //     nm_mgr->recvHandler(msg);
  20 //     return NULL;
  21 // }
  22
  23 // static void* NodeMulticastCallbackHandler(void *msg){
  24 //     ComlibPrintf("[%d]:In Node MulticastCallbackHandler\n", CkMyPe());
  25 //     register envelope *env = (envelope *)msg;
  26 //     CkUnpackMessage(&env);
  27 //     //nm_mgr->getCallback().send(EnvToUsr(env));
  28
  29 //     //nm_mgr->getHandler()(env);
  30 //     return NULL;
  31 // }
  32
  33 // //Handles multicast by sending only one message to a nodes and making
  34 // //them multicast locally
  35 // void NodeMulticast::setDestinationArray(CkArrayID a, int nelem,
  36 //                                      CkArrayIndex **idx, int ep){
  37
  38 //     mode = ARRAY_MODE;
  39 //     messageBuf = NULL;
  40 //     pes_per_node = 4;
  41 //     if(getenv("RMS_NODES") != NULL)
  42 //         pes_per_node = CkNumPes()/atoi(getenv("RMS_NODES"));
  43
  44 //     mAid = a;
  45 //     nelements = nelem;
  46 //     entryPoint = ep;
  47
  48 //     numNodes = CkNumPes()/pes_per_node;
  49 //     numCurDestPes = CkNumPes();
  50 //     myRank = 0;
  51 //     nodeMap = new int[numNodes];
  52
  53 //     ComlibPrintf("In SetDestinationArray %d, %d, %d, %d\n", numNodes,
  54 //                  pes_per_node, nelements, ep);
  55
  56 //     indexVec = new CkVec<CkArrayIndex> [CkNumPes()];
  57
  58 //     for(int count = 0; count < nelements; count++) {
  59 //         ComlibPrintf("Before lastKnown %d\n", count);
  60 //         int dest_proc = CkArrayID::CkLocalBranch(a)->lastKnown(*idx[count]);
  61 //         ComlibPrintf("After lastKnown %d\n", dest_proc);
  62 //         nodeMap[dest_proc/pes_per_node] = 1;
  63
  64 //         indexVec[dest_proc].insertAtEnd(*idx[count]);
  65 //     }
  66
  67 //     ComlibPrintf("After SetDestinationArray\n");
  68 // }
  69 // /*
  70 // void NodeMulticast::setPeList(int npes, int *pelist, ComlibMulticastHandler handler){
  71 //     mode = PROCESSOR_MODE;
  72 //     messageBuf = NULL;
  73 //     pes_per_node = 4;
  74 //     //if(getenv("RMS_NODES") != NULL)
  75 //     //pes_per_node = CkNumPes()/atoi(getenv("RMS_NODES"));
  76
  77 //     //cb = callback;
  78 //     this->handler = (long)handler;
  79
  80 //     numNodes = CkNumPes()/pes_per_node;
  81 //     numCurDestPes = npes;
  82
  83 //     myRank = 0;
  84 //     nodeMap = new int[numNodes];
  85
  86 //     this->npes = npes;
  87 //     this->pelist = new int[npes];
  88 //     memcpy(this->pelist, pelist, npes * sizeof(int));
  89
  90 //     ComlibPrintf("In setPeList %d, %d, %d\n", numNodes,
  91 //                  pes_per_node, npes);
  92
  93 //     for(int count = 0; count < npes; count++)
  94 //         nodeMap[pelist[count]/pes_per_node] = 1;
  95
  96 //     ComlibPrintf("After setPeList\n");
  97 // }
  98 // */
  99
 100 // void NodeMulticast::recvHandler(void *msg) {
 101 //     register envelope* env = (envelope *)msg;
 102 //     void *charm_msg = (void *)EnvToUsr(env);
 103
 104 //     env->setUsed(0);
 105 //     ComlibPrintf("In receive Handler\n");
 106 //     if(mode == ARRAY_MODE) {
 107 //         env->setArrayMgr(mAid);
 108 //      env->getsetArrayEp()=entryPoint;
 109 //      env->getsetArrayHops()=0;
 110 //      CkUnpackMessage(&env);
 111
 112 //      for(int count = 0; count < pes_per_node; count ++){
 113 //          int dest_pe = (CkMyPe()/pes_per_node) * pes_per_node + count;
 114 //          int size = indexVec[dest_pe].size();
 115
 116 //          ComlibPrintf("[%d], %d elements to send to %d of size %d\n", CkMyPe(), size, dest_pe, env->getTotalsize());
 117
 118 //          CkArrayIndex * idx_arr = indexVec[dest_pe].getVec();
 119 //          for(int itr = 0; itr < size; itr ++) {
 120 //              void *newcharmmsg = CkCopyMsg(&charm_msg);
 121 //              envelope* newenv = UsrToEnv(newcharmmsg);
 122 //              CProxyElement_ArrayBase ap(mAid, idx_arr[itr]);
 123 //              newenv->getsetArrayIndex()=idx_arr[itr];
 124 //              ap.ckSend((CkArrayMessage *)newcharmmsg, entryPoint);
 125 //          }
 126 //      }
 127 //     }
 128 //     else {
 129 //       CkUnpackMessage(&env);
 130 //       for(int count = 0; count < pes_per_node; count++)
 131 //      if(validRank[count]){
 132 //             void *newcharmmsg;
 133 //             envelope* newenv;
 134
 135 //             if(count <  pes_per_node - 1) {
 136 //                 newcharmmsg = CkCopyMsg(&charm_msg);
 137 //                 newenv = UsrToEnv(newcharmmsg);
 138 //             }
 139 //             else {
 140 //                 newcharmmsg = charm_msg;
 141 //                 newenv = UsrToEnv(newcharmmsg);
 142 //             }
 143
 144 //             CmiSetHandler(newenv, NodeMulticastCallbackHandlerId);
 145 //             ComlibPrintf("[%d] In receive Handler (proc mode), sending message to %d at handler %d\n",
 146 //                          CkMyPe(), (CkMyPe()/pes_per_node) * pes_per_node
 147 //                          + count, NodeMulticastCallbackHandlerId);
 148
 149 //             CkPackMessage(&newenv);
 150 //             CmiSyncSendAndFree((CkMyPe()/pes_per_node) *pes_per_node + count,
 151 //                                newenv->getTotalsize(), (char *)newenv);
 152 //      }
 153 //     }
 154 //     ComlibPrintf("[%d] CmiFree (Code) (%x)\n", CkMyPe(),
 155 //                  (long) msg - 2*sizeof(int));
 156 //     //CmiFree(msg);
 157 // }
 158
 159 // void NodeMulticast::insertMessage(CharmMessageHolder *cmsg){
 160
 161 //     ComlibPrintf("In insertMessage \n");
 162 //     envelope *env = UsrToEnv(cmsg->getCharmMessage());
 163
 164 //     CmiSetHandler(env, NodeMulticastHandlerId);
 165 //     messageBuf->enq(cmsg);
 166 // }
 167
 168 // void NodeMulticast::doneInserting(){
 169 //     CharmMessageHolder *cmsg;
 170 //     char *msg;
 171 //     register envelope *env;
 172
 173 //     ComlibPrintf("NodeMulticast :: doneInserting\n");
 174
 175 //     if(messageBuf->length() > 1) {
 176 //         //CkPrintf("NodeMulticast :: doneInserting length > 1\n");
 177 //         /*
 178 //         char **msgComps;
 179 //         int *sizes, msg_count;
 180
 181 //         msgComps = new char*[messageBuf->length()];
 182 //         sizes = new int[messageBuf->length()];
 183 //         msg_count = 0;
 184 //         while (!messageBuf->isEmpty()) {
 185 //             cmsg = messageBuf->deq();
 186 //             msg = cmsg->getCharmMessage();
 187 //             env = UsrToEnv(msg);
 188 //             sizes[msg_count] = env->getTotalsize();
 189 //             msgComps[msg_count] = (char *)env;
 190 //             msg_count++;
 191
 192 //             delete cmsg;
 193 //         }
 194
 195 //         for(int count = 0; count < numNodes; count++)
 196 //             if(nodeMap[count])
 197 //                 CmiMultipleSend(count * pes_per_node + myRank, msg_count,
 198 //                                 sizes, msgComps);
 199
 200 //         delete [] msgComps;
 201 //         delete [] sizes;
 202 //         */
 203 //     }
 204 //     else if (messageBuf->length() == 1){
 205 //         static int prevCount = 0;
 206 //         int count = 0;
 207 //         ComlibPrintf("Sending Node Multicast\n");
 208 //         cmsg = messageBuf->deq();
 209 //         msg = cmsg->getCharmMessage();
 210 //         env = UsrToEnv(msg);
 211
 212 //      if(mode == ARRAY_MODE)
 213 //          env->getsetArraySrcPe()=CkMyPe();
 214 //      CkPackMessage(&env);
 215
 216 //         CmiSetHandler(env, NodeMulticastHandlerId);
 217 //         ComlibPrintf("After set handler\n");
 218
 219 //         //CmiPrintf("cursedtpes = %d, %d\n", cmsg->npes, numCurDestPes);
 220
 221 //         if((mode != ARRAY_MODE) && cmsg->npes < numCurDestPes) {
 222 //             numCurDestPes = cmsg->npes;
 223 //             for(count = 0; count < numNodes; count++)
 224 //                 nodeMap[count] = 0;
 225
 226 //             for(count = 0; count < cmsg->npes; count++)
 227 //                 nodeMap[(cmsg->pelist[count])/pes_per_node] = 1;
 228 //         }
 229
 230 //         for(count = prevCount; count < numNodes; count++) {
 231 //          //int dest_node = count;
 232 //          int dest_node = (count + (CkMyPe()/pes_per_node))%numNodes;
 233 //          if(nodeMap[dest_node]) {
 234 //                 void *newcharmmsg;
 235 //                 envelope* newenv;
 236
 237 //                 if(count < numNodes - 1) {
 238 //                     newcharmmsg = CkCopyMsg((void **)&msg);
 239 //                     newenv = UsrToEnv(newcharmmsg);
 240 //                 }
 241 //                 else {
 242 //                     newcharmmsg = msg;
 243 //                     newenv = UsrToEnv(newcharmmsg);
 244 //                 }
 245
 246 //              ComlibPrintf("[%d]In cmisyncsend to %d\n", CkMyPe(),
 247 //                           dest_node * pes_per_node + myRank);
 248 // #if CMK_PERSISTENT_COMM
 249 //              if(env->getTotalsize() < MAX_BUF_SIZE)
 250 //                     CmiUsePersistentHandle(&persistentHandlerArray[dest_node],1);
 251 // #endif
 252 //              CkPackMessage(&newenv);
 253 //              CmiSyncSendAndFree(dest_node * pes_per_node + myRank,
 254 //                                 newenv->getTotalsize(), (char *)newenv);
 255 // #if CMK_PERSISTENT_COMM
 256 //              if(env->getTotalsize() < MAX_BUF_SIZE)
 257 //                     CmiUsePersistentHandle(NULL, 0);
 258 // #endif
 259 //             }
 260 //             prevCount ++;
 261 //             if((prevCount % MAX_SENDS_PER_BATCH == 0) &&
 262 //                (prevCount != numNodes)) {
 263 //                 CcdCallFnAfterOnPE((CcdVoidFn)call_doneInserting, (void *)this,
 264 //                                MULTICAST_DELAY, CkMyPe());
 265 //                 return;
 266 //             }
 267 //             prevCount = 0;
 268 //      }
 269
 270 //         ComlibPrintf("[%d] CmiFree (Code) (%x)\n", CkMyPe(), (char *)env - 2*sizeof(int));
 271 //         //CmiFree(env);
 272 //         delete cmsg;
 273 //     }
 274 // }
 275
 276 // void NodeMulticast::pup(PUP::er &p){
 277
 278 //     CharmStrategy::pup(p);
 279
 280 //     p | pes_per_node;
 281 //     p | numNodes;
 282 //     p | nelements;
 283 //     p | entryPoint;
 284 //     p | npes;
 285 //     p | mode;
 286 //     p | numCurDestPes;
 287 //     p | mAid;
 288
 289 //     if(p.isUnpacking()) {
 290 //         nodeMap = new int[numNodes];
 291
 292 //      if(mode == ARRAY_MODE) {
 293 //          typedef CkVec<CkArrayIndex> CkVecArrayIndex;
 294 //          CkVecArrayIndex *vec = new CkVecArrayIndex[CkNumPes()];
 295 //          indexVec = vec;
 296 //      }
 297
 298 //      if(mode == PROCESSOR_MODE)
 299 //          pelist = new int[npes];
 300 //     }
 301
 302 //     p | cb;
 303 //     p | handler;
 304 //     p(nodeMap, numNodes);
 305
 306 //     if(mode == PROCESSOR_MODE)
 307 //       p(pelist, npes);
 308
 309 //     if(mode == ARRAY_MODE)
 310 //       for(int count = 0; count < CkNumPes(); count++)
 311 //         p | indexVec[count];
 312
 313 //     if(p.isUnpacking()) {
 314 //         messageBuf = new CkQ <CharmMessageHolder *>;
 315 //         myRank = CkMyPe() % pes_per_node;
 316
 317 //         NodeMulticastHandlerId = CkRegisterHandler((CmiHandler)NodeMulticastHandler);
 318 //      NodeMulticastCallbackHandlerId = CkRegisterHandler
 319 //          ((CmiHandler)NodeMulticastCallbackHandler);
 320
 321 //         nm_mgr = this;
 322
 323 //      //validRank[0] =  validRank[1] = validRank[2] = validRank[3] = 0;
 324 //         memset(validRank, 0, MAX_PES_PER_NODE * sizeof(int));
 325 //      for(int count = 0; count < npes; count ++){
 326 //          if(CkMyPe()/pes_per_node == pelist[count] / pes_per_node)
 327 //              validRank[pelist[count] % pes_per_node] = 1;
 328 //      }
 329
 330 // #if CMK_PERSISTENT_COMM
 331 //      persistentHandlerArray = new PersistentHandle[numNodes];
 332 //      for(int count = 0; count < numNodes; count ++)
 333 //             //if(nodeMap[count])
 334 //             persistentHandlerArray[count] = CmiCreatePersistent
 335 //                 (count * pes_per_node + myRank, MAX_BUF_SIZE);
 336 // #endif
 337 //     }
 338 // }
 339
 340 // //PUPable_def(NodeMulticast);
 341
 342 // #endif