Logo AND Algorithmique Numérique Distribuée

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