src/smpi/smpi_comm.cpp

   1 /* Copyright (c) 2010-2015. 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 #include <limits.h>
   9
  10 #include <xbt/dict.h>
  11 #include <xbt/ex.h>
  12 #include <xbt/ex.hpp>
  13
  14 #include <simgrid/s4u/host.hpp>
  15
  16 #include "private.h"
  17 #include "smpi_mpi_dt_private.h"
  18 #include "src/simix/smx_private.h"
  19 #include "colls/colls.h"
  20
  21 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_comm, smpi, "Logging specific to SMPI (comm)");
  22
  23 xbt_dict_t smpi_comm_keyvals = nullptr;
  24 int comm_keyval_id = 0;//avoid collisions
  25
  26
  27 simgrid::smpi::Comm mpi_MPI_COMM_UNINITIALIZED;
  28 MPI_Comm MPI_COMM_UNINITIALIZED=&mpi_MPI_COMM_UNINITIALIZED;
  29
  30 /* Support for cartesian topology was added, but there are 2 other types of topology, graph et dist graph. In order to
  31  * support them, we have to add a field MPIR_Topo_type, and replace the MPI_Topology field by an union. */
  32
  33 static int smpi_compare_rankmap(const void *a, const void *b)
  34 {
  35   const int* x = static_cast<const int*>(a);
  36   const int* y = static_cast<const int*>(b);
  37
  38   if (x[1] < y[1]) {
  39     return -1;
  40   }
  41   if (x[1] == y[1]) {
  42     if (x[0] < y[0]) {
  43       return -1;
  44     }
  45     if (x[0] == y[0]) {
  46       return 0;
  47     }
  48     return 1;
  49   }
  50   return 1;
  51 }
  52
  53 namespace simgrid{
  54 namespace smpi{
  55
  56 Comm::Comm(){}
  57
  58 Comm::Comm(MPI_Group group, MPI_Topology topo) : _group(group), _topo(topo)
  59 {
  60   _refcount=1;
  61   _topoType = MPI_INVALID_TOPO;
  62   _intra_comm = MPI_COMM_NULL;
  63   _leaders_comm = MPI_COMM_NULL;
  64   _is_uniform=1;
  65   _non_uniform_map = nullptr;
  66   _leaders_map = nullptr;
  67   _is_blocked=0;
  68   _attributes=nullptr;
  69 }
  70
  71 void Comm::destroy()
  72 {
  73   if (this == MPI_COMM_UNINITIALIZED)
  74     smpi_process_comm_world()->destroy();
  75   delete _topo; // there's no use count on topos
  76   this->unuse();
  77 }
  78
  79 int Comm::dup(MPI_Comm* newcomm){
  80   if(smpi_privatize_global_variables){ //we need to switch as the called function may silently touch global variables
  81      smpi_switch_data_segment(smpi_process_index());
  82    }
  83   MPI_Group cp = new simgrid::smpi::Group(this->group());
  84   (*newcomm) = new simgrid::smpi::Comm(cp, this->topo());
  85   int ret = MPI_SUCCESS;
  86
  87   if(_attributes !=nullptr){
  88     (*newcomm)->_attributes   = xbt_dict_new_homogeneous(nullptr);
  89     xbt_dict_cursor_t cursor = nullptr;
  90     char* key;
  91     int flag;
  92     void* value_in;
  93     void* value_out;
  94     xbt_dict_foreach (_attributes, cursor, key, value_in) {
  95       smpi_comm_key_elem elem =
  96           static_cast<smpi_comm_key_elem>(xbt_dict_get_or_null_ext(smpi_comm_keyvals, key, sizeof(int)));
  97       if (elem != nullptr && elem->copy_fn != MPI_NULL_COPY_FN) {
  98         ret = elem->copy_fn(this, atoi(key), nullptr, value_in, &value_out, &flag);
  99         if (ret != MPI_SUCCESS) {
 100           (*newcomm)->destroy();
 101           *newcomm = MPI_COMM_NULL;
 102           xbt_dict_cursor_free(&cursor);
 103           return ret;
 104         }
 105         if (flag)
 106           xbt_dict_set_ext((*newcomm)->_attributes, key, sizeof(int), value_out, nullptr);
 107       }
 108       }
 109     }
 110   return ret;
 111 }
 112
 113 MPI_Group Comm::group()
 114 {
 115   if (this == MPI_COMM_UNINITIALIZED)
 116     return smpi_process_comm_world()->group();
 117   return _group;
 118 }
 119
 120 MPI_Topology Comm::topo() {
 121   return _topo;
 122 }
 123
 124 int Comm::size()
 125 {
 126   if (this == MPI_COMM_UNINITIALIZED)
 127     return smpi_process_comm_world()->size();
 128   return _group->size();
 129 }
 130
 131 int Comm::rank()
 132 {
 133   if (this == MPI_COMM_UNINITIALIZED)
 134     return smpi_process_comm_world()->rank();
 135   return _group->rank(smpi_process_index());
 136 }
 137
 138 void Comm::get_name (char* name, int* len)
 139 {
 140   if (this == MPI_COMM_UNINITIALIZED)
 141     smpi_process_comm_world()->get_name(name, len);
 142   if(this == MPI_COMM_WORLD) {
 143     strncpy(name, "WORLD",5);
 144     *len = 5;
 145   } else {
 146     *len = snprintf(name, MPI_MAX_NAME_STRING, "%p", this);
 147   }
 148 }
 149
 150 void Comm::set_leaders_comm(MPI_Comm leaders){
 151   if (this == MPI_COMM_UNINITIALIZED)
 152     smpi_process_comm_world()->set_leaders_comm(leaders);
 153   _leaders_comm=leaders;
 154 }
 155
 156 void Comm::set_intra_comm(MPI_Comm leaders){
 157   _intra_comm=leaders;
 158 }
 159
 160 int* Comm::get_non_uniform_map(){
 161   if (this == MPI_COMM_UNINITIALIZED)
 162     return smpi_process_comm_world()->get_non_uniform_map();
 163   return _non_uniform_map;
 164 }
 165
 166 int* Comm::get_leaders_map(){
 167   if (this == MPI_COMM_UNINITIALIZED)
 168     return smpi_process_comm_world()->get_leaders_map();
 169   return _leaders_map;
 170 }
 171
 172 MPI_Comm Comm::get_leaders_comm(){
 173   if (this == MPI_COMM_UNINITIALIZED)
 174     return smpi_process_comm_world()->get_leaders_comm();
 175   return _leaders_comm;
 176 }
 177
 178 MPI_Comm Comm::get_intra_comm(){
 179   if (this == MPI_COMM_UNINITIALIZED || this==MPI_COMM_WORLD)
 180     return smpi_process_get_comm_intra();
 181   else return _intra_comm;
 182 }
 183
 184 int Comm::is_uniform(){
 185   if (this == MPI_COMM_UNINITIALIZED)
 186     return smpi_process_comm_world()->is_uniform();
 187   return _is_uniform;
 188 }
 189
 190 int Comm::is_blocked(){
 191   if (this == MPI_COMM_UNINITIALIZED)
 192     return smpi_process_comm_world()->is_blocked();
 193   return _is_blocked;
 194 }
 195
 196 MPI_Comm Comm::split(int color, int key)
 197 {
 198   if (this == MPI_COMM_UNINITIALIZED)
 199     return smpi_process_comm_world()->split(color, key);
 200   int system_tag = 123;
 201   int* recvbuf;
 202
 203   MPI_Group group_root = nullptr;
 204   MPI_Group group_out  = nullptr;
 205   MPI_Group group      = this->group();
 206   int rank             = this->rank();
 207   int size             = this->size();
 208   /* Gather all colors and keys on rank 0 */
 209   int* sendbuf = xbt_new(int, 2);
 210   sendbuf[0] = color;
 211   sendbuf[1] = key;
 212   if(rank == 0) {
 213     recvbuf = xbt_new(int, 2 * size);
 214   } else {
 215     recvbuf = nullptr;
 216   }
 217   smpi_mpi_gather(sendbuf, 2, MPI_INT, recvbuf, 2, MPI_INT, 0, this);
 218   xbt_free(sendbuf);
 219   /* Do the actual job */
 220   if(rank == 0) {
 221     MPI_Group* group_snd = xbt_new(MPI_Group, size);
 222     int* rankmap         = xbt_new(int, 2 * size);
 223     for (int i = 0; i < size; i++) {
 224       if (recvbuf[2 * i] != MPI_UNDEFINED) {
 225         int count = 0;
 226         for (int j = i + 1; j < size; j++) {
 227           if(recvbuf[2 * i] == recvbuf[2 * j]) {
 228             recvbuf[2 * j] = MPI_UNDEFINED;
 229             rankmap[2 * count] = j;
 230             rankmap[2 * count + 1] = recvbuf[2 * j + 1];
 231             count++;
 232           }
 233         }
 234         /* Add self in the group */
 235         recvbuf[2 * i] = MPI_UNDEFINED;
 236         rankmap[2 * count] = i;
 237         rankmap[2 * count + 1] = recvbuf[2 * i + 1];
 238         count++;
 239         qsort(rankmap, count, 2 * sizeof(int), &smpi_compare_rankmap);
 240         group_out = new simgrid::smpi::Group(count);
 241         if (i == 0) {
 242           group_root = group_out; /* Save root's group */
 243         }
 244         for (int j = 0; j < count; j++) {
 245           int index = group->index(rankmap[2 * j]);
 246           group_out->set_mapping(index, j);
 247         }
 248         MPI_Request* requests = xbt_new(MPI_Request, count);
 249         int reqs              = 0;
 250         for (int j = 0; j < count; j++) {
 251           if(rankmap[2 * j] != 0) {
 252             group_snd[reqs]=new simgrid::smpi::Group(group_out);
 253             requests[reqs] = smpi_mpi_isend(&(group_snd[reqs]), 1, MPI_PTR, rankmap[2 * j], system_tag, this);
 254             reqs++;
 255           }
 256         }
 257         if(i != 0) {
 258           group_out->destroy();
 259         }
 260         smpi_mpi_waitall(reqs, requests, MPI_STATUS_IGNORE);
 261         xbt_free(requests);
 262       }
 263     }
 264     xbt_free(recvbuf);
 265     xbt_free(rankmap);
 266     xbt_free(group_snd);
 267     group_out = group_root; /* exit with root's group */
 268   } else {
 269     if(color != MPI_UNDEFINED) {
 270       smpi_mpi_recv(&group_out, 1, MPI_PTR, 0, system_tag, this, MPI_STATUS_IGNORE);
 271     } /* otherwise, exit with group_out == nullptr */
 272   }
 273   return group_out!=nullptr ? new simgrid::smpi::Comm(group_out, nullptr) : MPI_COMM_NULL;
 274 }
 275
 276 void Comm::use(){
 277   if (this == MPI_COMM_UNINITIALIZED)
 278     smpi_process_comm_world()->use();
 279   _group->use();
 280   _refcount++;
 281 }
 282
 283 void Comm::cleanup_attributes(){
 284   if(_attributes !=nullptr){
 285     xbt_dict_cursor_t cursor = nullptr;
 286     char* key;
 287     void* value;
 288     int flag;
 289     xbt_dict_foreach (_attributes, cursor, key, value) {
 290       smpi_comm_key_elem elem = static_cast<smpi_comm_key_elem>(xbt_dict_get_or_null(smpi_comm_keyvals, key));
 291       if (elem != nullptr && elem->delete_fn != nullptr)
 292         elem->delete_fn(this, atoi(key), value, &flag);
 293     }
 294     xbt_dict_free(&_attributes);
 295   }
 296 }
 297
 298 void Comm::cleanup_smp(){
 299   if (_intra_comm != MPI_COMM_NULL)
 300     _intra_comm->unuse();
 301   if (_leaders_comm != MPI_COMM_NULL)
 302     _leaders_comm->unuse();
 303   if (_non_uniform_map != nullptr)
 304     xbt_free(_non_uniform_map);
 305   if (_leaders_map != nullptr)
 306     xbt_free(_leaders_map);
 307 }
 308
 309 void Comm::unuse(){
 310   if (this == MPI_COMM_UNINITIALIZED)
 311     smpi_process_comm_world()->unuse();
 312   _refcount--;
 313   _group->unuse();
 314
 315   if(_refcount==0){
 316     this->cleanup_smp();
 317     this->cleanup_attributes();
 318     delete this;
 319   }
 320 }
 321
 322 static int compare_ints (const void *a, const void *b)
 323 {
 324   const int *da = static_cast<const int *>(a);
 325   const int *db = static_cast<const int *>(b);
 326
 327   return static_cast<int>(*da > *db) - static_cast<int>(*da < *db);
 328 }
 329
 330 void Comm::init_smp(){
 331   int leader = -1;
 332
 333   if (this == MPI_COMM_UNINITIALIZED)
 334     smpi_process_comm_world()->init_smp();
 335
 336   int comm_size = this->size();
 337
 338   // If we are in replay - perform an ugly hack
 339   // tell SimGrid we are not in replay for a while, because we need the buffers to be copied for the following calls
 340   bool replaying = false; //cache data to set it back again after
 341   if(smpi_process_get_replaying()){
 342    replaying=true;
 343    smpi_process_set_replaying(false);
 344   }
 345
 346   if(smpi_privatize_global_variables){ //we need to switch as the called function may silently touch global variables
 347      smpi_switch_data_segment(smpi_process_index());
 348    }
 349   //identify neighbours in comm
 350   //get the indexes of all processes sharing the same simix host
 351   xbt_swag_t process_list = SIMIX_host_self()->processes();
 352   int intra_comm_size = 0;
 353   int i =0;
 354   int min_index=INT_MAX;//the minimum index will be the leader
 355   smx_actor_t process = nullptr;
 356   xbt_swag_foreach(process, process_list) {
 357     int index = process->pid -1;
 358
 359     if(this->group()->rank(index)!=MPI_UNDEFINED){
 360         intra_comm_size++;
 361       //the process is in the comm
 362       if(index < min_index)
 363         min_index=index;
 364       i++;
 365     }
 366   }
 367   XBT_DEBUG("number of processes deployed on my node : %d", intra_comm_size);
 368   MPI_Group group_intra = new simgrid::smpi::Group(intra_comm_size);
 369   i=0;
 370   process = nullptr;
 371   xbt_swag_foreach(process, process_list) {
 372     int index = process->pid -1;
 373     if(this->group()->rank(index)!=MPI_UNDEFINED){
 374       group_intra->set_mapping(index, i);
 375       i++;
 376     }
 377   }
 378
 379   MPI_Comm comm_intra = new simgrid::smpi::Comm(group_intra, nullptr);
 380   leader=min_index;
 381
 382   int * leaders_map= static_cast<int*>(xbt_malloc0(sizeof(int)*comm_size));
 383   int * leader_list= static_cast<int*>(xbt_malloc0(sizeof(int)*comm_size));
 384   for(i=0; i<comm_size; i++){
 385       leader_list[i]=-1;
 386   }
 387
 388   smpi_coll_tuned_allgather_mpich(&leader, 1, MPI_INT , leaders_map, 1, MPI_INT, this);
 389
 390   if(smpi_privatize_global_variables){ //we need to switch as the called function may silently touch global variables
 391      smpi_switch_data_segment(smpi_process_index());
 392    }
 393
 394   if(_leaders_map==nullptr){
 395     _leaders_map= leaders_map;
 396   }else{
 397     xbt_free(leaders_map);
 398   }
 399   int j=0;
 400   int leader_group_size = 0;
 401   for(i=0; i<comm_size; i++){
 402       int already_done=0;
 403       for(j=0;j<leader_group_size; j++){
 404         if(_leaders_map[i]==leader_list[j]){
 405             already_done=1;
 406         }
 407       }
 408       if(already_done==0){
 409         leader_list[leader_group_size]=_leaders_map[i];
 410         leader_group_size++;
 411       }
 412   }
 413   qsort(leader_list, leader_group_size, sizeof(int),compare_ints);
 414
 415   MPI_Group leaders_group = new simgrid::smpi::Group(leader_group_size);
 416
 417   MPI_Comm leader_comm = MPI_COMM_NULL;
 418   if(MPI_COMM_WORLD!=MPI_COMM_UNINITIALIZED && this!=MPI_COMM_WORLD){
 419     //create leader_communicator
 420     for (i=0; i< leader_group_size;i++)
 421       leaders_group->set_mapping(leader_list[i], i);
 422     leader_comm = new simgrid::smpi::Comm(leaders_group, nullptr);
 423     this->set_leaders_comm(leader_comm);
 424     this->set_intra_comm(comm_intra);
 425
 426    //create intracommunicator
 427   }else{
 428     for (i=0; i< leader_group_size;i++)
 429       leaders_group->set_mapping(leader_list[i], i);
 430
 431     if(this->get_leaders_comm()==MPI_COMM_NULL){
 432       leader_comm = new simgrid::smpi::Comm(leaders_group, nullptr);
 433       this->set_leaders_comm(leader_comm);
 434     }else{
 435       leader_comm=this->get_leaders_comm();
 436       leaders_group->unuse();
 437     }
 438     smpi_process_set_comm_intra(comm_intra);
 439   }
 440
 441   int is_uniform = 1;
 442
 443   // Are the nodes uniform ? = same number of process/node
 444   int my_local_size=comm_intra->size();
 445   if(comm_intra->rank()==0) {
 446     int* non_uniform_map = xbt_new0(int,leader_group_size);
 447     smpi_coll_tuned_allgather_mpich(&my_local_size, 1, MPI_INT,
 448         non_uniform_map, 1, MPI_INT, leader_comm);
 449     for(i=0; i < leader_group_size; i++) {
 450       if(non_uniform_map[0] != non_uniform_map[i]) {
 451         is_uniform = 0;
 452         break;
 453       }
 454     }
 455     if(is_uniform==0 && this->is_uniform()!=0){
 456         _non_uniform_map= non_uniform_map;
 457     }else{
 458         xbt_free(non_uniform_map);
 459     }
 460     _is_uniform=is_uniform;
 461   }
 462   smpi_coll_tuned_bcast_mpich(&(_is_uniform),1, MPI_INT, 0, comm_intra );
 463
 464   if(smpi_privatize_global_variables){ //we need to switch as the called function may silently touch global variables
 465      smpi_switch_data_segment(smpi_process_index());
 466    }
 467   // Are the ranks blocked ? = allocated contiguously on the SMP nodes
 468   int is_blocked=1;
 469   int prev=this->group()->rank(comm_intra->group()->index(0));
 470     for (i=1; i<my_local_size; i++){
 471       int that=this->group()->rank(comm_intra->group()->index(i));
 472       if(that!=prev+1){
 473         is_blocked=0;
 474         break;
 475       }
 476       prev = that;
 477   }
 478
 479   int global_blocked;
 480   smpi_mpi_allreduce(&is_blocked, &(global_blocked), 1, MPI_INT, MPI_LAND, this);
 481
 482   if(MPI_COMM_WORLD==MPI_COMM_UNINITIALIZED || this==MPI_COMM_WORLD){
 483     if(this->rank()==0){
 484         _is_blocked=global_blocked;
 485     }
 486   }else{
 487     _is_blocked=global_blocked;
 488   }
 489   xbt_free(leader_list);
 490
 491   if(replaying)
 492     smpi_process_set_replaying(true);
 493 }
 494
 495 int Comm::attr_delete(int keyval){
 496   smpi_comm_key_elem elem =
 497      static_cast<smpi_comm_key_elem>(xbt_dict_get_or_null_ext(smpi_comm_keyvals, reinterpret_cast<const char*>(&keyval), sizeof(int)));
 498   if(elem==nullptr)
 499     return MPI_ERR_ARG;
 500   if(elem->delete_fn!=MPI_NULL_DELETE_FN){
 501     void* value = nullptr;
 502     int flag;
 503     if(this->attr_get(keyval, &value, &flag)==MPI_SUCCESS){
 504       int ret = elem->delete_fn(this, keyval, value, &flag);
 505       if(ret!=MPI_SUCCESS)
 506         return ret;
 507     }
 508   }
 509   if(_attributes==nullptr)
 510     return MPI_ERR_ARG;
 511
 512   xbt_dict_remove_ext(_attributes, reinterpret_cast<const char*>(&keyval), sizeof(int));
 513   return MPI_SUCCESS;
 514 }
 515
 516 int Comm::attr_get(int keyval, void* attr_value, int* flag){
 517   smpi_comm_key_elem elem =
 518     static_cast<smpi_comm_key_elem>(xbt_dict_get_or_null_ext(smpi_comm_keyvals, reinterpret_cast<const char*>(&keyval), sizeof(int)));
 519   if(elem==nullptr)
 520     return MPI_ERR_ARG;
 521   if(_attributes==nullptr){
 522     *flag=0;
 523     return MPI_SUCCESS;
 524   }
 525   try {
 526     *static_cast<void**>(attr_value) =
 527         xbt_dict_get_ext(_attributes, reinterpret_cast<const char*>(&keyval), sizeof(int));
 528     *flag=1;
 529   }
 530   catch (xbt_ex& ex) {
 531     *flag=0;
 532   }
 533   return MPI_SUCCESS;
 534 }
 535
 536 int Comm::attr_put(int keyval, void* attr_value){
 537   if(smpi_comm_keyvals==nullptr)
 538     smpi_comm_keyvals = xbt_dict_new_homogeneous(nullptr);
 539   smpi_comm_key_elem elem =
 540     static_cast<smpi_comm_key_elem>(xbt_dict_get_or_null_ext(smpi_comm_keyvals,  reinterpret_cast<const char*>(&keyval), sizeof(int)));
 541   if(elem==nullptr)
 542     return MPI_ERR_ARG;
 543   int flag;
 544   void* value = nullptr;
 545   this->attr_get(keyval, &value, &flag);
 546   if(flag!=0 && elem->delete_fn!=MPI_NULL_DELETE_FN){
 547     int ret = elem->delete_fn(this, keyval, value, &flag);
 548     if(ret!=MPI_SUCCESS)
 549       return ret;
 550   }
 551   if(_attributes==nullptr)
 552     _attributes = xbt_dict_new_homogeneous(nullptr);
 553
 554   xbt_dict_set_ext(_attributes,  reinterpret_cast<const char*>(&keyval), sizeof(int), attr_value, nullptr);
 555   return MPI_SUCCESS;
 556 }
 557
 558 }
 559 }
 560
 561 int smpi_comm_keyval_create(MPI_Comm_copy_attr_function* copy_fn, MPI_Comm_delete_attr_function* delete_fn, int* keyval,
 562                             void* extra_state){
 563   if(smpi_comm_keyvals==nullptr)
 564     smpi_comm_keyvals = xbt_dict_new_homogeneous(nullptr);
 565
 566   smpi_comm_key_elem value = static_cast<smpi_comm_key_elem>(xbt_new0(s_smpi_mpi_comm_key_elem_t,1));
 567
 568   value->copy_fn=copy_fn;
 569   value->delete_fn=delete_fn;
 570
 571   *keyval = comm_keyval_id;
 572   xbt_dict_set_ext(smpi_comm_keyvals, reinterpret_cast<const char*>(keyval), sizeof(int),static_cast<void*>(value), nullptr);
 573   comm_keyval_id++;
 574   return MPI_SUCCESS;
 575 }
 576
 577 int smpi_comm_keyval_free(int* keyval){
 578   smpi_comm_key_elem elem =
 579      static_cast<smpi_comm_key_elem>(xbt_dict_get_or_null_ext(smpi_comm_keyvals,  reinterpret_cast<const char*>(keyval), sizeof(int)));
 580   if(elem==nullptr)
 581     return MPI_ERR_ARG;
 582   xbt_dict_remove_ext(smpi_comm_keyvals,  reinterpret_cast<const char*>(keyval), sizeof(int));
 583   xbt_free(elem);
 584   return MPI_SUCCESS;
 585 }