Logo AND Algorithmique Numérique Distribuée

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