Make internal collective flags negative (incorrect in MPI), to avoid confusion with...

[simgrid.git] / src / smpi / smpi_base.c

diff --git a/src/smpi/smpi_base.c b/src/smpi/smpi_base.c
index 8664524..e106bb6 100644 (file)
--- a/src/smpi/smpi_base.c
+++ b/src/smpi/smpi_base.c
@@ -12,7 +12,7 @@
 #include "simix/smx_private.h"
 #include "surf/surf.h"
 #include "simgrid/sg_config.h"
-
+#include "colls/colls.h"
 
 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi, "Logging specific to SMPI (base)");
 
@@ -25,7 +25,7 @@ static int match_recv(void* a, void* b, smx_action_t ignored) {
   xbt_assert(ref, "Cannot match recv against null reference");
   xbt_assert(req, "Cannot match recv against null request");
   if((ref->src == MPI_ANY_SOURCE || req->src == ref->src)
-    && (ref->tag == MPI_ANY_TAG || req->tag == ref->tag)){
+    && ((ref->tag == MPI_ANY_TAG && req->tag >=0) || req->tag == ref->tag)){
     //we match, we can transfer some values
     // FIXME : move this to the copy function ?
     if(ref->src == MPI_ANY_SOURCE)ref->real_src = req->src;
@@ -34,6 +34,7 @@ static int match_recv(void* a, void* b, smx_action_t ignored) {
     if(req->detached==1){
         ref->detached_sender=req; //tie the sender to the receiver, as it is detached and has to be freed in the receiver
     }
+    XBT_DEBUG("match succeeded");
     return 1;
   }else return 0;
 }
@@ -46,7 +47,7 @@ static int match_send(void* a, void* b,smx_action_t ignored) {
    xbt_assert(req, "Cannot match send against null request");
 
    if((req->src == MPI_ANY_SOURCE || req->src == ref->src)
-             && (req->tag == MPI_ANY_TAG || req->tag == ref->tag))
+             && ((req->tag == MPI_ANY_TAG && ref->tag >=0)|| req->tag == ref->tag))
    {
      if(req->src == MPI_ANY_SOURCE)req->real_src = ref->src;
      if(req->tag == MPI_ANY_TAG)req->real_tag = ref->tag;
@@ -54,7 +55,7 @@ static int match_send(void* a, void* b,smx_action_t ignored) {
      if(ref->detached==1){
          req->detached_sender=ref; //tie the sender to the receiver, as it is detached and has to be freed in the receiver
      }
-
+    XBT_DEBUG("match succeeded");
      return 1;
    } else return 0;
 }
@@ -527,7 +528,12 @@ void smpi_mpi_sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 {
   MPI_Request requests[2];
   MPI_Status stats[2];
-
+  int myid=smpi_process_index();
+  if ((smpi_group_index(smpi_comm_group(comm), dst) == myid) && (smpi_group_index(smpi_comm_group(comm), src) == myid)) {
+      smpi_datatype_copy(sendbuf, sendcount, sendtype,
+                                     recvbuf, recvcount, recvtype);
+      return;
+  }
   requests[0] =
     smpi_isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm);
   requests[1] =
@@ -583,7 +589,7 @@ static void finish_wait(MPI_Request * request, MPI_Status * status)
     if (TRACE_smpi_view_internals()) {
       if(req->flags & RECV){
         int rank = smpi_process_index();
-        int  src_traced = smpi_group_index(smpi_comm_group(req->comm), req->src == MPI_ANY_SOURCE ? req->real_src : req->src);
+        int src_traced = (req->src == MPI_ANY_SOURCE ? req->real_src : req->src);
         TRACE_smpi_recv(rank, src_traced, rank);
       }
     }
@@ -907,7 +913,7 @@ void smpi_mpi_gather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
                      void *recvbuf, int recvcount, MPI_Datatype recvtype,
                      int root, MPI_Comm comm)
 {
-  int system_tag = 666;
+  int system_tag = COLL_TAG_GATHER;
   int rank, size, src, index;
   MPI_Aint lb = 0, recvext = 0;
   MPI_Request *requests;
@@ -970,7 +976,7 @@ void smpi_mpi_gatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
                       void *recvbuf, int *recvcounts, int *displs,
                       MPI_Datatype recvtype, int root, MPI_Comm comm)
 {
-  int system_tag = 666;
+  int system_tag = COLL_TAG_GATHERV;
   int rank, size, src, index;
   MPI_Aint lb = 0, recvext = 0;
   MPI_Request *requests;
@@ -1010,7 +1016,7 @@ void smpi_mpi_allgather(void *sendbuf, int sendcount,
                         int recvcount, MPI_Datatype recvtype,
                         MPI_Comm comm)
 {
-  int system_tag = 666;
+  int system_tag = COLL_TAG_ALLGATHER;
   int rank, size, other, index;
   MPI_Aint lb = 0, recvext = 0;
   MPI_Request *requests;
@@ -1049,7 +1055,7 @@ void smpi_mpi_allgatherv(void *sendbuf, int sendcount,
                          int *recvcounts, int *displs,
                          MPI_Datatype recvtype, MPI_Comm comm)
 {
-  int system_tag = 666;
+  int system_tag = COLL_TAG_ALLGATHERV;
   int rank, size, other, index;
   MPI_Aint lb = 0, recvext = 0;
   MPI_Request *requests;
@@ -1087,7 +1093,7 @@ void smpi_mpi_scatter(void *sendbuf, int sendcount, MPI_Datatype sendtype,
                       void *recvbuf, int recvcount, MPI_Datatype recvtype,
                       int root, MPI_Comm comm)
 {
-  int system_tag = 666;
+  int system_tag = COLL_TAG_SCATTER;
   int rank, size, dst, index;
   MPI_Aint lb = 0, sendext = 0;
   MPI_Request *requests;
@@ -1126,7 +1132,7 @@ void smpi_mpi_scatterv(void *sendbuf, int *sendcounts, int *displs,
                        MPI_Datatype sendtype, void *recvbuf, int recvcount,
                        MPI_Datatype recvtype, int root, MPI_Comm comm)
 {
-  int system_tag = 666;
+  int system_tag = COLL_TAG_SCATTERV;
   int rank, size, dst, index;
   MPI_Aint lb = 0, sendext = 0;
   MPI_Request *requests;
@@ -1165,23 +1171,35 @@ void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count,
                      MPI_Datatype datatype, MPI_Op op, int root,
                      MPI_Comm comm)
 {
-  int system_tag = 666;
+  int system_tag = COLL_TAG_REDUCE;
   int rank, size, src, index;
   MPI_Aint lb = 0, dataext = 0;
   MPI_Request *requests;
   void **tmpbufs;
 
+  char* sendtmpbuf = (char*) sendbuf;
+  if( sendbuf == MPI_IN_PLACE ) {
+      sendtmpbuf = (char *)recvbuf;
+  }
+
   rank = smpi_comm_rank(comm);
   size = smpi_comm_size(comm);
+  //non commutative case, use a working algo from openmpi
+  if(!smpi_op_is_commute(op)){
+    smpi_coll_tuned_reduce_ompi_basic_linear(sendtmpbuf, recvbuf, count,
+                     datatype, op, root, comm);
+    return;
+  }
+  
   if(rank != root) {
     // Send buffer to root
-    smpi_mpi_send(sendbuf, count, datatype, root, system_tag, comm);
+    smpi_mpi_send(sendtmpbuf, count, datatype, root, system_tag, comm);
   } else {
     // FIXME: check for errors
     smpi_datatype_extent(datatype, &lb, &dataext);
     // Local copy from root
-    if (sendbuf && recvbuf)
-      smpi_datatype_copy(sendbuf, count, datatype, recvbuf, count, datatype);
+    if (sendtmpbuf && recvbuf)
+      smpi_datatype_copy(sendtmpbuf, count, datatype, recvbuf, count, datatype);
     // Receive buffers from senders
     //TODO: make a MPI_barrier here ?
     requests = xbt_new(MPI_Request, size - 1);
@@ -1227,7 +1245,7 @@ void smpi_mpi_allreduce(void *sendbuf, void *recvbuf, int count,
 void smpi_mpi_scan(void *sendbuf, void *recvbuf, int count,
                    MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
 {
-  int system_tag = 666;
+  int system_tag = 888;
   int rank, size, other, index;
   MPI_Aint lb = 0, dataext = 0;
   MPI_Request *requests;