src/smpi/smpi_comm.c

   1 /* Copyright (c) 2010-2014. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include <stdlib.h>
   8
   9 #include "private.h"
  10 #include "smpi_mpi_dt_private.h"
  11 #include "limits.h"
  12 #include "simix/smx_private.h"
  13 #include "xbt/replay.h"
  14 #include "colls/colls.h"
  15
  16 extern int is_replay_active ;
  17
  18 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_comm, smpi,
  19                                 "Logging specific to SMPI (comm)");
  20
  21
  22
  23 /* Support for cartesian topology was added, but there are 2 other types of
  24  * topology, graph et dist graph. In order to support them, we have to add a
  25  * field MPIR_Topo_type, and replace the MPI_Topology field by an union. */
  26
  27 typedef struct s_smpi_mpi_communicator {
  28   MPI_Group group;
  29   MPIR_Topo_type topoType;
  30   MPI_Topology topo; // to be replaced by an union
  31   int refcount;
  32   MPI_Comm leaders_comm;//inter-node communicator
  33   MPI_Comm intra_comm;//intra-node communicator . For MPI_COMM_WORLD this can't be used, as var is global.
  34   //use an intracomm stored in the process data instead
  35   int* leaders_map; //who is the leader of each process
  36   int is_uniform;
  37   int* non_uniform_map; //set if smp nodes have a different number of processes allocated
  38   int is_blocked;// are ranks allocated on the same smp node contiguous ?
  39 } s_smpi_mpi_communicator_t;
  40
  41 static int smpi_compare_rankmap(const void *a, const void *b)
  42 {
  43   const int* x = (const int*)a;
  44   const int* y = (const int*)b;
  45
  46   if (x[1] < y[1]) {
  47     return -1;
  48   }
  49   if (x[1] == y[1]) {
  50     if (x[0] < y[0]) {
  51       return -1;
  52     }
  53     if (x[0] == y[0]) {
  54       return 0;
  55     }
  56     return 1;
  57   }
  58   return 1;
  59 }
  60
  61 MPI_Comm smpi_comm_new(MPI_Group group, MPI_Topology topo)
  62 {
  63   MPI_Comm comm;
  64
  65   comm = xbt_new(s_smpi_mpi_communicator_t, 1);
  66   comm->group = group;
  67   smpi_group_use(comm->group);
  68   comm->refcount=1;
  69   comm->topoType = -1;
  70   comm->topo = topo;
  71   comm->intra_comm = MPI_COMM_NULL;
  72   comm->leaders_comm = MPI_COMM_NULL;
  73   comm->is_uniform=1;
  74   comm->non_uniform_map = NULL;
  75   comm->leaders_map = NULL;
  76   comm->is_blocked=0;
  77   return comm;
  78 }
  79
  80 void smpi_comm_destroy(MPI_Comm comm)
  81 {
  82   if (comm == MPI_COMM_UNINITIALIZED)
  83     comm = smpi_process_comm_world();
  84   smpi_group_unuse(comm->group);
  85   smpi_topo_destroy(comm->topo); // there's no use count on topos
  86   smpi_comm_unuse(comm);
  87 }
  88
  89 MPI_Group smpi_comm_group(MPI_Comm comm)
  90 {
  91   if (comm == MPI_COMM_UNINITIALIZED)
  92     comm = smpi_process_comm_world();
  93
  94   return comm->group;
  95 }
  96
  97 MPI_Topology smpi_comm_topo(MPI_Comm comm) {
  98   if (comm != MPI_COMM_NULL)
  99     return comm->topo;
 100   return NULL;
 101 }
 102
 103 int smpi_comm_size(MPI_Comm comm)
 104 {
 105   if (comm == MPI_COMM_UNINITIALIZED)
 106     comm = smpi_process_comm_world();
 107
 108   return smpi_group_size(smpi_comm_group(comm));
 109 }
 110
 111 int smpi_comm_rank(MPI_Comm comm)
 112 {
 113   if (comm == MPI_COMM_UNINITIALIZED)
 114     comm = smpi_process_comm_world();
 115   return smpi_group_rank(smpi_comm_group(comm), smpi_process_index());
 116 }
 117
 118 void smpi_comm_get_name (MPI_Comm comm, char* name, int* len)
 119 {
 120   if (comm == MPI_COMM_UNINITIALIZED)
 121     comm = smpi_process_comm_world();
 122   if(comm == MPI_COMM_WORLD) {
 123     strcpy(name, "WORLD");
 124     *len = 5;
 125   } else {
 126     *len = snprintf(name, MPI_MAX_NAME_STRING, "%p", comm);
 127   }
 128 }
 129
 130 void smpi_comm_set_leaders_comm(MPI_Comm comm, MPI_Comm leaders){
 131   comm->leaders_comm=leaders;
 132 }
 133
 134 void smpi_comm_set_intra_comm(MPI_Comm comm, MPI_Comm leaders){
 135   comm->intra_comm=leaders;
 136 }
 137
 138 int* smpi_comm_get_non_uniform_map(MPI_Comm comm){
 139   return comm->non_uniform_map;
 140 }
 141
 142 int* smpi_comm_get_leaders_map(MPI_Comm comm){
 143   return comm->leaders_map;
 144 }
 145
 146 MPI_Comm smpi_comm_get_leaders_comm(MPI_Comm comm){
 147   return comm->leaders_comm;
 148 }
 149
 150 MPI_Comm smpi_comm_get_intra_comm(MPI_Comm comm){
 151   if(comm==MPI_COMM_WORLD) return smpi_process_get_comm_intra();
 152   else return comm->intra_comm;
 153 }
 154
 155 int smpi_comm_is_uniform(MPI_Comm comm){
 156   return comm->is_uniform;
 157 }
 158
 159 int smpi_comm_is_blocked(MPI_Comm comm){
 160   return comm->is_blocked;
 161 }
 162
 163 MPI_Comm smpi_comm_split(MPI_Comm comm, int color, int key)
 164 {
 165   if (comm == MPI_COMM_UNINITIALIZED)
 166     comm = smpi_process_comm_world();
 167   int system_tag = 123;
 168   int index, rank, size, i, j, count, reqs;
 169   int* sendbuf;
 170   int* recvbuf;
 171   int* rankmap;
 172   MPI_Group group, group_root, group_out;
 173   MPI_Request* requests;
 174
 175   group_root = group_out = NULL;
 176   group = smpi_comm_group(comm);
 177   rank = smpi_comm_rank(comm);
 178   size = smpi_comm_size(comm);
 179   /* Gather all colors and keys on rank 0 */
 180   sendbuf = xbt_new(int, 2);
 181   sendbuf[0] = color;
 182   sendbuf[1] = key;
 183   if(rank == 0) {
 184     recvbuf = xbt_new(int, 2 * size);
 185   } else {
 186     recvbuf = NULL;
 187   }
 188   smpi_mpi_gather(sendbuf, 2, MPI_INT, recvbuf, 2, MPI_INT, 0, comm);
 189   xbt_free(sendbuf);
 190   /* Do the actual job */
 191   if(rank == 0) {
 192     rankmap = xbt_new(int, 2 * size);
 193     for(i = 0; i < size; i++) {
 194       if(recvbuf[2 * i] == MPI_UNDEFINED) {
 195         continue;
 196       }
 197       count = 0;
 198       for(j = i + 1; j < size; j++)  {
 199         if(recvbuf[2 * i] == recvbuf[2 * j]) {
 200           recvbuf[2 * j] = MPI_UNDEFINED;
 201           rankmap[2 * count] = j;
 202           rankmap[2 * count + 1] = recvbuf[2 * j + 1];
 203           count++;
 204         }
 205       }
 206       /* Add self in the group */
 207       recvbuf[2 * i] = MPI_UNDEFINED;
 208       rankmap[2 * count] = i;
 209       rankmap[2 * count + 1] = recvbuf[2 * i + 1];
 210       count++;
 211       qsort(rankmap, count, 2 * sizeof(int), &smpi_compare_rankmap);
 212       group_out = smpi_group_new(count);
 213       if(i == 0) {
 214         group_root = group_out; /* Save root's group */
 215       }
 216       for(j = 0; j < count; j++) {
 217         //increment refcounter in order to avoid freeing the group too quick before copy
 218         index = smpi_group_index(group, rankmap[2 * j]);
 219         smpi_group_set_mapping(group_out, index, j);
 220       }
 221       requests = xbt_new(MPI_Request, count);
 222       reqs = 0;
 223       for(j = 0; j < count; j++) {
 224         if(rankmap[2 * j] != 0) {
 225           requests[reqs] = smpi_isend_init(&group_out, 1, MPI_PTR, rankmap[2 * j], system_tag, comm);
 226           reqs++;
 227         }
 228       }
 229       smpi_mpi_startall(reqs, requests);
 230       smpi_mpi_waitall(reqs, requests, MPI_STATUS_IGNORE);
 231       xbt_free(requests);
 232     }
 233     xbt_free(recvbuf);
 234     group_out = group_root; /* exit with root's group */
 235   } else {
 236     if(color != MPI_UNDEFINED) {
 237       smpi_mpi_recv(&group_out, 1, MPI_PTR, 0, system_tag, comm, MPI_STATUS_IGNORE);
 238       if(group_out){
 239         group_out=smpi_group_copy(group_out);
 240       }
 241     } /* otherwise, exit with group_out == NULL */
 242   }
 243   return group_out ? smpi_comm_new(group_out, NULL) : MPI_COMM_NULL;
 244 }
 245
 246 void smpi_comm_use(MPI_Comm comm){
 247   if (comm == MPI_COMM_UNINITIALIZED)
 248     comm = smpi_process_comm_world();
 249   comm->refcount++;
 250 }
 251
 252 void smpi_comm_unuse(MPI_Comm comm){
 253   if (comm == MPI_COMM_UNINITIALIZED)
 254     comm = smpi_process_comm_world();
 255   comm->refcount--;
 256   if(comm->refcount==0){
 257     if(comm->intra_comm != MPI_COMM_NULL)
 258       smpi_comm_unuse(comm->intra_comm);
 259     if(comm->leaders_comm != MPI_COMM_NULL)
 260       smpi_comm_unuse(comm->leaders_comm);
 261     if(comm->non_uniform_map !=NULL)
 262       xbt_free(comm->non_uniform_map);
 263     if(comm->leaders_map !=NULL)
 264       xbt_free(comm->leaders_map);
 265     xbt_free(comm);
 266   }
 267 }
 268
 269 static int
 270 compare_ints (const void *a, const void *b)
 271 {
 272   const int *da = (const int *) a;
 273   const int *db = (const int *) b;
 274
 275   return (*da > *db) - (*da < *db);
 276 }
 277
 278 void smpi_comm_init_smp(MPI_Comm comm){
 279   int leader = -1;
 280   int comm_size =smpi_comm_size(comm);
 281
 282   // If we are in replay - perform an ugly hack
 283   // say to SimGrid that we are not in replay for a while, because we need
 284   // the buffers to be copied for the following calls
 285   int replaying = 0; //cache data to set it back again after
 286   if(_xbt_replay_is_active()){
 287     replaying = 1;
 288     is_replay_active = 0 ;
 289   }
 290
 291   if(smpi_privatize_global_variables){ //we need to switch here, as the called function may silently touch global variables
 292      smpi_switch_data_segment(smpi_process_index());
 293    }
 294   //identify neighbours in comm
 295   //get the indexes of all processes sharing the same simix host
 296   xbt_swag_t process_list = simcall_host_get_process_list(SIMIX_host_self());
 297   int intra_comm_size = 0;
 298   //only one process/node, disable SMP support and return
 299 //  if(intra_comm_size==1){
 300 //      smpi_comm_set_intra_comm(comm, MPI_COMM_SELF);
 301 //      //smpi_comm_set_leaders_comm(comm, comm);
 302 //      smpi_process_set_comm_intra(MPI_COMM_SELF);
 303 //      return;
 304 //  }
 305
 306
 307   int i =0;
 308   int min_index=INT_MAX;//the minimum index will be the leader
 309   msg_process_t process = NULL;
 310   xbt_swag_foreach(process, process_list) {
 311     //is_in_comm=0;
 312     int index = SIMIX_process_get_PID(process) -1;
 313
 314     if(smpi_group_rank(smpi_comm_group(comm),  index)!=MPI_UNDEFINED){
 315         intra_comm_size++;
 316       //the process is in the comm
 317       if(index < min_index)
 318         min_index=index;
 319       i++;
 320     }
 321   }
 322   XBT_DEBUG("number of processes deployed on my node : %d", intra_comm_size);
 323   MPI_Group group_intra = smpi_group_new(intra_comm_size);
 324   i=0;
 325   process = NULL;
 326   xbt_swag_foreach(process, process_list) {
 327     //is_in_comm=0;
 328     int index = SIMIX_process_get_PID(process) -1;
 329     if(smpi_group_rank(smpi_comm_group(comm),  index)!=MPI_UNDEFINED){
 330       smpi_group_set_mapping(group_intra, index, i);
 331       i++;
 332     }
 333   }
 334
 335
 336   MPI_Comm comm_intra = smpi_comm_new(group_intra, NULL);
 337   //MPI_Comm shmem_comm = smpi_process_comm_intra();
 338   //int intra_rank = smpi_comm_rank(shmem_comm);
 339
 340
 341   //if(smpi_process_index()==min_index)
 342   leader=min_index;
 343
 344   int * leaders_map= (int*)xbt_malloc0(sizeof(int)*comm_size);
 345   int * leader_list= (int*)xbt_malloc0(sizeof(int)*comm_size);
 346   for(i=0; i<comm_size; i++){
 347       leader_list[i]=-1;
 348   }
 349
 350   smpi_coll_tuned_allgather_mpich(&leader, 1, MPI_INT , leaders_map, 1, MPI_INT, comm);
 351
 352   if(smpi_privatize_global_variables){ //we need to switch here, as the called function may silently touch global variables
 353      switch_data_segment(smpi_process_index());
 354    }
 355
 356   if(!comm->leaders_map){
 357     comm->leaders_map= leaders_map;
 358   }else{
 359     xbt_free(leaders_map);
 360   }
 361   int j=0;
 362   int leader_group_size = 0;
 363   for(i=0; i<comm_size; i++){
 364       int already_done=0;
 365       for(j=0;j<leader_group_size; j++){
 366         if(comm->leaders_map[i]==leader_list[j]){
 367             already_done=1;
 368         }
 369       }
 370       if(!already_done){
 371         leader_list[leader_group_size]=comm->leaders_map[i];
 372         leader_group_size++;
 373       }
 374   }
 375   qsort(leader_list, leader_group_size, sizeof(int),compare_ints);
 376
 377   MPI_Group leaders_group = smpi_group_new(leader_group_size);
 378
 379
 380   MPI_Comm leader_comm = MPI_COMM_NULL;
 381   if(comm!=MPI_COMM_WORLD){
 382     //create leader_communicator
 383     for (i=0; i< leader_group_size;i++)
 384       smpi_group_set_mapping(leaders_group, leader_list[i], i);
 385     leader_comm = smpi_comm_new(leaders_group, NULL);
 386     smpi_comm_set_leaders_comm(comm, leader_comm);
 387     smpi_comm_set_intra_comm(comm, comm_intra);
 388
 389     //create intracommunicator
 390    // smpi_comm_set_intra_comm(comm, smpi_comm_split(comm, *(int*)SIMIX_host_self(), comm_rank));
 391   }else{
 392     for (i=0; i< leader_group_size;i++)
 393       smpi_group_set_mapping(leaders_group, leader_list[i], i);
 394
 395     leader_comm = smpi_comm_new(leaders_group, NULL);
 396     if(smpi_comm_get_leaders_comm(comm)==MPI_COMM_NULL)
 397       smpi_comm_set_leaders_comm(comm, leader_comm);
 398     smpi_process_set_comm_intra(comm_intra);
 399   }
 400
 401   int is_uniform = 1;
 402
 403   // Are the nodes uniform ? = same number of process/node
 404   int my_local_size=smpi_comm_size(comm_intra);
 405   if(smpi_comm_rank(comm_intra)==0) {
 406     int* non_uniform_map = xbt_malloc(sizeof(int)*leader_group_size);
 407     smpi_coll_tuned_allgather_mpich(&my_local_size, 1, MPI_INT,
 408         non_uniform_map, 1, MPI_INT, leader_comm);
 409     for(i=0; i < leader_group_size; i++) {
 410       if(non_uniform_map[0] != non_uniform_map[i]) {
 411         is_uniform = 0;
 412         break;
 413       }
 414     }
 415     if(!is_uniform && smpi_comm_is_uniform(comm)){
 416         comm->non_uniform_map= non_uniform_map;
 417     }else{
 418         xbt_free(non_uniform_map);
 419     }
 420     comm->is_uniform=is_uniform;
 421   }
 422   smpi_coll_tuned_bcast_mpich(&(comm->is_uniform),1, MPI_INT, 0, comm_intra );
 423
 424   if(smpi_privatize_global_variables){ //we need to switch here, as the called function may silently touch global variables
 425      switch_data_segment(smpi_process_index());
 426    }
 427   // Are the ranks blocked ? = allocated contiguously on the SMP nodes
 428   int is_blocked=1;
 429   int prev=smpi_group_rank(smpi_comm_group(comm), smpi_group_index(smpi_comm_group(comm_intra), 0));
 430     for (i=1; i<my_local_size; i++){
 431       int this=smpi_group_rank(smpi_comm_group(comm),smpi_group_index(smpi_comm_group(comm_intra), i));
 432       if(this!=prev+1){
 433         is_blocked=0;
 434         break;
 435       }
 436       prev = this;
 437   }
 438
 439   int global_blocked;
 440   smpi_mpi_allreduce(&is_blocked, &(global_blocked), 1,
 441             MPI_INT, MPI_LAND, comm);
 442
 443   if(comm==MPI_COMM_WORLD){
 444     if(smpi_comm_rank(comm)==0){
 445         comm->is_blocked=global_blocked;
 446     }
 447   }else{
 448     comm->is_blocked=global_blocked;
 449   }
 450   xbt_free(leader_list);
 451
 452   if(replaying==1)
 453     is_replay_active = 1;
 454 }
 455