X-Git-Url: http://bilbo.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/149c63f36e15b8500b1e826bda5138318ff7ba2b..5d3cd01db4d6a80961d0189d02094491d272c982:/src/smpi/mpi/smpi_request.cpp

diff --git a/src/smpi/mpi/smpi_request.cpp b/src/smpi/mpi/smpi_request.cpp
index 6a8ac60b4c..7812f96a1e 100644
--- a/src/smpi/mpi/smpi_request.cpp
+++ b/src/smpi/mpi/smpi_request.cpp
@@ -36,22 +36,6 @@ 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)
 {
-  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();
@@ -72,6 +56,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 +124,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 = const_cast<void*>(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);
@@ -355,6 +359,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();
@@ -376,7 +385,7 @@ void Request::start()
       mailbox = process->mailbox_small();
       XBT_DEBUG("Is there a corresponding send already posted in the small mailbox %s (in case of SSEND)?",
                 mailbox->get_cname());
-      smx_activity_t action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
+      simgrid::kernel::activity::ActivityImplPtr action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
 
       if (action == nullptr) {
         mailbox = process->mailbox();
@@ -392,7 +401,7 @@ void Request::start()
     } else {
       mailbox = process->mailbox_small();
       XBT_DEBUG("Is there a corresponding send already posted the small mailbox?");
-      smx_activity_t action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
+      simgrid::kernel::activity::ActivityImplPtr action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
 
       if (action == nullptr) {
         XBT_DEBUG("No, nothing in the permanent receive mailbox");
@@ -470,7 +479,7 @@ void Request::start()
     } else if (((flags_ & MPI_REQ_RMA) != 0) || static_cast<int>(size_) < smpi_cfg_async_small_thresh()) { // eager mode
       mailbox = process->mailbox();
       XBT_DEBUG("Is there a corresponding recv already posted in the large mailbox %s?", mailbox->get_cname());
-      smx_activity_t action = mailbox->iprobe(1, &match_send, static_cast<void*>(this));
+      simgrid::kernel::activity::ActivityImplPtr action = mailbox->iprobe(1, &match_send, static_cast<void*>(this));
       if (action == nullptr) {
         if ((flags_ & MPI_REQ_SSEND) == 0) {
           mailbox = process->mailbox_small();
@@ -496,8 +505,9 @@ void Request::start()
 
     // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
     real_size_=size_;
+    size_t payload_size_ = size_ + 16;//MPI enveloppe size (tag+dest+communicator)
     action_   = simcall_comm_isend(
-        simgrid::s4u::Actor::by_pid(src_)->get_impl(), mailbox->get_impl(), size_, -1.0, buf, real_size_, &match_send,
+        simgrid::s4u::Actor::by_pid(src_)->get_impl(), mailbox->get_impl(), payload_size_, -1.0, buf, real_size_, &match_send,
         &xbt_free_f, // how to free the userdata if a detached send fails
         process->replaying() ? &smpi_comm_null_copy_buffer_callback : smpi_comm_copy_data_callback, this,
         // detach if msg size < eager/rdv switch limit
@@ -559,10 +569,10 @@ 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_);
+        *flag = simcall_comm_test((*request)->action_.get());
       } catch (const Exception&) {
         *flag = 0;
         return ret;
@@ -846,12 +856,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;
         }
       }
     }
@@ -909,7 +921,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;
   }
@@ -917,7 +929,7 @@ int Request::wait(MPI_Request * request, MPI_Status * status)
   if ((*request)->action_ != nullptr){
       try{
         // this is not a detached send
-        simcall_comm_wait((*request)->action_, -1.0);
+        simcall_comm_wait((*request)->action_.get(), -1.0);
       } catch (const Exception&) {
         XBT_VERB("Request cancelled");
       }