Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
3999fe7d574006c0bd8429b17719631296fa454f
[simgrid.git] / src / simix / smx_network.cpp
1 /* Copyright (c) 2009-2016. 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 <algorithm>
7
8 #include <boost/range/algorithm.hpp>
9
10 #include <xbt/ex.hpp>
11
12 #include <simgrid/s4u/host.hpp>
13
14 #include "src/surf/surf_interface.hpp"
15 #include "src/simix/smx_private.h"
16 #include "xbt/log.h"
17 #include "mc/mc.h"
18 #include "src/mc/mc_replay.h"
19 #include "xbt/dict.h"
20 #include "simgrid/s4u/Mailbox.hpp"
21
22 #include "src/kernel/activity/SynchroComm.hpp"
23 #include "src/surf/network_interface.hpp"
24
25 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, "SIMIX network-related synchronization");
26
27 static void SIMIX_mbox_free(void *data);
28 static xbt_dict_t mailboxes = xbt_dict_new_homogeneous(SIMIX_mbox_free);
29
30 static void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall);
31 static void SIMIX_comm_copy_data(smx_activity_t comm);
32 static inline void SIMIX_mbox_push(smx_mailbox_t mbox, smx_activity_t comm);
33 static smx_activity_t _find_matching_comm(boost::circular_buffer_space_optimized<smx_activity_t> *deque, e_smx_comm_type_t type,
34     int (*match_fun)(void *, void *,smx_activity_t), void *user_data, smx_activity_t my_synchro, bool remove_matching);
35 static void SIMIX_comm_start(smx_activity_t synchro);
36
37 void SIMIX_mailbox_exit()
38 {
39   xbt_dict_free(&mailboxes);
40 }
41
42 /******************************************************************************/
43 /*                           Rendez-Vous Points                               */
44 /******************************************************************************/
45
46 smx_mailbox_t SIMIX_mbox_create(const char *name)
47 {
48   xbt_assert(name, "Mailboxes must have a name");
49   /* two processes may have pushed the same mbox_create simcall at the same time */
50   smx_mailbox_t mbox = static_cast<smx_mailbox_t>(xbt_dict_get_or_null(mailboxes, name));
51   if (!mbox) {
52     mbox = new simgrid::simix::Mailbox(name);
53     XBT_DEBUG("Creating a mailbox at %p with name %s", mbox, name);
54     xbt_dict_set(mailboxes, mbox->name, mbox, nullptr);
55   }
56   return mbox;
57 }
58
59 void SIMIX_mbox_free(void *data)
60 {
61   XBT_DEBUG("mbox free %p", data);
62   smx_mailbox_t mbox = static_cast<smx_mailbox_t>(data);
63   delete mbox;
64 }
65
66 smx_mailbox_t SIMIX_mbox_get_by_name(const char *name)
67 {
68   return static_cast<smx_mailbox_t>(xbt_dict_get_or_null(mailboxes, name));
69 }
70
71 /**
72  *  \brief set the receiver of the rendez vous point to allow eager sends
73  *  \param mbox The rendez-vous point
74  *  \param process The receiving process
75  */
76 void SIMIX_mbox_set_receiver(smx_mailbox_t mbox, smx_actor_t process)
77 {
78   mbox->permanent_receiver = process;
79 }
80
81 /**
82  *  \brief Pushes a communication synchro into a rendez-vous point
83  *  \param mbox The mailbox
84  *  \param synchro The communication synchro
85  */
86 static inline void SIMIX_mbox_push(smx_mailbox_t mbox, smx_activity_t synchro)
87 {
88   simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
89   mbox->comm_queue.push_back(comm);
90   comm->mbox = mbox;
91 }
92
93 /**
94  *  \brief Removes a communication synchro from a rendez-vous point
95  *  \param mbox The rendez-vous point
96  *  \param synchro The communication synchro
97  */
98 void SIMIX_mbox_remove(smx_mailbox_t mbox, smx_activity_t synchro)
99 {
100   simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
101
102   comm->mbox = nullptr;
103   for (auto it = mbox->comm_queue.begin(); it != mbox->comm_queue.end(); it++)
104     if (*it == comm) {
105       mbox->comm_queue. erase(it);
106       return;
107     }
108   xbt_die("Cannot remove the comm %p that is not part of the mailbox %s",comm, mbox->name);
109 }
110
111 /**
112  *  \brief Checks if there is a communication synchro queued in a deque matching our needs
113  *  \param type The type of communication we are looking for (comm_send, comm_recv)
114  *  \return The communication synchro if found, nullptr otherwise
115  */
116 static smx_activity_t _find_matching_comm(boost::circular_buffer_space_optimized<smx_activity_t> *deque, e_smx_comm_type_t type,
117     int (*match_fun)(void *, void *,smx_activity_t), void *this_user_data, smx_activity_t my_synchro, bool remove_matching)
118 {
119   void* other_user_data = nullptr;
120
121   for(auto it = deque->begin(); it != deque->end(); it++){
122     smx_activity_t synchro = *it;
123     simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
124
125     if (comm->type == SIMIX_COMM_SEND) {
126       other_user_data = comm->src_data;
127     } else if (comm->type == SIMIX_COMM_RECEIVE) {
128       other_user_data = comm->dst_data;
129     }
130     if (comm->type == type &&
131         (!      match_fun ||       match_fun(this_user_data,  other_user_data, synchro)) &&
132         (!comm->match_fun || comm->match_fun(other_user_data, this_user_data,  my_synchro))) {
133       XBT_DEBUG("Found a matching communication synchro %p", comm);
134       if (remove_matching)
135         deque->erase(it);
136       comm->ref();
137 #if HAVE_MC
138       comm->mbox_cpy = comm->mbox;
139 #endif
140       comm->mbox = nullptr;
141       return comm;
142     }
143     XBT_DEBUG("Sorry, communication synchro %p does not match our needs:"
144               " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
145               comm, (int)comm->type, (int)type);
146   }
147   XBT_DEBUG("No matching communication synchro found");
148   return nullptr;
149 }
150
151 /******************************************************************************/
152 /*                          Communication synchros                            */
153 /******************************************************************************/
154 XBT_PRIVATE void simcall_HANDLER_comm_send(smx_simcall_t simcall, smx_actor_t src, smx_mailbox_t mbox,
155                                   double task_size, double rate,
156                                   void *src_buff, size_t src_buff_size,
157                                   int (*match_fun)(void *, void *,smx_activity_t),
158                                   void (*copy_data_fun)(smx_activity_t, void*, size_t),
159           void *data, double timeout){
160   smx_activity_t comm = simcall_HANDLER_comm_isend(simcall, src, mbox, task_size, rate,
161                            src_buff, src_buff_size, match_fun, nullptr, copy_data_fun,
162                data, 0);
163   SIMCALL_SET_MC_VALUE(simcall, 0);
164   simcall_HANDLER_comm_wait(simcall, comm, timeout);
165 }
166 XBT_PRIVATE smx_activity_t simcall_HANDLER_comm_isend(smx_simcall_t simcall, smx_actor_t src_proc, smx_mailbox_t mbox,
167                                   double task_size, double rate,
168                                   void *src_buff, size_t src_buff_size,
169                                   int (*match_fun)(void *, void *,smx_activity_t),
170                                   void (*clean_fun)(void *), // used to free the synchro in case of problem after a detached send
171                                   void (*copy_data_fun)(smx_activity_t, void*, size_t),// used to copy data if not default one
172                           void *data, int detached)
173 {
174   XBT_DEBUG("send from %p", mbox);
175
176   /* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
177   simgrid::kernel::activity::Comm* this_synchro = new simgrid::kernel::activity::Comm(SIMIX_COMM_SEND);
178
179   /* Look for communication synchro matching our needs. We also provide a description of
180    * ourself so that the other side also gets a chance of choosing if it wants to match with us.
181    *
182    * If it is not found then push our communication into the rendez-vous point */
183   smx_activity_t other_synchro =
184       _find_matching_comm(&mbox->comm_queue, SIMIX_COMM_RECEIVE, match_fun, data, this_synchro, /*remove_matching*/true);
185   simgrid::kernel::activity::Comm *other_comm = static_cast<simgrid::kernel::activity::Comm*>(other_synchro);
186
187
188   if (!other_synchro) {
189     other_synchro = this_synchro;
190     other_comm = static_cast<simgrid::kernel::activity::Comm*>(other_synchro);
191
192     if (mbox->permanent_receiver!=nullptr){
193       //this mailbox is for small messages, which have to be sent right now
194       other_synchro->state = SIMIX_READY;
195       other_comm->dst_proc=mbox->permanent_receiver.get();
196       other_comm->ref();
197       mbox->done_comm_queue.push_back(other_synchro);
198       XBT_DEBUG("pushing a message into the permanent receive fifo %p, comm %p", mbox, &(other_comm));
199
200     }else{
201       SIMIX_mbox_push(mbox, this_synchro);
202     }
203   } else {
204     XBT_DEBUG("Receive already pushed");
205     this_synchro->unref();
206
207     other_comm->state = SIMIX_READY;
208     other_comm->type = SIMIX_COMM_READY;
209
210   }
211   xbt_fifo_push(src_proc->comms, other_synchro);
212
213
214   if (detached) {
215     other_comm->detached = true;
216     other_comm->clean_fun = clean_fun;
217   } else {
218     other_comm->clean_fun = nullptr;
219   }
220
221   /* Setup the communication synchro */
222   other_comm->src_proc = src_proc;
223   other_comm->task_size = task_size;
224   other_comm->rate = rate;
225   other_comm->src_buff = src_buff;
226   other_comm->src_buff_size = src_buff_size;
227   other_comm->src_data = data;
228
229   other_comm->match_fun = match_fun;
230   other_comm->copy_data_fun = copy_data_fun;
231
232
233   if (MC_is_active() || MC_record_replay_is_active()) {
234     other_comm->state = SIMIX_RUNNING;
235     return (detached ? nullptr : other_comm);
236   }
237
238   SIMIX_comm_start(other_comm);
239   return (detached ? nullptr : other_comm);
240 }
241
242 XBT_PRIVATE void simcall_HANDLER_comm_recv(smx_simcall_t simcall, smx_actor_t receiver, smx_mailbox_t mbox,
243                          void *dst_buff, size_t *dst_buff_size,
244                          int (*match_fun)(void *, void *, smx_activity_t),
245                          void (*copy_data_fun)(smx_activity_t, void*, size_t),
246                          void *data, double timeout, double rate)
247 {
248   smx_activity_t comm = SIMIX_comm_irecv(receiver, mbox, dst_buff, dst_buff_size, match_fun, copy_data_fun, data, rate);
249   SIMCALL_SET_MC_VALUE(simcall, 0);
250   simcall_HANDLER_comm_wait(simcall, comm, timeout);
251 }
252
253 XBT_PRIVATE smx_activity_t simcall_HANDLER_comm_irecv(smx_simcall_t simcall, smx_actor_t receiver, smx_mailbox_t mbox,
254     void *dst_buff, size_t *dst_buff_size,
255     int (*match_fun)(void *, void *, smx_activity_t),
256     void (*copy_data_fun)(smx_activity_t, void*, size_t),
257     void *data, double rate)
258 {
259   return SIMIX_comm_irecv(receiver, mbox, dst_buff, dst_buff_size, match_fun, copy_data_fun, data, rate);
260 }
261
262 smx_activity_t SIMIX_comm_irecv(smx_actor_t dst_proc, smx_mailbox_t mbox, void *dst_buff, size_t *dst_buff_size,
263     int (*match_fun)(void *, void *, smx_activity_t),
264     void (*copy_data_fun)(smx_activity_t, void*, size_t), // used to copy data if not default one
265     void *data, double rate)
266 {
267   XBT_DEBUG("recv from %p %p", mbox, &mbox->comm_queue);
268   simgrid::kernel::activity::Comm* this_synchro = new simgrid::kernel::activity::Comm(SIMIX_COMM_RECEIVE);
269
270   smx_activity_t other_synchro;
271   //communication already done, get it inside the fifo of completed comms
272   if (mbox->permanent_receiver != nullptr && ! mbox->done_comm_queue.empty()) {
273
274     XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication");
275     //find a match in the already received fifo
276     other_synchro = _find_matching_comm(&mbox->done_comm_queue, SIMIX_COMM_SEND, match_fun, data, this_synchro,/*remove_matching*/true);
277     //if not found, assume the receiver came first, register it to the mailbox in the classical way
278     if (!other_synchro)  {
279       XBT_DEBUG("We have messages in the permanent receive list, but not the one we are looking for, pushing request into fifo");
280       other_synchro = this_synchro;
281       SIMIX_mbox_push(mbox, this_synchro);
282     } else {
283       simgrid::kernel::activity::Comm *other_comm = static_cast<simgrid::kernel::activity::Comm*>(other_synchro);
284
285       if(other_comm->surf_comm && other_comm->remains() < 1e-12) {
286         XBT_DEBUG("comm %p has been already sent, and is finished, destroy it",other_comm);
287         other_comm->state = SIMIX_DONE;
288         other_comm->type = SIMIX_COMM_DONE;
289         other_comm->mbox = nullptr;
290       }
291       other_comm->unref();
292       static_cast<simgrid::kernel::activity::Comm*>(this_synchro)->unref();
293     }
294   } else {
295     /* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
296
297     /* Look for communication synchro matching our needs. We also provide a description of
298      * ourself so that the other side also gets a chance of choosing if it wants to match with us.
299      *
300      * If it is not found then push our communication into the rendez-vous point */
301     other_synchro = _find_matching_comm(&mbox->comm_queue, SIMIX_COMM_SEND, match_fun, data, this_synchro,/*remove_matching*/true);
302
303     if (!other_synchro) {
304       XBT_DEBUG("Receive pushed first %zu", mbox->comm_queue.size());
305       other_synchro = this_synchro;
306       SIMIX_mbox_push(mbox, this_synchro);
307     } else {
308       this_synchro->unref();
309       simgrid::kernel::activity::Comm *other_comm = static_cast<simgrid::kernel::activity::Comm*>(other_synchro);
310
311       other_comm->state = SIMIX_READY;
312       other_comm->type = SIMIX_COMM_READY;
313     }
314     xbt_fifo_push(dst_proc->comms, other_synchro);
315   }
316
317   /* Setup communication synchro */
318   simgrid::kernel::activity::Comm *other_comm = static_cast<simgrid::kernel::activity::Comm*>(other_synchro);
319   other_comm->dst_proc = dst_proc;
320   other_comm->dst_buff = dst_buff;
321   other_comm->dst_buff_size = dst_buff_size;
322   other_comm->dst_data = data;
323
324   if (rate > -1.0 && (other_comm->rate < 0.0 || rate < other_comm->rate))
325     other_comm->rate = rate;
326
327   other_comm->match_fun = match_fun;
328   other_comm->copy_data_fun = copy_data_fun;
329
330   if (MC_is_active() || MC_record_replay_is_active()) {
331     other_synchro->state = SIMIX_RUNNING;
332     return other_synchro;
333   }
334
335   SIMIX_comm_start(other_synchro);
336   return other_synchro;
337 }
338
339 smx_activity_t simcall_HANDLER_comm_iprobe(smx_simcall_t simcall, smx_mailbox_t mbox,
340                                    int type, int src, int tag,
341                                    int (*match_fun)(void *, void *, smx_activity_t),
342                                    void *data){
343   return SIMIX_comm_iprobe(simcall->issuer, mbox, type, src, tag, match_fun, data);
344 }
345
346 smx_activity_t SIMIX_comm_iprobe(smx_actor_t dst_proc, smx_mailbox_t mbox, int type, int src,
347                               int tag, int (*match_fun)(void *, void *, smx_activity_t), void *data)
348 {
349   XBT_DEBUG("iprobe from %p %p", mbox, &mbox->comm_queue);
350   simgrid::kernel::activity::Comm* this_comm;
351   int smx_type;
352   if(type == 1){
353     this_comm = new simgrid::kernel::activity::Comm(SIMIX_COMM_SEND);
354     smx_type = SIMIX_COMM_RECEIVE;
355   } else{
356     this_comm = new simgrid::kernel::activity::Comm(SIMIX_COMM_RECEIVE);
357     smx_type = SIMIX_COMM_SEND;
358   } 
359   smx_activity_t other_synchro=nullptr;
360   if (mbox->permanent_receiver != nullptr && !mbox->done_comm_queue.empty()) {
361     XBT_DEBUG("first check in the permanent recv mailbox, to see if we already got something");
362     other_synchro = _find_matching_comm(&mbox->done_comm_queue,
363       (e_smx_comm_type_t) smx_type, match_fun, data, this_comm,/*remove_matching*/false);
364   }
365   if (!other_synchro){
366     XBT_DEBUG("check if we have more luck in the normal mailbox");
367     other_synchro = _find_matching_comm(&mbox->comm_queue,
368       (e_smx_comm_type_t) smx_type, match_fun, data, this_comm,/*remove_matching*/false);
369   }
370
371   if(other_synchro)
372     other_synchro->unref();
373
374   this_comm->unref();
375   return other_synchro;
376 }
377
378 void simcall_HANDLER_comm_wait(smx_simcall_t simcall, smx_activity_t synchro, double timeout)
379 {
380   /* Associate this simcall to the wait synchro */
381   XBT_DEBUG("simcall_HANDLER_comm_wait, %p", synchro);
382
383   synchro->simcalls.push_back(simcall);
384   simcall->issuer->waiting_synchro = synchro;
385
386   if (MC_is_active() || MC_record_replay_is_active()) {
387     int idx = SIMCALL_GET_MC_VALUE(simcall);
388     if (idx == 0) {
389       synchro->state = SIMIX_DONE;
390     } else {
391       /* If we reached this point, the wait simcall must have a timeout */
392       /* Otherwise it shouldn't be enabled and executed by the MC */
393       if (timeout < 0.0)
394         THROW_IMPOSSIBLE;
395
396       simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
397       if (comm->src_proc == simcall->issuer)
398         comm->state = SIMIX_SRC_TIMEOUT;
399       else
400         comm->state = SIMIX_DST_TIMEOUT;
401     }
402
403     SIMIX_comm_finish(synchro);
404     return;
405   }
406
407   /* If the synchro has already finish perform the error handling, */
408   /* otherwise set up a waiting timeout on the right side          */
409   if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) {
410     SIMIX_comm_finish(synchro);
411   } else { /* if (timeout >= 0) { we need a surf sleep action even when there is no timeout, otherwise surf won't tell us when the host fails */
412     surf_action_t sleep = surf_host_sleep(simcall->issuer->host, timeout);
413     sleep->setData(synchro);
414
415     simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
416     if (simcall->issuer == comm->src_proc)
417       comm->src_timeout = sleep;
418     else
419       comm->dst_timeout = sleep;
420   }
421 }
422
423 void simcall_HANDLER_comm_test(smx_simcall_t simcall, smx_activity_t synchro)
424 {
425   simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
426
427   if (MC_is_active() || MC_record_replay_is_active()){
428     simcall_comm_test__set__result(simcall, comm->src_proc && comm->dst_proc);
429     if (simcall_comm_test__get__result(simcall)){
430       synchro->state = SIMIX_DONE;
431       synchro->simcalls.push_back(simcall);
432       SIMIX_comm_finish(synchro);
433     } else {
434       SIMIX_simcall_answer(simcall);
435     }
436     return;
437   }
438
439   simcall_comm_test__set__result(simcall, (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING));
440   if (simcall_comm_test__get__result(simcall)) {
441     synchro->simcalls.push_back(simcall);
442     SIMIX_comm_finish(synchro);
443   } else {
444     SIMIX_simcall_answer(simcall);
445   }
446 }
447
448 void simcall_HANDLER_comm_testany(
449   smx_simcall_t simcall, simgrid::kernel::activity::ActivityImpl* comms[], size_t count)
450 {
451   // The default result is -1 -- this means, "nothing is ready".
452   // It can be changed below, but only if something matches.
453   simcall_comm_testany__set__result(simcall, -1);
454
455   if (MC_is_active() || MC_record_replay_is_active()){
456     int idx = SIMCALL_GET_MC_VALUE(simcall);
457     if(idx == -1){
458       SIMIX_simcall_answer(simcall);
459     }else{
460       simgrid::kernel::activity::ActivityImpl* synchro = comms[idx];
461       simcall_comm_testany__set__result(simcall, idx);
462       synchro->simcalls.push_back(simcall);
463       synchro->state = SIMIX_DONE;
464       SIMIX_comm_finish(synchro);
465     }
466     return;
467   }
468
469   for (std::size_t i = 0; i != count; ++i) {
470     simgrid::kernel::activity::ActivityImpl* synchro = comms[i];
471     if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) {
472       simcall_comm_testany__set__result(simcall, i);
473       synchro->simcalls.push_back(simcall);
474       SIMIX_comm_finish(synchro);
475       return;
476     }
477   }
478   SIMIX_simcall_answer(simcall);
479 }
480
481 void simcall_HANDLER_comm_waitany(smx_simcall_t simcall, xbt_dynar_t synchros, double timeout)
482 {
483   smx_activity_t synchro;
484   unsigned int cursor = 0;
485
486   if (MC_is_active() || MC_record_replay_is_active()){
487     if (timeout > 0.0)
488       xbt_die("Timeout not implemented for waitany in the model-checker"); 
489     int idx = SIMCALL_GET_MC_VALUE(simcall);
490     synchro = xbt_dynar_get_as(synchros, idx, smx_activity_t);
491     synchro->simcalls.push_back(simcall);
492     simcall_comm_waitany__set__result(simcall, idx);
493     synchro->state = SIMIX_DONE;
494     SIMIX_comm_finish(synchro);
495     return;
496   }
497   
498   if (timeout < 0.0){
499     simcall->timer = NULL;
500   } else {
501     simcall->timer = SIMIX_timer_set(SIMIX_get_clock() + timeout, [simcall]() {
502       SIMIX_waitany_remove_simcall_from_actions(simcall);
503       simcall_comm_waitany__set__result(simcall, -1);
504       SIMIX_simcall_answer(simcall);
505     });
506   }
507   
508   xbt_dynar_foreach(synchros, cursor, synchro){
509     /* associate this simcall to the the synchro */
510     synchro->simcalls.push_back(simcall);
511
512     /* see if the synchro is already finished */
513     if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING){
514       SIMIX_comm_finish(synchro);
515       break;
516     }
517   }
518 }
519
520 void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall)
521 {
522   smx_activity_t synchro;
523   unsigned int cursor = 0;
524   xbt_dynar_t synchros = simcall_comm_waitany__get__comms(simcall);
525
526   xbt_dynar_foreach(synchros, cursor, synchro) {
527     // Remove the first occurence of simcall:
528     auto i = boost::range::find(synchro->simcalls, simcall);
529     if (i !=  synchro->simcalls.end())
530       synchro->simcalls.erase(i);
531   }
532 }
533
534 /**
535  *  \brief Starts the simulation of a communication synchro.
536  *  \param synchro the communication synchro
537  */
538 static inline void SIMIX_comm_start(smx_activity_t synchro)
539 {
540   simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
541
542   /* If both the sender and the receiver are already there, start the communication */
543   if (synchro->state == SIMIX_READY) {
544
545     simgrid::s4u::Host* sender   = comm->src_proc->host;
546     simgrid::s4u::Host* receiver = comm->dst_proc->host;
547
548     XBT_DEBUG("Starting communication %p from '%s' to '%s'", synchro, sg_host_get_name(sender), sg_host_get_name(receiver));
549
550     comm->surf_comm = surf_network_model->communicate(sender, receiver, comm->task_size, comm->rate);
551     comm->surf_comm->setData(synchro);
552     comm->state = SIMIX_RUNNING;
553
554     /* If a link is failed, detect it immediately */
555     if (comm->surf_comm->getState() == simgrid::surf::Action::State::failed) {
556       XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure",
557                 sg_host_get_name(sender), sg_host_get_name(receiver));
558       comm->state = SIMIX_LINK_FAILURE;
559       comm->cleanupSurf();
560     }
561
562     /* If any of the process is suspend, create the synchro but stop its execution,
563        it will be restarted when the sender process resume */
564     if (SIMIX_process_is_suspended(comm->src_proc) || SIMIX_process_is_suspended(comm->dst_proc)) {
565       if (SIMIX_process_is_suspended(comm->src_proc))
566         XBT_DEBUG("The communication is suspended on startup because src (%s@%s) was suspended since it initiated the communication",
567             comm->src_proc->name.c_str(), sg_host_get_name(comm->src_proc->host));
568       else
569         XBT_DEBUG("The communication is suspended on startup because dst (%s@%s) was suspended since it initiated the communication",
570             comm->dst_proc->name.c_str(), sg_host_get_name(comm->dst_proc->host));
571
572       comm->surf_comm->suspend();
573     }
574   }
575 }
576
577 /**
578  * \brief Answers the SIMIX simcalls associated to a communication synchro.
579  * \param synchro a finished communication synchro
580  */
581 void SIMIX_comm_finish(smx_activity_t synchro)
582 {
583   simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
584   unsigned int destroy_count = 0;
585
586   while (!synchro->simcalls.empty()) {
587     smx_simcall_t simcall = synchro->simcalls.front();
588     synchro->simcalls.pop_front();
589
590     /* If a waitany simcall is waiting for this synchro to finish, then remove
591        it from the other synchros in the waitany list. Afterwards, get the
592        position of the actual synchro in the waitany dynar and
593        return it as the result of the simcall */
594
595     if (simcall->call == SIMCALL_NONE) //FIXME: maybe a better way to handle this case
596       continue; // if process handling comm is killed
597     if (simcall->call == SIMCALL_COMM_WAITANY) {
598       SIMIX_waitany_remove_simcall_from_actions(simcall);
599       if (simcall->timer) {
600         SIMIX_timer_remove(simcall->timer);
601         simcall->timer = nullptr;
602       }
603       if (!MC_is_active() && !MC_record_replay_is_active())
604         simcall_comm_waitany__set__result(simcall, xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro));
605     }
606
607     /* If the synchro is still in a rendez-vous point then remove from it */
608     if (comm->mbox)
609       SIMIX_mbox_remove(comm->mbox, synchro);
610
611     XBT_DEBUG("SIMIX_comm_finish: synchro state = %d", (int)synchro->state);
612
613     /* Check out for errors */
614
615     if (simcall->issuer->host->isOff()) {
616       simcall->issuer->context->iwannadie = 1;
617       SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
618     } else {
619       switch (synchro->state) {
620
621       case SIMIX_DONE:
622         XBT_DEBUG("Communication %p complete!", synchro);
623         SIMIX_comm_copy_data(synchro);
624         break;
625
626       case SIMIX_SRC_TIMEOUT:
627         SMX_EXCEPTION(simcall->issuer, timeout_error, 0, "Communication timeouted because of sender");
628         break;
629
630       case SIMIX_DST_TIMEOUT:
631         SMX_EXCEPTION(simcall->issuer, timeout_error, 0, "Communication timeouted because of receiver");
632         break;
633
634       case SIMIX_SRC_HOST_FAILURE:
635         if (simcall->issuer == comm->src_proc)
636           simcall->issuer->context->iwannadie = 1;
637   //          SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
638         else
639           SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
640         break;
641
642       case SIMIX_DST_HOST_FAILURE:
643         if (simcall->issuer == comm->dst_proc)
644           simcall->issuer->context->iwannadie = 1;
645   //          SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
646         else
647           SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
648         break;
649
650       case SIMIX_LINK_FAILURE:
651
652         XBT_DEBUG("Link failure in synchro %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d",
653                   synchro,
654                   comm->src_proc ? sg_host_get_name(comm->src_proc->host) : nullptr,
655                   comm->dst_proc ? sg_host_get_name(comm->dst_proc->host) : nullptr,
656                   simcall->issuer->name.c_str(), simcall->issuer, comm->detached);
657         if (comm->src_proc == simcall->issuer) {
658           XBT_DEBUG("I'm source");
659         } else if (comm->dst_proc == simcall->issuer) {
660           XBT_DEBUG("I'm dest");
661         } else {
662           XBT_DEBUG("I'm neither source nor dest");
663         }
664         SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure");
665         break;
666
667       case SIMIX_CANCELED:
668         if (simcall->issuer == comm->dst_proc)
669           SMX_EXCEPTION(simcall->issuer, cancel_error, 0, "Communication canceled by the sender");
670         else
671           SMX_EXCEPTION(simcall->issuer, cancel_error, 0, "Communication canceled by the receiver");
672         break;
673
674       default:
675         xbt_die("Unexpected synchro state in SIMIX_comm_finish: %d", (int)synchro->state);
676       }
677     }
678
679     /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */
680     if (simcall->issuer->exception) {
681       // In order to modify the exception we have to rethrow it:
682       try {
683         std::rethrow_exception(simcall->issuer->exception);
684       }
685       catch(xbt_ex& e) {
686         if (simcall->call == SIMCALL_COMM_WAITANY) {
687           e.value = xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro);
688         }
689         else if (simcall->call == SIMCALL_COMM_TESTANY) {
690           e.value = -1;
691           auto comms = simcall_comm_testany__get__comms(simcall);
692           auto count = simcall_comm_testany__get__count(simcall);
693           auto element = std::find(comms, comms + count, synchro);
694           if (element == comms + count)
695             e.value = -1;
696           else
697             e.value = element - comms;
698         }
699         simcall->issuer->exception = std::make_exception_ptr(e);
700       }
701       catch(...) {
702         // Nothing to do
703       }
704     }
705
706     if (simcall->issuer->host->isOff()) {
707       simcall->issuer->context->iwannadie = 1;
708     }
709
710     simcall->issuer->waiting_synchro = nullptr;
711     xbt_fifo_remove(simcall->issuer->comms, synchro);
712     if(comm->detached){
713       if(simcall->issuer == comm->src_proc){
714         if(comm->dst_proc)
715           xbt_fifo_remove(comm->dst_proc->comms, synchro);
716       }
717       if(simcall->issuer == comm->dst_proc){
718         if(comm->src_proc)
719           xbt_fifo_remove(comm->src_proc->comms, synchro);
720         //in case of a detached comm we have an extra ref to remove, as the sender won't do it
721         destroy_count++;
722       }
723     }
724     SIMIX_simcall_answer(simcall);
725     destroy_count++;
726   }
727
728   while (destroy_count-- > 0)
729     static_cast<simgrid::kernel::activity::Comm*>(synchro)->unref();
730 }
731
732 /******************************************************************************/
733 /*                    SIMIX_comm_copy_data callbacks                       */
734 /******************************************************************************/
735 static void (*SIMIX_comm_copy_data_callback) (smx_activity_t, void*, size_t) = &SIMIX_comm_copy_pointer_callback;
736
737 void SIMIX_comm_set_copy_data_callback(void (*callback) (smx_activity_t, void*, size_t))
738 {
739   SIMIX_comm_copy_data_callback = callback;
740 }
741
742 void SIMIX_comm_copy_pointer_callback(smx_activity_t synchro, void* buff, size_t buff_size)
743 {
744   simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
745
746   xbt_assert((buff_size == sizeof(void *)), "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
747   *(void **) (comm->dst_buff) = buff;
748 }
749
750 void SIMIX_comm_copy_buffer_callback(smx_activity_t synchro, void* buff, size_t buff_size)
751 {
752   simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
753
754   XBT_DEBUG("Copy the data over");
755   memcpy(comm->dst_buff, buff, buff_size);
756   if (comm->detached) { // if this is a detached send, the source buffer was duplicated by SMPI sender to make the original buffer available to the application ASAP
757     xbt_free(buff);
758     comm->src_buff = nullptr;
759   }
760 }
761
762
763 /**
764  *  \brief Copy the communication data from the sender's buffer to the receiver's one
765  *  \param comm The communication
766  */
767 void SIMIX_comm_copy_data(smx_activity_t synchro)
768 {
769   simgrid::kernel::activity::Comm *comm = static_cast<simgrid::kernel::activity::Comm*>(synchro);
770
771   size_t buff_size = comm->src_buff_size;
772   /* If there is no data to copy then return */
773   if (!comm->src_buff || !comm->dst_buff || comm->copied)
774     return;
775
776   XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)",
777             comm,
778             comm->src_proc ? sg_host_get_name(comm->src_proc->host) : "a finished process",
779             comm->src_buff,
780             comm->dst_proc ? sg_host_get_name(comm->dst_proc->host) : "a finished process",
781             comm->dst_buff, buff_size);
782
783   /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
784   if (comm->dst_buff_size)
785     buff_size = MIN(buff_size, *(comm->dst_buff_size));
786
787   /* Update the receiver's buffer size to the copied amount */
788   if (comm->dst_buff_size)
789     *comm->dst_buff_size = buff_size;
790
791   if (buff_size > 0){
792       if(comm->copy_data_fun)
793         comm->copy_data_fun (comm, comm->src_buff, buff_size);
794       else
795         SIMIX_comm_copy_data_callback (comm, comm->src_buff, buff_size);
796   }
797
798
799   /* Set the copied flag so we copy data only once */
800   /* (this function might be called from both communication ends) */
801   comm->copied = 1;
802 }