add a refcount for MPI_Op

[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 694a2b1..0762073 100644 (file)
--- a/src/smpi/mpi/smpi_request.cpp
+++ b/src/smpi/mpi/smpi_request.cpp
@@ -36,8 +36,8 @@ 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)
-    : buf_(buf), old_type_(datatype), src_(src), dst_(dst), tag_(tag), comm_(comm), flags_(flags)
+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)
 {
   void *old_buf = nullptr;
 // FIXME Handle the case of a partial shared malloc.
@@ -58,6 +58,8 @@ Request::Request(void* buf, int count, MPI_Datatype datatype, int src, int dst,
   size_ = datatype->size() * count;
   datatype->ref();
   comm_->ref();
+  if(op != MPI_REPLACE && op != MPI_OP_NULL)
+    op_->ref();
   action_          = nullptr;
   detached_        = 0;
   detached_sender_ = nullptr;
@@ -69,8 +71,10 @@ Request::Request(void* buf, int count, MPI_Datatype datatype, int src, int dst,
     refcount_ = 1;
   else
     refcount_ = 0;
-  op_   = MPI_REPLACE;
   cancelled_ = 0;
+  generalized_funcs=nullptr;
+  nbc_requests_=nullptr;
+  nbc_requests_size_=0;
 }
 
 void Request::ref(){
@@ -92,6 +96,9 @@ void Request::unref(MPI_Request* request)
         Comm::unref((*request)->comm_);
         Datatype::unref((*request)->old_type_);
       }
+      if ((*request)->op_!=MPI_REPLACE && (*request)->op_!=MPI_OP_NULL)
+        Op::unref(&(*request)->op_);
+
       (*request)->print_request("Destroying");
       delete *request;
       *request = MPI_REQUEST_NULL;
@@ -198,8 +205,7 @@ MPI_Request Request::rma_send_init(void *buf, int count, MPI_Datatype datatype,
     request      = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, comm->group()->actor(src)->get_pid(),
                           comm->group()->actor(dst)->get_pid(), tag, comm,
                           MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND | MPI_REQ_PREPARED |
-                              MPI_REQ_ACCUMULATE);
-    request->op_ = op;
+                              MPI_REQ_ACCUMULATE, op);
   }
   return request;
 }
@@ -223,8 +229,7 @@ MPI_Request Request::rma_recv_init(void *buf, int count, MPI_Datatype datatype,
   }else{
     request      = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, comm->group()->actor(src)->get_pid(),
                           comm->group()->actor(dst)->get_pid(), tag, comm,
-                          MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED | MPI_REQ_ACCUMULATE);
-    request->op_ = op;
+                          MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED | MPI_REQ_ACCUMULATE, op);
   }
   return request;
 }
@@ -520,6 +525,19 @@ int Request::test(MPI_Request * request, MPI_Status * status, int* flag) {
   // multiplier to the sleeptime, to increase speed of execution, each failed test will increase it
   static int nsleeps = 1;
   int ret = MPI_SUCCESS;
+  
+  // Are we testing a request meant for non blocking collectives ?
+  // If so, test all the subrequests.
+  if ((*request)->nbc_requests_size_>0){
+    ret = testall((*request)->nbc_requests_size_, (*request)->nbc_requests_, flag, MPI_STATUSES_IGNORE);
+    if(*flag){
+      delete[] (*request)->nbc_requests_;
+      (*request)->nbc_requests_size_=0;
+      unref(request);
+    }
+    return ret;
+  }
+  
   if(smpi_test_sleep > 0)
     simcall_process_sleep(nsleeps*smpi_test_sleep);
 
@@ -844,6 +862,20 @@ void Request::finish_wait(MPI_Request* request, MPI_Status * status)
 int Request::wait(MPI_Request * request, MPI_Status * status)
 {
   int ret=MPI_SUCCESS;
+  // Are we waiting on a request meant for non blocking collectives ?
+  // If so, wait for all the subrequests.
+  if ((*request)->nbc_requests_size_>0){
+    ret = waitall((*request)->nbc_requests_size_, (*request)->nbc_requests_, MPI_STATUSES_IGNORE);
+    for (int i = 0; i < (*request)->nbc_requests_size_; i++) {
+      if((*request)->nbc_requests_[i]!=MPI_REQUEST_NULL)
+        Request::unref(&((*request)->nbc_requests_[i]));
+    }
+    delete[] (*request)->nbc_requests_;
+    (*request)->nbc_requests_size_=0;
+    unref(request);
+    return ret;
+  }
+
   (*request)->print_request("Waiting");
   if ((*request)->flags_ & MPI_REQ_PREPARED) {
     Status::empty(status);
@@ -1063,7 +1095,7 @@ int Request::get_status(MPI_Request req, int* flag, MPI_Status * status){
 
   if(req != MPI_REQUEST_NULL && req->action_ != nullptr) {
     req->iprobe(req->src_, req->tag_, req->comm_, flag, status);
-    if(flag)
+    if(*flag)
       return MPI_SUCCESS;
   }
   if (req != MPI_REQUEST_NULL && 
@@ -1074,7 +1106,8 @@ int Request::get_status(MPI_Request req, int* flag, MPI_Status * status){
   }
 
   *flag=1;
-  if(status != MPI_STATUS_IGNORE) {
+  if(req != MPI_REQUEST_NULL &&
+     status != MPI_STATUS_IGNORE) {
     int src = req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_;
     status->MPI_SOURCE = req->comm_->group()->rank(src);
     status->MPI_TAG = req->tag_ == MPI_ANY_TAG ? req->real_tag_ : req->tag_;
@@ -1110,5 +1143,18 @@ int Request::grequest_complete( MPI_Request request){
   return MPI_SUCCESS;
 }
 
+void Request::set_nbc_requests(MPI_Request* reqs, int size){
+  nbc_requests_=reqs;
+  nbc_requests_size_=size;
+}
+
+int Request::get_nbc_requests_size(){
+  return nbc_requests_size_;
+}
+
+MPI_Request* Request::get_nbc_requests(){
+  return nbc_requests_;
+}
+
 }
 }