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-2019. The SimGrid Team. All rights reserved.          */
   4
   5 /* This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the license (GNU LGPL) which comes with this package. */
   7
   8 #include "colls_private.hpp"
   9 #include "src/smpi/include/smpi_actor.hpp"
  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   if (size <= 0)
 216     return MPI_ERR_COMM;
 217   //non commutative case, use a working algo from openmpi
 218   if (op != MPI_OP_NULL && not op->is_commutative()) {
 219     return Coll_reduce_ompi_basic_linear::reduce(sendtmpbuf, recvbuf, count, datatype, op, root, comm);
 220   }
 221
 222   if( sendbuf == MPI_IN_PLACE ) {
 223     sendtmpbuf = static_cast<char *>(smpi_get_tmp_sendbuffer(count*datatype->get_extent()));
 224     Datatype::copy(recvbuf, count, datatype,sendtmpbuf, count, datatype);
 225   }
 226
 227   if(rank != root) {
 228     // Send buffer to root
 229     Request::send(sendtmpbuf, count, datatype, root, system_tag, comm);
 230   } else {
 231     datatype->extent(&lb, &dataext);
 232     // Local copy from root
 233     if (sendtmpbuf != nullptr && recvbuf != nullptr)
 234       Datatype::copy(sendtmpbuf, count, datatype, recvbuf, count, datatype);
 235     // Receive buffers from senders
 236     MPI_Request *requests = xbt_new(MPI_Request, size - 1);
 237     void **tmpbufs = xbt_new(void *, size - 1);
 238     int index = 0;
 239     for (int src = 0; src < size; src++) {
 240       if (src != root) {
 241         if (not smpi_process()->replaying())
 242           tmpbufs[index] = xbt_malloc(count * dataext);
 243          else
 244            tmpbufs[index] = smpi_get_tmp_sendbuffer(count * dataext);
 245         requests[index] =
 246           Request::irecv_init(tmpbufs[index], count, datatype, src, system_tag, comm);
 247         index++;
 248       }
 249     }
 250     // Wait for completion of irecv's.
 251     Request::startall(size - 1, requests);
 252     for (int src = 0; src < size - 1; src++) {
 253       index = Request::waitany(size - 1, requests, MPI_STATUS_IGNORE);
 254       XBT_DEBUG("finished waiting any request with index %d", index);
 255       if(index == MPI_UNDEFINED) {
 256         break;
 257       }else{
 258         Request::unref(&requests[index]);
 259       }
 260       if(op) /* op can be MPI_OP_NULL that does nothing */
 261         if(op!=MPI_OP_NULL) op->apply( tmpbufs[index], recvbuf, &count, datatype);
 262     }
 263       for(index = 0; index < size - 1; index++) {
 264         smpi_free_tmp_buffer(tmpbufs[index]);
 265       }
 266     xbt_free(tmpbufs);
 267     xbt_free(requests);
 268
 269   }
 270   if( sendbuf == MPI_IN_PLACE ) {
 271     smpi_free_tmp_buffer(sendtmpbuf);
 272   }
 273   return MPI_SUCCESS;
 274 }
 275
 276 int Coll_allreduce_default::allreduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
 277 {
 278   int ret;
 279   ret = Coll_reduce_default::reduce(sendbuf, recvbuf, count, datatype, op, 0, comm);
 280   if(ret==MPI_SUCCESS)
 281     ret = Coll_bcast_default::bcast(recvbuf, count, datatype, 0, comm);
 282   return ret;
 283 }
 284
 285 int Coll_alltoall_default::alltoall( void *sbuf, int scount, MPI_Datatype sdtype, void* rbuf, int rcount, MPI_Datatype rdtype, MPI_Comm comm)
 286 {
 287   return Coll_alltoall_ompi::alltoall(sbuf, scount, sdtype, rbuf, rcount, rdtype, comm);
 288 }
 289
 290
 291
 292 int Coll_alltoallv_default::alltoallv(void *sendbuf, int *sendcounts, int *senddisps, MPI_Datatype sendtype,
 293                               void *recvbuf, int *recvcounts, int *recvdisps, MPI_Datatype recvtype, MPI_Comm comm)
 294 {
 295   int system_tag = 889;
 296   int i;
 297   int count;
 298   MPI_Aint lb = 0;
 299   MPI_Aint sendext = 0;
 300   MPI_Aint recvext = 0;
 301   MPI_Request *requests;
 302
 303   /* Initialize. */
 304   int rank = comm->rank();
 305   int size = comm->size();
 306   XBT_DEBUG("<%d> algorithm basic_alltoallv() called.", rank);
 307   sendtype->extent(&lb, &sendext);
 308   recvtype->extent(&lb, &recvext);
 309   /* Local copy from self */
 310   int err = Datatype::copy(static_cast<char *>(sendbuf) + senddisps[rank] * sendext, sendcounts[rank], sendtype,
 311                                static_cast<char *>(recvbuf) + recvdisps[rank] * recvext, recvcounts[rank], recvtype);
 312   if (err == MPI_SUCCESS && size > 1) {
 313     /* Initiate all send/recv to/from others. */
 314     requests = xbt_new(MPI_Request, 2 * (size - 1));
 315     count = 0;
 316     /* Create all receives that will be posted first */
 317     for (i = 0; i < size; ++i) {
 318       if (i != rank && recvcounts[i] != 0) {
 319         requests[count] = Request::irecv_init(static_cast<char *>(recvbuf) + recvdisps[i] * recvext,
 320                                           recvcounts[i], recvtype, i, system_tag, comm);
 321         count++;
 322       }else{
 323         XBT_DEBUG("<%d> skip request creation [src = %d, recvcounts[src] = %d]", rank, i, recvcounts[i]);
 324       }
 325     }
 326     /* Now create all sends  */
 327     for (i = 0; i < size; ++i) {
 328       if (i != rank && sendcounts[i] != 0) {
 329       requests[count] = Request::isend_init(static_cast<char *>(sendbuf) + senddisps[i] * sendext,
 330                                         sendcounts[i], sendtype, i, system_tag, comm);
 331       count++;
 332       }else{
 333         XBT_DEBUG("<%d> skip request creation [dst = %d, sendcounts[dst] = %d]", rank, i, sendcounts[i]);
 334       }
 335     }
 336     /* Wait for them all. */
 337     Request::startall(count, requests);
 338     XBT_DEBUG("<%d> wait for %d requests", rank, count);
 339     Request::waitall(count, requests, MPI_STATUS_IGNORE);
 340     for(i = 0; i < count; i++) {
 341       if(requests[i]!=MPI_REQUEST_NULL)
 342         Request::unref(&requests[i]);
 343     }
 344     xbt_free(requests);
 345   }
 346   return err;
 347 }
 348
 349 }
 350 }
 351