src/smpi/colls/smpi_mvapich2_selector.cpp

   1 /* selector for collective algorithms based on mvapich decision logic */
   2
   3 /* Copyright (c) 2009-2010, 2013-2017. The SimGrid Team.
   4  * All rights reserved.                                                     */
   5
   6 /* This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the license (GNU LGPL) which comes with this package. */
   8
   9 #include "colls_private.h"
  10
  11 #include "smpi_mvapich2_selector_stampede.h"
  12
  13 namespace simgrid{
  14 namespace smpi{
  15
  16
  17 int Coll_alltoall_mvapich2::alltoall( void *sendbuf, int sendcount,
  18     MPI_Datatype sendtype,
  19     void* recvbuf, int recvcount,
  20     MPI_Datatype recvtype,
  21     MPI_Comm comm)
  22 {
  23
  24   if(mv2_alltoall_table_ppn_conf==NULL)
  25     init_mv2_alltoall_tables_stampede();
  26
  27   int sendtype_size, recvtype_size, comm_size;
  28   char * tmp_buf = NULL;
  29   int mpi_errno=MPI_SUCCESS;
  30   int range = 0;
  31   int range_threshold = 0;
  32   int conf_index = 0;
  33   comm_size =  comm->size();
  34
  35   sendtype_size=sendtype->size();
  36   recvtype_size=recvtype->size();
  37   long nbytes = sendtype_size * sendcount;
  38
  39   /* check if safe to use partial subscription mode */
  40
  41   /* Search for the corresponding system size inside the tuning table */
  42   while ((range < (mv2_size_alltoall_tuning_table[conf_index] - 1)) &&
  43       (comm_size > mv2_alltoall_thresholds_table[conf_index][range].numproc)) {
  44       range++;
  45   }
  46   /* Search for corresponding inter-leader function */
  47   while ((range_threshold < (mv2_alltoall_thresholds_table[conf_index][range].size_table - 1))
  48       && (nbytes >
  49   mv2_alltoall_thresholds_table[conf_index][range].algo_table[range_threshold].max)
  50   && (mv2_alltoall_thresholds_table[conf_index][range].algo_table[range_threshold].max != -1)) {
  51       range_threshold++;
  52   }
  53   MV2_Alltoall_function = mv2_alltoall_thresholds_table[conf_index][range].algo_table[range_threshold]
  54                                                                                       .MV2_pt_Alltoall_function;
  55
  56   if(sendbuf != MPI_IN_PLACE) {
  57       mpi_errno = MV2_Alltoall_function(sendbuf, sendcount, sendtype,
  58           recvbuf, recvcount, recvtype,
  59           comm);
  60   } else {
  61       range_threshold = 0;
  62       if(nbytes <
  63           mv2_alltoall_thresholds_table[conf_index][range].in_place_algo_table[range_threshold].min
  64           ||nbytes > mv2_alltoall_thresholds_table[conf_index][range].in_place_algo_table[range_threshold].max
  65       ) {
  66           tmp_buf = (char *)smpi_get_tmp_sendbuffer( comm_size * recvcount * recvtype_size );
  67           mpi_errno = Datatype::copy((char *)recvbuf,
  68               comm_size*recvcount, recvtype,
  69               (char *)tmp_buf,
  70               comm_size*recvcount, recvtype);
  71
  72           mpi_errno = MV2_Alltoall_function(tmp_buf, recvcount, recvtype,
  73               recvbuf, recvcount, recvtype,
  74               comm );
  75           smpi_free_tmp_buffer(tmp_buf);
  76       } else {
  77           mpi_errno = MPIR_Alltoall_inplace_MV2(sendbuf, sendcount, sendtype,
  78               recvbuf, recvcount, recvtype,
  79               comm );
  80       }
  81   }
  82
  83
  84   return (mpi_errno);
  85 }
  86
  87 int Coll_allgather_mvapich2::allgather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
  88     void *recvbuf, int recvcount, MPI_Datatype recvtype,
  89     MPI_Comm comm)
  90 {
  91
  92   int mpi_errno = MPI_SUCCESS;
  93   long nbytes = 0, comm_size, recvtype_size;
  94   int range = 0;
  95   int partial_sub_ok = 0;
  96   int conf_index = 0;
  97   int range_threshold = 0;
  98   int is_two_level = 0;
  99   int local_size = -1;
 100   MPI_Comm shmem_comm;
 101   //MPI_Comm *shmem_commptr=NULL;
 102   /* Get the size of the communicator */
 103   comm_size = comm->size();
 104   recvtype_size=recvtype->size();
 105   nbytes = recvtype_size * recvcount;
 106
 107   if(mv2_allgather_table_ppn_conf==NULL)
 108     init_mv2_allgather_tables_stampede();
 109
 110   if(comm->get_leaders_comm()==MPI_COMM_NULL){
 111     comm->init_smp();
 112   }
 113
 114   int i;
 115   if (comm->is_uniform()){
 116     shmem_comm = comm->get_intra_comm();
 117     local_size = shmem_comm->size();
 118     i = 0;
 119     if (mv2_allgather_table_ppn_conf[0] == -1) {
 120       // Indicating user defined tuning
 121       conf_index = 0;
 122       goto conf_check_end;
 123     }
 124     do {
 125       if (local_size == mv2_allgather_table_ppn_conf[i]) {
 126         conf_index = i;
 127         partial_sub_ok = 1;
 128         break;
 129       }
 130       i++;
 131     } while(i < mv2_allgather_num_ppn_conf);
 132   }
 133   conf_check_end:
 134   if (partial_sub_ok != 1) {
 135     conf_index = 0;
 136   }
 137
 138   /* Search for the corresponding system size inside the tuning table */
 139   while ((range < (mv2_size_allgather_tuning_table[conf_index] - 1)) &&
 140       (comm_size >
 141   mv2_allgather_thresholds_table[conf_index][range].numproc)) {
 142       range++;
 143   }
 144   /* Search for corresponding inter-leader function */
 145   while ((range_threshold <
 146       (mv2_allgather_thresholds_table[conf_index][range].size_inter_table - 1))
 147       && (nbytes > mv2_allgather_thresholds_table[conf_index][range].inter_leader[range_threshold].max)
 148       && (mv2_allgather_thresholds_table[conf_index][range].inter_leader[range_threshold].max !=
 149           -1)) {
 150       range_threshold++;
 151   }
 152
 153   /* Set inter-leader pt */
 154   MV2_Allgatherction =
 155       mv2_allgather_thresholds_table[conf_index][range].inter_leader[range_threshold].
 156       MV2_pt_Allgatherction;
 157
 158   is_two_level =  mv2_allgather_thresholds_table[conf_index][range].two_level[range_threshold];
 159
 160   /* intracommunicator */
 161   if(is_two_level ==1){
 162     if(partial_sub_ok ==1){
 163       if (comm->is_blocked()){
 164       mpi_errno = MPIR_2lvl_Allgather_MV2(sendbuf, sendcount, sendtype,
 165                             recvbuf, recvcount, recvtype,
 166                             comm);
 167       }else{
 168       mpi_errno = Coll_allgather_mpich::allgather(sendbuf, sendcount, sendtype,
 169                             recvbuf, recvcount, recvtype,
 170                             comm);
 171       }
 172     } else {
 173       mpi_errno = MPIR_Allgather_RD_MV2(sendbuf, sendcount, sendtype,
 174           recvbuf, recvcount, recvtype,
 175           comm);
 176     }
 177   } else if(MV2_Allgatherction == &MPIR_Allgather_Bruck_MV2
 178       || MV2_Allgatherction == &MPIR_Allgather_RD_MV2
 179       || MV2_Allgatherction == &MPIR_Allgather_Ring_MV2) {
 180       mpi_errno = MV2_Allgatherction(sendbuf, sendcount, sendtype,
 181           recvbuf, recvcount, recvtype,
 182           comm);
 183   }else{
 184       return MPI_ERR_OTHER;
 185   }
 186
 187   return mpi_errno;
 188 }
 189
 190 int Coll_gather_mvapich2::gather(void *sendbuf,
 191     int sendcnt,
 192     MPI_Datatype sendtype,
 193     void *recvbuf,
 194     int recvcnt,
 195     MPI_Datatype recvtype,
 196     int root, MPI_Comm  comm)
 197 {
 198   if(mv2_gather_thresholds_table==NULL)
 199     init_mv2_gather_tables_stampede();
 200
 201   int mpi_errno = MPI_SUCCESS;
 202   int range = 0;
 203   int range_threshold = 0;
 204   int range_intra_threshold = 0;
 205   long nbytes = 0;
 206   int comm_size = 0;
 207   int recvtype_size, sendtype_size;
 208   int rank = -1;
 209   comm_size = comm->size();
 210   rank = comm->rank();
 211
 212   if (rank == root) {
 213       recvtype_size=recvtype->size();
 214       nbytes = recvcnt * recvtype_size;
 215   } else {
 216       sendtype_size=sendtype->size();
 217       nbytes = sendcnt * sendtype_size;
 218   }
 219
 220   /* Search for the corresponding system size inside the tuning table */
 221   while ((range < (mv2_size_gather_tuning_table - 1)) &&
 222       (comm_size > mv2_gather_thresholds_table[range].numproc)) {
 223       range++;
 224   }
 225   /* Search for corresponding inter-leader function */
 226   while ((range_threshold < (mv2_gather_thresholds_table[range].size_inter_table - 1))
 227       && (nbytes >
 228   mv2_gather_thresholds_table[range].inter_leader[range_threshold].max)
 229   && (mv2_gather_thresholds_table[range].inter_leader[range_threshold].max !=
 230       -1)) {
 231       range_threshold++;
 232   }
 233
 234   /* Search for corresponding intra node function */
 235   while ((range_intra_threshold < (mv2_gather_thresholds_table[range].size_intra_table - 1))
 236       && (nbytes >
 237   mv2_gather_thresholds_table[range].intra_node[range_intra_threshold].max)
 238   && (mv2_gather_thresholds_table[range].intra_node[range_intra_threshold].max !=
 239       -1)) {
 240       range_intra_threshold++;
 241   }
 242
 243     if (comm->is_blocked() ) {
 244         // Set intra-node function pt for gather_two_level
 245         MV2_Gather_intra_node_function =
 246                               mv2_gather_thresholds_table[range].intra_node[range_intra_threshold].
 247                               MV2_pt_Gather_function;
 248         //Set inter-leader pt
 249         MV2_Gather_inter_leader_function =
 250                               mv2_gather_thresholds_table[range].inter_leader[range_threshold].
 251                               MV2_pt_Gather_function;
 252         // We call Gather function
 253         mpi_errno =
 254             MV2_Gather_inter_leader_function(sendbuf, sendcnt, sendtype, recvbuf, recvcnt,
 255                                              recvtype, root, comm);
 256
 257     } else {
 258   // Indeed, direct (non SMP-aware)gather is MPICH one
 259   mpi_errno = Coll_gather_mpich::gather(sendbuf, sendcnt, sendtype,
 260       recvbuf, recvcnt, recvtype,
 261       root, comm);
 262   }
 263
 264   return mpi_errno;
 265 }
 266
 267 int Coll_allgatherv_mvapich2::allgatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 268     void *recvbuf, int *recvcounts, int *displs,
 269     MPI_Datatype recvtype, MPI_Comm  comm )
 270 {
 271   int mpi_errno = MPI_SUCCESS;
 272   int range = 0, comm_size, total_count, recvtype_size, i;
 273   int range_threshold = 0;
 274   long nbytes = 0;
 275
 276   if(mv2_allgatherv_thresholds_table==NULL)
 277     init_mv2_allgatherv_tables_stampede();
 278
 279   comm_size = comm->size();
 280   total_count = 0;
 281   for (i = 0; i < comm_size; i++)
 282     total_count += recvcounts[i];
 283
 284   recvtype_size=recvtype->size();
 285   nbytes = total_count * recvtype_size;
 286
 287   /* Search for the corresponding system size inside the tuning table */
 288   while ((range < (mv2_size_allgatherv_tuning_table - 1)) &&
 289       (comm_size > mv2_allgatherv_thresholds_table[range].numproc)) {
 290       range++;
 291   }
 292   /* Search for corresponding inter-leader function */
 293   while ((range_threshold < (mv2_allgatherv_thresholds_table[range].size_inter_table - 1))
 294       && (nbytes >
 295   comm_size * mv2_allgatherv_thresholds_table[range].inter_leader[range_threshold].max)
 296   && (mv2_allgatherv_thresholds_table[range].inter_leader[range_threshold].max !=
 297       -1)) {
 298       range_threshold++;
 299   }
 300   /* Set inter-leader pt */
 301   MV2_Allgatherv_function =
 302       mv2_allgatherv_thresholds_table[range].inter_leader[range_threshold].
 303       MV2_pt_Allgatherv_function;
 304
 305   if (MV2_Allgatherv_function == &MPIR_Allgatherv_Rec_Doubling_MV2)
 306     {
 307     if (not(comm_size & (comm_size - 1))) {
 308       mpi_errno =
 309           MPIR_Allgatherv_Rec_Doubling_MV2(sendbuf, sendcount, sendtype, recvbuf, recvcounts, displs, recvtype, comm);
 310         } else {
 311             mpi_errno =
 312                 MPIR_Allgatherv_Bruck_MV2(sendbuf, sendcount,
 313                     sendtype, recvbuf,
 314                     recvcounts, displs,
 315                     recvtype, comm);
 316         }
 317     } else {
 318         mpi_errno =
 319             MV2_Allgatherv_function(sendbuf, sendcount, sendtype,
 320                 recvbuf, recvcounts, displs,
 321                 recvtype, comm);
 322     }
 323
 324   return mpi_errno;
 325 }
 326
 327
 328
 329 int Coll_allreduce_mvapich2::allreduce(void *sendbuf,
 330     void *recvbuf,
 331     int count,
 332     MPI_Datatype datatype,
 333     MPI_Op op, MPI_Comm comm)
 334 {
 335
 336   int mpi_errno = MPI_SUCCESS;
 337   //int rank = 0,
 338   int comm_size = 0;
 339
 340   comm_size = comm->size();
 341   //rank = comm->rank();
 342
 343   if (count == 0) {
 344       return MPI_SUCCESS;
 345   }
 346
 347   if (mv2_allreduce_thresholds_table == NULL)
 348     init_mv2_allreduce_tables_stampede();
 349
 350   /* check if multiple threads are calling this collective function */
 351
 352   MPI_Aint sendtype_size = 0;
 353   long nbytes = 0;
 354   int range = 0, range_threshold = 0, range_threshold_intra = 0;
 355   int is_two_level = 0;
 356   int is_commutative = 0;
 357   MPI_Aint true_lb, true_extent;
 358
 359   sendtype_size=datatype->size();
 360   nbytes = count * sendtype_size;
 361
 362   datatype->extent(&true_lb, &true_extent);
 363   //MPI_Op *op_ptr;
 364   //is_commutative = op->is_commutative();
 365
 366   {
 367     /* Search for the corresponding system size inside the tuning table */
 368     while ((range < (mv2_size_allreduce_tuning_table - 1)) &&
 369         (comm_size > mv2_allreduce_thresholds_table[range].numproc)) {
 370         range++;
 371     }
 372     /* Search for corresponding inter-leader function */
 373     /* skip mcast poiters if mcast is not available */
 374     if(mv2_allreduce_thresholds_table[range].mcast_enabled != 1){
 375         while ((range_threshold < (mv2_allreduce_thresholds_table[range].size_inter_table - 1))
 376             && ((mv2_allreduce_thresholds_table[range].
 377                 inter_leader[range_threshold].MV2_pt_Allreducection
 378                 == &MPIR_Allreduce_mcst_reduce_redscat_gather_MV2) ||
 379                 (mv2_allreduce_thresholds_table[range].
 380                     inter_leader[range_threshold].MV2_pt_Allreducection
 381                     == &MPIR_Allreduce_mcst_reduce_two_level_helper_MV2)
 382             )) {
 383             range_threshold++;
 384         }
 385     }
 386     while ((range_threshold < (mv2_allreduce_thresholds_table[range].size_inter_table - 1))
 387         && (nbytes >
 388     mv2_allreduce_thresholds_table[range].inter_leader[range_threshold].max)
 389     && (mv2_allreduce_thresholds_table[range].inter_leader[range_threshold].max != -1)) {
 390         range_threshold++;
 391     }
 392     if(mv2_allreduce_thresholds_table[range].is_two_level_allreduce[range_threshold] == 1){
 393         is_two_level = 1;
 394     }
 395     /* Search for corresponding intra-node function */
 396     while ((range_threshold_intra <
 397         (mv2_allreduce_thresholds_table[range].size_intra_table - 1))
 398         && (nbytes >
 399     mv2_allreduce_thresholds_table[range].intra_node[range_threshold_intra].max)
 400     && (mv2_allreduce_thresholds_table[range].intra_node[range_threshold_intra].max !=
 401         -1)) {
 402         range_threshold_intra++;
 403     }
 404
 405     MV2_Allreducection = mv2_allreduce_thresholds_table[range].inter_leader[range_threshold]
 406                                                                                 .MV2_pt_Allreducection;
 407
 408     MV2_Allreduce_intra_function = mv2_allreduce_thresholds_table[range].intra_node[range_threshold_intra]
 409                                                                                     .MV2_pt_Allreducection;
 410
 411     /* check if mcast is ready, otherwise replace mcast with other algorithm */
 412     if((MV2_Allreducection == &MPIR_Allreduce_mcst_reduce_redscat_gather_MV2)||
 413         (MV2_Allreducection == &MPIR_Allreduce_mcst_reduce_two_level_helper_MV2)){
 414         {
 415           MV2_Allreducection = &MPIR_Allreduce_pt2pt_rd_MV2;
 416         }
 417         if(is_two_level != 1) {
 418             MV2_Allreducection = &MPIR_Allreduce_pt2pt_rd_MV2;
 419         }
 420     }
 421
 422     if(is_two_level == 1){
 423         // check if shm is ready, if not use other algorithm first
 424         if (is_commutative) {
 425           if(comm->get_leaders_comm()==MPI_COMM_NULL){
 426             comm->init_smp();
 427           }
 428           mpi_errno = MPIR_Allreduce_two_level_MV2(sendbuf, recvbuf, count,
 429                                                      datatype, op, comm);
 430                 } else {
 431         mpi_errno = MPIR_Allreduce_pt2pt_rd_MV2(sendbuf, recvbuf, count,
 432             datatype, op, comm);
 433         }
 434     } else {
 435         mpi_errno = MV2_Allreducection(sendbuf, recvbuf, count,
 436             datatype, op, comm);
 437     }
 438   }
 439
 440   //comm->ch.intra_node_done=0;
 441
 442   return (mpi_errno);
 443
 444
 445 }
 446
 447
 448 int Coll_alltoallv_mvapich2::alltoallv(void *sbuf, int *scounts, int *sdisps,
 449     MPI_Datatype sdtype,
 450     void *rbuf, int *rcounts, int *rdisps,
 451     MPI_Datatype rdtype,
 452     MPI_Comm  comm
 453 )
 454 {
 455
 456   if (sbuf == MPI_IN_PLACE) {
 457       return Coll_alltoallv_ompi_basic_linear::alltoallv(sbuf, scounts, sdisps, sdtype,
 458           rbuf, rcounts, rdisps,rdtype,
 459           comm);
 460   } else     /* For starters, just keep the original algorithm. */
 461   return Coll_alltoallv_ring::alltoallv(sbuf, scounts, sdisps, sdtype,
 462       rbuf, rcounts, rdisps,rdtype,
 463       comm);
 464 }
 465
 466
 467 int Coll_barrier_mvapich2::barrier(MPI_Comm  comm)
 468 {
 469   return Coll_barrier_mvapich2_pair::barrier(comm);
 470 }
 471
 472
 473
 474
 475 int Coll_bcast_mvapich2::bcast(void *buffer,
 476     int count,
 477     MPI_Datatype datatype,
 478     int root, MPI_Comm comm)
 479 {
 480     int mpi_errno = MPI_SUCCESS;
 481     int comm_size/*, rank*/;
 482     int two_level_bcast = 1;
 483     long nbytes = 0;
 484     int range = 0;
 485     int range_threshold = 0;
 486     int range_threshold_intra = 0;
 487     int is_homogeneous, is_contig;
 488     MPI_Aint type_size;
 489     //, position;
 490     void *tmp_buf = NULL;
 491     MPI_Comm shmem_comm;
 492     //MPID_Datatype *dtp;
 493
 494     if (count == 0)
 495         return MPI_SUCCESS;
 496     if(comm->get_leaders_comm()==MPI_COMM_NULL){
 497       comm->init_smp();
 498     }
 499     if (not mv2_bcast_thresholds_table)
 500       init_mv2_bcast_tables_stampede();
 501     comm_size = comm->size();
 502     //rank = comm->rank();
 503
 504     is_contig=1;
 505 /*    if (HANDLE_GET_KIND(datatype) == HANDLE_KIND_BUILTIN)*/
 506 /*        is_contig = 1;*/
 507 /*    else {*/
 508 /*        MPID_Datatype_get_ptr(datatype, dtp);*/
 509 /*        is_contig = dtp->is_contig;*/
 510 /*    }*/
 511
 512     is_homogeneous = 1;
 513
 514     /* MPI_Type_size() might not give the accurate size of the packed
 515      * datatype for heterogeneous systems (because of padding, encoding,
 516      * etc). On the other hand, MPI_Pack_size() can become very
 517      * expensive, depending on the implementation, especially for
 518      * heterogeneous systems. We want to use MPI_Type_size() wherever
 519      * possible, and MPI_Pack_size() in other places.
 520      */
 521     //if (is_homogeneous) {
 522         type_size=datatype->size();
 523
 524    /* } else {
 525         MPIR_Pack_size_impl(1, datatype, &type_size);
 526     }*/
 527     nbytes =  (count) * (type_size);
 528
 529     /* Search for the corresponding system size inside the tuning table */
 530     while ((range < (mv2_size_bcast_tuning_table - 1)) &&
 531            (comm_size > mv2_bcast_thresholds_table[range].numproc)) {
 532         range++;
 533     }
 534     /* Search for corresponding inter-leader function */
 535     while ((range_threshold < (mv2_bcast_thresholds_table[range].size_inter_table - 1))
 536            && (nbytes >
 537                mv2_bcast_thresholds_table[range].inter_leader[range_threshold].max)
 538            && (mv2_bcast_thresholds_table[range].inter_leader[range_threshold].max != -1)) {
 539         range_threshold++;
 540     }
 541
 542     /* Search for corresponding intra-node function */
 543     while ((range_threshold_intra <
 544             (mv2_bcast_thresholds_table[range].size_intra_table - 1))
 545            && (nbytes >
 546                mv2_bcast_thresholds_table[range].intra_node[range_threshold_intra].max)
 547            && (mv2_bcast_thresholds_table[range].intra_node[range_threshold_intra].max !=
 548                -1)) {
 549         range_threshold_intra++;
 550     }
 551
 552     MV2_Bcast_function =
 553         mv2_bcast_thresholds_table[range].inter_leader[range_threshold].
 554         MV2_pt_Bcast_function;
 555
 556     MV2_Bcast_intra_node_function =
 557         mv2_bcast_thresholds_table[range].
 558         intra_node[range_threshold_intra].MV2_pt_Bcast_function;
 559
 560 /*    if (mv2_user_bcast_intra == NULL && */
 561 /*            MV2_Bcast_intra_node_function == &MPIR_Knomial_Bcast_intra_node_MV2) {*/
 562 /*            MV2_Bcast_intra_node_function = &MPIR_Shmem_Bcast_MV2;*/
 563 /*    }*/
 564
 565     if (mv2_bcast_thresholds_table[range].inter_leader[range_threshold].
 566         zcpy_pipelined_knomial_factor != -1) {
 567         zcpy_knomial_factor =
 568             mv2_bcast_thresholds_table[range].inter_leader[range_threshold].
 569             zcpy_pipelined_knomial_factor;
 570     }
 571
 572     if (mv2_pipelined_zcpy_knomial_factor != -1) {
 573         zcpy_knomial_factor = mv2_pipelined_zcpy_knomial_factor;
 574     }
 575
 576     if(MV2_Bcast_intra_node_function == NULL) {
 577         /* if tuning table do not have any intra selection, set func pointer to
 578         ** default one for mcast intra node */
 579         MV2_Bcast_intra_node_function = &MPIR_Shmem_Bcast_MV2;
 580     }
 581
 582     /* Set value of pipeline segment size */
 583     bcast_segment_size = mv2_bcast_thresholds_table[range].bcast_segment_size;
 584
 585     /* Set value of inter node knomial factor */
 586     mv2_inter_node_knomial_factor = mv2_bcast_thresholds_table[range].inter_node_knomial_factor;
 587
 588     /* Set value of intra node knomial factor */
 589     mv2_intra_node_knomial_factor = mv2_bcast_thresholds_table[range].intra_node_knomial_factor;
 590
 591     /* Check if we will use a two level algorithm or not */
 592     two_level_bcast =
 593 #if defined(_MCST_SUPPORT_)
 594         mv2_bcast_thresholds_table[range].is_two_level_bcast[range_threshold]
 595         || comm->ch.is_mcast_ok;
 596 #else
 597         mv2_bcast_thresholds_table[range].is_two_level_bcast[range_threshold];
 598 #endif
 599      if (two_level_bcast == 1) {
 600        if (not is_contig || not is_homogeneous) {
 601          tmp_buf = (void*)smpi_get_tmp_sendbuffer(nbytes);
 602
 603          /*            position = 0;*/
 604          /*            if (rank == root) {*/
 605          /*                mpi_errno =*/
 606          /*                    MPIR_Pack_impl(buffer, count, datatype, tmp_buf, nbytes, &position);*/
 607          /*                if (mpi_errno)*/
 608          /*                    MPIU_ERR_POP(mpi_errno);*/
 609          /*            }*/
 610         }
 611 #ifdef CHANNEL_MRAIL_GEN2
 612         if ((mv2_enable_zcpy_bcast == 1) &&
 613               (&MPIR_Pipelined_Bcast_Zcpy_MV2 == MV2_Bcast_function)) {
 614           if (not is_contig || not is_homogeneous) {
 615             mpi_errno = MPIR_Pipelined_Bcast_Zcpy_MV2(tmp_buf, nbytes, MPI_BYTE, root, comm);
 616             } else {
 617                 mpi_errno = MPIR_Pipelined_Bcast_Zcpy_MV2(buffer, count, datatype,
 618                                                  root, comm);
 619             }
 620         } else
 621 #endif /* defined(CHANNEL_MRAIL_GEN2) */
 622         {
 623             shmem_comm = comm->get_intra_comm();
 624             if (not is_contig || not is_homogeneous) {
 625               mpi_errno = MPIR_Bcast_tune_inter_node_helper_MV2(tmp_buf, nbytes, MPI_BYTE, root, comm);
 626             } else {
 627                 mpi_errno =
 628                     MPIR_Bcast_tune_inter_node_helper_MV2(buffer, count, datatype, root,
 629                                                           comm);
 630             }
 631
 632             /* We are now done with the inter-node phase */
 633
 634
 635                     root = INTRA_NODE_ROOT;
 636
 637                     if (not is_contig || not is_homogeneous) {
 638                       mpi_errno = MV2_Bcast_intra_node_function(tmp_buf, nbytes, MPI_BYTE, root, shmem_comm);
 639                 } else {
 640                     mpi_errno = MV2_Bcast_intra_node_function(buffer, count,
 641                                                               datatype, root, shmem_comm);
 642
 643                 }
 644         }
 645         /*        if (not is_contig || not is_homogeneous) {*/
 646         /*            if (rank != root) {*/
 647         /*                position = 0;*/
 648         /*                mpi_errno = MPIR_Unpack_impl(tmp_buf, nbytes, &position, buffer,*/
 649         /*                                             count, datatype);*/
 650         /*            }*/
 651         /*        }*/
 652     } else {
 653         /* We use Knomial for intra node */
 654         MV2_Bcast_intra_node_function = &MPIR_Knomial_Bcast_intra_node_MV2;
 655 /*        if (mv2_enable_shmem_bcast == 0) {*/
 656             /* Fall back to non-tuned version */
 657 /*            MPIR_Bcast_intra_MV2(buffer, count, datatype, root, comm);*/
 658 /*        } else {*/
 659             mpi_errno = MV2_Bcast_function(buffer, count, datatype, root,
 660                                            comm);
 661
 662 /*        }*/
 663     }
 664
 665
 666     return mpi_errno;
 667
 668 }
 669
 670
 671
 672 int Coll_reduce_mvapich2::reduce( void *sendbuf,
 673     void *recvbuf,
 674     int count,
 675     MPI_Datatype datatype,
 676     MPI_Op op, int root, MPI_Comm comm)
 677 {
 678   if(mv2_reduce_thresholds_table == NULL)
 679     init_mv2_reduce_tables_stampede();
 680
 681   int mpi_errno = MPI_SUCCESS;
 682   int range = 0;
 683   int range_threshold = 0;
 684   int range_intra_threshold = 0;
 685   int is_commutative, pof2;
 686   int comm_size = 0;
 687   long nbytes = 0;
 688   int sendtype_size;
 689   int is_two_level = 0;
 690
 691   comm_size = comm->size();
 692   sendtype_size=datatype->size();
 693   nbytes = count * sendtype_size;
 694
 695   if (count == 0)
 696     return MPI_SUCCESS;
 697
 698   is_commutative = (op==MPI_OP_NULL || op->is_commutative());
 699
 700   /* find nearest power-of-two less than or equal to comm_size */
 701   for( pof2 = 1; pof2 <= comm_size; pof2 <<= 1 );
 702   pof2 >>=1;
 703
 704
 705   /* Search for the corresponding system size inside the tuning table */
 706   while ((range < (mv2_size_reduce_tuning_table - 1)) &&
 707       (comm_size > mv2_reduce_thresholds_table[range].numproc)) {
 708       range++;
 709   }
 710   /* Search for corresponding inter-leader function */
 711   while ((range_threshold < (mv2_reduce_thresholds_table[range].size_inter_table - 1))
 712       && (nbytes >
 713   mv2_reduce_thresholds_table[range].inter_leader[range_threshold].max)
 714   && (mv2_reduce_thresholds_table[range].inter_leader[range_threshold].max !=
 715       -1)) {
 716       range_threshold++;
 717   }
 718
 719   /* Search for corresponding intra node function */
 720   while ((range_intra_threshold < (mv2_reduce_thresholds_table[range].size_intra_table - 1))
 721       && (nbytes >
 722   mv2_reduce_thresholds_table[range].intra_node[range_intra_threshold].max)
 723   && (mv2_reduce_thresholds_table[range].intra_node[range_intra_threshold].max !=
 724       -1)) {
 725       range_intra_threshold++;
 726   }
 727
 728   /* Set intra-node function pt for reduce_two_level */
 729   MV2_Reduce_intra_function =
 730       mv2_reduce_thresholds_table[range].intra_node[range_intra_threshold].
 731       MV2_pt_Reduce_function;
 732   /* Set inter-leader pt */
 733   MV2_Reduce_function =
 734       mv2_reduce_thresholds_table[range].inter_leader[range_threshold].
 735       MV2_pt_Reduce_function;
 736
 737   if(mv2_reduce_intra_knomial_factor<0)
 738     {
 739       mv2_reduce_intra_knomial_factor = mv2_reduce_thresholds_table[range].intra_k_degree;
 740     }
 741   if(mv2_reduce_inter_knomial_factor<0)
 742     {
 743       mv2_reduce_inter_knomial_factor = mv2_reduce_thresholds_table[range].inter_k_degree;
 744     }
 745   if(mv2_reduce_thresholds_table[range].is_two_level_reduce[range_threshold] == 1){
 746       is_two_level = 1;
 747   }
 748   /* We call Reduce function */
 749   if(is_two_level == 1)
 750     {
 751        if (is_commutative == 1) {
 752          if(comm->get_leaders_comm()==MPI_COMM_NULL){
 753            comm->init_smp();
 754          }
 755          mpi_errno = MPIR_Reduce_two_level_helper_MV2(sendbuf, recvbuf, count,
 756                                            datatype, op, root, comm);
 757         } else {
 758       mpi_errno = MPIR_Reduce_binomial_MV2(sendbuf, recvbuf, count,
 759           datatype, op, root, comm);
 760       }
 761     } else if(MV2_Reduce_function == &MPIR_Reduce_inter_knomial_wrapper_MV2 ){
 762         if(is_commutative ==1)
 763           {
 764             mpi_errno = MV2_Reduce_function(sendbuf, recvbuf, count,
 765                 datatype, op, root, comm);
 766           } else {
 767               mpi_errno = MPIR_Reduce_binomial_MV2(sendbuf, recvbuf, count,
 768                   datatype, op, root, comm);
 769           }
 770     } else if(MV2_Reduce_function == &MPIR_Reduce_redscat_gather_MV2){
 771         if (/*(HANDLE_GET_KIND(op) == HANDLE_KIND_BUILTIN) &&*/ (count >= pof2))
 772           {
 773             mpi_errno = MV2_Reduce_function(sendbuf, recvbuf, count,
 774                 datatype, op, root, comm);
 775           } else {
 776               mpi_errno = MPIR_Reduce_binomial_MV2(sendbuf, recvbuf, count,
 777                   datatype, op, root, comm);
 778           }
 779     } else {
 780         mpi_errno = MV2_Reduce_function(sendbuf, recvbuf, count,
 781             datatype, op, root, comm);
 782     }
 783
 784
 785   return mpi_errno;
 786
 787 }
 788
 789
 790 int Coll_reduce_scatter_mvapich2::reduce_scatter(void *sendbuf, void *recvbuf, int *recvcnts,
 791     MPI_Datatype datatype, MPI_Op op,
 792     MPI_Comm comm)
 793 {
 794   int mpi_errno = MPI_SUCCESS;
 795   int i = 0, comm_size = comm->size(), total_count = 0, type_size =
 796       0, nbytes = 0;
 797   int range = 0;
 798   int range_threshold = 0;
 799   int is_commutative = 0;
 800   int *disps = static_cast<int*>(xbt_malloc(comm_size * sizeof (int)));
 801
 802   if(mv2_red_scat_thresholds_table==NULL)
 803     init_mv2_reduce_scatter_tables_stampede();
 804
 805   is_commutative=(op==MPI_OP_NULL || op->is_commutative());
 806   for (i = 0; i < comm_size; i++) {
 807       disps[i] = total_count;
 808       total_count += recvcnts[i];
 809   }
 810
 811   type_size=datatype->size();
 812   nbytes = total_count * type_size;
 813
 814   if (is_commutative) {
 815
 816       /* Search for the corresponding system size inside the tuning table */
 817       while ((range < (mv2_size_red_scat_tuning_table - 1)) &&
 818           (comm_size > mv2_red_scat_thresholds_table[range].numproc)) {
 819           range++;
 820       }
 821       /* Search for corresponding inter-leader function */
 822       while ((range_threshold < (mv2_red_scat_thresholds_table[range].size_inter_table - 1))
 823           && (nbytes >
 824       mv2_red_scat_thresholds_table[range].inter_leader[range_threshold].max)
 825       && (mv2_red_scat_thresholds_table[range].inter_leader[range_threshold].max !=
 826           -1)) {
 827           range_threshold++;
 828       }
 829
 830       /* Set inter-leader pt */
 831       MV2_Red_scat_function =
 832           mv2_red_scat_thresholds_table[range].inter_leader[range_threshold].
 833           MV2_pt_Red_scat_function;
 834
 835       mpi_errno = MV2_Red_scat_function(sendbuf, recvbuf,
 836           recvcnts, datatype,
 837           op, comm);
 838   } else {
 839       int is_block_regular = 1;
 840       for (i = 0; i < (comm_size - 1); ++i) {
 841           if (recvcnts[i] != recvcnts[i+1]) {
 842               is_block_regular = 0;
 843               break;
 844           }
 845       }
 846       int pof2 = 1;
 847       while (pof2 < comm_size) pof2 <<= 1;
 848       if (pof2 == comm_size && is_block_regular) {
 849           /* noncommutative, pof2 size, and block regular */
 850           mpi_errno = MPIR_Reduce_scatter_non_comm_MV2(sendbuf, recvbuf,
 851               recvcnts, datatype,
 852               op, comm);
 853       }
 854       mpi_errno =  Coll_reduce_scatter_mpich_rdb::reduce_scatter(sendbuf, recvbuf,
 855           recvcnts, datatype,
 856           op, comm);
 857   }
 858   xbt_free(disps);
 859   return mpi_errno;
 860
 861 }
 862
 863
 864
 865 int Coll_scatter_mvapich2::scatter(void *sendbuf,
 866     int sendcnt,
 867     MPI_Datatype sendtype,
 868     void *recvbuf,
 869     int recvcnt,
 870     MPI_Datatype recvtype,
 871     int root, MPI_Comm comm)
 872 {
 873   int range = 0, range_threshold = 0, range_threshold_intra = 0;
 874   int mpi_errno = MPI_SUCCESS;
 875   //   int mpi_errno_ret = MPI_SUCCESS;
 876   int rank, nbytes, comm_size;
 877   int recvtype_size, sendtype_size;
 878   int partial_sub_ok = 0;
 879   int conf_index = 0;
 880     int local_size = -1;
 881     int i;
 882      MPI_Comm shmem_comm;
 883   //    MPID_Comm *shmem_commptr=NULL;
 884   if(mv2_scatter_thresholds_table==NULL)
 885     init_mv2_scatter_tables_stampede();
 886
 887   if(comm->get_leaders_comm()==MPI_COMM_NULL){
 888     comm->init_smp();
 889   }
 890
 891   comm_size = comm->size();
 892
 893   rank = comm->rank();
 894
 895   if (rank == root) {
 896       sendtype_size=sendtype->size();
 897       nbytes = sendcnt * sendtype_size;
 898   } else {
 899       recvtype_size=recvtype->size();
 900       nbytes = recvcnt * recvtype_size;
 901   }
 902
 903     // check if safe to use partial subscription mode
 904     if (comm->is_uniform()) {
 905
 906         shmem_comm = comm->get_intra_comm();
 907         local_size = shmem_comm->size();
 908         i = 0;
 909         if (mv2_scatter_table_ppn_conf[0] == -1) {
 910             // Indicating user defined tuning
 911             conf_index = 0;
 912         }else{
 913             do {
 914                 if (local_size == mv2_scatter_table_ppn_conf[i]) {
 915                     conf_index = i;
 916                     partial_sub_ok = 1;
 917                     break;
 918                 }
 919                 i++;
 920             } while(i < mv2_scatter_num_ppn_conf);
 921         }
 922     }
 923
 924   if (partial_sub_ok != 1) {
 925       conf_index = 0;
 926   }
 927
 928   /* Search for the corresponding system size inside the tuning table */
 929   while ((range < (mv2_size_scatter_tuning_table[conf_index] - 1)) &&
 930       (comm_size > mv2_scatter_thresholds_table[conf_index][range].numproc)) {
 931       range++;
 932   }
 933   /* Search for corresponding inter-leader function */
 934   while ((range_threshold < (mv2_scatter_thresholds_table[conf_index][range].size_inter_table - 1))
 935       && (nbytes >
 936   mv2_scatter_thresholds_table[conf_index][range].inter_leader[range_threshold].max)
 937   && (mv2_scatter_thresholds_table[conf_index][range].inter_leader[range_threshold].max != -1)) {
 938       range_threshold++;
 939   }
 940
 941   /* Search for corresponding intra-node function */
 942   while ((range_threshold_intra <
 943       (mv2_scatter_thresholds_table[conf_index][range].size_intra_table - 1))
 944       && (nbytes >
 945   mv2_scatter_thresholds_table[conf_index][range].intra_node[range_threshold_intra].max)
 946   && (mv2_scatter_thresholds_table[conf_index][range].intra_node[range_threshold_intra].max !=
 947       -1)) {
 948       range_threshold_intra++;
 949   }
 950
 951   MV2_Scatter_function = mv2_scatter_thresholds_table[conf_index][range].inter_leader[range_threshold]
 952                                                                                       .MV2_pt_Scatter_function;
 953
 954   if(MV2_Scatter_function == &MPIR_Scatter_mcst_wrap_MV2) {
 955 #if defined(_MCST_SUPPORT_)
 956       if(comm->ch.is_mcast_ok == 1
 957           && mv2_use_mcast_scatter == 1
 958           && comm->ch.shmem_coll_ok == 1) {
 959           MV2_Scatter_function = &MPIR_Scatter_mcst_MV2;
 960       } else
 961 #endif /*#if defined(_MCST_SUPPORT_) */
 962         {
 963           if(mv2_scatter_thresholds_table[conf_index][range].inter_leader[range_threshold + 1].
 964               MV2_pt_Scatter_function != NULL) {
 965               MV2_Scatter_function = mv2_scatter_thresholds_table[conf_index][range].inter_leader[range_threshold + 1]
 966                                                                                                   .MV2_pt_Scatter_function;
 967           } else {
 968               /* Fallback! */
 969               MV2_Scatter_function = &MPIR_Scatter_MV2_Binomial;
 970           }
 971         }
 972   }
 973
 974   if( (MV2_Scatter_function == &MPIR_Scatter_MV2_two_level_Direct) ||
 975       (MV2_Scatter_function == &MPIR_Scatter_MV2_two_level_Binomial)) {
 976        if( comm->is_blocked()) {
 977              MV2_Scatter_intra_function = mv2_scatter_thresholds_table[conf_index][range].intra_node[range_threshold_intra]
 978                                 .MV2_pt_Scatter_function;
 979
 980              mpi_errno =
 981                    MV2_Scatter_function(sendbuf, sendcnt, sendtype,
 982                                         recvbuf, recvcnt, recvtype, root,
 983                                         comm);
 984          } else {
 985       mpi_errno = MPIR_Scatter_MV2_Binomial(sendbuf, sendcnt, sendtype,
 986           recvbuf, recvcnt, recvtype, root,
 987           comm);
 988
 989       }
 990   } else {
 991       mpi_errno = MV2_Scatter_function(sendbuf, sendcnt, sendtype,
 992           recvbuf, recvcnt, recvtype, root,
 993           comm);
 994   }
 995   return (mpi_errno);
 996 }
 997
 998 }
 999 }
1000
1001 void smpi_coll_cleanup_mvapich2()
1002 {
1003   int i = 0;
1004   if (mv2_alltoall_thresholds_table)
1005     xbt_free(mv2_alltoall_thresholds_table[i]);
1006   xbt_free(mv2_alltoall_thresholds_table);
1007   xbt_free(mv2_size_alltoall_tuning_table);
1008   xbt_free(mv2_alltoall_table_ppn_conf);
1009
1010   xbt_free(mv2_gather_thresholds_table);
1011   if (mv2_allgather_thresholds_table)
1012     xbt_free(mv2_allgather_thresholds_table[0]);
1013   xbt_free(mv2_size_allgather_tuning_table);
1014   xbt_free(mv2_allgather_table_ppn_conf);
1015   xbt_free(mv2_allgather_thresholds_table);
1016
1017   xbt_free(mv2_allgatherv_thresholds_table);
1018   xbt_free(mv2_reduce_thresholds_table);
1019   xbt_free(mv2_red_scat_thresholds_table);
1020   xbt_free(mv2_allreduce_thresholds_table);
1021   xbt_free(mv2_bcast_thresholds_table);
1022   if (mv2_scatter_thresholds_table)
1023     xbt_free(mv2_scatter_thresholds_table[0]);
1024   xbt_free(mv2_scatter_thresholds_table);
1025   xbt_free(mv2_size_scatter_tuning_table);
1026   xbt_free(mv2_scatter_table_ppn_conf);
1027 }