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