src/smpi/mpi/smpi_request.cpp

   1 /* Copyright (c) 2007-2022. The SimGrid Team. All rights reserved.          */
   2
   3 /* This program is free software; you can redistribute it and/or modify it
   4  * under the terms of the license (GNU LGPL) which comes with this package. */
   5
   6 #include "smpi_request.hpp"
   7
   8 #include "mc/mc.h"
   9 #include "private.hpp"
  10 #include "simgrid/Exception.hpp"
  11 #include "simgrid/s4u/ConditionVariable.hpp"
  12 #include "simgrid/s4u/Exec.hpp"
  13 #include "simgrid/s4u/Mutex.hpp"
  14 #include "smpi_comm.hpp"
  15 #include "smpi_datatype.hpp"
  16 #include "smpi_host.hpp"
  17 #include "smpi_op.hpp"
  18 #include "src/kernel/EngineImpl.hpp"
  19 #include "src/kernel/activity/CommImpl.hpp"
  20 #include "src/kernel/actor/ActorImpl.hpp"
  21 #include "src/kernel/actor/SimcallObserver.hpp"
  22 #include "src/mc/mc_replay.hpp"
  23 #include "src/smpi/include/smpi_actor.hpp"
  24
  25 #include <algorithm>
  26 #include <array>
  27
  28 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_request, smpi, "Logging specific to SMPI (request)");
  29
  30 static simgrid::config::Flag<double> smpi_iprobe_sleep(
  31   "smpi/iprobe", "Minimum time to inject inside a call to MPI_Iprobe", 1e-4);
  32 static simgrid::config::Flag<double> smpi_test_sleep(
  33   "smpi/test", "Minimum time to inject inside a call to MPI_Test", 1e-4);
  34
  35 std::vector<s_smpi_factor_t> smpi_ois_values;
  36
  37 extern std::function<void(simgrid::kernel::activity::CommImpl*, void*, size_t)> smpi_comm_copy_data_callback;
  38
  39 namespace simgrid::smpi {
  40
  41 Request::Request(const void* buf, int count, MPI_Datatype datatype, aid_t src, aid_t dst, int tag, MPI_Comm comm,
  42                  unsigned flags, MPI_Op op)
  43     : buf_(const_cast<void*>(buf))
  44     , old_buf_(buf_)
  45     , type_(datatype)
  46     , size_(datatype->size() * count)
  47     , src_(src)
  48     , dst_(dst)
  49     , tag_(tag)
  50     , comm_(comm)
  51     , flags_(flags)
  52     , op_(op)
  53 {
  54   datatype->ref();
  55   comm_->ref();
  56   if(op != MPI_REPLACE && op != MPI_OP_NULL)
  57     op_->ref();
  58   action_          = nullptr;
  59   detached_        = false;
  60   detached_sender_ = nullptr;
  61   real_src_        = 0;
  62   // get src_host if it's available (src is valid)
  63   if (auto src_process = simgrid::s4u::Actor::by_pid(src))
  64     src_host_ = src_process->get_host();
  65   truncated_       = false;
  66   unmatched_types_ = false;
  67   real_size_       = 0;
  68   real_tag_        = 0;
  69   if (flags & MPI_REQ_PERSISTENT)
  70     refcount_ = 1;
  71   else
  72     refcount_ = 0;
  73   message_id_ = 0;
  74   init_buffer(count);
  75   this->add_f();
  76 }
  77
  78 void Request::ref(){
  79   refcount_++;
  80 }
  81
  82 void Request::unref(MPI_Request* request)
  83 {
  84   xbt_assert(*request != MPI_REQUEST_NULL, "freeing an already free request");
  85
  86   (*request)->refcount_--;
  87   if ((*request)->refcount_ < 0) {
  88     (*request)->print_request("wrong refcount");
  89     xbt_die("Whoops, wrong refcount");
  90   }
  91   if ((*request)->refcount_ == 0) {
  92     if ((*request)->flags_ & MPI_REQ_GENERALIZED) {
  93       ((*request)->generalized_funcs)->free_fn(((*request)->generalized_funcs)->extra_state);
  94     } else {
  95       Comm::unref((*request)->comm_);
  96       Datatype::unref((*request)->type_);
  97     }
  98     if ((*request)->op_ != MPI_REPLACE && (*request)->op_ != MPI_OP_NULL)
  99       Op::unref(&(*request)->op_);
 100
 101     (*request)->print_request("Destroying");
 102     F2C::free_f((*request)->f2c_id());
 103     delete *request;
 104     *request = MPI_REQUEST_NULL;
 105   } else {
 106     (*request)->print_request("Decrementing");
 107   }
 108 }
 109
 110 bool Request::match_types(MPI_Datatype stype, MPI_Datatype rtype){
 111   bool match = false;
 112   if ((stype == rtype) ||
 113      //byte and packed always match with anything
 114      (stype == MPI_PACKED || rtype == MPI_PACKED || stype == MPI_BYTE || rtype == MPI_BYTE) ||
 115      //complex datatypes - we don't properly match these yet, as it would mean checking each subtype recursively.
 116      (stype->flags() & DT_FLAG_DERIVED || rtype->flags() & DT_FLAG_DERIVED) ||
 117      //duplicated datatypes, check if underlying is ok
 118      (stype->duplicated_datatype()!=MPI_DATATYPE_NULL && match_types(stype->duplicated_datatype(), rtype)) ||
 119      (rtype->duplicated_datatype()!=MPI_DATATYPE_NULL && match_types(stype, rtype->duplicated_datatype())))
 120     match = true;
 121   if (!match)
 122     XBT_WARN("Mismatched datatypes : sending %s and receiving %s", stype->name().c_str(), rtype->name().c_str());
 123   return match;
 124 }
 125
 126
 127 bool Request::match_common(MPI_Request req, MPI_Request sender, MPI_Request receiver)
 128 {
 129   xbt_assert(sender, "Cannot match against null sender");
 130   xbt_assert(receiver, "Cannot match against null receiver");
 131   XBT_DEBUG("Trying to match %s of sender src %ld against %ld, tag %d against %d, id %d against %d",
 132             (req == receiver ? "send" : "recv"), sender->src_, receiver->src_, sender->tag_, receiver->tag_,
 133             sender->comm_->id(), receiver->comm_->id());
 134
 135   if ((receiver->comm_->id() == MPI_UNDEFINED || sender->comm_->id() == MPI_UNDEFINED ||
 136        receiver->comm_->id() == sender->comm_->id()) &&
 137       ((receiver->src_ == MPI_ANY_SOURCE && (receiver->comm_->group()->rank(sender->src_) != MPI_UNDEFINED)) ||
 138        receiver->src_ == sender->src_) &&
 139       ((receiver->tag_ == MPI_ANY_TAG && sender->tag_ >= 0) || receiver->tag_ == sender->tag_)) {
 140     // we match, we can transfer some values
 141     if (receiver->src_ == MPI_ANY_SOURCE) {
 142       receiver->real_src_ = sender->src_;
 143       receiver->src_host_ = sender->src_host_;
 144     }
 145     if (receiver->tag_ == MPI_ANY_TAG)
 146       receiver->real_tag_ = sender->tag_;
 147     if ((receiver->flags_ & MPI_REQ_PROBE) == 0 && receiver->real_size_ < sender->real_size_) {
 148       XBT_DEBUG("Truncating message - should not happen: receiver size : %zu < sender size : %zu", receiver->real_size_,
 149                 sender->real_size_);
 150       receiver->truncated_ = true;
 151     }
 152     //0-sized datatypes/counts should not interfere and match
 153     if ( sender->real_size_ != 0 && receiver->real_size_ != 0 &&
 154          !match_types(sender->type_, receiver->type_))
 155       receiver->unmatched_types_ = true;
 156     if (sender->detached_)
 157       receiver->detached_sender_ = sender; // tie the sender to the receiver, as it is detached and has to be freed in
 158                                            // the receiver
 159     req->flags_ |= MPI_REQ_MATCHED; // mark as impossible to cancel anymore
 160     XBT_DEBUG("match succeeded");
 161     return true;
 162   }
 163   return false;
 164 }
 165
 166 void Request::init_buffer(int count){
 167 // FIXME Handle the case of a partial shared malloc.
 168   // This part handles the problem of non-contiguous memory (for the unserialization at the reception)
 169   if (not smpi_process()->replaying() &&
 170      ((((flags_ & MPI_REQ_RECV) != 0) && ((flags_ & MPI_REQ_ACCUMULATE) != 0)) || (type_->flags() & DT_FLAG_DERIVED))) {
 171     // This part handles the problem of non-contiguous memory
 172     old_buf_ = buf_;
 173     if (count==0){
 174       buf_ = nullptr;
 175     }else {
 176       buf_ = xbt_malloc(count*type_->size());
 177       if ((type_->flags() & DT_FLAG_DERIVED) && ((flags_ & MPI_REQ_SEND) != 0)) {
 178         type_->serialize(old_buf_, buf_, count);
 179       }
 180     }
 181   }
 182 }
 183
 184 bool Request::match_recv(void* a, void* b, simgrid::kernel::activity::CommImpl*)
 185 {
 186   auto ref = static_cast<MPI_Request>(a);
 187   auto req = static_cast<MPI_Request>(b);
 188   bool match = match_common(req, req, ref);
 189   if (not match || ref->comm_ == MPI_COMM_UNINITIALIZED || ref->comm_->is_smp_comm())
 190     return match;
 191
 192   if (ref->comm_->get_received_messages_count(ref->comm_->group()->rank(req->src_),
 193                                               ref->comm_->group()->rank(req->dst_), req->tag_) == req->message_id_) {
 194     if (((ref->flags_ & MPI_REQ_PROBE) == 0) && ((req->flags_ & MPI_REQ_PROBE) == 0)) {
 195       XBT_DEBUG("increasing count in comm %p, which was %u from pid %ld, to pid %ld with tag %d", ref->comm_,
 196                 ref->comm_->get_received_messages_count(ref->comm_->group()->rank(req->src_),
 197                                                         ref->comm_->group()->rank(req->dst_), req->tag_),
 198                 req->src_, req->dst_, req->tag_);
 199       ref->comm_->increment_received_messages_count(ref->comm_->group()->rank(req->src_),
 200                                                     ref->comm_->group()->rank(req->dst_), req->tag_);
 201       if (ref->real_size_ > req->real_size_) {
 202         ref->real_size_ = req->real_size_;
 203       }
 204     }
 205   } else {
 206     match = false;
 207     req->flags_ &= ~MPI_REQ_MATCHED;
 208     ref->detached_sender_ = nullptr;
 209     XBT_DEBUG("Refusing to match message, as its ID is not the one I expect. in comm %p, %u != %u, "
 210               "from pid %ld to pid %ld, with tag %d",
 211               ref->comm_,
 212               ref->comm_->get_received_messages_count(ref->comm_->group()->rank(req->src_),
 213                                                       ref->comm_->group()->rank(req->dst_), req->tag_),
 214               req->message_id_, req->src_, req->dst_, req->tag_);
 215   }
 216   return match;
 217 }
 218
 219 bool Request::match_send(void* a, void* b, simgrid::kernel::activity::CommImpl*)
 220 {
 221   auto ref = static_cast<MPI_Request>(a);
 222   auto req = static_cast<MPI_Request>(b);
 223   return match_common(req, ref, req);
 224 }
 225
 226 void Request::print_request(const char* message) const
 227 {
 228   XBT_VERB("%s  request %p  [buf = %p, size = %zu, src = %ld, dst = %ld, tag = %d, flags = %x]", message, this, buf_,
 229            size_, src_, dst_, tag_, flags_);
 230 }
 231
 232 /* factories, to hide the internal flags from the caller */
 233 MPI_Request Request::bsend_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 234 {
 235   return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
 236                      dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 237                      MPI_REQ_PERSISTENT | MPI_REQ_SEND | MPI_REQ_PREPARED | MPI_REQ_BSEND);
 238 }
 239
 240 MPI_Request Request::send_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 241 {
 242   return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
 243                      dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 244                      MPI_REQ_PERSISTENT | MPI_REQ_SEND | MPI_REQ_PREPARED);
 245 }
 246
 247 MPI_Request Request::ssend_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 248 {
 249   return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
 250                      dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 251                      MPI_REQ_PERSISTENT | MPI_REQ_SSEND | MPI_REQ_SEND | MPI_REQ_PREPARED);
 252 }
 253
 254 MPI_Request Request::isend_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 255 {
 256   return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
 257                      dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 258                      MPI_REQ_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND | MPI_REQ_PREPARED);
 259 }
 260
 261 MPI_Request Request::rma_send_init(const void *buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm,
 262                                MPI_Op op)
 263 {
 264   MPI_Request request;
 265   if(op==MPI_OP_NULL){
 266     request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, comm->group()->actor(src),
 267                           dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 268                           MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND | MPI_REQ_PREPARED);
 269   }else{
 270     request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, comm->group()->actor(src),
 271                           dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 272                           MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND | MPI_REQ_PREPARED |
 273                               MPI_REQ_ACCUMULATE,
 274                           op);
 275   }
 276   return request;
 277 }
 278
 279 MPI_Request Request::recv_init(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm)
 280 {
 281   aid_t source = MPI_PROC_NULL;
 282   if (src == MPI_ANY_SOURCE)
 283     source = MPI_ANY_SOURCE;
 284   else if (src != MPI_PROC_NULL)
 285     source = comm->group()->actor(src);
 286   return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype,
 287                      source,
 288                      simgrid::s4u::this_actor::get_pid(), tag, comm,
 289                      MPI_REQ_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED);
 290 }
 291
 292 MPI_Request Request::rma_recv_init(void *buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm,
 293                                MPI_Op op)
 294 {
 295   aid_t source        = MPI_PROC_NULL;
 296   if (src == MPI_ANY_SOURCE)
 297     source = MPI_ANY_SOURCE;
 298   else if (src != MPI_PROC_NULL)
 299     source = comm->group()->actor(src);
 300   MPI_Request request;
 301   if(op==MPI_OP_NULL){
 302     request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, source,
 303                           dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 304                           MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED);
 305   }else{
 306     request =
 307         new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, source,
 308                     dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 309                     MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED | MPI_REQ_ACCUMULATE, op);
 310   }
 311   return request;
 312 }
 313
 314 MPI_Request Request::irecv_init(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm)
 315 {
 316   aid_t source = MPI_PROC_NULL;
 317   if (src == MPI_ANY_SOURCE)
 318     source = MPI_ANY_SOURCE;
 319   else if (src != MPI_PROC_NULL)
 320     source = comm->group()->actor(src);
 321   return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype,
 322                      source, simgrid::s4u::this_actor::get_pid(), tag, comm,
 323                      MPI_REQ_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED);
 324 }
 325
 326 MPI_Request Request::ibsend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 327 {
 328   auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
 329                              dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 330                              MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND | MPI_REQ_BSEND);
 331   if(dst != MPI_PROC_NULL)
 332     request->start();
 333   return request;
 334 }
 335
 336 MPI_Request Request::isend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 337 {
 338   auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
 339                              dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 340                              MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND);
 341   if(dst != MPI_PROC_NULL)
 342     request->start();
 343   return request;
 344 }
 345
 346 MPI_Request Request::issend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 347 {
 348   auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
 349                              dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 350                              MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SSEND | MPI_REQ_SEND);
 351   if(dst != MPI_PROC_NULL)
 352     request->start();
 353   return request;
 354 }
 355
 356 MPI_Request Request::irecv(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm)
 357 {
 358   aid_t source        = MPI_PROC_NULL;
 359   if (src == MPI_ANY_SOURCE)
 360     source = MPI_ANY_SOURCE;
 361   else if (src != MPI_PROC_NULL)
 362     source = comm->group()->actor(src);
 363   auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, source,
 364                              simgrid::s4u::this_actor::get_pid(), tag, comm, MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV);
 365   if(src != MPI_PROC_NULL)
 366     request->start();
 367   return request;
 368 }
 369
 370 int Request::recv(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status * status)
 371 {
 372   MPI_Request request = irecv(buf, count, datatype, src, tag, comm);
 373   int retval = wait(&request,status);
 374   request = nullptr;
 375   return retval;
 376 }
 377
 378 void Request::bsend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 379 {
 380   auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
 381                              dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 382                              MPI_REQ_NON_PERSISTENT | MPI_REQ_SEND | MPI_REQ_BSEND);
 383
 384   if(dst != MPI_PROC_NULL)
 385    request->start();
 386   wait(&request, MPI_STATUS_IGNORE);
 387   request = nullptr;
 388 }
 389
 390 void Request::send(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 391 {
 392   auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
 393                              dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 394                              MPI_REQ_NON_PERSISTENT | MPI_REQ_SEND);
 395   if(dst != MPI_PROC_NULL)
 396    request->start();
 397   wait(&request, MPI_STATUS_IGNORE);
 398   request = nullptr;
 399 }
 400
 401 void Request::ssend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 402 {
 403   auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
 404                              dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm,
 405                              MPI_REQ_NON_PERSISTENT | MPI_REQ_SSEND | MPI_REQ_SEND);
 406
 407   if(dst != MPI_PROC_NULL)
 408    request->start();
 409   wait(&request,MPI_STATUS_IGNORE);
 410   request = nullptr;
 411 }
 412
 413 void Request::sendrecv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,int dst, int sendtag,
 414                        void *recvbuf, int recvcount, MPI_Datatype recvtype, int src, int recvtag,
 415                        MPI_Comm comm, MPI_Status * status)
 416 {
 417   aid_t source = MPI_PROC_NULL;
 418   if (src == MPI_ANY_SOURCE)
 419     source = MPI_ANY_SOURCE;
 420   else if (src != MPI_PROC_NULL)
 421     source = comm->group()->actor(src);
 422   aid_t destination = dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL;
 423
 424   std::array<MPI_Request, 2> requests;
 425   std::array<MPI_Status, 2> stats;
 426   if (aid_t myid = simgrid::s4u::this_actor::get_pid(); (destination == myid) && (source == myid)) {
 427     Datatype::copy(sendbuf, sendcount, sendtype, recvbuf, recvcount, recvtype);
 428     if (status != MPI_STATUS_IGNORE) {
 429       status->MPI_SOURCE = source;
 430       status->MPI_TAG    = recvtag;
 431       status->MPI_ERROR  = MPI_SUCCESS;
 432       status->count      = sendcount * sendtype->size();
 433     }
 434     return;
 435   }
 436   requests[0] = isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm);
 437   requests[1] = irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm);
 438   startall(2, requests.data());
 439   waitall(2, requests.data(), stats.data());
 440   unref(&requests[0]);
 441   unref(&requests[1]);
 442   if(status != MPI_STATUS_IGNORE) {
 443     // Copy receive status
 444     *status = stats[1];
 445   }
 446 }
 447
 448 void Request::start()
 449 {
 450   s4u::Mailbox* mailbox;
 451
 452   xbt_assert(action_ == nullptr, "Cannot (re-)start unfinished communication");
 453   //reinitialize temporary buffer for persistent requests
 454   if(real_size_ > 0 && flags_ & MPI_REQ_FINISHED){
 455     buf_ = old_buf_;
 456     init_buffer(real_size_/type_->size());
 457   }
 458   flags_ &= ~MPI_REQ_PREPARED;
 459   flags_ &= ~MPI_REQ_FINISHED;
 460   this->ref();
 461
 462   // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
 463   real_size_=size_;
 464   if ((flags_ & MPI_REQ_RECV) != 0) {
 465     this->print_request("New recv");
 466
 467     simgrid::smpi::ActorExt* process = smpi_process_remote(simgrid::s4u::Actor::by_pid(dst_));
 468
 469     simgrid::s4u::MutexPtr mut = process->mailboxes_mutex();
 470     if (smpi_cfg_async_small_thresh() != 0 || (flags_ & MPI_REQ_RMA) != 0)
 471       mut->lock();
 472
 473     bool is_probe = ((flags_ & MPI_REQ_PROBE) != 0);
 474     flags_ |= MPI_REQ_PROBE;
 475
 476     if (smpi_cfg_async_small_thresh() == 0 && (flags_ & MPI_REQ_RMA) == 0) {
 477       mailbox = process->mailbox();
 478     } else if (((flags_ & MPI_REQ_RMA) != 0) || static_cast<int>(size_) < smpi_cfg_async_small_thresh()) {
 479       //We have to check both mailboxes (because SSEND messages are sent to the large mbox).
 480       //begin with the more appropriate one : the small one.
 481       mailbox = process->mailbox_small();
 482       XBT_DEBUG("Is there a corresponding send already posted in the small mailbox %s (in case of SSEND)?",
 483                 mailbox->get_cname());
 484       simgrid::kernel::activity::ActivityImplPtr action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
 485
 486       if (action == nullptr) {
 487         mailbox = process->mailbox();
 488         XBT_DEBUG("No, nothing in the small mailbox test the other one : %s", mailbox->get_cname());
 489         action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
 490         if (action == nullptr) {
 491           XBT_DEBUG("Still nothing, switch back to the small mailbox : %s", mailbox->get_cname());
 492           mailbox = process->mailbox_small();
 493         }
 494       } else {
 495         XBT_DEBUG("yes there was something for us in the small mailbox");
 496       }
 497     } else {
 498       mailbox = process->mailbox_small();
 499       XBT_DEBUG("Is there a corresponding send already posted the small mailbox?");
 500       simgrid::kernel::activity::ActivityImplPtr action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
 501
 502       if (action == nullptr) {
 503         XBT_DEBUG("No, nothing in the permanent receive mailbox");
 504         mailbox = process->mailbox();
 505       } else {
 506         XBT_DEBUG("yes there was something for us in the small mailbox");
 507       }
 508     }
 509     if(!is_probe)
 510       flags_ &= ~MPI_REQ_PROBE;
 511     kernel::actor::CommIrecvSimcall observer{process->get_actor()->get_impl(),
 512                                              mailbox->get_impl(),
 513                                              static_cast<unsigned char*>(buf_),
 514                                              &real_size_,
 515                                              &match_recv,
 516                                              process->replaying() ? &smpi_comm_null_copy_buffer_callback
 517                                                                   : smpi_comm_copy_data_callback,
 518                                              this,
 519                                              -1.0};
 520     observer.set_tag(tag_);
 521
 522     action_ = kernel::actor::simcall_answered([&observer] { return kernel::activity::CommImpl::irecv(&observer); },
 523                                               &observer);
 524
 525     XBT_DEBUG("recv simcall posted");
 526
 527     if (smpi_cfg_async_small_thresh() != 0 || (flags_ & MPI_REQ_RMA) != 0)
 528       mut->unlock();
 529   } else { /* the RECV flag was not set, so this is a send */
 530     const simgrid::smpi::ActorExt* process = smpi_process_remote(simgrid::s4u::Actor::by_pid(dst_));
 531     xbt_assert(process, "Actor pid=%ld is gone??", dst_);
 532     if (TRACE_smpi_view_internals())
 533       TRACE_smpi_send(src_, src_, dst_, tag_, size_);
 534     this->print_request("New send");
 535
 536     message_id_=comm_->get_sent_messages_count(comm_->group()->rank(src_), comm_->group()->rank(dst_), tag_);
 537     comm_->increment_sent_messages_count(comm_->group()->rank(src_), comm_->group()->rank(dst_), tag_);
 538
 539     void* buf = buf_;
 540     if ((flags_ & MPI_REQ_SSEND) == 0 &&
 541         ((flags_ & MPI_REQ_RMA) != 0 || (flags_ & MPI_REQ_BSEND) != 0 ||
 542          static_cast<int>(size_) < smpi_cfg_detached_send_thresh())) {
 543       void *oldbuf = nullptr;
 544       detached_    = true;
 545       XBT_DEBUG("Send request %p is detached", this);
 546       this->ref();
 547       if (not(type_->flags() & DT_FLAG_DERIVED)) {
 548         oldbuf = buf_;
 549         if (not process->replaying() && oldbuf != nullptr && size_ != 0) {
 550           if (smpi_switch_data_segment(simgrid::s4u::Actor::by_pid(src_), buf_))
 551             XBT_DEBUG("Privatization : We are sending from a zone inside global memory. Switch data segment ");
 552
 553           //we need this temporary buffer even for bsend, as it will be released in the copy callback and we don't have a way to differentiate it
 554           //so actually ... don't use manually attached buffer space.
 555           buf = xbt_malloc(size_);
 556           memcpy(buf,oldbuf,size_);
 557           XBT_DEBUG("buf %p copied into %p",oldbuf,buf);
 558         }
 559       }
 560     }
 561
 562     //if we are giving back the control to the user without waiting for completion, we have to inject timings
 563     double sleeptime = 0.0;
 564     if (detached_ || ((flags_ & (MPI_REQ_ISEND | MPI_REQ_SSEND)) != 0)) { // issend should be treated as isend
 565       // isend and send timings may be different
 566       sleeptime =
 567           ((flags_ & MPI_REQ_ISEND) != 0)
 568               ? simgrid::s4u::Actor::self()->get_host()->extension<simgrid::smpi::Host>()->oisend(
 569                     size_, simgrid::s4u::Actor::by_pid(src_)->get_host(), simgrid::s4u::Actor::by_pid(dst_)->get_host())
 570               : simgrid::s4u::Actor::self()->get_host()->extension<simgrid::smpi::Host>()->osend(
 571                     size_, simgrid::s4u::Actor::by_pid(src_)->get_host(),
 572                     simgrid::s4u::Actor::by_pid(dst_)->get_host());
 573     }
 574
 575     if(sleeptime > 0.0){
 576       simgrid::s4u::this_actor::sleep_for(sleeptime);
 577       XBT_DEBUG("sending size of %zu : sleep %f ", size_, sleeptime);
 578     }
 579
 580     simgrid::s4u::MutexPtr mut = process->mailboxes_mutex();
 581
 582     if (smpi_cfg_async_small_thresh() != 0 || (flags_ & MPI_REQ_RMA) != 0)
 583       mut->lock();
 584
 585     if (not(smpi_cfg_async_small_thresh() != 0 || (flags_ & MPI_REQ_RMA) != 0)) {
 586       mailbox = process->mailbox();
 587     } else if (((flags_ & MPI_REQ_RMA) != 0) || static_cast<int>(size_) < smpi_cfg_async_small_thresh()) { // eager mode
 588       bool is_probe = ((flags_ & MPI_REQ_PROBE) != 0);
 589       flags_ |= MPI_REQ_PROBE;
 590
 591       mailbox = process->mailbox();
 592       XBT_DEBUG("Is there a corresponding recv already posted in the large mailbox %s?", mailbox->get_cname());
 593       if (not mailbox->iprobe(1, &match_send, static_cast<void*>(this))) {
 594         if ((flags_ & MPI_REQ_SSEND) == 0) {
 595           mailbox = process->mailbox_small();
 596           XBT_DEBUG("No, nothing in the large mailbox, message is to be sent on the small one %s",
 597                     mailbox->get_cname());
 598         } else {
 599           mailbox = process->mailbox_small();
 600           XBT_DEBUG("SSEND : Is there a corresponding recv already posted in the small mailbox %s?",
 601                     mailbox->get_cname());
 602           if (not mailbox->iprobe(1, &match_send, static_cast<void*>(this))) {
 603             XBT_DEBUG("No, we are first, send to large mailbox");
 604             mailbox = process->mailbox();
 605           }
 606         }
 607       } else {
 608         XBT_DEBUG("Yes there was something for us in the large mailbox");
 609       }
 610       if(!is_probe)
 611         flags_ &= ~MPI_REQ_PROBE;
 612     } else {
 613       mailbox = process->mailbox();
 614       XBT_DEBUG("Send request %p is in the large mailbox %s (buf: %p)", this, mailbox->get_cname(), buf_);
 615     }
 616
 617     size_t payload_size_ = size_ + 16;//MPI enveloppe size (tag+dest+communicator)
 618     kernel::actor::CommIsendSimcall observer{
 619         simgrid::kernel::EngineImpl::get_instance()->get_actor_by_pid(src_), mailbox->get_impl(),
 620         static_cast<double>(payload_size_), -1, static_cast<unsigned char*>(buf), real_size_, &match_send,
 621         &xbt_free_f, // how to free the userdata if a detached send fails
 622         process->replaying() ? &smpi_comm_null_copy_buffer_callback : smpi_comm_copy_data_callback, this,
 623         // detach if msg size < eager/rdv switch limit
 624         detached_};
 625     observer.set_tag(tag_);
 626     action_ = kernel::actor::simcall_answered([&observer] { return kernel::activity::CommImpl::isend(&observer); },
 627                                               &observer);
 628     XBT_DEBUG("send simcall posted");
 629
 630     /* FIXME: detached sends are not traceable (action_ == nullptr) */
 631     if (action_ != nullptr) {
 632       boost::static_pointer_cast<kernel::activity::CommImpl>(action_)->set_tracing_category(
 633           smpi_process()->get_tracing_category());
 634     }
 635
 636     if (smpi_cfg_async_small_thresh() != 0 || ((flags_ & MPI_REQ_RMA) != 0))
 637       mut->unlock();
 638   }
 639 }
 640
 641 void Request::startall(int count, MPI_Request * requests)
 642 {
 643   if(requests== nullptr)
 644     return;
 645
 646   for(int i = 0; i < count; i++) {
 647     if(requests[i]->src_ != MPI_PROC_NULL && requests[i]->dst_ != MPI_PROC_NULL)
 648       requests[i]->start();
 649   }
 650 }
 651
 652 void Request::cancel()
 653 {
 654   this->flags_ |= MPI_REQ_CANCELLED;
 655   if (this->action_ != nullptr)
 656     (boost::static_pointer_cast<simgrid::kernel::activity::CommImpl>(this->action_))->cancel();
 657 }
 658
 659 int Request::test(MPI_Request * request, MPI_Status * status, int* flag) {
 660   // assume that *request is not MPI_REQUEST_NULL (filtered in PMPI_Test or testall before)
 661   // to avoid deadlocks if used as a break condition, such as
 662   //     while (MPI_Test(request, flag, status) && flag) dostuff...
 663   // because the time will not normally advance when only calls to MPI_Test are made -> deadlock
 664   // multiplier to the sleeptime, to increase speed of execution, each failed test will increase it
 665   xbt_assert(*request != MPI_REQUEST_NULL);
 666
 667   static int nsleeps = 1;
 668   int ret = MPI_SUCCESS;
 669
 670   if(smpi_test_sleep > 0)
 671     simgrid::s4u::this_actor::sleep_for(nsleeps * smpi_test_sleep);
 672
 673   Status::empty(status);
 674   *flag = 1;
 675
 676   if ((*request)->flags_ & MPI_REQ_NBC){
 677     *flag = finish_nbc_requests(request, 1);
 678   }
 679
 680   if (((*request)->flags_ & (MPI_REQ_PREPARED | MPI_REQ_FINISHED)) == 0) {
 681     if ((*request)->action_ != nullptr && ((*request)->flags_ & MPI_REQ_CANCELLED) == 0){
 682       try{
 683         kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self();
 684         kernel::actor::ActivityTestSimcall observer{issuer, (*request)->action_.get()};
 685         *flag = kernel::actor::simcall_answered(
 686             [&observer] { return observer.get_activity()->test(observer.get_issuer()); }, &observer);
 687       } catch (const Exception&) {
 688         *flag = 0;
 689         return ret;
 690       }
 691     }
 692     if (((*request)->flags_ & MPI_REQ_GENERALIZED) && not((*request)->flags_ & MPI_REQ_COMPLETE))
 693       *flag=0;
 694     if (*flag) {
 695       finish_wait(request, status); // may invalidate *request
 696       if (*request != MPI_REQUEST_NULL && ((*request)->flags_ & MPI_REQ_GENERALIZED)){
 697         MPI_Status tmp_status;
 698         MPI_Status* mystatus;
 699         if (status == MPI_STATUS_IGNORE) {
 700           mystatus = &tmp_status;
 701           Status::empty(mystatus);
 702         } else {
 703           mystatus = status;
 704         }
 705         ret = ((*request)->generalized_funcs)->query_fn(((*request)->generalized_funcs)->extra_state, mystatus);
 706       }
 707       nsleeps=1;//reset the number of sleeps we will do next time
 708       if (*request != MPI_REQUEST_NULL && ((*request)->flags_ & MPI_REQ_PERSISTENT) == 0)
 709         *request = MPI_REQUEST_NULL;
 710     } else if (smpi_cfg_grow_injected_times()) {
 711       nsleeps++;
 712     }
 713   }
 714   return ret;
 715 }
 716
 717 int Request::testsome(int incount, MPI_Request requests[], int *count, int *indices, MPI_Status status[])
 718 {
 719   int error=0;
 720   int count_dead = 0;
 721   int flag = 0;
 722   MPI_Status stat;
 723   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 724
 725   *count = 0;
 726   for (int i = 0; i < incount; i++) {
 727     if (requests[i] != MPI_REQUEST_NULL && not (requests[i]->flags_ & MPI_REQ_FINISHED)) {
 728       if (test(&requests[i], pstat, &flag) != MPI_SUCCESS)
 729         error = 1;
 730       if(flag) {
 731         indices[*count] = i;
 732         if (status != MPI_STATUSES_IGNORE)
 733           status[*count] = *pstat;
 734         (*count)++;
 735         if ((requests[i] != MPI_REQUEST_NULL) && (requests[i]->flags_ & MPI_REQ_NON_PERSISTENT))
 736           requests[i] = MPI_REQUEST_NULL;
 737       }
 738     } else {
 739       count_dead++;
 740     }
 741   }
 742   if(count_dead==incount)*count=MPI_UNDEFINED;
 743   if(error!=0)
 744     return MPI_ERR_IN_STATUS;
 745   else
 746     return MPI_SUCCESS;
 747 }
 748
 749 int Request::testany(int count, MPI_Request requests[], int *index, int* flag, MPI_Status * status)
 750 {
 751   std::vector<simgrid::kernel::activity::ActivityImpl*> comms;
 752   comms.reserve(count);
 753
 754   *flag = 0;
 755   int ret = MPI_SUCCESS;
 756   *index = MPI_UNDEFINED;
 757
 758   std::vector<int> map; /** Maps all matching comms back to their location in requests **/
 759   for (int i = 0; i < count; i++) {
 760     if ((requests[i] != MPI_REQUEST_NULL) && requests[i]->action_ && not(requests[i]->flags_ & MPI_REQ_PREPARED)) {
 761       comms.push_back(requests[i]->action_.get());
 762       map.push_back(i);
 763     }
 764   }
 765   if (not map.empty()) {
 766     //multiplier to the sleeptime, to increase speed of execution, each failed testany will increase it
 767     static int nsleeps = 1;
 768     if(smpi_test_sleep > 0)
 769       simgrid::s4u::this_actor::sleep_for(nsleeps * smpi_test_sleep);
 770     ssize_t i;
 771     try{
 772       kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self();
 773       kernel::actor::ActivityTestanySimcall observer{issuer, comms};
 774       i = kernel::actor::simcall_answered(
 775           [&observer] {
 776             return kernel::activity::ActivityImpl::test_any(observer.get_issuer(), observer.get_activities());
 777           },
 778           &observer);
 779     } catch (const Exception&) {
 780       XBT_DEBUG("Exception in testany");
 781       return 0;
 782     }
 783
 784     if (i != -1) { // -1 is not MPI_UNDEFINED but a SIMIX return code. (nothing matches)
 785       *index = map[i];
 786       if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & MPI_REQ_GENERALIZED) &&
 787           not(requests[*index]->flags_ & MPI_REQ_COMPLETE)) {
 788         *flag=0;
 789       } else {
 790         finish_wait(&requests[*index],status);
 791       if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & MPI_REQ_GENERALIZED)){
 792         MPI_Status tmp_status;
 793         MPI_Status* mystatus;
 794         if (status == MPI_STATUS_IGNORE) {
 795           mystatus = &tmp_status;
 796           Status::empty(mystatus);
 797         } else {
 798           mystatus = status;
 799         }
 800         ret=(requests[*index]->generalized_funcs)->query_fn((requests[*index]->generalized_funcs)->extra_state, mystatus);
 801       }
 802
 803       if (requests[*index] != MPI_REQUEST_NULL && requests[*index]->flags_ & MPI_REQ_NBC){
 804         *flag = finish_nbc_requests(&requests[*index] , 1);
 805       }
 806
 807       if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & MPI_REQ_NON_PERSISTENT))
 808           requests[*index] = MPI_REQUEST_NULL;
 809         XBT_DEBUG("Testany - returning with index %d", *index);
 810         *flag=1;
 811       }
 812       nsleeps = 1;
 813     } else {
 814       nsleeps++;
 815     }
 816   } else {
 817       XBT_DEBUG("Testany on inactive handles, returning flag=1 but empty status");
 818       //all requests are null or inactive, return true
 819       *flag = 1;
 820       *index = MPI_UNDEFINED;
 821       Status::empty(status);
 822   }
 823
 824   return ret;
 825 }
 826
 827 int Request::testall(int count, MPI_Request requests[], int* outflag, MPI_Status status[])
 828 {
 829   MPI_Status stat;
 830   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 831   int flag;
 832   int error = 0;
 833   *outflag = 1;
 834   for(int i=0; i<count; i++){
 835     if (requests[i] != MPI_REQUEST_NULL && not(requests[i]->flags_ & MPI_REQ_PREPARED)) {
 836       int ret = test(&requests[i], pstat, &flag);
 837       if (flag){
 838         flag=0;
 839       }else{
 840         *outflag=0;
 841       }
 842       if (ret != MPI_SUCCESS)
 843         error = 1;
 844     }else{
 845       Status::empty(pstat);
 846     }
 847     if(status != MPI_STATUSES_IGNORE) {
 848       status[i] = *pstat;
 849     }
 850   }
 851   if (error == 1)
 852     return MPI_ERR_IN_STATUS;
 853   else
 854     return MPI_SUCCESS;
 855 }
 856
 857 void Request::probe(int source, int tag, MPI_Comm comm, MPI_Status* status){
 858   int flag=0;
 859   //FIXME find another way to avoid busy waiting ?
 860   // the issue here is that we have to wait on a nonexistent comm
 861   while(flag==0){
 862     iprobe(source, tag, comm, &flag, status);
 863     XBT_DEBUG("Busy Waiting on probing : %d", flag);
 864   }
 865 }
 866
 867 void Request::iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status* status){
 868   // to avoid deadlock, we have to sleep some time here, or the timer won't advance and we will only do iprobe simcalls
 869   // especially when used as a break condition, such as while (MPI_Iprobe(...)) dostuff...
 870   // nsleeps is a multiplier to the sleeptime, to increase speed of execution, each failed iprobe will increase it
 871   // This can speed up the execution of certain applications by an order of magnitude, such as HPL
 872   static int nsleeps = 1;
 873   double speed        = s4u::this_actor::get_host()->get_speed();
 874   double maxrate      = smpi_cfg_iprobe_cpu_usage();
 875   auto request =
 876       new Request(nullptr, 0, MPI_CHAR, source == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : comm->group()->actor(source),
 877                   simgrid::s4u::this_actor::get_pid(), tag, comm, MPI_REQ_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PROBE);
 878   if (smpi_iprobe_sleep > 0) {
 879     /** Compute the number of flops we will sleep **/
 880     s4u::this_actor::exec_init(/*nsleeps: See comment above */ nsleeps *
 881                                /*(seconds * flop/s -> total flops)*/ smpi_iprobe_sleep * speed * maxrate)
 882         ->set_name("iprobe")
 883         /* Not the entire CPU can be used when iprobing: This is important for
 884          * the energy consumption caused by polling with iprobes.
 885          * Note also that the number of flops that was
 886          * computed above contains a maxrate factor and is hence reduced (maxrate < 1)
 887          */
 888         ->set_bound(maxrate * speed)
 889         ->start()
 890         ->wait();
 891   }
 892   // behave like a receive, but don't do it
 893   s4u::Mailbox* mailbox;
 894
 895   request->print_request("New iprobe");
 896   // We have to test both mailboxes as we don't know if we will receive one or another
 897   if (smpi_cfg_async_small_thresh() > 0) {
 898     mailbox = smpi_process()->mailbox_small();
 899     XBT_DEBUG("Trying to probe the perm recv mailbox");
 900     request->action_ = mailbox->iprobe(0, &match_recv, static_cast<void*>(request));
 901   }
 902
 903   if (request->action_ == nullptr){
 904     mailbox = smpi_process()->mailbox();
 905     XBT_DEBUG("trying to probe the other mailbox");
 906     request->action_ = mailbox->iprobe(0, &match_recv, static_cast<void*>(request));
 907   }
 908
 909   if (request->action_ != nullptr){
 910     kernel::activity::CommImplPtr sync_comm = boost::static_pointer_cast<kernel::activity::CommImpl>(request->action_);
 911     const Request* req                      = static_cast<MPI_Request>(sync_comm->src_data_);
 912     *flag = 1;
 913     if (status != MPI_STATUS_IGNORE && (req->flags_ & MPI_REQ_PREPARED) == 0) {
 914       status->MPI_SOURCE = comm->group()->rank(req->src_);
 915       status->MPI_TAG    = req->tag_;
 916       status->MPI_ERROR  = MPI_SUCCESS;
 917       status->count      = req->real_size_;
 918     }
 919     nsleeps = 1;//reset the number of sleeps we will do next time
 920   }
 921   else {
 922     *flag = 0;
 923     if (smpi_cfg_grow_injected_times())
 924       nsleeps++;
 925   }
 926   unref(&request);
 927   xbt_assert(request == MPI_REQUEST_NULL);
 928 }
 929
 930 int Request::finish_nbc_requests(MPI_Request* request, int test){
 931   int flag = 1;
 932   int ret = 0;
 933   if(test == 0)
 934     ret = waitall((*request)->nbc_requests_.size(), (*request)->nbc_requests_.data(), MPI_STATUSES_IGNORE);
 935   else{
 936     ret = testall((*request)->nbc_requests_.size(), (*request)->nbc_requests_.data(), &flag, MPI_STATUSES_IGNORE);
 937   }
 938   if(ret!=MPI_SUCCESS)
 939     xbt_die("Failure when waiting on non blocking collective sub-requests");
 940   if(flag == 1){
 941     XBT_DEBUG("Finishing non blocking collective request with %zu sub-requests", (*request)->nbc_requests_.size());
 942     for(auto& req: (*request)->nbc_requests_){
 943       if((*request)->buf_!=nullptr && req!=MPI_REQUEST_NULL){//reduce case
 944         void * buf=req->buf_;
 945         if((*request)->type_->flags() & DT_FLAG_DERIVED)
 946           buf=req->old_buf_;
 947         if(req->flags_ & MPI_REQ_RECV ){
 948           if((*request)->op_!=MPI_OP_NULL){
 949             int count=(*request)->size_/ (*request)->type_->size();
 950             (*request)->op_->apply(buf, (*request)->buf_, &count, (*request)->type_);
 951           }
 952           smpi_free_tmp_buffer(static_cast<unsigned char*>(buf));
 953         }
 954       }
 955       if(req!=MPI_REQUEST_NULL)
 956         Request::unref(&req);
 957     }
 958     (*request)->nbc_requests_.clear();
 959   }
 960   return flag;
 961 }
 962
 963 void Request::finish_wait(MPI_Request* request, MPI_Status * status)
 964 {
 965   MPI_Request req = *request;
 966   Status::empty(status);
 967   if((req->flags_ & MPI_REQ_CANCELLED) != 0 && (req->flags_ & MPI_REQ_MATCHED) == 0) {
 968     if (status!=MPI_STATUS_IGNORE)
 969       status->cancelled=1;
 970     if(req->detached_sender_ != nullptr)
 971       unref(&(req->detached_sender_));
 972     unref(request);
 973     return;
 974   }
 975
 976   if ((req->flags_ & (MPI_REQ_PREPARED | MPI_REQ_GENERALIZED | MPI_REQ_FINISHED)) == 0) {
 977     if (status != MPI_STATUS_IGNORE) {
 978       if (req->src_== MPI_PROC_NULL || req->dst_== MPI_PROC_NULL){
 979         Status::empty(status);
 980         status->MPI_SOURCE = MPI_PROC_NULL;
 981       } else {
 982         aid_t src          = req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_;
 983         status->MPI_SOURCE = req->comm_->group()->rank(src);
 984         status->MPI_TAG = req->tag_ == MPI_ANY_TAG ? req->real_tag_ : req->tag_;
 985         status->MPI_ERROR  = req->truncated_ ? MPI_ERR_TRUNCATE : MPI_SUCCESS;
 986       }
 987       // this handles the case were size in receive differs from size in send
 988       status->count = req->real_size_;
 989     }
 990     //detached send will be finished at the other end
 991     if (not(req->detached_ && ((req->flags_ & MPI_REQ_SEND) != 0))) {
 992       req->print_request("Finishing");
 993       MPI_Datatype datatype = req->type_;
 994
 995       // FIXME Handle the case of a partial shared malloc.
 996       if (not smpi_process()->replaying() &&
 997         (((req->flags_ & MPI_REQ_ACCUMULATE) != 0) || (datatype->flags() & DT_FLAG_DERIVED))) {
 998         if (smpi_switch_data_segment(simgrid::s4u::Actor::self(), req->old_buf_))
 999           XBT_VERB("Privatization : We are unserializing to a zone in global memory  Switch data segment ");
1000
1001         if(datatype->flags() & DT_FLAG_DERIVED){
1002           // This part handles the problem of non-contiguous memory the unserialization at the reception
1003           if ((req->flags_ & MPI_REQ_RECV) && datatype->size() != 0)
1004             datatype->unserialize(req->buf_, req->old_buf_, req->real_size_/datatype->size() , req->op_);
1005           xbt_free(req->buf_);
1006           req->buf_=nullptr;
1007         } else if (req->flags_ & MPI_REQ_RECV) { // apply op on contiguous buffer for accumulate
1008           if (datatype->size() != 0) {
1009             int n = req->real_size_ / datatype->size();
1010             req->op_->apply(req->buf_, req->old_buf_, &n, datatype);
1011           }
1012           xbt_free(req->buf_);
1013           req->buf_=nullptr;
1014         }
1015       }
1016     }
1017   }
1018
1019   if (TRACE_smpi_view_internals() && ((req->flags_ & MPI_REQ_RECV) != 0)) {
1020     aid_t rank       = simgrid::s4u::this_actor::get_pid();
1021     aid_t src_traced = (req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_);
1022     TRACE_smpi_recv(src_traced, rank,req->tag_);
1023   }
1024   if(req->detached_sender_ != nullptr){
1025     //integrate pseudo-timing for buffering of small messages, do not bother to execute the simcall if 0
1026     simgrid::s4u::Host* dst_host = simgrid::s4u::Actor::by_pid(req->dst_)->get_host();
1027     if (double sleeptime = simgrid::s4u::Actor::self()->get_host()->extension<simgrid::smpi::Host>()->orecv(
1028             req->real_size(), req->src_host_, dst_host);
1029         sleeptime > 0.0) {
1030       simgrid::s4u::this_actor::sleep_for(sleeptime);
1031       XBT_DEBUG("receiving size of %zu : sleep %f ", req->real_size_, sleeptime);
1032     }
1033     unref(&(req->detached_sender_));
1034   }
1035   if (req->flags_ & MPI_REQ_PERSISTENT)
1036     req->action_ = nullptr;
1037   req->flags_ |= MPI_REQ_FINISHED;
1038
1039   if (req->truncated_ || req->unmatched_types_) {
1040     char error_string[MPI_MAX_ERROR_STRING];
1041     int error_size;
1042     int errkind;
1043     if(req->truncated_ )
1044       errkind = MPI_ERR_TRUNCATE;
1045     else
1046       errkind = MPI_ERR_TYPE;
1047     PMPI_Error_string(errkind, error_string, &error_size);
1048     MPI_Errhandler err = (req->comm_) ? (req->comm_)->errhandler() : MPI_ERRHANDLER_NULL;
1049     if (err == MPI_ERRHANDLER_NULL || err == MPI_ERRORS_RETURN)
1050       XBT_WARN("recv - returned %.*s instead of MPI_SUCCESS", error_size, error_string);
1051     else if (err == MPI_ERRORS_ARE_FATAL)
1052       xbt_die("recv - returned %.*s instead of MPI_SUCCESS", error_size, error_string);
1053     else
1054       err->call((req->comm_), errkind);
1055     if (err != MPI_ERRHANDLER_NULL)
1056       simgrid::smpi::Errhandler::unref(err);
1057     MC_assert(not MC_is_active()); /* Only fail in MC mode */
1058   }
1059   if(req->src_ != MPI_PROC_NULL && req->dst_ != MPI_PROC_NULL)
1060     unref(request);
1061 }
1062
1063 int Request::wait(MPI_Request * request, MPI_Status * status)
1064 {
1065   // assume that *request is not MPI_REQUEST_NULL (filtered in PMPI_Wait before)
1066   xbt_assert(*request != MPI_REQUEST_NULL);
1067
1068   int ret=MPI_SUCCESS;
1069
1070   if((*request)->src_ == MPI_PROC_NULL || (*request)->dst_ == MPI_PROC_NULL){
1071     if (status != MPI_STATUS_IGNORE) {
1072       Status::empty(status);
1073       status->MPI_SOURCE = MPI_PROC_NULL;
1074     }
1075     (*request)=MPI_REQUEST_NULL;
1076     return ret;
1077   }
1078
1079   (*request)->print_request("Waiting");
1080   if ((*request)->flags_ & (MPI_REQ_PREPARED | MPI_REQ_FINISHED)) {
1081     Status::empty(status);
1082     return ret;
1083   }
1084
1085   if ((*request)->action_ != nullptr){
1086       try{
1087         // this is not a detached send
1088         kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self();
1089         kernel::actor::ActivityWaitSimcall observer{issuer, (*request)->action_.get(), -1};
1090         kernel::actor::simcall_blocking([issuer, &observer] { observer.get_activity()->wait_for(issuer, -1); },
1091                                         &observer);
1092       } catch (const CancelException&) {
1093         XBT_VERB("Request cancelled");
1094       }
1095   }
1096
1097   if ((*request)->flags_ & MPI_REQ_GENERALIZED) {
1098     if (not((*request)->flags_ & MPI_REQ_COMPLETE)) {
1099       ((*request)->generalized_funcs)->mutex->lock();
1100       ((*request)->generalized_funcs)->cond->wait(((*request)->generalized_funcs)->mutex);
1101       ((*request)->generalized_funcs)->mutex->unlock();
1102     }
1103     MPI_Status tmp_status;
1104     MPI_Status* mystatus;
1105     if (status == MPI_STATUS_IGNORE) {
1106       mystatus = &tmp_status;
1107       Status::empty(mystatus);
1108     } else {
1109       mystatus = status;
1110     }
1111     ret = ((*request)->generalized_funcs)->query_fn(((*request)->generalized_funcs)->extra_state, mystatus);
1112   }
1113
1114   if ((*request)->truncated_)
1115     ret = MPI_ERR_TRUNCATE;
1116
1117   if ((*request)->flags_ & MPI_REQ_NBC)
1118     finish_nbc_requests(request, 0);
1119
1120   finish_wait(request, status); // may invalidate *request
1121   if (*request != MPI_REQUEST_NULL && (((*request)->flags_ & MPI_REQ_NON_PERSISTENT) != 0))
1122     *request = MPI_REQUEST_NULL;
1123   return ret;
1124 }
1125
1126 int Request::waitany(int count, MPI_Request requests[], MPI_Status * status)
1127 {
1128   int index = MPI_UNDEFINED;
1129
1130   if(count > 0) {
1131     // Wait for a request to complete
1132     std::vector<simgrid::kernel::activity::ActivityImpl*> comms;
1133     std::vector<int> map;
1134     XBT_DEBUG("Wait for one of %d", count);
1135     for(int i = 0; i < count; i++) {
1136       if (requests[i] != MPI_REQUEST_NULL && not(requests[i]->flags_ & MPI_REQ_PREPARED) &&
1137           not(requests[i]->flags_ & MPI_REQ_FINISHED)) {
1138         if (requests[i]->action_ != nullptr) {
1139           XBT_DEBUG("Waiting any %p ", requests[i]);
1140           comms.push_back(requests[i]->action_.get());
1141           map.push_back(i);
1142         } else {
1143           // This is a finished detached request, let's return this one
1144           comms.clear(); // don't do the waitany call afterwards
1145           index = i;
1146           if (requests[index]->flags_ & MPI_REQ_NBC)
1147             finish_nbc_requests(&requests[index], 0);
1148           finish_wait(&requests[i], status); // cleanup if refcount = 0
1149           if (requests[i] != MPI_REQUEST_NULL && (requests[i]->flags_ & MPI_REQ_NON_PERSISTENT))
1150             requests[i] = MPI_REQUEST_NULL; // set to null
1151           break;
1152         }
1153       }
1154     }
1155     if (not comms.empty()) {
1156       XBT_DEBUG("Enter waitany for %zu comms", comms.size());
1157       ssize_t i;
1158       try{
1159         kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self();
1160         kernel::actor::ActivityWaitanySimcall observer{issuer, comms, -1};
1161         i = kernel::actor::simcall_blocking(
1162             [&observer] {
1163               kernel::activity::ActivityImpl::wait_any_for(observer.get_issuer(), observer.get_activities(),
1164                                                            observer.get_timeout());
1165             },
1166             &observer);
1167       } catch (const CancelException&) {
1168         XBT_INFO("request cancelled");
1169         i = -1;
1170       }
1171
1172       // not MPI_UNDEFINED, as this is a simix return code
1173       if (i != -1) {
1174         index = map[i];
1175         //in case of an accumulate, we have to wait the end of all requests to apply the operation, ordered correctly.
1176         if ((requests[index] == MPI_REQUEST_NULL) ||
1177             (not((requests[index]->flags_ & MPI_REQ_ACCUMULATE) && (requests[index]->flags_ & MPI_REQ_RECV)))) {
1178           finish_wait(&requests[index],status);
1179           if (requests[index] != MPI_REQUEST_NULL && (requests[index]->flags_ & MPI_REQ_NON_PERSISTENT))
1180             requests[index] = MPI_REQUEST_NULL;
1181         }
1182       }
1183     }
1184   }
1185
1186
1187   if (index==MPI_UNDEFINED)
1188     Status::empty(status);
1189
1190   return index;
1191 }
1192
1193 static int sort_accumulates(const Request* a, const Request* b)
1194 {
1195   return (a->tag() > b->tag());
1196 }
1197
1198 int Request::waitall(int count, MPI_Request requests[], MPI_Status status[])
1199 {
1200   std::vector<MPI_Request> accumulates;
1201   int index;
1202   MPI_Status stat;
1203   MPI_Status *pstat = (status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat);
1204   int retvalue = MPI_SUCCESS;
1205   //tag invalid requests in the set
1206   if (status != MPI_STATUSES_IGNORE) {
1207     for (int c = 0; c < count; c++) {
1208       if (requests[c] == MPI_REQUEST_NULL || requests[c]->dst_ == MPI_PROC_NULL ||
1209           (requests[c]->flags_ & MPI_REQ_PREPARED)) {
1210         Status::empty(&status[c]);
1211       } else if (requests[c]->src_ == MPI_PROC_NULL) {
1212         Status::empty(&status[c]);
1213         status[c].MPI_SOURCE = MPI_PROC_NULL;
1214       }
1215     }
1216   }
1217   for (int c = 0; c < count; c++) {
1218     if (MC_is_active() || MC_record_replay_is_active()) {
1219       wait(&requests[c],pstat);
1220       index = c;
1221     } else {
1222       index = waitany(count, requests, pstat);
1223
1224       if (index == MPI_UNDEFINED)
1225         break;
1226
1227       if (requests[index] != MPI_REQUEST_NULL && (requests[index]->flags_ & MPI_REQ_RECV) &&
1228           (requests[index]->flags_ & MPI_REQ_ACCUMULATE))
1229         accumulates.push_back(requests[index]);
1230       if (requests[index] != MPI_REQUEST_NULL && (requests[index]->flags_ & MPI_REQ_NON_PERSISTENT))
1231         requests[index] = MPI_REQUEST_NULL;
1232     }
1233     if (status != MPI_STATUSES_IGNORE) {
1234       status[index] = *pstat;
1235       if (status[index].MPI_ERROR == MPI_ERR_TRUNCATE)
1236         retvalue = MPI_ERR_IN_STATUS;
1237     }
1238   }
1239
1240   std::sort(accumulates.begin(), accumulates.end(), sort_accumulates);
1241   for (auto& req : accumulates)
1242     finish_wait(&req, status);
1243
1244   return retvalue;
1245 }
1246
1247 int Request::waitsome(int incount, MPI_Request requests[], int *indices, MPI_Status status[])
1248 {
1249   int count = 0;
1250   int flag = 0;
1251   int index = 0;
1252   MPI_Status stat;
1253   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
1254   index             = waitany(incount, requests, pstat);
1255   if(index==MPI_UNDEFINED) return MPI_UNDEFINED;
1256   if(status != MPI_STATUSES_IGNORE) {
1257     status[count] = *pstat;
1258   }
1259   indices[count] = index;
1260   count++;
1261   for (int i = 0; i < incount; i++) {
1262     if (i != index && requests[i] != MPI_REQUEST_NULL && not(requests[i]->flags_ & MPI_REQ_FINISHED)) {
1263       test(&requests[i], pstat,&flag);
1264       if (flag==1){
1265         indices[count] = i;
1266         if(status != MPI_STATUSES_IGNORE) {
1267           status[count] = *pstat;
1268         }
1269         if (requests[i] != MPI_REQUEST_NULL && (requests[i]->flags_ & MPI_REQ_NON_PERSISTENT))
1270           requests[i]=MPI_REQUEST_NULL;
1271         count++;
1272       }
1273     }
1274   }
1275   return count;
1276 }
1277
1278 MPI_Request Request::f2c(int id)
1279 {
1280   if(id==MPI_FORTRAN_REQUEST_NULL)
1281     return MPI_REQUEST_NULL;
1282   return static_cast<MPI_Request>(F2C::lookup()->at(id));
1283 }
1284
1285 void Request::free_f(int id)
1286 {
1287   if (id != MPI_FORTRAN_REQUEST_NULL) {
1288     F2C::lookup()->erase(id);
1289   }
1290 }
1291
1292 int Request::get_status(const Request* req, int* flag, MPI_Status* status)
1293 {
1294   *flag=0;
1295
1296   if(req != MPI_REQUEST_NULL && req->action_ != nullptr) {
1297     req->iprobe(req->comm_->group()->rank(req->src_), req->tag_, req->comm_, flag, status);
1298     if(*flag)
1299       return MPI_SUCCESS;
1300   }
1301   if (req != MPI_REQUEST_NULL && (req->flags_ & MPI_REQ_GENERALIZED) && not(req->flags_ & MPI_REQ_COMPLETE)) {
1302     *flag = 0;
1303     return MPI_SUCCESS;
1304   }
1305
1306   *flag=1;
1307   if(req != MPI_REQUEST_NULL &&
1308      status != MPI_STATUS_IGNORE) {
1309     aid_t src          = req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_;
1310     status->MPI_SOURCE = req->comm_->group()->rank(src);
1311     status->MPI_TAG = req->tag_ == MPI_ANY_TAG ? req->real_tag_ : req->tag_;
1312     status->MPI_ERROR = req->truncated_ ? MPI_ERR_TRUNCATE : MPI_SUCCESS;
1313     status->count = req->real_size_;
1314   }
1315   return MPI_SUCCESS;
1316 }
1317
1318 int Request::grequest_start(MPI_Grequest_query_function* query_fn, MPI_Grequest_free_function* free_fn,
1319                             MPI_Grequest_cancel_function* cancel_fn, void* extra_state, MPI_Request* request)
1320 {
1321   *request = new Request();
1322   (*request)->flags_ |= MPI_REQ_GENERALIZED;
1323   (*request)->flags_ |= MPI_REQ_PERSISTENT;
1324   (*request)->refcount_ = 1;
1325   ((*request)->generalized_funcs)             = std::make_unique<smpi_mpi_generalized_request_funcs_t>();
1326   ((*request)->generalized_funcs)->query_fn=query_fn;
1327   ((*request)->generalized_funcs)->free_fn=free_fn;
1328   ((*request)->generalized_funcs)->cancel_fn=cancel_fn;
1329   ((*request)->generalized_funcs)->extra_state=extra_state;
1330   ((*request)->generalized_funcs)->cond = simgrid::s4u::ConditionVariable::create();
1331   ((*request)->generalized_funcs)->mutex = simgrid::s4u::Mutex::create();
1332   return MPI_SUCCESS;
1333 }
1334
1335 int Request::grequest_complete(MPI_Request request)
1336 {
1337   if ((not(request->flags_ & MPI_REQ_GENERALIZED)) || request->generalized_funcs->mutex == nullptr)
1338     return MPI_ERR_REQUEST;
1339   request->generalized_funcs->mutex->lock();
1340   request->flags_ |= MPI_REQ_COMPLETE; // in case wait would be called after complete
1341   request->generalized_funcs->cond->notify_one();
1342   request->generalized_funcs->mutex->unlock();
1343   return MPI_SUCCESS;
1344 }
1345
1346 void Request::start_nbc_requests(std::vector<MPI_Request> reqs){
1347   if (not reqs.empty()) {
1348     nbc_requests_ = reqs;
1349     Request::startall(reqs.size(), reqs.data());
1350   }
1351 }
1352
1353 std::vector<MPI_Request> Request::get_nbc_requests() const
1354 {
1355   return nbc_requests_;
1356 }
1357 } // namespace simgrid::smpi