X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/cb5f454f56e825bc82b74dd65b22a395ac754064..467fd11cdff5e25ee58862a196463f2bfff7796e:/src/smpi/mpi/smpi_request.cpp

diff --git a/src/smpi/mpi/smpi_request.cpp b/src/smpi/mpi/smpi_request.cpp
index 2057fd80b4..fe8e35b36c 100644
--- a/src/smpi/mpi/smpi_request.cpp
+++ b/src/smpi/mpi/smpi_request.cpp
@@ -39,7 +39,7 @@ namespace smpi{
 Request::Request(const void* buf, int count, MPI_Datatype datatype, aid_t src, aid_t dst, int tag, MPI_Comm comm,
                  unsigned flags, MPI_Op op)
     : buf_(const_cast<void*>(buf))
-    , old_type_(datatype)
+    , type_(datatype)
     , size_(datatype->size() * count)
     , src_(src)
     , dst_(dst)
@@ -57,6 +57,7 @@ Request::Request(const void* buf, int count, MPI_Datatype datatype, aid_t src, a
   detached_sender_ = nullptr;
   real_src_        = 0;
   truncated_       = false;
+  unmatched_types_ = false;
   real_size_       = 0;
   real_tag_        = 0;
   if (flags & MPI_REQ_PERSISTENT)
@@ -85,7 +86,7 @@ void Request::unref(MPI_Request* request)
       ((*request)->generalized_funcs)->free_fn(((*request)->generalized_funcs)->extra_state);
     } else {
       Comm::unref((*request)->comm_);
-      Datatype::unref((*request)->old_type_);
+      Datatype::unref((*request)->type_);
     }
     if ((*request)->op_ != MPI_REPLACE && (*request)->op_ != MPI_OP_NULL)
       Op::unref(&(*request)->op_);
@@ -99,6 +100,23 @@ void Request::unref(MPI_Request* request)
   }
 }
 
+bool Request::match_types(MPI_Datatype stype, MPI_Datatype rtype){
+  bool match = false;
+  if ((stype == rtype) ||
+     //byte and packed always match with anything
+     (stype == MPI_PACKED || rtype == MPI_PACKED || stype == MPI_BYTE || rtype == MPI_BYTE) ||
+     //complex datatypes - we don't properly match these yet, as it would mean checking each subtype recursively.
+     (stype->flags() & DT_FLAG_DERIVED || rtype->flags() & DT_FLAG_DERIVED) ||
+     //duplicated datatypes, check if underlying is ok
+     (stype->duplicated_datatype()!=MPI_DATATYPE_NULL && match_types(stype->duplicated_datatype(), rtype)) ||
+     (rtype->duplicated_datatype()!=MPI_DATATYPE_NULL && match_types(stype, rtype->duplicated_datatype())))
+    match = true;
+  if (!match)
+    XBT_WARN("Mismatched datatypes : sending %s and receiving %s", stype->name().c_str(), rtype->name().c_str());
+  return match;
+}
+
+
 bool Request::match_common(MPI_Request req, MPI_Request sender, MPI_Request receiver)
 {
   xbt_assert(sender, "Cannot match against null sender");
@@ -125,6 +143,10 @@ bool Request::match_common(MPI_Request req, MPI_Request sender, MPI_Request rece
         receiver->real_size_=sender->real_size_;
       }
     }
+    //0-sized datatypes/counts should not interfere and match
+    if ( sender->real_size_ != 0 && receiver->real_size_ != 0 &&
+         !match_types(sender->type_, receiver->type_))
+      receiver->unmatched_types_ = true;
     if (sender->detached_)
       receiver->detached_sender_ = sender; // tie the sender to the receiver, as it is detached and has to be freed in
                                            // the receiver
@@ -139,15 +161,15 @@ 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)) {
+  if ((((flags_ & MPI_REQ_RECV) != 0) && ((flags_ & MPI_REQ_ACCUMULATE) != 0)) || (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);
+      buf_ = xbt_malloc(count*type_->size());
+      if ((type_->flags() & DT_FLAG_DERIVED) && ((flags_ & MPI_REQ_SEND) != 0)) {
+        type_->serialize(old_buf, buf_, count);
       }
     }
   }
@@ -408,7 +430,7 @@ void Request::start()
   //reinitialize temporary buffer for persistent requests
   if(real_size_ > 0 && flags_ & MPI_REQ_FINISHED){
     buf_ = old_buf_;
-    init_buffer(real_size_/old_type_->size());
+    init_buffer(real_size_/type_->size());
   }
   flags_ &= ~MPI_REQ_PREPARED;
   flags_ &= ~MPI_REQ_FINISHED;
@@ -481,7 +503,7 @@ void Request::start()
       detached_    = true;
       XBT_DEBUG("Send request %p is detached", this);
       this->ref();
-      if (not(old_type_->flags() & DT_FLAG_DERIVED)) {
+      if (not(type_->flags() & DT_FLAG_DERIVED)) {
         oldbuf = buf_;
         if (not process->replaying() && oldbuf != nullptr && size_ != 0) {
           if ((smpi_cfg_privatization() != SmpiPrivStrategies::NONE) &&
@@ -616,7 +638,7 @@ int Request::test(MPI_Request * request, MPI_Status * status, int* flag) {
         return ret;
       }
     }
-    if (((*request)->flags_ & MPI_REQ_GENERALIZED) && !((*request)->flags_ & MPI_REQ_COMPLETE))
+    if (((*request)->flags_ & MPI_REQ_GENERALIZED) && not((*request)->flags_ & MPI_REQ_COMPLETE))
       *flag=0;
     if (*flag) {
       finish_wait(request, status); // may invalidate *request
@@ -705,9 +727,8 @@ int Request::testany(int count, MPI_Request requests[], int *index, int* flag, M
     
     if (i != -1) { // -1 is not MPI_UNDEFINED but a SIMIX return code. (nothing matches)
       *index = map[i];
-      if (requests[*index] != MPI_REQUEST_NULL && 
-          (requests[*index]->flags_ & MPI_REQ_GENERALIZED)
-          && !(requests[*index]->flags_ & MPI_REQ_COMPLETE)) {
+      if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & MPI_REQ_GENERALIZED) &&
+          not(requests[*index]->flags_ & MPI_REQ_COMPLETE)) {
         *flag=0;
       } else {
         finish_wait(&requests[*index],status);
@@ -865,12 +886,12 @@ int Request::finish_nbc_requests(MPI_Request* request, int test){
     for(auto& req: (*request)->nbc_requests_){
       if((*request)->buf_!=nullptr && req!=MPI_REQUEST_NULL){//reduce case
         void * buf=req->buf_;
-        if((*request)->old_type_->flags() & DT_FLAG_DERIVED)
+        if((*request)->type_->flags() & DT_FLAG_DERIVED)
           buf=req->old_buf_;
         if(req->flags_ & MPI_REQ_RECV ){
           if((*request)->op_!=MPI_OP_NULL){
-            int count=(*request)->size_/ (*request)->old_type_->size();
-            (*request)->op_->apply(buf, (*request)->buf_, &count, (*request)->old_type_);
+            int count=(*request)->size_/ (*request)->type_->size();
+            (*request)->op_->apply(buf, (*request)->buf_, &count, (*request)->type_);
           }
           smpi_free_tmp_buffer(static_cast<unsigned char*>(buf));
         }
@@ -913,7 +934,7 @@ void Request::finish_wait(MPI_Request* request, MPI_Status * status)
     //detached send will be finished at the other end
     if (not(req->detached_ && ((req->flags_ & MPI_REQ_SEND) != 0))) {
       req->print_request("Finishing");
-      MPI_Datatype datatype = req->old_type_;
+      MPI_Datatype datatype = req->type_;
 
       // FIXME Handle the case of a partial shared malloc.
       if (((req->flags_ & MPI_REQ_ACCUMULATE) != 0) ||
@@ -962,17 +983,22 @@ void Request::finish_wait(MPI_Request* request, MPI_Status * status)
     req->action_ = nullptr;
   req->flags_ |= MPI_REQ_FINISHED;
 
-  if (req->truncated_) {
+  if (req->truncated_ || req->unmatched_types_) {
     char error_string[MPI_MAX_ERROR_STRING];
     int error_size;
-    PMPI_Error_string(MPI_ERR_TRUNCATE, error_string, &error_size);
+    int errkind;
+    if(req->truncated_ )
+      errkind = MPI_ERR_TRUNCATE;
+    else
+      errkind = MPI_ERR_TYPE;
+    PMPI_Error_string(errkind, error_string, &error_size);
     MPI_Errhandler err = (req->comm_) ? (req->comm_)->errhandler() : MPI_ERRHANDLER_NULL;
     if (err == MPI_ERRHANDLER_NULL || err == MPI_ERRORS_RETURN)
       XBT_WARN("recv - returned %.*s instead of MPI_SUCCESS", error_size, error_string);
     else if (err == MPI_ERRORS_ARE_FATAL)
       xbt_die("recv - returned %.*s instead of MPI_SUCCESS", error_size, error_string);
     else
-      err->call((req->comm_), MPI_ERR_TRUNCATE);
+      err->call((req->comm_), errkind);
     if (err != MPI_ERRHANDLER_NULL)
       simgrid::smpi::Errhandler::unref(err);
     MC_assert(not MC_is_active()); /* Only fail in MC mode */
@@ -1013,7 +1039,7 @@ int Request::wait(MPI_Request * request, MPI_Status * status)
   }
 
   if ((*request)->flags_ & MPI_REQ_GENERALIZED) {
-    if(!((*request)->flags_ & MPI_REQ_COMPLETE)){
+    if (not((*request)->flags_ & MPI_REQ_COMPLETE)) {
       ((*request)->generalized_funcs)->mutex->lock();
       ((*request)->generalized_funcs)->cond->wait(((*request)->generalized_funcs)->mutex);
       ((*request)->generalized_funcs)->mutex->unlock();
@@ -1148,12 +1174,9 @@ int Request::waitall(int count, MPI_Request requests[], MPI_Status status[])
     }
   }
 
-  if (not accumulates.empty()) {
-    std::sort(accumulates.begin(), accumulates.end(), sort_accumulates);
-    for (auto& req : accumulates) {
-      finish_wait(&req, status);
-    }
-  }
+  std::sort(accumulates.begin(), accumulates.end(), sort_accumulates);
+  for (auto& req : accumulates)
+    finish_wait(&req, status);
 
   return retvalue;
 }
@@ -1213,10 +1236,8 @@ int Request::get_status(const Request* req, int* flag, MPI_Status* status)
     if(*flag)
       return MPI_SUCCESS;
   }
-  if (req != MPI_REQUEST_NULL && 
-     (req->flags_ & MPI_REQ_GENERALIZED)
-     && !(req->flags_ & MPI_REQ_COMPLETE)) {
-     *flag=0;
+  if (req != MPI_REQUEST_NULL && (req->flags_ & MPI_REQ_GENERALIZED) && not(req->flags_ & MPI_REQ_COMPLETE)) {
+    *flag = 0;
     return MPI_SUCCESS;
   }
 
@@ -1251,7 +1272,7 @@ int Request::grequest_start(MPI_Grequest_query_function* query_fn, MPI_Grequest_
 
 int Request::grequest_complete(MPI_Request request)
 {
-  if ((!(request->flags_ & MPI_REQ_GENERALIZED)) || request->generalized_funcs->mutex == nullptr)
+  if ((not(request->flags_ & MPI_REQ_GENERALIZED)) || request->generalized_funcs->mutex == nullptr)
     return MPI_ERR_REQUEST;
   request->generalized_funcs->mutex->lock();
   request->flags_ |= MPI_REQ_COMPLETE; // in case wait would be called after complete