src/smpi/smpi_request.cpp

   1 /* Copyright (c) 2007-2017. 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 <xbt/config.hpp>
   7 #include <algorithm>
   8 #include "private.h"
   9 #include "mc/mc.h"
  10 #include "src/mc/mc_replay.h"
  11 #include "src/simix/smx_private.h"
  12 #include "simgrid/sg_config.h"
  13 #include "smpi/smpi_utils.hpp"
  14 #include <simgrid/s4u/host.hpp>
  15 #include "src/kernel/activity/SynchroComm.hpp"
  16
  17 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_request, smpi, "Logging specific to SMPI (reques)");
  18
  19 static simgrid::config::Flag<double> smpi_iprobe_sleep(
  20   "smpi/iprobe", "Minimum time to inject inside a call to MPI_Iprobe", 1e-4);
  21 static simgrid::config::Flag<double> smpi_test_sleep(
  22   "smpi/test", "Minimum time to inject inside a call to MPI_Test", 1e-4);
  23
  24 std::vector<s_smpi_factor_t> smpi_os_values;
  25 std::vector<s_smpi_factor_t> smpi_or_values;
  26 std::vector<s_smpi_factor_t> smpi_ois_values;
  27
  28 extern void (*smpi_comm_copy_data_callback) (smx_activity_t, void*, size_t);
  29
  30 static double smpi_os(size_t size)
  31 {
  32   if (smpi_os_values.empty()) {
  33     smpi_os_values = parse_factor(xbt_cfg_get_string("smpi/os"));
  34   }
  35   double current=smpi_os_values.empty()?0.0:smpi_os_values[0].values[0]+smpi_os_values[0].values[1]*size;
  36   // Iterate over all the sections that were specified and find the right
  37   // value. (fact.factor represents the interval sizes; we want to find the
  38   // section that has fact.factor <= size and no other such fact.factor <= size)
  39   // Note: parse_factor() (used before) already sorts the vector we iterate over!
  40   for (auto& fact : smpi_os_values) {
  41     if (size <= fact.factor) { // Values already too large, use the previously computed value of current!
  42       XBT_DEBUG("os : %zu <= %zu return %.10f", size, fact.factor, current);
  43       return current;
  44     }else{
  45       // If the next section is too large, the current section must be used.
  46       // Hence, save the cost, as we might have to use it.
  47       current = fact.values[0]+fact.values[1]*size;
  48     }
  49   }
  50   XBT_DEBUG("Searching for smpi/os: %zu is larger than the largest boundary, return %.10f", size, current);
  51
  52   return current;
  53 }
  54
  55 static double smpi_ois(size_t size)
  56 {
  57   if (smpi_ois_values.empty()) {
  58     smpi_ois_values = parse_factor(xbt_cfg_get_string("smpi/ois"));
  59   }
  60   double current=smpi_ois_values.empty()?0.0:smpi_ois_values[0].values[0]+smpi_ois_values[0].values[1]*size;
  61   // Iterate over all the sections that were specified and find the right value. (fact.factor represents the interval
  62   // sizes; we want to find the section that has fact.factor <= size and no other such fact.factor <= size)
  63   // Note: parse_factor() (used before) already sorts the vector we iterate over!
  64   for (auto& fact : smpi_ois_values) {
  65     if (size <= fact.factor) { // Values already too large, use the previously  computed value of current!
  66       XBT_DEBUG("ois : %zu <= %zu return %.10f", size, fact.factor, current);
  67       return current;
  68     }else{
  69       // If the next section is too large, the current section must be used.
  70       // Hence, save the cost, as we might have to use it.
  71       current = fact.values[0]+fact.values[1]*size;
  72     }
  73   }
  74   XBT_DEBUG("Searching for smpi/ois: %zu is larger than the largest boundary, return %.10f", size, current);
  75
  76   return current;
  77 }
  78
  79 static double smpi_or(size_t size)
  80 {
  81   if (smpi_or_values.empty()) {
  82     smpi_or_values = parse_factor(xbt_cfg_get_string("smpi/or"));
  83   }
  84
  85   double current=smpi_or_values.empty()?0.0:smpi_or_values.front().values[0]+smpi_or_values.front().values[1]*size;
  86
  87   // Iterate over all the sections that were specified and find the right value. (fact.factor represents the interval
  88   // sizes; we want to find the section that has fact.factor <= size and no other such fact.factor <= size)
  89   // Note: parse_factor() (used before) already sorts the vector we iterate over!
  90   for (auto fact : smpi_or_values) {
  91     if (size <= fact.factor) { // Values already too large, use the previously computed value of current!
  92       XBT_DEBUG("or : %zu <= %zu return %.10f", size, fact.factor, current);
  93       return current;
  94     } else {
  95       // If the next section is too large, the current section must be used.
  96       // Hence, save the cost, as we might have to use it.
  97       current=fact.values[0]+fact.values[1]*size;
  98     }
  99   }
 100   XBT_DEBUG("smpi_or: %zu is larger than largest boundary, return %.10f", size, current);
 101
 102   return current;
 103 }
 104
 105 namespace simgrid{
 106 namespace smpi{
 107
 108 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)
 109 {
 110   void *old_buf = nullptr;
 111   if(((((flags & RECV) != 0) && ((flags & ACCUMULATE) !=0)) || (datatype->flags() & DT_FLAG_DERIVED)) && (!smpi_is_shared(buf_))){
 112     // This part handles the problem of non-contiguous memory
 113     old_buf = buf;
 114     if (count==0){
 115       buf_ = nullptr;
 116     }else {
 117       buf_ = xbt_malloc(count*datatype->size());
 118       if ((datatype->flags() & DT_FLAG_DERIVED) && ((flags & SEND) != 0)) {
 119         datatype->serialize(old_buf, buf_, count);
 120       }
 121     }
 122   }
 123   // This part handles the problem of non-contiguous memory (for the unserialisation at the reception)
 124   old_buf_  = old_buf;
 125   size_ = datatype->size() * count;
 126   datatype->ref();
 127   comm_->ref();
 128   action_          = nullptr;
 129   detached_        = 0;
 130   detached_sender_ = nullptr;
 131   real_src_        = 0;
 132   truncated_       = 0;
 133   real_size_       = 0;
 134   real_tag_        = 0;
 135   if (flags & PERSISTENT)
 136     refcount_ = 1;
 137   else
 138     refcount_ = 0;
 139   op_   = MPI_REPLACE;
 140 }
 141
 142 MPI_Comm Request::comm(){
 143   return comm_;
 144 }
 145
 146 int Request::src(){
 147   return src_;
 148 }
 149
 150 int Request::dst(){
 151   return dst_;
 152 }
 153
 154 int Request::tag(){
 155   return tag_;
 156 }
 157
 158 int Request::flags(){
 159   return flags_;
 160 }
 161
 162 int Request::detached(){
 163   return detached_;
 164 }
 165
 166 size_t Request::size(){
 167   return size_;
 168 }
 169
 170 size_t Request::real_size(){
 171   return real_size_;
 172 }
 173
 174 void Request::unref(MPI_Request* request)
 175 {
 176   if((*request) != MPI_REQUEST_NULL){
 177     (*request)->refcount_--;
 178     if((*request)->refcount_<0) xbt_die("wrong refcount");
 179     if((*request)->refcount_==0){
 180         Datatype::unref((*request)->old_type_);
 181         Comm::unref((*request)->comm_);
 182         (*request)->print_request("Destroying");
 183         delete *request;
 184         *request = MPI_REQUEST_NULL;
 185     }else{
 186       (*request)->print_request("Decrementing");
 187     }
 188   }else{
 189     xbt_die("freeing an already free request");
 190   }
 191 }
 192
 193 int Request::match_recv(void* a, void* b, smx_activity_t ignored) {
 194   MPI_Request ref = static_cast<MPI_Request>(a);
 195   MPI_Request req = static_cast<MPI_Request>(b);
 196   XBT_DEBUG("Trying to match a recv of src %d against %d, tag %d against %d",ref->src_,req->src_, ref->tag_, req->tag_);
 197
 198   xbt_assert(ref, "Cannot match recv against null reference");
 199   xbt_assert(req, "Cannot match recv against null request");
 200   if((ref->src_ == MPI_ANY_SOURCE || req->src_ == ref->src_)
 201     && ((ref->tag_ == MPI_ANY_TAG && req->tag_ >=0) || req->tag_ == ref->tag_)){
 202     //we match, we can transfer some values
 203     if(ref->src_ == MPI_ANY_SOURCE)
 204       ref->real_src_ = req->src_;
 205     if(ref->tag_ == MPI_ANY_TAG)
 206       ref->real_tag_ = req->tag_;
 207     if(ref->real_size_ < req->real_size_)
 208       ref->truncated_ = 1;
 209     if(req->detached_==1)
 210       ref->detached_sender_=req; //tie the sender to the receiver, as it is detached and has to be freed in the receiver
 211     XBT_DEBUG("match succeeded");
 212     return 1;
 213   }else return 0;
 214 }
 215
 216 int Request::match_send(void* a, void* b,smx_activity_t ignored) {
 217   MPI_Request ref = static_cast<MPI_Request>(a);
 218   MPI_Request req = static_cast<MPI_Request>(b);
 219   XBT_DEBUG("Trying to match a send of src %d against %d, tag %d against %d",ref->src_,req->src_, ref->tag_, req->tag_);
 220   xbt_assert(ref, "Cannot match send against null reference");
 221   xbt_assert(req, "Cannot match send against null request");
 222
 223   if((req->src_ == MPI_ANY_SOURCE || req->src_ == ref->src_)
 224       && ((req->tag_ == MPI_ANY_TAG && ref->tag_ >=0)|| req->tag_ == ref->tag_)){
 225     if(req->src_ == MPI_ANY_SOURCE)
 226       req->real_src_ = ref->src_;
 227     if(req->tag_ == MPI_ANY_TAG)
 228       req->real_tag_ = ref->tag_;
 229     if(req->real_size_ < ref->real_size_)
 230       req->truncated_ = 1;
 231     if(ref->detached_==1)
 232       req->detached_sender_=ref; //tie the sender to the receiver, as it is detached and has to be freed in the receiver
 233     XBT_DEBUG("match succeeded");
 234     return 1;
 235   } else
 236     return 0;
 237 }
 238
 239 void Request::print_request(const char *message)
 240 {
 241   XBT_VERB("%s  request %p  [buf = %p, size = %zu, src = %d, dst = %d, tag = %d, flags = %x]",
 242        message, this, buf_, size_, src_, dst_, tag_, flags_);
 243 }
 244
 245
 246 /* factories, to hide the internal flags from the caller */
 247 MPI_Request Request::send_init(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 248 {
 249
 250   return new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype, smpi_process()->index(),
 251                           comm->group()->index(dst), tag, comm, PERSISTENT | SEND | PREPARED);
 252 }
 253
 254 MPI_Request Request::ssend_init(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, smpi_process()->index(),
 257                         comm->group()->index(dst), tag, comm, PERSISTENT | SSEND | SEND | PREPARED);
 258 }
 259
 260 MPI_Request Request::isend_init(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 261 {
 262   return new Request(buf==MPI_BOTTOM ? nullptr : buf , count, datatype, smpi_process()->index(),
 263                           comm->group()->index(dst), tag,comm, PERSISTENT | ISEND | SEND | PREPARED);
 264 }
 265
 266
 267 MPI_Request Request::rma_send_init(void *buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm,
 268                                MPI_Op op)
 269 {
 270   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
 271   if(op==MPI_OP_NULL){
 272     request = new Request(buf==MPI_BOTTOM ? nullptr : buf , count, datatype, src, dst, tag,
 273                             comm, RMA | NON_PERSISTENT | ISEND | SEND | PREPARED);
 274   }else{
 275     request = new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype,  src, dst, tag,
 276                             comm, RMA | NON_PERSISTENT | ISEND | SEND | PREPARED | ACCUMULATE);
 277     request->op_ = op;
 278   }
 279   return request;
 280 }
 281
 282 MPI_Request Request::recv_init(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm)
 283 {
 284   return new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype,
 285                           src == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : comm->group()->index(src),
 286                           smpi_process()->index(), tag, comm, PERSISTENT | RECV | PREPARED);
 287 }
 288
 289 MPI_Request Request::rma_recv_init(void *buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm,
 290                                MPI_Op op)
 291 {
 292   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
 293   if(op==MPI_OP_NULL){
 294     request = new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype,  src, dst, tag,
 295                             comm, RMA | NON_PERSISTENT | RECV | PREPARED);
 296   }else{
 297     request = new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype,  src, dst, tag,
 298                             comm, RMA | NON_PERSISTENT | RECV | PREPARED | ACCUMULATE);
 299     request->op_ = op;
 300   }
 301   return request;
 302 }
 303
 304 MPI_Request Request::irecv_init(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm)
 305 {
 306   return new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype, src == MPI_ANY_SOURCE ? MPI_ANY_SOURCE :
 307                           comm->group()->index(src), smpi_process()->index(), tag,
 308                           comm, PERSISTENT | RECV | PREPARED);
 309 }
 310
 311 MPI_Request Request::isend(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 312 {
 313   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
 314   request =  new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype, smpi_process()->index(),
 315                            comm->group()->index(dst), tag, comm, NON_PERSISTENT | ISEND | SEND);
 316   request->start();
 317   return request;
 318 }
 319
 320 MPI_Request Request::issend(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 321 {
 322   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
 323   request = new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype, smpi_process()->index(),
 324                         comm->group()->index(dst), tag,comm, NON_PERSISTENT | ISEND | SSEND | SEND);
 325   request->start();
 326   return request;
 327 }
 328
 329
 330 MPI_Request Request::irecv(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm)
 331 {
 332   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
 333   request = new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype, src == MPI_ANY_SOURCE ? MPI_ANY_SOURCE :
 334                           comm->group()->index(src), smpi_process()->index(), tag, comm,
 335                           NON_PERSISTENT | RECV);
 336   request->start();
 337   return request;
 338 }
 339
 340 void Request::recv(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status * status)
 341 {
 342   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
 343   request = irecv(buf, count, datatype, src, tag, comm);
 344   wait(&request,status);
 345   request = nullptr;
 346 }
 347
 348 void Request::send(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 349 {
 350   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
 351   request = new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype, smpi_process()->index(),
 352                           comm->group()->index(dst), tag, comm, NON_PERSISTENT | SEND);
 353
 354   request->start();
 355   wait(&request, MPI_STATUS_IGNORE);
 356   request = nullptr;
 357 }
 358
 359 void Request::ssend(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 360 {
 361   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
 362   request = new Request(buf==MPI_BOTTOM ? nullptr : buf, count, datatype, smpi_process()->index(),
 363                           comm->group()->index(dst), tag, comm, NON_PERSISTENT | SSEND | SEND);
 364
 365   request->start();
 366   wait(&request,MPI_STATUS_IGNORE);
 367   request = nullptr;
 368 }
 369
 370 void Request::sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,int dst, int sendtag,
 371                        void *recvbuf, int recvcount, MPI_Datatype recvtype, int src, int recvtag,
 372                        MPI_Comm comm, MPI_Status * status)
 373 {
 374   MPI_Request requests[2];
 375   MPI_Status stats[2];
 376   int myid=smpi_process()->index();
 377   if ((comm->group()->index(dst) == myid) && (comm->group()->index(src) == myid)){
 378       Datatype::copy(sendbuf, sendcount, sendtype, recvbuf, recvcount, recvtype);
 379       return;
 380   }
 381   requests[0] = isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm);
 382   requests[1] = irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm);
 383   startall(2, requests);
 384   waitall(2, requests, stats);
 385   unref(&requests[0]);
 386   unref(&requests[1]);
 387   if(status != MPI_STATUS_IGNORE) {
 388     // Copy receive status
 389     *status = stats[1];
 390   }
 391 }
 392
 393 void Request::start()
 394 {
 395   smx_mailbox_t mailbox;
 396
 397   xbt_assert(action_ == nullptr, "Cannot (re-)start unfinished communication");
 398   flags_ &= ~PREPARED;
 399   flags_ &= ~FINISHED;
 400   refcount_++;
 401
 402   if ((flags_ & RECV) != 0) {
 403     this->print_request("New recv");
 404
 405     simgrid::smpi::Process* process = smpi_process_remote(dst_);
 406
 407     int async_small_thresh = xbt_cfg_get_int("smpi/async-small-thresh");
 408
 409     xbt_mutex_t mut = process->mailboxes_mutex();
 410     if (async_small_thresh != 0 || (flags_ & RMA) != 0)
 411       xbt_mutex_acquire(mut);
 412
 413     if (async_small_thresh == 0 && (flags_ & RMA) == 0 ) {
 414       mailbox = process->mailbox();
 415     }
 416     else if (((flags_ & RMA) != 0) || static_cast<int>(size_) < async_small_thresh) {
 417       //We have to check both mailboxes (because SSEND messages are sent to the large mbox).
 418       //begin with the more appropriate one : the small one.
 419       mailbox = process->mailbox_small();
 420       XBT_DEBUG("Is there a corresponding send already posted in the small mailbox %p (in case of SSEND)?", mailbox);
 421       smx_activity_t action = simcall_comm_iprobe(mailbox, 0, src_,tag_, &match_recv,
 422                                                   static_cast<void*>(this));
 423
 424       if (action == nullptr) {
 425         mailbox = process->mailbox();
 426         XBT_DEBUG("No, nothing in the small mailbox test the other one : %p", mailbox);
 427         action = simcall_comm_iprobe(mailbox, 0, src_,tag_, &match_recv, static_cast<void*>(this));
 428         if (action == nullptr) {
 429           XBT_DEBUG("Still nothing, switch back to the small mailbox : %p", mailbox);
 430           mailbox = process->mailbox_small();
 431         }
 432       } else {
 433         XBT_DEBUG("yes there was something for us in the large mailbox");
 434       }
 435     } else {
 436       mailbox = process->mailbox_small();
 437       XBT_DEBUG("Is there a corresponding send already posted the small mailbox?");
 438       smx_activity_t action = simcall_comm_iprobe(mailbox, 0, src_,tag_, &match_recv, static_cast<void*>(this));
 439
 440       if (action == nullptr) {
 441         XBT_DEBUG("No, nothing in the permanent receive mailbox");
 442         mailbox = process->mailbox();
 443       } else {
 444         XBT_DEBUG("yes there was something for us in the small mailbox");
 445       }
 446     }
 447
 448     // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
 449     real_size_=size_;
 450     action_ = simcall_comm_irecv(process->process(), mailbox, buf_, &real_size_, &match_recv,
 451                                          ! process->replaying()? smpi_comm_copy_data_callback
 452                                          : &smpi_comm_null_copy_buffer_callback, this, -1.0);
 453     XBT_DEBUG("recv simcall posted");
 454
 455     if (async_small_thresh != 0 || (flags_ & RMA) != 0 )
 456       xbt_mutex_release(mut);
 457   } else { /* the RECV flag was not set, so this is a send */
 458     simgrid::smpi::Process* process = smpi_process_remote(dst_);
 459     int rank = src_;
 460     if (TRACE_smpi_view_internals()) {
 461       TRACE_smpi_send(rank, rank, dst_, tag_, size_);
 462     }
 463     this->print_request("New send");
 464
 465     void* buf = buf_;
 466     if ((flags_ & SSEND) == 0 && ( (flags_ & RMA) != 0
 467         || static_cast<int>(size_) < xbt_cfg_get_int("smpi/send-is-detached-thresh") ) ) {
 468       void *oldbuf = nullptr;
 469       detached_ = 1;
 470       XBT_DEBUG("Send request %p is detached", this);
 471       refcount_++;
 472       if(!(old_type_->flags() & DT_FLAG_DERIVED)){
 473         oldbuf = buf_;
 474         if (!process->replaying() && oldbuf != nullptr && size_!=0){
 475           if((smpi_privatize_global_variables != 0)
 476             && (static_cast<char*>(buf_) >= smpi_start_data_exe)
 477             && (static_cast<char*>(buf_) < smpi_start_data_exe + smpi_size_data_exe )){
 478             XBT_DEBUG("Privatization : We are sending from a zone inside global memory. Switch data segment ");
 479             smpi_switch_data_segment(src_);
 480           }
 481           buf = xbt_malloc(size_);
 482           memcpy(buf,oldbuf,size_);
 483           XBT_DEBUG("buf %p copied into %p",oldbuf,buf);
 484         }
 485       }
 486     }
 487
 488     //if we are giving back the control to the user without waiting for completion, we have to inject timings
 489     double sleeptime = 0.0;
 490     if(detached_ != 0 || ((flags_ & (ISEND|SSEND)) != 0)){// issend should be treated as isend
 491       //isend and send timings may be different
 492       sleeptime = ((flags_ & ISEND) != 0) ? smpi_ois(size_) : smpi_os(size_);
 493     }
 494
 495     if(sleeptime > 0.0){
 496       simcall_process_sleep(sleeptime);
 497       XBT_DEBUG("sending size of %zu : sleep %f ", size_, sleeptime);
 498     }
 499
 500     int async_small_thresh = xbt_cfg_get_int("smpi/async-small-thresh");
 501
 502     xbt_mutex_t mut=process->mailboxes_mutex();
 503
 504     if (async_small_thresh != 0 || (flags_ & RMA) != 0)
 505       xbt_mutex_acquire(mut);
 506
 507     if (!(async_small_thresh != 0 || (flags_ & RMA) !=0)) {
 508       mailbox = process->mailbox();
 509     } else if (((flags_ & RMA) != 0) || static_cast<int>(size_) < async_small_thresh) { // eager mode
 510       mailbox = process->mailbox();
 511       XBT_DEBUG("Is there a corresponding recv already posted in the large mailbox %p?", mailbox);
 512       smx_activity_t action = simcall_comm_iprobe(mailbox, 1,dst_, tag_, &match_send,
 513                                                   static_cast<void*>(this));
 514       if (action == nullptr) {
 515         if ((flags_ & SSEND) == 0){
 516           mailbox = process->mailbox_small();
 517           XBT_DEBUG("No, nothing in the large mailbox, message is to be sent on the small one %p", mailbox);
 518         } else {
 519           mailbox = process->mailbox_small();
 520           XBT_DEBUG("SSEND : Is there a corresponding recv already posted in the small mailbox %p?", mailbox);
 521           action = simcall_comm_iprobe(mailbox, 1,dst_, tag_, &match_send, static_cast<void*>(this));
 522           if (action == nullptr) {
 523             XBT_DEBUG("No, we are first, send to large mailbox");
 524             mailbox = process->mailbox();
 525           }
 526         }
 527       } else {
 528         XBT_DEBUG("Yes there was something for us in the large mailbox");
 529       }
 530     } else {
 531       mailbox = process->mailbox();
 532       XBT_DEBUG("Send request %p is in the large mailbox %p (buf: %p)",mailbox, this,buf_);
 533     }
 534
 535     // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
 536     real_size_=size_;
 537     action_ = simcall_comm_isend(SIMIX_process_from_PID(src_+1), mailbox, size_, -1.0,
 538                                          buf, real_size_, &match_send,
 539                          &xbt_free_f, // how to free the userdata if a detached send fails
 540                          !process->replaying() ? smpi_comm_copy_data_callback
 541                          : &smpi_comm_null_copy_buffer_callback, this,
 542                          // detach if msg size < eager/rdv switch limit
 543                          detached_);
 544     XBT_DEBUG("send simcall posted");
 545
 546     /* FIXME: detached sends are not traceable (action_ == nullptr) */
 547     if (action_ != nullptr)
 548       simcall_set_category(action_, TRACE_internal_smpi_get_category());
 549     if (async_small_thresh != 0 || ((flags_ & RMA)!=0))
 550       xbt_mutex_release(mut);
 551   }
 552 }
 553
 554 void Request::startall(int count, MPI_Request * requests)
 555 {
 556   if(requests== nullptr)
 557     return;
 558
 559   for(int i = 0; i < count; i++) {
 560     requests[i]->start();
 561   }
 562 }
 563
 564 int Request::test(MPI_Request * request, MPI_Status * status) {
 565   //assume that request is not MPI_REQUEST_NULL (filtered in PMPI_Test or testall before)
 566   // to avoid deadlocks if used as a break condition, such as
 567   //     while (MPI_Test(request, flag, status) && flag) dostuff...
 568   // because the time will not normally advance when only calls to MPI_Test are made -> deadlock
 569   // multiplier to the sleeptime, to increase speed of execution, each failed test will increase it
 570   static int nsleeps = 1;
 571   if(smpi_test_sleep > 0)
 572     simcall_process_sleep(nsleeps*smpi_test_sleep);
 573
 574   Status::empty(status);
 575   int flag = 1;
 576   if (((*request)->flags_ & PREPARED) == 0) {
 577     if ((*request)->action_ != nullptr)
 578       flag = simcall_comm_test((*request)->action_);
 579     if (flag) {
 580       finish_wait(request,status);
 581       nsleeps=1;//reset the number of sleeps we will do next time
 582       if (*request != MPI_REQUEST_NULL && ((*request)->flags_ & PERSISTENT)==0)
 583       *request = MPI_REQUEST_NULL;
 584     } else if (xbt_cfg_get_boolean("smpi/grow-injected-times")){
 585       nsleeps++;
 586     }
 587   }
 588   return flag;
 589 }
 590
 591 int Request::testsome(int incount, MPI_Request requests[], int *indices, MPI_Status status[])
 592 {
 593   int i;
 594   int count = 0;
 595   int count_dead = 0;
 596   MPI_Status stat;
 597   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 598
 599   for(i = 0; i < incount; i++) {
 600     if((requests[i] != MPI_REQUEST_NULL)) {
 601       if(test(&requests[i], pstat)) {
 602          indices[i] = 1;
 603          count++;
 604          if(status != MPI_STATUSES_IGNORE) {
 605            status[i] = *pstat;
 606          }
 607          if ((requests[i] != MPI_REQUEST_NULL) && requests[i]->flags_ & NON_PERSISTENT)
 608          requests[i]=MPI_REQUEST_NULL;
 609       }
 610     }else{
 611       count_dead++;
 612     }
 613   }
 614   if(count_dead==incount)
 615     return MPI_UNDEFINED;
 616   else return count;
 617 }
 618
 619 int Request::testany(int count, MPI_Request requests[], int *index, MPI_Status * status)
 620 {
 621   std::vector<simgrid::kernel::activity::ActivityImpl*> comms;
 622   comms.reserve(count);
 623
 624   int i;
 625   int flag = 0;
 626
 627   *index = MPI_UNDEFINED;
 628
 629   std::vector<int> map; /** Maps all matching comms back to their location in requests **/
 630   for(i = 0; i < count; i++) {
 631     if ((requests[i] != MPI_REQUEST_NULL) && requests[i]->action_ && !(requests[i]->flags_ & PREPARED)) {
 632        comms.push_back(requests[i]->action_);
 633        map.push_back(i);
 634     }
 635   }
 636   if(!map.empty()) {
 637     //multiplier to the sleeptime, to increase speed of execution, each failed testany will increase it
 638     static int nsleeps = 1;
 639     if(smpi_test_sleep > 0)
 640       simcall_process_sleep(nsleeps*smpi_test_sleep);
 641
 642     i = simcall_comm_testany(comms.data(), comms.size()); // The i-th element in comms matches!
 643     if (i != -1) { // -1 is not MPI_UNDEFINED but a SIMIX return code. (nothing matches)
 644       *index = map[i];
 645       finish_wait(&requests[*index],status);
 646       flag             = 1;
 647       nsleeps          = 1;
 648       if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & NON_PERSISTENT)) {
 649         requests[*index] = MPI_REQUEST_NULL;
 650       }
 651     } else {
 652       nsleeps++;
 653     }
 654   } else {
 655       //all requests are null or inactive, return true
 656       flag = 1;
 657       Status::empty(status);
 658   }
 659
 660   return flag;
 661 }
 662
 663 int Request::testall(int count, MPI_Request requests[], MPI_Status status[])
 664 {
 665   MPI_Status stat;
 666   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 667   int flag=1;
 668   for(int i=0; i<count; i++){
 669     if (requests[i] != MPI_REQUEST_NULL && !(requests[i]->flags_ & PREPARED)) {
 670       if (test(&requests[i], pstat)!=1){
 671         flag=0;
 672       }else{
 673           requests[i]=MPI_REQUEST_NULL;
 674       }
 675     }else{
 676       Status::empty(pstat);
 677     }
 678     if(status != MPI_STATUSES_IGNORE) {
 679       status[i] = *pstat;
 680     }
 681   }
 682   return flag;
 683 }
 684
 685 void Request::probe(int source, int tag, MPI_Comm comm, MPI_Status* status){
 686   int flag=0;
 687   //FIXME find another way to avoid busy waiting ?
 688   // the issue here is that we have to wait on a nonexistent comm
 689   while(flag==0){
 690     iprobe(source, tag, comm, &flag, status);
 691     XBT_DEBUG("Busy Waiting on probing : %d", flag);
 692   }
 693 }
 694
 695 void Request::iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status* status){
 696   // to avoid deadlock, we have to sleep some time here, or the timer won't advance and we will only do iprobe simcalls
 697   // especially when used as a break condition, such as while (MPI_Iprobe(...)) dostuff...
 698   // nsleeps is a multiplier to the sleeptime, to increase speed of execution, each failed iprobe will increase it
 699   // This can speed up the execution of certain applications by an order of magnitude, such as HPL
 700   static int nsleeps = 1;
 701   double speed       = simgrid::s4u::Actor::self()->host()->speed();
 702   double maxrate = xbt_cfg_get_double("smpi/iprobe-cpu-usage");
 703   MPI_Request request = new Request(nullptr, 0, MPI_CHAR, source == MPI_ANY_SOURCE ? MPI_ANY_SOURCE :
 704                  comm->group()->index(source), comm->rank(), tag, comm, PERSISTENT | RECV);
 705   if (smpi_iprobe_sleep > 0) {
 706     smx_activity_t iprobe_sleep = simcall_execution_start("iprobe", /* flops to executek*/nsleeps*smpi_iprobe_sleep*speed*maxrate, /* priority */1.0, /* performance bound */maxrate*speed);
 707     simcall_execution_wait(iprobe_sleep);
 708   }
 709   // behave like a receive, but don't do it
 710   smx_mailbox_t mailbox;
 711
 712   request->print_request("New iprobe");
 713   // We have to test both mailboxes as we don't know if we will receive one one or another
 714   if (xbt_cfg_get_int("smpi/async-small-thresh") > 0){
 715       mailbox = smpi_process()->mailbox_small();
 716       XBT_DEBUG("Trying to probe the perm recv mailbox");
 717       request->action_ = simcall_comm_iprobe(mailbox, 0, request->src_, request->tag_, &match_recv,
 718                                             static_cast<void*>(request));
 719   }
 720
 721   if (request->action_ == nullptr){
 722     mailbox = smpi_process()->mailbox();
 723     XBT_DEBUG("trying to probe the other mailbox");
 724     request->action_ = simcall_comm_iprobe(mailbox, 0, request->src_,request->tag_, &match_recv,
 725                                           static_cast<void*>(request));
 726   }
 727
 728   if (request->action_ != nullptr){
 729     simgrid::kernel::activity::Comm *sync_comm = static_cast<simgrid::kernel::activity::Comm*>(request->action_);
 730     MPI_Request req                            = static_cast<MPI_Request>(sync_comm->src_data);
 731     *flag = 1;
 732     if(status != MPI_STATUS_IGNORE && (req->flags_ & PREPARED) == 0) {
 733       status->MPI_SOURCE = comm->group()->rank(req->src_);
 734       status->MPI_TAG    = req->tag_;
 735       status->MPI_ERROR  = MPI_SUCCESS;
 736       status->count      = req->real_size_;
 737     }
 738     nsleeps = 1;//reset the number of sleeps we will do next time
 739   }
 740   else {
 741     *flag = 0;
 742     if (xbt_cfg_get_boolean("smpi/grow-injected-times"))
 743       nsleeps++;
 744   }
 745   unref(&request);
 746 }
 747
 748 void Request::finish_wait(MPI_Request* request, MPI_Status * status)
 749 {
 750   MPI_Request req = *request;
 751   Status::empty(status);
 752
 753   if(!((req->detached_ != 0) && ((req->flags_ & SEND) != 0)) && ((req->flags_ & PREPARED) == 0)){
 754     if(status != MPI_STATUS_IGNORE) {
 755       int src = req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_;
 756       status->MPI_SOURCE = req->comm_->group()->rank(src);
 757       status->MPI_TAG = req->tag_ == MPI_ANY_TAG ? req->real_tag_ : req->tag_;
 758       status->MPI_ERROR = req->truncated_ != 0 ? MPI_ERR_TRUNCATE : MPI_SUCCESS;
 759       // this handles the case were size in receive differs from size in send
 760       status->count = req->real_size_;
 761     }
 762
 763     req->print_request("Finishing");
 764     MPI_Datatype datatype = req->old_type_;
 765
 766     if((((req->flags_ & ACCUMULATE) != 0) || (datatype->flags() & DT_FLAG_DERIVED)) && (!smpi_is_shared(req->old_buf_))){
 767
 768       if (!smpi_process()->replaying()){
 769         if( smpi_privatize_global_variables != 0 && (static_cast<char*>(req->old_buf_) >= smpi_start_data_exe)
 770             && ((char*)req->old_buf_ < smpi_start_data_exe + smpi_size_data_exe )){
 771             XBT_VERB("Privatization : We are unserializing to a zone in global memory  Switch data segment ");
 772             smpi_switch_data_segment(smpi_process()->index());
 773         }
 774       }
 775
 776       if(datatype->flags() & DT_FLAG_DERIVED){
 777         // This part handles the problem of non-contignous memory the unserialization at the reception
 778         if((req->flags_ & RECV) && datatype->size()!=0)
 779           datatype->unserialize(req->buf_, req->old_buf_, req->real_size_/datatype->size() , req->op_);
 780         xbt_free(req->buf_);
 781       }else if(req->flags_ & RECV){//apply op on contiguous buffer for accumulate
 782           if(datatype->size()!=0){
 783             int n =req->real_size_/datatype->size();
 784             req->op_->apply(req->buf_, req->old_buf_, &n, datatype);
 785           }
 786           xbt_free(req->buf_);
 787       }
 788     }
 789   }
 790
 791   if (TRACE_smpi_view_internals() && ((req->flags_ & RECV) != 0)){
 792     int rank = smpi_process()->index();
 793     int src_traced = (req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_);
 794     TRACE_smpi_recv(rank, src_traced, rank,req->tag_);
 795   }
 796   if(req->detached_sender_ != nullptr){
 797     //integrate pseudo-timing for buffering of small messages, do not bother to execute the simcall if 0
 798     double sleeptime = smpi_or(req->real_size_);
 799     if(sleeptime > 0.0){
 800       simcall_process_sleep(sleeptime);
 801       XBT_DEBUG("receiving size of %zu : sleep %f ", req->real_size_, sleeptime);
 802     }
 803     unref(&(req->detached_sender_));
 804   }
 805   if(req->flags_ & PERSISTENT)
 806     req->action_ = nullptr;
 807   req->flags_ |= FINISHED;
 808   unref(request);
 809 }
 810
 811 void Request::wait(MPI_Request * request, MPI_Status * status)
 812 {
 813   (*request)->print_request("Waiting");
 814   if ((*request)->flags_ & PREPARED) {
 815     Status::empty(status);
 816     return;
 817   }
 818
 819   if ((*request)->action_ != nullptr)
 820     // this is not a detached send
 821     simcall_comm_wait((*request)->action_, -1.0);
 822
 823   finish_wait(request,status);
 824   if (*request != MPI_REQUEST_NULL && (((*request)->flags_ & NON_PERSISTENT)!=0))
 825     *request = MPI_REQUEST_NULL;
 826 }
 827
 828 int Request::waitany(int count, MPI_Request requests[], MPI_Status * status)
 829 {
 830   s_xbt_dynar_t comms; // Keep it on stack to save some extra mallocs
 831   int i;
 832   int size = 0;
 833   int index = MPI_UNDEFINED;
 834   int *map;
 835
 836   if(count > 0) {
 837     // Wait for a request to complete
 838     xbt_dynar_init(&comms, sizeof(smx_activity_t), nullptr);
 839     map = xbt_new(int, count);
 840     XBT_DEBUG("Wait for one of %d", count);
 841     for(i = 0; i < count; i++) {
 842       if (requests[i] != MPI_REQUEST_NULL && !(requests[i]->flags_ & PREPARED) && !(requests[i]->flags_ & FINISHED)) {
 843         if (requests[i]->action_ != nullptr) {
 844           XBT_DEBUG("Waiting any %p ", requests[i]);
 845           xbt_dynar_push(&comms, &requests[i]->action_);
 846           map[size] = i;
 847           size++;
 848         } else {
 849           // This is a finished detached request, let's return this one
 850           size  = 0; // so we free the dynar but don't do the waitany call
 851           index = i;
 852           finish_wait(&requests[i], status); // cleanup if refcount = 0
 853           if (requests[i] != MPI_REQUEST_NULL && (requests[i]->flags_ & NON_PERSISTENT))
 854             requests[i] = MPI_REQUEST_NULL; // set to null
 855           break;
 856         }
 857       }
 858     }
 859     if(size > 0) {
 860       i = simcall_comm_waitany(&comms, -1);
 861
 862       // not MPI_UNDEFINED, as this is a simix return code
 863       if (i != -1) {
 864         index = map[i];
 865         //in case of an accumulate, we have to wait the end of all requests to apply the operation, ordered correctly.
 866         if ((requests[index] == MPI_REQUEST_NULL)
 867              ||  (!((requests[index]->flags_ & ACCUMULATE) && (requests[index]->flags_ & RECV)))){
 868           finish_wait(&requests[index],status);
 869           if (requests[i] != MPI_REQUEST_NULL && (requests[i]->flags_ & NON_PERSISTENT))
 870             requests[index] = MPI_REQUEST_NULL;
 871         }
 872       }
 873     }
 874
 875     xbt_dynar_free_data(&comms);
 876     xbt_free(map);
 877   }
 878
 879   if (index==MPI_UNDEFINED)
 880     Status::empty(status);
 881
 882   return index;
 883 }
 884
 885 static int sort_accumulates(MPI_Request a, MPI_Request b)
 886 {
 887   return (a->tag() > b->tag());
 888 }
 889
 890 int Request::waitall(int count, MPI_Request requests[], MPI_Status status[])
 891 {
 892   std::vector<MPI_Request> accumulates;
 893   int index;
 894   MPI_Status stat;
 895   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 896   int retvalue = MPI_SUCCESS;
 897   //tag invalid requests in the set
 898   if (status != MPI_STATUSES_IGNORE) {
 899     for (int c = 0; c < count; c++) {
 900       if (requests[c] == MPI_REQUEST_NULL || requests[c]->dst_ == MPI_PROC_NULL || (requests[c]->flags_ & PREPARED)) {
 901         Status::empty(&status[c]);
 902       } else if (requests[c]->src_ == MPI_PROC_NULL) {
 903         Status::empty(&status[c]);
 904         status[c].MPI_SOURCE = MPI_PROC_NULL;
 905       }
 906     }
 907   }
 908   for (int c = 0; c < count; c++) {
 909     if (MC_is_active() || MC_record_replay_is_active()) {
 910       wait(&requests[c],pstat);
 911       index = c;
 912     } else {
 913       index = waitany(count, requests, pstat);
 914       if (index == MPI_UNDEFINED)
 915         break;
 916
 917       if (requests[index] != MPI_REQUEST_NULL
 918            && (requests[index]->flags_ & RECV)
 919            && (requests[index]->flags_ & ACCUMULATE))
 920         accumulates.push_back(requests[index]);
 921       if (requests[index] != MPI_REQUEST_NULL && (requests[index]->flags_ & NON_PERSISTENT))
 922         requests[index] = MPI_REQUEST_NULL;
 923     }
 924     if (status != MPI_STATUSES_IGNORE) {
 925       status[index] = *pstat;
 926       if (status[index].MPI_ERROR == MPI_ERR_TRUNCATE)
 927         retvalue = MPI_ERR_IN_STATUS;
 928     }
 929   }
 930
 931   if (!accumulates.empty()) {
 932     std::sort(accumulates.begin(), accumulates.end(), sort_accumulates);
 933     for (auto req : accumulates) {
 934       finish_wait(&req, status);
 935     }
 936   }
 937
 938   return retvalue;
 939 }
 940
 941 int Request::waitsome(int incount, MPI_Request requests[], int *indices, MPI_Status status[])
 942 {
 943   int i;
 944   int count = 0;
 945   int index;
 946   MPI_Status stat;
 947   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 948
 949   for(i = 0; i < incount; i++)
 950   {
 951     index=waitany(incount, requests, pstat);
 952     if(index!=MPI_UNDEFINED){
 953       indices[count] = index;
 954       count++;
 955       if(status != MPI_STATUSES_IGNORE) {
 956         status[index] = *pstat;
 957       }
 958      if (requests[index] != MPI_REQUEST_NULL && (requests[index]->flags_ & NON_PERSISTENT))
 959      requests[index]=MPI_REQUEST_NULL;
 960     }else{
 961       return MPI_UNDEFINED;
 962     }
 963   }
 964   return count;
 965 }
 966
 967 MPI_Request Request::f2c(int id) {
 968   char key[KEY_SIZE];
 969   if(id==MPI_FORTRAN_REQUEST_NULL)
 970     return static_cast<MPI_Request>(MPI_REQUEST_NULL);
 971   return static_cast<MPI_Request>(xbt_dict_get(F2C::f2c_lookup(), get_key_id(key, id)));
 972 }
 973
 974 int Request::add_f() {
 975   if(F2C::f2c_lookup()==nullptr){
 976     F2C::set_f2c_lookup(xbt_dict_new_homogeneous(nullptr));
 977   }
 978   char key[KEY_SIZE];
 979   xbt_dict_set(F2C::f2c_lookup(), get_key_id(key, F2C::f2c_id()), this, nullptr);
 980   F2C::f2c_id_increment();
 981   return F2C::f2c_id()-1;
 982 }
 983
 984 void Request::free_f(int id) {
 985   char key[KEY_SIZE];
 986   if(id!=MPI_FORTRAN_REQUEST_NULL)
 987     xbt_dict_remove(F2C::f2c_lookup(), get_key_id(key, id));
 988 }
 989
 990 }
 991 }
 992
 993
 994