src/smpi/colls/smpi_default_selector.cpp

   1 /* selector with default/naive Simgrid algorithms. These should not be trusted for performance evaluations */
   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 namespace simgrid{
  12 namespace smpi{
  13
  14 int Coll_bcast_default::bcast(void *buf, int count, MPI_Datatype datatype, int root, MPI_Comm comm)
  15 {
  16   return Coll_bcast_binomial_tree::bcast(buf, count, datatype, root, comm);
  17 }
  18
  19 int Coll_barrier_default::barrier(MPI_Comm comm)
  20 {
  21   return Coll_barrier_ompi_basic_linear::barrier(comm);
  22 }
  23
  24
  25 int Coll_gather_default::gather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
  26                      void *recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm)
  27 {
  28   int system_tag = COLL_TAG_GATHER;
  29   MPI_Aint lb = 0;
  30   MPI_Aint recvext = 0;
  31
  32   int rank = comm->rank();
  33   int size = comm->size();
  34   if(rank != root) {
  35     // Send buffer to root
  36     Request::send(sendbuf, sendcount, sendtype, root, system_tag, comm);
  37   } else {
  38     recvtype->extent(&lb, &recvext);
  39     // Local copy from root
  40     Datatype::copy(sendbuf, sendcount, sendtype, static_cast<char*>(recvbuf) + root * recvcount * recvext,
  41                        recvcount, recvtype);
  42     // Receive buffers from senders
  43     MPI_Request *requests = xbt_new(MPI_Request, size - 1);
  44     int index = 0;
  45     for (int src = 0; src < size; src++) {
  46       if(src != root) {
  47         requests[index] = Request::irecv_init(static_cast<char*>(recvbuf) + src * recvcount * recvext, recvcount, recvtype,
  48                                           src, system_tag, comm);
  49         index++;
  50       }
  51     }
  52     // Wait for completion of irecv's.
  53     Request::startall(size - 1, requests);
  54     Request::waitall(size - 1, requests, MPI_STATUS_IGNORE);
  55     for (int src = 0; src < size-1; src++) {
  56       Request::unref(&requests[src]);
  57     }
  58     xbt_free(requests);
  59   }
  60   return MPI_SUCCESS;
  61 }
  62
  63 int Coll_reduce_scatter_default::reduce_scatter(void *sendbuf, void *recvbuf, int *recvcounts, MPI_Datatype datatype, MPI_Op op,
  64                              MPI_Comm comm)
  65 {
  66   int rank = comm->rank();
  67
  68   /* arbitrarily choose root as rank 0 */
  69   int size = comm->size();
  70   int count = 0;
  71   int *displs = xbt_new(int, size);
  72   for (int i = 0; i < size; i++) {
  73     displs[i] = count;
  74     count += recvcounts[i];
  75   }
  76   void *tmpbuf = static_cast<void*>(smpi_get_tmp_sendbuffer(count*datatype->get_extent()));
  77   int ret = MPI_SUCCESS;
  78
  79   ret = Coll_reduce_default::reduce(sendbuf, tmpbuf, count, datatype, op, 0, comm);
  80   if(ret==MPI_SUCCESS)
  81     ret = Colls::scatterv(tmpbuf, recvcounts, displs, datatype, recvbuf, recvcounts[rank], datatype, 0, comm);
  82   xbt_free(displs);
  83   smpi_free_tmp_buffer(tmpbuf);
  84   return ret;
  85 }
  86
  87
  88 int Coll_allgather_default::allgather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
  89                         void *recvbuf,int recvcount, MPI_Datatype recvtype, MPI_Comm comm)
  90 {
  91   int system_tag = COLL_TAG_ALLGATHER;
  92   MPI_Aint lb = 0;
  93   MPI_Aint recvext = 0;
  94   MPI_Request *requests;
  95
  96   int rank = comm->rank();
  97   int size = comm->size();
  98   // FIXME: check for errors
  99   recvtype->extent(&lb, &recvext);
 100   // Local copy from self
 101   Datatype::copy(sendbuf, sendcount, sendtype, static_cast<char *>(recvbuf) + rank * recvcount * recvext, recvcount,
 102                      recvtype);
 103   // Send/Recv buffers to/from others;
 104   requests = xbt_new(MPI_Request, 2 * (size - 1));
 105   int index = 0;
 106   for (int other = 0; other < size; other++) {
 107     if(other != rank) {
 108       requests[index] = Request::isend_init(sendbuf, sendcount, sendtype, other, system_tag,comm);
 109       index++;
 110       requests[index] = Request::irecv_init(static_cast<char *>(recvbuf) + other * recvcount * recvext, recvcount, recvtype,
 111                                         other, system_tag, comm);
 112       index++;
 113     }
 114   }
 115   // Wait for completion of all comms.
 116   Request::startall(2 * (size - 1), requests);
 117   Request::waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
 118   for (int other = 0; other < 2*(size-1); other++) {
 119     Request::unref(&requests[other]);
 120   }
 121   xbt_free(requests);
 122   return MPI_SUCCESS;
 123 }
 124
 125 int Coll_allgatherv_default::allgatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype, void *recvbuf,
 126                          int *recvcounts, int *displs, MPI_Datatype recvtype, MPI_Comm comm)
 127 {
 128   int system_tag = COLL_TAG_ALLGATHERV;
 129   MPI_Aint lb = 0;
 130   MPI_Aint recvext = 0;
 131
 132   int rank = comm->rank();
 133   int size = comm->size();
 134   recvtype->extent(&lb, &recvext);
 135   // Local copy from self
 136   Datatype::copy(sendbuf, sendcount, sendtype,
 137                      static_cast<char *>(recvbuf) + displs[rank] * recvext,recvcounts[rank], recvtype);
 138   // Send buffers to others;
 139   MPI_Request *requests = xbt_new(MPI_Request, 2 * (size - 1));
 140   int index = 0;
 141   for (int other = 0; other < size; other++) {
 142     if(other != rank) {
 143       requests[index] =
 144         Request::isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm);
 145       index++;
 146       requests[index] = Request::irecv_init(static_cast<char *>(recvbuf) + displs[other] * recvext, recvcounts[other],
 147                           recvtype, other, system_tag, comm);
 148       index++;
 149     }
 150   }
 151   // Wait for completion of all comms.
 152   Request::startall(2 * (size - 1), requests);
 153   Request::waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
 154   for (int other = 0; other < 2*(size-1); other++) {
 155     Request::unref(&requests[other]);
 156   }
 157   xbt_free(requests);
 158   return MPI_SUCCESS;
 159 }
 160
 161 int Coll_scatter_default::scatter(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 162                       void *recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm)
 163 {
 164   int system_tag = COLL_TAG_SCATTER;
 165   MPI_Aint lb = 0;
 166   MPI_Aint sendext = 0;
 167   MPI_Request *requests;
 168
 169   int rank = comm->rank();
 170   int size = comm->size();
 171   if(rank != root) {
 172     // Recv buffer from root
 173     Request::recv(recvbuf, recvcount, recvtype, root, system_tag, comm, MPI_STATUS_IGNORE);
 174   } else {
 175     sendtype->extent(&lb, &sendext);
 176     // Local copy from root
 177     if(recvbuf!=MPI_IN_PLACE){
 178         Datatype::copy(static_cast<char *>(sendbuf) + root * sendcount * sendext,
 179                            sendcount, sendtype, recvbuf, recvcount, recvtype);
 180     }
 181     // Send buffers to receivers
 182     requests = xbt_new(MPI_Request, size - 1);
 183     int index = 0;
 184     for(int dst = 0; dst < size; dst++) {
 185       if(dst != root) {
 186         requests[index] = Request::isend_init(static_cast<char *>(sendbuf) + dst * sendcount * sendext, sendcount, sendtype,
 187                                           dst, system_tag, comm);
 188         index++;
 189       }
 190     }
 191     // Wait for completion of isend's.
 192     Request::startall(size - 1, requests);
 193     Request::waitall(size - 1, requests, MPI_STATUS_IGNORE);
 194     for (int dst = 0; dst < size-1; dst++) {
 195       Request::unref(&requests[dst]);
 196     }
 197     xbt_free(requests);
 198   }
 199   return MPI_SUCCESS;
 200 }
 201
 202
 203
 204 int Coll_reduce_default::reduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype, MPI_Op op, int root,
 205                      MPI_Comm comm)
 206 {
 207   int system_tag = COLL_TAG_REDUCE;
 208   MPI_Aint lb = 0;
 209   MPI_Aint dataext = 0;
 210
 211   char* sendtmpbuf = static_cast<char *>(sendbuf);
 212
 213   int rank = comm->rank();
 214   int size = comm->size();
 215   //non commutative case, use a working algo from openmpi
 216   if(op != MPI_OP_NULL && !op->is_commutative()){
 217     return Coll_reduce_ompi_basic_linear::reduce(sendtmpbuf, recvbuf, count, datatype, op, root, comm);
 218   }
 219
 220   if( sendbuf == MPI_IN_PLACE ) {
 221     sendtmpbuf = static_cast<char *>(smpi_get_tmp_sendbuffer(count*datatype->get_extent()));
 222     Datatype::copy(recvbuf, count, datatype,sendtmpbuf, count, datatype);
 223   }
 224
 225   if(rank != root) {
 226     // Send buffer to root
 227     Request::send(sendtmpbuf, count, datatype, root, system_tag, comm);
 228   } else {
 229     datatype->extent(&lb, &dataext);
 230     // Local copy from root
 231     if (sendtmpbuf != nullptr && recvbuf != nullptr)
 232       Datatype::copy(sendtmpbuf, count, datatype, recvbuf, count, datatype);
 233     // Receive buffers from senders
 234     MPI_Request *requests = xbt_new(MPI_Request, size - 1);
 235     void **tmpbufs = xbt_new(void *, size - 1);
 236     int index = 0;
 237     for (int src = 0; src < size; src++) {
 238       if (src != root) {
 239          if (!smpi_process()->replaying())
 240           tmpbufs[index] = xbt_malloc(count * dataext);
 241          else
 242            tmpbufs[index] = smpi_get_tmp_sendbuffer(count * dataext);
 243         requests[index] =
 244           Request::irecv_init(tmpbufs[index], count, datatype, src, system_tag, comm);
 245         index++;
 246       }
 247     }
 248     // Wait for completion of irecv's.
 249     Request::startall(size - 1, requests);
 250     for (int src = 0; src < size - 1; src++) {
 251       index = Request::waitany(size - 1, requests, MPI_STATUS_IGNORE);
 252       XBT_DEBUG("finished waiting any request with index %d", index);
 253       if(index == MPI_UNDEFINED) {
 254         break;
 255       }else{
 256         Request::unref(&requests[index]);
 257       }
 258       if(op) /* op can be MPI_OP_NULL that does nothing */
 259         if(op!=MPI_OP_NULL) op->apply( tmpbufs[index], recvbuf, &count, datatype);
 260     }
 261       for(index = 0; index < size - 1; index++) {
 262         smpi_free_tmp_buffer(tmpbufs[index]);
 263       }
 264     xbt_free(tmpbufs);
 265     xbt_free(requests);
 266
 267   }
 268   if( sendbuf == MPI_IN_PLACE ) {
 269     smpi_free_tmp_buffer(sendtmpbuf);
 270   }
 271   return MPI_SUCCESS;
 272 }
 273
 274 int Coll_allreduce_default::allreduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
 275 {
 276   int ret;
 277   ret = Colls::reduce(sendbuf, recvbuf, count, datatype, op, 0, comm);
 278   if(ret==MPI_SUCCESS)
 279     ret = Colls::bcast(recvbuf, count, datatype, 0, comm);
 280   return ret;
 281 }
 282
 283 int Coll_alltoall_default::alltoall( void *sbuf, int scount, MPI_Datatype sdtype, void* rbuf, int rcount, MPI_Datatype rdtype, MPI_Comm comm)
 284 {
 285   return Coll_alltoall_ompi::alltoall(sbuf, scount, sdtype, rbuf, rcount, rdtype, comm);
 286 }
 287
 288
 289
 290 int Coll_alltoallv_default::alltoallv(void *sendbuf, int *sendcounts, int *senddisps, MPI_Datatype sendtype,
 291                               void *recvbuf, int *recvcounts, int *recvdisps, MPI_Datatype recvtype, MPI_Comm comm)
 292 {
 293   int system_tag = 889;
 294   int i;
 295   int count;
 296   MPI_Aint lb = 0;
 297   MPI_Aint sendext = 0;
 298   MPI_Aint recvext = 0;
 299   MPI_Request *requests;
 300
 301   /* Initialize. */
 302   int rank = comm->rank();
 303   int size = comm->size();
 304   XBT_DEBUG("<%d> algorithm basic_alltoallv() called.", rank);
 305   sendtype->extent(&lb, &sendext);
 306   recvtype->extent(&lb, &recvext);
 307   /* Local copy from self */
 308   int err = Datatype::copy(static_cast<char *>(sendbuf) + senddisps[rank] * sendext, sendcounts[rank], sendtype,
 309                                static_cast<char *>(recvbuf) + recvdisps[rank] * recvext, recvcounts[rank], recvtype);
 310   if (err == MPI_SUCCESS && size > 1) {
 311     /* Initiate all send/recv to/from others. */
 312     requests = xbt_new(MPI_Request, 2 * (size - 1));
 313     count = 0;
 314     /* Create all receives that will be posted first */
 315     for (i = 0; i < size; ++i) {
 316       if (i != rank && recvcounts[i] != 0) {
 317         requests[count] = Request::irecv_init(static_cast<char *>(recvbuf) + recvdisps[i] * recvext,
 318                                           recvcounts[i], recvtype, i, system_tag, comm);
 319         count++;
 320       }else{
 321         XBT_DEBUG("<%d> skip request creation [src = %d, recvcounts[src] = %d]", rank, i, recvcounts[i]);
 322       }
 323     }
 324     /* Now create all sends  */
 325     for (i = 0; i < size; ++i) {
 326       if (i != rank && sendcounts[i] != 0) {
 327       requests[count] = Request::isend_init(static_cast<char *>(sendbuf) + senddisps[i] * sendext,
 328                                         sendcounts[i], sendtype, i, system_tag, comm);
 329       count++;
 330       }else{
 331         XBT_DEBUG("<%d> skip request creation [dst = %d, sendcounts[dst] = %d]", rank, i, sendcounts[i]);
 332       }
 333     }
 334     /* Wait for them all. */
 335     Request::startall(count, requests);
 336     XBT_DEBUG("<%d> wait for %d requests", rank, count);
 337     Request::waitall(count, requests, MPI_STATUS_IGNORE);
 338     for(i = 0; i < count; i++) {
 339       if(requests[i]!=MPI_REQUEST_NULL)
 340         Request::unref(&requests[i]);
 341     }
 342     xbt_free(requests);
 343   }
 344   return err;
 345 }
 346
 347 }
 348 }
 349