src/smpi/colls/smpi_nbc_impl.cpp

   1 /* Asynchronous parts of the basic collective algorithms, meant to be used both for the naive default implementation, but also for non blocking collectives */
   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
  15 int Colls::ibarrier(MPI_Comm comm, MPI_Request* request)
  16 {
  17   int i;
  18   int size = comm->size();
  19   int rank = comm->rank();
  20   MPI_Request* requests;
  21   (*request) = new Request( nullptr, 0, MPI_BYTE,
  22                          rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
  23   if (rank > 0) {
  24     requests = new MPI_Request[2];
  25     requests[0] = Request::isend (nullptr, 0, MPI_BYTE, 0,
  26                              COLL_TAG_BARRIER,
  27                              comm);
  28     requests[1] = Request::irecv (nullptr, 0, MPI_BYTE, 0,
  29                              COLL_TAG_BARRIER,
  30                              comm);
  31     (*request)->set_nbc_requests(requests, 2);
  32   }
  33   else {
  34     requests = new MPI_Request[(size-1)*2];
  35     for (i = 1; i < 2*size-1; i+=2) {
  36         requests[i-1] = Request::irecv(nullptr, 0, MPI_BYTE, MPI_ANY_SOURCE,
  37                                  COLL_TAG_BARRIER, comm
  38                                  );
  39         requests[i] = Request::isend(nullptr, 0, MPI_BYTE, (i+1)/2,
  40                                  COLL_TAG_BARRIER,
  41                                  comm
  42                                  );
  43     }
  44     (*request)->set_nbc_requests(requests, 2*(size-1));
  45   }
  46   return MPI_SUCCESS;
  47 }
  48
  49 int Colls::ibcast(void *buf, int count, MPI_Datatype datatype, int root, MPI_Comm comm, MPI_Request* request)
  50 {
  51   int i;
  52   int size = comm->size();
  53   int rank = comm->rank();
  54   MPI_Request* requests;
  55   (*request) = new Request( nullptr, 0, MPI_BYTE,
  56                          rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
  57   if (rank != root) {
  58     requests = new MPI_Request[1];
  59     requests[0] = Request::irecv (buf, count, datatype, root,
  60                              COLL_TAG_BCAST,
  61                              comm);
  62     (*request)->set_nbc_requests(requests, 1);
  63   }
  64   else {
  65     requests = new MPI_Request[size-1];
  66     int n = 0;
  67     for (i = 0; i < size; i++) {
  68       if(i!=root){
  69         requests[n] = Request::isend(buf, count, datatype, i,
  70                                  COLL_TAG_BCAST,
  71                                  comm
  72                                  );
  73         n++;
  74       }
  75     }
  76     (*request)->set_nbc_requests(requests, size-1);
  77   }
  78   return MPI_SUCCESS;
  79 }
  80
  81 int Colls::iallgather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
  82                         void *recvbuf,int recvcount, MPI_Datatype recvtype, MPI_Comm comm, MPI_Request* request)
  83 {
  84
  85   const int system_tag = COLL_TAG_ALLGATHER;
  86   MPI_Aint lb = 0;
  87   MPI_Aint recvext = 0;
  88   MPI_Request *requests;
  89
  90   int rank = comm->rank();
  91   int size = comm->size();
  92   (*request) = new Request( nullptr, 0, MPI_BYTE,
  93                          rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
  94   // FIXME: check for errors
  95   recvtype->extent(&lb, &recvext);
  96   // Local copy from self
  97   Datatype::copy(sendbuf, sendcount, sendtype, static_cast<char *>(recvbuf) + rank * recvcount * recvext, recvcount,
  98                      recvtype);
  99   // Send/Recv buffers to/from others;
 100   requests = new MPI_Request[2 * (size - 1)];
 101   int index = 0;
 102   for (int other = 0; other < size; other++) {
 103     if(other != rank) {
 104       requests[index] = Request::isend_init(sendbuf, sendcount, sendtype, other, system_tag,comm);
 105       index++;
 106       requests[index] = Request::irecv_init(static_cast<char *>(recvbuf) + other * recvcount * recvext, recvcount, recvtype,
 107                                         other, system_tag, comm);
 108       index++;
 109     }
 110   }
 111   Request::startall(2 * (size - 1), requests);
 112   (*request)->set_nbc_requests(requests, 2 * (size - 1));
 113   return MPI_SUCCESS;
 114 }
 115
 116 int Colls::iscatter(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 117                       void *recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm, MPI_Request* request)
 118 {
 119   const int system_tag = COLL_TAG_SCATTER;
 120   MPI_Aint lb = 0;
 121   MPI_Aint sendext = 0;
 122   MPI_Request *requests;
 123
 124   int rank = comm->rank();
 125   int size = comm->size();
 126   (*request) = new Request( nullptr, 0, MPI_BYTE,
 127                          rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
 128   if(rank != root) {
 129     requests = new MPI_Request[1];
 130     // Recv buffer from root
 131     requests[0] = Request::irecv(recvbuf, recvcount, recvtype, root, system_tag, comm);
 132     (*request)->set_nbc_requests(requests, 1);
 133   } else {
 134     sendtype->extent(&lb, &sendext);
 135     // Local copy from root
 136     if(recvbuf!=MPI_IN_PLACE){
 137         Datatype::copy(static_cast<char *>(sendbuf) + root * sendcount * sendext,
 138                            sendcount, sendtype, recvbuf, recvcount, recvtype);
 139     }
 140     // Send buffers to receivers
 141     requests = new MPI_Request[size - 1];
 142     int index = 0;
 143     for(int dst = 0; dst < size; dst++) {
 144       if(dst != root) {
 145         requests[index] = Request::isend_init(static_cast<char *>(sendbuf) + dst * sendcount * sendext, sendcount, sendtype,
 146                                           dst, system_tag, comm);
 147         index++;
 148       }
 149     }
 150     // Wait for completion of isend's.
 151     Request::startall(size - 1, requests);
 152     (*request)->set_nbc_requests(requests, size - 1);
 153   }
 154   return MPI_SUCCESS;
 155 }
 156
 157 int Colls::iallgatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype, void *recvbuf,
 158                          int *recvcounts, int *displs, MPI_Datatype recvtype, MPI_Comm comm, MPI_Request* request)
 159 {
 160   const int system_tag = COLL_TAG_ALLGATHERV;
 161   MPI_Aint lb = 0;
 162   MPI_Aint recvext = 0;
 163
 164   int rank = comm->rank();
 165   int size = comm->size();
 166   (*request) = new Request( nullptr, 0, MPI_BYTE,
 167                          rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
 168   recvtype->extent(&lb, &recvext);
 169   // Local copy from self
 170   Datatype::copy(sendbuf, sendcount, sendtype,
 171                      static_cast<char *>(recvbuf) + displs[rank] * recvext,recvcounts[rank], recvtype);
 172   // Send buffers to others;
 173   MPI_Request *requests = new MPI_Request[2 * (size - 1)];
 174   int index = 0;
 175   for (int other = 0; other < size; other++) {
 176     if(other != rank) {
 177       requests[index] =
 178         Request::isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm);
 179       index++;
 180       requests[index] = Request::irecv_init(static_cast<char *>(recvbuf) + displs[other] * recvext, recvcounts[other],
 181                           recvtype, other, system_tag, comm);
 182       index++;
 183     }
 184   }
 185   // Wait for completion of all comms.
 186   Request::startall(2 * (size - 1), requests);
 187   (*request)->set_nbc_requests(requests, 2 * (size - 1));
 188   return MPI_SUCCESS;
 189 }
 190
 191 int Colls::ialltoall( void *sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, MPI_Comm comm, MPI_Request* request){
 192 int system_tag = COLL_TAG_ALLTOALL;
 193   int i;
 194   int count;
 195   MPI_Aint lb = 0, sendext = 0, recvext = 0;
 196   MPI_Request *requests;
 197
 198   /* Initialize. */
 199   int rank = comm->rank();
 200   int size = comm->size();
 201   (*request) = new Request( nullptr, 0, MPI_BYTE,
 202                          rank,rank, COLL_TAG_ALLTOALL, comm, MPI_REQ_PERSISTENT);
 203   sendtype->extent(&lb, &sendext);
 204   recvtype->extent(&lb, &recvext);
 205   /* simple optimization */
 206   int err = Datatype::copy(static_cast<char *>(sendbuf) + rank * sendcount * sendext, sendcount, sendtype,
 207                                static_cast<char *>(recvbuf) + rank * recvcount * recvext, recvcount, recvtype);
 208   if (err == MPI_SUCCESS && size > 1) {
 209     /* Initiate all send/recv to/from others. */
 210     requests = new MPI_Request[2 * (size - 1)];
 211     /* Post all receives first -- a simple optimization */
 212     count = 0;
 213     for (i = (rank + 1) % size; i != rank; i = (i + 1) % size) {
 214       requests[count] = Request::irecv_init(static_cast<char *>(recvbuf) + i * recvcount * recvext, recvcount,
 215                                         recvtype, i, system_tag, comm);
 216       count++;
 217     }
 218     /* Now post all sends in reverse order
 219      *   - We would like to minimize the search time through message queue
 220      *     when messages actually arrive in the order in which they were posted.
 221      * TODO: check the previous assertion
 222      */
 223     for (i = (rank + size - 1) % size; i != rank; i = (i + size - 1) % size) {
 224       requests[count] = Request::isend_init(static_cast<char *>(sendbuf) + i * sendcount * sendext, sendcount,
 225                                         sendtype, i, system_tag, comm);
 226       count++;
 227     }
 228     /* Wait for them all. */
 229     Request::startall(count, requests);
 230     (*request)->set_nbc_requests(requests, count);
 231   }
 232   return MPI_SUCCESS;
 233 }
 234
 235 int Colls::ialltoallv(void *sendbuf, int *sendcounts, int *senddisps, MPI_Datatype sendtype,
 236                               void *recvbuf, int *recvcounts, int *recvdisps, MPI_Datatype recvtype, MPI_Comm comm, MPI_Request *request){
 237   const int system_tag = COLL_TAG_ALLTOALLV;
 238   MPI_Aint lb = 0;
 239   MPI_Aint sendext = 0;
 240   MPI_Aint recvext = 0;
 241   MPI_Request *requests;
 242
 243   /* Initialize. */
 244   int rank = comm->rank();
 245   int size = comm->size();
 246   (*request) = new Request( nullptr, 0, MPI_BYTE,
 247                          rank,rank, COLL_TAG_ALLTOALLV, comm, MPI_REQ_PERSISTENT);
 248   sendtype->extent(&lb, &sendext);
 249   recvtype->extent(&lb, &recvext);
 250   /* Local copy from self */
 251   int err = Datatype::copy(static_cast<char *>(sendbuf) + senddisps[rank] * sendext, sendcounts[rank], sendtype,
 252                                static_cast<char *>(recvbuf) + recvdisps[rank] * recvext, recvcounts[rank], recvtype);
 253   if (err == MPI_SUCCESS && size > 1) {
 254     /* Initiate all send/recv to/from others. */
 255     requests = new MPI_Request[2 * (size - 1)];
 256     int count = 0;
 257     /* Create all receives that will be posted first */
 258     for (int i = 0; i < size; ++i) {
 259       if (i != rank && recvcounts[i] != 0) {
 260         requests[count] = Request::irecv_init(static_cast<char *>(recvbuf) + recvdisps[i] * recvext,
 261                                           recvcounts[i], recvtype, i, system_tag, comm);
 262         count++;
 263       }else{
 264         XBT_DEBUG("<%d> skip request creation [src = %d, recvcounts[src] = %d]", rank, i, recvcounts[i]);
 265       }
 266     }
 267     /* Now create all sends  */
 268     for (int i = 0; i < size; ++i) {
 269       if (i != rank && sendcounts[i] != 0) {
 270       requests[count] = Request::isend_init(static_cast<char *>(sendbuf) + senddisps[i] * sendext,
 271                                         sendcounts[i], sendtype, i, system_tag, comm);
 272       count++;
 273       }else{
 274         XBT_DEBUG("<%d> skip request creation [dst = %d, sendcounts[dst] = %d]", rank, i, sendcounts[i]);
 275       }
 276     }
 277     /* Wait for them all. */
 278     Request::startall(count, requests);
 279     (*request)->set_nbc_requests(requests, count);
 280   }
 281   return err;
 282 }
 283
 284 int Colls::igather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 285                      void *recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm, MPI_Request *request)
 286 {
 287   const int system_tag = COLL_TAG_GATHER;
 288   MPI_Aint lb = 0;
 289   MPI_Aint recvext = 0;
 290   MPI_Request *requests;
 291
 292   int rank = comm->rank();
 293   int size = comm->size();
 294   (*request) = new Request( nullptr, 0, MPI_BYTE,
 295                          rank,rank, COLL_TAG_GATHER, comm, MPI_REQ_PERSISTENT);
 296   if(rank != root) {
 297     // Send buffer to root
 298     requests = new MPI_Request[1];
 299     requests[0]=Request::isend(sendbuf, sendcount, sendtype, root, system_tag, comm);
 300     (*request)->set_nbc_requests(requests, 1);
 301   } else {
 302     recvtype->extent(&lb, &recvext);
 303     // Local copy from root
 304     Datatype::copy(sendbuf, sendcount, sendtype, static_cast<char*>(recvbuf) + root * recvcount * recvext,
 305                        recvcount, recvtype);
 306     // Receive buffers from senders
 307     requests = new MPI_Request[size - 1];
 308     int index = 0;
 309     for (int src = 0; src < size; src++) {
 310       if(src != root) {
 311         requests[index] = Request::irecv_init(static_cast<char*>(recvbuf) + src * recvcount * recvext, recvcount, recvtype,
 312                                           src, system_tag, comm);
 313         index++;
 314       }
 315     }
 316     // Wait for completion of irecv's.
 317     Request::startall(size - 1, requests);
 318     (*request)->set_nbc_requests(requests, size - 1);
 319   }
 320   return MPI_SUCCESS;
 321 }
 322
 323 int Colls::igatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype, void *recvbuf, int *recvcounts, int *displs,
 324                       MPI_Datatype recvtype, int root, MPI_Comm comm, MPI_Request *request)
 325 {
 326   int system_tag = COLL_TAG_GATHERV;
 327   MPI_Aint lb = 0;
 328   MPI_Aint recvext = 0;
 329   MPI_Request *requests;
 330
 331   int rank = comm->rank();
 332   int size = comm->size();
 333   (*request) = new Request( nullptr, 0, MPI_BYTE,
 334                          rank,rank, COLL_TAG_GATHERV, comm, MPI_REQ_PERSISTENT);
 335   if (rank != root) {
 336     // Send buffer to root
 337     requests = new MPI_Request[1];
 338     requests[0]=Request::isend(sendbuf, sendcount, sendtype, root, system_tag, comm);
 339     (*request)->set_nbc_requests(requests, 1);
 340   } else {
 341     recvtype->extent(&lb, &recvext);
 342     // Local copy from root
 343     Datatype::copy(sendbuf, sendcount, sendtype, static_cast<char*>(recvbuf) + displs[root] * recvext,
 344                        recvcounts[root], recvtype);
 345     // Receive buffers from senders
 346     requests = new MPI_Request[size - 1];
 347     int index = 0;
 348     for (int src = 0; src < size; src++) {
 349       if(src != root) {
 350         requests[index] = Request::irecv_init(static_cast<char*>(recvbuf) + displs[src] * recvext,
 351                           recvcounts[src], recvtype, src, system_tag, comm);
 352         index++;
 353       }
 354     }
 355     // Wait for completion of irecv's.
 356     Request::startall(size - 1, requests);
 357     (*request)->set_nbc_requests(requests, size - 1);
 358   }
 359   return MPI_SUCCESS;
 360 }
 361 int Colls::iscatterv(void *sendbuf, int *sendcounts, int *displs, MPI_Datatype sendtype, void *recvbuf, int recvcount,
 362                        MPI_Datatype recvtype, int root, MPI_Comm comm, MPI_Request *request)
 363 {
 364   int system_tag = COLL_TAG_SCATTERV;
 365   MPI_Aint lb = 0;
 366   MPI_Aint sendext = 0;
 367   MPI_Request* requests;
 368
 369   int rank = comm->rank();
 370   int size = comm->size();
 371   (*request) = new Request( nullptr, 0, MPI_BYTE,
 372                          rank,rank, COLL_TAG_SCATTERV, comm, MPI_REQ_PERSISTENT);
 373   if(rank != root) {
 374     // Recv buffer from root
 375     requests = new MPI_Request[1];
 376     requests[0]=Request::irecv(recvbuf, recvcount, recvtype, root, system_tag, comm);
 377     (*request)->set_nbc_requests(requests, 1);
 378   } else {
 379     sendtype->extent(&lb, &sendext);
 380     // Local copy from root
 381     if(recvbuf!=MPI_IN_PLACE){
 382       Datatype::copy(static_cast<char *>(sendbuf) + displs[root] * sendext, sendcounts[root],
 383                        sendtype, recvbuf, recvcount, recvtype);
 384     }
 385     // Send buffers to receivers
 386     MPI_Request *requests = new MPI_Request[size - 1];
 387     int index = 0;
 388     for (int dst = 0; dst < size; dst++) {
 389       if (dst != root) {
 390         requests[index] = Request::isend_init(static_cast<char *>(sendbuf) + displs[dst] * sendext, sendcounts[dst],
 391                             sendtype, dst, system_tag, comm);
 392         index++;
 393       }
 394     }
 395     // Wait for completion of isend's.
 396     Request::startall(size - 1, requests);
 397     (*request)->set_nbc_requests(requests, size - 1);
 398   }
 399   return MPI_SUCCESS;
 400 }
 401 }
 402 }