src/smpi/smpi_request.cpp

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