Have the communicators created together share a unique ID.

[simgrid.git] / src / smpi / mpi / smpi_request.cpp

diff --git a/src/smpi/mpi/smpi_request.cpp b/src/smpi/mpi/smpi_request.cpp
index 197e030..e0113e8 100644 (file)
--- a/src/smpi/mpi/smpi_request.cpp
+++ b/src/smpi/mpi/smpi_request.cpp
@@ -36,14 +36,14 @@ extern void (*smpi_comm_copy_data_callback)(simgrid::kernel::activity::CommImpl*
 namespace simgrid{
 namespace smpi{
 
-Request::Request(void* buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, unsigned flags, MPI_Op op)
-    : buf_(buf), old_type_(datatype), src_(src), dst_(dst), tag_(tag), comm_(comm), flags_(flags), op_(op)
+Request::Request(const void* buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, unsigned flags, MPI_Op op)
+    : buf_(const_cast<void*>(buf)), old_type_(datatype), src_(src), dst_(dst), tag_(tag), comm_(comm), flags_(flags), op_(op)
 {
   void *old_buf = nullptr;
 // FIXME Handle the case of a partial shared malloc.
   if ((((flags & MPI_REQ_RECV) != 0) && ((flags & MPI_REQ_ACCUMULATE) != 0)) || (datatype->flags() & DT_FLAG_DERIVED)) {
     // This part handles the problem of non-contiguous memory
-    old_buf = buf;
+    old_buf = const_cast<void*>(buf);
     if (count==0){
       buf_ = nullptr;
     }else {
@@ -92,6 +92,7 @@ void Request::unref(MPI_Request* request)
     if((*request)->refcount_==0){
       if ((*request)->flags_ & MPI_REQ_GENERALIZED){
         ((*request)->generalized_funcs)->free_fn(((*request)->generalized_funcs)->extra_state);
+        delete (*request)->generalized_funcs;
       }else{
         Comm::unref((*request)->comm_);
         Datatype::unref((*request)->old_type_);
@@ -114,11 +115,12 @@ int Request::match_recv(void* a, void* b, simgrid::kernel::activity::CommImpl*)
 {
   MPI_Request ref = static_cast<MPI_Request>(a);
   MPI_Request req = static_cast<MPI_Request>(b);
-  XBT_DEBUG("Trying to match a recv of src %d against %d, tag %d against %d",ref->src_,req->src_, ref->tag_, req->tag_);
+  XBT_DEBUG("Trying to match a recv of src %d against %d, tag %d against %d, id %d against %d",ref->src_,req->src_, ref->tag_, req->tag_,ref->comm_->id(),req->comm_->id());
 
   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_)
+  if((ref->comm_->id()==MPI_UNDEFINED || req->comm_->id() == MPI_UNDEFINED || (ref->comm_->id()==req->comm_->id()))
+    && ((ref->src_ == MPI_ANY_SOURCE  && (ref->comm_->group()->rank(req->src_) != MPI_UNDEFINED)) || req->src_ == ref->src_)
     && ((ref->tag_ == MPI_ANY_TAG && req->tag_ >=0) || req->tag_ == ref->tag_)){
     //we match, we can transfer some values
     if(ref->src_ == MPI_ANY_SOURCE)
@@ -140,11 +142,12 @@ int Request::match_send(void* a, void* b, simgrid::kernel::activity::CommImpl*)
 {
   MPI_Request ref = static_cast<MPI_Request>(a);
   MPI_Request req = static_cast<MPI_Request>(b);
-  XBT_DEBUG("Trying to match a send of src %d against %d, tag %d against %d",ref->src_,req->src_, ref->tag_, req->tag_);
+  XBT_DEBUG("Trying to match a send of src %d against %d, tag %d against %d, id %d against %d",ref->src_,req->src_, ref->tag_, req->tag_,ref->comm_->id(),req->comm_->id());
   xbt_assert(ref, "Cannot match send against null reference");
   xbt_assert(req, "Cannot match send against null request");
 
-  if((req->src_ == MPI_ANY_SOURCE || req->src_ == ref->src_)
+  if((ref->comm_->id()==MPI_UNDEFINED || req->comm_->id() == MPI_UNDEFINED || (ref->comm_->id()==req->comm_->id()))
+      && ((req->src_ == MPI_ANY_SOURCE  && (req->comm_->group()->rank(ref->src_) != MPI_UNDEFINED)) || req->src_ == ref->src_)
       && ((req->tag_ == MPI_ANY_TAG && ref->tag_ >=0)|| req->tag_ == ref->tag_)){
     if(req->src_ == MPI_ANY_SOURCE)
       req->real_src_ = ref->src_;
@@ -170,7 +173,7 @@ void Request::print_request(const char *message)
 
 
 /* factories, to hide the internal flags from the caller */
-MPI_Request Request::send_init(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
+MPI_Request Request::send_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 {
 
   return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
@@ -178,14 +181,14 @@ MPI_Request Request::send_init(void *buf, int count, MPI_Datatype datatype, int
                      MPI_REQ_PERSISTENT | MPI_REQ_SEND | MPI_REQ_PREPARED);
 }
 
-MPI_Request Request::ssend_init(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
+MPI_Request Request::ssend_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 {
   return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
                      comm->group()->actor(dst)->get_pid(), tag, comm,
                      MPI_REQ_PERSISTENT | MPI_REQ_SSEND | MPI_REQ_SEND | MPI_REQ_PREPARED);
 }
 
-MPI_Request Request::isend_init(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
+MPI_Request Request::isend_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 {
   return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
                      comm->group()->actor(dst)->get_pid(), tag, comm,
@@ -193,7 +196,7 @@ MPI_Request Request::isend_init(void *buf, int count, MPI_Datatype datatype, int
 }
 
 
-MPI_Request Request::rma_send_init(void *buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm,
+MPI_Request Request::rma_send_init(const void *buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm,
                                MPI_Op op)
 {
   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
@@ -242,7 +245,7 @@ MPI_Request Request::irecv_init(void *buf, int count, MPI_Datatype datatype, int
                      MPI_REQ_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED);
 }
 
-MPI_Request Request::isend(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
+MPI_Request Request::isend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 {
   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
   request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
@@ -252,7 +255,7 @@ MPI_Request Request::isend(void *buf, int count, MPI_Datatype datatype, int dst,
   return request;
 }
 
-MPI_Request Request::issend(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
+MPI_Request Request::issend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 {
   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
   request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
@@ -281,7 +284,7 @@ void Request::recv(void *buf, int count, MPI_Datatype datatype, int src, int tag
   request = nullptr;
 }
 
-void Request::send(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
+void Request::send(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 {
   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
   request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
@@ -292,7 +295,7 @@ void Request::send(void *buf, int count, MPI_Datatype datatype, int dst, int tag
   request = nullptr;
 }
 
-void Request::ssend(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
+void Request::ssend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 {
   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
   request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
@@ -304,7 +307,7 @@ void Request::ssend(void *buf, int count, MPI_Datatype datatype, int dst, int ta
   request = nullptr;
 }
 
-void Request::sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,int dst, int sendtag,
+void Request::sendrecv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,int dst, int sendtag,
                        void *recvbuf, int recvcount, MPI_Datatype recvtype, int src, int recvtag,
                        MPI_Comm comm, MPI_Status * status)
 {
@@ -541,14 +544,13 @@ int Request::test(MPI_Request * request, MPI_Status * status, int* flag) {
   if(smpi_test_sleep > 0)
     simcall_process_sleep(nsleeps*smpi_test_sleep);
 
-  MPI_Status* mystatus;
   Status::empty(status);
   *flag = 1;
   if (((*request)->flags_ & MPI_REQ_PREPARED) == 0) {
     if ((*request)->action_ != nullptr){
       try{
         *flag = simcall_comm_test((*request)->action_);
-      }catch (xbt_ex& e) {
+      } catch (const Exception&) {
         *flag = 0;
         return ret;
       }
@@ -560,6 +562,7 @@ int Request::test(MPI_Request * request, MPI_Status * status, int* flag) {
     if (*flag) {
       finish_wait(request,status);
       if (*request != MPI_REQUEST_NULL && ((*request)->flags_ & MPI_REQ_GENERALIZED)){
+        MPI_Status* mystatus;
         if(status==MPI_STATUS_IGNORE){
           mystatus=new MPI_Status();
           Status::empty(mystatus);
@@ -622,7 +625,6 @@ int Request::testany(int count, MPI_Request requests[], int *index, int* flag, M
   int i;
   *flag = 0;
   int ret = MPI_SUCCESS;
-  MPI_Status* mystatus;
   *index = MPI_UNDEFINED;
 
   std::vector<int> map; /** Maps all matching comms back to their location in requests **/
@@ -639,7 +641,8 @@ int Request::testany(int count, MPI_Request requests[], int *index, int* flag, M
       simcall_process_sleep(nsleeps*smpi_test_sleep);
     try{
       i = simcall_comm_testany(comms.data(), comms.size()); // The i-th element in comms matches!
-    }catch (xbt_ex& e) {
+    } catch (const Exception&) {
+      XBT_DEBUG("Exception in testany");
       return 0;
     }
     
@@ -652,6 +655,7 @@ int Request::testany(int count, MPI_Request requests[], int *index, int* flag, M
       } else {
         finish_wait(&requests[*index],status);
       if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & MPI_REQ_GENERALIZED)){
+        MPI_Status* mystatus;
         if(status==MPI_STATUS_IGNORE){
           mystatus=new MPI_Status();
           Status::empty(mystatus);
@@ -665,6 +669,7 @@ int Request::testany(int count, MPI_Request requests[], int *index, int* flag, M
 
         if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & MPI_REQ_NON_PERSISTENT)) 
           requests[*index] = MPI_REQUEST_NULL;
+        XBT_DEBUG("Testany - returning with index %d", *index);
         *flag=1;
       }
       nsleeps = 1;
@@ -672,8 +677,10 @@ int Request::testany(int count, MPI_Request requests[], int *index, int* flag, M
       nsleeps++;
     }
   } else {
+      XBT_DEBUG("Testany on inactive handles, returning flag=1 but empty status");
       //all requests are null or inactive, return true
       *flag = 1;
+      *index = MPI_UNDEFINED;
       Status::empty(status);
   }
 
@@ -684,7 +691,8 @@ int Request::testall(int count, MPI_Request requests[], int* outflag, MPI_Status
 {
   MPI_Status stat;
   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
-  int flag, error=0;
+  int flag;
+  int error = 0;
   int ret=MPI_SUCCESS;
   *outflag = 1;
   for(int i=0; i<count; i++){
@@ -781,6 +789,7 @@ void Request::iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status*
       nsleeps++;
   }
   unref(&request);
+  xbt_assert(request == MPI_REQUEST_NULL);
 }
 
 void Request::finish_wait(MPI_Request* request, MPI_Status * status)
@@ -875,7 +884,7 @@ int Request::wait(MPI_Request * request, MPI_Status * status)
             int count=(*request)->size_/ (*request)->old_type_->size();
             (*request)->op_->apply(buf, (*request)->buf_, &count, (*request)->old_type_);
           }
-          smpi_free_tmp_buffer(buf);
+          smpi_free_tmp_buffer(static_cast<unsigned char*>(buf));
         }
       }
       if((*request)->nbc_requests_[i]!=MPI_REQUEST_NULL)
@@ -898,7 +907,7 @@ int Request::wait(MPI_Request * request, MPI_Status * status)
       try{
         // this is not a detached send
         simcall_comm_wait((*request)->action_, -1.0);
-      }catch (xbt_ex& e) {
+      } catch (const Exception&) {
         XBT_VERB("Request cancelled");
       }
   }
@@ -961,8 +970,8 @@ int Request::waitany(int count, MPI_Request requests[], MPI_Status * status)
       try{
         // this is not a detached send
         i = simcall_comm_waitany(comms.data(), comms.size(), -1);
-      }catch (xbt_ex& e) {
-      XBT_INFO("request %d cancelled ",i);
+      } catch (const Exception&) {
+        XBT_INFO("request %d cancelled ", i);
         return i;
       }
 
@@ -1059,7 +1068,7 @@ int Request::waitsome(int incount, MPI_Request requests[], int *indices, MPI_Sta
   indices[count] = index;
   count++;
   for (int i = 0; i < incount; i++) {
-    if((requests[i] != MPI_REQUEST_NULL)) {
+    if (requests[i] != MPI_REQUEST_NULL) {
       test(&requests[i], pstat,&flag);
       if (flag==1){
         indices[count] = i;
@@ -1135,7 +1144,7 @@ int Request::grequest_start( MPI_Grequest_query_function *query_fn, MPI_Grequest
   (*request)->flags_ |= MPI_REQ_GENERALIZED;
   (*request)->flags_ |= MPI_REQ_PERSISTENT;
   (*request)->refcount_ = 1;
-  ((*request)->generalized_funcs)=xbt_new0(s_smpi_mpi_generalized_request_funcs_t ,1);
+  ((*request)->generalized_funcs) = new s_smpi_mpi_generalized_request_funcs_t;
   ((*request)->generalized_funcs)->query_fn=query_fn;
   ((*request)->generalized_funcs)->free_fn=free_fn;
   ((*request)->generalized_funcs)->cancel_fn=cancel_fn;
@@ -1156,8 +1165,13 @@ int Request::grequest_complete( MPI_Request request){
 }
 
 void Request::set_nbc_requests(MPI_Request* reqs, int size){
-  nbc_requests_=reqs;
-  nbc_requests_size_=size;
+  nbc_requests_size_ = size;
+  if (size > 0) {
+    nbc_requests_ = reqs;
+  } else {
+    delete[] reqs;
+    nbc_requests_ = nullptr;
+  }
 }
 
 int Request::get_nbc_requests_size(){