Logo AND Algorithmique Numérique Distribuée

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