X-Git-Url: http://bilbo.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/009ff7e4463fae3c7a2181c1f8147fd100afbfdd..503aa37006d22d261b792dd00d526976fafa7052:/src/smpi/mpi/smpi_request.cpp

diff --git a/src/smpi/mpi/smpi_request.cpp b/src/smpi/mpi/smpi_request.cpp
index b12e4b0a75..2bc696f74f 100644
--- a/src/smpi/mpi/smpi_request.cpp
+++ b/src/smpi/mpi/smpi_request.cpp
@@ -33,26 +33,18 @@ extern void (*smpi_comm_copy_data_callback)(simgrid::kernel::activity::CommImpl*
 namespace simgrid{
 namespace smpi{
 
-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)
+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)
+    , size_(datatype->size() * count)
+    , 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 = const_cast<void*>(buf);
-    if (count==0){
-      buf_ = nullptr;
-    }else {
-      buf_ = xbt_malloc(count*datatype->size());
-      if ((datatype->flags() & DT_FLAG_DERIVED) && ((flags & MPI_REQ_SEND) != 0)) {
-        datatype->serialize(old_buf, buf_, count);
-      }
-    }
-  }
-  // This part handles the problem of non-contiguous memory (for the unserialization at the reception)
-  old_buf_  = old_buf;
-  size_ = datatype->size() * count;
   datatype->ref();
   comm_->ref();
   if(op != MPI_REPLACE && op != MPI_OP_NULL)
@@ -72,6 +64,7 @@ Request::Request(const void* buf, int count, MPI_Datatype datatype, int src, int
   generalized_funcs=nullptr;
   nbc_requests_=nullptr;
   nbc_requests_size_=0;
+  init_buffer(count);
 }
 
 void Request::ref(){
@@ -139,6 +132,25 @@ bool Request::match_common(MPI_Request req, MPI_Request sender, MPI_Request rece
   return false;
 }
 
+void Request::init_buffer(int count){
+  void *old_buf = nullptr;
+// FIXME Handle the case of a partial shared malloc.
+  // This part handles the problem of non-contiguous memory (for the unserialization at the reception)
+  if ((((flags_ & MPI_REQ_RECV) != 0) && ((flags_ & MPI_REQ_ACCUMULATE) != 0)) || (old_type_->flags() & DT_FLAG_DERIVED)) {
+    // This part handles the problem of non-contiguous memory
+    old_buf = buf_;
+    if (count==0){
+      buf_ = nullptr;
+    }else {
+      buf_ = xbt_malloc(count*old_type_->size());
+      if ((old_type_->flags() & DT_FLAG_DERIVED) && ((flags_ & MPI_REQ_SEND) != 0)) {
+        old_type_->serialize(old_buf, buf_, count);
+      }
+    }
+  }
+  old_buf_  = old_buf;
+}
+
 bool Request::match_recv(void* a, void* b, simgrid::kernel::activity::CommImpl*)
 {
   MPI_Request ref = static_cast<MPI_Request>(a);
@@ -153,7 +165,7 @@ bool Request::match_send(void* a, void* b, simgrid::kernel::activity::CommImpl*)
   return match_common(req, ref, req);
 }
 
-void Request::print_request(const char *message)
+void Request::print_request(const char* message) const
 {
   XBT_VERB("%s  request %p  [buf = %p, size = %zu, src = %d, dst = %d, tag = %d, flags = %x]",
        message, this, buf_, size_, src_, dst_, tag_, flags_);
@@ -355,6 +367,11 @@ void Request::start()
   s4u::Mailbox* mailbox;
 
   xbt_assert(action_ == nullptr, "Cannot (re-)start unfinished communication");
+  //reinitialize temporary buffer for persistent requests
+  if(real_size_ > 0 && flags_ & MPI_REQ_FINISHED){
+    buf_ = old_buf_;
+    init_buffer(real_size_/old_type_->size());
+  }
   flags_ &= ~MPI_REQ_PREPARED;
   flags_ &= ~MPI_REQ_FINISHED;
   this->ref();
@@ -412,7 +429,7 @@ void Request::start()
     if (smpi_cfg_async_small_thresh() != 0 || (flags_ & MPI_REQ_RMA) != 0)
       mut->unlock();
   } else { /* the RECV flag was not set, so this is a send */
-    simgrid::smpi::ActorExt* process = smpi_process_remote(simgrid::s4u::Actor::by_pid(dst_));
+    const simgrid::smpi::ActorExt* process = smpi_process_remote(simgrid::s4u::Actor::by_pid(dst_));
     xbt_assert(process, "Actor pid=%d is gone??", dst_);
     int rank = src_;
     if (TRACE_smpi_view_internals()) {
@@ -560,7 +577,7 @@ int Request::test(MPI_Request * request, MPI_Status * status, int* flag) {
 
   Status::empty(status);
   *flag = 1;
-  if (((*request)->flags_ & MPI_REQ_PREPARED) == 0) {
+  if (((*request)->flags_ & (MPI_REQ_PREPARED | MPI_REQ_FINISHED)) == 0) {
     if ((*request)->action_ != nullptr && (*request)->cancelled_ != 1){
       try{
         *flag = simcall_comm_test((*request)->action_.get());
@@ -847,12 +864,14 @@ void Request::finish_wait(MPI_Request* request, MPI_Status * status)
           if ((req->flags_ & MPI_REQ_RECV) && datatype->size() != 0)
             datatype->unserialize(req->buf_, req->old_buf_, req->real_size_/datatype->size() , req->op_);
           xbt_free(req->buf_);
+          req->buf_=nullptr;
         } else if (req->flags_ & MPI_REQ_RECV) { // apply op on contiguous buffer for accumulate
           if (datatype->size() != 0) {
             int n = req->real_size_ / datatype->size();
             req->op_->apply(req->buf_, req->old_buf_, &n, datatype);
           }
           xbt_free(req->buf_);
+          req->buf_=nullptr;
         }
       }
     }
@@ -910,7 +929,7 @@ int Request::wait(MPI_Request * request, MPI_Status * status)
   }
 
   (*request)->print_request("Waiting");
-  if ((*request)->flags_ & MPI_REQ_PREPARED) {
+  if ((*request)->flags_ & (MPI_REQ_PREPARED | MPI_REQ_FINISHED)) {
     Status::empty(status);
     return ret;
   }
@@ -1007,7 +1026,7 @@ int Request::waitany(int count, MPI_Request requests[], MPI_Status * status)
   return index;
 }
 
-static int sort_accumulates(MPI_Request a, MPI_Request b)
+static int sort_accumulates(const Request* a, const Request* b)
 {
   return (a->tag() > b->tag());
 }
@@ -1036,8 +1055,8 @@ int Request::waitall(int count, MPI_Request requests[], MPI_Status status[])
       wait(&requests[c],pstat);
       index = c;
     } else {
-      index = waitany(count, (MPI_Request*)requests, pstat);
-      
+      index = waitany(count, requests, pstat);
+
       if (index == MPI_UNDEFINED)
         break;
 
@@ -1071,7 +1090,7 @@ int Request::waitsome(int incount, MPI_Request requests[], int *indices, MPI_Sta
   int index = 0;
   MPI_Status stat;
   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
-  index = waitany(incount, (MPI_Request*)requests, pstat);
+  index             = waitany(incount, requests, pstat);
   if(index==MPI_UNDEFINED) return MPI_UNDEFINED;
   if(status != MPI_STATUSES_IGNORE) {
     status[count] = *pstat;
@@ -1100,7 +1119,7 @@ MPI_Request Request::f2c(int id)
 {
   char key[KEY_SIZE];
   if(id==MPI_FORTRAN_REQUEST_NULL)
-    return static_cast<MPI_Request>(MPI_REQUEST_NULL);
+    return MPI_REQUEST_NULL;
   return static_cast<MPI_Request>(F2C::f2c_lookup()->at(get_key(key,id)));
 }
 
@@ -1178,13 +1197,14 @@ void Request::set_nbc_requests(MPI_Request* reqs, int size){
   }
 }
 
-int Request::get_nbc_requests_size(){
+int Request::get_nbc_requests_size() const
+{
   return nbc_requests_size_;
 }
 
-MPI_Request* Request::get_nbc_requests(){
+MPI_Request* Request::get_nbc_requests() const
+{
   return nbc_requests_;
 }
-
 }
 }