X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/7a40637a7db0cf339c5a09ba5594a004a5304a3c..ec1872016478e2a44665e1e40a8608bcb22f08a3:/src/smpi/colls/smpi_nbc_impl.cpp

diff --git a/src/smpi/colls/smpi_nbc_impl.cpp b/src/smpi/colls/smpi_nbc_impl.cpp
index f1db5eac28..6c234e69d0 100644
--- a/src/smpi/colls/smpi_nbc_impl.cpp
+++ b/src/smpi/colls/smpi_nbc_impl.cpp
@@ -53,7 +53,7 @@ int Colls::ibcast(void *buf, int count, MPI_Datatype datatype, int root, MPI_Com
   int rank = comm->rank();
   MPI_Request* requests;
   (*request) = new Request( nullptr, 0, MPI_BYTE,
-                         rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
+                         rank,rank, COLL_TAG_BCAST, comm, MPI_REQ_PERSISTENT);
   if (rank != root) {
     requests = new MPI_Request[1];
     requests[0] = Request::irecv (buf, count, datatype, root,
@@ -90,7 +90,7 @@ int Colls::iallgather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
   int rank = comm->rank();
   int size = comm->size();
   (*request) = new Request( nullptr, 0, MPI_BYTE,
-                         rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
   // FIXME: check for errors
   recvtype->extent(&lb, &recvext);
   // Local copy from self
@@ -124,7 +124,7 @@ int Colls::iscatter(void *sendbuf, int sendcount, MPI_Datatype sendtype,
   int rank = comm->rank();
   int size = comm->size();
   (*request) = new Request( nullptr, 0, MPI_BYTE,
-                         rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
   if(rank != root) {
     requests = new MPI_Request[1];
     // Recv buffer from root
@@ -164,7 +164,7 @@ int Colls::iallgatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype, void
   int rank = comm->rank();
   int size = comm->size();
   (*request) = new Request( nullptr, 0, MPI_BYTE,
-                         rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
   recvtype->extent(&lb, &recvext);
   // Local copy from self
   Datatype::copy(sendbuf, sendcount, sendtype,
@@ -199,7 +199,7 @@ int system_tag = COLL_TAG_ALLTOALL;
   int rank = comm->rank();
   int size = comm->size();
   (*request) = new Request( nullptr, 0, MPI_BYTE,
-                         rank,rank, COLL_TAG_ALLTOALL, comm, MPI_REQ_PERSISTENT);
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
   sendtype->extent(&lb, &sendext);
   recvtype->extent(&lb, &recvext);
   /* simple optimization */
@@ -244,7 +244,7 @@ int Colls::ialltoallv(void *sendbuf, int *sendcounts, int *senddisps, MPI_Dataty
   int rank = comm->rank();
   int size = comm->size();
   (*request) = new Request( nullptr, 0, MPI_BYTE,
-                         rank,rank, COLL_TAG_ALLTOALLV, comm, MPI_REQ_PERSISTENT);
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
   sendtype->extent(&lb, &sendext);
   recvtype->extent(&lb, &recvext);
   /* Local copy from self */
@@ -256,7 +256,7 @@ int Colls::ialltoallv(void *sendbuf, int *sendcounts, int *senddisps, MPI_Dataty
     int count = 0;
     /* Create all receives that will be posted first */
     for (int i = 0; i < size; ++i) {
-      if (i != rank && recvcounts[i] != 0) {
+      if (i != rank) {
         requests[count] = Request::irecv_init(static_cast<char *>(recvbuf) + recvdisps[i] * recvext,
                                           recvcounts[i], recvtype, i, system_tag, comm);
         count++;
@@ -266,7 +266,7 @@ int Colls::ialltoallv(void *sendbuf, int *sendcounts, int *senddisps, MPI_Dataty
     }
     /* Now create all sends  */
     for (int i = 0; i < size; ++i) {
-      if (i != rank && sendcounts[i] != 0) {
+      if (i != rank) {
       requests[count] = Request::isend_init(static_cast<char *>(sendbuf) + senddisps[i] * sendext,
                                         sendcounts[i], sendtype, i, system_tag, comm);
       count++;
@@ -281,6 +281,50 @@ int Colls::ialltoallv(void *sendbuf, int *sendcounts, int *senddisps, MPI_Dataty
   return err;
 }
 
+int Colls::ialltoallw(void *sendbuf, int *sendcounts, int *senddisps, MPI_Datatype* sendtypes,
+                              void *recvbuf, int *recvcounts, int *recvdisps, MPI_Datatype* recvtypes, MPI_Comm comm, MPI_Request *request){
+  const int system_tag = COLL_TAG_ALLTOALLV;
+  MPI_Request *requests;
+
+  /* Initialize. */
+  int rank = comm->rank();
+  int size = comm->size();
+  (*request) = new Request( nullptr, 0, MPI_BYTE,
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
+  /* Local copy from self */
+  int err = (sendcounts[rank]>0 && recvcounts[rank]) ? Datatype::copy(static_cast<char *>(sendbuf) + senddisps[rank], sendcounts[rank], sendtypes[rank],
+                               static_cast<char *>(recvbuf) + recvdisps[rank], recvcounts[rank], recvtypes[rank]): MPI_SUCCESS;
+  if (err == MPI_SUCCESS && size > 1) {
+    /* Initiate all send/recv to/from others. */
+    requests = new MPI_Request[2 * (size - 1)];
+    int count = 0;
+    /* Create all receives that will be posted first */
+    for (int i = 0; i < size; ++i) {
+      if (i != rank) {
+        requests[count] = Request::irecv_init(static_cast<char *>(recvbuf) + recvdisps[i],
+                                          recvcounts[i], recvtypes[i], i, system_tag, comm);
+        count++;
+      }else{
+        XBT_DEBUG("<%d> skip request creation [src = %d, recvcounts[src] = %d]", rank, i, recvcounts[i]);
+      }
+    }
+    /* Now create all sends  */
+    for (int i = 0; i < size; ++i) {
+      if (i != rank) {
+      requests[count] = Request::isend_init(static_cast<char *>(sendbuf) + senddisps[i] ,
+                                        sendcounts[i], sendtypes[i], i, system_tag, comm);
+      count++;
+      }else{
+        XBT_DEBUG("<%d> skip request creation [dst = %d, sendcounts[dst] = %d]", rank, i, sendcounts[i]);
+      }
+    }
+    /* Wait for them all. */
+    Request::startall(count, requests);
+    (*request)->set_nbc_requests(requests, count);
+  }
+  return err;
+}
+
 int Colls::igather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
                      void *recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm, MPI_Request *request)
 {
@@ -292,7 +336,7 @@ int Colls::igather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
   int rank = comm->rank();
   int size = comm->size();
   (*request) = new Request( nullptr, 0, MPI_BYTE,
-                         rank,rank, COLL_TAG_GATHER, comm, MPI_REQ_PERSISTENT);
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
   if(rank != root) {
     // Send buffer to root
     requests = new MPI_Request[1];
@@ -331,7 +375,7 @@ int Colls::igatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype, void *r
   int rank = comm->rank();
   int size = comm->size();
   (*request) = new Request( nullptr, 0, MPI_BYTE,
-                         rank,rank, COLL_TAG_GATHERV, comm, MPI_REQ_PERSISTENT);
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
   if (rank != root) {
     // Send buffer to root
     requests = new MPI_Request[1];
@@ -369,7 +413,7 @@ int Colls::iscatterv(void *sendbuf, int *sendcounts, int *displs, MPI_Datatype s
   int rank = comm->rank();
   int size = comm->size();
   (*request) = new Request( nullptr, 0, MPI_BYTE,
-                         rank,rank, COLL_TAG_SCATTERV, comm, MPI_REQ_PERSISTENT);
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
   if(rank != root) {
     // Recv buffer from root
     requests = new MPI_Request[1];
@@ -398,5 +442,202 @@ int Colls::iscatterv(void *sendbuf, int *sendcounts, int *displs, MPI_Datatype s
   }
   return MPI_SUCCESS;
 }
+
+int Colls::ireduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype, MPI_Op op, int root,
+                     MPI_Comm comm, MPI_Request* request)
+{
+  const int system_tag = COLL_TAG_REDUCE;
+  MPI_Aint lb = 0;
+  MPI_Aint dataext = 0;
+  MPI_Request* requests;
+
+  char* sendtmpbuf = static_cast<char *>(sendbuf);
+
+  int rank = comm->rank();
+  int size = comm->size();
+
+  if (size <= 0)
+    return MPI_ERR_COMM;
+
+  if( sendbuf == MPI_IN_PLACE ) {
+    sendtmpbuf = static_cast<char *>(smpi_get_tmp_sendbuffer(count*datatype->get_extent()));
+    Datatype::copy(recvbuf, count, datatype,sendtmpbuf, count, datatype);
+  }
+
+  if(rank == root){
+    (*request) =  new Request( recvbuf, count, datatype,
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT, op);
+  }
+  else
+    (*request) = new Request( nullptr, count, datatype,
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
+
+  if(rank != root) {
+    // Send buffer to root
+    requests = new MPI_Request[1];
+    requests[0]=Request::isend(sendtmpbuf, count, datatype, root, system_tag, comm);
+    (*request)->set_nbc_requests(requests, 1);
+  } else {
+    datatype->extent(&lb, &dataext);
+    // Local copy from root
+    if (sendtmpbuf != nullptr && recvbuf != nullptr)
+      Datatype::copy(sendtmpbuf, count, datatype, recvbuf, count, datatype);
+    // Receive buffers from senders
+    MPI_Request *requests = new MPI_Request[size - 1];
+    int index = 0;
+    for (int src = 0; src < size; src++) {
+      if (src != root) {
+        requests[index] =
+          Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype, src, system_tag, comm);
+        index++;
+      }
+    }
+    // Wait for completion of irecv's.
+    Request::startall(size - 1, requests);
+    (*request)->set_nbc_requests(requests, size - 1);
+  }	
+  if( sendbuf == MPI_IN_PLACE ) {
+    smpi_free_tmp_buffer(sendtmpbuf);
+  }
+  return MPI_SUCCESS;
+}
+
+int Colls::iallreduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype,
+                      MPI_Op op, MPI_Comm comm, MPI_Request* request)
+{
+
+  const int system_tag = COLL_TAG_ALLREDUCE;
+  MPI_Aint lb = 0;
+  MPI_Aint dataext = 0;
+  MPI_Request *requests;
+
+  int rank = comm->rank();
+  int size = comm->size();
+  (*request) = new Request( recvbuf, count, datatype,
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT, op);
+  // FIXME: check for errors
+  datatype->extent(&lb, &dataext);
+  // Local copy from self
+  Datatype::copy(sendbuf, count, datatype, recvbuf, count, datatype);
+  // Send/Recv buffers to/from others;
+  requests = new MPI_Request[2 * (size - 1)];
+  int index = 0;
+  for (int other = 0; other < size; other++) {
+    if(other != rank) {
+      requests[index] = Request::isend_init(sendbuf, count, datatype, other, system_tag,comm);
+      index++;
+      requests[index] = Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype,
+                                        other, system_tag, comm);
+      index++;
+    }
+  }
+  Request::startall(2 * (size - 1), requests);
+  (*request)->set_nbc_requests(requests, 2 * (size - 1));
+  return MPI_SUCCESS;
+}
+
+int Colls::iscan(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm, MPI_Request* request)
+{
+  int system_tag = -888;
+  MPI_Aint lb      = 0;
+  MPI_Aint dataext = 0;
+
+  int rank = comm->rank();
+  int size = comm->size();
+  (*request) = new Request( recvbuf, count, datatype,
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT, op);
+  datatype->extent(&lb, &dataext);
+
+  // Local copy from self
+  Datatype::copy(sendbuf, count, datatype, recvbuf, count, datatype);
+
+  // Send/Recv buffers to/from others
+  MPI_Request *requests = new MPI_Request[size - 1];
+  void **tmpbufs = xbt_new(void *, rank);
+  int index = 0;
+  for (int other = 0; other < rank; other++) {
+    tmpbufs[index] = smpi_get_tmp_sendbuffer(count * dataext);
+    requests[index] = Request::irecv_init(tmpbufs[index], count, datatype, other, system_tag, comm);
+    index++;
+  }
+  for (int other = rank + 1; other < size; other++) {
+    requests[index] = Request::isend_init(sendbuf, count, datatype, other, system_tag, comm);
+    index++;
+  }
+  // Wait for completion of all comms.
+  Request::startall(size - 1, requests);
+  (*request)->set_nbc_requests(requests, size - 1);
+  return MPI_SUCCESS;
+}
+
+int Colls::iexscan(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm, MPI_Request* request)
+{
+  int system_tag = -888;
+  MPI_Aint lb         = 0;
+  MPI_Aint dataext    = 0;
+  int rank = comm->rank();
+  int size = comm->size();
+  (*request) = new Request( recvbuf, count, datatype,
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT, op);
+  datatype->extent(&lb, &dataext);
+  if(rank != 0)
+    memset(recvbuf, 0, count*dataext);
+
+  // Send/Recv buffers to/from others
+  MPI_Request *requests = new MPI_Request[size - 1];
+  void **tmpbufs = xbt_new(void *, rank);
+  int index = 0;
+  for (int other = 0; other < rank; other++) {
+    tmpbufs[index] = smpi_get_tmp_sendbuffer(count * dataext);
+    requests[index] = Request::irecv_init(tmpbufs[index], count, datatype, other, system_tag, comm);
+    index++;
+  }
+  for (int other = rank + 1; other < size; other++) {
+    requests[index] = Request::isend_init(sendbuf, count, datatype, other, system_tag, comm);
+    index++;
+  }
+  // Wait for completion of all comms.
+  Request::startall(size - 1, requests);
+  (*request)->set_nbc_requests(requests, size - 1);
+  return MPI_SUCCESS;
+}
+
+int Colls::ireduce_scatter(void *sendbuf, void *recvbuf, int *recvcounts, MPI_Datatype datatype, MPI_Op op,
+                             MPI_Comm comm, MPI_Request* request){
+//Version where each process performs the reduce for its own part. Alltoall pattern for comms.
+  const int system_tag = COLL_TAG_REDUCE_SCATTER;
+  MPI_Aint lb = 0;
+  MPI_Aint dataext = 0;
+  MPI_Request *requests;
+
+  int rank = comm->rank();
+  int size = comm->size();
+  int count=recvcounts[rank];
+  (*request) = new Request( recvbuf, count, datatype,
+                         rank,rank, system_tag, comm, MPI_REQ_PERSISTENT, op);
+  datatype->extent(&lb, &dataext);
+
+  // Send/Recv buffers to/from others;
+  requests = new MPI_Request[2 * (size - 1)];
+  int index = 0;
+  int recvdisp=0;
+  for (int other = 0; other < size; other++) {
+    if(other != rank) {
+      requests[index] = Request::isend_init(static_cast<char *>(sendbuf) + recvdisp * dataext, recvcounts[other], datatype, other, system_tag,comm);
+      XBT_VERB("sending with recvdisp %d", recvdisp);
+      index++;
+      requests[index] = Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype,
+                                        other, system_tag, comm);
+      index++;
+    }else{
+      Datatype::copy(static_cast<char *>(sendbuf) + recvdisp * dataext, count, datatype, recvbuf, count, datatype);
+    }
+    recvdisp+=recvcounts[other];
+  }
+  Request::startall(2 * (size - 1), requests);
+  (*request)->set_nbc_requests(requests, 2 * (size - 1));
+  return MPI_SUCCESS;
+}
+
 }
 }